TimescaleDB'deki verileri okuyup alarm yapısına dönüştürmek

Alarm üretim mantığı genelde manuel yazılıyor. Bu ya PostgreSQL tarafında bir trigger + NOTIFY akışıyla başlatılıyor, ya da uygulama tarafında çalışan bir servis ile okuyup dönüştürülüyor.

Elimizde TimescaleDB üzerinde bir hypertable olduğunu düşünelim:

• sensör verisi geliyor
• bazı kayıtlar alarm üretmeli
• bazı alarmlar tek satırdan çıkıyor, bazıları ise pencere bazlı hesap gerektiriyor, yani son 5 dakikanın ortalaması %80'i geçti mi? gibi

TimescaleDB bu tip akışlar için ne kadar uygun?

TimescaleDB, PostgreSQL'in üzerinde çalıştığı için PostgreSQL'in trigger mekanizmasını destekliyor. Resmi dokümana göre hypertable üzerinde oluşturulan trigger'lar alttaki chunk'lara da yayılıyor.

Ayrıca TimescaleDB'de job mekanizması var ve continuous aggregate desteği de var. Yani alarmı doğrudan her satır insert edildiğinde üretmek zorunda değiliz. Özellikle pencere bazlı kurallarda önce aggregate üretmek, sonra alarm çıkarmak daha mantıklı.

Yaklaşım 1: PostgreSQL LISTEN/NOTIFY ile alarm akışı

Bu modelin mantığı basit:

1. TimescaleDB hypertable'ına veri gelir.
2. 'AFTER INSERT' trigger çalışır.
3. Trigger, uygun gördüğü kayıt için pg_notify(...) çağırır.
4. Uygulamadaki bir listener servis ilgili kanalı dinler.
5. Listener, payload'daki anahtar bilgiyle gerçek kaydı okur.
6. Alarm nesnesini oluşturur ve alarms tablosuna yazar.

Artıları nelerdir?

Öncelikle gecikme düşük. NOTIFY, commit sonrası dinleyicilere sinyal gönderir. Yani veri başarılı şekilde commit olduktan hemen sonra uygulaman uyanabilir. Ayrı bir queue ürünü kurmadan, veritabanının içinden olay sinyali alabiliyorsun.

Dikkat edilmesi gereken noktalar:

PostgreSQL dokümanına göre:

• bildirimler transaction commit edilmeden teslim edilmiyor
• dinleyici transaction içindeyse mesaj ona da transaction bitince teslim ediliyor
• aynı transaction içinde aynı kanal ve aynı payload tekrar tekrar gönderilirse tek bildirime indirgenebiliyor
• payload varsayılan yapılandırmada 8000 bayttan kısa olmak zorunda
• queue dolarsa NOTIFY çağıran transaction commit anında hata alabiliyor

Yani kısaca NOTIFY veri taşımıyor. Sadece olay sinyali veriyor.

NOTIFY için basit örnek

CREATE TABLE sensor_events (
  event_id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  ts timestamptz NOT NULL,
  device_id text NOT NULL,
  metric text NOT NULL,
  value double precision NOT NULL
);

SELECT create_hypertable('sensor_events', by_range('ts'));

CREATE OR REPLACE FUNCTION notify_alarm_candidate()
RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
  IF NEW.metric = 'temperature' AND NEW.value >= 80 THEN
    PERFORM pg_notify('alarm_candidates', NEW.event_id::text);
  END IF;

  RETURN NEW;
END;
$$;

CREATE TRIGGER trg_notify_alarm_candidate
AFTER INSERT ON sensor_events
FOR EACH ROW
EXECUTE FUNCTION notify_alarm_candidate();

Bu örnekte trigger sadece bu kayıt alarm adayı olabilir diyor. Alarmın gerçekten açılıp açılmayacağına uygulama karar veriyor.

Yaklaşım 2: Yazılımla okumak ve alarma dönüştürmek

1. Uygulama tarafında bir worker veya servis çalışır.
2. Bu servis TimescaleDB'den yeni verileri belli bir mantıkla okur.
3. Okuduğu veriyi alarm modeline dönüştürür.

Bu modelde raw hypertable ayrı kalır, alarm işleme için ayrı bir alarm_candidates tablosu tutulur.

Uygulama worker'ları bu tablodan kayıt çekip işler. Çoklu worker varsa PostgreSQL'in FOR UPDATE SKIP LOCKED özelliği ile çakışmadan paralel tüketim yapılabilir. PostgreSQL dokümanı da SKIP LOCKED kullanımının queue benzeri yapılarda lock çakışmasını azaltmak için uygun olduğunu söylüyor.

Örnek:

WITH picked AS (
  SELECT id
  FROM alarm_outbox
  WHERE status = 'new'
  ORDER BY id
  FOR UPDATE SKIP LOCKED
  LIMIT 100
)
UPDATE alarm_outbox o
SET status = 'processing',
    picked_at = now()
FROM picked
WHERE o.id = picked.id
RETURNING o.*;

Sonuç

NOTIFY, iyi bir tetikleme sinyali, uygulama worker'ı ise daha iyi bir işleme motoru.