From 814d14a6f6e40ddcc1802cdc569c0a8a7d5ca666 Mon Sep 17 00:00:00 2001 From: "Viv B." Date: Thu, 27 Aug 2026 13:16:03 +1000 Subject: [PATCH] feat(place/driver_health): include driver processes running on edges (#2) A core node already reports the processes on each edge attached to it, in the same responses we were making, but only the `local` half was being read. So a driver running only on an edge was invisible, and one running on both a node and an edge was reported for the node alone. Now the union of `loaded.local` and every `loaded.edge` entry is walked, and a state is emitted per location a driver runs in, taking memory from `status.local` or `status.edge[edge_id]` as appropriate. This costs no extra requests, a single driver status response covers the node and all of its edges. Edge processes are named `.`. Edges aren't queried directly and the node an edge happens to be attached to isn't part of the path to it, so neither the edge hostname nor the core hostname belongs in the name. Results are deduplicated by name, so an edge reported by more than one core node is counted once. Not yet exercised against a live edge, the spec covers it with fake core nodes. Co-authored-by: Claude Opus 5 (1M context) --- drivers/place/driver_health.cr | 48 ++++++++++++++++---- drivers/place/driver_health_spec.cr | 70 ++++++++++++++++++++++------- 2 files changed, 94 insertions(+), 24 deletions(-) diff --git a/drivers/place/driver_health.cr b/drivers/place/driver_health.cr index 1b63ab82e0..fb7d20ec8c 100644 --- a/drivers/place/driver_health.cr +++ b/drivers/place/driver_health.cr @@ -4,17 +4,23 @@ require "placeos-core-client" require "redis_service_manager" # Reports whether the driver processes on every core node in the cluster are -# running. +# running, including those running on an edge. # # Core nodes are discovered from the service registration that # `Proxy::RemoteDriver` already uses to route module requests, so no API key or # request to an external PlaceOS instance is required. Each node is then asked # about its own driver processes over core's internal API (`/api/core/v1`), the # same data rest-api aggregates for its `/cluster` routes. +# +# Edges are never queried directly. A core node reports the processes on each +# edge attached to it in the same responses, forwarding the request over the edge +# protocol where it is served by the same `ProcessManager::Common#driver_status` +# a core node uses for its own processes - so edge memory figures mean the same +# thing local ones do. class Place::DriverHealth < PlaceOS::Driver descriptive_name "PlaceOS Driver Health" generic_name :DriverHealth - description %(Checks that the driver processes on every core node in the cluster are running, exposing a running state (1 or 0) per driver for backoffice and InfluxDB) + description %(Checks that the driver processes on every core node and edge in the cluster are running, exposing a running state (1 or 0) per driver for backoffice and InfluxDB) default_settings({ # how often to check the cluster, set to 0 to only check on request @@ -41,7 +47,8 @@ class Place::DriverHealth < PlaceOS::Driver # `.`, unique across the cluster getter name : String - # the core node the driver process is on, i.e. `core-0` + # the core node the driver process is on, i.e. `core-0`, or the id of the + # edge it's on, i.e. `edge-KjO683qopP` getter hostname : String # the driver source path, i.e. `drivers_place_bookings` @@ -114,6 +121,9 @@ class Place::DriverHealth < PlaceOS::Driver end drivers.sort_by!(&.name) + + # an edge attached to more than one core node would be reported by each + drivers.uniq!(&.name) not_running = drivers.select(&.running.zero?).map(&.name) self[:clusters] = clusters @@ -132,23 +142,43 @@ class Place::DriverHealth < PlaceOS::Driver drivers end - # returns the nodes hostname and the state of the drivers running on it + # returns the nodes hostname and the state of the drivers running on it, along + # with those running on any edge attached to it protected def check_node(uri : URI) : Tuple(String, Array(DriverState)) PlaceOS::Core::Client.client(uri, retries: CORE_RETRIES) do |client| # the hostname of the pod, i.e. `core-0` hostname = client.core_load.local.hostname - # a mapping of driver => the modules that driver is running - states = client.loaded.local.keys.map do |driver| - # no status or no memory in use means the process isn't running - memory = client.driver_status(driver).local.try(&.memory_usage) || 0_i64 - DriverState.new(hostname, driver, memory.zero? ? 0 : 1, Time.utc.to_unix) + # `local` and each entry of `edge` map driver => the modules it's running + loaded = client.loaded + states = [] of DriverState + + # a driver can be running on the node, on one of its edges, or both, and a + # single status request covers all of them + drivers = loaded.local.keys.to_set + loaded.edge.each_value { |processes| drivers.concat processes.keys } + + drivers.each do |driver| + status = client.driver_status driver + checked = Time.utc.to_unix + + states << DriverState.new(hostname, driver, running(status.local), checked) if loaded.local.has_key? driver + + loaded.edge.each do |edge_id, processes| + next unless processes.has_key? driver + states << DriverState.new(edge_id, driver, running(status.edge[edge_id]?), checked) + end end {hostname, states} end end + # a driver process that isn't using any memory isn't running + protected def running(status : PlaceOS::Core::Client::DriverStatus::Metadata?) : Int32 + (status.try(&.memory_usage) || 0_i64).zero? ? 0 : 1 + end + # the configured nodes, otherwise the nodes registered in the cluster protected def core_nodes : Hash(String, URI) nodes = @core_nodes diff --git a/drivers/place/driver_health_spec.cr b/drivers/place/driver_health_spec.cr index b169da80a1..0b937c6335 100644 --- a/drivers/place/driver_health_spec.cr +++ b/drivers/place/driver_health_spec.cr @@ -15,6 +15,24 @@ CORE_2_ID = "01M073D9GRBDTX8Q1XH5ZYQN9Y" DISPLAY = "drivers_place_demo_display_4894a36_arm64" BOOKINGS = "drivers_place_bookings_1a2b3c4_arm64" ROUTER = "drivers_place_router_9f8e7d6_arm64" +KIOSK = "drivers_place_kiosk_2c3d4e5_arm64" + +EDGE_A = "edge-abc" +EDGE_B = "edge-xyz" + +# maps a driver to the memory it's using, nil for a driver core has no status for +alias Drivers = Hash(String, Int64?) +alias Metadata = NamedTuple(running: Bool, memory_usage: Int64) + +# :nodoc: +def metadata(memory : Int64?) : Metadata? + memory.nil? ? nil : {running: memory > 0, memory_usage: memory} +end + +# :nodoc: +def modules(drivers : Drivers) + drivers.keys.to_h { |driver| {driver, ["mod-#{driver}"]} } +end # a driver binary that wasn't named by the build service LEGACY = "legacy_driver" @@ -36,9 +54,9 @@ def system_load(hostname : String) end # :nodoc: -# `drivers` maps a driver to the memory it's using, nil for a driver core has no -# status for at all -def serve_core(port : Int32, hostname : String, drivers : Hash(String, Int64?)) +# a core node reports its own driver processes plus those on each edge attached +# to it, keyed by edge id +def serve_core(port : Int32, hostname : String, drivers : Drivers, edges : Hash(String, Drivers) = {} of String => Drivers) server = HTTP::Server.new do |context| context.response.content_type = "application/json" @@ -46,12 +64,15 @@ def serve_core(port : Int32, hostname : String, drivers : Hash(String, Int64?)) when "/api/core/v1/status/load" context.response.print({local: system_load(hostname), edge: {} of String => String}.to_json) when "/api/core/v1/status/loaded" - loaded = drivers.keys.to_h { |driver| {driver, ["mod-#{driver}"]} } - context.response.print({local: loaded, edge: {} of String => String}.to_json) + context.response.print({ + local: modules(drivers), + edge: edges.transform_values { |edge| modules(edge) }, + }.to_json) when "/api/core/v1/status/driver" - memory = drivers[context.request.query_params["path"]] - local = memory.nil? ? nil : {running: memory > 0, memory_usage: memory} - context.response.print({local: local, edge: {} of String => String}.to_json) + driver = context.request.query_params["path"] + edge_status = {} of String => Metadata? + edges.each { |edge_id, edge| edge_status[edge_id] = metadata(edge[driver]) if edge.has_key? driver } + context.response.print({local: metadata(drivers[driver]?), edge: edge_status}.to_json) else context.response.status_code = 404 end @@ -75,8 +96,14 @@ def serve_broken_core(port : Int32) server end -serve_core(CORE_0_PORT, "core-0", {DISPLAY => 12_345_i64, BOOKINGS => 0_i64}) -serve_core(CORE_1_PORT, "core-1", {ROUTER => nil, LEGACY => 6_789_i64}) +# EDGE_A is attached to both core nodes, so it gets reported twice and has to be +# deduplicated. it reports the same memory either way so the result is unambiguous +serve_core(CORE_0_PORT, "core-0", + Drivers{DISPLAY => 12_345_i64, BOOKINGS => 0_i64}, + {EDGE_A => Drivers{KIOSK => 0_i64}}) +serve_core(CORE_1_PORT, "core-1", + Drivers{ROUTER => nil, LEGACY => 6_789_i64}, + {EDGE_A => Drivers{DISPLAY => 4_000_i64, KIOSK => 0_i64}, EDGE_B => Drivers{KIOSK => 9_000_i64}}) serve_broken_core(CORE_2_PORT) DriverSpecs.mock_driver "Place::DriverHealth" do @@ -102,7 +129,7 @@ DriverSpecs.mock_driver "Place::DriverHealth" do results = Array(DriverState) .from_json exec(:check_drivers).get.not_nil!.to_json - results.size.should eq 4 + results.size.should eq 7 # a driver using memory is running, the commit and architecture are split # out of the executable name @@ -129,18 +156,27 @@ DriverSpecs.mock_driver "Place::DriverHealth" do results[3][:commit].should eq "" results[3][:running].should eq 1 + # a process on an edge is named for the edge, not the node reporting it. + # EDGE_A is attached to both core nodes yet appears once + results[4..].map { |result| {result[:name], result[:hostname], result[:running]} }.should eq [ + {"#{EDGE_A}.drivers_place_demo_display", EDGE_A, 1}, # on an edge, using memory + {"#{EDGE_A}.drivers_place_kiosk", EDGE_A, 0}, # on an edge, using no memory + {"#{EDGE_B}.drivers_place_kiosk", EDGE_B, 1}, # a second edge of the same node + ] + # each result is stamped with when it was checked results.each do |result| result[:timestamp].should be >= before result[:timestamp].should be <= Time.utc.to_unix end - status[:driver_count].should eq 4 - status[:running_count].should eq 2 + status[:driver_count].should eq 7 + status[:running_count].should eq 4 Array(String).from_json(status[:not_running].to_json).should eq [ "core-0.drivers_place_bookings", "core-1.drivers_place_router", + "#{EDGE_A}.drivers_place_kiosk", ] Array(NamedTuple(id: String, name: String)).from_json(status[:clusters].to_json).should eq [ @@ -172,13 +208,17 @@ DriverSpecs.mock_driver "Place::DriverHealth" do results = Array(DriverState) .from_json exec(:check_drivers).get.not_nil!.to_json - results.map(&.[](:name)).should eq ["core-0.drivers_place_bookings", "core-0.drivers_place_demo_display"] + results.map(&.[](:name)).should eq [ + "core-0.drivers_place_bookings", + "core-0.drivers_place_demo_display", + "#{EDGE_A}.drivers_place_kiosk", + ] Array(String).from_json(status[:unreachable_clusters].to_json).should eq [CORE_2_ID] Array(NamedTuple(id: String, name: String)).from_json(status[:clusters].to_json).should eq [ {id: CORE_0_ID, name: "core-0"}, ] - status[:driver_count].should eq 2 + status[:driver_count].should eq 3 status[:running_count].should eq 1 end end