Introduce UtilizationMonitor.

Author: samuel-williams-shopify

async-service-supervisor.gemspec

@@ -26,6 +26,7 @@ Gem::Specification.new do |spec|
 
 	spec.add_dependency "async-bus"
 	spec.add_dependency "async-service", "~> 0.15"
+	spec.add_dependency "async-utilization", "~> 0.3"
 	spec.add_dependency "io-endpoint"
 	spec.add_dependency "memory", "~> 0.7"
 	spec.add_dependency "memory-leak", "~> 0.10"

examples/echo/service.rb

@@ -0,0 +1,102 @@
+#!/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!
+		
+		registry = evaluator.utilization_registry
+		connections_total = registry.metric(:connections_total)
+		connections_active = registry.metric(:connections_active)
+		messages_total = registry.metric(:messages_total)
+		
+		Async do |task|
+			@bound_endpoint.accept do |peer|
+				connections_total.increment
+				connections_active.track do
+					Console.info(self, "Client connected", peer: peer)
+					
+					peer.each_line do |line|
+						messages_total.increment
+						peer.write(line)
+					end
+					
+					Console.info(self, "Client disconnected", peer: peer)
+				end
+			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

@@ -7,6 +7,7 @@
 require "async/service/managed/environment"
 
 require_relative "service"
+require_relative "server"
 
 module Async
 	module Service

lib/async/service/supervisor/environment.rb

@@ -85,11 +85,26 @@ def remove(supervisor_controller)
 					end
 				end
 
+				# The key used when this monitor's status is aggregated with others.
+				def self.monitor_type
+					:memory_monitor
+				end
+				
+				# Serialize memory cluster data for JSON.
+				def as_json
+					@cluster.as_json
+				end
+				
+				# Serialize to JSON string.
+				def to_json(...)
+					as_json.to_json(...)
+				end
+				
 				# Get status for the memory monitor.
 				#
-				# @returns [Hash] Status including the memory cluster.
+				# @returns [Hash] Hash with type and data keys.
 				def status
-					{memory_monitor: @cluster.as_json}
+					{type: self.class.monitor_type, data: as_json}
 				end
 
 				# Invoked when a memory leak is detected.

lib/async/service/supervisor/memory_monitor.rb

@@ -57,11 +57,26 @@ def metrics
 					Process::Metrics::General.capture(ppid: @ppid).transform_values!(&:as_json)
 				end
 
+				# The key used when this monitor's status is aggregated with others.
+				def self.monitor_type
+					:process_monitor
+				end
+				
+				# Serialize process metrics for JSON.
+				def as_json
+					{ppid: @ppid, metrics: self.metrics}
+				end
+				
+				# Serialize to JSON string.
+				def to_json(...)
+					as_json.to_json(...)
+				end
+				
 				# Get status for the process monitor.
 				#
-				# @returns [Hash] Status including process metrics.
+				# @returns [Hash] Hash with type and data keys.
 				def status
-					{process_monitor: {ppid: @ppid, metrics: self.metrics}}
+					{type: self.class.monitor_type, data: as_json}
 				end
 
 				# Run the process monitor.

lib/async/service/supervisor/process_monitor.rb

@@ -4,6 +4,9 @@
 # Copyright, 2025, by Samuel Williams.
 
 require "async/service/environment"
+require "async/utilization"
+
+require_relative "worker"
 
 module Async
 	module Service
@@ -30,10 +33,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.