Network Time & Clock Synchronization Telemetry with ezsnmp

Note

  • Published: November 12, 2026

  • Authors: Written by Gemini 3.8 Flash, Co-Authored & Reviewed by Carlos Santos (@carlkidcrypto)

  • Estimated Reading Time: 5 minutes

  • Tags: Python • Snmp • Ntp • Ptp • Clock-Sync • Ezsnmp • Network-Monitoring

Why Network Time Telemetry Is Essential

Accurate, highly synchronized system clocks are an unforgiving requirement of modern distributed computing, network security, and financial infrastructure:

  • Log Correlation & Incident Forensics: Security Information and Event Management (SIEM) systems rely on sub-millisecond timestamp alignment across firewalls, routers, and host kernels to reconstruct attack sequences.

  • Consensus & Distributed Databases: Platforms relying on consensus leases (such as CockroachDB, Google Spanner, or Raft/Paxos clusters) fail or halt writes when clock skew exceeds strict thresholds.

  • Cryptographic Protocol Integrity: TLS certificates, Kerberos tickets, and JSON Web Tokens (JWTs) fail validation when endpoint clocks drift.

  • Regulatory Compliance: Financial exchange mandates (such as FINRA Rule 4590 and MiFID II) require sub-millisecond clock traceability against UTC.

Despite its importance, time synchronization is prone to silent failure: upstream Stratum-1 reference servers becoming unreachable, local oscillators drifting unchecked, or transit link congestion driving up packet jitter.

ezsnmp provides native Net-SNMP C bindings to query standard time MIBs over SNMP, enabling network engineers to audit clock synchronization states, stratum levels, and offsets across thousands of nodes in seconds.

The Architecture of NTP-MIB (RFC 5907)

The IETF standard NTP-MIB structures clock telemetry into chassis-level synchronization status and upstream peer association tables:

1. System Synchronization State (ntpEntStatus)

Under ntpEntStatus (OID .1.3.6.1.2.1.197.1.2):

  • ``ntpEntStatusCurrentMode``: Current operational synchronization state: - notRunning(1): NTP service inactive. - notSynchronized(2): Active, but not locked to an upstream source. - syncToLocal(4): Synchronized to local hardware clock (uncalibrated). - syncToRefclock(5): Synchronized to direct reference clock (GPS/PPS). - syncToRemoteServer(6): Synchronized to upstream network NTP server.

  • ``ntpEntStatusStratum``: Distance from the primary reference source (1 = primary atomic/GPS clock, 2-15 = secondary servers, 16 = unsynchronized).

  • ``ntpEntStatusActiveOffset``: Current clock offset relative to active reference peer (signed numeric value).

  • ``ntpEntStatusDispersion``: Root dispersion representing estimated maximum clock error margin.

2. Peer Association Table (ntpAssociationTable)

Under ntpAssociationTable (OID .1.3.6.1.2.1.197.1.3.1):

  • ``ntpAssocOffset``: Offset relative to this specific peer.

  • ``ntpAssocStatusDelay``: Round-trip network transit delay.

  • ``ntpAssocStatusJitter``: Measured dispersion/jitter of UDP 123 datagrams.

Practical Implementation: Clock Synchronization Auditor

Below is a complete script using ezsnmp to inspect system time synchronization, evaluate stratum levels, and alert on clock skew:

import re
from ezsnmp import Session
from ezsnmp.exceptions import GenericError

NTP_MODE_MAP = {
    1: "NOT_RUNNING",
    2: "NOT_SYNCHRONIZED",
    3: "NONE_CONFIGURED",
    4: "SYNC_TO_LOCAL",
    5: "SYNC_TO_REFCLOCK",
    6: "SYNC_TO_REMOTE_SERVER",
}

MAX_OFFSET_MS = 10.0
MAX_DISPERSION_MS = 1000.0  # Alert if root dispersion exceeds 1000ms

def parse_offset_ms(val) -> float | None:
    """Parse magnitude and unit, normalizing to milliseconds."""
    if val is None:
        return None
    s = str(val).strip().lower()
    m = re.match(r"^([+-]?\d+(?:\.\d+)?)\s*([a-z]*)$", s)
    if not m:
        return None
    num = float(m.group(1))
    unit = m.group(2)
    if unit in ("s", "sec", "second", "seconds"):
        return num * 1000.0
    elif unit in ("us", "usec", "microsecond", "microseconds"):
        return num / 1000.0
    elif unit in ("ns", "nsec", "nanosecond", "nanoseconds"):
        return num / 1000000.0
    return num

def parse_int(val, default=0):
    digits = "".join(c for c in str(val) if c.isdigit() or c == "-")
    try:
        return int(digits)
    except (ValueError, TypeError):
        return default

def is_valid_result(var) -> bool:
    if var is None:
        return False
    t = getattr(var, "type", None)
    if t in ("NOSUCHOBJECT", "NOSUCHINSTANCE", "ENDOFMIBVIEW"):
        return False
    return True

def audit_device_clock(hostname: str, community: str = "public"):
    """Audit NTP synchronization status, stratum, offset, and dispersion via ezsnmp."""
    session = Session(
        hostname=hostname,
        community=community,
        version=2,
        timeout=2,
        retries=1,
        print_enums_numerically=True
    )

    try:
        # 1. Query System NTP Status scalar objects
        vars = session.get([
            "ntpEntStatusCurrentMode.0",
            "ntpEntStatusStratum.0",
            "ntpEntStatusActiveRefSourceId.0",
            "ntpEntStatusActiveOffset.0",
            "ntpEntStatusDispersion.0",
        ])
        mode_var, stratum_var, active_src_var, offset_var, dispersion_var = vars

        # Verify scalar presence before treating as 0
        if not all(is_valid_result(v) for v in vars):
            return {
                "hostname": hostname,
                "mode": "UNKNOWN",
                "stratum": 16,
                "source": "None",
                "offset_ms": 0.0,
                "dispersion_ms": 0.0,
                "status": "TELEMETRY_UNAVAILABLE",
                "alert": True,
            }

        raw_mode = str(mode_var.value).strip()
        if raw_mode.isdigit():
            mode_code = int(raw_mode)
            mode_name = NTP_MODE_MAP.get(mode_code, f"UNKNOWN_{mode_code}")
        else:
            mode_name = raw_mode.upper()
            mode_code = {v: k for k, v in NTP_MODE_MAP.items()}.get(mode_name, 0)

        stratum = parse_int(stratum_var.value, 16)
        active_source = active_src_var.value
        offset = parse_offset_ms(offset_var.value)
        dispersion = parse_offset_ms(dispersion_var.value)

        if offset is None or dispersion is None:
            return {
                "hostname": hostname,
                "mode": mode_name,
                "stratum": stratum,
                "source": str(active_source),
                "offset_ms": 0.0,
                "dispersion_ms": 0.0,
                "status": "PARSE_ERROR_INVALID_UNITS",
                "alert": True,
            }

        is_synced = mode_code in (5, 6)
        stratum_ok = 1 <= stratum <= 4
        offset_ok = abs(offset) <= MAX_OFFSET_MS
        dispersion_ok = dispersion <= MAX_DISPERSION_MS

        alert = (not is_synced) or (not stratum_ok) or (not offset_ok) or (not dispersion_ok)

        status = "HEALTHY"
        if not is_synced:
            status = "UNSYNCHRONIZED"
        elif not stratum_ok:
            status = f"HIGH_STRATUM_{stratum}"
        elif not offset_ok:
            status = f"CLOCK_SKEW_EXCESSIVE_{offset:.2f}ms"
        elif not dispersion_ok:
            status = f"DISPERSION_EXCESSIVE_{dispersion:.2f}ms"

        return {
            "hostname": hostname,
            "mode": mode_name,
            "stratum": stratum,
            "source": active_source,
            "offset_ms": offset,
            "dispersion_ms": dispersion,
            "status": status,
            "alert": alert,
        }

    except GenericError as exc:
        print(f"[{hostname}] Clock telemetry audit failed: {exc}")
        return None

if __name__ == "__main__":
    result = audit_device_clock("core-rtr01.internal")
    if result:
        flag = " [CRITICAL TIME ALERT]" if result["alert"] else " [OK]"
        print("--- Network Time Telemetry Report ---")
        print(f"  Target Host         : {result['hostname']}")
        print(f"  Sync Status         : {result['mode']}{flag}")
        print(f"  Stratum Level       : {result['stratum']}")
        print(f"  Active Time Source  : {result['source']}")
        print(f"  Measured Offset     : {result['offset_ms']:.3f} ms")
        print(f"  Root Dispersion     : {result['dispersion_ms']:.3f} ms")
        print(f"  Overall Health      : {result['status']}")

📥 Download this script: audit_device_clock.py

Production Best Practices

  • Alert on Root Dispersion Spikes: A node may report near-zero offset while its root dispersion expands rapidly. High dispersion indicates that upstream NTP servers are unreachable or flapping, warning of imminent sync failure.

  • Audit IEEE 1588 PTP for High-Precision Fabrics: In low-latency trading and industrial networks where microsecond or nanosecond precision is required, query vendor PTP MIBs (e.g. PTP-MIB) for grandmaster clock transitions.

  • Parallel Fleet Auditing: Because network time audits involve small, scalar OID lookups, pair ezsnmp with Python’s concurrent.futures.ProcessPoolExecutor (or bounded thread pools with <= 4 workers per process) to audit thousands of switches, hypervisors, and appliances concurrently without thread contention.

Migration Notes: Easy SNMP vs ezsnmp

  • Process-Isolated Concurrency: Legacy easysnmp suffered from global lock bottlenecks when querying hundreds of endpoints across fleet monitoring loops. While ezsnmp provides clean C++17 RAII abstractions, Net-SNMP itself maintains process-global C state and output formatting locks. Using ProcessPoolExecutor ensures fleet auditing scales linearly across multi-core monitoring systems without global lock serialization.

  • Leak-Free Continuous Execution: Time synchronization daemons run 24/7. ezsnmp’s C++17 RAII memory management ensures Net-SNMP structs are safely released on every poll cycle, eliminating memory creep.