From fd9867c9155550982fff6eb77ed75a98c84fda49 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Thu, 6 Aug 2026 20:02:14 +1200 Subject: [PATCH 01/12] Add dedicated cluster and endpoint discovery services --- lib/async/grpc/xds.rb | 4 + lib/async/grpc/xds/cluster.rb | 10 +- .../grpc/xds/cluster_discovery_service.rb | 56 +++++ lib/async/grpc/xds/config_source.rb | 46 ++++ lib/async/grpc/xds/control_plane.rb | 6 +- lib/async/grpc/xds/discovery_service.rb | 155 ++++++++++++ .../grpc/xds/endpoint_discovery_service.rb | 56 +++++ lib/async/grpc/xds/server.rb | 10 +- lib/async/grpc/xds/service.rb | 108 +------- readme.md | 16 +- releases.md | 5 + test/async/grpc/xds/cluster.rb | 11 + test/async/grpc/xds/config_source.rb | 24 ++ test/async/grpc/xds/discovery_service.rb | 238 ++++++++++++++++++ 14 files changed, 630 insertions(+), 115 deletions(-) create mode 100644 lib/async/grpc/xds/cluster_discovery_service.rb create mode 100644 lib/async/grpc/xds/config_source.rb create mode 100644 lib/async/grpc/xds/discovery_service.rb create mode 100644 lib/async/grpc/xds/endpoint_discovery_service.rb create mode 100644 test/async/grpc/xds/config_source.rb create mode 100644 test/async/grpc/xds/discovery_service.rb diff --git a/lib/async/grpc/xds.rb b/lib/async/grpc/xds.rb index df3b6dc..2468f31 100644 --- a/lib/async/grpc/xds.rb +++ b/lib/async/grpc/xds.rb @@ -13,12 +13,16 @@ require_relative "xds/load_balancer" require_relative "xds/context" require_relative "xds/client" +require_relative "xds/config_source" require_relative "xds/cluster" require_relative "xds/endpoint" require_relative "xds/http_health_check" require_relative "xds/client_side_weighted_round_robin" require_relative "xds/control_plane" +require_relative "xds/discovery_service" require_relative "xds/service" +require_relative "xds/cluster_discovery_service" +require_relative "xds/endpoint_discovery_service" require_relative "xds/server" module Async diff --git a/lib/async/grpc/xds/cluster.rb b/lib/async/grpc/xds/cluster.rb index 31b5ae2..37ee62c 100644 --- a/lib/async/grpc/xds/cluster.rb +++ b/lib/async/grpc/xds/cluster.rb @@ -6,9 +6,10 @@ require "google/protobuf/duration_pb" require "envoy/config/cluster/v3/cluster_pb" -require "envoy/config/core/v3/config_source_pb" require "envoy/config/core/v3/protocol_pb" +require_relative "config_source" + module Async module GRPC module XDS @@ -21,21 +22,20 @@ module Cluster # Build an EDS cluster resource. # @parameter name [String] The cluster name. # @parameter service_name [String] The EDS service name. + # @parameter eds_config [Envoy::Config::Core::V3::ConfigSource] The source used to discover endpoint assignments. # @parameter load_balancing_policy [Envoy::Config::Cluster::V3::LoadBalancingPolicy | Nil] The typed Envoy load-balancing policy. # @parameter health_checks [Array(Envoy::Config::Core::V3::HealthCheck)] The active health checks applied to cluster endpoints. # @parameter connect_timeout [Numeric] The upstream connection timeout in seconds. # @parameter protocol [Symbol] The canonical upstream protocol, either `:http1` or `:http2`. # @returns [Envoy::Config::Cluster::V3::Cluster] The generated cluster resource. # @raises [ArgumentError] If the upstream protocol is unsupported. - def build(name, service_name: name, load_balancing_policy: nil, health_checks: [], connect_timeout: 5, protocol: :http2) + def build(name, service_name: name, eds_config: ConfigSource.ads, load_balancing_policy: nil, health_checks: [], connect_timeout: 5, protocol: :http2) options = { name: name.to_s, type: Envoy::Config::Cluster::V3::Cluster::DiscoveryType::EDS, eds_cluster_config: Envoy::Config::Cluster::V3::Cluster::EdsClusterConfig.new( service_name: service_name.to_s, - eds_config: Envoy::Config::Core::V3::ConfigSource.new( - ads: Envoy::Config::Core::V3::AggregatedConfigSource.new - ) + eds_config: eds_config ), connect_timeout: duration(connect_timeout), health_checks: health_checks, diff --git a/lib/async/grpc/xds/cluster_discovery_service.rb b/lib/async/grpc/xds/cluster_discovery_service.rb new file mode 100644 index 0000000..a1aaa9f --- /dev/null +++ b/lib/async/grpc/xds/cluster_discovery_service.rb @@ -0,0 +1,56 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "envoy/service/discovery/v3/discovery_pb" +require "protocol/grpc/interface" + +require_relative "control_plane" +require_relative "discovery_service" + +module Async + module GRPC + module XDS + # Serves Cluster Discovery Service requests from a {ControlPlane}. + class ClusterDiscoveryService < DiscoveryService + SERVICE_NAME = "envoy.service.cluster.v3.ClusterDiscoveryService" + RESOURCE_TYPE = ControlPlane::CLUSTER_TYPE + + # The gRPC interface for cluster discovery. + class Interface < Protocol::GRPC::Interface + rpc :StreamClusters, + request_class: Envoy::Service::Discovery::V3::DiscoveryRequest, + response_class: Envoy::Service::Discovery::V3::DiscoveryResponse, + streaming: :bidirectional + + rpc :DeltaClusters, + request_class: Envoy::Service::Discovery::V3::DeltaDiscoveryRequest, + response_class: Envoy::Service::Discovery::V3::DeltaDiscoveryResponse, + streaming: :bidirectional + end + + # Initialize a Cluster Discovery Service. + # @parameter control_plane [ControlPlane] The control plane that provides clusters. + def initialize(control_plane) + super(Interface, SERVICE_NAME, control_plane, resource_type: RESOURCE_TYPE) + end + + # Serve a state-of-the-world cluster discovery stream. + # @parameter input [Enumerable] The stream of discovery requests. + # @parameter output [Interface(:write)] The discovery response stream. + # @parameter call [Object] The gRPC call context. + # @asynchronous + def stream_clusters(input, output, call) + stream_resources(input, output) + end + + # Reject a delta cluster discovery stream, which is not supported. + # @raises [Protocol::GRPC::Error] Always raised because delta xDS is not implemented. + def delta_clusters(input, output, call) + delta_resources + end + end + end + end +end diff --git a/lib/async/grpc/xds/config_source.rb b/lib/async/grpc/xds/config_source.rb new file mode 100644 index 0000000..89275eb --- /dev/null +++ b/lib/async/grpc/xds/config_source.rb @@ -0,0 +1,46 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "envoy/config/core/v3/config_source_pb" +require "envoy/config/core/v3/grpc_service_pb" + +module Async + module GRPC + module XDS + # Builds Envoy xDS configuration sources. + module ConfigSource + extend self + + # Build a configuration source that uses the global ADS server. + # @returns [Envoy::Config::Core::V3::ConfigSource] The ADS configuration source. + def ads + Envoy::Config::Core::V3::ConfigSource.new( + ads: Envoy::Config::Core::V3::AggregatedConfigSource.new + ) + end + + # Build a configuration source for a dedicated gRPC discovery service. + # @parameter cluster_name [String] The static Envoy cluster used to reach the management server. + # @returns [Envoy::Config::Core::V3::ConfigSource] The gRPC API configuration source. + def grpc(cluster_name) + Envoy::Config::Core::V3::ConfigSource.new( + resource_api_version: :V3, + api_config_source: Envoy::Config::Core::V3::ApiConfigSource.new( + api_type: :GRPC, + transport_api_version: :V3, + grpc_services: [ + Envoy::Config::Core::V3::GrpcService.new( + envoy_grpc: Envoy::Config::Core::V3::GrpcService::EnvoyGrpc.new( + cluster_name: cluster_name.to_s + ) + ) + ] + ) + ) + end + end + end + end +end diff --git a/lib/async/grpc/xds/control_plane.rb b/lib/async/grpc/xds/control_plane.rb index ca2ea35..4521b57 100644 --- a/lib/async/grpc/xds/control_plane.rb +++ b/lib/async/grpc/xds/control_plane.rb @@ -19,7 +19,7 @@ module Async module GRPC module XDS - # Maintains xDS resource snapshots and notifies ADS streams when resources change. + # Maintains xDS resource snapshots and notifies discovery streams when resources change. class ControlPlane CLUSTER_TYPE = Cluster::TYPE_URL ENDPOINT_TYPE = Endpoint::TYPE_URL @@ -152,7 +152,7 @@ def response(type_url, names = nil) end # Register a stream to receive resource-change notifications. - # @parameter stream [Service::Stream] The stream to register. + # @parameter stream [DiscoveryService::Stream] The stream to register. def register_stream(stream) @mutex.synchronize do @streams.add(stream) @@ -160,7 +160,7 @@ def register_stream(stream) end # Remove a registered stream. - # @parameter stream [Service::Stream] The stream to remove. + # @parameter stream [DiscoveryService::Stream] The stream to remove. def remove_stream(stream) @mutex.synchronize do @streams.delete(stream) diff --git a/lib/async/grpc/xds/discovery_service.rb b/lib/async/grpc/xds/discovery_service.rb new file mode 100644 index 0000000..1310889 --- /dev/null +++ b/lib/async/grpc/xds/discovery_service.rb @@ -0,0 +1,155 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "async" +require "async/grpc/service" +require "async/queue" +require "protocol/grpc/error" +require "protocol/grpc/status" +require "set" + +require_relative "control_plane" + +module Async + module GRPC + module XDS + # Shared implementation for state-of-the-world xDS discovery services. + class DiscoveryService < Async::GRPC::Service + # Initialize a discovery service. + # @parameter interface [Class] The gRPC service interface. + # @parameter service_name [String] The fully qualified gRPC service name. + # @parameter control_plane [ControlPlane] The control plane that provides resources. + # @parameter resource_type [String | Nil] The fixed resource type, or `nil` for aggregated discovery. + def initialize(interface, service_name, control_plane, resource_type: nil) + super(interface, service_name) + + @control_plane = control_plane + @resource_type = resource_type + end + + # Serve a state-of-the-world discovery stream. + # @parameter input [Enumerable] The stream of discovery requests. + # @parameter output [Interface(:write)] The discovery response stream. + # @asynchronous + def stream_resources(input, output) + stream = Stream.new(@control_plane, output, resource_type: @resource_type) + @control_plane.register_stream(stream) + + reader = Async::Task.current.async do + input.each do |request| + stream.request(request) + end + end + + writer = Async::Task.current.async do + stream.run + end + + reader.wait + ensure + stream&.close + reader&.stop + writer&.stop + @control_plane.remove_stream(stream) if stream + end + + # Reject a delta discovery stream, which is not supported. + # @raises [Protocol::GRPC::Error] Always raised because delta xDS is not implemented. + def delta_resources + raise Protocol::GRPC::Error.new( + Protocol::GRPC::Status::UNIMPLEMENTED, + "Delta xDS is not implemented." + ) + end + + # Represents one discovery stream and its subscribed resources. + class Stream + # Initialize a discovery stream. + # @parameter control_plane [ControlPlane] The control plane that provides resources. + # @parameter output [Interface(:write)] The discovery response stream. + # @parameter resource_type [String | Nil] The fixed resource type, or `nil` for aggregated discovery. + def initialize(control_plane, output, resource_type: nil) + @control_plane = control_plane + @output = output + @resource_type = resource_type + @subscriptions = Hash.new{|hash, type_url| hash[type_url] = Set.new} + @versions = {} + @queue = Async::Queue.new + @closed = false + end + + # Process a discovery request and update the stream's subscriptions. + # @parameter request [Envoy::Service::Discovery::V3::DiscoveryRequest] The discovery request. + def request(request) + type_url = request.type_url + + if @resource_type + if type_url.nil? || type_url.empty? + type_url = @resource_type + elsif type_url != @resource_type + raise Protocol::GRPC::Error.new( + Protocol::GRPC::Status::INVALID_ARGUMENT, + "Expected resource type #{@resource_type.inspect}, but received #{type_url.inspect}." + ) + end + elsif type_url.nil? || type_url.empty? + return + end + + if request.error_detail + Console.warn(self, "Received xDS NACK.", type_url: type_url, error_detail: request.error_detail) + return + end + + if request.resource_names.any? + @subscriptions[type_url].merge(request.resource_names) + else + @subscriptions[type_url] + end + + @queue << type_url + end + + # Schedule a resource type for delivery after it changes. + # @parameter type_url [String] The changed xDS resource type URL. + def changed(type_url) + return if @resource_type && type_url != @resource_type + + @queue << type_url unless @closed + end + + # Deliver scheduled resource updates until the stream closes. + # @asynchronous + def run + until @closed + type_url = @queue.dequeue + flush(type_url) + end + end + + # Deliver the latest resource version for a subscribed type. + # @parameter type_url [String] The xDS resource type URL. + def flush(type_url) + names = @subscriptions[type_url] + return unless names + + version = @control_plane.version(type_url) + return if @versions[type_url] == version + + response = @control_plane.response(type_url, names) + @output.write(response) + @versions[type_url] = version + end + + # Close the stream and stop waiting for changes. + def close + @closed = true + @queue.close + end + end + end + end + end +end diff --git a/lib/async/grpc/xds/endpoint_discovery_service.rb b/lib/async/grpc/xds/endpoint_discovery_service.rb new file mode 100644 index 0000000..90869f4 --- /dev/null +++ b/lib/async/grpc/xds/endpoint_discovery_service.rb @@ -0,0 +1,56 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "envoy/service/discovery/v3/discovery_pb" +require "protocol/grpc/interface" + +require_relative "control_plane" +require_relative "discovery_service" + +module Async + module GRPC + module XDS + # Serves Endpoint Discovery Service requests from a {ControlPlane}. + class EndpointDiscoveryService < DiscoveryService + SERVICE_NAME = "envoy.service.endpoint.v3.EndpointDiscoveryService" + RESOURCE_TYPE = ControlPlane::ENDPOINT_TYPE + + # The gRPC interface for endpoint discovery. + class Interface < Protocol::GRPC::Interface + rpc :StreamEndpoints, + request_class: Envoy::Service::Discovery::V3::DiscoveryRequest, + response_class: Envoy::Service::Discovery::V3::DiscoveryResponse, + streaming: :bidirectional + + rpc :DeltaEndpoints, + request_class: Envoy::Service::Discovery::V3::DeltaDiscoveryRequest, + response_class: Envoy::Service::Discovery::V3::DeltaDiscoveryResponse, + streaming: :bidirectional + end + + # Initialize an Endpoint Discovery Service. + # @parameter control_plane [ControlPlane] The control plane that provides endpoint assignments. + def initialize(control_plane) + super(Interface, SERVICE_NAME, control_plane, resource_type: RESOURCE_TYPE) + end + + # Serve a state-of-the-world endpoint discovery stream. + # @parameter input [Enumerable] The stream of discovery requests. + # @parameter output [Interface(:write)] The discovery response stream. + # @parameter call [Object] The gRPC call context. + # @asynchronous + def stream_endpoints(input, output, call) + stream_resources(input, output) + end + + # Reject a delta endpoint discovery stream, which is not supported. + # @raises [Protocol::GRPC::Error] Always raised because delta xDS is not implemented. + def delta_endpoints(input, output, call) + delta_resources + end + end + end + end +end diff --git a/lib/async/grpc/xds/server.rb b/lib/async/grpc/xds/server.rb index 3f250bd..6a1b37e 100644 --- a/lib/async/grpc/xds/server.rb +++ b/lib/async/grpc/xds/server.rb @@ -16,16 +16,22 @@ module XDS class Server # Initialize an xDS server. # @parameter control_plane [ControlPlane] The control plane to serve. + # @parameter services [Array(Class)] The discovery service classes to register. # @parameter options [Hash] Default options forwarded to `Async::HTTP::Server`. - def initialize(control_plane = ControlPlane.new, **options) + def initialize(control_plane = ControlPlane.new, services: [Service], **options) @control_plane = control_plane @dispatcher = Async::GRPC::Dispatcher.new - @dispatcher.register(Service.new(@control_plane)) + @services = services.map do |service| + service.new(@control_plane).tap do |instance| + @dispatcher.register(instance) + end + end @options = options end attr :control_plane attr :dispatcher + attr :services # Run the xDS server on an endpoint. # @parameter endpoint [Async::HTTP::Endpoint] The endpoint to bind. diff --git a/lib/async/grpc/xds/service.rb b/lib/async/grpc/xds/service.rb index 56931dc..97c95e4 100644 --- a/lib/async/grpc/xds/service.rb +++ b/lib/async/grpc/xds/service.rb @@ -3,29 +3,21 @@ # Released under the MIT License. # Copyright, 2026, by Samuel Williams. -require "async" -require "async/queue" -require "async/grpc/service" -require "protocol/grpc/error" -require "protocol/grpc/status" -require "set" - require "envoy/service/discovery/v3/aggregated_discovery_service" -require_relative "control_plane" +require_relative "discovery_service" module Async module GRPC module XDS # Serves Envoy Aggregated Discovery Service requests from a {ControlPlane}. - class Service < Async::GRPC::Service + class Service < DiscoveryService SERVICE_NAME = "envoy.service.discovery.v3.AggregatedDiscoveryService" # Initialize an Aggregated Discovery Service. # @parameter control_plane [ControlPlane] The control plane that provides resources. def initialize(control_plane) - super(Envoy::Service::Discovery::V3::AggregatedDiscoveryService, SERVICE_NAME) - @control_plane = control_plane + super(Envoy::Service::Discovery::V3::AggregatedDiscoveryService, SERVICE_NAME, control_plane) end # Serve a state-of-the-world Aggregated Discovery Service stream. @@ -34,25 +26,7 @@ def initialize(control_plane) # @parameter call [Object] The gRPC call context. # @asynchronous def stream_aggregated_resources(input, output, call) - stream = Stream.new(@control_plane, output) - @control_plane.register_stream(stream) - - reader = Async do - input.each do |request| - stream.request(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 + stream_resources(input, output) end # Reject a delta Aggregated Discovery Service stream, which is not supported. @@ -61,79 +35,7 @@ def stream_aggregated_resources(input, output, call) # @parameter call [Object] The gRPC call context. # @raises [Protocol::GRPC::Error] Always raised because delta xDS is not implemented. def delta_aggregated_resources(input, output, call) - raise Protocol::GRPC::Error.new( - Protocol::GRPC::Status::UNIMPLEMENTED, - "Delta xDS is not implemented." - ) - end - - # Represents one ADS stream and its subscribed resources. - class Stream - # Initialize an ADS stream. - # @parameter control_plane [ControlPlane] The control plane that provides resources. - # @parameter output [Interface(:write)] The discovery response stream. - def initialize(control_plane, output) - @control_plane = control_plane - @output = output - @subscriptions = Hash.new{|hash, type_url| hash[type_url] = Set.new} - @versions = {} - @queue = Async::Queue.new - @closed = false - end - - # Process a discovery request and update the stream's subscriptions. - # @parameter request [Envoy::Service::Discovery::V3::DiscoveryRequest] The discovery request. - def request(request) - return if request.type_url.nil? || request.type_url.empty? - - if request.error_detail - Console.warn(self, "Received xDS NACK.", type_url: request.type_url, error_detail: request.error_detail) - return - end - - if request.resource_names.any? - @subscriptions[request.type_url].merge(request.resource_names) - else - @subscriptions[request.type_url] - end - - @queue << request.type_url - end - - # Schedule a resource type for delivery after it changes. - # @parameter type_url [String] The changed xDS resource type URL. - def changed(type_url) - @queue << type_url unless @closed - end - - # Deliver scheduled resource updates until the stream closes. - # @asynchronous - def run - until @closed - type_url = @queue.dequeue - flush(type_url) - end - end - - # Deliver the latest resource version for a subscribed type. - # @parameter type_url [String] The xDS resource type URL. - def flush(type_url) - names = @subscriptions[type_url] - return unless names - - version = @control_plane.version(type_url) - return if @versions[type_url] == version - - response = @control_plane.response(type_url, names) - @output.write(response) - @versions[type_url] = version - end - - # Close the stream and stop waiting for changes. - def close - @closed = true - @queue.close - end + delta_resources end end end diff --git a/readme.md b/readme.md index 66e0ada..e1db12d 100644 --- a/readme.md +++ b/readme.md @@ -2,7 +2,7 @@ xDS support for `async-grpc` clients. -This gem contains the experimental xDS implementation extracted from `async-grpc`, including Envoy xDS protobuf definitions, ADS discovery, CDS/EDS resource handling, and client-side load balancing. +This gem contains the experimental xDS implementation extracted from `async-grpc`, including Envoy xDS protobuf definitions, aggregated and resource-specific discovery, CDS/EDS resource handling, and client-side load balancing. [![Development Status](https://github.com/socketry/async-grpc-xds/workflows/Test/badge.svg)](https://github.com/socketry/async-grpc-xds/actions?workflow=Test) @@ -10,9 +10,21 @@ This gem contains the experimental xDS implementation extracted from `async-grpc Please see the [project documentation](https://socketry.github.io/async-grpc-xds/) for more details. +By default, {ruby Async::GRPC::XDS::Server} serves the Aggregated Discovery Service. To expose dedicated CDS and EDS streams while leaving ADS available to another control plane, select the resource-specific services: + +``` ruby +server = Async::GRPC::XDS::Server.new( + control_plane, + services: [ + Async::GRPC::XDS::ClusterDiscoveryService, + Async::GRPC::XDS::EndpointDiscoveryService, + ] +) +``` + ## Status -This is an early implementation focused on ADS with CDS and EDS. LDS/RDS, full routing semantics, NACK handling, locality weighting, and delta xDS are not complete yet. +This is an early implementation focused on CDS and EDS over ADS or dedicated discovery streams. LDS/RDS, full routing semantics, NACK handling, locality weighting, and delta xDS are not complete yet. ## Testing diff --git a/releases.md b/releases.md index 8d6cacc..870063b 100644 --- a/releases.md +++ b/releases.md @@ -1,5 +1,10 @@ # Releases +## Unreleased + + - Add dedicated Cluster Discovery Service and Endpoint Discovery Service implementations, allowing resource-specific xDS streams without claiming ADS. + - Allow generated clusters to use a dedicated EDS configuration source instead of ADS. + ## v0.3.0 - Add Envoy HTTP active health-check resources and attach health checks to generated clusters. diff --git a/test/async/grpc/xds/cluster.rb b/test/async/grpc/xds/cluster.rb index 9ca0e52..636a590 100644 --- a/test/async/grpc/xds/cluster.rb +++ b/test/async/grpc/xds/cluster.rb @@ -4,6 +4,7 @@ # Copyright, 2026, by Samuel Williams. require "async/grpc/xds/cluster" +require "async/grpc/xds/config_source" require "async/grpc/xds/http_health_check" describe Async::GRPC::XDS::Cluster do @@ -26,6 +27,16 @@ expect(cluster.http2_protocol_options).to be == nil end + it "uses a dedicated EDS configuration source" do + eds_config = Async::GRPC::XDS::ConfigSource.grpc("xds_cluster") + cluster = subject.build("myservice", eds_config: eds_config) + api_config_source = cluster.eds_cluster_config.eds_config.api_config_source + + expect(api_config_source.api_type).to be == :GRPC + expect(api_config_source.transport_api_version).to be == :V3 + expect(api_config_source.grpc_services.first.envoy_grpc.cluster_name).to be == "xds_cluster" + end + it "attaches active health checks" do health_check = Async::GRPC::XDS::HTTPHealthCheck.build("/services/ping") cluster = subject.build("myservice", health_checks: [health_check]) diff --git a/test/async/grpc/xds/config_source.rb b/test/async/grpc/xds/config_source.rb new file mode 100644 index 0000000..c5a42a3 --- /dev/null +++ b/test/async/grpc/xds/config_source.rb @@ -0,0 +1,24 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "async/grpc/xds/config_source" + +describe Async::GRPC::XDS::ConfigSource do + it "builds an ADS configuration source" do + config_source = subject.ads + + expect(config_source.ads).not.to be_nil + end + + it "builds a dedicated gRPC configuration source" do + config_source = subject.grpc(:xds_cluster) + api_config_source = config_source.api_config_source + + expect(config_source.resource_api_version).to be == :V3 + expect(api_config_source.api_type).to be == :GRPC + expect(api_config_source.transport_api_version).to be == :V3 + expect(api_config_source.grpc_services.first.envoy_grpc.cluster_name).to be == "xds_cluster" + end +end diff --git a/test/async/grpc/xds/discovery_service.rb b/test/async/grpc/xds/discovery_service.rb new file mode 100644 index 0000000..307abbd --- /dev/null +++ b/test/async/grpc/xds/discovery_service.rb @@ -0,0 +1,238 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "async/grpc/xds/cluster_discovery_service" +require "async/grpc/xds/endpoint_discovery_service" +require "async/grpc/xds/server" +require "async/grpc/xds/service" +require "async/notification" +require "envoy/config/cluster/v3/cluster_pb" +require "envoy/config/endpoint/v3/endpoint_pb" +require "google/rpc/status_pb" +require "sus/fixtures/async" + +describe Async::GRPC::XDS::DiscoveryService do + include Sus::Fixtures::Async::ReactorContext + + let(:control_plane) {Async::GRPC::XDS::ControlPlane.new} + + def output + [].tap do |responses| + responses.define_singleton_method(:write){|response| self << response} + end + end + + def request(type_url: nil, resource_names: ["myservice"]) + Envoy::Service::Discovery::V3::DiscoveryRequest.new( + type_url: type_url, + resource_names: resource_names + ) + end + + def stream_for(resource_type, output) + subject::Stream.new(control_plane, output, resource_type: resource_type) + end + + it "serves clusters with an explicit resource type" do + control_plane.update_cluster("myservice") + responses = output + stream = stream_for(Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE, responses) + + stream.request(request(type_url: Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE)) + stream.flush(Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE) + + cluster = Envoy::Config::Cluster::V3::Cluster.decode(responses.first.resources.first.value) + expect(cluster.name).to be == "myservice" + ensure + stream&.close + end + + it "uses the implied endpoint resource type when omitted" do + control_plane.update_endpoints("myservice", [ + {addresses: [{address: "127.0.0.1", port: 50051}], healthy: true} + ]) + responses = output + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, responses) + + stream.request(request) + stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + + assignment = Envoy::Config::Endpoint::V3::ClusterLoadAssignment.decode(responses.first.resources.first.value) + expect(assignment.cluster_name).to be == "myservice" + ensure + stream&.close + end + + it "requires a resource type for aggregated streams" do + responses = output + stream = subject::Stream.new(control_plane, responses) + + stream.request(request) + + expect(responses).to be(:empty?) + ensure + stream&.close + end + + it "subscribes to every resource when no names are specified" do + control_plane.update_endpoints("myservice", []) + responses = output + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, responses) + + stream.request(request(resource_names: [])) + stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + + expect(responses.first.resources.size).to be == 1 + ensure + stream&.close + end + + it "logs and ignores a NACK" do + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, output) + nack = request + nack.error_detail = Google::Rpc::Status.new(message: "invalid resource") + + stream.request(nack) + + expect(stream).not.to be_nil + ensure + stream&.close + end + + it "rejects a resource type that does not belong to the service" do + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, output) + + expect do + stream.request(request(type_url: Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE)) + end.to raise_exception(Protocol::GRPC::Error) + ensure + stream&.close + end + + it "ignores changes for resource types that do not belong to the service" do + responses = output + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, responses) + stream.request(request) + stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + + stream.changed(Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE) + stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + + expect(responses.size).to be == 1 + ensure + stream&.close + end + + it "accepts changes for its resource type" do + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, output) + + stream.changed(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + + expect(stream).not.to be_nil + ensure + stream&.close + end + + it "serves endpoint requests through the dedicated discovery stream" do + control_plane.update_endpoints("myservice", [ + {addresses: [{address: "127.0.0.1", port: 50051}], healthy: true} + ]) + service = Async::GRPC::XDS::EndpointDiscoveryService.new(control_plane) + request = self.request + responses = [] + response_written = Async::Notification.new + input = Object.new + input.define_singleton_method(:each) do |&block| + block.call(request) + response_written.wait + end + responses.define_singleton_method(:write) do |response| + self << response + response_written.signal + end + + service.stream_endpoints(input, responses, nil) + + expect(responses.first.type_url).to be == Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE + end + + it "delegates cluster requests to the discovery stream" do + service = Async::GRPC::XDS::ClusterDiscoveryService.new(control_plane) + input = Object.new + output = Object.new + arguments = nil + service.define_singleton_method(:stream_resources) do |*given| + arguments = given + end + + service.stream_clusters(input, output, nil) + + expect(arguments).to be == [input, output] + end + + it "delegates aggregated requests to the discovery stream" do + service = Async::GRPC::XDS::Service.new(control_plane) + input = Object.new + output = Object.new + arguments = nil + service.define_singleton_method(:stream_resources) do |*given| + arguments = given + end + + service.stream_aggregated_resources(input, output, nil) + + expect(arguments).to be == [input, output] + end + + it "configures a server with dedicated cluster and endpoint services" do + server = Async::GRPC::XDS::Server.new( + control_plane, + services: [ + Async::GRPC::XDS::ClusterDiscoveryService, + Async::GRPC::XDS::EndpointDiscoveryService, + ] + ) + + expect(server.services.map(&:service_name)).to be == [ + "envoy.service.cluster.v3.ClusterDiscoveryService", + "envoy.service.endpoint.v3.EndpointDiscoveryService", + ] + end + + it "runs the configured HTTP server" do + http_server = Object.new + http_server.define_singleton_method(:run){:result} + server = Async::GRPC::XDS::Server.new(control_plane, timeout: 1) + + mock(Async::HTTP::Server) do |mock| + mock.replace(:new) do |dispatcher, endpoint, **options| + expect(dispatcher).to be == server.dispatcher + expect(endpoint).to be == :endpoint + expect(options).to be == {timeout: 1, reuse_port: true} + http_server + end + end + + expect(server.run(:endpoint, reuse_port: true)).to be == :result + end + + it "rejects delta discovery" do + service = Async::GRPC::XDS::EndpointDiscoveryService.new(control_plane) + + expect do + service.delta_endpoints(nil, nil, nil) + end.to raise_exception(Protocol::GRPC::Error) + + cluster_service = Async::GRPC::XDS::ClusterDiscoveryService.new(control_plane) + expect do + cluster_service.delta_clusters(nil, nil, nil) + end.to raise_exception(Protocol::GRPC::Error) + + aggregated_service = Async::GRPC::XDS::Service.new(control_plane) + expect do + aggregated_service.delta_aggregated_resources(nil, nil, nil) + end.to raise_exception(Protocol::GRPC::Error) + end +end From 3c92457ee484722096da59346da741e7a270084e Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Thu, 6 Aug 2026 21:21:30 +1200 Subject: [PATCH 02/12] Avoid asynchronous discovery stream test fixtures --- test/async/grpc/xds/discovery_service.rb | 69 ++++++++++++++++++------ 1 file changed, 53 insertions(+), 16 deletions(-) diff --git a/test/async/grpc/xds/discovery_service.rb b/test/async/grpc/xds/discovery_service.rb index 307abbd..34840ba 100644 --- a/test/async/grpc/xds/discovery_service.rb +++ b/test/async/grpc/xds/discovery_service.rb @@ -7,15 +7,11 @@ require "async/grpc/xds/endpoint_discovery_service" require "async/grpc/xds/server" require "async/grpc/xds/service" -require "async/notification" require "envoy/config/cluster/v3/cluster_pb" require "envoy/config/endpoint/v3/endpoint_pb" require "google/rpc/status_pb" -require "sus/fixtures/async" describe Async::GRPC::XDS::DiscoveryService do - include Sus::Fixtures::Async::ReactorContext - let(:control_plane) {Async::GRPC::XDS::ControlPlane.new} def output @@ -135,27 +131,68 @@ def stream_for(resource_type, output) stream&.close end - it "serves endpoint requests through the dedicated discovery stream" do + it "delivers queued resource changes" do control_plane.update_endpoints("myservice", [ {addresses: [{address: "127.0.0.1", port: 50051}], healthy: true} ]) - service = Async::GRPC::XDS::EndpointDiscoveryService.new(control_plane) - request = self.request - responses = [] - response_written = Async::Notification.new - input = Object.new - input.define_singleton_method(:each) do |&block| - block.call(request) - response_written.wait - end + responses = output + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, responses) + stream.request(request) + responses.define_singleton_method(:write) do |response| self << response - response_written.signal + stream.close end - service.stream_endpoints(input, responses, nil) + stream.run expect(responses.first.type_url).to be == Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE + ensure + stream&.close + end + + it "coordinates endpoint discovery stream tasks" do + service = Async::GRPC::XDS::EndpointDiscoveryService.new(control_plane) + input = [request] + output = Object.new + events = [] + stream = Object.new + stream.define_singleton_method(:request){|request| events << [:request, request]} + stream.define_singleton_method(:run){events << :run} + stream.define_singleton_method(:close){events << :close} + + task = Object.new + task.define_singleton_method(:async) do |&block| + block.call + + Object.new.tap do |handle| + handle.define_singleton_method(:wait){events << :wait} + handle.define_singleton_method(:stop){events << :stop} + end + end + + mock(subject::Stream) do |stream_mock| + stream_mock.replace(:new) do |given_control_plane, given_output, resource_type:| + expect(given_control_plane).to be == control_plane + expect(given_output).to be == output + expect(resource_type).to be == Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE + stream + end + + mock(Async::Task) do |task_mock| + task_mock.replace(:current){task} + service.stream_endpoints(input, output, nil) + end + end + + expect(events).to be == [ + [:request, input.first], + :run, + :wait, + :close, + :stop, + :stop, + ] end it "delegates cluster requests to the discovery stream" do From 9b6e6da4497f09b9c2483600f14b73944c9f56ba Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Thu, 6 Aug 2026 21:55:33 +1200 Subject: [PATCH 03/12] Add getting started guide --- async-grpc-xds.gemspec | 2 +- context/getting-started.md | 363 +++++++++++++++++++++++++++++++ context/index.yaml | 12 + guides/getting-started/readme.md | 363 +++++++++++++++++++++++++++++++ guides/links.yaml | 2 + readme.md | 12 +- releases.md | 1 + 7 files changed, 743 insertions(+), 12 deletions(-) create mode 100644 context/getting-started.md create mode 100644 context/index.yaml create mode 100644 guides/getting-started/readme.md create mode 100644 guides/links.yaml diff --git a/async-grpc-xds.gemspec b/async-grpc-xds.gemspec index c8552d0..b79324f 100644 --- a/async-grpc-xds.gemspec +++ b/async-grpc-xds.gemspec @@ -20,7 +20,7 @@ Gem::Specification.new do |spec| "source_code_uri" => "https://github.com/socketry/async-grpc-xds.git", } - spec.files = Dir.glob(["{fixtures,lib,proto,xds}/**/*", "*.md"], File::FNM_DOTMATCH, base: __dir__) + spec.files = Dir.glob(["{context,fixtures,lib,proto,xds}/**/*", "*.md"], File::FNM_DOTMATCH, base: __dir__) spec.required_ruby_version = ">= 3.3" diff --git a/context/getting-started.md b/context/getting-started.md new file mode 100644 index 0000000..0170193 --- /dev/null +++ b/context/getting-started.md @@ -0,0 +1,363 @@ +# Getting Started + +This guide explains how to use `async-grpc-xds` to publish CDS and EDS resources to Envoy, or to discover gRPC backends from Ruby using xDS. + +## Installation + +Add the gem to your project: + +~~~ bash +$ bundle add async-grpc-xds +~~~ + +The gem provides both sides of an xDS integration: + + - A control-plane server which publishes in-memory resources to Envoy over ADS or dedicated discovery services. + - An experimental Ruby gRPC client which discovers clusters and endpoints from an ADS server. + +## Core Concepts + +xDS separates logical proxy configuration from the concrete destinations which currently provide it: + +| API | Resource | Purpose | +| --- | --- | --- | +| CDS | `Cluster` | Describes an upstream service, protocol, health checks, load-balancing policy, and how to discover its endpoints. | +| EDS | `ClusterLoadAssignment` | Supplies the concrete IP or Unix-socket endpoints for a cluster. | +| RDS | `RouteConfiguration` | Maps incoming requests to clusters. | +| LDS | `Listener` | Describes the addresses and network filters where Envoy accepts traffic. | + +`async-grpc-xds` currently builds and serves CDS and EDS resources. Routes and listeners normally remain in Envoy's bootstrap configuration or come from another control plane. + +Three names connect the configuration: + + - The **management cluster** is a static Envoy cluster, such as `xds_cluster`, which reaches the Ruby xDS server. + - The **application cluster** is the logical upstream service, such as `application`. + - The EDS `service_name` and `ClusterLoadAssignment#cluster_name` identify the endpoint assignment used by that application cluster. They default to the application cluster name. + +The following diagram shows the dedicated CDS and EDS arrangement used throughout this guide: + +``` mermaid +flowchart LR + Bootstrap[Envoy bootstrap] --> Management[xds_cluster] + Management -->|CDS stream| ControlPlane[Ruby control plane] + Management -->|EDS stream| ControlPlane + ControlPlane -->|Cluster| Application[application cluster] + ControlPlane -->|ClusterLoadAssignment| Workers[worker endpoints] + Traffic[Application traffic] --> Envoy + Envoy --> Application + Application --> Workers +``` + +## Serving Dedicated CDS and EDS + +Dedicated discovery services are a good fit when this control plane owns application clusters and their local workers, but should not claim Envoy's single ADS connection. Another control plane can then use ADS for coordinated listener, route, or other configuration. + +Create a control plane, publish an application cluster and its initial workers, then serve the dedicated CDS and EDS interfaces: + +``` ruby +require "async" +require "async/grpc/xds" +require "async/http/endpoint" + +control_plane = Async::GRPC::XDS::ControlPlane.new(identifier: "application-supervisor") +eds_config = Async::GRPC::XDS::ConfigSource.grpc("xds_cluster") +health_check = Async::GRPC::XDS::HTTPHealthCheck.build( + "/health", + interval: 2, + timeout: 1 +) + +control_plane.update_cluster( + "application", + protocol: :http1, + eds_config: eds_config, + health_checks: [health_check] +) + +control_plane.update_endpoints("application", [ + { + hostname: "worker-1", + addresses: [{address: "127.0.0.1", port: 9292}], + healthy: true, + }, + { + hostname: "worker-2", + addresses: [{address: "127.0.0.1", port: 9293}], + healthy: true, + }, +]) + +server = Async::GRPC::XDS::Server.new( + control_plane, + services: [ + Async::GRPC::XDS::ClusterDiscoveryService, + Async::GRPC::XDS::EndpointDiscoveryService, + ] +) + +endpoint = Async::HTTP::Endpoint.parse( + "http://0.0.0.0:18000", + protocol: Async::HTTP::Protocol::HTTP2 +) + +Sync do + server.run(endpoint) +end +``` + +The management endpoint uses HTTP/2 because xDS is a gRPC protocol. This example uses plaintext HTTP/2 within a trusted local network; deployments can instead configure TLS on the endpoint. + +### Envoy Bootstrap + +Envoy must know how to reach the management server before it can discover anything else. Define `xds_cluster` statically and configure CDS to use it: + +``` yaml +node: + id: application-proxy-1 + cluster: application-proxies + +admin: + address: + socket_address: + address: 127.0.0.1 + port_value: 19000 + +dynamic_resources: + cds_config: + resource_api_version: V3 + api_config_source: + api_type: GRPC + transport_api_version: V3 + grpc_services: + - envoy_grpc: + cluster_name: xds_cluster + +static_resources: + listeners: + - name: ingress + address: + socket_address: + address: 0.0.0.0 + port_value: 8080 + filter_chains: + - filters: + - name: envoy.filters.network.http_connection_manager + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager + stat_prefix: ingress_http + route_config: + name: local_route + validate_clusters: false + virtual_hosts: + - name: application + domains: ["*"] + routes: + - match: + prefix: "/" + route: + cluster: application + http_filters: + - name: envoy.filters.http.router + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router + + clusters: + - name: xds_cluster + connect_timeout: 1s + type: STATIC + typed_extension_protocol_options: + envoy.extensions.upstreams.http.v3.HttpProtocolOptions: + "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions + explicit_http_config: + http2_protocol_options: {} + load_assignment: + cluster_name: xds_cluster + endpoints: + - lb_endpoints: + - endpoint: + address: + socket_address: + address: 127.0.0.1 + port_value: 18000 +``` + +The static route names `application` before CDS has delivered that cluster, so `validate_clusters: false` allows the bootstrap configuration to load. Envoy keeps the discovered cluster warming until its EDS assignment arrives. + +The `xds_cluster` name must match the name passed to {ruby Async::GRPC::XDS::ConfigSource.grpc}. The generated application cluster then opens its dedicated EDS stream through the same management cluster. + +## Publishing Endpoint Changes + +The control plane is thread-safe and immediately notifies connected discovery streams after a resource changes. Publish the complete current endpoint assignment whenever workers start, stop, or change health: + +``` ruby +control_plane.update_endpoints("application", [ + { + hostname: "worker-2", + addresses: [{address: "127.0.0.1", port: 9293}], + healthy: true, + }, +]) +``` + +This implementation uses state-of-the-world discovery: each EDS response contains the complete assignment requested by Envoy. It does not currently implement the delta discovery RPCs. + +Every call to `update_cluster` or `update_endpoints` increments the version for that resource type, even if the generated resource is unchanged. Producers should therefore avoid publishing redundant updates. + +To remove a service completely, remove both resources: + +``` ruby +control_plane.remove_endpoints("application") +control_plane.remove_cluster("application") +``` + +### IP and Unix-Socket Addresses + +An endpoint must contain one or more addresses. An IP address uses `:address` and `:port`: + +``` ruby +{ + addresses: [{address: "127.0.0.1", port: 9292}], + healthy: true, +} +``` + +A Unix domain socket uses `:path`: + +``` ruby +{ + addresses: [{path: "/run/application/worker-1.ipc"}], + healthy: true, +} +``` + +Several addresses in one `:addresses` array describe alternative addresses for one logical load-balancer endpoint. The first becomes Envoy's primary address and the remainder become `additional_addresses`; they do not represent additional workers. + +### Health + +The `:healthy` value sets the EDS `health_status` for an endpoint. It accepts healthy, unhealthy, degraded, or unknown states; booleans map to healthy and unhealthy. + +This reported state is separate from active health checks. Adding a health check to the cluster tells Envoy to probe every published endpoint itself: + +``` ruby +health_check = Async::GRPC::XDS::HTTPHealthCheck.build( + "/health", + interval: 2, + timeout: 1, + unhealthy_threshold: 2, + healthy_threshold: 1 +) + +control_plane.update_cluster( + "application", + protocol: :http1, + eds_config: Async::GRPC::XDS::ConfigSource.grpc("xds_cluster"), + health_checks: [health_check] +) +``` + +Use reported health for information already known by the resource owner, such as whether a worker remains registered. Use active health checks when Envoy should independently verify that it can send application traffic to the endpoint. + +## ADS or Dedicated Services + +Both transports carry the same `DiscoveryRequest` and `DiscoveryResponse` messages. The difference is how Envoy organizes its streams and management servers: + +| | ADS | Dedicated CDS and EDS | +| --- | --- | --- | +| Streams | One aggregated stream for several resource types. | One stream per resource type. | +| Coordination | Provides ordering across related resource types from one control plane. | Each resource type progresses independently. | +| Ownership | Envoy has one ADS management server. | Each resource type can use its own configuration source. | +| Best fit | One control plane owns coordinated proxy configuration. | A focused control plane owns only clusters or endpoints. | + +The default server exposes ADS: + +``` ruby +control_plane = Async::GRPC::XDS::ControlPlane.new +control_plane.update_cluster("application", protocol: :http1) +control_plane.update_endpoints("application", endpoints) + +server = Async::GRPC::XDS::Server.new(control_plane) +``` + +{ruby Async::GRPC::XDS::Cluster.build} uses ADS for EDS by default, so no explicit `eds_config` is needed in this mode. Configure Envoy accordingly: + +``` yaml +dynamic_resources: + ads_config: + api_type: GRPC + transport_api_version: V3 + grpc_services: + - envoy_grpc: + cluster_name: xds_cluster + cds_config: + resource_api_version: V3 + ads: {} +``` + +Do not use the default ADS EDS source when serving dedicated CDS and EDS. In that arrangement, pass `ConfigSource.grpc("xds_cluster")` while building or updating every application cluster. + +## Using the Ruby xDS Client + +{ruby Async::GRPC::XDS::Client} is an experimental gRPC client which discovers a named cluster and its healthy endpoints over ADS, then load-balances RPC calls between them. + +Supply bootstrap configuration directly: + +``` ruby +bootstrap = { + xds_servers: [ + { + server_uri: "127.0.0.1:18000", + channel_creds: [{type: "insecure"}], + } + ], + node: { + id: "orders-client-1", + cluster: "orders-clients", + } +} + +Sync do + client = Async::GRPC::XDS::Client.new("application", bootstrap: bootstrap) + + begin + stub = client.stub(Greeter::Interface, "example.Greeter") + response = stub.say_hello(Greeter::Request.new(name: "World")) + ensure + client.close + end +end +``` + +Alternatively, pass the path to a JSON bootstrap file. When `bootstrap` is omitted, the client checks `GRPC_XDS_BOOTSTRAP` and then `~/.config/grpc/bootstrap.json`. + +The Ruby client currently consumes ADS, supports CDS and EDS, filters endpoints by reported health, and provides basic client-side load balancing and retry behavior. Dedicated CDS and EDS client streams are not yet implemented. + +## Operational Checks + +Envoy's admin interface shows whether it accepted the resources and whether the application cluster has usable members: + +~~~ bash +$ curl -s http://127.0.0.1:19000/config_dump +$ curl -s http://127.0.0.1:19000/clusters +$ curl -s http://127.0.0.1:19000/stats | grep -E 'cluster\.application\.(warming|membership_healthy|update_rejected)' +~~~ + +When a cluster remains in `warming`, check: + + - The `xds_cluster` address and port reach the Ruby server using HTTP/2. + - The management cluster name matches the name in every gRPC configuration source. + - The CDS cluster name, EDS service name, and `ClusterLoadAssignment#cluster_name` agree. + - A dedicated CDS cluster uses a dedicated EDS configuration source rather than `ads: {}`. + - Envoy has not incremented `update_rejected`; rejected responses are also returned to the control plane as NACKs and logged. + +## Current Scope + +`async-grpc-xds` currently provides: + + - xDS v3 protobuf definitions. + - State-of-the-world ADS, CDS, and EDS server streams. + - In-memory cluster and endpoint resources with per-type versions. + - HTTP/1 and HTTP/2 upstream cluster configuration. + - IP, Unix-socket, health-status, active HTTP health-check, and out-of-band ORCA policy resource builders. + - An experimental ADS-based Ruby gRPC client. + +Delta discovery, complete NACK recovery, persistent resource storage, LDS/RDS serving, locality weighting, and complete routing semantics are not implemented yet. diff --git a/context/index.yaml b/context/index.yaml new file mode 100644 index 0000000..0de09cc --- /dev/null +++ b/context/index.yaml @@ -0,0 +1,12 @@ +# 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: xDS support for Async::GRPC clients. +metadata: + documentation_uri: https://socketry.github.io/async-grpc-xds/ + source_code_uri: https://github.com/socketry/async-grpc-xds.git +files: +- path: getting-started.md + title: Getting Started + description: This guide explains how to use `async-grpc-xds` to publish CDS and + EDS resources to Envoy, or to discover gRPC backends from Ruby using xDS. diff --git a/guides/getting-started/readme.md b/guides/getting-started/readme.md new file mode 100644 index 0000000..0170193 --- /dev/null +++ b/guides/getting-started/readme.md @@ -0,0 +1,363 @@ +# Getting Started + +This guide explains how to use `async-grpc-xds` to publish CDS and EDS resources to Envoy, or to discover gRPC backends from Ruby using xDS. + +## Installation + +Add the gem to your project: + +~~~ bash +$ bundle add async-grpc-xds +~~~ + +The gem provides both sides of an xDS integration: + + - A control-plane server which publishes in-memory resources to Envoy over ADS or dedicated discovery services. + - An experimental Ruby gRPC client which discovers clusters and endpoints from an ADS server. + +## Core Concepts + +xDS separates logical proxy configuration from the concrete destinations which currently provide it: + +| API | Resource | Purpose | +| --- | --- | --- | +| CDS | `Cluster` | Describes an upstream service, protocol, health checks, load-balancing policy, and how to discover its endpoints. | +| EDS | `ClusterLoadAssignment` | Supplies the concrete IP or Unix-socket endpoints for a cluster. | +| RDS | `RouteConfiguration` | Maps incoming requests to clusters. | +| LDS | `Listener` | Describes the addresses and network filters where Envoy accepts traffic. | + +`async-grpc-xds` currently builds and serves CDS and EDS resources. Routes and listeners normally remain in Envoy's bootstrap configuration or come from another control plane. + +Three names connect the configuration: + + - The **management cluster** is a static Envoy cluster, such as `xds_cluster`, which reaches the Ruby xDS server. + - The **application cluster** is the logical upstream service, such as `application`. + - The EDS `service_name` and `ClusterLoadAssignment#cluster_name` identify the endpoint assignment used by that application cluster. They default to the application cluster name. + +The following diagram shows the dedicated CDS and EDS arrangement used throughout this guide: + +``` mermaid +flowchart LR + Bootstrap[Envoy bootstrap] --> Management[xds_cluster] + Management -->|CDS stream| ControlPlane[Ruby control plane] + Management -->|EDS stream| ControlPlane + ControlPlane -->|Cluster| Application[application cluster] + ControlPlane -->|ClusterLoadAssignment| Workers[worker endpoints] + Traffic[Application traffic] --> Envoy + Envoy --> Application + Application --> Workers +``` + +## Serving Dedicated CDS and EDS + +Dedicated discovery services are a good fit when this control plane owns application clusters and their local workers, but should not claim Envoy's single ADS connection. Another control plane can then use ADS for coordinated listener, route, or other configuration. + +Create a control plane, publish an application cluster and its initial workers, then serve the dedicated CDS and EDS interfaces: + +``` ruby +require "async" +require "async/grpc/xds" +require "async/http/endpoint" + +control_plane = Async::GRPC::XDS::ControlPlane.new(identifier: "application-supervisor") +eds_config = Async::GRPC::XDS::ConfigSource.grpc("xds_cluster") +health_check = Async::GRPC::XDS::HTTPHealthCheck.build( + "/health", + interval: 2, + timeout: 1 +) + +control_plane.update_cluster( + "application", + protocol: :http1, + eds_config: eds_config, + health_checks: [health_check] +) + +control_plane.update_endpoints("application", [ + { + hostname: "worker-1", + addresses: [{address: "127.0.0.1", port: 9292}], + healthy: true, + }, + { + hostname: "worker-2", + addresses: [{address: "127.0.0.1", port: 9293}], + healthy: true, + }, +]) + +server = Async::GRPC::XDS::Server.new( + control_plane, + services: [ + Async::GRPC::XDS::ClusterDiscoveryService, + Async::GRPC::XDS::EndpointDiscoveryService, + ] +) + +endpoint = Async::HTTP::Endpoint.parse( + "http://0.0.0.0:18000", + protocol: Async::HTTP::Protocol::HTTP2 +) + +Sync do + server.run(endpoint) +end +``` + +The management endpoint uses HTTP/2 because xDS is a gRPC protocol. This example uses plaintext HTTP/2 within a trusted local network; deployments can instead configure TLS on the endpoint. + +### Envoy Bootstrap + +Envoy must know how to reach the management server before it can discover anything else. Define `xds_cluster` statically and configure CDS to use it: + +``` yaml +node: + id: application-proxy-1 + cluster: application-proxies + +admin: + address: + socket_address: + address: 127.0.0.1 + port_value: 19000 + +dynamic_resources: + cds_config: + resource_api_version: V3 + api_config_source: + api_type: GRPC + transport_api_version: V3 + grpc_services: + - envoy_grpc: + cluster_name: xds_cluster + +static_resources: + listeners: + - name: ingress + address: + socket_address: + address: 0.0.0.0 + port_value: 8080 + filter_chains: + - filters: + - name: envoy.filters.network.http_connection_manager + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager + stat_prefix: ingress_http + route_config: + name: local_route + validate_clusters: false + virtual_hosts: + - name: application + domains: ["*"] + routes: + - match: + prefix: "/" + route: + cluster: application + http_filters: + - name: envoy.filters.http.router + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router + + clusters: + - name: xds_cluster + connect_timeout: 1s + type: STATIC + typed_extension_protocol_options: + envoy.extensions.upstreams.http.v3.HttpProtocolOptions: + "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions + explicit_http_config: + http2_protocol_options: {} + load_assignment: + cluster_name: xds_cluster + endpoints: + - lb_endpoints: + - endpoint: + address: + socket_address: + address: 127.0.0.1 + port_value: 18000 +``` + +The static route names `application` before CDS has delivered that cluster, so `validate_clusters: false` allows the bootstrap configuration to load. Envoy keeps the discovered cluster warming until its EDS assignment arrives. + +The `xds_cluster` name must match the name passed to {ruby Async::GRPC::XDS::ConfigSource.grpc}. The generated application cluster then opens its dedicated EDS stream through the same management cluster. + +## Publishing Endpoint Changes + +The control plane is thread-safe and immediately notifies connected discovery streams after a resource changes. Publish the complete current endpoint assignment whenever workers start, stop, or change health: + +``` ruby +control_plane.update_endpoints("application", [ + { + hostname: "worker-2", + addresses: [{address: "127.0.0.1", port: 9293}], + healthy: true, + }, +]) +``` + +This implementation uses state-of-the-world discovery: each EDS response contains the complete assignment requested by Envoy. It does not currently implement the delta discovery RPCs. + +Every call to `update_cluster` or `update_endpoints` increments the version for that resource type, even if the generated resource is unchanged. Producers should therefore avoid publishing redundant updates. + +To remove a service completely, remove both resources: + +``` ruby +control_plane.remove_endpoints("application") +control_plane.remove_cluster("application") +``` + +### IP and Unix-Socket Addresses + +An endpoint must contain one or more addresses. An IP address uses `:address` and `:port`: + +``` ruby +{ + addresses: [{address: "127.0.0.1", port: 9292}], + healthy: true, +} +``` + +A Unix domain socket uses `:path`: + +``` ruby +{ + addresses: [{path: "/run/application/worker-1.ipc"}], + healthy: true, +} +``` + +Several addresses in one `:addresses` array describe alternative addresses for one logical load-balancer endpoint. The first becomes Envoy's primary address and the remainder become `additional_addresses`; they do not represent additional workers. + +### Health + +The `:healthy` value sets the EDS `health_status` for an endpoint. It accepts healthy, unhealthy, degraded, or unknown states; booleans map to healthy and unhealthy. + +This reported state is separate from active health checks. Adding a health check to the cluster tells Envoy to probe every published endpoint itself: + +``` ruby +health_check = Async::GRPC::XDS::HTTPHealthCheck.build( + "/health", + interval: 2, + timeout: 1, + unhealthy_threshold: 2, + healthy_threshold: 1 +) + +control_plane.update_cluster( + "application", + protocol: :http1, + eds_config: Async::GRPC::XDS::ConfigSource.grpc("xds_cluster"), + health_checks: [health_check] +) +``` + +Use reported health for information already known by the resource owner, such as whether a worker remains registered. Use active health checks when Envoy should independently verify that it can send application traffic to the endpoint. + +## ADS or Dedicated Services + +Both transports carry the same `DiscoveryRequest` and `DiscoveryResponse` messages. The difference is how Envoy organizes its streams and management servers: + +| | ADS | Dedicated CDS and EDS | +| --- | --- | --- | +| Streams | One aggregated stream for several resource types. | One stream per resource type. | +| Coordination | Provides ordering across related resource types from one control plane. | Each resource type progresses independently. | +| Ownership | Envoy has one ADS management server. | Each resource type can use its own configuration source. | +| Best fit | One control plane owns coordinated proxy configuration. | A focused control plane owns only clusters or endpoints. | + +The default server exposes ADS: + +``` ruby +control_plane = Async::GRPC::XDS::ControlPlane.new +control_plane.update_cluster("application", protocol: :http1) +control_plane.update_endpoints("application", endpoints) + +server = Async::GRPC::XDS::Server.new(control_plane) +``` + +{ruby Async::GRPC::XDS::Cluster.build} uses ADS for EDS by default, so no explicit `eds_config` is needed in this mode. Configure Envoy accordingly: + +``` yaml +dynamic_resources: + ads_config: + api_type: GRPC + transport_api_version: V3 + grpc_services: + - envoy_grpc: + cluster_name: xds_cluster + cds_config: + resource_api_version: V3 + ads: {} +``` + +Do not use the default ADS EDS source when serving dedicated CDS and EDS. In that arrangement, pass `ConfigSource.grpc("xds_cluster")` while building or updating every application cluster. + +## Using the Ruby xDS Client + +{ruby Async::GRPC::XDS::Client} is an experimental gRPC client which discovers a named cluster and its healthy endpoints over ADS, then load-balances RPC calls between them. + +Supply bootstrap configuration directly: + +``` ruby +bootstrap = { + xds_servers: [ + { + server_uri: "127.0.0.1:18000", + channel_creds: [{type: "insecure"}], + } + ], + node: { + id: "orders-client-1", + cluster: "orders-clients", + } +} + +Sync do + client = Async::GRPC::XDS::Client.new("application", bootstrap: bootstrap) + + begin + stub = client.stub(Greeter::Interface, "example.Greeter") + response = stub.say_hello(Greeter::Request.new(name: "World")) + ensure + client.close + end +end +``` + +Alternatively, pass the path to a JSON bootstrap file. When `bootstrap` is omitted, the client checks `GRPC_XDS_BOOTSTRAP` and then `~/.config/grpc/bootstrap.json`. + +The Ruby client currently consumes ADS, supports CDS and EDS, filters endpoints by reported health, and provides basic client-side load balancing and retry behavior. Dedicated CDS and EDS client streams are not yet implemented. + +## Operational Checks + +Envoy's admin interface shows whether it accepted the resources and whether the application cluster has usable members: + +~~~ bash +$ curl -s http://127.0.0.1:19000/config_dump +$ curl -s http://127.0.0.1:19000/clusters +$ curl -s http://127.0.0.1:19000/stats | grep -E 'cluster\.application\.(warming|membership_healthy|update_rejected)' +~~~ + +When a cluster remains in `warming`, check: + + - The `xds_cluster` address and port reach the Ruby server using HTTP/2. + - The management cluster name matches the name in every gRPC configuration source. + - The CDS cluster name, EDS service name, and `ClusterLoadAssignment#cluster_name` agree. + - A dedicated CDS cluster uses a dedicated EDS configuration source rather than `ads: {}`. + - Envoy has not incremented `update_rejected`; rejected responses are also returned to the control plane as NACKs and logged. + +## Current Scope + +`async-grpc-xds` currently provides: + + - xDS v3 protobuf definitions. + - State-of-the-world ADS, CDS, and EDS server streams. + - In-memory cluster and endpoint resources with per-type versions. + - HTTP/1 and HTTP/2 upstream cluster configuration. + - IP, Unix-socket, health-status, active HTTP health-check, and out-of-band ORCA policy resource builders. + - An experimental ADS-based Ruby gRPC client. + +Delta discovery, complete NACK recovery, persistent resource storage, LDS/RDS serving, locality weighting, and complete routing semantics are not implemented yet. diff --git a/guides/links.yaml b/guides/links.yaml new file mode 100644 index 0000000..7f527b0 --- /dev/null +++ b/guides/links.yaml @@ -0,0 +1,2 @@ +getting-started: + order: 1 diff --git a/readme.md b/readme.md index e1db12d..78299bc 100644 --- a/readme.md +++ b/readme.md @@ -10,17 +10,7 @@ This gem contains the experimental xDS implementation extracted from `async-grpc Please see the [project documentation](https://socketry.github.io/async-grpc-xds/) for more details. -By default, {ruby Async::GRPC::XDS::Server} serves the Aggregated Discovery Service. To expose dedicated CDS and EDS streams while leaving ADS available to another control plane, select the resource-specific services: - -``` ruby -server = Async::GRPC::XDS::Server.new( - control_plane, - services: [ - Async::GRPC::XDS::ClusterDiscoveryService, - Async::GRPC::XDS::EndpointDiscoveryService, - ] -) -``` + - [Getting Started](https://socketry.github.io/async-grpc-xds/guides/getting-started/index) - This guide explains how to use `async-grpc-xds` to publish CDS and EDS resources to Envoy, or to discover gRPC backends from Ruby using xDS. ## Status diff --git a/releases.md b/releases.md index 870063b..da60e2b 100644 --- a/releases.md +++ b/releases.md @@ -4,6 +4,7 @@ - Add dedicated Cluster Discovery Service and Endpoint Discovery Service implementations, allowing resource-specific xDS streams without claiming ADS. - Allow generated clusters to use a dedicated EDS configuration source instead of ADS. + - Add a guide covering control-plane setup, Envoy bootstrap configuration, resource updates, and Ruby client usage. ## v0.3.0 From d44a284f40f56756741b040274c0ca74aab8191d Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Thu, 6 Aug 2026 22:15:51 +1200 Subject: [PATCH 04/12] Remove redundant status section --- readme.md | 4 ---- 1 file changed, 4 deletions(-) diff --git a/readme.md b/readme.md index 78299bc..0ab0738 100644 --- a/readme.md +++ b/readme.md @@ -12,10 +12,6 @@ Please see the [project documentation](https://socketry.github.io/async-grpc-xds - [Getting Started](https://socketry.github.io/async-grpc-xds/guides/getting-started/index) - This guide explains how to use `async-grpc-xds` to publish CDS and EDS resources to Envoy, or to discover gRPC backends from Ruby using xDS. -## Status - -This is an early implementation focused on CDS and EDS over ADS or dedicated discovery streams. LDS/RDS, full routing semantics, NACK handling, locality weighting, and delta xDS are not complete yet. - ## Testing The `xds/` directory contains a Docker Compose integration environment with a Go xDS control plane and Ruby gRPC backends. From f71218f0e305a027918a51f12a798e238386c743 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Thu, 6 Aug 2026 22:16:52 +1200 Subject: [PATCH 05/12] Remove guide from release notes --- releases.md | 1 - 1 file changed, 1 deletion(-) diff --git a/releases.md b/releases.md index da60e2b..870063b 100644 --- a/releases.md +++ b/releases.md @@ -4,7 +4,6 @@ - Add dedicated Cluster Discovery Service and Endpoint Discovery Service implementations, allowing resource-specific xDS streams without claiming ADS. - Allow generated clusters to use a dedicated EDS configuration source instead of ADS. - - Add a guide covering control-plane setup, Envoy bootstrap configuration, resource updates, and Ruby client usage. ## v0.3.0 From 635bdae2e8725b5d832262cd52c8f42883c44dbb Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Thu, 6 Aug 2026 22:17:20 +1200 Subject: [PATCH 06/12] Exclude test fixtures from gem --- async-grpc-xds.gemspec | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/async-grpc-xds.gemspec b/async-grpc-xds.gemspec index b79324f..323435b 100644 --- a/async-grpc-xds.gemspec +++ b/async-grpc-xds.gemspec @@ -20,7 +20,7 @@ Gem::Specification.new do |spec| "source_code_uri" => "https://github.com/socketry/async-grpc-xds.git", } - spec.files = Dir.glob(["{context,fixtures,lib,proto,xds}/**/*", "*.md"], File::FNM_DOTMATCH, base: __dir__) + spec.files = Dir.glob(["{context,lib,proto,xds}/**/*", "*.md"], File::FNM_DOTMATCH, base: __dir__) spec.required_ruby_version = ">= 3.3" From e193e3b2bfbde4c3bd726b7b759d3c6c19d575bc Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Thu, 6 Aug 2026 22:20:40 +1200 Subject: [PATCH 07/12] Extract XDS stream --- lib/async/grpc/xds.rb | 1 + lib/async/grpc/xds/control_plane.rb | 4 +- lib/async/grpc/xds/discovery_service.rb | 88 +------------------- lib/async/grpc/xds/stream.rb | 101 +++++++++++++++++++++++ test/async/grpc/xds/discovery_service.rb | 6 +- 5 files changed, 108 insertions(+), 92 deletions(-) create mode 100644 lib/async/grpc/xds/stream.rb diff --git a/lib/async/grpc/xds.rb b/lib/async/grpc/xds.rb index 2468f31..ac6d80f 100644 --- a/lib/async/grpc/xds.rb +++ b/lib/async/grpc/xds.rb @@ -19,6 +19,7 @@ require_relative "xds/http_health_check" require_relative "xds/client_side_weighted_round_robin" require_relative "xds/control_plane" +require_relative "xds/stream" require_relative "xds/discovery_service" require_relative "xds/service" require_relative "xds/cluster_discovery_service" diff --git a/lib/async/grpc/xds/control_plane.rb b/lib/async/grpc/xds/control_plane.rb index 4521b57..b9c00a2 100644 --- a/lib/async/grpc/xds/control_plane.rb +++ b/lib/async/grpc/xds/control_plane.rb @@ -152,7 +152,7 @@ def response(type_url, names = nil) end # Register a stream to receive resource-change notifications. - # @parameter stream [DiscoveryService::Stream] The stream to register. + # @parameter stream [Stream] The stream to register. def register_stream(stream) @mutex.synchronize do @streams.add(stream) @@ -160,7 +160,7 @@ def register_stream(stream) end # Remove a registered stream. - # @parameter stream [DiscoveryService::Stream] The stream to remove. + # @parameter stream [Stream] The stream to remove. def remove_stream(stream) @mutex.synchronize do @streams.delete(stream) diff --git a/lib/async/grpc/xds/discovery_service.rb b/lib/async/grpc/xds/discovery_service.rb index 1310889..68934d4 100644 --- a/lib/async/grpc/xds/discovery_service.rb +++ b/lib/async/grpc/xds/discovery_service.rb @@ -5,12 +5,11 @@ require "async" require "async/grpc/service" -require "async/queue" require "protocol/grpc/error" require "protocol/grpc/status" -require "set" require_relative "control_plane" +require_relative "stream" module Async module GRPC @@ -64,91 +63,6 @@ def delta_resources ) end - # Represents one discovery stream and its subscribed resources. - class Stream - # Initialize a discovery stream. - # @parameter control_plane [ControlPlane] The control plane that provides resources. - # @parameter output [Interface(:write)] The discovery response stream. - # @parameter resource_type [String | Nil] The fixed resource type, or `nil` for aggregated discovery. - def initialize(control_plane, output, resource_type: nil) - @control_plane = control_plane - @output = output - @resource_type = resource_type - @subscriptions = Hash.new{|hash, type_url| hash[type_url] = Set.new} - @versions = {} - @queue = Async::Queue.new - @closed = false - end - - # Process a discovery request and update the stream's subscriptions. - # @parameter request [Envoy::Service::Discovery::V3::DiscoveryRequest] The discovery request. - def request(request) - type_url = request.type_url - - if @resource_type - if type_url.nil? || type_url.empty? - type_url = @resource_type - elsif type_url != @resource_type - raise Protocol::GRPC::Error.new( - Protocol::GRPC::Status::INVALID_ARGUMENT, - "Expected resource type #{@resource_type.inspect}, but received #{type_url.inspect}." - ) - end - elsif type_url.nil? || type_url.empty? - return - end - - if request.error_detail - Console.warn(self, "Received xDS NACK.", type_url: type_url, error_detail: request.error_detail) - return - end - - if request.resource_names.any? - @subscriptions[type_url].merge(request.resource_names) - else - @subscriptions[type_url] - end - - @queue << type_url - end - - # Schedule a resource type for delivery after it changes. - # @parameter type_url [String] The changed xDS resource type URL. - def changed(type_url) - return if @resource_type && type_url != @resource_type - - @queue << type_url unless @closed - end - - # Deliver scheduled resource updates until the stream closes. - # @asynchronous - def run - until @closed - type_url = @queue.dequeue - flush(type_url) - end - end - - # Deliver the latest resource version for a subscribed type. - # @parameter type_url [String] The xDS resource type URL. - def flush(type_url) - names = @subscriptions[type_url] - return unless names - - version = @control_plane.version(type_url) - return if @versions[type_url] == version - - response = @control_plane.response(type_url, names) - @output.write(response) - @versions[type_url] = version - end - - # Close the stream and stop waiting for changes. - def close - @closed = true - @queue.close - end - end end end end diff --git a/lib/async/grpc/xds/stream.rb b/lib/async/grpc/xds/stream.rb new file mode 100644 index 0000000..71b80c9 --- /dev/null +++ b/lib/async/grpc/xds/stream.rb @@ -0,0 +1,101 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "async/queue" +require "protocol/grpc/error" +require "protocol/grpc/status" +require "set" + +module Async + module GRPC + module XDS + # Represents one discovery stream and its subscribed resources. + class Stream + # Initialize a discovery stream. + # @parameter control_plane [ControlPlane] The control plane that provides resources. + # @parameter output [Interface(:write)] The discovery response stream. + # @parameter resource_type [String | Nil] The fixed resource type, or `nil` for aggregated discovery. + def initialize(control_plane, output, resource_type: nil) + @control_plane = control_plane + @output = output + @resource_type = resource_type + @subscriptions = Hash.new{|hash, type_url| hash[type_url] = Set.new} + @versions = {} + @queue = Async::Queue.new + @closed = false + end + + # Process a discovery request and update the stream's subscriptions. + # @parameter request [Envoy::Service::Discovery::V3::DiscoveryRequest] The discovery request. + def request(request) + type_url = request.type_url + + if @resource_type + if type_url.nil? || type_url.empty? + type_url = @resource_type + elsif type_url != @resource_type + raise Protocol::GRPC::Error.new( + Protocol::GRPC::Status::INVALID_ARGUMENT, + "Expected resource type #{@resource_type.inspect}, but received #{type_url.inspect}." + ) + end + elsif type_url.nil? || type_url.empty? + return + end + + if request.error_detail + Console.warn(self, "Received xDS NACK.", type_url: type_url, error_detail: request.error_detail) + return + end + + if request.resource_names.any? + @subscriptions[type_url].merge(request.resource_names) + else + @subscriptions[type_url] + end + + @queue << type_url + end + + # Schedule a resource type for delivery after it changes. + # @parameter type_url [String] The changed xDS resource type URL. + def changed(type_url) + return if @resource_type && type_url != @resource_type + + @queue << type_url unless @closed + end + + # Deliver scheduled resource updates until the stream closes. + # @asynchronous + def run + until @closed + type_url = @queue.dequeue + flush(type_url) + end + end + + # Deliver the latest resource version for a subscribed type. + # @parameter type_url [String] The xDS resource type URL. + def flush(type_url) + names = @subscriptions[type_url] + return unless names + + version = @control_plane.version(type_url) + return if @versions[type_url] == version + + response = @control_plane.response(type_url, names) + @output.write(response) + @versions[type_url] = version + end + + # Close the stream and stop waiting for changes. + def close + @closed = true + @queue.close + end + end + end + end +end diff --git a/test/async/grpc/xds/discovery_service.rb b/test/async/grpc/xds/discovery_service.rb index 34840ba..c7125cf 100644 --- a/test/async/grpc/xds/discovery_service.rb +++ b/test/async/grpc/xds/discovery_service.rb @@ -28,7 +28,7 @@ def request(type_url: nil, resource_names: ["myservice"]) end def stream_for(resource_type, output) - subject::Stream.new(control_plane, output, resource_type: resource_type) + Async::GRPC::XDS::Stream.new(control_plane, output, resource_type: resource_type) end it "serves clusters with an explicit resource type" do @@ -63,7 +63,7 @@ def stream_for(resource_type, output) it "requires a resource type for aggregated streams" do responses = output - stream = subject::Stream.new(control_plane, responses) + stream = Async::GRPC::XDS::Stream.new(control_plane, responses) stream.request(request) @@ -171,7 +171,7 @@ def stream_for(resource_type, output) end end - mock(subject::Stream) do |stream_mock| + mock(Async::GRPC::XDS::Stream) do |stream_mock| stream_mock.replace(:new) do |given_control_plane, given_output, resource_type:| expect(given_control_plane).to be == control_plane expect(given_output).to be == output From 94c1934dd4f123bad61278537ee2142dd005e6b8 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Thu, 6 Aug 2026 22:25:55 +1200 Subject: [PATCH 08/12] Improve XDS stream coverage --- lib/async/grpc/xds/stream.rb | 11 +- test/async/grpc/xds/discovery_service.rb | 133 ------------ test/async/grpc/xds/stream.rb | 248 +++++++++++++++++++++++ 3 files changed, 254 insertions(+), 138 deletions(-) create mode 100644 test/async/grpc/xds/stream.rb diff --git a/lib/async/grpc/xds/stream.rb b/lib/async/grpc/xds/stream.rb index 71b80c9..5223ea0 100644 --- a/lib/async/grpc/xds/stream.rb +++ b/lib/async/grpc/xds/stream.rb @@ -21,7 +21,7 @@ def initialize(control_plane, output, resource_type: nil) @control_plane = control_plane @output = output @resource_type = resource_type - @subscriptions = Hash.new{|hash, type_url| hash[type_url] = Set.new} + @subscriptions = {} @versions = {} @queue = Async::Queue.new @closed = false @@ -50,10 +50,10 @@ def request(request) return end - if request.resource_names.any? - @subscriptions[type_url].merge(request.resource_names) - else - @subscriptions[type_url] + names = Set.new(request.resource_names) + if @subscriptions[type_url] != names + @subscriptions[type_url] = names + @versions.delete(type_url) end @queue << type_url @@ -63,6 +63,7 @@ def request(request) # @parameter type_url [String] The changed xDS resource type URL. def changed(type_url) return if @resource_type && type_url != @resource_type + return unless @subscriptions.key?(type_url) @queue << type_url unless @closed end diff --git a/test/async/grpc/xds/discovery_service.rb b/test/async/grpc/xds/discovery_service.rb index c7125cf..38a2c7c 100644 --- a/test/async/grpc/xds/discovery_service.rb +++ b/test/async/grpc/xds/discovery_service.rb @@ -7,19 +7,10 @@ require "async/grpc/xds/endpoint_discovery_service" require "async/grpc/xds/server" require "async/grpc/xds/service" -require "envoy/config/cluster/v3/cluster_pb" -require "envoy/config/endpoint/v3/endpoint_pb" -require "google/rpc/status_pb" describe Async::GRPC::XDS::DiscoveryService do let(:control_plane) {Async::GRPC::XDS::ControlPlane.new} - def output - [].tap do |responses| - responses.define_singleton_method(:write){|response| self << response} - end - end - def request(type_url: nil, resource_names: ["myservice"]) Envoy::Service::Discovery::V3::DiscoveryRequest.new( type_url: type_url, @@ -27,130 +18,6 @@ def request(type_url: nil, resource_names: ["myservice"]) ) end - def stream_for(resource_type, output) - Async::GRPC::XDS::Stream.new(control_plane, output, resource_type: resource_type) - end - - it "serves clusters with an explicit resource type" do - control_plane.update_cluster("myservice") - responses = output - stream = stream_for(Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE, responses) - - stream.request(request(type_url: Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE)) - stream.flush(Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE) - - cluster = Envoy::Config::Cluster::V3::Cluster.decode(responses.first.resources.first.value) - expect(cluster.name).to be == "myservice" - ensure - stream&.close - end - - it "uses the implied endpoint resource type when omitted" do - control_plane.update_endpoints("myservice", [ - {addresses: [{address: "127.0.0.1", port: 50051}], healthy: true} - ]) - responses = output - stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, responses) - - stream.request(request) - stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) - - assignment = Envoy::Config::Endpoint::V3::ClusterLoadAssignment.decode(responses.first.resources.first.value) - expect(assignment.cluster_name).to be == "myservice" - ensure - stream&.close - end - - it "requires a resource type for aggregated streams" do - responses = output - stream = Async::GRPC::XDS::Stream.new(control_plane, responses) - - stream.request(request) - - expect(responses).to be(:empty?) - ensure - stream&.close - end - - it "subscribes to every resource when no names are specified" do - control_plane.update_endpoints("myservice", []) - responses = output - stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, responses) - - stream.request(request(resource_names: [])) - stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) - - expect(responses.first.resources.size).to be == 1 - ensure - stream&.close - end - - it "logs and ignores a NACK" do - stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, output) - nack = request - nack.error_detail = Google::Rpc::Status.new(message: "invalid resource") - - stream.request(nack) - - expect(stream).not.to be_nil - ensure - stream&.close - end - - it "rejects a resource type that does not belong to the service" do - stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, output) - - expect do - stream.request(request(type_url: Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE)) - end.to raise_exception(Protocol::GRPC::Error) - ensure - stream&.close - end - - it "ignores changes for resource types that do not belong to the service" do - responses = output - stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, responses) - stream.request(request) - stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) - - stream.changed(Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE) - stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) - - expect(responses.size).to be == 1 - ensure - stream&.close - end - - it "accepts changes for its resource type" do - stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, output) - - stream.changed(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) - - expect(stream).not.to be_nil - ensure - stream&.close - end - - it "delivers queued resource changes" do - control_plane.update_endpoints("myservice", [ - {addresses: [{address: "127.0.0.1", port: 50051}], healthy: true} - ]) - responses = output - stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, responses) - stream.request(request) - - responses.define_singleton_method(:write) do |response| - self << response - stream.close - end - - stream.run - - expect(responses.first.type_url).to be == Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE - ensure - stream&.close - end - it "coordinates endpoint discovery stream tasks" do service = Async::GRPC::XDS::EndpointDiscoveryService.new(control_plane) input = [request] diff --git a/test/async/grpc/xds/stream.rb b/test/async/grpc/xds/stream.rb new file mode 100644 index 0000000..9d9178e --- /dev/null +++ b/test/async/grpc/xds/stream.rb @@ -0,0 +1,248 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "async/grpc/xds/stream" +require "async/grpc/xds/control_plane" +require "envoy/config/cluster/v3/cluster_pb" +require "envoy/config/endpoint/v3/endpoint_pb" +require "google/rpc/status_pb" + +describe Async::GRPC::XDS::Stream do + let(:control_plane) {Async::GRPC::XDS::ControlPlane.new} + + def output + [].tap do |responses| + responses.define_singleton_method(:write){|response| self << response} + end + end + + def request(type_url: nil, resource_names: ["myservice"]) + Envoy::Service::Discovery::V3::DiscoveryRequest.new( + type_url: type_url, + resource_names: resource_names + ) + end + + def stream_for(resource_type, output) + subject.new(control_plane, output, resource_type: resource_type) + end + + it "serves clusters with an explicit resource type" do + control_plane.update_cluster("myservice") + responses = output + stream = stream_for(Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE, responses) + + stream.request(request(type_url: Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE)) + stream.flush(Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE) + + cluster = Envoy::Config::Cluster::V3::Cluster.decode(responses.first.resources.first.value) + expect(cluster.name).to be == "myservice" + ensure + stream&.close + end + + it "uses the implied endpoint resource type when omitted" do + control_plane.update_endpoints("myservice", [ + {addresses: [{address: "127.0.0.1", port: 50051}], healthy: true} + ]) + responses = output + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, responses) + + stream.request(request) + stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + + assignment = Envoy::Config::Endpoint::V3::ClusterLoadAssignment.decode(responses.first.resources.first.value) + expect(assignment.cluster_name).to be == "myservice" + ensure + stream&.close + end + + it "requires a resource type for aggregated streams" do + responses = output + stream = subject.new(control_plane, responses) + + stream.request(request) + + expect(responses).to be(:empty?) + ensure + stream&.close + end + + it "subscribes to every resource when no names are specified" do + control_plane.update_endpoints("myservice", []) + responses = output + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, responses) + + stream.request(request(resource_names: [])) + stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + + expect(responses.first.resources.size).to be == 1 + ensure + stream&.close + end + + it "does not publish resources before subscribing" do + control_plane.update_endpoints("myservice", []) + responses = output + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, responses) + + stream.changed(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + + expect(responses).to be(:empty?) + ensure + stream&.close + end + + it "replaces subscriptions and responds without a resource change" do + control_plane.update_endpoints("first", []) + control_plane.update_endpoints("second", []) + responses = output + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, responses) + + stream.request(request(resource_names: ["first"])) + stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + stream.request(request(resource_names: ["second"])) + stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + + assignments = responses.map do |response| + response.resources.map do |resource| + Envoy::Config::Endpoint::V3::ClusterLoadAssignment.decode(resource.value).cluster_name + end + end + + expect(assignments).to be == [["first"], ["second"]] + ensure + stream&.close + end + + it "does not respond again when an acknowledgement keeps the same subscription" do + control_plane.update_endpoints("myservice", []) + responses = output + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, responses) + + stream.request(request) + stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + stream.request(request) + stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + + expect(responses.size).to be == 1 + ensure + stream&.close + end + + it "logs and ignores a NACK" do + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, output) + nack = request + nack.error_detail = Google::Rpc::Status.new(message: "invalid resource") + + stream.request(nack) + + expect(stream).not.to be_nil + ensure + stream&.close + end + + it "rejects a resource type that does not belong to the service" do + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, output) + + expect do + stream.request(request(type_url: Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE)) + end.to raise_exception(Protocol::GRPC::Error) + ensure + stream&.close + end + + it "ignores changes for resource types that do not belong to the service" do + responses = output + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, responses) + stream.request(request) + stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + + stream.changed(Async::GRPC::XDS::ControlPlane::CLUSTER_TYPE) + stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + + expect(responses.size).to be == 1 + ensure + stream&.close + end + + it "publishes resource changes after subscribing" do + control_plane.update_endpoints("myservice", []) + responses = output + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, responses) + stream.request(request) + stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + + control_plane.update_endpoints("myservice", [ + {addresses: [{address: "127.0.0.1", port: 50051}], healthy: true} + ]) + stream.changed(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + + expect(responses.size).to be == 2 + ensure + stream&.close + end + + it "ignores resource changes after closing" do + control_plane.update_endpoints("myservice", []) + responses = output + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, responses) + stream.request(request) + stream.flush(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + stream.close + + control_plane.update_endpoints("myservice", [ + {addresses: [{address: "127.0.0.1", port: 50051}], healthy: true} + ]) + stream.changed(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE) + stream.run + + expect(responses.size).to be == 1 + ensure + stream&.close + end + + it "delivers queued resource changes" do + control_plane.update_endpoints("myservice", [ + {addresses: [{address: "127.0.0.1", port: 50051}], healthy: true} + ]) + responses = output + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, responses) + stream.request(request) + + responses.define_singleton_method(:write) do |response| + self << response + stream.close + end + + stream.run + + expect(responses.first.type_url).to be == Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE + ensure + stream&.close + end + + it "unblocks a running stream when closed" do + stream = stream_for(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, output) + finished = false + + Sync do |task| + runner = task.async do + stream.run + finished = true + end + + task.yield + stream.close + runner.wait + end + + expect(finished).to be == true + ensure + stream&.close + end +end From 93f22f75a608ff5001331cc3101306872beb1eea Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Thu, 6 Aug 2026 23:23:39 +1200 Subject: [PATCH 09/12] Add Envoy discovery integration test --- .github/workflows/test-xds.yaml | 10 +- .../client}/Dockerfile.backend | 2 +- .../client}/Dockerfile.control-plane | 0 {xds => integration/client}/backend_server.rb | 0 .../client}/docker-compose.yaml | 22 +-- {xds => integration/client}/go.mod | 0 {xds => integration/client}/go.sum | 0 {xds => integration/client}/readme.md | 6 +- .../client}/test/async/grpc/xds/client.rb | 0 .../test/async/grpc/xds/control_plane.rb | 0 {xds => integration/client}/test_server.go | 0 integration/envoy/docker-compose.yaml | 29 ++++ integration/envoy/envoy.yaml | 39 +++++ integration/envoy/readme.md | 12 ++ integration/envoy/test.rb | 135 ++++++++++++++++++ readme.md | 5 +- 16 files changed, 242 insertions(+), 18 deletions(-) rename {xds => integration/client}/Dockerfile.backend (89%) rename {xds => integration/client}/Dockerfile.control-plane (100%) rename {xds => integration/client}/backend_server.rb (100%) rename {xds => integration/client}/docker-compose.yaml (77%) rename {xds => integration/client}/go.mod (100%) rename {xds => integration/client}/go.sum (100%) rename {xds => integration/client}/readme.md (94%) rename {xds => integration/client}/test/async/grpc/xds/client.rb (100%) rename {xds => integration/client}/test/async/grpc/xds/control_plane.rb (100%) rename {xds => integration/client}/test_server.go (100%) create mode 100644 integration/envoy/docker-compose.yaml create mode 100644 integration/envoy/envoy.yaml create mode 100644 integration/envoy/readme.md create mode 100644 integration/envoy/test.rb diff --git a/.github/workflows/test-xds.yaml b/.github/workflows/test-xds.yaml index 12fb10a..9a381f6 100644 --- a/.github/workflows/test-xds.yaml +++ b/.github/workflows/test-xds.yaml @@ -31,6 +31,14 @@ jobs: - name: Run tests timeout-minutes: 10 + working-directory: integration/client env: RUBY_VERSION: ${{matrix.ruby}} - run: docker compose -f xds/docker-compose.yaml up --build --exit-code-from tests + run: docker compose up --build --exit-code-from tests + + - name: Test dedicated discovery services with Envoy + timeout-minutes: 10 + working-directory: integration/envoy + env: + RUBY_VERSION: ${{matrix.ruby}} + run: docker compose up --build --exit-code-from tests diff --git a/xds/Dockerfile.backend b/integration/client/Dockerfile.backend similarity index 89% rename from xds/Dockerfile.backend rename to integration/client/Dockerfile.backend index 0f68433..07df6e9 100644 --- a/xds/Dockerfile.backend +++ b/integration/client/Dockerfile.backend @@ -21,4 +21,4 @@ COPY . . EXPOSE ${PORT:-50051} # Run gRPC server -CMD bundle exec ruby xds/backend_server.rb +CMD bundle exec ruby integration/client/backend_server.rb diff --git a/xds/Dockerfile.control-plane b/integration/client/Dockerfile.control-plane similarity index 100% rename from xds/Dockerfile.control-plane rename to integration/client/Dockerfile.control-plane diff --git a/xds/backend_server.rb b/integration/client/backend_server.rb similarity index 100% rename from xds/backend_server.rb rename to integration/client/backend_server.rb diff --git a/xds/docker-compose.yaml b/integration/client/docker-compose.yaml similarity index 77% rename from xds/docker-compose.yaml rename to integration/client/docker-compose.yaml index 5f10883..3821595 100644 --- a/xds/docker-compose.yaml +++ b/integration/client/docker-compose.yaml @@ -19,9 +19,9 @@ services: # Backend gRPC server 1 backend-1: build: - context: .. - dockerfile: xds/Dockerfile.backend - command: bundle exec ruby xds/backend_server.rb + context: ../.. + dockerfile: integration/client/Dockerfile.backend + command: bundle exec ruby integration/client/backend_server.rb environment: - BUNDLE_WITHOUT=maintenance - PORT=50051 @@ -36,9 +36,9 @@ services: # Backend gRPC server 2 backend-2: build: - context: .. - dockerfile: xds/Dockerfile.backend - command: bundle exec ruby xds/backend_server.rb + context: ../.. + dockerfile: integration/client/Dockerfile.backend + command: bundle exec ruby integration/client/backend_server.rb environment: - BUNDLE_WITHOUT=maintenance - PORT=50052 @@ -53,9 +53,9 @@ services: # Backend gRPC server 3 backend-3: build: - context: .. - dockerfile: xds/Dockerfile.backend - command: bundle exec ruby xds/backend_server.rb + context: ../.. + dockerfile: integration/client/Dockerfile.backend + command: bundle exec ruby integration/client/backend_server.rb environment: - BUNDLE_WITHOUT=maintenance - PORT=50053 @@ -71,9 +71,9 @@ services: tests: image: ruby:${RUBY_VERSION:-latest} volumes: - - ../:/code + - ../../:/code working_dir: /code - command: bash -c "bundle install && bundle exec sus xds/test" + command: bash -c "bundle install && bundle exec sus integration/client/test" environment: - BUNDLE_GEMFILE=/code/gems.rb - BUNDLE_WITHOUT=maintenance diff --git a/xds/go.mod b/integration/client/go.mod similarity index 100% rename from xds/go.mod rename to integration/client/go.mod diff --git a/xds/go.sum b/integration/client/go.sum similarity index 100% rename from xds/go.sum rename to integration/client/go.sum diff --git a/xds/readme.md b/integration/client/readme.md similarity index 94% rename from xds/readme.md rename to integration/client/readme.md index 093f204..85fef31 100644 --- a/xds/readme.md +++ b/integration/client/readme.md @@ -1,4 +1,4 @@ -# xDS Integration Tests +# xDS Client Integration Test This directory contains Docker Compose configuration and test files for xDS integration testing, following the same pattern as `async-redis` (Sentinel and Cluster tests). @@ -14,13 +14,13 @@ The Docker Compose setup includes: From this directory: ```bash -cd xds +cd integration/client docker compose up --build --exit-code-from tests ``` ## Test Structure -Tests are located in `xds/test/async/grpc/xds/` and follow the same pattern as other async-grpc tests: +Tests are located in `integration/client/test/async/grpc/xds/` and follow the same pattern as other async-grpc tests: - `client.rb`: Tests for `Async::GRPC::XDS::Client` - Tests use `Sus::Fixtures::Async::ReactorContext` for async test support diff --git a/xds/test/async/grpc/xds/client.rb b/integration/client/test/async/grpc/xds/client.rb similarity index 100% rename from xds/test/async/grpc/xds/client.rb rename to integration/client/test/async/grpc/xds/client.rb diff --git a/xds/test/async/grpc/xds/control_plane.rb b/integration/client/test/async/grpc/xds/control_plane.rb similarity index 100% rename from xds/test/async/grpc/xds/control_plane.rb rename to integration/client/test/async/grpc/xds/control_plane.rb diff --git a/xds/test_server.go b/integration/client/test_server.go similarity index 100% rename from xds/test_server.go rename to integration/client/test_server.go diff --git a/integration/envoy/docker-compose.yaml b/integration/envoy/docker-compose.yaml new file mode 100644 index 0000000..297f7cb --- /dev/null +++ b/integration/envoy/docker-compose.yaml @@ -0,0 +1,29 @@ +services: + tests: + image: ruby:${RUBY_VERSION:-latest} + volumes: + - ../../:/code + working_dir: /code + command: bash -c "bundle install && bundle exec sus integration/envoy/test.rb" + environment: + - BUNDLE_GEMFILE=/code/gems.rb + - BUNDLE_WITHOUT=maintenance + - CONSOLE_OUTPUT=XTerm + - COVERAGE=${COVERAGE} + - ENVOY_ADMIN_URI=http://envoy:19000 + - READINESS_BIND=http://0.0.0.0:18001 + - XDS_BIND=http://0.0.0.0:18000 + healthcheck: + test: ["CMD", "curl", "--fail", "--silent", "--output", "/dev/null", "http://127.0.0.1:18001/"] + interval: 1s + timeout: 1s + retries: 300 + + envoy: + image: envoyproxy/envoy:v1.39-latest + command: ["envoy", "-c", "/etc/envoy/envoy.yaml", "--log-level", "warning"] + volumes: + - ./envoy.yaml:/etc/envoy/envoy.yaml:ro + depends_on: + tests: + condition: service_healthy diff --git a/integration/envoy/envoy.yaml b/integration/envoy/envoy.yaml new file mode 100644 index 0000000..42c157f --- /dev/null +++ b/integration/envoy/envoy.yaml @@ -0,0 +1,39 @@ +node: + id: async-grpc-xds-envoy-integration + cluster: async-grpc-xds-envoy-integration + +admin: + address: + socket_address: + address: 0.0.0.0 + port_value: 19000 + +dynamic_resources: + cds_config: + resource_api_version: V3 + api_config_source: + api_type: GRPC + transport_api_version: V3 + grpc_services: + - envoy_grpc: + cluster_name: xds_cluster + +static_resources: + clusters: + - name: xds_cluster + connect_timeout: 1s + type: STRICT_DNS + typed_extension_protocol_options: + envoy.extensions.upstreams.http.v3.HttpProtocolOptions: + "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions + explicit_http_config: + http2_protocol_options: {} + load_assignment: + cluster_name: xds_cluster + endpoints: + - lb_endpoints: + - endpoint: + address: + socket_address: + address: tests + port_value: 18000 diff --git a/integration/envoy/readme.md b/integration/envoy/readme.md new file mode 100644 index 0000000..7fa20d7 --- /dev/null +++ b/integration/envoy/readme.md @@ -0,0 +1,12 @@ +# Envoy Integration Test + +This integration test exercises the Ruby xDS control plane's dedicated CDS and EDS services against Envoy. + +It publishes a cluster and endpoint, confirms Envoy accepted both, replaces the endpoint, and confirms Envoy applied the update without rejecting either resource. + +Run it from this directory: + +```bash +cd integration/envoy +docker compose up --build --exit-code-from tests +``` diff --git a/integration/envoy/test.rb b/integration/envoy/test.rb new file mode 100644 index 0000000..886dc85 --- /dev/null +++ b/integration/envoy/test.rb @@ -0,0 +1,135 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "async/grpc/xds/cluster_discovery_service" +require "async/grpc/xds/config_source" +require "async/grpc/xds/control_plane" +require "async/grpc/xds/endpoint_discovery_service" +require "async/grpc/xds/server" +require "async/http/endpoint" +require "async/http/internet" +require "async/http/server" +require "json" +require "protocol/http/response" +require "sus/fixtures/async" + +describe "dedicated discovery services with Envoy" do + include Sus::Fixtures::Async::ReactorContext + + let(:admin_uri) {ENV["ENVOY_ADMIN_URI"]} + let(:bind_uri) {ENV["XDS_BIND"] || "http://0.0.0.0:18000"} + let(:readiness_uri) {ENV["READINESS_BIND"] || "http://0.0.0.0:18001"} + let(:cluster_name) {"application"} + let(:internet) {Async::HTTP::Internet.new} + + def get_admin(path) + response = internet.get("#{admin_uri}#{path}") + raise "Envoy admin request failed: #{response.status}" unless response.success? + + JSON.parse(response.read) + ensure + response&.close + end + + def eventually(timeout: 15, interval: 0.1) + deadline = Time.now + timeout + last_error = nil + + while Time.now < deadline + begin + if result = yield + return result + end + rescue => error + last_error = error + end + + sleep(interval) + end + + raise last_error if last_error + raise "Timed out waiting for condition" + end + + def cluster_status + get_admin("/clusters?format=json").fetch("cluster_statuses").find do |status| + status["name"] == cluster_name + end + end + + def addresses(status) + status.fetch("host_statuses", []).map do |host| + socket_address = host.fetch("address").fetch("socket_address") + [socket_address.fetch("address"), socket_address.fetch("port_value")] + end + end + + def stats + get_admin("/stats?format=json").fetch("stats").filter_map do |stat| + [stat.fetch("name"), stat.fetch("value")] if stat.key?("name") + end.to_h + end + + it "applies CDS and EDS updates from the Ruby control plane" do + skip "Requires Envoy (ENVOY_ADMIN_URI)" unless admin_uri + + control_plane = Async::GRPC::XDS::ControlPlane.new + control_plane.update_cluster( + cluster_name, + eds_config: Async::GRPC::XDS::ConfigSource.grpc("xds_cluster"), + protocol: :http1 + ) + control_plane.update_endpoints(cluster_name, [ + {addresses: [{address: "192.0.2.1", port: 8001}], healthy: true} + ]) + + server = Async::GRPC::XDS::Server.new( + control_plane, + services: [ + Async::GRPC::XDS::ClusterDiscoveryService, + Async::GRPC::XDS::EndpointDiscoveryService + ] + ) + endpoint = Async::HTTP::Endpoint.parse( + bind_uri, + protocol: Async::HTTP::Protocol::HTTP2 + ) + server_task = Async{server.run(endpoint)} + readiness_endpoint = Async::HTTP::Endpoint.parse(readiness_uri) + readiness_server = Async::HTTP::Server.new( + ->(_request){Protocol::HTTP::Response[200, {}, []]}, + readiness_endpoint + ) + readiness_task = Async{readiness_server.run} + + status = eventually do + status = cluster_status + status if status && addresses(status) == [["192.0.2.1", 8001]] + end + + expect(status["added_via_api"]).to be == true + + control_plane.update_endpoints(cluster_name, [ + {addresses: [{address: "192.0.2.2", port: 8002}], healthy: true} + ]) + + status = eventually do + status = cluster_status + status if status && addresses(status) == [["192.0.2.2", 8002]] + end + + expect(addresses(status)).to be == [["192.0.2.2", 8002]] + + stats = self.stats + expect(stats.fetch("cluster_manager.cds.update_success")).to be >= 1 + expect(stats.fetch("cluster_manager.cds.update_rejected")).to be == 0 + expect(stats.fetch("cluster.#{cluster_name}.update_success")).to be >= 2 + expect(stats.fetch("cluster.#{cluster_name}.update_rejected")).to be == 0 + ensure + internet.close + readiness_task&.stop + server_task&.stop + end +end diff --git a/readme.md b/readme.md index 0ab0738..fe15022 100644 --- a/readme.md +++ b/readme.md @@ -14,10 +14,11 @@ Please see the [project documentation](https://socketry.github.io/async-grpc-xds ## Testing -The `xds/` directory contains a Docker Compose integration environment with a Go xDS control plane and Ruby gRPC backends. +The `integration/` directory contains Docker Compose environments for testing the Ruby xDS client against a Go control plane and the Ruby control plane against Envoy. ``` bash -docker compose -f xds/docker-compose.yaml up --build --exit-code-from tests +cd integration/client +docker compose up --build --exit-code-from tests ``` ## Releases From 813d52faefa105d99007bd18e3d3a8798f903f4f Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Thu, 6 Aug 2026 23:43:45 +1200 Subject: [PATCH 10/12] Fix relocated fixture paths --- integration/client/backend_server.rb | 2 +- integration/client/test/async/grpc/xds/client.rb | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/integration/client/backend_server.rb b/integration/client/backend_server.rb index 70bf533..ba2df22 100644 --- a/integration/client/backend_server.rb +++ b/integration/client/backend_server.rb @@ -9,7 +9,7 @@ require "async/http/endpoint" require "async/grpc/dispatcher" require "async/grpc/service" -require_relative "../fixtures/async/grpc/test_interface" +require_relative "../../fixtures/async/grpc/test_interface" class TestBackendService < Async::GRPC::Service def initialize(interface_class, service_name, backend_id) diff --git a/integration/client/test/async/grpc/xds/client.rb b/integration/client/test/async/grpc/xds/client.rb index 5727806..0ef8414 100644 --- a/integration/client/test/async/grpc/xds/client.rb +++ b/integration/client/test/async/grpc/xds/client.rb @@ -8,7 +8,7 @@ require "async/grpc/service" require "sus/fixtures/async" require "async/http/endpoint" -require_relative "../../../../../fixtures/async/grpc/test_interface" +require_relative "../../../../../../fixtures/async/grpc/test_interface" require "json" require "net/http" require "set" From 6f15eade0e910408a67ca5293dc3546e81512df4 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 7 Aug 2026 00:03:10 +1200 Subject: [PATCH 11/12] Use bake-test-integration --- .github/workflows/test-xds.yaml | 16 ++++++---------- gems.rb | 1 + integration/client/readme.md | 5 ++--- integration/envoy/readme.md | 5 ++--- readme.md | 3 +-- 5 files changed, 12 insertions(+), 18 deletions(-) diff --git a/.github/workflows/test-xds.yaml b/.github/workflows/test-xds.yaml index 9a381f6..23dd513 100644 --- a/.github/workflows/test-xds.yaml +++ b/.github/workflows/test-xds.yaml @@ -28,17 +28,13 @@ jobs: steps: - uses: actions/checkout@v6 + - uses: ruby/setup-ruby@v1 + with: + ruby-version: "4.0" + bundler-cache: true - - name: Run tests + - name: Run integration tests timeout-minutes: 10 - working-directory: integration/client env: RUBY_VERSION: ${{matrix.ruby}} - run: docker compose up --build --exit-code-from tests - - - name: Test dedicated discovery services with Envoy - timeout-minutes: 10 - working-directory: integration/envoy - env: - RUBY_VERSION: ${{matrix.ruby}} - run: docker compose up --build --exit-code-from tests + run: bundle exec bake test:integration diff --git a/gems.rb b/gems.rb index ea04e96..65f592a 100644 --- a/gems.rb +++ b/gems.rb @@ -29,4 +29,5 @@ gem "bake-test" gem "bake-test-external" + gem "bake-test-integration" end diff --git a/integration/client/readme.md b/integration/client/readme.md index 85fef31..16cb277 100644 --- a/integration/client/readme.md +++ b/integration/client/readme.md @@ -11,11 +11,10 @@ The Docker Compose setup includes: ## Running Tests -From this directory: +From the project root: ```bash -cd integration/client -docker compose up --build --exit-code-from tests +bundle exec bake test:integration name=client ``` ## Test Structure diff --git a/integration/envoy/readme.md b/integration/envoy/readme.md index 7fa20d7..9293129 100644 --- a/integration/envoy/readme.md +++ b/integration/envoy/readme.md @@ -4,9 +4,8 @@ This integration test exercises the Ruby xDS control plane's dedicated CDS and E It publishes a cluster and endpoint, confirms Envoy accepted both, replaces the endpoint, and confirms Envoy applied the update without rejecting either resource. -Run it from this directory: +Run it from the project root: ```bash -cd integration/envoy -docker compose up --build --exit-code-from tests +bundle exec bake test:integration name=envoy ``` diff --git a/readme.md b/readme.md index fe15022..b84f0e0 100644 --- a/readme.md +++ b/readme.md @@ -17,8 +17,7 @@ Please see the [project documentation](https://socketry.github.io/async-grpc-xds The `integration/` directory contains Docker Compose environments for testing the Ruby xDS client against a Go control plane and the Ruby control plane against Envoy. ``` bash -cd integration/client -docker compose up --build --exit-code-from tests +bundle exec bake test:integration ``` ## Releases From d03fff1971a1fa095c1fcef28f5249276b05587e Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 7 Aug 2026 13:17:32 +1200 Subject: [PATCH 12/12] Rename integration test workflow --- .github/workflows/{test-xds.yaml => test-integration.yaml} | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) rename .github/workflows/{test-xds.yaml => test-integration.yaml} (96%) diff --git a/.github/workflows/test-xds.yaml b/.github/workflows/test-integration.yaml similarity index 96% rename from .github/workflows/test-xds.yaml rename to .github/workflows/test-integration.yaml index 23dd513..d8e7e0c 100644 --- a/.github/workflows/test-xds.yaml +++ b/.github/workflows/test-integration.yaml @@ -1,4 +1,4 @@ -name: Test xDS +name: Test Integration on: [push, pull_request]