feat(place/driver_health): report driver process health across the cluster

drivers/place/driver_health.cr

@@ -0,0 +1,171 @@
+require "placeos-driver"
+require "placeos-driver/proxy/remote_driver"
+require "placeos-core-client"
+require "redis_service_manager"
+
+# Reports whether the driver processes on every core node in the cluster are
+# running.
+#
+# 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.
+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)
+
+  default_settings({
+    # how often to check the cluster, set to 0 to only check on request
+    check_every_minutes: 5,
+
+    # optionally check a fixed set of core nodes, `node id => core URI`.
+    # the cluster is discovered when this is empty, which is what you want
+    # in a normal deployment
+    core_nodes: {} of String => String,
+  })
+
+  # attempts made against a core node before it's considered unreachable.
+  # the client default of 10 (with a 40 second max interval) would stall the
+  # check for minutes against a node that is down
+  CORE_RETRIES = 2
+
+  # driver executables are named `<source path>_<short commit>_<arch>` by the
+  # build service, i.e. `drivers_place_bookings_4894a36_arm64`
+  EXECUTABLE_NAME = /\A(?<driver>.+)_(?<commit>[0-9a-f]{7})_(?<arch>[a-z0-9]+)\z/
+
+  struct DriverState
+    include JSON::Serializable
+
+    # `<hostname>.<driver>`, unique across the cluster
+    getter name : String
+
+    # the core node the driver process is on, i.e. `core-0`
+    getter hostname : String
+
+    # the driver source path, i.e. `drivers_place_bookings`
+    getter driver : String
+
+    # the short commit hash the driver was built from, i.e. `4894a36`
+    getter commit : String
+
+    # 1 when the driver process was using memory when we asked, otherwise 0.
+    # numeric rather than boolean so InfluxDB can aggregate it (mean, sum)
+    getter running : Int32
+
+    # when the running state was checked, unix seconds
+    getter timestamp : Int64
+
+    def initialize(@hostname, executable : String, @running, @timestamp)
+      if match = EXECUTABLE_NAME.match(executable)
+        @driver = match["driver"]
+        @commit = match["commit"]
+      else
+        @driver = executable
+        @commit = ""
+      end
+      @name = "#{@hostname}.#{@driver}"
+    end
+  end
+
+  @check_every : Time::Span = 5.minutes
+  @core_nodes : Hash(String, URI) = {} of String => URI
+  @discovery : Clustering::Discovery? = nil
+
+  def on_load
+    on_update
+  end
+
+  def on_update
+    @check_every = (setting?(Int32, :check_every_minutes) || 5).minutes
+    @core_nodes = (setting?(Hash(String, String), :core_nodes) || {} of String => String)
+      .transform_values { |uri| URI.parse uri }
+
+    schedule.clear
+    return unless @check_every > Time::Span.zero
+
+    # let the cluster settle before the first check, drivers are still launching
+    # for a while after a core node starts
+    schedule.in(30.seconds) { check_drivers }
+    schedule.every(@check_every) { check_drivers }
+  end
+
+  # the core nodes that make up the cluster, `node id => core URI`
+  def cluster_nodes : Hash(String, String)
+    core_nodes.transform_values(&.to_s)
+  end
+
+  # checks every driver process on every core node in the cluster
+  def check_drivers : Array(DriverState)
+    clusters = [] of NamedTuple(id: String, name: String)
+    unreachable = [] of String
+    drivers = [] of DriverState
+
+    core_nodes.each do |id, uri|
+      begin
+        hostname, states = check_node uri
+        clusters << {id: id, name: hostname}
+        drivers.concat states
+      rescue error
+        logger.warn(exception: error) { "failed to query core node #{id} on #{uri}" }
+        unreachable << id
+      end
+    end
+
+    drivers.sort_by!(&.name)
+    not_running = drivers.select(&.running.zero?).map(&.name)
+
+    self[:clusters] = clusters
+    self[:unreachable_clusters] = unreachable
+    # exposed as InfluxDB tags so each driver process is its own series
+    self[:drivers] = {
+      value:       drivers,
+      ts_hint:     "complex",
+      ts_tag_keys: ["name", "hostname"],
+    }
+    self[:driver_count] = drivers.size
+    self[:running_count] = drivers.size - not_running.size
+    self[:not_running] = not_running
+    self[:last_checked] = Time.utc.to_unix
+
+    drivers
+  end
+
+  # returns the nodes hostname and the state of the drivers running on 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)
+      end
+
+      {hostname, states}
+    end
+  end
+
+  # the configured nodes, otherwise the nodes registered in the cluster
+  protected def core_nodes : Hash(String, URI)
+    nodes = @core_nodes
+    return nodes unless nodes.empty?
+    discovery.node_hash
+  end
+
+  # reads the core service registration, the same one `Proxy::RemoteDriver` uses
+  # to work out which node is running a module. we only ever read from it, this
+  # process is not a member of the cluster
+  protected def discovery : Clustering::Discovery
+    @discovery ||= Clustering::Discovery.new(
+      RedisServiceManager.new(
+        service: PlaceOS::Driver::Proxy::RemoteDriver::CORE_NAMESPACE,
+        redis: PlaceOS::Driver::RedisStorage.shared_redis_client,
+        lock: PlaceOS::Driver::RedisStorage.redis_lock
+      )
+    )
+  end
+end

drivers/place/driver_health_spec.cr

@@ -0,0 +1,184 @@
+require "placeos-driver/spec"
+require "http/server"
+
+# :nodoc:
+# the driver talks to each core node over that nodes internal API, so we stand up
+# fake core pods on local ports and point the driver at them
+CORE_0_PORT = 8341
+CORE_1_PORT = 8342
+CORE_2_PORT = 8343
+
+CORE_0_ID = "01M073D9GRBDTX8Q1XH5ZYQN9W"
+CORE_1_ID = "01M073D9GRBDTX8Q1XH5ZYQN9X"
+CORE_2_ID = "01M073D9GRBDTX8Q1XH5ZYQN9Y"
+
+DISPLAY  = "drivers_place_demo_display_4894a36_arm64"
+BOOKINGS = "drivers_place_bookings_1a2b3c4_arm64"
+ROUTER   = "drivers_place_router_9f8e7d6_arm64"
+
+# a driver binary that wasn't named by the build service
+LEGACY = "legacy_driver"
+
+# :nodoc:
+alias DriverState = NamedTuple(name: String, hostname: String, driver: String, commit: String, running: Int32, timestamp: Int64)
+
+# :nodoc:
+def system_load(hostname : String)
+  {
+    hostname:     hostname,
+    cpu_count:    4,
+    core_cpu:     0.5,
+    total_cpu:    1.5,
+    memory_total: 8_000_000_i64,
+    memory_usage: 4_000_000_i64,
+    core_memory:  100_000_i64,
+  }
+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?))
+  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)
+    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)
+    else
+      context.response.status_code = 404
+    end
+  end
+
+  # bind before spawning so the port is accepting connections by the time the
+  # driver makes a request
+  server.bind_tcp "127.0.0.1", port
+  spawn { server.listen }
+  server
+end
+
+# :nodoc:
+def serve_broken_core(port : Int32)
+  server = HTTP::Server.new do |context|
+    context.response.status_code = 500
+    context.response.print("core is not well")
+  end
+  server.bind_tcp "127.0.0.1", port
+  spawn { server.listen }
+  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})
+serve_broken_core(CORE_2_PORT)
+
+DriverSpecs.mock_driver "Place::DriverHealth" do
+  # 0 disables the schedule so the checks below are the only ones that run
+  settings({
+    check_every_minutes: 0,
+    core_nodes:          {
+      CORE_0_ID => "http://127.0.0.1:#{CORE_0_PORT}",
+      CORE_1_ID => "http://127.0.0.1:#{CORE_1_PORT}",
+    },
+  })
+
+  it "reports the configured cluster nodes" do
+    nodes = Hash(String, String).from_json exec(:cluster_nodes).get.not_nil!.to_json
+    nodes.should eq({
+      CORE_0_ID => "http://127.0.0.1:#{CORE_0_PORT}",
+      CORE_1_ID => "http://127.0.0.1:#{CORE_1_PORT}",
+    })
+  end
+
+  it "checks the memory use of every driver process in the cluster" do
+    before = Time.utc.to_unix
+    results = Array(DriverState)
+      .from_json exec(:check_drivers).get.not_nil!.to_json
+
+    results.size.should eq 4
+
+    # a driver using memory is running, the commit and architecture are split
+    # out of the executable name
+    results[1][:name].should eq "core-0.drivers_place_demo_display"
+    results[1][:hostname].should eq "core-0"
+    results[1][:driver].should eq "drivers_place_demo_display"
+    results[1][:commit].should eq "4894a36"
+    results[1][:running].should eq 1
+
+    # a driver using no memory is not
+    results[0][:name].should eq "core-0.drivers_place_bookings"
+    results[0][:commit].should eq "1a2b3c4"
+    results[0][:running].should eq 0
+
+    # neither is one core has no status for
+    results[2][:name].should eq "core-1.drivers_place_router"
+    results[2][:hostname].should eq "core-1"
+    results[2][:commit].should eq "9f8e7d6"
+    results[2][:running].should eq 0
+
+    # an executable the build service didn't name is reported as is
+    results[3][:name].should eq "core-1.#{LEGACY}"
+    results[3][:driver].should eq LEGACY
+    results[3][:commit].should eq ""
+    results[3][:running].should eq 1
+
+    # 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
+
+    Array(String).from_json(status[:not_running].to_json).should eq [
+      "core-0.drivers_place_bookings",
+      "core-1.drivers_place_router",
+    ]
+
+    Array(NamedTuple(id: String, name: String)).from_json(status[:clusters].to_json).should eq [
+      {id: CORE_0_ID, name: "core-0"},
+      {id: CORE_1_ID, name: "core-1"},
+    ]
+
+    Array(String).from_json(status[:unreachable_clusters].to_json).should be_empty
+    status[:last_checked].as_i64.should be >= before
+
+    # the state matches what the function returned, shaped so the influx
+    # exporter tags each point with the driver name and node
+    drivers = status[:drivers]
+    drivers["ts_hint"].should eq "complex"
+    Array(String).from_json(drivers["ts_tag_keys"].to_json).should eq ["name", "hostname"]
+    Array(DriverState)
+      .from_json(drivers["value"].to_json).should eq results
+  end
+
+  it "flags a node it can't reach and still checks the rest" do
+    settings({
+      check_every_minutes: 0,
+      core_nodes:          {
+        CORE_0_ID => "http://127.0.0.1:#{CORE_0_PORT}",
+        CORE_2_ID => "http://127.0.0.1:#{CORE_2_PORT}",
+      },
+    })
+
+    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"]
+
+    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[:running_count].should eq 1
+  end
+end