diff --git a/async-service-supervisor-envoy.gemspec b/async-service-supervisor-envoy.gemspec index fa60084..a1ab9e1 100644 --- a/async-service-supervisor-envoy.gemspec +++ b/async-service-supervisor-envoy.gemspec @@ -6,7 +6,7 @@ Gem::Specification.new do |spec| spec.name = "async-service-supervisor-envoy" spec.version = Async::Service::Supervisor::Envoy::VERSION - spec.summary = "Envoy xDS monitor for async-service-supervisor." + spec.summary = "Envoy endpoint discovery monitor for async-service-supervisor." spec.authors = ["Samuel Williams"] spec.license = "MIT" diff --git a/context/getting-started.md b/context/getting-started.md index c1ea351..3b93541 100644 --- a/context/getting-started.md +++ b/context/getting-started.md @@ -1,6 +1,6 @@ # Getting Started -This guide explains how to use `async-service-supervisor-envoy` to publish supervised worker endpoints to Envoy using xDS. +This guide explains how to use `async-service-supervisor-envoy` to publish supervised worker endpoints to Envoy using EDS. ## Installation @@ -16,10 +16,13 @@ The gem depends on `async-service-supervisor` and `async-grpc-xds`. `async-service-supervisor-envoy` provides: - - {ruby Async::Service::Supervisor::Envoy::Monitor} - A supervisor monitor that publishes worker endpoints through xDS. + - {ruby Async::Service::Supervisor::Envoy::Monitor} - A supervisor monitor that publishes worker endpoints through EDS. + - {ruby Async::Service::Supervisor::Envoy::EndpointDiscoveryService} - The gRPC endpoint discovery service Envoy subscribes to. - {ruby Async::Service::Supervisor::Envoy::Endpoint} - A small value object for endpoint state. -The monitor runs an xDS control plane endpoint. Envoy connects to it using ADS and receives CDS/EDS updates derived from supervisor worker state. +The monitor serves Envoy's Endpoint Discovery Service. Envoy subscribes to it and receives endpoint assignments derived from supervisor worker state. + +Endpoints are the only resource that changes as workers come and go. Clusters, listeners and routes are static for a given deployment, so the monitor does not publish them, and it does not use the aggregated discovery service. Configure them in Envoy directly, or with a separate control plane which is then free to use ADS. ## Endpoint State @@ -68,6 +71,47 @@ Async::Service::Supervisor::Envoy::Monitor.new( By default, workers are grouped into clusters by `state[:name]`. +## Envoy Configuration + +Declare each cluster in Envoy and point its endpoint configuration at the monitor: + +``` yaml +static_resources: + clusters: + - name: myservice + connect_timeout: 1s + type: EDS + http2_protocol_options: {} + eds_cluster_config: + service_name: myservice + eds_config: + resource_api_version: V3 + api_config_source: + api_type: GRPC + transport_api_version: V3 + grpc_services: + - envoy_grpc: + cluster_name: supervisor_eds + + - name: supervisor_eds + connect_timeout: 1s + type: STRICT_DNS + http2_protocol_options: {} + load_assignment: + cluster_name: supervisor_eds + endpoints: + - lb_endpoints: + - endpoint: + address: + socket_address: + address: 127.0.0.1 + port_value: 18000 +``` + +`service_name` must match the cluster name the monitor publishes. Everything else about the cluster is yours: set `http2_protocol_options` when your workers serve HTTP/2, add a transport socket for TLS upstreams, and declare any active health checks. The monitor does not inspect or constrain them. + +If clusters are delivered by a separate control plane rather than declared statically, use the same `eds_cluster_config` block. Do not use `eds_config: {ads: {}}`, which would route the subscription to that control plane instead of the monitor. + ## Custom Mapping You can customize cluster grouping, endpoint selection, and health with a delegate: @@ -113,10 +157,29 @@ utilization_monitor = Async::Service::Supervisor::UtilizationMonitor.new(interva ] ``` -The supervisor utilization monitor manages each worker's shared-memory allocation and registration. The Envoy monitor samples it through `sample_by_worker`, combines each worker's `requests_total` counter with processor usage from `process-metrics`, and serves the resulting ORCA reports from the same HTTP/2 endpoint as ADS. It also configures each discovered cluster to use Envoy's client-side weighted-round-robin policy. +The supervisor utilization monitor manages each worker's shared-memory allocation and registration. The Envoy monitor samples it through `sample_by_worker`, combines each worker's `requests_total` counter with processor usage from `process-metrics`, and serves the resulting ORCA reports from the same HTTP/2 endpoint as endpoint discovery. The first sample establishes a baseline. Subsequent reports contain normalized `cpu_utilization` and `rps_fractional` values for each worker. Reports are removed immediately when a worker disconnects. +Workers are identified by the `hostname` published with each endpoint, which Envoy sends as the request authority when it opens an out-of-band reporting stream. Enabling ORCA therefore publishes one endpoint per worker rather than one per shared listener. + +Configure the cluster to use client-side weighted round-robin, pointing at the monitor's port: + +``` yaml +load_balancing_policy: + policies: + - typed_extension_config: + name: envoy.load_balancing_policies.client_side_weighted_round_robin + typed_config: + "@type": type.googleapis.com/envoy.extensions.load_balancing_policies.client_side_weighted_round_robin.v3.ClientSideWeightedRoundRobin + enable_oob_load_report: true + oob_reporting_period: 1s + oob_reporting_config: + port_value: 18000 +``` + +Set `oob_reporting_period` to the monitor's `interval`. Envoy dials this port on each endpoint's own address, which reaches the monitor because it shares a network namespace with the workers. + Out-of-band ORCA requires: - Envoy 1.39 or later. diff --git a/context/index.yaml b/context/index.yaml index 5e12684..f33dc9d 100644 --- a/context/index.yaml +++ b/context/index.yaml @@ -1,7 +1,7 @@ # Automatically generated context index for Utopia::Project guides. # Do not edit then files in this directory directly, instead edit the guides and then run `bake utopia:project:agent:context:update`. --- -description: Envoy xDS monitor for async-service-supervisor. +description: Envoy endpoint discovery monitor for async-service-supervisor. metadata: documentation_uri: https://socketry.github.io/async-service-supervisor-envoy/ source_code_uri: https://github.com/socketry/async-service-supervisor-envoy.git @@ -9,4 +9,4 @@ files: - path: getting-started.md title: Getting Started description: This guide explains how to use `async-service-supervisor-envoy` to - publish supervised worker endpoints to Envoy using xDS. + publish supervised worker endpoints to Envoy using EDS. diff --git a/control-plane/envoy.yaml b/control-plane/envoy.yaml index 205858f..9b8b891 100644 --- a/control-plane/envoy.yaml +++ b/control-plane/envoy.yaml @@ -8,14 +8,6 @@ admin: address: 0.0.0.0 port_value: 19000 -dynamic_resources: - ads_config: - api_type: GRPC - transport_api_version: V3 - grpc_services: - - envoy_grpc: - cluster_name: xds_cluster - static_resources: listeners: - name: listener_http @@ -45,6 +37,8 @@ static_resources: "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router clusters: + # The cluster is static. Only its endpoints are discovered, from the + # supervisor's endpoint discovery service. - name: app-http1 connect_timeout: 1s type: EDS @@ -52,13 +46,19 @@ static_resources: eds_cluster_config: service_name: app-http1 eds_config: - ads: {} resource_api_version: V3 - - name: xds_cluster + api_config_source: + api_type: GRPC + transport_api_version: V3 + grpc_services: + - envoy_grpc: + cluster_name: supervisor_eds + + - name: supervisor_eds connect_timeout: 1s type: STRICT_DNS load_assignment: - cluster_name: xds_cluster + cluster_name: supervisor_eds endpoints: - lb_endpoints: - endpoint: diff --git a/control-plane/readme.md b/control-plane/readme.md index 8a3c72c..fecca37 100644 --- a/control-plane/readme.md +++ b/control-plane/readme.md @@ -4,11 +4,13 @@ This scenario exercises the intended Envoy control plane topology: - Falcon workers register with `Async::Service::Supervisor::Worker`. - Each worker publishes `state[:endpoint]` with its name, scheme, supported protocols, and concrete addresses. -- `Async::Service::Supervisor::Envoy::Monitor` maps supervisor state into xDS endpoint resources. -- Envoy connects to the supervisor's xDS server and subscribes to endpoint updates. +- `Async::Service::Supervisor::Envoy::Monitor` maps supervisor state into Envoy endpoint assignments. +- Envoy connects to the supervisor's endpoint discovery service and subscribes to endpoint updates. - Envoy routes HTTP traffic to the supervised Falcon workers. -Envoy initiates the xDS connection. The supervisor does not call Envoy's admin API or mutate Envoy directly. Once Envoy has connected and subscribed, the supervisor streams updates over that connection. This matches the normal xDS control plane model and gives Envoy ownership of reconnects, resource ACK/NACK handling, and sidecar lifecycle. +The cluster itself is declared statically in `envoy.yaml`; only its endpoints are discovered. + +Envoy initiates the connection. The supervisor does not call Envoy's admin API or mutate Envoy directly. Once Envoy has connected and subscribed, the supervisor streams updates over that connection. This matches the normal xDS control plane model and gives Envoy ownership of reconnects, resource ACK/NACK handling, and sidecar lifecycle. ## Running Tests @@ -26,9 +28,9 @@ $ docker compose -f control-plane/docker-compose.yaml down --remove-orphans The test verifies the happy path for the desired architecture: -- The supervisor can run an xDS server. +- The supervisor can run an endpoint discovery server. - Supervised Falcon workers can publish endpoints. -- Envoy can subscribe to those endpoints using ADS-backed EDS. +- Envoy can subscribe to those endpoints using EDS. - Envoy can load balance requests across the supervised workers. This is a framework for lifecycle testing rather than the complete production story. Follow-up cases should cover worker removal, worker recovery, health changes, and xDS stream reconnects. diff --git a/control-plane/test/envoy.rb b/control-plane/test/envoy.rb index 76f2258..2d33f25 100644 --- a/control-plane/test/envoy.rb +++ b/control-plane/test/envoy.rb @@ -47,18 +47,25 @@ def eventually(timeout: 20, interval: 0.5) expect(responses.filter_map{|response| response["x-backend-id"]}.uniq.sort).to be == ["backend-a", "backend-b"] end - it "loads the xDS cluster from the supervisor monitor" do + it "discovers cluster endpoints from the supervisor monitor" do uri = admin_uri + "/clusters?format=json" - cluster_status = eventually do + host_statuses = eventually do if (response = Net::HTTP.get_response(uri)).code.to_i == 200 clusters = JSON.parse(response.body) - clusters.fetch("cluster_statuses").find do |cluster| + cluster_status = clusters.fetch("cluster_statuses").find do |cluster| cluster.fetch("name") == "app-http1" end + + hosts = cluster_status&.fetch("host_statuses", nil) + hosts if hosts&.size == 2 end end - expect(cluster_status).not.to be_nil + addresses = host_statuses.map do |host| + host.fetch("address").fetch("socket_address").fetch("port_value") + end + + expect(addresses.sort).to be == [9292, 9293] end end diff --git a/guides/getting-started/readme.md b/guides/getting-started/readme.md index c1ea351..3b93541 100644 --- a/guides/getting-started/readme.md +++ b/guides/getting-started/readme.md @@ -1,6 +1,6 @@ # Getting Started -This guide explains how to use `async-service-supervisor-envoy` to publish supervised worker endpoints to Envoy using xDS. +This guide explains how to use `async-service-supervisor-envoy` to publish supervised worker endpoints to Envoy using EDS. ## Installation @@ -16,10 +16,13 @@ The gem depends on `async-service-supervisor` and `async-grpc-xds`. `async-service-supervisor-envoy` provides: - - {ruby Async::Service::Supervisor::Envoy::Monitor} - A supervisor monitor that publishes worker endpoints through xDS. + - {ruby Async::Service::Supervisor::Envoy::Monitor} - A supervisor monitor that publishes worker endpoints through EDS. + - {ruby Async::Service::Supervisor::Envoy::EndpointDiscoveryService} - The gRPC endpoint discovery service Envoy subscribes to. - {ruby Async::Service::Supervisor::Envoy::Endpoint} - A small value object for endpoint state. -The monitor runs an xDS control plane endpoint. Envoy connects to it using ADS and receives CDS/EDS updates derived from supervisor worker state. +The monitor serves Envoy's Endpoint Discovery Service. Envoy subscribes to it and receives endpoint assignments derived from supervisor worker state. + +Endpoints are the only resource that changes as workers come and go. Clusters, listeners and routes are static for a given deployment, so the monitor does not publish them, and it does not use the aggregated discovery service. Configure them in Envoy directly, or with a separate control plane which is then free to use ADS. ## Endpoint State @@ -68,6 +71,47 @@ Async::Service::Supervisor::Envoy::Monitor.new( By default, workers are grouped into clusters by `state[:name]`. +## Envoy Configuration + +Declare each cluster in Envoy and point its endpoint configuration at the monitor: + +``` yaml +static_resources: + clusters: + - name: myservice + connect_timeout: 1s + type: EDS + http2_protocol_options: {} + eds_cluster_config: + service_name: myservice + eds_config: + resource_api_version: V3 + api_config_source: + api_type: GRPC + transport_api_version: V3 + grpc_services: + - envoy_grpc: + cluster_name: supervisor_eds + + - name: supervisor_eds + connect_timeout: 1s + type: STRICT_DNS + http2_protocol_options: {} + load_assignment: + cluster_name: supervisor_eds + endpoints: + - lb_endpoints: + - endpoint: + address: + socket_address: + address: 127.0.0.1 + port_value: 18000 +``` + +`service_name` must match the cluster name the monitor publishes. Everything else about the cluster is yours: set `http2_protocol_options` when your workers serve HTTP/2, add a transport socket for TLS upstreams, and declare any active health checks. The monitor does not inspect or constrain them. + +If clusters are delivered by a separate control plane rather than declared statically, use the same `eds_cluster_config` block. Do not use `eds_config: {ads: {}}`, which would route the subscription to that control plane instead of the monitor. + ## Custom Mapping You can customize cluster grouping, endpoint selection, and health with a delegate: @@ -113,10 +157,29 @@ utilization_monitor = Async::Service::Supervisor::UtilizationMonitor.new(interva ] ``` -The supervisor utilization monitor manages each worker's shared-memory allocation and registration. The Envoy monitor samples it through `sample_by_worker`, combines each worker's `requests_total` counter with processor usage from `process-metrics`, and serves the resulting ORCA reports from the same HTTP/2 endpoint as ADS. It also configures each discovered cluster to use Envoy's client-side weighted-round-robin policy. +The supervisor utilization monitor manages each worker's shared-memory allocation and registration. The Envoy monitor samples it through `sample_by_worker`, combines each worker's `requests_total` counter with processor usage from `process-metrics`, and serves the resulting ORCA reports from the same HTTP/2 endpoint as endpoint discovery. The first sample establishes a baseline. Subsequent reports contain normalized `cpu_utilization` and `rps_fractional` values for each worker. Reports are removed immediately when a worker disconnects. +Workers are identified by the `hostname` published with each endpoint, which Envoy sends as the request authority when it opens an out-of-band reporting stream. Enabling ORCA therefore publishes one endpoint per worker rather than one per shared listener. + +Configure the cluster to use client-side weighted round-robin, pointing at the monitor's port: + +``` yaml +load_balancing_policy: + policies: + - typed_extension_config: + name: envoy.load_balancing_policies.client_side_weighted_round_robin + typed_config: + "@type": type.googleapis.com/envoy.extensions.load_balancing_policies.client_side_weighted_round_robin.v3.ClientSideWeightedRoundRobin + enable_oob_load_report: true + oob_reporting_period: 1s + oob_reporting_config: + port_value: 18000 +``` + +Set `oob_reporting_period` to the monitor's `interval`. Envoy dials this port on each endpoint's own address, which reaches the monitor because it shares a network namespace with the workers. + Out-of-band ORCA requires: - Envoy 1.39 or later. diff --git a/lib/async/service/supervisor/envoy.rb b/lib/async/service/supervisor/envoy.rb index d0b53b0..a9b4d8a 100644 --- a/lib/async/service/supervisor/envoy.rb +++ b/lib/async/service/supervisor/envoy.rb @@ -6,6 +6,7 @@ require_relative "envoy/version" require_relative "envoy/delegate" require_relative "envoy/endpoint" +require_relative "envoy/endpoint_discovery_service" require_relative "envoy/monitor" require_relative "envoy/orca_service" require_relative "envoy/supervised" diff --git a/lib/async/service/supervisor/envoy/endpoint_discovery_service.rb b/lib/async/service/supervisor/envoy/endpoint_discovery_service.rb new file mode 100644 index 0000000..e02792e --- /dev/null +++ b/lib/async/service/supervisor/envoy/endpoint_discovery_service.rb @@ -0,0 +1,86 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "async/grpc/service" +require "async/grpc/xds/control_plane" +require "async/grpc/xds/service" +require "envoy/service/discovery/v3/discovery_pb" +require "protocol/grpc/interface" + +module Async + module Service + module Supervisor + module Envoy + # Serves Envoy Endpoint Discovery Service requests from a control plane. + # + # The supervisor only ever publishes endpoint assignments, so it serves EDS + # directly rather than aggregating resource types. This leaves the ADS + # transport unclaimed for cluster, listener and route configuration. + class EndpointDiscoveryService < Async::GRPC::Service + SERVICE_NAME = "envoy.service.endpoint.v3.EndpointDiscoveryService" + + # The only resource type this service serves. + RESOURCE_TYPE = Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE + + # Interface definition for the Endpoint Discovery Service. + class Interface < Protocol::GRPC::Interface + rpc :StreamEndpoints, + request_class: ::Envoy::Service::Discovery::V3::DiscoveryRequest, + response_class: ::Envoy::Service::Discovery::V3::DiscoveryResponse, + streaming: :bidirectional + end + + # Initialize an Endpoint Discovery Service. + # @parameter control_plane [Async::GRPC::XDS::ControlPlane] The control plane that provides endpoint assignments. + def initialize(control_plane) + super(Interface, SERVICE_NAME) + + @control_plane = control_plane + end + + # Serve a state-of-the-world Endpoint Discovery Service stream. + # @parameter input [Enumerable] The stream of discovery requests. + # @parameter output [Interface(:write)] The discovery response stream. + # @parameter call [Protocol::GRPC::Call] The gRPC call context. + # @asynchronous + def stream_endpoints(input, output, call) + stream = Async::GRPC::XDS::Service::Stream.new(@control_plane, output) + @control_plane.register_stream(stream) + + reader = Async do + input.each do |request| + stream.request(normalize(request)) + end + end + + writer = Async do + stream.run + end + + reader.wait + ensure + stream&.close + reader&.stop + writer&.stop + @control_plane.remove_stream(stream) if stream + end + + private + + # Pin a request to this service's resource type. A dedicated discovery + # service implies its own type, so clients may omit it, and a client + # cannot use this stream to subscribe to anything else. + def normalize(request) + return request if request.type_url == RESOURCE_TYPE + + request = request.dup + request.type_url = RESOURCE_TYPE + request + end + end + end + end + end +end diff --git a/lib/async/service/supervisor/envoy/monitor.rb b/lib/async/service/supervisor/envoy/monitor.rb index 433b9bf..e2663b7 100644 --- a/lib/async/service/supervisor/envoy/monitor.rb +++ b/lib/async/service/supervisor/envoy/monitor.rb @@ -3,17 +3,18 @@ # Released under the MIT License. # Copyright, 2026, by Samuel Williams. +require "async/grpc/dispatcher" +require "async/grpc/xds/control_plane" require "async/http/endpoint" +require "async/http/server" require "async/service/supervisor/monitor" require "async/service/supervisor/utilization_monitor" -require "async/grpc/xds/client_side_weighted_round_robin" -require "async/grpc/xds/control_plane" -require "async/grpc/xds/server" require "process/metrics" require "xds/data/orca/v3/orca_load_report_pb" require_relative "delegate" require_relative "endpoint" +require_relative "endpoint_discovery_service" require_relative "endpoint_group" require_relative "orca_service" @@ -22,13 +23,17 @@ module Service module Supervisor # Provides Envoy integration for supervisor-managed services. module Envoy - # Represents a supervisor monitor that publishes worker endpoints to Envoy using xDS. + # Represents a supervisor monitor that publishes worker endpoints to Envoy using EDS. + # + # Endpoints are the only resource that changes as workers come and go, so the + # monitor publishes endpoint assignments and nothing else. Clusters, listeners + # and routes are static per deployment and belong in Envoy's configuration, or + # in a separate control plane which is free to use the ADS transport. class Monitor < Async::Service::Supervisor::Monitor # Initialize the monitor. - # @parameter bind [String | Nil] The optional address for the xDS control plane server. + # @parameter bind [String | Nil] The optional address for the endpoint 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 health_checks [Array(Envoy::Config::Core::V3::HealthCheck)] The active health checks applied to published clusters. + # @parameter control_plane [Async::GRPC::XDS::ControlPlane] The control plane to update. # @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. @@ -37,7 +42,6 @@ def initialize( bind: nil, delegate: Delegate.new, control_plane: Async::GRPC::XDS::ControlPlane.new, - health_checks: [], orca: false, processor: nil, utilization_monitor: nil, @@ -49,20 +53,18 @@ def initialize( @bind = bind @delegate = delegate @control_plane = control_plane - @health_checks = health_checks @interval = interval @orca = orca @controllers = {} - @published_clusters = {} - @server_task = nil + @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? + port = server_endpoint.url.port + raise ArgumentError, "ORCA reporting requires a fixed TCP port!" unless port&.positive? @processor = processor || Process::Metrics::Processor.new @utilization_monitor = utilization_monitor @@ -72,7 +74,7 @@ def initialize( end end - # @attribute [Async::GRPC::XDS::ControlPlane] The xDS control plane receiving cluster and endpoint updates. + # @attribute [Async::GRPC::XDS::ControlPlane] The control plane receiving endpoint updates. attr :control_plane # @attribute [Delegate] The delegate used to map supervisor state into Envoy endpoints. @@ -105,17 +107,26 @@ def remove(supervisor_controller) end end - # Run the monitor and optional xDS server task. - # @parameter parent [Async::Task] The parent task used for the xDS server. + # Build the gRPC services published on the bind address. + # @returns [Array(Async::GRPC::Service)] The endpoint discovery service, and the ORCA service if enabled. + def services + services = [EndpointDiscoveryService.new(@control_plane)] + services << ORCAService.new(self, minimum_interval: @interval) if @orca + services + end + + # Run the monitor and optional endpoint discovery server task. + # @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 - @server_task = parent.async do - server = Async::GRPC::XDS::Server.new(@control_plane) - server.dispatcher.register(ORCAService.new(self, minimum_interval: @interval)) if @orca - server.run(server_endpoint) + dispatcher = Async::GRPC::Dispatcher.new + services.each{|service| dispatcher.register(service)} + + parent.async do + Async::HTTP::Server.new(dispatcher, server_endpoint).run end end @@ -189,20 +200,15 @@ def build_records(supervisor_controller) end def reconcile - records_by_cluster = build_records_by_cluster - clusters = build_clusters(records_by_cluster) + clusters = build_clusters - records_by_cluster.each do |cluster, records| - configuration = cluster_configuration(records) + # Skipping an identical assignment only avoids a redundant version bump: + (@published_endpoints.keys | clusters.keys).each do |cluster| + endpoints = clusters.fetch(cluster, []) + next if @published_endpoints[cluster] == endpoints - unless @published_clusters[cluster] == configuration - @control_plane.update_cluster(cluster, **configuration) - @published_clusters[cluster] = configuration - end - end - - (@published_clusters.keys | clusters.keys).each do |cluster| - @control_plane.update_endpoints(cluster, clusters.fetch(cluster, [])) + @control_plane.update_endpoints(cluster, endpoints) + @published_endpoints[cluster] = endpoints end end @@ -218,8 +224,11 @@ def build_clusters(records_by_cluster = build_records_by_cluster) records_by_cluster.transform_values do |records| if @orca records.map do |record| + # Envoy sends the endpoint hostname as the request authority when + # it opens an out-of-band reporting stream, which is how a single + # ORCA service identifies which worker a report is for: { - addresses: record[:endpoint].addresses, + addresses: orca_addresses(record[:endpoint]), healthy: record[:healthy], hostname: worker_hostname(record[:worker]), } @@ -237,32 +246,14 @@ def build_clusters(records_by_cluster = build_records_by_cluster) end end - def cluster_configuration(records) - schemes = records.map{|record| record[:endpoint].scheme}.uniq - protocols = records.map{|record| record[:endpoint].protocols} - common_protocols = protocols.reduce{|common, names| common & names} - - raise ArgumentError, "Envoy cluster contains incompatible schemes: #{schemes.inspect}" if schemes.size > 1 - raise ArgumentError, "Envoy cluster contains no common protocols: #{protocols.inspect}" if common_protocols.empty? - raise ArgumentError, "HTTPS upstream endpoints are not yet supported!" if schemes.first == :https - - configuration = { - protocol: envoy_protocol(common_protocols), - health_checks: @health_checks, - } - - if @orca - if records.any?{|record| record[:endpoint].addresses.any?{|address| address[:path]}} - raise ArgumentError, "Out-of-band ORCA reporting requires IP endpoints!" - end - - configuration[:load_balancing_policy] = Async::GRPC::XDS::ClientSideWeightedRoundRobin.build( - @orca_port, - reporting_period: @interval - ) + # Out-of-band ORCA identifies a worker by the address Envoy dials for its + # reporting stream, which a Unix socket cannot provide. + def orca_addresses(endpoint) + if endpoint.addresses.any?{|address| address[:path]} + raise ArgumentError, "Out-of-band ORCA reporting requires IP endpoints!" end - configuration + endpoint.addresses end def sample_load_reports @@ -315,19 +306,6 @@ def sample_load_reports ) end end - - def envoy_protocol(protocols) - protocols.each do |protocol| - case protocol - when "h2" - return :http2 - when "http/1.1", "http/1.0" - return :http1 - end - end - - raise ArgumentError, "Envoy cluster contains no supported protocols: #{protocols.inspect}" - end end end end diff --git a/readme.md b/readme.md index 56d214d..3b090c0 100644 --- a/readme.md +++ b/readme.md @@ -1,6 +1,6 @@ # Async::Service::Supervisor::Envoy -Provides an Envoy xDS monitor for `async-service-supervisor`. +Provides an Envoy endpoint discovery monitor for `async-service-supervisor`. [![Development Status](https://github.com/socketry/async-service-supervisor-envoy/workflows/Test/badge.svg)](https://github.com/socketry/async-service-supervisor-envoy/actions?workflow=Test) @@ -8,7 +8,7 @@ Provides an Envoy xDS monitor for `async-service-supervisor`. `async-service-supervisor-envoy` publishes supervised worker endpoints to Envoy: - - **xDS control plane** - Runs an ADS server backed by `async-grpc-xds`. + - **Endpoint discovery** - Runs an EDS server backed by `async-grpc-xds`, leaving the ADS transport free for cluster, listener and route configuration. - **Supervisor integration** - Registers and removes endpoints from supervisor worker lifecycle events. - **Multiple clusters** - Groups workers by `state[:name]` by default. - **Endpoint contract** - Converts concrete post-bind worker listeners into Envoy upstream endpoints, including grouped IP or Unix socket addresses. diff --git a/releases.md b/releases.md index 49c886d..c8345c1 100644 --- a/releases.md +++ b/releases.md @@ -2,6 +2,11 @@ ## Unreleased + - Serve endpoints from a dedicated `EndpointDiscoveryService` instead of the aggregated discovery service, leaving ADS available for cluster, listener and route configuration. + - Stop publishing cluster resources. Declare clusters in Envoy, or in a separate control plane, including their protocol options, active health checks and load-balancing policy. + - Remove the `health_checks:` monitor option, which only configured published clusters. + - Stop validating endpoint schemes and protocols, which only constrained the cluster resources the monitor no longer publishes. Out-of-band ORCA still rejects Unix socket endpoints, because it cannot identify a worker without a distinct address. + - Publish endpoint assignments only when they change, instead of on every reconciliation. - Use normalized processor utilization from `process-metrics` v0.13. ## v0.4.0 diff --git a/test/async/service/supervisor/envoy/endpoint_discovery_service.rb b/test/async/service/supervisor/envoy/endpoint_discovery_service.rb new file mode 100644 index 0000000..fde621c --- /dev/null +++ b/test/async/service/supervisor/envoy/endpoint_discovery_service.rb @@ -0,0 +1,99 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "async/service/supervisor/envoy/endpoint_discovery_service" +require "envoy/config/endpoint/v3/endpoint_pb" +require "sus/fixtures/async" + +describe Async::Service::Supervisor::Envoy::EndpointDiscoveryService do + include Sus::Fixtures::Async::ReactorContext + + let(:control_plane) {Async::GRPC::XDS::ControlPlane.new} + let(:service) {subject.new(control_plane)} + + # Yields the given requests and then blocks, as a live Envoy stream does: + def input_for(requests) + Object.new.tap do |input| + input.define_singleton_method(:each) do |&block| + requests.each(&block) + Async::Task.current.sleep + end + end + end + + def output + [].tap do |output| + output.define_singleton_method(:write){|value| self << value} + end + end + + def until_written(output) + 100.times do + return if output.any? + Async::Task.current.yield + end + + raise "No response was written!" + end + + def stream(request) + responses = output + + task = Async do + service.stream_endpoints(input_for([request]), responses, nil) + end + + until_written(responses) + task.stop + + responses.first + end + + def assignment_for(response) + Envoy::Config::Endpoint::V3::ClusterLoadAssignment.decode(response.resources.first.value) + end + + it "streams the requested assignment" do + control_plane.update_endpoints("myservice", [ + {addresses: [{address: "127.0.0.1", port: 50051}], healthy: true} + ]) + + response = stream(Envoy::Service::Discovery::V3::DiscoveryRequest.new( + type_url: subject::RESOURCE_TYPE, + resource_names: ["myservice"] + )) + assignment = assignment_for(response) + + expect(response.type_url).to be == subject::RESOURCE_TYPE + expect(assignment.cluster_name).to be == "myservice" + expect(assignment.endpoints.first.lb_endpoints.first.endpoint.address.socket_address.port_value).to be == 50051 + end + + it "serves a subscriber that omits the implied resource type" do + control_plane.update_endpoints("myservice", [ + {addresses: [{address: "127.0.0.1", port: 50051}], healthy: true} + ]) + + response = stream(Envoy::Service::Discovery::V3::DiscoveryRequest.new(resource_names: ["myservice"])) + + expect(response.type_url).to be == subject::RESOURCE_TYPE + expect(assignment_for(response).cluster_name).to be == "myservice" + end + + it "refuses to serve any other resource type" do + control_plane.update_cluster("myservice") + control_plane.update_endpoints("myservice", [ + {addresses: [{address: "127.0.0.1", port: 50051}], healthy: true} + ]) + + response = stream(Envoy::Service::Discovery::V3::DiscoveryRequest.new( + type_url: Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE, + resource_names: ["myservice"] + )) + + expect(response.type_url).to be == subject::RESOURCE_TYPE + expect(assignment_for(response).cluster_name).to be == "myservice" + end +end diff --git a/test/async/service/supervisor/envoy/monitor.rb b/test/async/service/supervisor/envoy/monitor.rb index fdd7d28..d57ffb4 100644 --- a/test/async/service/supervisor/envoy/monitor.rb +++ b/test/async/service/supervisor/envoy/monitor.rb @@ -4,7 +4,6 @@ # Copyright, 2026, by Samuel Williams. require "async/service/supervisor/envoy/monitor" -require "async/grpc/xds/http_health_check" require "envoy/config/endpoint/v3/endpoint_pb" describe Async::Service::Supervisor::Envoy::Monitor do @@ -49,20 +48,17 @@ def processor(samples = []) expect(load_balancer_endpoint.endpoint.address.socket_address.port_value).to be == 50051 end - it "publishes active health checks" do - health_check = Async::GRPC::XDS::HTTPHealthCheck.build("/services/ping") - monitor = subject.new(health_checks: [health_check]) + it "never publishes cluster resources" do controller = Controller.new(1, { endpoint: {name: "myservice", scheme: "http", protocols: ["h2"], addresses: [{address: "127.0.0.1", port: 50051}]} }) monitor.register(controller) - cluster = monitor.control_plane.resources(Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE).first - expect(cluster.health_checks).to be == [health_check] + expect(control_plane.resources(Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE)).to be(:empty?) end - it "publishes ORCA worker identity and load-balancing configuration" do + it "publishes ORCA worker identity" do monitor = subject.new( bind: "http://127.0.0.1:18000", orca: true, @@ -80,15 +76,8 @@ def processor(samples = []) ["myservice"] ) assignment = Envoy::Config::Endpoint::V3::ClusterLoadAssignment.decode(response.resources.first.value) - cluster = monitor.control_plane.resources(Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE).first - typed_configuration = cluster.load_balancing_policy.policies.first.typed_extension_config - configuration = Envoy::Extensions::LoadBalancingPolicies::ClientSideWeightedRoundRobin::V3::ClientSideWeightedRoundRobin.decode( - typed_configuration.typed_config.value - ) expect(assignment.endpoints.first.lb_endpoints.first.endpoint.hostname).to be == "worker-1" - expect(configuration.enable_oob_load_report.value).to be == true - expect(configuration.oob_reporting_config.port_value).to be == 18000 end it "samples per-worker ORCA load reports" do @@ -171,9 +160,6 @@ def processor(samples = []) load_balancer_endpoint = assignment.endpoints.first.lb_endpoints.first expect(load_balancer_endpoint.endpoint.address.pipe.path).to be == "/tmp/falcon.ipc" - - cluster = control_plane.resources(Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE).first - expect(cluster.http2_protocol_options).to be_nil end it "keeps one endpoint's addresses grouped" do @@ -394,78 +380,93 @@ def healthy?(supervisor_controller, endpoint) } end - it "selects the preferred common endpoint protocol" do + it "publishes distinct endpoints reported by several workers" do monitor.register(Controller.new(1, { - endpoint: {name: "myservice", scheme: "http", protocols: ["h2", "http/1.1"], addresses: [{path: "/tmp/one.ipc"}]} + endpoint: {name: "myservice", scheme: "http", protocols: ["h2"], addresses: [{path: "/tmp/one.ipc"}]} })) monitor.register(Controller.new(2, { endpoint: {name: "myservice", scheme: "http", protocols: ["h2"], addresses: [{path: "/tmp/two.ipc"}]} })) - cluster = control_plane.resources(Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE).first - expect(cluster.http2_protocol_options).not.to be_nil + assignment = endpoint_assignment("myservice") + + expect(assignment.endpoints.first.lb_endpoints.size).to be == 2 end - it "rejects endpoints without a common protocol in one cluster" do + it "does not republish unchanged endpoints" do monitor.register(Controller.new(1, { - endpoint: {name: "myservice", scheme: "http", protocols: ["http/1.1"], addresses: [{path: "/tmp/one.ipc"}]} + endpoint: {name: "myservice", scheme: "http", protocols: ["h2"], addresses: [{address: "127.0.0.1", port: 50051}]} })) - expect do - monitor.register(Controller.new(2, { - endpoint: {name: "myservice", scheme: "http", protocols: ["h2"], addresses: [{path: "/tmp/two.ipc"}]} - })) - end.to raise_exception(ArgumentError) + version = control_plane.version(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + monitor.run_once + + expect(control_plane.version(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE)).to be == version end - it "rejects unsupported endpoint protocols" do - expect do - monitor.register(Controller.new(1, { - endpoint: {name: "myservice", scheme: "http", protocols: ["unsupported"], addresses: [{path: "/tmp/one.ipc"}]} - })) - end.to raise_exception(ArgumentError) + it "keeps unchanged assignments in the state of the world when another cluster changes" do + monitor.register(Controller.new(1, { + endpoint: {name: "service-a", scheme: "http", protocols: ["h2"], addresses: [{address: "127.0.0.1", port: 50051}]} + })) + monitor.register(Controller.new(2, { + endpoint: {name: "service-b", scheme: "http", protocols: ["h2"], addresses: [{address: "127.0.0.2", port: 50052}]} + })) + + response = control_plane.response( + Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, + ["service-a", "service-b"] + ) + + names = response.resources.map do |resource| + Envoy::Config::Endpoint::V3::ClusterLoadAssignment.decode(resource.value).cluster_name + end + + expect(names.sort).to be == ["service-a", "service-b"] + end + + it "serves endpoint discovery and nothing else" do + monitor = subject.new(bind: "http://127.0.0.1:18000") + + expect(monitor.services.map(&:service_name)).to be == [ + "envoy.service.endpoint.v3.EndpointDiscoveryService" + ] + end + + it "serves ORCA alongside endpoint discovery" do + monitor = subject.new( + bind: "http://127.0.0.1:18000", + orca: true, + processor: processor, + utilization_monitor: utilization_monitor + ) + + expect(monitor.services.map(&:service_name)).to be == [ + "envoy.service.endpoint.v3.EndpointDiscoveryService", + "xds.service.orca.v3.OpenRcaService" + ] end - it "runs an xDS server when bound" do + it "runs a server task only when bound" do parent = Class.new do def initialize - @count = 0 + @blocks = [] end - def async(&block) - @count += 1 - - if @count == 1 - :monitor_task - else - block.call - :server_task - end - end - end.new - - calls = [] - original_server = Async::GRPC::XDS.send(:remove_const, :Server) - - fake_server = Class.new do - define_method(:initialize) do |control_plane| - calls << [:initialize, control_plane] - end + attr :blocks - define_method(:run) do |endpoint| - calls << [:run, endpoint] + def async(&block) + @blocks << block + :"task-#{@blocks.size}" end end - Async::GRPC::XDS.const_set(:Server, fake_server) + unbound = parent.new + monitor.run(parent: unbound) - monitor = subject.new(bind: "http://127.0.0.1:18000") + bound = parent.new + expect(subject.new(bind: "http://127.0.0.1:18000").run(parent: bound)).to be == :"task-1" - expect(monitor.run(parent: parent)).to be == :monitor_task - expect(calls.first).to be == [:initialize, monitor.control_plane] - expect(calls.last.last).to be_a(Async::HTTP::Endpoint) - ensure - Async::GRPC::XDS.send(:remove_const, :Server) - Async::GRPC::XDS.const_set(:Server, original_server) + expect(unbound.blocks.size).to be == 1 + expect(bound.blocks.size).to be == 2 end end