Repository: eclipse-bluechi/bluechi (language: python)

Issue:
Add agent heartbeat timestamps to client node monitoring.
This addresses issue #337, adding the last heartbeat timestamp in client node monitoring.

```
NODE                          |     STATE|      LAST HEARTBEAT|
===============================================================
agent-001                     |    online|     Jun 04 12:28:03|
agent-002                     |   offline|                    |
```

If the node goes offline, you can determine from the last heartbeat timestamp when it went down.
Add last heartbeat timestamp to node monitoring
#186 added node monitoring with `hirtectl monitor node-connection`.  This issue is to add a column with the last heartbeat timestamp from a node.  This can be used to determine how long an agent has been down.



Discussion / hints:
Should we add a test for this somewhere?
Yep!  I suggest I add an integration test in a follow-up PR.
On Mon, Jun 05, 2023 at 05:08:48AM -0700, Steve Dunnagan wrote:
> Yep!  I suggest I add an integration test in a follow-up PR.

Why splitting this into two PR? Due to time availability or is there another
reason?

> On Mon, Jun 05, 2023 at 05:08:48AM -0700, Steve Dunnagan wrote: Yep! I suggest I add an integration test in a follow-up PR.
> Why splitting this into two PR? Due to time availability or is there another reason?

The monitoring in `hirtectl` can't really be tested, I think (but maybe I am missing something). So we'd have to use the D-Bus API, create a monitor + subscription with custom node-connection handling to reliably test this - which is a bit more effort. Is it ok to create a gh issue (and add it to the integration test epic in jira) to keep track of it? @pypingou 
P.S.: This is probably one of the use cases for the Python SDK @dougsland was talking about :) 
We decided not to add timestamps to the agent heartbeat signals.
We decided not to add a timestamp to the agent heartbeat.

Fix patch:
diff --git a/data/org.containers.hirte.Node.xml b/data/org.containers.hirte.Node.xml
index 7b3f511a51..e92b8afbc6 100644
--- a/data/org.containers.hirte.Node.xml
+++ b/data/org.containers.hirte.Node.xml
@@ -60,5 +60,6 @@
 
     <property name="Name" type="s" access="read" />
     <property name="Status" type="s" access="read" />
+    <property name="LastSeenTimestamp" type="t" access="read" />
   </interface>
 </node>
\ No newline at end of file
diff --git a/data/org.containers.hirte.internal.Agent.xml b/data/org.containers.hirte.internal.Agent.xml
index da2e343087..9833338f07 100644
--- a/data/org.containers.hirte.internal.Agent.xml
+++ b/data/org.containers.hirte.internal.Agent.xml
@@ -107,8 +107,6 @@
       <arg name="nodeName" type="s" />
       <arg name="unitName" type="s" />
     </signal>
-    <signal name="Heartbeat">
-      <arg name="agent_name" type="s" />
-    </signal>
+    <signal name="Heartbeat" />
   </interface>
 </node>
\ No newline at end of file
diff --git a/doc/docs/api/description.md b/doc/docs/api/description.md
index 15ccb34414..5db426de15 100644
--- a/doc/docs/api/description.md
+++ b/doc/docs/api/description.md
@@ -203,6 +203,10 @@ Object path: `/org/containers/hirte/node/$name`
   * `Status` - `s`
 
     Status of the node, currently one of: `online`, `offline`. Emits changed when this changes.
+  
+  * `LastSeenTimestamp` - `t`
+
+    Timestamp of the last successfully received heartbeat of the node.
 
 ### interface org.containers.hirte.Job
 
@@ -375,7 +379,7 @@ This is the main interface that the node implements and that is used by the mana
     This is emitted when a proxy is not needed anymore because the service requiring the proxy
     service is stopped.
 
-  * `Heartbeat(s nodeName)`
+  * `Heartbeat()`
 
     This is a periodic signal from the node to the manager.
 
diff --git a/src/agent/agent.c b/src/agent/agent.c
index f735f70a6b..525d7cd8e5 100644
--- a/src/agent/agent.c
+++ b/src/agent/agent.c
@@ -124,8 +124,7 @@ static int agent_heartbeat_timer_callback(sd_event_source *event_source, UNUSED
                                 INTERNAL_AGENT_OBJECT_PATH,
                                 INTERNAL_AGENT_INTERFACE,
                                 "Heartbeat",
-                                "s",
-                                agent->name);
+                                "");
                 if (r < 0) {
                         hirte_log_errorf("Failed to emit heartbeat signal: %s", strerror(-r));
                 }
@@ -1389,7 +1388,7 @@ static const sd_bus_vtable internal_agent_vtable[] = {
                                         SD_BUS_PARAM(reason),
                         0),
         SD_BUS_SIGNAL_WITH_NAMES("UnitRemoved", "s", SD_BUS_PARAM(unit), 0),
-        SD_BUS_SIGNAL_WITH_NAMES("Heartbeat", "s", SD_BUS_PARAM(agent_name), 0),
+        SD_BUS_SIGNAL("Heartbeat", "", 0),
         SD_BUS_METHOD("StartDep", "s", "", agent_method_start_dep, 0),
         SD_BUS_METHOD("StopDep", "s", "", agent_method_stop_dep, 0),
         SD_BUS_METHOD("EnableUnitFiles", "asbb", "ba(sss)", agent_method_passthrough_to_systemd, 0),
diff --git a/src/client/method_monitor.c b/src/client/method_monitor.c
index 75f4e63a13..1e97c9f2d0 100644
--- a/src/client/method_monitor.c
+++ b/src/client/method_monitor.c
@@ -1,6 +1,7 @@
 /* SPDX-License-Identifier: GPL-2.0-or-later */
 #include "libhirte/common/common.h"
 #include "libhirte/common/list.h"
+#include "libhirte/common/time-util.h"
 #include "libhirte/service/shutdown.h"
 
 #include "method_monitor.h"
@@ -299,37 +300,61 @@ typedef struct NodeConnection NodeConnection;
 
 struct NodeConnection {
         char *name;
+        char *node_path;
         char *state;
+        uint64_t last_seen;
         LIST_FIELDS(NodeConnection, nodes);
 };
 
 typedef struct Nodes {
+        sd_bus *api_bus;
         LIST_HEAD(NodeConnection, nodes);
 } Nodes;
 
-static Nodes *nodes_new() {
+static Nodes *nodes_new(sd_bus *api_bus) {
         Nodes *nodes = malloc0(sizeof(Nodes));
+        nodes->api_bus = api_bus;
         LIST_HEAD_INIT(nodes->nodes);
         return nodes;
 }
 
-static void nodes_add_connection(Nodes *nodes, const char *node_name, const char *node_state) {
+static char *node_connection_fmt_last_seen(NodeConnection *con) {
+        if (streq(con->state, "online")) {
+                return strdup("now");
+        } else if (streq(con->state, "offline") && con->last_seen == 0) {
+                return strdup("never");
+        }
+
+        struct timespec t;
+        t.tv_sec = (time_t) con->last_seen;
+        t.tv_nsec = 0;
+        return get_formatted_log_timestamp_for_timespec(t, false);
+}
+
+static void nodes_add_connection(
+                Nodes *nodes,
+                const char *node_name,
+                const char *node_path,
+                const char *node_state,
+                uint64_t last_seen_timestamp) {
         NodeConnection *node_connection = malloc0(sizeof(NodeConnection));
         node_connection->name = strdup(node_name);
+        node_connection->node_path = strdup(node_path);
         node_connection->state = strdup(node_state);
+        node_connection->last_seen = last_seen_timestamp;
 
         LIST_APPEND(nodes, nodes->nodes, node_connection);
 }
 
-static void nodes_update_connection(Nodes *nodes, const char *node_name, const char *node_state) {
+static NodeConnection *nodes_find_by_name(Nodes *nodes, const char *node_name) {
         NodeConnection *curr = NULL;
         NodeConnection *next = NULL;
         LIST_FOREACH_SAFE(nodes, curr, next, nodes->nodes) {
                 if (streq(curr->name, node_name)) {
-                        free(curr->state);
-                        curr->state = strdup(node_state);
+                        return curr;
                 }
         }
+        return NULL;
 }
 
 static void nodes_unref(Nodes *nodes) {
@@ -341,6 +366,7 @@ static void nodes_unref(Nodes *nodes) {
         NodeConnection *next = NULL;
         LIST_FOREACH_SAFE(nodes, curr, next, nodes->nodes) {
                 free(curr->name);
+                free(curr->node_path);
                 free(curr->state);
         }
 
@@ -357,14 +383,44 @@ static void print_nodes(Nodes *nodes) {
         printf("\033[%d;%dH", 0, 0);
 
         /* print monitor header */
-        printf("%-30.30s|%10s\n", "NODE", "STATE");
-        printf("=========================================\n");
+        printf("%-30.30s| %-10.10s| %-28.28s\n", "NODE", "STATE", "LAST SEEN");
+        printf("=========================================================================\n");
 
         NodeConnection *curr = NULL;
         NodeConnection *next = NULL;
         LIST_FOREACH_SAFE(nodes, curr, next, nodes->nodes) {
-                printf("%-30.30s|%10s\n", curr->name, curr->state);
+                _cleanup_free_ char *last_seen = node_connection_fmt_last_seen(curr);
+                printf("%-30.30s| %-10.10s| %-28.28s\n", curr->name, curr->state, last_seen);
+        }
+}
+
+
+static int fetch_last_seen_timestamp_property(
+                sd_bus *api_bus, const char *node_path, uint64_t *ret_last_seen_timestamp) {
+        _cleanup_sd_bus_error_ sd_bus_error prop_error = SD_BUS_ERROR_NULL;
+        _cleanup_sd_bus_message_ sd_bus_message *prop_reply = NULL;
+
+        int r = sd_bus_get_property(
+                        api_bus,
+                        HIRTE_INTERFACE_BASE_NAME,
+                        node_path,
+                        NODE_INTERFACE,
+                        "LastSeenTimestamp",
+                        &prop_error,
+                        &prop_reply,
+                        "t");
+        if (r < 0) {
+                return r;
+        }
+
+        uint64_t last_seen_timestamp = 0;
+        r = sd_bus_message_read(prop_reply, "t", &last_seen_timestamp);
+        if (r < 0) {
+                return r;
         }
+        *ret_last_seen_timestamp = last_seen_timestamp;
+
+        return 0;
 }
 
 static int on_node_connection_state_changed(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *error) {
@@ -379,7 +435,21 @@ static int on_node_connection_state_changed(sd_bus_message *m, void *userdata, U
                 return 0;
         }
 
-        nodes_update_connection(nodes, node_name, con_state);
+        NodeConnection *con = nodes_find_by_name(nodes, node_name);
+        if (con == NULL) {
+                fprintf(stderr, "Couldn't find connection for node '%s'", node_name);
+                return 0;
+        }
+
+        r = fetch_last_seen_timestamp_property(nodes->api_bus, con->node_path, &con->last_seen);
+        if (r < 0) {
+                fprintf(stderr, "Failed to get last seen property of node %s: %s\n", con->name, strerror(-r));
+                return r;
+        }
+
+        free(con->state);
+        con->state = strdup(con_state);
+
         print_nodes(nodes);
 
         return 0;
@@ -404,7 +474,7 @@ int method_monitor_node_connection_state(sd_bus *api_bus) {
                 return r;
         }
 
-        _cleanup_nodes_ Nodes *nodes = nodes_new();
+        _cleanup_nodes_ Nodes *nodes = nodes_new(api_bus);
 
         r = sd_bus_message_enter_container(reply, SD_BUS_TYPE_ARRAY, "(sos)");
         if (r < 0) {
@@ -413,16 +483,30 @@ int method_monitor_node_connection_state(sd_bus *api_bus) {
         }
         while (sd_bus_message_at_end(reply, false) == 0) {
                 const char *name = NULL;
-                UNUSED const char *path = NULL;
+                const char *path = NULL;
                 const char *state = NULL;
+                uint64_t last_seen_timestamp = 0;
 
                 r = sd_bus_message_read(reply, "(sos)", &name, &path, &state);
                 if (r < 0) {
                         fprintf(stderr, "Failed to read node information: %s\n", strerror(-r));
                         return r;
                 }
-                nodes_add_connection(nodes, name, state);
+
+                if (streq(state, "offline")) {
+                        r = fetch_last_seen_timestamp_property(api_bus, path, &last_seen_timestamp);
+                        if (r < 0) {
+                                fprintf(stderr,
+                                        "Failed to get last seen property of node %s: %s\n",
+                                        name,
+                                        strerror(-r));
+                                return r;
+                        }
+                }
+
+                nodes_add_connection(nodes, name, path, state, last_seen_timestamp);
         }
+
         print_nodes(nodes);
 
         r = sd_bus_match_signal(
diff --git a/src/libhirte/common/time-util.c b/src/libhirte/common/time-util.c
index e811e021c7..6ab5762cd5 100644
--- a/src/libhirte/common/time-util.c
+++ b/src/libhirte/common/time-util.c
@@ -16,6 +16,20 @@ uint64_t get_time_micros() {
         return now_micros;
 }
 
+int get_time_seconds(time_t *ret_time_seconds) {
+        if (ret_time_seconds == NULL) {
+                return -EINVAL;
+        }
+
+        struct timespec now;
+        int r = clock_gettime(CLOCK_REALTIME, &now);
+        if (r < 0) {
+                return r;
+        }
+        *ret_time_seconds = now.tv_sec;
+        return 0;
+}
+
 uint64_t finalize_time_interval_micros(int64_t start_time_micros) {
         return get_time_micros() - start_time_micros;
 }
diff --git a/src/libhirte/common/time-util.h b/src/libhirte/common/time-util.h
index 6b28393abb..e0e10b6828 100644
--- a/src/libhirte/common/time-util.h
+++ b/src/libhirte/common/time-util.h
@@ -12,6 +12,7 @@ static const double nanosec_to_millisec_multiplier = 1e-6;
 static const uint64_t millis_in_second = 1000;
 
 uint64_t get_time_micros();
+int get_time_seconds(time_t *ret_time_seconds);
 uint64_t finalize_time_interval_micros(int64_t start_time_micros);
 double micros_to_millis(uint64_t time_micros);
 char *get_formatted_log_timestamp();
diff --git a/src/manager/node.c b/src/manager/node.c
index 9a5f3adf0e..1d5626fc65 100644
--- a/src/manager/node.c
+++ b/src/manager/node.c
@@ -45,6 +45,14 @@ static int node_property_get_status(
                 sd_bus_message *reply,
                 void *userdata,
                 sd_bus_error *ret_error);
+static int node_property_get_last_seen(
+                sd_bus *bus,
+                const char *path,
+                const char *interface,
+                const char *property,
+                sd_bus_message *reply,
+                void *userdata,
+                sd_bus_error *ret_error);
 
 static const sd_bus_vtable internal_manager_controller_vtable[] = {
         SD_BUS_VTABLE_START(0), SD_BUS_METHOD("Register", "s", "", node_method_register, 0), SD_BUS_VTABLE_END
@@ -66,6 +74,7 @@ static const sd_bus_vtable node_vtable[] = {
         SD_BUS_METHOD("SetLogLevel", "s", "", node_method_set_log_level, 0),
         SD_BUS_PROPERTY("Name", "s", node_property_get_nodename, 0, SD_BUS_VTABLE_PROPERTY_CONST),
         SD_BUS_PROPERTY("Status", "s", node_property_get_status, 0, SD_BUS_VTABLE_PROPERTY_EMITS_CHANGE),
+        SD_BUS_PROPERTY("LastSeenTimestamp", "t", node_property_get_last_seen, 0, SD_BUS_VTABLE_PROPERTY_EXPLICIT),
         SD_BUS_VTABLE_END
 };
 
@@ -145,6 +154,8 @@ Node *node_new(Manager *manager, const char *name) {
                 return NULL;
         }
 
+        node->last_seen = 0;
+
         if (name) {
                 node->name = strdup(name);
                 if (node->name == NULL) {
@@ -303,8 +314,7 @@ static struct hashmap *node_compute_unique_monitor_subscriptions(Node *node, con
 }
 
 
-static int node_match_job_state_changed(
-                UNUSED sd_bus_message *m, UNUSED void *userdata, UNUSED sd_bus_error *error) {
+static int node_match_job_state_changed(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *error) {
         Node *node = userdata;
         Manager *manager = node->manager;
         uint32_t hirte_job_id = 0;
@@ -480,15 +490,14 @@ static int node_match_job_done(UNUSED sd_bus_message *m, UNUSED void *userdata,
         return 1;
 }
 
-static int node_match_heartbeat(UNUSED sd_bus_message *m, UNUSED void *userdata, UNUSED sd_bus_error *error) {
-        char *node_name = NULL;
+static int node_match_heartbeat(UNUSED sd_bus_message *m, void *userdata, UNUSED sd_bus_error *error) {
+        Node *node = userdata;
 
-        int r = sd_bus_message_read(m, "s", &node_name);
+        int r = get_time_seconds(&node->last_seen);
         if (r < 0) {
-                hirte_log_errorf("Error reading heartbeat: %s", strerror(-r));
+                hirte_log_errorf("Failed to get current time on heartbeat: %s", strerror(-r));
                 return 0;
         }
-
         return 1;
 }
 
@@ -749,7 +758,7 @@ bool node_set_agent_bus(Node *node, sd_bus *bus) {
                                 INTERNAL_AGENT_INTERFACE,
                                 AGENT_HEARTBEAT_SIGNAL_NAME,
                                 node_match_heartbeat,
-                                NULL);
+                                node);
                 if (r < 0) {
                         hirte_log_errorf("Failed to add heartbeat signal match: %s", strerror(-r));
                         return false;
@@ -1000,6 +1009,19 @@ static int node_property_get_status(
         return sd_bus_message_append(reply, "s", node_get_status(node));
 }
 
+static int node_property_get_last_seen(
+                UNUSED sd_bus *bus,
+                UNUSED const char *path,
+                UNUSED const char *interface,
+                UNUSED const char *property,
+                sd_bus_message *reply,
+                void *userdata,
+                UNUSED sd_bus_error *ret_error) {
+        Node *node = userdata;
+
+        return sd_bus_message_append(reply, "t", node->last_seen);
+}
+
 
 AgentRequest *agent_request_ref(AgentRequest *req) {
         req->ref_count++;
diff --git a/src/manager/node.h b/src/manager/node.h
index 9ec63e7c85..cb029902a6 100644
--- a/src/manager/node.h
+++ b/src/manager/node.h
@@ -52,6 +52,7 @@ struct Node {
         LIST_HEAD(ProxyDependency, proxy_dependencies);
 
         struct hashmap *unit_subscriptions;
+        time_t last_seen;
 };
 
 Node *node_new(Manager *manager, const char *name);


Test patch:
diff --git a/src/libhirte/test/common/meson.build b/src/libhirte/test/common/meson.build
index c2f9b2e1f2..7c7f18a68d 100644
--- a/src/libhirte/test/common/meson.build
+++ b/src/libhirte/test/common/meson.build
@@ -19,3 +19,4 @@ foreach src : common_src
 endforeach
 
 subdir('network')
+subdir('time-util')
diff --git a/src/libhirte/test/common/time-util/get_log_timestamp.c b/src/libhirte/test/common/time-util/get_log_timestamp.c
new file mode 100644
index 0000000000..2a8257a027
--- /dev/null
+++ b/src/libhirte/test/common/time-util/get_log_timestamp.c
@@ -0,0 +1,45 @@
+/* SPDX-License-Identifier: GPL-2.0-or-later */
+#include <assert.h>
+#include <stdbool.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <time.h>
+
+#include "libhirte/common/common.h"
+#include "libhirte/common/string-util.h"
+#include "libhirte/common/time-util.h"
+
+bool test_get_log_timestamp(__time_t tv_sec, __time_t tv_nsec, char *expected_timestamp) {
+        struct timespec time;
+        time.tv_sec = tv_sec;
+        time.tv_nsec = tv_nsec;
+        _cleanup_free_ char *timestamp = get_formatted_log_timestamp_for_timespec(time, true);
+
+        if (streq(timestamp, expected_timestamp)) {
+                return true;
+        }
+
+        fprintf(stdout,
+                "FAILED: for time (%ldsec, %ldnsec) - Expected %s, but got %s\n",
+                tv_sec,
+                tv_nsec,
+                expected_timestamp,
+                timestamp);
+        return false;
+}
+
+int main() {
+        bool result = true;
+
+        result = result && test_get_log_timestamp(0, 0, "1970-01-01 00:00:00,000+0000");
+        result = result && test_get_log_timestamp(1687075200, 0, "2023-06-18 08:00:00,000+0000");
+        result = result && test_get_log_timestamp(1687075200, 999999999, "2023-06-18 08:00:00,999+0000");
+        result = result && test_get_log_timestamp(1687075200, 999000000, "2023-06-18 08:00:00,999+0000");
+        result = result && test_get_log_timestamp(1687075200, 998900000, "2023-06-18 08:00:00,998+0000");
+        result = result && test_get_log_timestamp(1687046399, 999000000, "2023-06-17 23:59:59,999+0000");
+
+        if (result) {
+                return EXIT_SUCCESS;
+        }
+        return EXIT_FAILURE;
+}
\ No newline at end of file
diff --git a/src/libhirte/test/common/time-util/get_time_seconds.c b/src/libhirte/test/common/time-util/get_time_seconds.c
new file mode 100644
index 0000000000..c9ee8007dc
--- /dev/null
+++ b/src/libhirte/test/common/time-util/get_time_seconds.c
@@ -0,0 +1,40 @@
+/* SPDX-License-Identifier: GPL-2.0-or-later */
+#include <assert.h>
+#include <stdbool.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <time.h>
+
+#include "libhirte/common/common.h"
+#include "libhirte/common/string-util.h"
+#include "libhirte/common/time-util.h"
+
+bool test_get_time_seconds(time_t *input, int expected_return) {
+        int r = get_time_seconds(input);
+
+        if (r == expected_return) {
+                return true;
+        }
+
+        fprintf(stdout,
+                "FAILED: get_time_seconds() - Expected return value %d (%s), but got %d (%s)\n",
+                expected_return,
+                strerror(-expected_return),
+                r,
+                strerror(-r));
+        return false;
+}
+
+int main() {
+        bool result = true;
+
+        result = result && test_get_time_seconds(NULL, -EINVAL);
+
+        time_t ret = 0;
+        result = result && test_get_time_seconds(&ret, 0);
+
+        if (result) {
+                return EXIT_SUCCESS;
+        }
+        return EXIT_FAILURE;
+}
\ No newline at end of file
diff --git a/src/libhirte/test/common/time-util/meson.build b/src/libhirte/test/common/time-util/meson.build
new file mode 100644
index 0000000000..b81b5301c2
--- /dev/null
+++ b/src/libhirte/test/common/time-util/meson.build
@@ -0,0 +1,16 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+
+network_src = [
+  'get_log_timestamp',
+  'get_time_seconds',
+]
+
+foreach src : network_src
+  exec_test = executable(src, src + '.c',
+    link_with: [
+      hirte_lib,
+    ],
+    include_directories: include_directories('../../../..'),
+  )
+  test(src, exec_test)
+endforeach