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

Issue:
Agent should try to reconnect when connection goes down
The agent should detect when the connection to the manager gets broken and do something about it. This could be either retry, with exponential backoff, or just exit with an error, or a combination of the two. Possibly this should be a configuration option.


Discussion / hints:
Related, we should similarly retry connect on startup if the manager is not yet up.

Fix patch:
diff --git a/src/agent/agent.c b/src/agent/agent.c
index 7eadffa04d..5ee1840128 100644
--- a/src/agent/agent.c
+++ b/src/agent/agent.c
@@ -35,27 +35,43 @@ static bool
                                 void *userdata,
                                 free_func_t free_userdata);
 
-static int agent_reset_heartbeat_timer(Agent *agent, sd_event_source **event_source);
+static bool agent_connect(Agent *agent);
 
-static int agent_heartbeat_timer_callback(sd_event_source *event_source, UNUSED uint64_t usec, void *userdata) {
-        Agent *agent = userdata;
-        int r = 0;
+static int agent_disconnected(UNUSED sd_bus_message *message, void *userdata, UNUSED sd_bus_error *error) {
+        Agent *agent = (Agent *) userdata;
 
-        assert(event_source);
+        if (!agent_connect(agent)) {
+                hirte_log_errorf("Agent '%s' disconnected", agent->name);
+                agent->connection_state = AGENT_CONNECTION_STATE_RETRY;
+        }
 
-        r = sd_bus_emit_signal(
-                        agent->peer_dbus,
-                        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));
-                return r;
+        return 0;
+}
+
+static int agent_reset_heartbeat_timer(Agent *agent, sd_event_source **event_source);
+
+static int agent_heartbeat_timer_callback(sd_event_source *event_source, UNUSED uint64_t usec, void *userdata) {
+        Agent *agent = (Agent *) userdata;
+
+        if (agent->connection_state == AGENT_CONNECTION_STATE_CONNECTED) {
+                int r = sd_bus_emit_signal(
+                                agent->peer_dbus,
+                                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));
+                }
+        } else if (agent->connection_state == AGENT_CONNECTION_STATE_RETRY) {
+                agent->connection_retry_count++;
+                if (!agent_connect(agent)) {
+                        hirte_log_errorf("Agent '%s' disconnected", agent->name);
+                }
         }
 
-        r = agent_reset_heartbeat_timer(agent, &event_source);
+        int r = agent_reset_heartbeat_timer(agent, &event_source);
         if (r < 0) {
                 hirte_log_errorf("Failed to reset agent heartbeat timer: %s", strerror(-r));
                 return r;
@@ -283,21 +299,24 @@ Agent *agent_new(void) {
                 return NULL;
         }
 
-        _cleanup_agent_ Agent *n = malloc0(sizeof(Agent));
-        n->ref_count = 1;
-        n->event = steal_pointer(&event);
-        n->api_bus_service_name = steal_pointer(&service_name);
-        n->port = HIRTE_DEFAULT_PORT;
-        LIST_HEAD_INIT(n->outstanding_requests);
-        LIST_HEAD_INIT(n->tracked_jobs);
+        _cleanup_agent_ Agent *agent = malloc0(sizeof(Agent));
+        agent->ref_count = 1;
+        agent->event = steal_pointer(&event);
+        agent->api_bus_service_name = steal_pointer(&service_name);
+        agent->port = HIRTE_DEFAULT_PORT;
+        LIST_HEAD_INIT(agent->outstanding_requests);
+        LIST_HEAD_INIT(agent->tracked_jobs);
 
-        n->unit_infos = hashmap_new(
+        agent->unit_infos = hashmap_new(
                         sizeof(AgentUnitInfo), 0, 0, 0, unit_info_hash, unit_info_compare, unit_info_clear, NULL);
-        if (n->unit_infos == NULL) {
+        if (agent->unit_infos == NULL) {
                 return NULL;
         }
 
-        return steal_pointer(&n);
+        agent->connection_state = AGENT_CONNECTION_STATE_DISCONNECTED;
+        agent->connection_retry_count = 0;
+
+        return steal_pointer(&agent);
 }
 
 Agent *agent_ref(Agent *agent) {
@@ -1472,13 +1491,56 @@ bool agent_start(Agent *agent) {
                 sd_bus_add_filter(agent->systemd_dbus, NULL, debug_systemd_message_handler, agent);
         }
 
+        r = shutdown_service_register(agent->api_bus, agent->event);
+        if (r < 0) {
+                hirte_log_errorf("Failed to register shutdown service: %s", strerror(-r));
+                return false;
+        }
+
+        r = event_loop_add_shutdown_signals(agent->event);
+        if (r < 0) {
+                hirte_log_errorf("Failed to add signals to agent event loop: %s", strerror(-r));
+                return false;
+        }
+
+        r = agent_setup_heartbeat_timer(agent);
+        if (r < 0) {
+                hirte_log_errorf("Failed to set up agent heartbeat timer: %s", strerror(-r));
+                return false;
+        }
+
+        if (!agent_connect(agent)) {
+                hirte_log_errorf("Agent '%s' disconnected", agent->name);
+                agent->connection_state = AGENT_CONNECTION_STATE_RETRY;
+        }
+
+        r = sd_event_loop(agent->event);
+        if (r < 0) {
+                hirte_log_errorf("Starting event loop failed: %s", strerror(-r));
+                return false;
+        }
+
+        return true;
+}
+
+bool agent_stop(Agent *agent) {
+        if (agent == NULL) {
+                return false;
+        }
+
+        return true;
+}
+
+static bool agent_connect(Agent *agent) {
+        peer_bus_close(agent->peer_dbus);
+
         agent->peer_dbus = peer_bus_open(agent->event, "peer-bus-to-orchestrator", agent->orch_addr);
         if (agent->peer_dbus == NULL) {
                 hirte_log_error("Failed to open peer dbus");
                 return false;
         }
 
-        r = bus_socket_set_no_delay(agent->peer_dbus);
+        int r = bus_socket_set_no_delay(agent->peer_dbus);
         if (r < 0) {
                 hirte_log_warn("Failed to set NO_DELAY on socket");
         }
@@ -1496,11 +1558,12 @@ bool agent_start(Agent *agent) {
                         internal_agent_vtable,
                         agent);
         if (r < 0) {
-                hirte_log_errorf("Failed to add manager vtable: %s", strerror(-r));
+                hirte_log_errorf("Failed to add agent vtable: %s", strerror(-r));
                 return false;
         }
 
-        _cleanup_sd_bus_message_ sd_bus_message *reg_m = NULL;
+        _cleanup_sd_bus_message_ sd_bus_message *bus_msg = NULL;
+        sd_bus_error error = SD_BUS_ERROR_NULL;
         r = sd_bus_call_method(
                         agent->peer_dbus,
                         HIRTE_DBUS_NAME,
@@ -1508,52 +1571,41 @@ bool agent_start(Agent *agent) {
                         INTERNAL_MANAGER_INTERFACE,
                         "Register",
                         &error,
-                        &reg_m,
+                        &bus_msg,
                         "s",
                         agent->name);
         if (r < 0) {
-                hirte_log_errorf("Failed to issue method call: %s", error.message);
+                hirte_log_errorf(
+                                "Agent '%s' connection attempt %d: failure",
+                                agent->name,
+                                agent->connection_retry_count);
                 sd_bus_error_free(&error);
                 return false;
         }
 
-        r = sd_bus_message_read(reg_m, "");
+        r = sd_bus_message_read(bus_msg, "");
         if (r < 0) {
                 hirte_log_errorf("Failed to parse response message: %s", strerror(-r));
                 return false;
         }
 
-        hirte_log_infof("Registered to '%s' as '%s'", agent->orch_addr, agent->name);
+        hirte_log_infof("Agent '%s' connection attempt %d: success", agent->name, agent->connection_retry_count);
 
-        r = shutdown_service_register(agent->api_bus, agent->event);
-        if (r < 0) {
-                hirte_log_errorf("Failed to register shutdown service: %s", strerror(-r));
-                return false;
-        }
+        agent->connection_state = AGENT_CONNECTION_STATE_CONNECTED;
+        agent->connection_retry_count = 0;
 
-        r = event_loop_add_shutdown_signals(agent->event);
-        if (r < 0) {
-                hirte_log_errorf("Failed to add signals to agent event loop: %s", strerror(-r));
-                return false;
-        }
-
-        r = agent_setup_heartbeat_timer(agent);
-        if (r < 0) {
-                hirte_log_errorf("Failed to set up agent heartbeat timer: %s", strerror(-r));
-                return false;
-        }
-
-        r = sd_event_loop(agent->event);
+        r = sd_bus_match_signal_async(
+                        agent->peer_dbus,
+                        NULL,
+                        "org.freedesktop.DBus.Local",
+                        "/org/freedesktop/DBus/Local",
+                        "org.freedesktop.DBus.Local",
+                        "Disconnected",
+                        agent_disconnected,
+                        NULL,
+                        agent);
         if (r < 0) {
-                hirte_log_errorf("Starting event loop failed: %s", strerror(-r));
-                return false;
-        }
-
-        return true;
-}
-
-bool agent_stop(Agent *agent) {
-        if (agent == NULL) {
+                hirte_log_errorf("Failed to request match for Disconnected signal: %s", strerror(-r));
                 return false;
         }
 
diff --git a/src/agent/agent.h b/src/agent/agent.h
index 1e23894214..e045b807be 100644
--- a/src/agent/agent.h
+++ b/src/agent/agent.h
@@ -35,6 +35,12 @@ DEFINE_CLEANUP_FUNC(SystemdRequest, systemd_request_unref)
 typedef void (*job_tracker_callback)(sd_bus_message *m, const char *result, void *userdata);
 typedef struct JobTracker JobTracker;
 
+typedef enum {
+        AGENT_CONNECTION_STATE_DISCONNECTED,
+        AGENT_CONNECTION_STATE_CONNECTED,
+        AGENT_CONNECTION_STATE_RETRY
+} AgentConnectionState;
+
 struct Agent {
         int ref_count;
 
@@ -42,6 +48,9 @@ struct Agent {
         char *host;
         int port;
 
+        AgentConnectionState connection_state;
+        uint64_t connection_retry_count;
+
         char *orch_addr;
         char *api_bus_service_name;
 
diff --git a/src/libhirte/bus/bus.c b/src/libhirte/bus/bus.c
index 0c2fe6bbf4..f080eed954 100644
--- a/src/libhirte/bus/bus.c
+++ b/src/libhirte/bus/bus.c
@@ -84,6 +84,20 @@ sd_bus *peer_bus_open(sd_event *event, const char *dbus_description, const char
         return steal_pointer(&dbus);
 }
 
+int peer_bus_close(sd_bus *peer_dbus) {
+        if (peer_dbus != NULL) {
+                int r = sd_bus_detach_event(peer_dbus);
+                if (r < 0) {
+                        hirte_log_errorf("Failed to detach bus from event: %s", strerror(-r));
+                        return r;
+                }
+
+                sd_bus_flush_close_unref(peer_dbus);
+        }
+
+        return 0;
+}
+
 /* This is just some helper to make the peer connections look like a bus for e.g busctl */
 static int method_peer_hello(sd_bus_message *m, UNUSED void *userdata, UNUSED sd_bus_error *ret_error) {
         /* Reply with the response */
diff --git a/src/libhirte/bus/bus.h b/src/libhirte/bus/bus.h
index 8fd7e1127d..f5ecc1ddc5 100644
--- a/src/libhirte/bus/bus.h
+++ b/src/libhirte/bus/bus.h
@@ -5,7 +5,8 @@
 
 char *assemble_tcp_address(const struct sockaddr_in *addr);
 sd_bus *peer_bus_open(sd_event *event, const char *dbus_description, const char *dbus_server_addr);
+int peer_bus_close(sd_bus *peer_dbus);
 sd_bus *peer_bus_open_server(sd_event *event, const char *dbus_description, const char *sender, int fd);
 sd_bus *system_bus_open(sd_event *event);
 sd_bus *systemd_bus_open(sd_event *event);
-sd_bus *user_bus_open(sd_event *event);
\ No newline at end of file
+sd_bus *user_bus_open(sd_event *event);
diff --git a/src/libhirte/bus/utils.c b/src/libhirte/bus/utils.c
index efe1789158..597706ffb3 100644
--- a/src/libhirte/bus/utils.c
+++ b/src/libhirte/bus/utils.c
@@ -6,10 +6,10 @@
 #include "utils.h"
 
 /* Number of seconds idle before sending keepalive packets */
-#define AGENT_KEEPALIVE_SOCKET_KEEPIDLE_SECS 10
+#define AGENT_KEEPALIVE_SOCKET_KEEPIDLE_SECS 1
 
 /* Number of seconds idle between each keepalive packet */
-#define AGENT_KEEPALIVE_SOCKET_KEEPINTVL_SECS 10
+#define AGENT_KEEPALIVE_SOCKET_KEEPINTVL_SECS 1
 
 int bus_parse_properties_foreach(sd_bus_message *m, bus_property_cb cb, void *userdata) {
         bool stop = false;
diff --git a/src/libhirte/common/protocol.h b/src/libhirte/common/protocol.h
index bd206e3e36..8272f4091d 100644
--- a/src/libhirte/common/protocol.h
+++ b/src/libhirte/common/protocol.h
@@ -81,7 +81,7 @@ UnitActiveState active_state_from_string(const char *s);
 
 /* Agent to Hirte heartbeat signals */
 
-// Application-level heartbeat set at 1 second.
-#define AGENT_HEARTBEAT_INTERVAL_USEC (1000000)
+// Application-level heartbeat set at 2 seconds.
+#define AGENT_HEARTBEAT_INTERVAL_USEC (2000000)
 
 #define AGENT_HEARTBEAT_SIGNAL_NAME "Heartbeat"
diff --git a/src/manager/node.c b/src/manager/node.c
index 9af98963ef..9f8113829d 100644
--- a/src/manager/node.c
+++ b/src/manager/node.c
@@ -371,11 +371,7 @@ static int node_match_heartbeat(UNUSED sd_bus_message *m, UNUSED void *userdata,
                 return 0;
         }
 
-        static bool first_heartbeat_received;
-        if (!first_heartbeat_received) {
-                hirte_log_infof("First heartbeat received from %s", node_name);
-                first_heartbeat_received = true;
-        }
+        hirte_log_debugf("Heartbeat received from %s", node_name);
 
         return 1;
 }


Test patch:
diff --git a/.github/workflows/e2e.yaml b/.github/workflows/e2e.yaml
index 34bda83b7a..f0c7658283 100644
--- a/.github/workflows/e2e.yaml
+++ b/.github/workflows/e2e.yaml
@@ -61,7 +61,7 @@ jobs:
           /hirte/example-hirte-agent.ini > bar.log 2>&1 &)"
           podman exec -it hirte-agent-bar /usr/bin/pkill --signal TERM hirte-agent
           podman exec -w /home/test-user -u test-user -it hirte-agent-bar \
-          grep "Registered.* as 'bar'" bar.log
+          grep "Agent 'bar' connection attempt 0: success" bar.log
 
       - name: run hirte agent-foo
         run: |
@@ -76,5 +76,4 @@ jobs:
           /hirte/example-hirte-agent.ini > foo.log 2>&1 &)"
           podman exec -it hirte-agent-foo /usr/bin/pkill --signal TERM hirte-agent
           podman exec -w /home/test-user -u test-user -it hirte-agent-foo \
-          grep "Registered.* as 'foo'" foo.log
-
+          grep "Agent 'foo' connection attempt 0: success" foo.log