class Monitor

parameter　bind　String | Nil

The optional address for the discovery server.

parameter　delegate　Delegate

The delegate used to map supervisor state into Envoy endpoints.

parameter　control_plane　Async::GRPC::XDS::ControlPlane

The xDS control plane to update.

parameter　management_cluster　String

The static Envoy cluster used to reach this discovery server.

parameter　publish_clusters　Boolean

Whether to publish derived cluster configuration through CDS.

parameter　health_checks　Array(Envoy::Config::Core::V3::HealthCheck)

The active health checks applied to published clusters.

parameter　orca　Boolean

Whether to collect and serve per-worker ORCA load reports.

parameter　processor　Process::Metrics::Processor | Nil

The optional process CPU sampler.

parameter　utilization_monitor　Async::Service::Supervisor::UtilizationMonitor | Nil

The per-worker utilization monitor used for ORCA reporting.

parameter　interval　Numeric

The endpoint reconciliation and ORCA reporting interval in seconds.

def initialize(
	bind: nil,
	delegate: Delegate.new,
	control_plane: Async::GRPC::XDS::ControlPlane.new,
	management_cluster: "xds_cluster",
	publish_clusters: true,
	health_checks: [],
	orca: false,
	processor: nil,
	utilization_monitor: nil,
	interval: 1,
	**options
)
	super(interval: interval, **options)
	
	@bind = bind
	@delegate = delegate
	@control_plane = control_plane
	@eds_config = Async::GRPC::XDS::ConfigSource.grpc(management_cluster)
	@publish_clusters = publish_clusters
	@health_checks = health_checks
	@interval = interval
	@orca = orca
	@controllers = {}
	@published_clusters = {}
	@published_endpoints = {}
	@mutex = Mutex.new
	
	if @orca
		raise ArgumentError, "ORCA reporting requires a TCP bind address!" unless @bind
		raise ArgumentError, "ORCA reporting requires a utilization monitor!" unless utilization_monitor
		
		@orca_port = server_endpoint.url.port
		raise ArgumentError, "ORCA reporting requires a fixed TCP port!" unless @orca_port&.positive?
		
		@processor = processor || Process::Metrics::Processor.new
		@utilization_monitor = utilization_monitor
		@request_totals = {}
		@load_reports = {}
		@authorities = {}
	end
end

attribute　Async::GRPC::XDS::ControlPlane

The xDS control plane receiving cluster and endpoint updates.

attribute　Delegate

The delegate used to map supervisor state into Envoy endpoints.

parameter　supervisor_controller　Object

The supervisor controller describing the worker.

returns　void

def register(supervisor_controller)
	@mutex.synchronize do
		@controllers[supervisor_controller.id] = supervisor_controller
		@authorities[worker_hostname(supervisor_controller)] = supervisor_controller.id if @orca
		reconcile
	end
end

parameter　supervisor_controller　Object

The supervisor controller describing the worker.

returns　void

def remove(supervisor_controller)
	@mutex.synchronize do
		@controllers.delete(supervisor_controller.id)
		if @orca
			hostname = worker_hostname(supervisor_controller)
			@authorities.delete(hostname)
			@load_reports.delete(hostname)
			@request_totals.delete(supervisor_controller.id)
		end
		reconcile
	end
end

parameter　parent　Async::Task

The parent task used for the server.

returns　Async::Task

The monitor task.

def run(parent: Async::Task.current)
	task = super(parent: parent)
	
	if @bind
		parent.async do
			services = [Async::GRPC::XDS::EndpointDiscoveryService]
			services.unshift(Async::GRPC::XDS::ClusterDiscoveryService) if @publish_clusters
			
			server = Async::GRPC::XDS::Server.new(
				@control_plane,
				services: services
			)
			server.dispatcher.register(ORCAService.new(self, minimum_interval: @interval)) if @orca
			server.run(server_endpoint)
		end
	end
	
	task
end

returns　Hash

The clusters and endpoint hashes.

def as_json
	@mutex.synchronize do
		{
			clusters: build_clusters
		}
	end
end

parameter　authority　String

The gRPC request authority.

returns　Boolean

Whether the authority identifies a live worker.

def worker?(authority)
	return false unless @orca
	
	@mutex.synchronize{@authorities.key?(authority)}
end

parameter　authority　String

The gRPC request authority.

returns　Xds::Data::Orca::V3::OrcaLoadReport | Nil

The latest valid report, if available.

def load_report(authority)
	return unless @orca
	
	@mutex.synchronize{@load_reports[authority]}
end

returns　void

def run_once
	sample_load_reports if @orca
	
	@mutex.synchronize do
		reconcile
	end
end