19941333032

Committed 04 Dec 2025 07:27PM UTC coverage: 82.516% (-0.1%) from 82.632%

Build # 19941333032

Build Type

push

github

Committed by

engelmi

Commit Message

Remove test data with too long log configs

The test cases for too long log configs (level, target and isquiet) causes
the bluechi-agent service to fail (as expected). However, this happens in
the test setup phase where we expect a running agent. Due to timing issues
this check might even pass, depending how fast the status is being checked.
This causes the test to be flacky.
Since such a long input for log configs is unrealistic and the proper level
for testing is on the unit level (test without the long data is already in
place), these cases have een removed.
In addition, the unused BluechiConfig variable for max line length has been
removed.

Signed-off-by: Michael Engel <mengel@redhat.com>

Run Details

5687 of 6892 relevant lines covered (82.52%)

1865.63 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.91

/src/controller/node.c

/*
 * Copyright Contributors to the Eclipse BlueChi project
 *
 * SPDX-License-Identifier: LGPL-2.1-or-later
 */
#include <systemd/sd-bus.h>

#ifdef CONFIG_H_USE_SELINUX
#        include <selinux/selinux.h>
#endif

#include "libbluechi/bus/bus.h"
#include "libbluechi/bus/utils.h"
#include "libbluechi/common/parse-util.h"
#include "libbluechi/common/time-util.h"
#include "libbluechi/log/log.h"

#include "controller.h"
#include "job.h"
#include "metrics.h"
#include "monitor.h"
#include "node.h"
#include "proxy_monitor.h"

#define DEBUG_AGENT_MESSAGES 0

static void node_send_agent_subscribe_all(Node *node);
static void node_start_proxy_dependency_all(Node *node);
static int node_run_unit_lifecycle_method(sd_bus_message *m, Node *node, const char *job_type, const char *method);

static int node_method_register(sd_bus_message *m, void *userdata, sd_bus_error *ret_error);
static int node_disconnected(sd_bus_message *message, void *userdata, sd_bus_error *error);
static int node_method_list_units(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error);
static int node_method_list_unit_files(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error);
static int node_method_set_unit_properties(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error);
static int node_method_start_unit(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error);
static int node_method_stop_unit(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error);
static int node_method_restart_unit(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error);
static int node_method_reload_unit(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error);
static int node_method_passthrough_to_agent(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error);
static int node_method_set_log_level(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error);
static int node_property_get_status(
                sd_bus *bus,
                const char *path,
                const char *interface,
                const char *property,
                sd_bus_message *reply,
                void *userdata,
                sd_bus_error *ret_error);
static int node_property_get_peer_ip(
                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_controller_controller_vtable[] = {
        SD_BUS_VTABLE_START(0),
        SD_BUS_METHOD("Register", "s", "", node_method_register, 0),
        SD_BUS_SIGNAL("Heartbeat", "", 0),
        SD_BUS_VTABLE_END
};

static const sd_bus_vtable node_vtable[] = {
        SD_BUS_VTABLE_START(0),
        SD_BUS_METHOD("ListUnits", "", UNIT_INFO_STRUCT_ARRAY_TYPESTRING, node_method_list_units, 0),
        SD_BUS_METHOD("ListUnitFiles", "", UNIT_FILE_INFO_STRUCT_ARRAY_TYPESTRING, node_method_list_unit_files, 0),
        SD_BUS_METHOD("GetUnitFileState", "s", "s", node_method_passthrough_to_agent, 0),
        SD_BUS_METHOD("StartUnit", "ss", "o", node_method_start_unit, 0),
        SD_BUS_METHOD("StartTransientUnit", "ssa(sv)a(sa(sv))", "o", node_method_passthrough_to_agent, 0),
        SD_BUS_METHOD("StopUnit", "ss", "o", node_method_stop_unit, 0),
        SD_BUS_METHOD("FreezeUnit", "s", "", node_method_passthrough_to_agent, 0),
        SD_BUS_METHOD("ThawUnit", "s", "", node_method_passthrough_to_agent, 0),
        SD_BUS_METHOD("RestartUnit", "ss", "o", node_method_restart_unit, 0),
        SD_BUS_METHOD("ReloadUnit", "ss", "o", node_method_reload_unit, 0),
        SD_BUS_METHOD("ResetFailed", "", "", node_method_passthrough_to_agent, 0),
        SD_BUS_METHOD("ResetFailedUnit", "s", "", node_method_passthrough_to_agent, 0),
        SD_BUS_METHOD("GetUnitProperties", "ss", "a{sv}", node_method_passthrough_to_agent, 0),
        SD_BUS_METHOD("GetUnitProperty", "sss", "v", node_method_passthrough_to_agent, 0),
        SD_BUS_METHOD("SetUnitProperties", "sba(sv)", "", node_method_set_unit_properties, 0),
        SD_BUS_METHOD("EnableUnitFiles", "asbb", "ba(sss)", node_method_passthrough_to_agent, 0),
        SD_BUS_METHOD("DisableUnitFiles", "asb", "a(sss)", node_method_passthrough_to_agent, 0),
        SD_BUS_METHOD("Reload", "", "", node_method_passthrough_to_agent, 0),
        SD_BUS_METHOD("KillUnit", "ssi", "", node_method_passthrough_to_agent, 0),
        SD_BUS_METHOD("SetLogLevel", "s", "", node_method_set_log_level, 0),
        SD_BUS_METHOD("GetDefaultTarget", "", "s", node_method_passthrough_to_agent, 0),
        SD_BUS_METHOD("SetDefaultTarget", "sb", "a(sss)", node_method_passthrough_to_agent, 0),
        SD_BUS_PROPERTY("Name", "s", NULL, offsetof(Node, name), SD_BUS_VTABLE_PROPERTY_CONST),
        SD_BUS_PROPERTY("Status", "s", node_property_get_status, 0, SD_BUS_VTABLE_PROPERTY_EMITS_CHANGE),
        SD_BUS_PROPERTY("PeerIp", "s", node_property_get_peer_ip, 0, SD_BUS_VTABLE_PROPERTY_EXPLICIT),
        SD_BUS_PROPERTY("LastSeenTimestamp", "t", NULL, offsetof(Node, last_seen), SD_BUS_VTABLE_PROPERTY_EXPLICIT),
        SD_BUS_PROPERTY("LastSeenTimestampMonotonic",
                        "t",
                        NULL,
                        offsetof(Node, last_seen_monotonic),
                        SD_BUS_VTABLE_PROPERTY_EXPLICIT),
        SD_BUS_VTABLE_END
};

struct ProxyDependency {
        char *unit_name;
        int n_deps;
        LIST_FIELDS(ProxyDependency, deps);
};

static void proxy_dependency_free(struct ProxyDependency *dep) {
        free_and_null(dep->unit_name);
        free_and_null(dep);
}

struct ProxyTarget {
        char *target_name;
        LIST_FIELDS(ProxyTarget, allowed_targets);
};

static void proxy_target_free(struct ProxyTarget *target) {
        free_and_null(target->target_name);
        free_and_null(target);
}

typedef struct UnitSubscription UnitSubscription;

struct UnitSubscription {
        Subscription *sub;
        LIST_FIELDS(UnitSubscription, subs);
};

typedef struct {
        char *unit;
        LIST_HEAD(UnitSubscription, subs);
        bool loaded;
        UnitActiveState active_state;
        char *substate;
} UnitSubscriptions;

typedef struct {
        char *unit;
} UnitSubscriptionsKey;

static void unit_subscriptions_clear(void *item) {
        UnitSubscriptions *usubs = item;
        free_and_null(usubs->unit);
        free_and_null(usubs->substate);
        assert(LIST_IS_EMPTY(usubs->subs));
}

static uint64_t unit_subscriptions_hash(const void *item, uint64_t seed0, uint64_t seed1) {
        const UnitSubscriptions *usubs = item;
        return hashmap_sip(usubs->unit, strlen(usubs->unit), seed0, seed1);
}

static int unit_subscriptions_compare(const void *a, const void *b, UNUSED void *udata) {
        const UnitSubscriptions *usubs_a = a;
        const UnitSubscriptions *usubs_b = b;

        return strcmp(usubs_a->unit, usubs_b->unit);
}


Node *node_new(Controller *controller, const char *name) {
        _cleanup_node_ Node *node = malloc0(sizeof(Node));
        if (node == NULL) {
                return NULL;
        }

        node->ref_count = 1;
        node->controller = controller;
        LIST_INIT(nodes, node);
        LIST_HEAD_INIT(node->outstanding_requests);
        LIST_HEAD_INIT(node->proxy_monitors);
        LIST_HEAD_INIT(node->proxy_dependencies);
        LIST_HEAD_INIT(node->allowed_proxy_targets);

        node->unit_subscriptions = hashmap_new(
                        sizeof(UnitSubscriptions),
                        0,
                        0,
                        0,
                        unit_subscriptions_hash,
                        unit_subscriptions_compare,
                        unit_subscriptions_clear,
                        NULL);
        if (node->unit_subscriptions == NULL) {
                return NULL;
        }

        node->last_seen = 0;
        node->last_seen_monotonic = 0;

        node->name = NULL;
        if (name) {
                node->name = strdup(name);
                if (node->name == NULL) {
                        return NULL;
                }

                int r = assemble_object_path_string(NODE_OBJECT_PATH_PREFIX, name, &node->object_path);
                if (r < 0) {
                        return NULL;
                }
        }
        node->peer_ip = NULL;
        node->peer_selinux_context = NULL;

        node->is_shutdown = false;

        return steal_pointer(&node);
}

Node *node_ref(Node *node) {
        node->ref_count++;
        return node;
}

void node_unref(Node *node) {
        node->ref_count--;
        if (node->ref_count != 0) {
                return;
        }

        ProxyMonitor *proxy_monitor = NULL;
        ProxyMonitor *next_proxy_monitor = NULL;
        LIST_FOREACH_SAFE(monitors, proxy_monitor, next_proxy_monitor, node->proxy_monitors) {
                node_remove_proxy_monitor(node, proxy_monitor);
        }


        ProxyDependency *dep = NULL;
        ProxyDependency *next_dep = NULL;
        LIST_FOREACH_SAFE(deps, dep, next_dep, node->proxy_dependencies) {
                proxy_dependency_free(dep);
        }

        ProxyTarget *target = NULL;
        ProxyTarget *next_target = NULL;
        LIST_FOREACH_SAFE(allowed_targets, target, next_target, node->allowed_proxy_targets) {
                proxy_target_free(target);
        }

        node_unset_agent_bus(node);
        sd_bus_slot_unrefp(&node->export_slot);

        hashmap_free(node->unit_subscriptions);

        free_and_null(node->name);
        free_and_null(node->object_path);
        free_and_null(node->peer_ip);
        free_and_null(node->required_selinux_context);
        free(node);
}

void node_shutdown(Node *node) {
        AgentRequest *req = NULL;
        AgentRequest *next_req = NULL;
        node->is_shutdown = true;
        LIST_FOREACH_SAFE(outstanding_requests, req, next_req, node->outstanding_requests) {
                agent_request_cancel(req);
        }
}

bool node_set_required_selinux_context(Node *node, const char *selinux_context) {
        node->required_selinux_context = strdup(selinux_context);
        if (node->required_selinux_context == NULL) {
                return false;
        }
        return true;
}

int node_add_allowed_proxy_target(Node *node, const char *target_name) {
        ProxyTarget *target = NULL;

        _cleanup_free_ char *target_name_copy = strdup(target_name);
        if (target_name_copy == NULL) {
                return -ENOMEM;
        }

        target = malloc0(sizeof(ProxyTarget));
        if (target == NULL) {
                return -ENOMEM;
        }

        target->target_name = steal_pointer(&target_name_copy);
        LIST_APPEND(allowed_targets, node->allowed_proxy_targets, target);

        return 0;
}

static ProxyTarget *node_find_allowed_proxy_target(Node *node, const char *name) {
        ProxyTarget *curr = NULL;
        ProxyTarget *next = NULL;
        LIST_FOREACH_SAFE(allowed_targets, curr, next, node->allowed_proxy_targets) {
                if (streq(curr->target_name, name)) {
                        return curr;
                }
        }
        return NULL;
}

bool node_export(Node *node) {
        Controller *controller = node->controller;

        assert(node->name != NULL);

        int r = sd_bus_add_object_vtable(
                        controller->api_bus,
                        &node->export_slot,
                        node->object_path,
                        NODE_INTERFACE,
                        node_vtable,
                        node);
        if (r < 0) {
                bc_log_errorf("Failed to add node vtable: %s", strerror(-r));
                return false;
        }

        return true;
}

static int debug_messages_handler(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error) {
        Node *node = userdata;
        if (node->name) {
                bc_log_infof("Incoming message from node '%s' (fd %d): path: %s, iface: %s, member: %s, signature: '%s'",
                             node->name,
                             sd_bus_get_fd(node->agent_bus),
                             sd_bus_message_get_path(m),
                             sd_bus_message_get_interface(m),
                             sd_bus_message_get_member(m),
                             sd_bus_message_get_signature(m, true));
        } else {
                bc_log_infof("Incoming message from node fd %d: path: %s, iface: %s, member: %s, signature: '%s'",
                             sd_bus_get_fd(node->agent_bus),
                             sd_bus_message_get_path(m),
                             sd_bus_message_get_interface(m),
                             sd_bus_message_get_member(m),
                             sd_bus_message_get_signature(m, true));
        }
        return 0;
}

bool node_has_agent(Node *node) {
        return node->agent_bus != NULL;
}

bool node_is_online(Node *node) {
        return node && node->name && node_has_agent(node);
}


static uint64_t subscription_hashmap_hash(const void *item, UNUSED uint64_t seed0, UNUSED uint64_t seed1) {
        const Subscription * const *subscriptionp = item;
        return (uint64_t) ((uintptr_t) *subscriptionp);
}

static int subscription_hashmap_compare(const void *a, const void *b, UNUSED void *udata) {
        const Subscription * const *subscription_a_p = a;
        const Subscription * const *subscription_b_p = b;
        if ((*subscription_a_p)->monitor == (*subscription_b_p)->monitor) {
                return 0;
        }
        return 1;
}

static struct hashmap *node_compute_unique_monitor_subscriptions(Node *node, const char *unit) {
        struct hashmap *unique_subs = hashmap_new(
                        sizeof(void *), 0, 0, 0, subscription_hashmap_hash, subscription_hashmap_compare, NULL, NULL);
        if (unique_subs == NULL) {
                return NULL;
        }

        const UnitSubscriptionsKey key = { (char *) unit };
        const UnitSubscriptions *usubs = hashmap_get(node->unit_subscriptions, &key);
        if (usubs != NULL) {
                UnitSubscription *usub = NULL;
                UnitSubscription *next_usub = NULL;
                LIST_FOREACH_SAFE(subs, usub, next_usub, usubs->subs) {
                        Subscription *sub = usub->sub;
                        hashmap_set(unique_subs, &sub);
                        if (hashmap_oom(unique_subs)) {
                                bc_log_error("Failed to compute vector of unique monitors, OOM");

                                hashmap_free(unique_subs);
                                unique_subs = NULL;
                                return NULL;
                        }
                }
        }

        /* Only check for wildcards if the unit itself is not one. */
        if (!streq(unit, SYMBOL_WILDCARD)) {
                const UnitSubscriptionsKey wildcard_key = { (char *) SYMBOL_WILDCARD };
                const UnitSubscriptions *usubs_wildcard = hashmap_get(node->unit_subscriptions, &wildcard_key);
                if (usubs_wildcard != NULL) {
                        UnitSubscription *usub = NULL;
                        UnitSubscription *next_usub = NULL;
                        LIST_FOREACH_SAFE(subs, usub, next_usub, usubs_wildcard->subs) {
                                Subscription *sub = usub->sub;
                                hashmap_set(unique_subs, &sub);
                                if (hashmap_oom(unique_subs)) {
                                        bc_log_error("Failed to compute vector of unique monitors, OOM");

                                        hashmap_free(unique_subs);
                                        unique_subs = NULL;
                                        return NULL;
                                }
                        }
                }
        }

        return unique_subs;
}


static int node_match_job_state_changed(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *error) {
        Node *node = userdata;
        Controller *controller = node->controller;
        uint32_t bc_job_id = 0;
        const char *state = NULL;

        int r = sd_bus_message_read(m, "us", &bc_job_id, &state);
        if (r < 0) {
                bc_log_errorf("Invalid JobStateChange signal: %s", strerror(-r));
                return 0;
        }

        controller_job_state_changed(controller, bc_job_id, state);
        return 1;
}

static int node_match_unit_properties_changed(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *error) {
        Node *node = userdata;
        const char *unit = NULL;
        const char *interface = NULL;

        int r = sd_bus_message_read(m, "ss", &unit, &interface);
        if (r >= 0) {
                r = sd_bus_message_rewind(m, false);
        }
        if (r < 0) {
                bc_log_error("Invalid UnitPropertiesChanged signal");
                return 0;
        }

        struct hashmap *unique_subs = node_compute_unique_monitor_subscriptions(node, unit);
        if (unique_subs != NULL) {
                Subscription **subp = NULL;
                size_t i = 0;
                while (hashmap_iter(unique_subs, &i, (void **) &subp)) {
                        Subscription *sub = *subp;
                        int r = sub->handle_unit_property_changed(sub->monitor, node->name, unit, interface, m);
                        if (r < 0) {
                                bc_log_error("Failed to emit UnitPropertyChanged signal");
                        }
                }
                hashmap_free(unique_subs);
        }

        return 1;
}


static int node_match_unit_new(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *error) {
        Node *node = userdata;
        const char *unit = NULL;
        const char *reason = NULL;

        int r = sd_bus_message_read(m, "ss", &unit, &reason);
        if (r < 0) {
                bc_log_errorf("Invalid UnitNew signal: %s", strerror(-r));
                return 0;
        }

        const UnitSubscriptionsKey key = { (char *) unit };
        UnitSubscriptions *usubs = (UnitSubscriptions *) hashmap_get(node->unit_subscriptions, &key);
        if (usubs != NULL) {
                usubs->loaded = true;
                if (is_wildcard(unit)) {
                        usubs->active_state = UNIT_ACTIVE;
                        free(usubs->substate);
                        usubs->substate = strdup("running");
                }
        }

        struct hashmap *unique_subs = node_compute_unique_monitor_subscriptions(node, unit);
        if (unique_subs != NULL) {
                Subscription **subp = NULL;
                size_t i = 0;
                while (hashmap_iter(unique_subs, &i, (void **) &subp)) {
                        Subscription *sub = *subp;
                        int r = sub->handle_unit_new(sub->monitor, node->name, unit, reason);
                        if (r < 0) {
                                bc_log_errorf("Failed to emit UnitNew signal: %s", strerror(-r));
                        }
                }
                hashmap_free(unique_subs);
        }

        return 1;
}

static int node_match_unit_state_changed(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *error) {
        Node *node = userdata;
        const char *unit = NULL;
        const char *active_state = NULL;
        const char *substate = NULL;
        const char *reason = NULL;

        int r = sd_bus_message_read(m, "ssss", &unit, &active_state, &substate, &reason);
        if (r < 0) {
                bc_log_errorf("Invalid UnitStateChanged signal: %s", strerror(-r));
                return 0;
        }

        const UnitSubscriptionsKey key = { (char *) unit };
        UnitSubscriptions *usubs = (UnitSubscriptions *) hashmap_get(node->unit_subscriptions, &key);
        if (usubs != NULL) {
                usubs->loaded = true;
                usubs->active_state = active_state_from_string(active_state);
                free(usubs->substate);
                usubs->substate = strdup(substate);
        }

        struct hashmap *unique_subs = node_compute_unique_monitor_subscriptions(node, unit);
        if (unique_subs != NULL) {
                Subscription **subp = NULL;
                size_t i = 0;
                while (hashmap_iter(unique_subs, &i, (void **) &subp)) {
                        Subscription *sub = *subp;
                        int r = sub->handle_unit_state_changed(
                                        sub->monitor, node->name, unit, active_state, substate, reason);
                        if (r < 0) {
                                bc_log_errorf("Failed to emit UnitStateChanged signal: %s", strerror(-r));
                        }
                }
                hashmap_free(unique_subs);
        }

        return 1;
}

static int node_match_unit_removed(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *error) {
        Node *node = userdata;
        const char *unit = NULL;

        int r = sd_bus_message_read(m, "s", &unit);
        if (r < 0) {
                bc_log_errorf("Invalid UnitRemoved signal: %s", strerror(-r));
                return 0;
        }

        const UnitSubscriptionsKey key = { (char *) unit };
        UnitSubscriptions *usubs = (UnitSubscriptions *) hashmap_get(node->unit_subscriptions, &key);
        if (usubs != NULL) {
                usubs->loaded = false;
        }

        struct hashmap *unique_subs = node_compute_unique_monitor_subscriptions(node, unit);
        if (unique_subs != NULL) {
                Subscription **subp = NULL;
                size_t i = 0;
                while (hashmap_iter(unique_subs, &i, (void **) &subp)) {
                        Subscription *sub = *subp;
                        int r = sub->handle_unit_removed(sub->monitor, node->name, unit, "real");
                        if (r < 0) {
                                bc_log_errorf("Failed to emit UnitRemoved signal: %s", strerror(-r));
                        }
                }
                hashmap_free(unique_subs);
        }

        return 1;
}

static int node_match_job_done(UNUSED sd_bus_message *m, UNUSED void *userdata, UNUSED sd_bus_error *error) {
        Node *node = userdata;
        Controller *controller = node->controller;
        uint32_t bc_job_id = 0;
        const char *result = NULL;

        int r = sd_bus_message_read(m, "us", &bc_job_id, &result);
        if (r < 0) {
                bc_log_errorf("Invalid JobDone signal: %s", strerror(-r));
                return 0;
        }

        controller_finish_job(controller, bc_job_id, result);
        return 1;
}

static int node_match_heartbeat(UNUSED sd_bus_message *m, void *userdata, UNUSED sd_bus_error *error) {
        Node *node = userdata;
        uint64_t now = 0;
        uint64_t now_monotonic = 0;

        now = get_time_micros();
        if (now == USEC_INFINITY) {
                bc_log_error("Failed to get current time on heartbeat");
                return 0;
        }

        now_monotonic = get_time_micros_monotonic();
        if (now_monotonic == USEC_INFINITY) {
                bc_log_error("Failed to get current monotonic time on heartbeat");
                return 0;
        }

        node->last_seen = now;
        node->last_seen_monotonic = now_monotonic;
        return 1;
}

static ProxyMonitor *node_find_proxy_monitor(Node *node, const char *target_node_name, const char *unit_name) {
        ProxyMonitor *proxy_monitor = NULL;
        LIST_FOREACH(monitors, proxy_monitor, node->proxy_monitors) {
                if (streq(proxy_monitor->target_node->name, target_node_name) &&
                    streq(proxy_monitor->unit_name, unit_name)) {
                        return proxy_monitor;
                }
        }

        return NULL;
}

static int node_on_match_proxy_new(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *error) {
        Node *node = userdata;
        Controller *controller = node->controller;

        const char *target_node_name = NULL;
        const char *unit_name = NULL;
        const char *proxy_object_path = NULL;

        int r = sd_bus_message_read(m, "sso", &target_node_name, &unit_name, &proxy_object_path);
        if (r < 0) {
                bc_log_errorf("Invalid arguments in ProxyNew signal: %s", strerror(-r));
                return r;
        }

        bc_log_infof("Node '%s' registered new proxy for unit '%s' on node '%s'",
                     node->name,
                     unit_name,
                     target_node_name);

        _cleanup_proxy_monitor_ ProxyMonitor *monitor = proxy_monitor_new(
                        node, target_node_name, unit_name, proxy_object_path);
        if (monitor == NULL) {
                bc_log_error("Failed to create proxy monitor, OOM");
                return -ENOMEM;
        }

        Node *target_node = controller_find_node(controller, target_node_name);
        if (target_node == NULL) {
                bc_log_error("Proxy requested for non-existing node");
                proxy_monitor_send_error(monitor, "No such node");
                return 0;
        }

        ProxyTarget *target = node_find_allowed_proxy_target(node, target_node_name);
        if (target == NULL) {
                bc_log_errorf("Proxy request denied for %s->%s", node->name, target_node_name);
                proxy_monitor_send_error(monitor, "Proxy on node not allowed");
                return 0;
        }

        ProxyMonitor *old_monitor = node_find_proxy_monitor(node, target_node_name, unit_name);
        if (old_monitor != NULL) {
                bc_log_warnf("Proxy for '%s' (on '%s') requested, but old proxy already exists, removing it",
                             unit_name,
                             target_node_name);
                node_remove_proxy_monitor(node, old_monitor);
        }

        r = proxy_monitor_set_target_node(monitor, target_node);
        if (r < 0) {
                bc_log_errorf("Failed to add proxy dependency: %s", strerror(-r));
                proxy_monitor_send_error(monitor, "Failed to add proxy dependency");
                return 0;
        }

        /* We now have a valid monitor, add it to the list and enable monitor.
           From this point we should not send errors. */
        controller_add_subscription(controller, monitor->subscription);
        LIST_APPEND(monitors, node->proxy_monitors, proxy_monitor_ref(monitor));

        /* TODO: What about !!node_is_online(target_node) ? Tell now or wait for it to connect? */

        return 0;
}

void node_remove_proxy_monitor(Node *node, ProxyMonitor *monitor) {
        Controller *controller = node->controller;

        proxy_monitor_close(monitor);

        controller_remove_subscription(controller, monitor->subscription);
        LIST_REMOVE(monitors, node->proxy_monitors, monitor);

        proxy_monitor_unref(monitor);
}

static int node_on_match_proxy_removed(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *error) {
        Node *node = userdata;
        const char *target_node_name = NULL;
        const char *unit_name = NULL;

        int r = sd_bus_message_read(m, "ss", &target_node_name, &unit_name);
        if (r < 0) {
                bc_log_errorf("Invalid arguments in ProxyRemoved signal: %s", strerror(-r));
                return r;
        }
        bc_log_infof("Node '%s' unregistered proxy for unit '%s' on node '%s'",
                     node->name,
                     unit_name,
                     target_node_name);

        ProxyMonitor *proxy_monitor = node_find_proxy_monitor(node, target_node_name, unit_name);
        if (proxy_monitor == NULL) {
                bc_log_error("Got ProxyRemoved for unknown monitor");
                return 0;
        }

        node_remove_proxy_monitor(node, proxy_monitor);
        return 0;
}

bool node_set_agent_bus(Node *node, sd_bus *bus) {
        int r = 0;

        if (node->agent_bus != NULL) {
                bc_log_error("Error: Trying to add two agents for a node");
                return false;
        }

        node->agent_bus = sd_bus_ref(bus);

        // If getting peer IP fails, only log and proceed as normal.
        _cleanup_free_ char *peer_ip = NULL;
        uint16_t peer_port = 0;
        r = get_peer_ip_address(node->agent_bus, &peer_ip, &peer_port);
        if (r < 0 && r != -EINVAL) {
                bc_log_errorf("Failed to get peer IP: %s", strerror(-r));
        } else {
                node->peer_ip = steal_pointer(&peer_ip);
        }

#ifdef CONFIG_H_USE_SELINUX
        char *peercon = NULL;
        if (getpeercon(sd_bus_get_fd(bus), &peercon) == 0) {
                node->peer_selinux_context = parse_selinux_type(peercon);
                if (node->peer_selinux_context == NULL) {
                        bc_log_errorf("Failed to parse peer selinux type '%s'", peercon);
                }
                freecon(peercon);
        }
#endif

        if (node->name == NULL) {
                // We only connect to this on the unnamed nodes so register
                // can be called. We can't reconnect it during migration.
                r = sd_bus_add_object_vtable(
                                bus,
                                &node->internal_controller_slot,
                                INTERNAL_CONTROLLER_OBJECT_PATH,
                                INTERNAL_CONTROLLER_INTERFACE,
                                internal_controller_controller_vtable,
                                node);
                if (r < 0) {
                        node_unset_agent_bus(node);
                        bc_log_errorf("Failed to add peer bus vtable: %s", strerror(-r));
                        return false;
                }
        } else {
                // Only listen to signals on named nodes
                r = sd_bus_match_signal(
                                bus,
                                NULL,
                                NULL,
                                INTERNAL_AGENT_OBJECT_PATH,
                                INTERNAL_AGENT_INTERFACE,
                                "JobDone",
                                node_match_job_done,
                                node);
                if (r < 0) {
                        bc_log_errorf("Failed to add JobDone peer bus match: %s", strerror(-r));
                        return false;
                }

                r = sd_bus_match_signal(
                                bus,
                                NULL,
                                NULL,
                                INTERNAL_AGENT_OBJECT_PATH,
                                INTERNAL_AGENT_INTERFACE,
                                "JobStateChanged",
                                node_match_job_state_changed,
                                node);
                if (r < 0) {
                        bc_log_errorf("Failed to add JobStateChanged peer bus match: %s", strerror(-r));
                        return false;
                }

                r = sd_bus_match_signal(
                                bus,
                                NULL,
                                NULL,
                                INTERNAL_AGENT_OBJECT_PATH,
                                INTERNAL_AGENT_INTERFACE,
                                "UnitPropertiesChanged",
                                node_match_unit_properties_changed,
                                node);
                if (r < 0) {
                        bc_log_errorf("Failed to add UnitPropertiesChanged peer bus match: %s", strerror(-r));
                        return false;
                }

                r = sd_bus_match_signal(
                                bus,
                                NULL,
                                NULL,
                                INTERNAL_AGENT_OBJECT_PATH,
                                INTERNAL_AGENT_INTERFACE,
                                "UnitNew",
                                node_match_unit_new,
                                node);
                if (r < 0) {
                        bc_log_errorf("Failed to add UnitNew peer bus match: %s", strerror(-r));
                        return false;
                }

                r = sd_bus_match_signal(
                                bus,
                                NULL,
                                NULL,
                                INTERNAL_AGENT_OBJECT_PATH,
                                INTERNAL_AGENT_INTERFACE,
                                "UnitStateChanged",
                                node_match_unit_state_changed,
                                node);
                if (r < 0) {
                        bc_log_errorf("Failed to add UnitStateChanged peer bus match: %s", strerror(-r));
                        return false;
                }

                r = sd_bus_match_signal(
                                bus,
                                NULL,
                                NULL,
                                INTERNAL_AGENT_OBJECT_PATH,
                                INTERNAL_AGENT_INTERFACE,
                                "UnitRemoved",
                                node_match_unit_removed,
                                node);
                if (r < 0) {
                        bc_log_errorf("Failed to add UnitRemoved peer bus match: %s", strerror(-r));
                        return false;
                }

                r = sd_bus_match_signal(
                                bus,
                                NULL,
                                NULL,
                                INTERNAL_AGENT_OBJECT_PATH,
                                INTERNAL_AGENT_INTERFACE,
                                "ProxyNew",
                                node_on_match_proxy_new,
                                node);
                if (r < 0) {
                        bc_log_errorf("Failed to add ProxyNew peer bus match: %s", strerror(-r));
                        return false;
                }

                r = sd_bus_match_signal(
                                bus,
                                NULL,
                                NULL,
                                INTERNAL_AGENT_OBJECT_PATH,
                                INTERNAL_AGENT_INTERFACE,
                                "ProxyRemoved",
                                node_on_match_proxy_removed,
                                node);
                if (r < 0) {
                        bc_log_errorf("Failed to add ProxyNew peer bus match: %s", strerror(-r));
                        return false;
                }

                r = sd_bus_emit_properties_changed(
                                node->controller->api_bus, node->object_path, NODE_INTERFACE, "Status", NULL);
                if (r < 0) {
                        bc_log_errorf("Failed to emit status property changed: %s", strerror(-r));
                }

                r = sd_bus_match_signal(
                                bus,
                                NULL,
                                NULL,
                                INTERNAL_AGENT_OBJECT_PATH,
                                INTERNAL_AGENT_INTERFACE,
                                AGENT_HEARTBEAT_SIGNAL_NAME,
                                node_match_heartbeat,
                                node);
                if (r < 0) {
                        bc_log_errorf("Failed to add heartbeat signal match: %s", strerror(-r));
                        return false;
                }
        }

        r = sd_bus_match_signal_async(
                        bus,
                        &node->disconnect_slot,
                        "org.freedesktop.DBus.Local",
                        "/org/freedesktop/DBus/Local",
                        "org.freedesktop.DBus.Local",
                        "Disconnected",
                        node_disconnected,
                        NULL,
                        node);
        if (r < 0) {
                node_unset_agent_bus(node);
                bc_log_errorf("Failed to request match for Disconnected message: %s", strerror(-r));
                return false;
        }

        if (DEBUG_AGENT_MESSAGES) {
                sd_bus_add_filter(bus, NULL, debug_messages_handler, node);
        }


        /* Register any active subscriptions with new agent */
        node_send_agent_subscribe_all(node);

        /* Register any active dependencies with new agent */
        node_start_proxy_dependency_all(node);

        return true;
}

void node_unset_agent_bus(Node *node) {
        bool was_online = node->name && node_has_agent(node);

        sd_bus_slot_unrefp(&node->disconnect_slot);
        node->disconnect_slot = NULL;

        sd_bus_slot_unrefp(&node->internal_controller_slot);
        node->internal_controller_slot = NULL;

        sd_bus_slot_unrefp(&node->metrics_matching_slot);
        node->metrics_matching_slot = NULL;

        sd_bus_unrefp(&node->agent_bus);
        node->agent_bus = NULL;

        free_and_null(node->peer_selinux_context);
        free_and_null(node->peer_ip);

        if (was_online) {
                int r = sd_bus_emit_properties_changed(
                                node->controller->api_bus, node->object_path, NODE_INTERFACE, "Status", NULL);
                if (r < 0) {
                        bc_log_errorf("Failed to emit status property changed: %s", strerror(-r));
                }
        }
}

/* org.eclipse.bluechi.internal.Controller.Register(in s name)) */
static int node_method_register(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error) {
        Node *node = userdata;
        Controller *controller = node->controller;
        char *name = NULL;
        _cleanup_free_ char *description = NULL;

        /* Once we're not anonymous, don't allow register calls */
        if (node->name != NULL) {
                return sd_bus_reply_method_errorf(m, SD_BUS_ERROR_ADDRESS_IN_USE, "Can't register twice");
        }

        /* Read the parameters */
        int r = sd_bus_message_read(m, "s", &name);
        if (r < 0) {
                bc_log_errorf("Failed to parse parameters: %s", strerror(-r));
                return r;
        }

        Node *named_node = controller_find_node(controller, name);
        if (named_node == NULL) {
                return sd_bus_reply_method_errorf(m, SD_BUS_ERROR_SERVICE_UNKNOWN, "Unexpected node name");
        }

        if (named_node->required_selinux_context &&
            (node->peer_selinux_context == NULL ||
             !streq(node->peer_selinux_context, named_node->required_selinux_context))) {
                bc_log_errorf("Node tried to register as '%s' with wrong selinux context '%s', expected '%s'",
                              name,
                              node->peer_selinux_context ? node->peer_selinux_context : "(missing)",
                              named_node->required_selinux_context);

                return sd_bus_reply_method_errorf(m, SD_BUS_ERROR_ACCESS_DENIED, "Node name not allowed");
        }

        if (node_has_agent(named_node)) {
                return sd_bus_reply_method_errorf(
                                m, SD_BUS_ERROR_ADDRESS_IN_USE, "The node is already connected");
        }

        named_node->last_seen = get_time_micros();
        named_node->last_seen_monotonic = get_time_micros_monotonic();

        r = asprintf(&description, "node-%s", name);
        if (r >= 0) {
                (void) sd_bus_set_description(node->agent_bus, description);
        }

        /* Migrate the agent connection to the named node */
        _cleanup_sd_bus_ sd_bus *agent_bus = sd_bus_ref(node->agent_bus);
        if (!node_set_agent_bus(named_node, agent_bus)) {
                return sd_bus_reply_method_errorf(
                                m, SD_BUS_ERROR_FAILED, "Internal error: Couldn't set agent bus");
        }

        if (controller->metrics_enabled) {
                node_enable_metrics(named_node);
        }

        node_unset_agent_bus(node);

        /* update number of online nodes and check the new system state */
        controller_check_system_status(controller, controller->number_of_nodes_online++);

        bc_log_infof("Registered managed node from fd %d as '%s'", sd_bus_get_fd(agent_bus), name);

        return sd_bus_reply_method_return(m, "");
}

static int node_disconnected(UNUSED sd_bus_message *message, void *userdata, UNUSED sd_bus_error *error) {
        Node *node = userdata;

        node_disconnect(node);

        return 0;
}

void node_disconnect(Node *node) {
        Controller *controller = node->controller;
        void *item = NULL;
        size_t i = 0;

        /* Send virtual unit remove and state change for any reported loaded units */
        while (hashmap_iter(node->unit_subscriptions, &i, &item)) {
                UnitSubscriptions *usubs = item;
                bool send_state_change = false;

                if (!usubs->loaded) {
                        continue;
                }

                if (usubs->active_state >= 0 && usubs->active_state != UNIT_INACTIVE) {
                        /* We previously reported an not-inactive valid state, send a virtual inactive state */
                        usubs->active_state = UNIT_INACTIVE;
                        free(usubs->substate);
                        usubs->substate = strdup("agent-offline");
                        send_state_change = true;
                }

                usubs->loaded = false;

                int r = 0;
                if (send_state_change) {
                        struct hashmap *unique_subs = node_compute_unique_monitor_subscriptions(
                                        node, usubs->unit);
                        if (unique_subs != NULL) {

                                Subscription **subp = NULL;
                                size_t s = 0;
                                while (hashmap_iter(unique_subs, &s, (void **) &subp)) {
                                        Subscription *sub = *subp;
                                        r = sub->handle_unit_state_changed(
                                                        sub->monitor,
                                                        node->name,
                                                        usubs->unit,
                                                        active_state_to_string(usubs->active_state),
                                                        usubs->substate,
                                                        "virtual");
                                        if (r < 0) {
                                                bc_log_error("Failed to emit UnitStateChanged signal");
                                        }
                                }
                                hashmap_free(unique_subs);
                        }
                }


                struct hashmap *unique_subs = node_compute_unique_monitor_subscriptions(node, usubs->unit);
                if (unique_subs != NULL) {

                        Subscription **subp = NULL;
                        size_t s = 0;
                        while (hashmap_iter(unique_subs, &s, (void **) &subp)) {
                                Subscription *sub = *subp;
                                r = sub->handle_unit_removed(sub->monitor, node->name, usubs->unit, "virtual");
                                if (r < 0) {
                                        bc_log_error("Failed to emit UnitRemoved signal");
                                }
                        }
                        hashmap_free(unique_subs);
                }
        }

        ProxyMonitor *proxy_monitor = NULL;
        ProxyMonitor *next_proxy_monitor = NULL;
        LIST_FOREACH_SAFE(monitors, proxy_monitor, next_proxy_monitor, node->proxy_monitors) {
                node_remove_proxy_monitor(node, proxy_monitor);
        }

        /* Remove anonymous nodes when they disconnect */
        if (node->name == NULL) {
                bc_log_info("Anonymous node disconnected");
                controller_remove_node(controller, node);
        } else {
                bc_log_infof("Node '%s' disconnected", node->name);
                /* Remove all jobs associated with the registered node that got disconnected. */
                if (!LIST_IS_EMPTY(controller->jobs)) {
                        Job *job = NULL;
                        Job *next_job = NULL;
                        LIST_FOREACH_SAFE(jobs, job, next_job, controller->jobs) {
                                if (strcmp(job->node->name, node->name) == 0) {
                                        bc_log_debugf("Removing job %d from node %s", job->id, job->node->name);
                                        LIST_REMOVE(jobs, controller->jobs, job);
                                        job_unref(job);
                                }
                        }
                }
                node_unset_agent_bus(node);

                /* update number of online nodes and check the new system state */
                controller_check_system_status(controller, controller->number_of_nodes_online--);
        }
}

const char *node_get_status(Node *node) {
        if (node_has_agent(node)) {
                return "online";
        }
        return "offline";
}

static int node_property_get_status(
                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, "s", node_get_status(node));
}

static int node_property_get_peer_ip(
                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, "s", node->peer_ip);
}

AgentRequest *agent_request_ref(AgentRequest *req) {
        req->ref_count++;
        return req;
}

void agent_request_unref(AgentRequest *req) {
        req->ref_count--;
        if (req->ref_count != 0) {
                return;
        }

        if (req->userdata && req->free_userdata) {
                req->free_userdata(req->userdata);
        }
        sd_bus_slot_unrefp(&req->slot);
        sd_bus_message_unrefp(&req->message);

        Node *node = req->node;
        LIST_REMOVE(outstanding_requests, node->outstanding_requests, req);
        node_unref(req->node);
        free(req);
}

int node_create_request(
                AgentRequest **ret,
                Node *node,
                const char *method,
                agent_request_response_t cb,
                void *userdata,
                free_func_t free_userdata) {
        AgentRequest *req = malloc0(sizeof(AgentRequest));
        if (req == NULL) {
                return -ENOMEM;
        }

        int r = sd_bus_message_new_method_call(
                        node->agent_bus,
                        &req->message,
                        BC_AGENT_DBUS_NAME,
                        INTERNAL_AGENT_OBJECT_PATH,
                        INTERNAL_AGENT_INTERFACE,
                        method);
        if (r < 0) {
                free(req);
                req = NULL;
                return r;
        }

        req->ref_count = 1;
        req->node = node_ref(node);
        LIST_INIT(outstanding_requests, req);
        req->cb = cb;
        req->userdata = userdata;
        req->free_userdata = free_userdata;
        req->is_cancelled = false;
        LIST_APPEND(outstanding_requests, node->outstanding_requests, req);

        *ret = req;
        return 0;
}

static int agent_request_callback(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error) {
        _cleanup_agent_request_ AgentRequest *req = userdata;
        if (req->is_cancelled) {
                bc_log_debugf("Response received to a cancelled request for node %s. Dropping message.",
                              req->node->name);
                return 0;
        }

        return req->cb(req, m, ret_error);
}

int agent_request_cancel(AgentRequest *r) {
        _cleanup_agent_request_ AgentRequest *req = r;
        req->is_cancelled = true;
        _cleanup_sd_bus_message_ sd_bus_message *m = NULL;
        sd_bus_message_new_method_errorf(req->message, &m, SD_BUS_ERROR_FAILED, "Request cancelled");

        return req->cb(req, m, NULL);
}

int agent_request_start(AgentRequest *req) {
        Node *node = req->node;

        int r = sd_bus_call_async(
                        node->agent_bus,
                        &req->slot,
                        req->message,
                        agent_request_callback,
                        req,
                        BC_DEFAULT_DBUS_TIMEOUT);
        if (r < 0) {
                return r;
        }

        agent_request_ref(req); /* Keep alive while operation is outstanding */
        return 1;
}

AgentRequest *node_request_list_units(
                Node *node, agent_request_response_t cb, void *userdata, free_func_t free_userdata) {
        if (!node_has_agent(node)) {
                return NULL;
        }

        _cleanup_agent_request_ AgentRequest *req = NULL;
        node_create_request(&req, node, "ListUnits", cb, userdata, free_userdata);
        if (req == NULL) {
                return NULL;
        }

        if (agent_request_start(req) < 0) {
                return NULL;
        }

        return steal_pointer(&req);
}

AgentRequest *node_request_list_unit_files(
                Node *node, agent_request_response_t cb, void *userdata, free_func_t free_userdata) {
        if (!node_has_agent(node)) {
                return NULL;
        }

        _cleanup_agent_request_ AgentRequest *req = NULL;
        node_create_request(&req, node, "ListUnitFiles", cb, userdata, free_userdata);
        if (req == NULL) {
                return NULL;
        }

        if (agent_request_start(req) < 0) {
                return NULL;
        }

        return steal_pointer(&req);
}

/*************************************************************************
 ********** org.eclipse.bluechi.Node.ListUnits **************************
 ************************************************************************/

static int method_list_units_callback(AgentRequest *req, sd_bus_message *m, UNUSED sd_bus_error *ret_error) {
        sd_bus_message *request_message = req->userdata;

        if (sd_bus_message_is_method_error(m, NULL)) {
                /* Forward error */
                return sd_bus_reply_method_error(request_message, sd_bus_message_get_error(m));
        }

        _cleanup_sd_bus_message_ sd_bus_message *reply = NULL;
        int r = sd_bus_message_new_method_return(request_message, &reply);
        if (r < 0) {
                return sd_bus_reply_method_errorf(
                                request_message,
                                SD_BUS_ERROR_FAILED,
                                "Failed to create a reply message for ListUnits request: %s",
                                strerror(-r));
        }

        r = sd_bus_message_copy(reply, m, true);
        if (r < 0) {
                return sd_bus_reply_method_errorf(
                                request_message,
                                SD_BUS_ERROR_FAILED,
                                "Failed to copy the bus message for ListUnits request: %s",
                                strerror(-r));
        }

        return sd_bus_message_send(reply);
}


static int node_method_list_units(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error) {
        Node *node = userdata;

        if (node->is_shutdown) {
                return sd_bus_reply_method_errorf(
                                m, SD_BUS_ERROR_FAILED, "Request not allowed: node is in shutdown state");
        }

        _cleanup_agent_request_ AgentRequest *agent_req = node_request_list_units(
                        node,
                        method_list_units_callback,
                        sd_bus_message_ref(m),
                        (free_func_t) sd_bus_message_unref);
        if (agent_req == NULL) {
                return sd_bus_reply_method_errorf(m, SD_BUS_ERROR_FAILED, "List units not found");
        }

        return 1;
}

/*************************************************************************
 ********** org.eclipse.bluechi.Node.ListUnitFiles ***********************
 ************************************************************************/

static int method_list_unit_files_callback(AgentRequest *req, sd_bus_message *m, UNUSED sd_bus_error *ret_error) {
        sd_bus_message *request_message = req->userdata;

        if (sd_bus_message_is_method_error(m, NULL)) {
                /* Forward error */
                return sd_bus_reply_method_error(request_message, sd_bus_message_get_error(m));
        }

        _cleanup_sd_bus_message_ sd_bus_message *reply = NULL;
        int r = sd_bus_message_new_method_return(request_message, &reply);
        if (r < 0) {
                return sd_bus_reply_method_errorf(
                                request_message,
                                SD_BUS_ERROR_FAILED,
                                "Failed to create a reply message for ListUnitFiles request: %s",
                                strerror(-r));
        }

        r = sd_bus_message_copy(reply, m, true);
        if (r < 0) {
                return sd_bus_reply_method_errorf(
                                request_message,
                                SD_BUS_ERROR_FAILED,
                                "Failed to copy the bus message for ListUnitFiles request: %s",
                                strerror(-r));
        }

        return sd_bus_message_send(reply);
}

static int node_method_list_unit_files(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error) {
        Node *node = userdata;

        if (node->is_shutdown) {
                return sd_bus_reply_method_errorf(
                                m, SD_BUS_ERROR_FAILED, "Request not allowed: node is in shutdown state");
        }

        _cleanup_agent_request_ AgentRequest *agent_req = node_request_list_unit_files(
                        node,
                        method_list_unit_files_callback,
                        sd_bus_message_ref(m),
                        (free_func_t) sd_bus_message_unref);
        if (agent_req == NULL) {
                return sd_bus_reply_method_errorf(m, SD_BUS_ERROR_FAILED, "List unit files not found");
        }
        return 1;
}

/*************************************************************************
 ********** org.eclipse.bluechi.Node.SetUnitProperty ******************
 ************************************************************************/

static int node_method_set_unit_properties_callback(
                AgentRequest *req, sd_bus_message *m, UNUSED sd_bus_error *ret_error) {
        sd_bus_message *request_message = req->userdata;

        if (sd_bus_message_is_method_error(m, NULL)) {
                /* Forward error */
                return sd_bus_reply_method_error(request_message, sd_bus_message_get_error(m));
        }

        _cleanup_sd_bus_message_ sd_bus_message *reply = NULL;
        int r = sd_bus_message_new_method_return(request_message, &reply);
        if (r < 0) {
                return sd_bus_reply_method_errorf(
                                request_message,
                                SD_BUS_ERROR_FAILED,
                                "Failed to create a reply message: %s",
                                strerror(-r));
        }

        return sd_bus_message_send(reply);
}

static int node_method_set_unit_properties(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error) {
        Node *node = userdata;
        const char *unit = NULL;
        int runtime = 0;

        if (node->is_shutdown) {
                return sd_bus_reply_method_errorf(
                                m, SD_BUS_ERROR_FAILED, "Request not allowed: node is in shutdown state");
        }

        int r = sd_bus_message_read(m, "sb", &unit, &runtime);
        if (r < 0) {
                return sd_bus_reply_method_errorf(
                                m,
                                SD_BUS_ERROR_INVALID_ARGS,
                                "Invalid argument for unit or runtime: %s",
                                strerror(-r));
        }

        _cleanup_agent_request_ AgentRequest *req = NULL;
        r = node_create_request(
                        &req,
                        node,
                        "SetUnitProperties",
                        node_method_set_unit_properties_callback,
                        sd_bus_message_ref(m),
                        (free_func_t) sd_bus_message_unref);
        if (req == NULL) {
                sd_bus_message_unref(m);

                return sd_bus_reply_method_errorf(
                                m, SD_BUS_ERROR_FAILED, "Failed to create an agent request: %s", strerror(-r));
        }

        r = sd_bus_message_append(req->message, "sb", unit, runtime);
        if (r < 0) {
                return sd_bus_reply_method_errorf(
                                m,
                                SD_BUS_ERROR_FAILED,
                                "Failed to append unit and runtime to the message: %s",
                                strerror(-r));
        }

        r = sd_bus_message_copy(req->message, m, false);
        if (r < 0) {
                return sd_bus_reply_method_errorf(
                                m, SD_BUS_ERROR_FAILED, "Failed to copy a message: %s", strerror(-r));
        }

        r = agent_request_start(req);
        if (r < 0) {
                return sd_bus_reply_method_errorf(
                                m,
                                SD_BUS_ERROR_FAILED,
                                "Failed to call the method to start the node: %s",
                                strerror(-r));
        }

        return 1;
}

static int node_method_passthrough_to_agent_callback(
                AgentRequest *req, sd_bus_message *m, UNUSED sd_bus_error *ret_error) {
        sd_bus_message *request_message = req->userdata;

        if (sd_bus_message_is_method_error(m, NULL)) {
                return sd_bus_reply_method_error(request_message, sd_bus_message_get_error(m));
        }

        _cleanup_sd_bus_message_ sd_bus_message *reply = NULL;
        int r = sd_bus_message_new_method_return(request_message, &reply);
        if (r < 0) {
                return sd_bus_reply_method_errorf(
                                request_message,
                                SD_BUS_ERROR_FAILED,
                                "Failed to create a reply message: %s",
                                strerror(-r));
        }

        r = sd_bus_message_copy(reply, m, true);
        if (r < 0) {
                return sd_bus_reply_method_errorf(
                                request_message,
                                SD_BUS_ERROR_FAILED,
                                "Failed to copy a reply message: %s",
                                strerror(-r));
        }

        return sd_bus_message_send(reply);
}

static int node_method_passthrough_to_agent(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error) {
        Node *node = userdata;

        if (node->is_shutdown) {
                return sd_bus_reply_method_errorf(
                                m, SD_BUS_ERROR_FAILED, "Request not allowed: node is in shutdown state");
        }

        _cleanup_agent_request_ AgentRequest *req = NULL;
        int r = node_create_request(
                        &req,
                        node,
                        sd_bus_message_get_member(m),
                        node_method_passthrough_to_agent_callback,
                        sd_bus_message_ref(m),
                        (free_func_t) sd_bus_message_unref);
        if (req == NULL) {
                sd_bus_message_unref(m);

                return sd_bus_reply_method_errorf(
                                m, SD_BUS_ERROR_FAILED, "Failed to create an agent request: %s", strerror(-r));
        }

        r = sd_bus_message_copy(req->message, m, true);
        if (r < 0) {
                return sd_bus_reply_method_errorf(
                                m, SD_BUS_ERROR_FAILED, "Failed to copy a reply message: %s", strerror(-r));
        }

        r = agent_request_start(req);
        if (r < 0) {
                return sd_bus_reply_method_errorf(
                                m,
                                SD_BUS_ERROR_FAILED,
                                "Failed to call the method to start the node: %s",
                                strerror(-r));
        }

        return 1;
}

/* Keep track of data related to setting up a job. For example calling
   the initial agent request before we know the job is actually going to
   happen. */
typedef struct {
        int ref_count;
        sd_bus_message *request_message;
        Job *job;
} JobSetup;

static JobSetup *job_setup_ref(JobSetup *setup) {
        setup->ref_count++;
        return setup;
}

static void job_setup_unref(JobSetup *setup) {
        setup->ref_count--;
        if (setup->ref_count != 0) {
                return;
        }

        job_unrefp(&setup->job);
        sd_bus_message_unrefp(&setup->request_message);
        free(setup);
}

DEFINE_CLEANUP_FUNC(JobSetup, job_setup_unref)
#define _cleanup_job_setup_ _cleanup_(job_setup_unrefp)

static JobSetup *job_setup_new(sd_bus_message *request_message, Node *node, const char *unit, const char *type) {
        _cleanup_job_setup_ JobSetup *setup = malloc0(sizeof(JobSetup));
        if (setup == NULL) {
                return NULL;
        }

        setup->ref_count = 1;
        setup->request_message = sd_bus_message_ref(request_message);
        setup->job = job_new(node, unit, type);
        if (setup->job == NULL) {
                NULL;
        }

        return steal_pointer(&setup);
}

static int unit_lifecycle_method_callback(AgentRequest *req, sd_bus_message *m, UNUSED sd_bus_error *ret_error) {
        Node *node = req->node;
        Controller *controller = node->controller;
        JobSetup *setup = req->userdata;

        if (sd_bus_message_is_method_error(m, NULL)) {
                /* Forward error */
                return sd_bus_reply_method_error(setup->request_message, sd_bus_message_get_error(m));
        }

        if (!controller_add_job(controller, setup->job)) {
                return sd_bus_reply_method_errorf(m, SD_BUS_ERROR_FAILED, "Failed to add a job");
        }

        return sd_bus_reply_method_return(setup->request_message, "o", setup->job->object_path);
}

static int node_run_unit_lifecycle_method(
                sd_bus_message *m, Node *node, const char *job_type, const char *method) {
        const char *unit = NULL;
        const char *mode = NULL;
        uint64_t start_time = get_time_micros();

        if (node->is_shutdown) {
                return sd_bus_reply_method_errorf(
                                m, SD_BUS_ERROR_FAILED, "Request not allowed: node is in shutdown state");
        }

        int r = sd_bus_message_read(m, "ss", &unit, &mode);
        if (r < 0) {
                return sd_bus_reply_method_errorf(
                                m,
                                SD_BUS_ERROR_INVALID_ARGS,
                                "Invalid argument for unit or mode: %s",
                                strerror(-r));
        }

        _cleanup_job_setup_ JobSetup *setup = job_setup_new(m, node, unit, job_type);
        if (setup == NULL) {
                return sd_bus_reply_method_errorf(m, SD_BUS_ERROR_NO_MEMORY, "Out of memory");
        }

        if (node->controller->metrics_enabled) {
                setup->job->job_start_micros = start_time;
        }

        _cleanup_agent_request_ AgentRequest *req = NULL;
        r = node_create_request(
                        &req,
                        node,
                        method,
                        unit_lifecycle_method_callback,
                        job_setup_ref(setup),
                        (free_func_t) job_setup_unref);
        if (req == NULL) {
                job_setup_unref(setup);

                return sd_bus_reply_method_errorf(
                                m, SD_BUS_ERROR_FAILED, "Failed to create an agent request: %s", strerror(-r));
        }

        r = sd_bus_message_append(req->message, "ssu", unit, mode, setup->job->id);
        if (r < 0) {
                return sd_bus_reply_method_errorf(
                                m,
                                SD_BUS_ERROR_FAILED,
                                "Failed to append unit, mode, and job ID to the message: %s",
                                strerror(-r));
        }

        r = agent_request_start(req);
        if (r < 0) {
                return sd_bus_reply_method_errorf(
                                m,
                                SD_BUS_ERROR_FAILED,
                                "Failed to call the method to start the node: %s",
                                strerror(-r));
        }

        return 1;
}


/*************************************************************************
 ********** org.eclipse.bluechi.Node.StartUnit **************************
 ************************************************************************/

static int node_method_start_unit(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error) {
        return node_run_unit_lifecycle_method(m, (Node *) userdata, "start", "StartUnit");
}

/*************************************************************************
 ********** org.eclipse.bluechi.Node.StopUnit ***************************
 ************************************************************************/


static int node_method_stop_unit(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error) {
        return node_run_unit_lifecycle_method(m, (Node *) userdata, "stop", "StopUnit");
}

/*************************************************************************
 ********** org.eclipse.bluechi.Node.RestartUnit ************************
 ************************************************************************/

static int node_method_restart_unit(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error) {
        return node_run_unit_lifecycle_method(m, (Node *) userdata, "restart", "RestartUnit");
}

/*************************************************************************
 ********** org.eclipse.bluechi.Node.ReloadUnit **************************
 ************************************************************************/

static int node_method_reload_unit(sd_bus_message *m, void *userdata, UNUSED sd_bus_error *ret_error) {
        return node_run_unit_lifecycle_method(m, (Node *) userdata, "reload", "ReloadUnit");
}

/*************************************************************************
 ********** org.eclipse.bluechi.Node.SetLogLevel *******************
 ************************************************************************/

static int node_method_set_log_level(sd_bus_message *m, UNUSED void *userdata, UNUSED sd_bus_error *ret_error) {
        const char *level = NULL;
        Node *node = (Node *) userdata;
        sd_bus_error error = SD_BUS_ERROR_NULL;
        _cleanup_sd_bus_message_ sd_bus_message *sub_m = NULL;

        int r = sd_bus_message_read(m, "s", &level);
        if (r < 0) {
                return sd_bus_reply_method_errorf(
                                m,
                                SD_BUS_ERROR_INVALID_ARGS,
                                "Invalid argument for the log-level: %s",
                                strerror(-r));
        }
        LogLevel loglevel = string_to_log_level(level);
        if (loglevel == LOG_LEVEL_INVALID) {
                r = sd_bus_reply_method_return(m, "");
                if (r < 0) {
                        return sd_bus_reply_method_errorf(
                                        m,
                                        SD_BUS_ERROR_INVALID_ARGS,
                                        "Invalid argument for the log level invalid");
                }
        }
        r = sd_bus_call_method(
                        node->agent_bus,
                        BC_AGENT_DBUS_NAME,
                        INTERNAL_AGENT_OBJECT_PATH,
                        INTERNAL_AGENT_INTERFACE,
                        "SetLogLevel",
                        &error,
                        &sub_m,
                        "s",
                        level);
        if (r < 0) {
                bc_log_errorf("Failed to set log level call: %s", error.message);
                sd_bus_error_free(&error);
                return sd_bus_reply_method_errorf(
                                m,
                                SD_BUS_ERROR_FAILED,
                                "Failed to call a method to set the log level: %s",
                                strerror(-r));
        }
        return sd_bus_reply_method_return(m, "");
}

static int send_agent_simple_message(Node *node, const char *method, const char *arg) {
        _cleanup_sd_bus_message_ sd_bus_message *m = NULL;
        int r = sd_bus_message_new_method_call(
                        node->agent_bus,
                        &m,
                        BC_AGENT_DBUS_NAME,
                        INTERNAL_AGENT_OBJECT_PATH,
                        INTERNAL_AGENT_INTERFACE,
                        method);
        if (r < 0) {
                return r;
        }

        if (arg != NULL) {
                r = sd_bus_message_append(m, "s", arg);
                if (r < 0) {
                        return r;
                }
        }

        return sd_bus_send(node->agent_bus, m, NULL);
}

static void node_send_agent_subscribe(Node *node, const char *unit) {
        if (!node_has_agent(node)) {
                return;
        }

        int r = send_agent_simple_message(node, "Subscribe", unit);
        if (r < 0) {
                bc_log_error("Failed to subscribe w/ agent");
        }
}


static void node_send_agent_unsubscribe(Node *node, const char *unit) {
        if (!node_has_agent(node)) {
                return;
        }

        int r = send_agent_simple_message(node, "Unsubscribe", unit);
        if (r < 0) {
                bc_log_error("Failed to unsubscribe w/ agent");
        }
}

/* Resubscribe to all subscriptions */
static void node_send_agent_subscribe_all(Node *node) {
        void *item = NULL;
        size_t i = 0;

        while (hashmap_iter(node->unit_subscriptions, &i, &item)) {
                UnitSubscriptions *usubs = item;
                node_send_agent_subscribe(node, usubs->unit);
        }
}

void node_subscribe(Node *node, Subscription *sub) {
        SubscribedUnit *sub_unit = NULL;
        SubscribedUnit *next_sub_unit = NULL;
        LIST_FOREACH_SAFE(units, sub_unit, next_sub_unit, sub->subscribed_units) {
                const UnitSubscriptionsKey key = { sub_unit->name };
                UnitSubscriptions *usubs = NULL;

                _cleanup_free_ UnitSubscription *usub = malloc0(sizeof(UnitSubscription));
                if (usub == NULL) {
                        bc_log_error("Failed to subscribe to unit, OOM");
                        return;
                }
                usub->sub = sub;

                usubs = (UnitSubscriptions *) hashmap_get(node->unit_subscriptions, &key);
                if (usubs == NULL) {
                        UnitSubscriptions v = { NULL, NULL, false, _UNIT_ACTIVE_STATE_INVALID, NULL };
                        v.unit = strdup(key.unit);
                        if (v.unit == NULL) {
                                bc_log_error("Failed to subscribe to unit, OOM");
                                return;
                        }

                        usubs = (UnitSubscriptions *) hashmap_set(node->unit_subscriptions, &v);
                        if (usubs == NULL && hashmap_oom(node->unit_subscriptions)) {
                                free(v.unit);
                                bc_log_error("Failed to subscribe to unit, OOM");
                                return;
                        }

                        /* First sub to this unit, pass to agent */
                        node_send_agent_subscribe(node, sub_unit->name);

                        usubs = (UnitSubscriptions *) hashmap_get(node->unit_subscriptions, &key);
                }

                LIST_APPEND(subs, usubs->subs, steal_pointer(&usub));

                /* We know this is loaded, so we won't get notified from
                   the agent, instead send a virtual event here. */
                if (usubs->loaded) {
                        int r = sub->handle_unit_new(sub->monitor, node->name, sub_unit->name, "virtual");
                        if (r < 0) {
                                bc_log_error("Failed to emit UnitNew signal");
                        }

                        if (usubs->active_state >= 0) {
                                r = sub->handle_unit_state_changed(
                                                sub->monitor,
                                                node->name,
                                                sub_unit->name,
                                                active_state_to_string(usubs->active_state),
                                                usubs->substate ? usubs->substate : "invalid",
                                                "virtual");
                                if (r < 0) {
                                        bc_log_error("Failed to emit UnitNew signal");
                                }
                        }
                }
        }
}

void node_unsubscribe(Node *node, Subscription *sub) {
        SubscribedUnit *sub_unit = NULL;
        SubscribedUnit *next_sub_unit = NULL;
        LIST_FOREACH_SAFE(units, sub_unit, next_sub_unit, sub->subscribed_units) {
                UnitSubscriptionsKey key = { sub_unit->name };
                UnitSubscriptions *usubs = NULL;
                UnitSubscription *usub = NULL;
                UnitSubscription *found = NULL;
                UnitSubscriptions *deleted = NULL;

                /* NOTE: If there are errors during subscribe we may still
                   call unsubscribe, so this must silently handle the
                   case of too many unsubscribes. */

                usubs = (UnitSubscriptions *) hashmap_get(node->unit_subscriptions, &key);
                if (usubs == NULL) {
                        continue;
                }

                LIST_FOREACH(subs, usub, usubs->subs) {
                        if (usub->sub == sub) {
                                found = usub;
                                break;
                        }
                }

                if (found == NULL) {
                        continue;
                }

                LIST_REMOVE(subs, usubs->subs, found);
                free_and_null(found);

                if (LIST_IS_EMPTY(usubs->subs)) {
                        /* Last subscription for this unit, tell agent */
                        node_send_agent_unsubscribe(node, sub_unit->name);
                        deleted = (UnitSubscriptions *) hashmap_delete(node->unit_subscriptions, &key);
                        if (deleted) {
                                unit_subscriptions_clear(deleted);
                        }
                }
        }
}

static void node_start_proxy_dependency(Node *node, ProxyDependency *dep) {
        if (!node_has_agent(node)) {
                return;
        }

        bc_log_infof("Starting dependency %s on node %s", dep->unit_name, node->name);

        int r = send_agent_simple_message(node, "StartDep", dep->unit_name);
        if (r < 0) {
                bc_log_error("Failed to send StartDep to agent");
        }
}

static void node_start_proxy_dependency_all(Node *node) {
        ProxyDependency *dep = NULL;
        LIST_FOREACH(deps, dep, node->proxy_dependencies) {
                node_start_proxy_dependency(node, dep);
        }
}

static void node_stop_proxy_dependency(Node *node, ProxyDependency *dep) {
        if (!node_has_agent(node)) {
                return;
        }

        bc_log_infof("Stopping dependency %s on node %s", dep->unit_name, node->name);

        int r = send_agent_simple_message(node, "StopDep", dep->unit_name);
        if (r < 0) {
                bc_log_error("Failed to send StopDep to agent");
        }
}

static struct ProxyDependency *node_find_proxy_dependency(Node *node, const char *unit_name) {
        ProxyDependency *dep = NULL;
        LIST_FOREACH(deps, dep, node->proxy_dependencies) {
                if (streq(dep->unit_name, unit_name)) {
                        return dep;
                }
        }

        return NULL;
}

int node_add_proxy_dependency(Node *node, const char *unit_name) {
        ProxyDependency *dep = NULL;

        dep = node_find_proxy_dependency(node, unit_name);
        if (dep) {
                dep->n_deps++;
                /* Always start, if the dep service was stopped by
                   the target service stopping */
                node_start_proxy_dependency(node, dep);
                return 0;
        }

        _cleanup_free_ char *unit_name_copy = strdup(unit_name);
        if (unit_name_copy == NULL) {
                return -ENOMEM;
        }

        dep = malloc0(sizeof(ProxyDependency));
        if (dep == NULL) {
                return -ENOMEM;
        }

        dep->unit_name = steal_pointer(&unit_name_copy);
        dep->n_deps = 1;
        LIST_APPEND(deps, node->proxy_dependencies, dep);

        node_start_proxy_dependency(node, dep);

        return 0;
}

int node_remove_proxy_dependency(Node *node, const char *unit_name) {
        ProxyDependency *dep = NULL;
        dep = node_find_proxy_dependency(node, unit_name);
        if (!dep) {
                return -ENOENT;
        }

        dep->n_deps--;

        if (dep->n_deps == 0) {
                /* Only stop on the last dep */
                node_stop_proxy_dependency(node, dep);

                LIST_REMOVE(deps, node->proxy_dependencies, dep);
                proxy_dependency_free(dep);
        }

        return 0;
}

int node_method_get_unit_uint64_property_sync(Node *node, char *unit, char *property, uint64_t *value) {
        int r = 0;
        _cleanup_sd_bus_message_ sd_bus_message *message = NULL;
        sd_bus_error error = SD_BUS_ERROR_NULL;
        r = sd_bus_call_method(
                        node->agent_bus,
                        BC_AGENT_DBUS_NAME,
                        INTERNAL_AGENT_OBJECT_PATH,
                        INTERNAL_AGENT_INTERFACE,
                        "GetUnitProperty",
                        &error,
                        &message,
                        "sss",
                        unit,
                        "org.freedesktop.systemd1.Unit",
                        property);
        if (r < 0) {
                bc_log_errorf("Failed to issue GetUnitProperty call: %s", error.message);
                sd_bus_error_free(&error);
                return r;
        }

        r = sd_bus_message_enter_container(message, SD_BUS_TYPE_VARIANT, "t");
        if (r < 0) {
                bc_log_errorf("Failed to parse response message: %s", strerror(-r));
                return r;
        }

        r = sd_bus_message_read_basic(message, SD_BUS_TYPE_UINT64, value);
        if (r < 0) {
                bc_log_errorf("Failed to parse response message: %s", strerror(-r));
                return r;
        }

        r = sd_bus_message_exit_container(message);
        if (r < 0) {
                bc_log_errorf("Failed to parse response message: %s", strerror(-r));
                return r;
        }

        return 0;
}

void node_enable_metrics(Node *node) {
        if (!node_has_agent(node)) {
                return;
        }

        int r = send_agent_simple_message(node, "EnableMetrics", NULL);
        if (r < 0) {
                bc_log_error("Failed to enable metrics on agent");
        }

        if (!metrics_node_signal_matching_register(node)) {
                bc_log_error("Failed to enable metrics on agent");
        }
}

void node_disable_metrics(Node *node) {
        if (!node_has_agent(node)) {
                return;
        }

        int r = send_agent_simple_message(node, "DisableMetrics", NULL);
        if (r < 0) {
                bc_log_error("Failed to disable metrics on agent");
        }

        sd_bus_slot_unrefp(&node->metrics_matching_slot);
        node->metrics_matching_slot = NULL;
}

eclipse-bluechi / bluechi / 19941333032

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous