diff --git a/drivers/place/driver_health.cr b/drivers/place/driver_health.cr new file mode 100644 index 0000000000..1b63ab82e0 --- /dev/null +++ b/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 `__` by the + # build service, i.e. `drivers_place_bookings_4894a36_arm64` + EXECUTABLE_NAME = /\A(?.+)_(?[0-9a-f]{7})_(?[a-z0-9]+)\z/ + + struct DriverState + include JSON::Serializable + + # `.`, 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 diff --git a/drivers/place/driver_health_spec.cr b/drivers/place/driver_health_spec.cr new file mode 100644 index 0000000000..b169da80a1 --- /dev/null +++ b/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