diff --git a/.github/workflows/test-xds.yaml b/.github/workflows/test-integration.yaml similarity index 71% rename from .github/workflows/test-xds.yaml rename to .github/workflows/test-integration.yaml index 12fb10a..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] @@ -28,9 +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 env: RUBY_VERSION: ${{matrix.ruby}} - run: docker compose -f xds/docker-compose.yaml up --build --exit-code-from tests + run: bundle exec bake test:integration diff --git a/async-grpc-xds.gemspec b/async-grpc-xds.gemspec index c8552d0..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(["{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" 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/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/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/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 96% rename from xds/backend_server.rb rename to integration/client/backend_server.rb index 70bf533..ba2df22 100644 --- a/xds/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/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 92% rename from xds/readme.md rename to integration/client/readme.md index 093f204..16cb277 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). @@ -11,16 +11,15 @@ The Docker Compose setup includes: ## Running Tests -From this directory: +From the project root: ```bash -cd xds -docker compose up --build --exit-code-from tests +bundle exec bake test:integration name=client ``` ## 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 99% rename from xds/test/async/grpc/xds/client.rb rename to integration/client/test/async/grpc/xds/client.rb index 5727806..0ef8414 100644 --- a/xds/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" 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..9293129 --- /dev/null +++ b/integration/envoy/readme.md @@ -0,0 +1,11 @@ +# 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 the project root: + +```bash +bundle exec bake test:integration name=envoy +``` 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/lib/async/grpc/xds.rb b/lib/async/grpc/xds.rb index df3b6dc..ac6d80f 100644 --- a/lib/async/grpc/xds.rb +++ b/lib/async/grpc/xds.rb @@ -13,12 +13,17 @@ 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/stream" +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..b9c00a2 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 [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 [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..68934d4 --- /dev/null +++ b/lib/async/grpc/xds/discovery_service.rb @@ -0,0 +1,69 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "async" +require "async/grpc/service" +require "protocol/grpc/error" +require "protocol/grpc/status" + +require_relative "control_plane" +require_relative "stream" + +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 + + 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/lib/async/grpc/xds/stream.rb b/lib/async/grpc/xds/stream.rb new file mode 100644 index 0000000..5223ea0 --- /dev/null +++ b/lib/async/grpc/xds/stream.rb @@ -0,0 +1,102 @@ +# 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 = {} + @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 + + names = Set.new(request.resource_names) + if @subscriptions[type_url] != names + @subscriptions[type_url] = names + @versions.delete(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 + return unless @subscriptions.key?(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 + end + end + end +end diff --git a/readme.md b/readme.md index 66e0ada..b84f0e0 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,16 +10,14 @@ 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. -## 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. + - [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. ## 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 +bundle exec bake test:integration ``` ## Releases 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..38a2c7c --- /dev/null +++ b/test/async/grpc/xds/discovery_service.rb @@ -0,0 +1,142 @@ +# 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" + +describe Async::GRPC::XDS::DiscoveryService do + let(:control_plane) {Async::GRPC::XDS::ControlPlane.new} + + def request(type_url: nil, resource_names: ["myservice"]) + Envoy::Service::Discovery::V3::DiscoveryRequest.new( + type_url: type_url, + resource_names: resource_names + ) + 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(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 + 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 + 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 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