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