feat(place/driver_health): include driver processes running on edges

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
     # `<hostname>.<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

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,22 +54,25 @@ 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"
 
     case context.request.path
     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