Introduce UtilizationMonitor.

Author: samuel-williams-shopify

async-service-supervisor.gemspec

@@ -26,8 +26,10 @@ Gem::Specification.new do |spec|
 
 	spec.add_dependency "async-bus"
 	spec.add_dependency "async-service", "~> 0.15"
+	spec.add_dependency "async-utilization", "~> 0.2"
 	spec.add_dependency "io-endpoint"
 	spec.add_dependency "memory", "~> 0.7"
 	spec.add_dependency "memory-leak", "~> 0.10"
 	spec.add_dependency "process-metrics"
+	spec.add_dependency "thread-local"
 end

examples/echo/service.rb

@@ -0,0 +1,97 @@
+#!/usr/bin/env async-service
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2025, by Samuel Williams.
+
+require "async/service/supervisor"
+require "async/service/managed/service"
+require "async/utilization"
+require "io/endpoint/host_endpoint"
+
+class EchoService < Async::Service::Managed::Service
+	def initialize(...)
+		super
+		
+		@bound_endpoint = nil
+		@endpoint = nil
+	end
+	
+	# Prepare the bound endpoint for the server.
+	def start
+		@endpoint = IO::Endpoint.tcp("127.0.0.1", 8080)
+		
+		Sync do
+			@bound_endpoint = @endpoint.bound
+		end
+		
+		Console.info(self, "Starting echo server on #{@endpoint}")
+		
+		super
+	end
+	
+	def run(instance, evaluator)
+		evaluator.prepare!(instance)
+		
+		instance.ready!
+		
+		Async do |task|
+			@bound_endpoint.accept do |peer|
+				Async::Utilization.increment(:total_connections)
+				Async::Utilization.increment(:active_connections) do
+					Console.info(self, "Client connected", peer: peer)
+					
+					peer.each_line do |line|
+						Async::Utilization.increment(:total_messages)
+						peer.write(line)
+					end
+				end
+				
+				Console.info(self, "Client disconnected", peer: peer)
+			end
+		end
+		
+		# Return the bound endpoint for health checking
+		@bound_endpoint
+	end
+	
+	# Close the bound endpoint.
+	def stop(...)
+		if @bound_endpoint
+			@bound_endpoint.close
+			@bound_endpoint = nil
+		end
+		
+		@endpoint = nil
+		
+		super
+	end
+end
+
+service "echo" do
+	include Async::Service::Managed::Environment
+	include Async::Service::Supervisor::Supervised
+	
+	service_class EchoService
+	
+	utilization_schema do
+		{
+			connections_total: :u64,
+			connections_active: :u32,
+			messages_total: :u64
+		}
+	end
+end
+
+service "supervisor" do
+	include Async::Service::Supervisor::Environment
+	
+	monitors do
+		[
+			Async::Service::Supervisor::UtilizationMonitor.new(
+				path: "utilization.shm",
+				interval: 1
+			)
+		]
+	end
+end

examples/echo/utilization.shm

@@ -12,6 +12,7 @@
 
 require_relative "supervisor/memory_monitor"
 require_relative "supervisor/process_monitor"
+require_relative "supervisor/utilization_monitor"
 
 require_relative "supervisor/environment"
 require_relative "supervisor/supervised"

lib/async/service/supervisor.rb

@@ -4,6 +4,7 @@
 # Copyright, 2025, by Samuel Williams.
 
 require "async/service/environment"
+require "async/utilization"
 
 module Async
 	module Service
@@ -30,10 +31,38 @@ def supervisor_worker_state
 					{name: self.name}
 				end
 
+				# A default schema for utilization metrics.
+				# @returns [Hash | Nil] The utilization schema or nil if utilization is disabled.
+				def utilization_schema
+					{
+						connections_active: :u32,
+						connections_total: :u64,
+						requests_active: :u32,
+						requests_total: :u64,
+					}
+				end
+				
+				# Get the utilization registry for this service.
+				#
+				# Creates a new registry instance for tracking utilization metrics.
+				# This registry is used by workers to emit metrics that can be collected
+				# by the supervisor's utilization monitor.
+				#
+				# @returns [Async::Utilization::Registry] A new utilization registry instance.
+				def utilization_registry
+					Async::Utilization::Registry.new
+				end
+				
 				# The supervised worker for the current process.
 				# @returns [Worker] The worker client.
 				def supervisor_worker
-					Worker.new(process_id: Process.pid, endpoint: supervisor_endpoint, state: self.supervisor_worker_state)
+					Worker.new(
+						process_id: Process.pid,
+						endpoint: supervisor_endpoint,
+						state: self.supervisor_worker_state,
+						utilization_schema: self.utilization_schema,
+						utilization_registry: self.utilization_registry,
+					)
 				end
 
 				# Create a supervised worker for the given instance.