Prepare 3.0.7 release

CHANGELOG.md

@@ -22,6 +22,14 @@ The current development release is 3.0.6.  This is the recommended version for
 anyone who wants the latest features.  It should be suitable for production
 deployment very soon.
 
+## [3.0.7] - 2024-01-05
+
+### Changed
+
+- Handle loss of telegraf socket more gracefully
+- Handle failure of subprocess more gracefully
+- Ensure that systemd always restarts ntpmon regardless of exit code
+
 ## [3.0.6] - 2024-01-02
 
 ### Added

debian/changelog

@@ -1,3 +1,9 @@
+ntpmon (3.0.7-1) focal; urgency=medium
+
+  * New upstream release.
+
+ -- Paul Gear <[email protected]>  Fri, 05 Jan 2024 11:25:25 +1000
+
 ntpmon (3.0.6-1) focal; urgency=medium
 
   * New upstream release.

src/ntpmon-telegraf.systemd

@@ -8,7 +8,7 @@ Requires=telegraf.service
 [Service]
 ExecStart={{ install_dir }}/ntpmon.py --interval 60 --mode telegraf --implementation {{ implementation }}
 KillMode=process
-Restart=on-failure
+Restart=always
 RestartSec=42s
 User={{ user }}
 Group={{ group }}

src/ntpmon.py

@@ -164,7 +164,7 @@ async def peer_stats_task(args: argparse.Namespace, output: outputs.Output) -> N
             if stats is not None:
                 if "peertype" not in stats:
                     stats["peertype"] = find_type(stats["source"], checkobjs["peers"].peers)
-                output.send_measurement(stats, debug=args.debug)
+                output.send_peer_measurements(stats, debug=args.debug)
 
 
 async def summary_stats_task(args: argparse.Namespace, output: outputs.Output) -> None:
@@ -187,6 +187,7 @@ async def start_tasks(args: argparse.Namespace) -> None:
     peer_stats = asyncio.create_task(peer_stats_task(args, output), name="peerstats")
     summary_stats = asyncio.create_task(summary_stats_task(args, output), name="summarystats")
     await asyncio.wait((peer_stats, summary_stats), return_when=asyncio.FIRST_COMPLETED)
+    sys.exit(1)
 
 
 if __name__ == "__main__":

src/outputs.py

@@ -4,10 +4,12 @@
 
 
 import argparse
+import datetime
 import socket
 import sys
 
 from io import TextIOWrapper
+import time
 from typing import ClassVar, Dict, List, Tuple
 
 
@@ -73,10 +75,10 @@ class Output:
     def send_info(self, metrics: dict, debug: bool = False) -> None:
         pass
 
-    def send_measurement(self, metrics: dict, debug: bool = False) -> None:
+    def send_peer_counts(self, metrics: dict, debug: bool = False) -> None:
         pass
 
-    def send_peer_counts(self, metrics: dict, debug: bool = False) -> None:
+    def send_peer_measurements(self, metrics: dict, debug: bool = False) -> None:
         pass
 
     def send_summary_stats(self, metrics: dict, debug: bool = False) -> None:
@@ -89,12 +91,12 @@ def __init__(self, args: argparse.Namespace) -> None:
 
     formatstr: ClassVar[str] = 'PUTVAL "%s/ntpmon-%s" interval=%d N:%.9f'
 
-    def send_measurement(self, metrics: dict, debug: bool = False) -> None:
-        self.send_stats(metrics, self.peerstatstypes, hostname=metrics["source"], debug=debug)
-
     def send_peer_counts(self, metrics: dict, debug: bool = False) -> None:
         self.send_stats(metrics, self.peertypes, debug=debug)
 
+    def send_peer_measurements(self, metrics: dict, debug: bool = False) -> None:
+        self.send_stats(metrics, self.peerstatstypes, hostname=metrics["source"], debug=debug)
+
     def send_stats(self, metrics: dict, types: dict, debug: bool = False, hostname: str = None) -> None:
         if hostname is None:
             hostname = self.args.hostname
@@ -191,16 +193,6 @@ def send_info(self, metrics: dict, debug: bool = False) -> None:
             debug=debug,
         )
 
-    def send_measurement(self, metrics: dict, debug: bool = False) -> None:
-        self.send_stats(
-            "ntpmon_peer",
-            metrics,
-            self.peerstatstypes,
-            [x for x in self.peerstatslabels if x in metrics],
-            [metrics[x] for x in self.peerstatslabels if x in metrics],
-            debug=debug,
-        )
-
     def send_peer_counts(self, metrics: dict, debug: bool = False) -> None:
         for metric in sorted(self.peertypes.keys()):
             if metric in metrics:
@@ -214,8 +206,15 @@ def send_peer_counts(self, metrics: dict, debug: bool = False) -> None:
                     debug=debug,
                 )
 
-    def send_summary_stats(self, metrics: dict, debug: bool = False) -> None:
-        self.send_stats("ntpmon", metrics, self.summarystatstypes, debug=debug)
+    def send_peer_measurements(self, metrics: dict, debug: bool = False) -> None:
+        self.send_stats(
+            "ntpmon_peer",
+            metrics,
+            self.peerstatstypes,
+            [x for x in self.peerstatslabels if x in metrics],
+            [metrics[x] for x in self.peerstatslabels if x in metrics],
+            debug=debug,
+        )
 
     def send_stats(
         self,
@@ -240,6 +239,9 @@ def send_stats(
                     value /= 100
                 self.set_prometheus_metric(name, description, value, fmt, labelnames, labels, debug=debug)
 
+    def send_summary_stats(self, metrics: dict, debug: bool = False) -> None:
+        self.send_stats("ntpmon", metrics, self.summarystatstypes, debug=debug)
+
     def set_prometheus_metric(
         self,
         name: str,
@@ -277,7 +279,8 @@ def set_prometheus_metric(
 class TelegrafOutput(Output):
     def __init__(self, args: argparse.Namespace) -> None:
         super().__init__()
-        self.file = sys.stdout if args.debug else self.get_telegraf_file(args.connect)
+        self.args = args
+        self.set_file()
 
     @staticmethod
     def get_telegraf_file(connect: str) -> TextIOWrapper:
@@ -288,27 +291,44 @@ def get_telegraf_file(connect: str) -> TextIOWrapper:
         s.connect((host, port))
         return s.makefile(mode="w")
 
-    def send_info(self, metrics: dict, debug: bool) -> None:
-        telegraf_line = line_protocol.to_line_protocol(metrics, "ntpmon_info")
-        print(telegraf_line, file=self.file)
+    def send(self, name: str, metrics: dict, tries: int = 0) -> None:
+        telegraf_line = line_protocol.to_line_protocol(metrics, name)
+        if tries >= 5:
+            print("Reached maximum retries on telegraf connection", file=sys.stderr)
+            print(telegraf_line)
+            sys.exit(3)
+        try:
+            print(telegraf_line, file=self.file)
+        except BrokenPipeError as bpe:
+            # If we have lost our connection to telegraf, wait a little, then
+            # reopen the socket and try again. We add a timestamp to metrics
+            # without it, in case it takes a while to make the connection.
+            if "datetime" not in metrics:
+                metrics["datetime"] = datetime.datetime.now(tz=datetime.timezone.utc)
+            time.sleep(0.1)
+            self.set_file()
+            self.send(name, metrics, tries + 1)
 
-    def send_measurement(self, metrics: dict, debug: bool = False) -> None:
-        telegraf_line = line_protocol.to_line_protocol(metrics, "ntpmon_peer")
-        print(telegraf_line, file=self.file)
+    def send_info(self, metrics: dict, debug: bool) -> None:
+        self.send("ntpmon_info", metrics)
 
     def send_peer_counts(self, metrics: dict, debug: bool = False) -> None:
         for metric in sorted(self.peertypes.keys()):
             telegraf_metrics = {
                 "count": metrics[metric],
                 "peertype": metric,
             }
-            output = line_protocol.to_line_protocol(telegraf_metrics, "ntpmon_peers")
-            print(output, file=self.file)
+            self.send("ntpmon_peers", telegraf_metrics)
+
+    def send_peer_measurements(self, metrics: dict, debug: bool = False) -> None:
+        self.send("ntpmon_peer", metrics)
 
     def send_summary_stats(self, metrics: dict, debug: bool = False) -> None:
         telegraf_metrics = {k: metrics[k] for k in sorted(self.summarytypes.keys()) if k in metrics}
-        telegraf_line = line_protocol.to_line_protocol(telegraf_metrics, "ntpmon")
-        print(telegraf_line, file=self.file)
+        self.send("ntpmon", telegraf_metrics)
+
+    def set_file(self) -> None:
+        self.file = sys.stdout if self.args.debug else self.get_telegraf_file(self.args.connect)
 
 
 def get_output(args: argparse.Namespace) -> Output:

src/process.py

@@ -35,6 +35,7 @@
 
 
 def execute_subprocess(cmd, timeout, debug, errfatal):
+    output = None
     try:
         output = subprocess.check_output(
             cmd,

src/version_data.py

@@ -2,4 +2,4 @@
 # and bash.
 MAJOR="3"
 MINOR="0"
-PATCH="6"
+PATCH="7"