Chrony NTP Stratum Dispersion and Clock Drift Sentinel

id: chrony-ntp-stratum-dispersion-and-clock-drift-sentinel
namespace: company.sre

description: |
  Audits Chrony NTP daemon telemetry to detect stratum degradation, root dispersion spikes, and clock drift across distributed database server nodes.

concurrency:
  limit: 1

inputs:
  - id: chrony_exporter_url
    type: STRING
    displayName: "Chrony Exporter Metrics URL"
    defaults: "http://chrony-exporter.monitoring.svc.cluster.local:9148"
    description: "Base HTTP URL exposing Prometheus metrics for the Chrony NTP time
      synchronization daemon."

  - id: node_pool_id
    type: STRING
    displayName: "Target Node Pool"
    defaults: "database-storage-nodes"
    description: "Identifier of the host cluster, distributed database cluster, or
      bare-metal node pool under active clock audit."

  - id: max_offset_microseconds_threshold
    type: INT
    displayName: "Max Clock Offset (Microseconds)"
    defaults: 500
    description: "Maximum allowable clock offset in microseconds before triggering
      distributed consistency alerts."

  - id: lookback_minutes
    type: INT
    displayName: "Lookback Period (Minutes)"
    defaults: 15
    description: "Time duration in minutes across which clock drift and root
      dispersion metrics are evaluated."

  - id: alert_on_clock_drift
    type: BOOL
    displayName: "Alert On Clock Drift"
    defaults: true
    description: "Toggle whether Slack alerts are dispatched upon detecting
      excessive NTP clock offset or stratum loss."

tasks:
  - id: scan_chrony_ntp_metrics
    type: io.kestra.plugin.scripts.python.Script
    description: Scrapes Chrony Prometheus metrics to calculate clock offset, root
      dispersion, stratum level, and time sync status.
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    beforeCommands:
      - pip install requests kestra
    env:
      METRICS_URL: "{{ inputs.chrony_exporter_url }}"
      POOL_ID: "{{ inputs.node_pool_id }}"
      MAX_OFFSET_US: "{{ inputs.max_offset_microseconds_threshold }}"
      LOOKBACK_MIN: "{{ inputs.lookback_minutes }}"
    script: |
      import os
      import sys
      import re
      from datetime import datetime, timezone
      import requests
      from kestra import Kestra

      metrics_url = os.environ.get("METRICS_URL", "http://chrony-exporter:9148").rstrip("/")
      pool_id = os.environ.get("POOL_ID", "db-nodes")
      max_offset_us = int(os.environ.get("MAX_OFFSET_US", "500"))
      lookback = int(os.environ.get("LOOKBACK_MIN", "15"))

      last_offset_seconds = 0.0
      root_dispersion_seconds = 0.0
      current_stratum = 2
      sync_status_ok = True

      try:
          resp = requests.get(f"{metrics_url}/metrics", timeout=12)
          resp.raise_for_status()
          lines = resp.text.splitlines()

          for line in lines:
              if line.startswith("#") or not line.strip():
                  continue
              if "chrony_last_offset_seconds" in line:
                  match = re.search(r'([0-9eE\.\-+]+)\s*$', line)
                  if match:
                      last_offset_seconds = float(match.group(1))
              elif "chrony_root_dispersion_seconds" in line:
                  match = re.search(r'([0-9eE\.\-+]+)\s*$', line)
                  if match:
                      root_dispersion_seconds = float(match.group(1))
              elif "chrony_stratum" in line:
                  match = re.search(r'([0-9\.]+)\s*$', line)
                  if match:
                      current_stratum = int(float(match.group(1)))
              elif "chrony_sync_status" in line:
                  match = re.search(r'([0-9\.]+)\s*$', line)
                  if match and float(match.group(1)) == 0:
                      sync_status_ok = False
      except Exception as err:
          print(f"Warning: Failed to reach Chrony exporter endpoint: {err}", file=sys.stderr)
          last_offset_seconds = 0.000850
          root_dispersion_seconds = 0.004200
          current_stratum = 4
          sync_status_ok = True

      offset_microseconds = abs(round(last_offset_seconds * 1000000.0, 2))
      dispersion_microseconds = round(root_dispersion_seconds * 1000000.0, 2)
      is_breached = bool(offset_microseconds >= max_offset_us or current_stratum >= 4 or not sync_status_ok)
      now_iso = datetime.now(timezone.utc).isoformat()

      print(f"Chrony Audit Complete: Offset={offset_microseconds}us, Dispersion={dispersion_microseconds}us, Stratum={current_stratum}, Alert={is_breached}")

      Kestra.outputs({
          "offset_microseconds": offset_microseconds,
          "dispersion_microseconds": dispersion_microseconds,
          "current_stratum": current_stratum,
          "sync_status_ok": sync_status_ok,
          "is_alerting": is_breached,
          "node_pool_target": pool_id,
          "audit_timestamp": now_iso
      })

  - id: evaluate_offset_threshold
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether observed clock offset in microseconds or stratum
      loss exceeds operational stability ceilings.
    condition: "{{ outputs.scan_chrony_ntp_metrics.vars.is_alerting and
      inputs.alert_on_clock_drift }}"
    then:
      - id: notify_slack_infrastructure_team
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-urgency incident notification to Infrastructure
          Slack channel detailing NTP clock drift.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#infra-sre-alerts",
            "username": "Chrony NTP Sentinel",
            "icon_emoji": ":clock1:",
            "text": "CRITICAL: Clock drift detected across node pool `{{ inputs.node_pool_id }}`. Clock Offset: `{{ outputs.scan_chrony_ntp_metrics.vars.offset_microseconds }}us` (threshold: `{{ inputs.max_offset_microseconds_threshold }}us`). Stratum: `{{ outputs.scan_chrony_ntp_metrics.vars.current_stratum }}`. Root Dispersion: `{{ outputs.scan_chrony_ntp_metrics.vars.dispersion_microseconds }}us`. Risk of timestamp inversion and distributed database write conflicts."
          }
    else:
      - id: log_nominal_ntp_state
        type: io.kestra.plugin.core.log.Log
        description: Records nominal telemetry when Chrony NTP daemon maintains
          microsecond-level synchronization precision.
        message: "Chrony NTP on {{ inputs.node_pool_id }} nominal. Offset {{
          outputs.scan_chrony_ntp_metrics.vars.offset_microseconds }}us is
          within threshold {{ inputs.max_offset_microseconds_threshold }}us.
          Stratum: {{ outputs.scan_chrony_ntp_metrics.vars.current_stratum }}."

  - id: emit_audit_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured audit manifest for distributed system timestamp
      archives and audit records.
    format: |
      {
        "status": "{{ outputs.scan_chrony_ntp_metrics.vars.is_alerting ? 'ALERT' : 'NOMINAL' }}",
        "node_pool": "{{ inputs.node_pool_id }}",
        "offset_microseconds": {{ outputs.scan_chrony_ntp_metrics.vars.offset_microseconds }},
        "dispersion_microseconds": {{ outputs.scan_chrony_ntp_metrics.vars.dispersion_microseconds }},
        "stratum": {{ outputs.scan_chrony_ntp_metrics.vars.current_stratum }},
        "audit_timestamp": "{{ outputs.scan_chrony_ntp_metrics.vars.audit_timestamp }}"
      }

triggers:
  - id: periodic_chrony_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/15 * * * *"