Telemetry — AI Observability Worker

Overview

AIObservabilityWorker is the central monitoring service that ties together the metrics registry, alert rules, anomaly detection, LLM incident summarization, and notification delivery. It runs a periodic evaluation loop that checks all registered alert rules against current metric data and dispatches notifications when alerts fire.

The worker is designed to be wired up once at application startup, alongside the SDK initialization. It is independent of the framework's HTTP request lifecycle — it runs on a timer and does not block incoming requests.

Key Concepts

• Evaluation loop — runs every intervalMs milliseconds (default: 30 seconds). On each tick it reads the current metric snapshot and evaluates every registered alert.
• Threshold evaluation — compares the metric's lastValue against the alert's threshold using the configured operator.
• Anomaly evaluation — feeds the metric value into the AnomalyDetector and fires if isAnomaly === true.
• Insight generation — before dispatching, the worker calls the IncidentSummarizer to produce a human-readable IncidentInsight (summary + recommendation + confidence score).
• Notification dispatch — the completed AlertNotification (with the insight attached) is sent to all configured AlertNotifier instances.
• metricsProvider — by default the worker calls getMetricsRegistry().snapshot(). Supply a custom provider for testing or when reading metrics from an external source.

API Reference

AIObservabilityWorker

import { AIObservabilityWorker } from '@tagna/udiot/telemetry';

const worker = new AIObservabilityWorker(options?);

Constructor: AIObservabilityWorkerOptions

.start(): void

Begins the periodic evaluation loop. Calling .start() on an already-started worker is a no-op — the timer is not restarted.

.stop(): void

Stops the evaluation loop by clearing the internal interval. The worker can be restarted by calling .start() again.

.evaluate(): Promise<void>

Runs one evaluation cycle manually. Useful in tests to trigger evaluation on demand without waiting for the timer.

AlertNotification

The object dispatched to onAlert and all notifiers when an alert fires:

Evaluation loop (step by step)

Every intervalMs:
  1. Get MetricSnapshot
       └─ metricsProvider?.() or getMetricsRegistry().snapshot()

  2. Get all alert rules
       └─ listAIAlerts()

  3. For each alert rule:
     a. Resolve metricName (from alert.options.metric or parsed condition)
     b. Look up metricValue in snapshot (uses lastValue ?? sum ?? count)
     c. Resolve detection mode (threshold or anomaly)

     If threshold mode:
       d. Parse operator and threshold
       e. compare(metricValue, threshold, operator)
       f. If true → proceed to step 4

     If anomaly mode:
       d. anomalyDetector.detect(metricName, metricValue, windowSize/minSamples/sensitivity)
       e. If result.isAnomaly → proceed to step 4

  4. Build AlertNotification:
       a. Call summarizer.summarize(IncidentContext) → IncidentInsight
       b. Attach insight to notification

  5. Dispatch:
       a. Call onAlert(notification) (if provided)
       b. Call notifier.dispatch(notification) for each registered notifier

Architecture diagram

                  ┌─────────────────────────────────┐
                  │       AIObservabilityWorker      │
                  │  (timer: every 30 s by default)  │
                  └────────────┬────────────────────┘
                               │ evaluate()
               ┌───────────────┼───────────────────┐
               ▼               ▼                   ▼
    MetricsRegistry       listAIAlerts()     AnomalyDetector
    .snapshot()           (alert rules)      .detect(metric, value)
               │               │                   │
               └───────────────┴───────────────────┘
                               │ alert triggered
                               ▼
                     IncidentSummarizer
                     .summarize(context)
                               │
                               ▼
                      AlertNotification
                    (with IncidentInsight)
                               │
               ┌───────────────┼──────────────┐
               ▼               ▼              ▼
           onAlert()    ConsoleNotifier  SlackNotifier  ...

Basic Usage

Wiring the worker at application startup

import {
  AIObservabilityWorker,
  registerAIAlert,
  LLMIncidentSummarizer,
  RuleBasedIncidentSummarizer,
  initializeTelemetry,
  OTLPMetricsExporter,
  shutdownTelemetry,
} from '@tagna/udiot/telemetry';

// 1. Initialize SDK
initializeTelemetry({
  metrics: {
    exporters: [new OTLPMetricsExporter()],
    flushIntervalMs: 60_000,
  },
});

// 2. Register alert rules
registerAIAlert({
  target: 'ApiGateway',
  methodName: 'handleRequest',
  options: {
    name: 'High error rate',
    metric: 'http.error_rate',
    mode: 'threshold',
    operator: '>',
    threshold: 5,
    severity: 'critical',
    playbookUrl: 'https://wiki.internal/runbooks/errors',
  },
});

// 3. Set up summarizer (LLM with rule-based fallback)
const summarizer = new LLMIncidentSummarizer({
  apiKey: process.env.OPENAI_API_KEY,
  model: 'gpt-4o-mini',
  fallback: new RuleBasedIncidentSummarizer(),
});

// 4. Create and start the worker
const worker = new AIObservabilityWorker({
  intervalMs: 30_000,
  // Deliver through onAlert using an application-owned notification integration.
  summarizer,
  onAlert: (notification) => {
    console.log('Alert fired:', notification.alert.options.name, notification.insight?.summary);
  },
});

worker.start();

// 5. Shut down gracefully
process.on('SIGTERM', async () => {
  worker.stop();
  await shutdownTelemetry();
  process.exit(0);
});

Testing alert evaluation

import {
  AIObservabilityWorker,
  registerAIAlert,
  resetAIAlerts,
  AlertNotification,
} from '@tagna/udiot/telemetry';

afterEach(() => resetAIAlerts());

test('fires when error rate exceeds threshold', async () => {
  registerAIAlert({
    target: 'TestTarget',
    methodName: 'testMethod',
    options: {
      metric: 'test.error_rate',
      mode: 'threshold',
      operator: '>',
      threshold: 10,
      severity: 'warning',
    },
  });

  const fired: AlertNotification[] = [];
  const worker = new AIObservabilityWorker({
    metricsProvider: () => [
      {
        name: 'test.error_rate',
        type: 'gauge',
        sum: 15,
        count: 1,
        min: 15,
        max: 15,
        lastValue: 15,
        updatedAt: Date.now(),
      },
    ],
    onAlert: (n) => { fired.push(n); },
  });

  await worker.evaluate();

  expect(fired).toHaveLength(1);
  expect(fired[0].metricValue).toBe(15);
  expect(fired[0].triggerType).toBe('threshold');
});

Common Patterns

Multiple notifiers — pass an array of notifiers to route the same alert to several channels simultaneously (Slack, webhook, console).

Severity routing — create the worker with a single onAlert callback that selects notifiers dynamically based on notification.alert.options.severity.

Custom metric source — use metricsProvider to read metrics from Redis, a database, or a mock during testing, without having to populate the global MetricsRegistry.

Pitfalls

• The evaluation timer is unreffed (timer.unref?.()) so it does not prevent Node.js from exiting if no other work is pending. Call worker.stop() explicitly in your shutdown handler to be safe.
• Notifier failures are caught and logged as warnings. A failing Slack webhook does not prevent other notifiers from being called.
• The worker reads listAIAlerts() on every tick. If you register many rules, each tick performs a linear scan of all rules. For typical production use (tens of rules) this is negligible.
• If the same metric name is recorded with different attributes, the registry holds multiple aggregates. The worker finds the metric by name without considering attributes — it returns the first match. Use unique metric names if attribute-based distinction is needed in alerts.