replacing ntpdate with ntplib

.github/workflows/lint.yaml

@@ -10,7 +10,7 @@ on:
 
 jobs:
   ament_lint:
-    name: Linting with ${{ matrix.linter }}
+    name: Lint ${{ matrix.linter }}
     strategy:
       fail-fast: false
       matrix:

.github/workflows/test.yaml

@@ -6,14 +6,22 @@ on:
       - ros2
   schedule:
     # Run every week at 20:00 on Sunday
-    - cron: '0 20 * * 0'
+    - cron: "0 20 * * 0"
 
 jobs:
   build_and_test:
-    name: Build and test on ${{ matrix.distro }}
+    name: ${{ matrix.package }} on ${{ matrix.distro }}
     strategy:
       fail-fast: false
       matrix:
+        package:
+          [
+            diagnostic_aggregator,
+            diagnostic_common_diagnostics,
+            diagnostic_updater,
+            self_test,
+          ]
+        distro: [foxy, humble, rolling]
         include:
           - distro: foxy
             os: ubuntu-20.04
@@ -31,16 +39,6 @@ jobs:
       - uses: ros-tooling/action-ros-ci@master
         with:
           target-ros2-distro: ${{ matrix.distro }}
-          package-name: diagnostic_aggregator
-            diagnostic_common_diagnostics
-            diagnostic_updater
-            self_test
-          vcs-repo-file-url: |
-            https://raw.githubusercontent.com/ros2/ros2/master/ros2.repos
-          colcon-defaults: |
-            {
-              "build": {
-                "mixin": ["coverage-gcc"]
-              }
-            }
-          colcon-mixin-repository: https://raw.githubusercontent.com/colcon/colcon-mixin-repository/master/index.yaml
+          package-name: ${{ matrix.package }}
+          # vcs-repo-file-url: |
+          #   https://raw.githubusercontent.com/ros2/ros2/master/ros2.repos

diagnostic_common_diagnostics/CMakeLists.txt

@@ -15,6 +15,12 @@ install(PROGRAMS
 if(BUILD_TESTING)
     find_package(ament_lint_auto REQUIRED)
     ament_lint_auto_find_test_dependencies()
+
+    find_package(ament_cmake_pytest REQUIRED)
+    ament_add_pytest_test(
+        test_ntp_monitor
+        test/systemtest/test_ntp_monitor.py
+        TIMEOUT 10)
 endif()
 
 ament_package()

diagnostic_common_diagnostics/diagnostic_common_diagnostics/ntp_monitor.py

@@ -32,15 +32,14 @@
 # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 # POSSIBILITY OF SUCH DAMAGE.
 
-import re
 import socket
-from subprocess import PIPE
-from subprocess import Popen
-from subprocess import TimeoutExpired
 import sys
 import threading
 
 import diagnostic_updater as DIAG
+
+import ntplib
+
 import rclpy
 from rclpy.node import Node
 
@@ -111,9 +110,12 @@ def checkCB(self):
 
     def ntp_diag(self, st):
         """
-        Add ntp diagnostics to the given status message.
+        Add ntp diagnostics to the given status message and return it.
+
+        Args:
+        ----
+            st: The diagnostic status object to populate
 
-        @param st: The status message to add diagnostics to.
         """
 
         def add_kv(stat_values, key, value):
@@ -122,47 +124,32 @@ def add_kv(stat_values, key, value):
             kv.value = value
             stat_values.append(kv)
 
-        PROCESS_NAME = 'ntpdate'
-        p = Popen([PROCESS_NAME, '-q', self.ntp_hostname],
-                  stdout=PIPE, stdin=PIPE, stderr=PIPE)
+        ntp_client = ntplib.NTPClient()
+        response = None
         try:
-            stdout, stderr = p.communicate(timeout=5)
-        except TimeoutExpired:
-            p.kill()
-            stdout, stderr = p.communicate()
-            self.get_logger().error(
-                'Timeout running %s: "%s"' %
-                (PROCESS_NAME, stderr))
-        rc = p.returncode
-
-        st.values = []
-        add_kv(st.values, 'Offset tolerance (us)', str(self.offset))
-        add_kv(st.values, 'Offset tolerance (us) for Error',
-               str(self.error_offset))
-        output = stdout.decode('UTF-8')
-        errors = stderr.decode('UTF-8')
-
-        if rc == 0:
-            re_match = re.search('offset (.*),', output)
-            measured_offset = float(re_match[1])*1000000
-
-            st.level = DIAG.DiagnosticStatus.OK
-            st.message = 'OK'
-            add_kv(st.values, 'Offset (us)', str(measured_offset))
+            response = ntp_client.request(self.ntp_hostname, version=3)
+        except ntplib.NTPException as e:
+            self.get_logger().error(f'NTP Error: {e}')
+            st.level = DIAG.DiagnosticStatus.ERROR
+            st.message = f'NTP Error: {e}'
+            add_kv(st.values, 'Offset (us)', 'N/A')
+            add_kv(st.values, 'Errors', str(e))
 
+        if response is not None:
+            measured_offset = response.offset * 1e6
             if (abs(measured_offset) > self.offset):
                 st.level = DIAG.DiagnosticStatus.WARN
-                st.message = 'NTP Offset Too High'
+                st.message = \
+                    f'NTP offset above threshold: {measured_offset}>'\
+                    f'{self.offset} us'
             if (abs(measured_offset) > self.error_offset):
                 st.level = DIAG.DiagnosticStatus.ERROR
-                st.message = 'NTP Offset Too High'
-
-        else:
-            st.level = DIAG.DiagnosticStatus.ERROR
-            st.message = 'Error Running ntpdate. Returned %d' % rc
-            add_kv(st.values, 'Offset (us)', 'N/A')
-            add_kv(st.values, 'Output', output)
-            add_kv(st.values, 'Errors', errors)
+                st.message = \
+                    f'NTP offset above error threshold: {measured_offset}>'\
+                    f'{self.error_offset} us'
+            if (abs(measured_offset) < self.offset):
+                st.level = DIAG.DiagnosticStatus.OK
+                st.message = f'NTP Offset OK: {measured_offset} us'
 
         return st
 
@@ -171,47 +158,35 @@ def ntp_monitor_main(argv=sys.argv[1:]):
     import argparse
     parser = argparse.ArgumentParser()
     parser.add_argument('--ntp_hostname',
-                        action='store',
-                        default='pool.ntp.org',
+                        action='store', default='0.pool.ntp.org',
                         type=str)
-    parser.add_argument('--offset-tolerance',
-                        dest='offset_tol',
-                        action='store',
-                        default=500,
-                        help='Offset from NTP host',
-                        metavar='OFFSET-TOL',
+    parser.add_argument('--offset-tolerance', dest='offset_tol',
+                        action='store', default=500,
+                        help='Offset from NTP host', metavar='OFFSET-TOL',
                         type=int)
-    parser.add_argument('--error-offset-tolerance',
-                        dest='error_offset_tol',
-                        action='store',
-                        default=5000000,
+    parser.add_argument('--error-offset-tolerance', dest='error_offset_tol',
+                        action='store', default=5000000,
                         help='Offset from NTP host. Above this is error',
-                        metavar='OFFSET-TOL',
-                        type=int)
-    parser.add_argument('--self_offset-tolerance',
-                        dest='self_offset_tol',
-                        action='store',
-                        default=500,
-                        help='Offset from self',
-                        metavar='SELF_OFFSET-TOL',
+                        metavar='OFFSET-TOL', type=int)
+    parser.add_argument('--self_offset-tolerance', dest='self_offset_tol',
+                        action='store', default=500,
+                        help='Offset from self', metavar='SELF_OFFSET-TOL',
                         type=int)
-    parser.add_argument('--diag-hostname',
-                        dest='diag_hostname',
-                        help="Computer name in diagnostics output (ex: 'c1')",
+    parser.add_argument('--diag-hostname', dest='diag_hostname',
+                        help='Computer name in diagnostics output (ex: "c1")',
                         metavar='DIAG_HOSTNAME',
-                        action='store',
-                        default=None,
+                        action='store', default=None,
                         type=str)
-    parser.add_argument('--no-self-test',
-                        dest='do_self_test',
+    parser.add_argument('--no-self-test', dest='do_self_test',
                         help='Disable self test',
-                        action='store_false',
-                        default=True)
+                        action='store_false', default=True)
     args = parser.parse_args(args=argv)
 
     offset = args.offset_tol
     self_offset = args.self_offset_tol
     error_offset = args.error_offset_tol
+    assert offset < error_offset, \
+        'Offset tolerance must be less than error offset tolerance'
 
     ntp_monitor = NTPMonitor(args.ntp_hostname, offset, self_offset,
                              args.diag_hostname, error_offset,

diagnostic_common_diagnostics/package.xml

@@ -20,7 +20,7 @@
 
   <exec_depend>rclpy</exec_depend>
   <exec_depend>diagnostic_updater</exec_depend>
-  <exec_depend>ntpdate</exec_depend>
+  <exec_depend>python3-ntplib</exec_depend>
 
   <test_depend>ament_lint_auto</test_depend>
   <!-- Usage of ament_lint_common is locked by https://github.com/ament/ament_lint/issues/423

diagnostic_common_diagnostics/test/systemtest/test_ntp_monitor.py

@@ -0,0 +1,115 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+# Software License Agreement (BSD License)
+#
+# Copyright (c) 2023, Robert Bosch GmbH
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions
+# are met:
+#
+#  * Redistributions of source code must retain the above copyright
+#    notice, this list of conditions and the following disclaimer.
+#  * Redistributions in binary form must reproduce the above
+#    copyright notice, this list of conditions and the following
+#    disclaimer in the documentation and/or other materials provided
+#    with the distribution.
+#  * Neither the name of the Willow Garage nor the names of its
+#    contributors may be used to endorse or promote products derived
+#    from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
+# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
+# COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
+# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+
+import os
+import subprocess
+import unittest
+
+import ament_index_python
+
+from diagnostic_msgs.msg import DiagnosticArray
+
+import rclpy
+
+TIMEOUT_MAX_S = 2.
+
+
+class TestNTPMonitor(unittest.TestCase):
+
+    def __init__(self, methodName: str = 'runTest') -> None:
+        super().__init__(methodName)
+        rclpy.init()
+        self.n_msgs_received = 0
+
+    def setUp(self):
+        self.n_msgs_received = 0
+        n = self._count_msgs(TIMEOUT_MAX_S)
+        self.assertEqual(n, 0)
+        self.subprocess = subprocess.Popen(
+            [
+                os.path.join(
+                    ament_index_python.get_package_prefix(
+                        'diagnostic_common_diagnostics'
+                    ),
+                    'lib',
+                    'diagnostic_common_diagnostics',
+                    'ntp_monitor.py'
+                )
+            ]
+        )
+
+    def tearDown(self):
+        self.subprocess.kill()
+
+    def _diagnostics_callback(self, msg):
+        search_strings = [
+            'NTP offset from',
+            'NTP self-offset for'
+        ]
+        for search_string in search_strings:
+            if search_string not in ''.join([
+                s.name for s in msg.status
+            ]):
+                return
+        self.n_msgs_received += 1
+
+    def _count_msgs(self, timeout_s):
+        self.n_msgs_received = 0
+        node = rclpy.create_node('test_ntp_monitor')
+        node.create_subscription(
+            DiagnosticArray,
+            'diagnostics',
+            self._diagnostics_callback,
+            1
+        )
+        TIME_D_S = .1
+        waited_s = 0.
+        while waited_s < timeout_s and self.n_msgs_received == 0:
+            rclpy.spin_once(node, timeout_sec=TIME_D_S)
+            waited_s += TIME_D_S
+        node.destroy_node()
+        return self.n_msgs_received
+
+    def test_publishing(self):
+        n = self._count_msgs(TIMEOUT_MAX_S)
+
+        self.assertGreater(
+            n,
+            0,
+            f'No messages received within {TIMEOUT_MAX_S}s'
+        )
+
+
+if __name__ == '__main__':
+    unittest.main()