InfluxDB & Kapacitor & Chronograf Kullanarak Çoklu Koşul Bildirimleri Oluşturmak

by TygrVrli · 16 Mayıs 2020

Arkadaşlar merhaba ,

Bir önceki yazımda InfluxDB , Kapacitor kullanarak bir değer üzerinden deadman bildirimleri oluşturmuştuk. O yazıma şuradan ulaşabilirsiniz.

Normalde Chronograf-UI kullanarak tek bir değer üzerinden kural oluşturmak oldukça kolay. Fakat birden fazla değer ile oluşturmak istediğimiz zaman biraz ellerimizi kirletmemiz gerekiyor.

Chronograf ve Kapacitor aslında InfluxDB ile sistemleri monitörize etmek için kullanılıyor. Yani sistem durumunu(ram , cpu , network , disk) InfluxDB’ye kayıt edip herhangi bir anomali oluştuğu zaman bunları takip etmek ve bildirmek için tasarlanmış. Fakat ben yazılarımda bu ekosistemi IoT dünyasına uyarlamaya çalışıyorum. Çoklu koşul bildirimlerine de IoT dünyasında çok sık ihtiyaç duyuluyor. Bir örnek vermemiz gerekirse ,

bir odamız olduğunu düşünelim. Bu odada bir adet sıcaklık sensörü var ve sıcaklık 23 derecenin üstüne çıktığında alarm oluşturmak istiyoruz. Bunu chronograf ile yapmak oldukça kolay. Fakat iş akışımızı şu şekilde değiştirelim. Oda Sıcaklığı 23 derecenin üstüne çıkmışsa ve klima çalışmıyor ise.

Bu durumda Chronograf arayüzü bize çoklu sensörler ile çalışma imkanı vermiyor. Bu noktada ise TICKScript imdadımıza yetişiyor.

Normalde UI kullanarak oluşturduğumuz kurallar için arka planda otomatik olarak bir TICKScript oluşturuluyor.

Aşağıdaki görüntüde kural isimleri ile aynı isimde TICKScriptleri görebilirsiniz.

TICKScript Nedir ?

Kapacitor tarafından kendisine beslenen verileri dönüştürmek , streamdan okumak ve değişiklikleri izlemek için kullanılan bir Domain Specific Language’dir. TICKScript’in kapacitor içerisindeki öncelikli görevi alarmları tanımlamaktır. Kaynak

TICKScript kullanarak stream ve ya batch datalar üzerinde işlem yapılabilir. Stream data ile işlem yapılırken point by point olarak data streamdan okunur ve stream için bir TICKScript yazıldığı zaman Kapacitor bu stream içerisindeki verilere otomatik olarak abone olur. Batch data üzerinde işlem yapılırken ise data çok historic data ile çalışılır. Ve sorgulamalar üzerinden ilgili datalara ulaşılır.

Aşağıdaki TICKScript daha önceki yazımızda UI kullanarak oluşturduğumuz DeadMan kuralının otomatik oluşturulan TICKScript çıktısı. Şimdi bu kodu parça parça inceleyelim.

var db = 'deadman_notification_sample'

var rp = 'autogen'

var measurement = 'HeartBeat'

var groupBy = []

var whereFilter = lambda: ("HeartBeatTag" == 'HeartBeatTag')

var period = 1m

0s

var db = 'deadman_notification_sample'

var rp = 'autogen'

var measurement = 'HeartBeat'

var groupBy = []

var whereFilter = lambda: ("HeartBeatTag" == 'HeartBeatTag')

var period = 1m

var db satırında kuralımızın çalışmasını istediğimiz InfluxDB ismini tanımlıyoruz.

var rp satırında Retention Policy tanımımızı yaptık. Bu alan influxDB’de özel bir belirteç kullanmazsak ‘autogen’ olarak geliyor.

var measurement satırı ise ‘deadman_notification_sample’ veritabanında kullanmak istediğimiz veriyi temsil ediyor.

Aşağıda önceki yazımdan bir kod parçası var. Burada InfluxDB’ye veri yazan uygulamamızda Point.Name alanında yazılı olan HeartBeat ismini görüyoruz. Kuralımızda kullanacağımız measurement de bu point aslında.

var point_model = new Point()
                    {
                        Name = "HeartBeat", // table name
                        Fields = new Dictionary<string, object>() { { "HeartBeatValue", "OK" } },
                        Timestamp = DateTime.Now,
                        Tags = new Dictionary<string, object>() { { "HeartBeatTag", "HeartBeatTag" }
                    }

var point_model = new Point()

{

Name = "HeartBeat", // table name

Fields = new Dictionary<string, object>() { { "HeartBeatValue", "OK" } },

Timestamp = DateTime.Now,

Tags = new Dictionary<string, object>() { { "HeartBeatTag", "HeartBeatTag" }

}

var whereFilter alanı ise bu veri üzerinde özel bir filtremiz var ise bunu çalıştırmak için kullanılıyor. Bizim InfluxDB’ye gönderdiğimiz HeartBeat verisi HeartBeatTag tagi ile işaretlenmiş durumda. Bu yüzden where filter’da bu tag’a sahip olan verileri alıyoruz.

var period alanı ise bu kuralın tetiklenme süresini belirtiyor. Bu TICKScript deadman olduğu için buradaki period alanı X süre ile veri alamamışsak anlamında kullanılıyor.

Scriptimizin 2.Kısmı

var name = 'DeadManAlert'

var idVar = name

var message = '"Bir şey oldu"'

var idTag = 'alertID'

var levelTag = 'level'

var messageField = 'message'

var durationField = 'duration'

var outputDB = 'chronograf'

var outputRP = 'autogen'

var outputMeasurement = 'alerts'

var triggerType = 'deadman'

var threshold = 0.0

var name = 'DeadManAlert'

var idVar = name

var message = '"Bir şey oldu"'

var idTag = 'alertID'

var levelTag = 'level'

var messageField = 'message'

var durationField = 'duration'

var outputDB = 'chronograf'

var outputRP = 'autogen'

var outputMeasurement = 'alerts'

var triggerType = 'deadman'

var threshold = 0.0

name alanı bu script tetiklendiği zaman X Rule Triggered alanında X yerine gelecek olan alarm adı.

message alanı ise script tetiklendiğinde oluşturulacak olan mesaj. CPU kullanımı %40’ın üstüne çıktı , Sıcaklık 35 derecenin altına düştü gibi.

levelTAG ise tanımladığımız alarmın durumunu belirtiyor.

outputDB ise alarm’ın çıktısını yazacağımız veritabanını gösteriyor. Bu script Chronograf arayüzü ile oluşturulduğu için veritabanımız chronograf veritabanı. Bu sayede Chronograf arayüzünü kullanarak alarmlarımızın geçmişini görüntüleyebiliyoruz.

triggerType alanı ise bu alarmın nasıl çalışacağını anlatıyor. Bu örnek deadman olduğu için type deadman. Burada kullanılabilecek diğer typelar threshold , relative , deadman.

TICKScript’in 3.Kısmı

Daha önce tanımladığımız sabitleri kullanarak kuralımızın çalışma yapısını oluşturuyoruz.

var data = stream
    |from()
        .database(db)
        .retentionPolicy(rp)
        .measurement(measurement)
        .groupBy(groupBy)
        .where(whereFilter)

var data = stream

|from()

.database(db)

.retentionPolicy(rp)

.measurement(measurement)

.groupBy(groupBy)

.where(whereFilter)

Yukarıda görebileceğimiz gibi aslında stream objesi üzerinde bir sorgu çalıştırıyoruz. Burada stream yerine batch kullanarak belli DB’ye kayıt edilmiş veriler üzerinde de çalışabiliriz. Fakat biz bu alarmda DeadMan yaptığımız için stream üzerinden çalışmak bize anlık alarm oluşturma şansı veriyor.

Aşağıda ise TICKScript’in son kısmını görüyoruz. Yukarıda streamden gelen datayı sorguladıktan sonra |deadman ile bir deadman alarmı register ediyoruz. Tanımladığımız sabitleri kullanarak alarm’ın özelliklerini ayarlıyoruz.

.post satırı ile ise bu alarm tetiklendiği zaman external bir api’ye notification gönderiyoruz.

Trigger kısmında ise bu alarm tetiklendiği zaman (Windows formsdaki OnClick , OnDrag gibi düşünebiliriz burayı. OnTriggered gibi)

InfluxDBOut fonksiyonunu kullanarak alarm tetiklenme bilgisini Chronograf veritabanına yazıyoruz. eval fonksiyonları expressionları evaluate etmek için kullanılıyor.

var trigger = data
    |deadman(threshold, period)
        .message(message)
        .id(idVar)
        .idTag(idTag)
        .levelTag(levelTag)
        .messageField(messageField)
        .durationField(durationField)
        .post('http://localhost:48711/api/values')
        .header('Content-Type', 'application/json')

trigger
    |eval(lambda: "emitted")
        .as('value')
        .keep('value', messageField, durationField)
    |eval(lambda: float("value"))
        .as('value')
        .keep()
    |influxDBOut()
        .create()
        .database(outputDB)
        .retentionPolicy(outputRP)
        .measurement(outputMeasurement)
        .tag('alertName', name)
        .tag('triggerType', triggerType)

trigger
    |httpOut('output')

var trigger = data

|deadman(threshold, period)

.message(message)

.id(idVar)

.idTag(idTag)

.levelTag(levelTag)

.messageField(messageField)

.durationField(durationField)

.post('http://localhost:48711/api/values')

.header('Content-Type', 'application/json')

trigger

|eval(lambda: "emitted")

.as('value')

.keep('value', messageField, durationField)

|eval(lambda: float("value"))

.as('value')

.keep()

|influxDBOut()

.create()

.database(outputDB)

.retentionPolicy(outputRP)

.measurement(outputMeasurement)

.tag('alertName', name)

.tag('triggerType', triggerType)

trigger

|httpOut('output')

Yazımızın buraya kadar olan kısmında tek bir sensör (InfluxDB’deki measurementlere sensör demeyi tercih edeceğim çünkü IoT dünyasında her measurement bir sensör’e karşılık geliyor) kullanarak bir deadman notification oluşturduk.

Yazımızın bundan sonraki kısmında ise çoklu sensör kullanarak ve sadece TICKScript yazarak nasıl bir kural oluşturabileceğimizi anlatacağım.

Önce sanki bir odamız ve bu oda içinde 2 adet sensörümüz varmış gibi çalışan bir InfluxDB data writer uygulaması yazalım.

Bu uygulamamız için ismi ‘multiple_Sensor_rules’ olan bir veritabanı yarattık.

 while (true)
            {
                Console.WriteLine("InfluxWriter");

                var infuxUrl = "http://localhost:8086/";
                var infuxUser = "admin";
                var infuxPwd = "admin";

                var clientDb = new InfluxDbClient(infuxUrl, infuxUser, infuxPwd, InfluxDbVersion.Latest);


                var point_model = new Point()
                {
                    Name = "ToplantiOdasi1", // table name
                    Fields = new Dictionary<string, object>() { { "Sicaklik", 10 }, { "KlimaDurum", 0 } },
                    Timestamp = DateTime.Now,
                    Tags = new Dictionary<string, object>() { { "Toplanti_Odasi_1", "Toplanti_Odasi_1" }
                    }
                };

                var dbName = "multiple_sensor_rules";
                var response = clientDb.Client.WriteAsync(point_model, dbName).Result;

                Console.WriteLine("Sicaklik 10 , KlimaDurum 0 Sended"); // Tetiklemez
                Console.ReadLine();
                point_model.Fields.Clear();
                point_model.Fields = new Dictionary<string, object>() { { "Sicaklik", 26 }, { "KlimaDurum", 1 } };
                response = clientDb.Client.WriteAsync(point_model, dbName).Result;
                Console.WriteLine("Sicaklik 26 , KlimaDurum 1 Sended"); // Tetiklemez
               Console.ReadLine();


                point_model.Fields.Clear();
                point_model.Fields = new Dictionary<string, object>() { { "Sicaklik", 20 }, { "KlimaDurum", 1 } };
                response = clientDb.Client.WriteAsync(point_model, dbName).Result;
                Console.WriteLine("Sicaklik 20 , KlimaDurum 1 Sended"); // Tetiklemez
              Console.ReadLine();

                point_model.Fields.Clear();
                point_model.Fields = new Dictionary<string, object>() { { "Sicaklik", 24 }, { "KlimaDurum", 0 } };
                response = clientDb.Client.WriteAsync(point_model, dbName).Result;
                Console.WriteLine("Sicaklik 24 , KlimaDurum 0 Sended"); // Tetikler
                Console.ReadLine();

              // Thread.Sleep(3000);
            }

while (true)

{

Console.WriteLine("InfluxWriter");

var infuxUrl = "http://localhost:8086/";

var infuxUser = "admin";

var infuxPwd = "admin";

var clientDb = new InfluxDbClient(infuxUrl, infuxUser, infuxPwd, InfluxDbVersion.Latest);

var point_model = new Point()

{

Name = "ToplantiOdasi1", // table name

Fields = new Dictionary<string, object>() { { "Sicaklik", 10 }, { "KlimaDurum", 0 } },

Timestamp = DateTime.Now,

Tags = new Dictionary<string, object>() { { "Toplanti_Odasi_1", "Toplanti_Odasi_1" }

}

};

var dbName = "multiple_sensor_rules";

var response = clientDb.Client.WriteAsync(point_model, dbName).Result;

Console.WriteLine("Sicaklik 10 , KlimaDurum 0 Sended"); // Tetiklemez

Console.ReadLine();

point_model.Fields.Clear();

point_model.Fields = new Dictionary<string, object>() { { "Sicaklik", 26 }, { "KlimaDurum", 1 } };

response = clientDb.Client.WriteAsync(point_model, dbName).Result;

Console.WriteLine("Sicaklik 26 , KlimaDurum 1 Sended"); // Tetiklemez

Console.ReadLine();

point_model.Fields.Clear();

point_model.Fields = new Dictionary<string, object>() { { "Sicaklik", 20 }, { "KlimaDurum", 1 } };

response = clientDb.Client.WriteAsync(point_model, dbName).Result;

Console.WriteLine("Sicaklik 20 , KlimaDurum 1 Sended"); // Tetiklemez

Console.ReadLine();

point_model.Fields.Clear();

point_model.Fields = new Dictionary<string, object>() { { "Sicaklik", 24 }, { "KlimaDurum", 0 } };

response = clientDb.Client.WriteAsync(point_model, dbName).Result;

Console.WriteLine("Sicaklik 24 , KlimaDurum 0 Sended"); // Tetikler

Console.ReadLine();

// Thread.Sleep(3000);

}

Sonrasında ise TICKScript kısmına geçebiliriz.

TICKScript yazmak için yine Chronograf arayüzünden yararlanacağız.

http://localhost:8888/sources/1/alert-rules

1	http://localhost:8888/sources/1/alert-rules

Yukarıda vermiş olduğum ‘Chronograf-> Manage Tasks’ menüsüne geçiyoruz.

Burada Write TICKScript butonuna tıklayıp yeni bir script oluşturuyoruz.

Karşımıza bu şekilde bir editör ekranı çıkıyor.

Burada Scriptimiz için bir ID yazabiliriz , ya da otomatik oluşturulması için boş bırakabiliriz.

Ben ID olarak ‘ airconditioning_alarm_rule’ kullanacağım.

Sağ taraftan çalışacağımız veritabanını seçiyoruz.

Ve artık Scriptimizi yazmaya başlayabiliriz.

Aşağıdaki kodu incelediğimiz zaman triggerType alanının thresHold olarak değiştiğini , sicaklikCrit ve klimaCrit isimli yeni sabitler eklendiğini göreceğiz.

windowSize stream içerisinde bize bir pencere oluşturup aradığımız değerleri bulmamıza yardımcı oluyor.

sicaklikCrit ve klimaCrit ise izlemekte olduğumuz iki sensörün değerlerini karşılaştırmak için kullandığımız sabitler.

var db = 'multiple_sensor_rules'

var rp = 'autogen'

var measurement = 'ToplantiOdasi1'

var whereFilter = lambda: ('Toplanti_Odasi_1' == 'Toplanti_Odasi_1')

var groupBy = []

var name = 'AirConditioningAlarm'

var idVar = name

var message = 'Sıcaklık 23 Derecenin Üstünde ve Klima Çalışmıyor'

var idTag = 'alertID'

var levelTag = 'level'

var messageField = 'message'

var durationField = 'duration'

var outputDB = 'chronograf'

var outputRP = 'autogen'

var outputMeasurement = 'alerts'

var triggerType = 'threshold'

var sicaklikCrit = 23

var klimaCrit = 0

var db = 'multiple_sensor_rules'

var rp = 'autogen'

var measurement = 'ToplantiOdasi1'

var whereFilter = lambda: ('Toplanti_Odasi_1' == 'Toplanti_Odasi_1')

var groupBy = []

var name = 'AirConditioningAlarm'

var idVar = name

var message = 'Sıcaklık 23 Derecenin Üstünde ve Klima Çalışmıyor'

var idTag = 'alertID'

var levelTag = 'level'

var messageField = 'message'

var durationField = 'duration'

var outputDB = 'chronograf'

var outputRP = 'autogen'

var outputMeasurement = 'alerts'

var triggerType = 'threshold'

var sicaklikCrit = 23

var klimaCrit = 0

Şimdi ise stream üzerinden izlemek istediğimiz sensörlerin değerlerini alalım.

var sicaklikStreamWindow = stream
    |from()
        .database(db)
        .retentionPolicy(rp)
        .measurement(measurement)
        .where(whereFilter)
    |eval(lambda: "Sicaklik")
        .as('value')
    |log()

var klimaStreamWindow = stream
    |from()
        .database(db)
        .retentionPolicy(rp)
        .measurement(measurement)
        .where(whereFilter)
    |eval(lambda: "KlimaDurum")
        .as('value')
    |log()

var data = sicaklikStreamWindow
    |join(klimaStreamWindow)
        .as('Sicaklik', 'Klima')
    |log()

var sicaklikStreamWindow = stream

|from()

.database(db)

.retentionPolicy(rp)

.measurement(measurement)

.where(whereFilter)

|eval(lambda: "Sicaklik")

.as('value')

|log()

var klimaStreamWindow = stream

|from()

.database(db)

.retentionPolicy(rp)

.measurement(measurement)

.where(whereFilter)

|eval(lambda: "KlimaDurum")

.as('value')

|log()

var data = sicaklikStreamWindow

|join(klimaStreamWindow)

.as('Sicaklik', 'Klima')

|log()

InfluxDB’ye veri yazan uygulamamızda 2 adet measurement – 2 adet sensör yazmıştık. Bunların isimleri Sicaklik ve Klima_Durum idi. Yukarıdaki kod parçasında stream içerisinde 1 dakikalık windowlar halinde aralıklar oluşturulup bu iki veriyi streamden anlık olarak aldık.

Aldığımız verileri

var data = sicaklikStreamWindow
    |join(klimaStreamWindow)
        .as('Sicaklik', 'Klima')
    |log()

var data = sicaklikStreamWindow

|join(klimaStreamWindow)

.as('Sicaklik', 'Klima')

|log()

şeklinde join kullanarak birleştirdik. Bu noktada |join kullanarak n adet veriyi birleştirmek mümkün. Bu da bize N adet sensörü kullanan oldukça kompleks alarm kuralları oluşturma olanağı sağlıyor.

Bundan sonraki kısımda klasik TICKScript devam ediyor.

sadece en önemli olduğunu düşündüğüm kısım .crit satırı. Burada and , or , == , != ,>,< gibi logic operatörlerin tamamı kullanılabiliyor.

Aşağıdaki kuralı yorumlarsak

Sicaklik.value>sicaklikCrit(23) && Klima_Durum.Value == klimaCrit(OFF) ise bu kural tetiklenecek.

var trigger = data
    |alert()
        .crit(lambda: "Sicaklik.value" > sicaklikCrit AND "Klima.value" == klimaCrit)
        .message(message)
        .id(idVar)
        .idTag(idTag)
        .levelTag(levelTag)
        .messageField(messageField)
        .durationField(durationField)
        .post('http://localhost:48711/api/values')
        .header('Content-Type', 'application/json')
    |log()

trigger
    |eval(lambda: float("Sicaklik.value"))
        .as('Sicaklik.value')
        .keep()
    |eval(lambda: float("Klima.value"))
        .as('Klima.value')
        .keep()
    |influxDBOut()
        .create()
        .database(outputDB)
        .retentionPolicy(outputRP)
        .measurement(outputMeasurement)
        .tag('alertName', name)
        .tag('triggerType', triggerType)

trigger
    |httpOut('output')

var trigger = data

|alert()

.crit(lambda: "Sicaklik.value" > sicaklikCrit AND "Klima.value" == klimaCrit)

.message(message)

.id(idVar)

.idTag(idTag)

.levelTag(levelTag)

.messageField(messageField)

.durationField(durationField)

.post('http://localhost:48711/api/values')

.header('Content-Type', 'application/json')

|log()

trigger

|eval(lambda: float("Sicaklik.value"))

.as('Sicaklik.value')

.keep()

|eval(lambda: float("Klima.value"))

.as('Klima.value')

.keep()

|influxDBOut()

.create()

.database(outputDB)

.retentionPolicy(outputRP)

.measurement(outputMeasurement)

.tag('alertName', name)

.tag('triggerType', triggerType)

trigger

|httpOut('output')

Tüm işimiz bittikten sonra kuralımızı save edip TICKScript editörü kapatabiliriz.

Save işleminden sonra Editörün alt kısmında TICKScript is Valid uyarısını görüp yazdığımız scriptte her hangi bir syntax Error olmadığından emin olmalıyız.

Bir önceki yazımda nasıl yapılacağını anlatmış olduğum NotificationHandler uygulamamızı başlatıyoruz. Alarm sonucu oluşan notificationları görebilmek için.

InfluxDB Writer Uygulamamız sırası ile her enter’a basışımızda ,

Sicaklik 10 , Klima_Durum OFF Sended // Tetiklemez Klima Off(0) Sıcaklık 23’den Küçük

Sicaklik 26 , Klima_Durum ON Sended // Tetiklemez Klima On(1) Sıcaklık 23’den Büyük

Sicaklik 20 , Klima_Durum ON Sended // Tetiklemez Klima ON(1) Sıcaklık 23’den Küçük

Sicaklik 24, Klima_Durum OFF Sended // Tetikler Klima OFF(0) Sıcaklık 23’den Büyük

datalarını gönderecek. Influx Writer uygulamamızı çalıştırıp her şeyin yolunda olduğundan emin olalım.

Sonrasında ise handler uygulamamızı takip edelim.

Yukarıdaki görüntüde görebileceğimiz gibi kuralımız tetiklendi. Tabii ki burada bitirmiyoruz ve DataWriter uygulamamızda bir kere daha Enter tuşuna basıp kuralı tetiklemeyen bir değerin gönderilmesini sağlıyoruz.

Gönderdiğimiz değerler kuralı sağlamamasına rağmen neden yeniden POST requesti atıldı ? Cevabı dto.Level ve dto.PreviousLevel alanlarında saklı.

İlk gönderdiğimiz 24,0 değeleri kuralı sağladı ve Kuralı OK durumundan CRITICAL durumuna çekti. Kuralımız hala CRITICAL.

Sonrasında kuralı sağlamayan 10,0 değerini gönderdiğimiz zaman ise bu değerler Kuralımızı CRITICAL durumundan OK durumuna çekti. Böyle kuralımızın değişim geçmişini de tutabiliyoruz.

Bu yazımda da sizlere TICKScript kullanarak çoklu değerler üzerinden çalışan Kurallar tanımlamayı anlatmaya çalıştım.

Sürç-ü lisan ettiysem affola.

Post Views: 3.147

InfluxDB & Kapacitor & Chronograf Kullanarak Çoklu Koşul Bildirimleri Oluşturmak

TICKScript Nedir ?

You may also like...

Bir yanıt yazın Yanıtı iptal et

InfluxDB & Kapacitor & Chronograf Kullanarak Çoklu Koşul Bildirimleri Oluşturmak

TICKScript Nedir ?

You may also like...

Arduino ile MQTT Haberleşmesi

InfluxDB & Kapacitor & Chronograf Kullanarak DeadMan Bildirimleri Oluşturmak

Arduino MQTT & InfluxDB & Grafana Kullanarak Uzaktan İzleme Sistemi

Bir yanıt yazın Yanıtı iptal et