NTP Time Sync Monitor

Problem Statement

Clock skew across distributed systems is a silent operational hazard. When local server clocks drift out of synchronization, microservices fail to correlate logs, distributed databases suffer split-brain scenarios or data corruption during timestamps-based conflicts, and security protocols like OAuth 2.0 or Kerberos reject authentication tokens due to time-window mismatches.

While background daemons like chronyd or systemd-timesyncd maintain local time, they can silently freeze, get blocked by firewall rules, or suffer from misconfiguration without raising immediate system alerts. This NTP Time Sync Monitor independently measures your system clock drift against target NTP servers, reports latency metrics, and exits with non-zero status codes when drift exceeds tolerable thresholds—making it ideal for cron jobs, monitoring agents, or container readiness probes.

How It Works

The script queries one or more NTP (Network Time Protocol) servers using the Network Time Protocol library (ntplib). For each target server, it sends standard SNTP UDP packets (port 123) and calculates:

1. Clock Offset: The time difference in milliseconds between the local system clock and the remote reference clock.
2. Round-Trip Delay (RTT): Network latency encountered during the query.
3. Stratum: The hierarchy level of the target NTP server.

The script evaluates the calculated offsets against a user-configured threshold (--alert-offset-ms). Results can be emitted in human-readable tabular text or structured JSON for log parsing pipelines. If any queried NTP server reports a time drift exceeding the threshold, or if all requested NTP servers fail to respond, the script exits with status code 1, signalling an alert condition to upstream orchestration tools.

Full Script

#!/usr/bin/env python3
"""
NTP Time Sync Monitor

Queries upstream NTP servers, checks local system clock offset, 
and alerts if time drift exceeds specified thresholds.
"""

import argparse
from dataclasses import asdict, dataclass
import json
import os
import sys
from typing import List, Optional

try:
    import ntplib
except ImportError:
    sys.exit(
        "Error: 'ntplib' dependency missing. Install it using 'pip install ntplib'."
    )


@dataclass
class NTPCheckResult:
    server: str
    stratum: int
    offset_ms: float
    delay_ms: float
    status: str  # "OK", "OFFSET_EXCEEDED", "UNREACHABLE"
    error: Optional[str] = None


def query_ntp_server(
    server: str, timeout: float, alert_offset_ms: float
) -> NTPCheckResult:
    """Queries a single NTP server and calculates time offset and delay."""
    client = ntplib.NTPClient()
    try:
        # Request NTP packet (version 3 is widely supported)
        response = client.request(server, version=3, timeout=timeout)
        
        # Convert offset and delay from seconds to milliseconds
        offset_ms = response.offset * 1000.0
        delay_ms = response.delay * 1000.0
        stratum = response.stratum

        # Determine status based on drift threshold
        status = "OFFSET_EXCEEDED" if abs(offset_ms) > alert_offset_ms else "OK"

        return NTPCheckResult(
            server=server,
            stratum=stratum,
            offset_ms=round(offset_ms, 3),
            delay_ms=round(delay_ms, 3),
            status=status,
        )
    except (ntplib.NTPException, Exception) as err:
        return NTPCheckResult(
            server=server,
            stratum=0,
            offset_ms=0.0,
            delay_ms=0.0,
            status="UNREACHABLE",
            error=str(err),
        )


def format_text_output(results: List[NTPCheckResult], alert_offset_ms: float) -> str:
    """Formats NTP check results into a human-readable table."""
    lines = [
        f"NTP Time Sync Status (Threshold: ±{alert_offset_ms} ms)",
        "=" * 70,
        f"{'SERVER':<25} {'STATUS':<17} {'OFFSET (ms)':<12} {'DELAY (ms)':<10} {'STRATUM'}",
        "-" * 70,
    ]
    
    for r in results:
        if r.status == "UNREACHABLE":
            lines.append(
                f"{r.server:<25} {r.status:<17} {'N/A':<12} {'N/A':<10} N/A ({r.error})"
            )
        else:
            lines.append(
                f"{r.server:<25} {r.status:<17} {r.offset_ms:<12.3f} {r.delay_ms:<10.3f} {r.stratum}"
            )
            
    lines.append("=" * 70)
    return "\n".join(lines)


def parse_args() -> argparse.Namespace:
    """Parses command-line arguments and merges defaults from environment variables."""
    default_servers = os.environ.get(
        "NTP_SERVERS", "pool.ntp.org,time.google.com,time.cloudflare.com"
    )
    default_threshold = float(os.environ.get("NTP_ALERT_OFFSET_MS", "100.0"))
    default_timeout = float(os.environ.get("NTP_TIMEOUT", "5.0"))

    parser = argparse.ArgumentParser(
        description="Monitor system clock drift against upstream NTP servers."
    )
    parser.add_argument(
        "--servers",
        type=str,
        default=default_servers,
        help=f"Comma-separated list of NTP servers (default: {default_servers})",
    )
    parser.add_argument(
        "--alert-offset-ms",
        type=float,
        default=default_threshold,
        help=f"Max acceptable offset in milliseconds before alerting (default: {default_threshold})",
    )
    parser.add_argument(
        "--timeout",
        type=float,
        default=default_timeout,
        help=f"Socket timeout in seconds per server query (default: {default_timeout})",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="Output check results in JSON format",
    )

    return parser.parse_args()


def main() -> None:
    args = parse_args()

    # Split server comma-separated list into cleaned string list
    server_list = [s.strip() for s in args.servers.split(",") if s.strip()]

    if not server_list:
        print("Error: No NTP servers specified.", file=sys.stderr)
        sys.exit(2)

    results: List[NTPCheckResult] = []
    has_alert = False
    reachable_count = 0

    for server in server_list:
        res = query_ntp_server(server, args.timeout, args.alert_offset_ms)
        results.append(res)

        if res.status != "UNREACHABLE":
            reachable_count += 1
        if res.status == "OFFSET_EXCEEDED":
            has_alert = True

    # Alert condition if all servers failed or if threshold exceeded
    all_failed = reachable_count == 0
    if all_failed:
        has_alert = True

    # Output generation
    if args.json:
        output_data = {
            "summary": {
                "total_queried": len(server_list),
                "reachable": reachable_count,
                "threshold_ms": args.alert_offset_ms,
                "alert": has_alert,
            },
            "results": [asdict(r) for r in results],
        }
        print(json.dumps(output_data, indent=2))
    else:
        print(format_text_output(results, args.alert_offset_ms))
        if all_failed:
            print("ALERT: All NTP servers were unreachable!")
        elif has_alert:
            print("ALERT: Local clock drift exceeds threshold on one or more servers!")

    # Non-zero exit code indicates unhealthy sync status or network failure
    if has_alert:
        sys.exit(1)
    
    sys.exit(0)


if __name__ == "__main__":
    main()

Configuration

The script supports configuration via command-line arguments or environment variables. Command-line flags take precedence over environment variables.

Usage Examples

1. Basic Check with Default Settings

Run the script using target servers and standard threshold defaults:

python3 ntp_monitor.py

Output:

NTP Time Sync Status (Threshold: ±100.0 ms)
======================================================================
SERVER                    STATUS            OFFSET (ms)  DELAY (ms)  STRATUM
----------------------------------------------------------------------
pool.ntp.org              OK                1.234        24.102      2
time.google.com           OK                0.811        18.450      1
time.cloudflare.com       OK                -0.412       12.301      1
======================================================================

2. Strict Threshold Check on Custom Servers

Target specific upstream servers with a strict threshold of 15 milliseconds:

python3 ntp_monitor.py --servers pool.ntp.org,time.apple.com --alert-offset-ms 15

Output:

NTP Time Sync Status (Threshold: ±15.0 ms)
======================================================================
SERVER                    STATUS            OFFSET (ms)  DELAY (ms)  STRATUM
----------------------------------------------------------------------
pool.ntp.org              OK                2.110        22.501      2
time.apple.com            OFFSET_EXCEEDED   18.452       45.120      1
======================================================================
ALERT: Local clock drift exceeds threshold on one or more servers!

(Exit code: 1)

3. JSON Output for Automated Ingestion

Use the --json flag to emit machine-readable metrics for vector agents, Datadog, or custom alerting scripts:

python3 ntp_monitor.py --servers time.google.com --json

Output:

{
  "summary": {
    "total_queried": 1,
    "reachable": 1,
    "threshold_ms": 100.0,
    "alert": false
  },
  "results": [
    {
      "server": "time.google.com",
      "stratum": 1,
      "offset_ms": 0.412,
      "delay_ms": 15.201,
      "status": "OK",
      "error": null
    }
  ]
}

Error Handling

• Missing Dependency: Checks for ntplib upon launch and exits with actionable installation steps if absent.
• Network Timeouts & UDP Blockade: If outbound UDP Port 123 is blocked by network firewalls or security groups, ntplib.NTPException is caught gracefully per host, marking that server as UNREACHABLE without crashing execution.
• DNS Resolution Failure: Handles domain lookup errors on individual hosts, logging the exception string into the result payload.
• Total Failure Handling: If all targeted servers fail to respond, the execution flags has_alert = True and exits with code 1 to avoid false-positive success reporting.
• Invalid Input Safety: Verifies that parsed server lists are non-empty after string trimming.

Extending Ideas

1. Prometheus Metrics Exporter: Wrap the execution logic inside a http.server module to expose standard /metrics endpoints for Prometheus scraping (e.g., node_ntp_offset_milliseconds).
2. Automated Clock Correction: Integrate an optional --auto-fix flag that executes /usr/sbin/chronyc step or systemctl restart systemd-timesyncd when severe drift (> 1000ms) is observed.
3. Outlier Filtering Algorithm: Implement Marzullo's algorithm or intersection algorithms to filter out individual faulty NTP sources ("falsetickers") from skew calculations.
4. Webhook Notifications: Add immediate Webhook dispatching (Slack, PagerDuty, Teams) when critical offset breaches occur.