From 9e9823b5cc3fe8aaf449e67fe58fdafe35b948f4 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Thu, 23 Jul 2026 21:17:40 +1200 Subject: [PATCH 01/10] Support grouped Unix endpoint addresses --- lib/async/grpc/xds/resource_builder.rb | 86 ++++++++++++++++++++----- releases.md | 5 ++ test/async/grpc/xds/resource_builder.rb | 33 ++++++++++ 3 files changed, 109 insertions(+), 15 deletions(-) diff --git a/lib/async/grpc/xds/resource_builder.rb b/lib/async/grpc/xds/resource_builder.rb index c94f0c2..38105c3 100644 --- a/lib/async/grpc/xds/resource_builder.rb +++ b/lib/async/grpc/xds/resource_builder.rb @@ -31,8 +31,8 @@ def self.pack(resource) ) end - def self.cluster(name, service_name: name, load_balancer_policy: :round_robin, connect_timeout: 5) - Envoy::Config::Cluster::V3::Cluster.new( + def self.cluster(name, service_name: name, load_balancer_policy: :round_robin, 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( @@ -43,8 +43,18 @@ def self.cluster(name, service_name: name, load_balancer_policy: :round_robin, c ), connect_timeout: duration(connect_timeout), lb_policy: load_balancer_policy_value(load_balancer_policy), - http2_protocol_options: Envoy::Config::Core::V3::Http2ProtocolOptions.new - ) + } + + case protocol + when :http1, "http1", "http/1.1" + # Envoy uses HTTP/1 by default. + when :http2, "http2", "h2" + options[:http2_protocol_options] = Envoy::Config::Core::V3::Http2ProtocolOptions.new + else + raise ArgumentError, "Unsupported upstream protocol: #{protocol.inspect}" + end + + Envoy::Config::Cluster::V3::Cluster.new(**options) end def self.cluster_load_assignment(cluster_name, endpoints) @@ -60,16 +70,17 @@ def self.cluster_load_assignment(cluster_name, endpoints) def self.load_balancer_endpoint(endpoint) endpoint = normalize_endpoint(endpoint) + addresses = endpoint.fetch(:addresses) + address = addresses.first Envoy::Config::Endpoint::V3::LbEndpoint.new( endpoint: Envoy::Config::Endpoint::V3::Endpoint.new( - address: Envoy::Config::Core::V3::Address.new( - socket_address: Envoy::Config::Core::V3::SocketAddress.new( - protocol: Envoy::Config::Core::V3::SocketAddress::Protocol::TCP, - address: endpoint[:address], - port_value: endpoint[:port] + address: build_address(address), + additional_addresses: addresses.drop(1).map do |additional_address| + Envoy::Config::Endpoint::V3::Endpoint::AdditionalAddress.new( + address: build_address(additional_address) ) - ), + end, hostname: endpoint[:hostname].to_s ), health_status: health_status_value(endpoint.fetch(:healthy, true)) @@ -80,16 +91,20 @@ def self.normalize_endpoint(endpoint) case endpoint when Hash { - address: endpoint.fetch(:address){endpoint.fetch("address")}, - port: endpoint.fetch(:port){endpoint.fetch("port")}.to_i, + addresses: normalize_addresses(endpoint), hostname: endpoint[:hostname] || endpoint["hostname"], healthy: endpoint.key?(:healthy) ? endpoint[:healthy] : endpoint.fetch("healthy", true) } else - if endpoint.respond_to?(:address) && endpoint.respond_to?(:port) + if endpoint.respond_to?(:addresses) + { + addresses: endpoint.addresses.map{|address| normalize_address(address)}, + hostname: endpoint.respond_to?(:hostname) ? endpoint.hostname : nil, + healthy: endpoint.respond_to?(:healthy?) ? endpoint.healthy? : true + } + elsif endpoint.respond_to?(:address) && endpoint.respond_to?(:port) { - address: endpoint.address, - port: endpoint.port.to_i, + addresses: [{address: endpoint.address, port: endpoint.port.to_i}], hostname: endpoint.respond_to?(:hostname) ? endpoint.hostname : nil, healthy: endpoint.respond_to?(:healthy?) ? endpoint.healthy? : true } @@ -99,6 +114,47 @@ def self.normalize_endpoint(endpoint) end end + def self.normalize_addresses(endpoint) + addresses = if addresses = endpoint[:addresses] || endpoint["addresses"] + addresses.map{|address| normalize_address(address)} + else + [normalize_address(endpoint)] + end + + raise ArgumentError, "An endpoint requires at least one address!" if addresses.empty? + + addresses + end + + def self.normalize_address(address) + if path = address[:path] || address["path"] + {path: path} + else + { + address: address.fetch(:address){address.fetch("address")}, + port: address.fetch(:port){address.fetch("port")}.to_i, + } + end + end + + def self.build_address(address) + if path = address[:path] + Envoy::Config::Core::V3::Address.new( + pipe: Envoy::Config::Core::V3::Pipe.new(path: path) + ) + else + Envoy::Config::Core::V3::Address.new( + socket_address: Envoy::Config::Core::V3::SocketAddress.new( + protocol: Envoy::Config::Core::V3::SocketAddress::Protocol::TCP, + address: address[:address], + port_value: address[:port] + ) + ) + end + end + + private_class_method :normalize_addresses, :normalize_address, :build_address + def self.duration(seconds) whole_seconds = seconds.to_i nanos = ((seconds.to_f - whole_seconds) * 1_000_000_000).to_i diff --git a/releases.md b/releases.md index c6d7f77..ef60117 100644 --- a/releases.md +++ b/releases.md @@ -1,5 +1,10 @@ # Releases +## Unreleased + + - Support grouped IP and Unix-domain-socket addresses in EDS endpoints. + - Support selecting HTTP/1 or HTTP/2 for generated clusters. + ## v0.1.0 - Initial extraction from `async-grpc`. diff --git a/test/async/grpc/xds/resource_builder.rb b/test/async/grpc/xds/resource_builder.rb index c7b8f36..65bf082 100644 --- a/test/async/grpc/xds/resource_builder.rb +++ b/test/async/grpc/xds/resource_builder.rb @@ -19,6 +19,12 @@ expect(cluster.http2_protocol_options).not.to be == nil end + it "builds an HTTP/1 EDS cluster resource" do + cluster = subject.cluster("myservice", protocol: :http1) + + expect(cluster.http2_protocol_options).to be == nil + end + it "packs resources using their protobuf type URL" do cluster = subject.cluster("myservice") resource = subject.pack(cluster) @@ -68,6 +74,33 @@ def healthy? expect(load_balancer_endpoint.health_status).to be == :UNHEALTHY end + it "builds grouped IP and Unix endpoint addresses" do + load_balancer_endpoint = subject.load_balancer_endpoint({ + addresses: [ + {path: "/tmp/falcon.ipc"}, + {address: "127.0.0.1", port: 9292}, + ] + }) + endpoint = load_balancer_endpoint.endpoint + + expect(endpoint.address.pipe.path).to be == "/tmp/falcon.ipc" + expect(endpoint.additional_addresses.size).to be == 1 + expect(endpoint.additional_addresses.first.address.socket_address.address).to be == "127.0.0.1" + expect(endpoint.additional_addresses.first.address.socket_address.port_value).to be == 9292 + end + + it "rejects unsupported upstream protocols" do + expect do + subject.cluster("myservice", protocol: :http3) + end.to raise_exception(ArgumentError) + end + + it "rejects endpoints without addresses" do + expect do + subject.load_balancer_endpoint(addresses: []) + end.to raise_exception(ArgumentError) + end + it "rejects invalid endpoint data" do expect do subject.load_balancer_endpoint(Object.new) From 9cbada4001e8c338450f76cc5af949c14eb81d90 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 24 Jul 2026 12:23:41 +1200 Subject: [PATCH 02/10] Fix RuboCop blank line indentation. --- lib/async/grpc/xds/resource_builder.rb | 27 ++++---- test/async/grpc/xds/resources.rb | 85 +++++++++++++------------- 2 files changed, 55 insertions(+), 57 deletions(-) diff --git a/lib/async/grpc/xds/resource_builder.rb b/lib/async/grpc/xds/resource_builder.rb index 38105c3..bd2ef11 100644 --- a/lib/async/grpc/xds/resource_builder.rb +++ b/lib/async/grpc/xds/resource_builder.rb @@ -67,12 +67,12 @@ def self.cluster_load_assignment(cluster_name, endpoints) ] ) end - + def self.load_balancer_endpoint(endpoint) endpoint = normalize_endpoint(endpoint) addresses = endpoint.fetch(:addresses) address = addresses.first - + Envoy::Config::Endpoint::V3::LbEndpoint.new( endpoint: Envoy::Config::Endpoint::V3::Endpoint.new( address: build_address(address), @@ -86,7 +86,7 @@ def self.load_balancer_endpoint(endpoint) health_status: health_status_value(endpoint.fetch(:healthy, true)) ) end - + def self.normalize_endpoint(endpoint) case endpoint when Hash @@ -113,19 +113,19 @@ def self.normalize_endpoint(endpoint) end end end - + def self.normalize_addresses(endpoint) addresses = if addresses = endpoint[:addresses] || endpoint["addresses"] addresses.map{|address| normalize_address(address)} else [normalize_address(endpoint)] end - + raise ArgumentError, "An endpoint requires at least one address!" if addresses.empty? - + addresses end - + def self.normalize_address(address) if path = address[:path] || address["path"] {path: path} @@ -136,7 +136,7 @@ def self.normalize_address(address) } end end - + def self.build_address(address) if path = address[:path] Envoy::Config::Core::V3::Address.new( @@ -152,16 +152,16 @@ def self.build_address(address) ) end end - + private_class_method :normalize_addresses, :normalize_address, :build_address - + def self.duration(seconds) whole_seconds = seconds.to_i nanos = ((seconds.to_f - whole_seconds) * 1_000_000_000).to_i - + Google::Protobuf::Duration.new(seconds: whole_seconds, nanos: nanos) end - + def self.load_balancer_policy_value(load_balancer_policy) case load_balancer_policy when :round_robin, :ROUND_ROBIN, "round_robin", "ROUND_ROBIN" @@ -174,7 +174,7 @@ def self.load_balancer_policy_value(load_balancer_policy) load_balancer_policy end end - + def self.health_status_value(healthy) case healthy when :healthy, :HEALTHY, "healthy", "HEALTHY", true @@ -191,4 +191,3 @@ def self.health_status_value(healthy) end end end - \ No newline at end of file diff --git a/test/async/grpc/xds/resources.rb b/test/async/grpc/xds/resources.rb index f526266..09e13c0 100644 --- a/test/async/grpc/xds/resources.rb +++ b/test/async/grpc/xds/resources.rb @@ -16,7 +16,7 @@ {type: :HTTP, interval: 5, timeout: 1, http_health_check: {path: "/ready"}} ] ) - + expect(cluster.name).to be == "myservice" expect(cluster.type).to be == :EDS expect(cluster.load_balancer_policy).to be == :LEAST_REQUEST @@ -25,19 +25,19 @@ {type: :HTTP, timeout: 1, interval: 5, path: "/ready"} ] end - + it "parses cluster type and load balancer policy variants" do expect(subject.new(name: "static", type: :STATIC).type).to be == :STATIC expect(subject.new(name: "logical", type: :LOGICAL_DNS).type).to be == :LOGICAL_DNS expect(subject.new(name: "strict", type: :STRICT_DNS).type).to be == :STRICT_DNS expect(subject.new(name: "unknown", type: :UNKNOWN).type).to be == :EDS - + expect(subject.new(name: "round-robin", lb_policy: :ROUND_ROBIN).load_balancer_policy).to be == :ROUND_ROBIN expect(subject.new(name: "ring-hash", lb_policy: :RING_HASH).load_balancer_policy).to be == :RING_HASH expect(subject.new(name: "maglev", lb_policy: :MAGLEV).load_balancer_policy).to be == :MAGLEV expect(subject.new(name: "unknown-policy", lb_policy: :UNKNOWN).load_balancer_policy).to be == :ROUND_ROBIN end - + it "parses hash health check variants" do cluster = subject.new( name: "myservice", @@ -48,29 +48,29 @@ {health_checker: {type: :unknown}} ] ) - + expect(cluster.health_checks[0][:type]).to be == :HTTP expect(cluster.health_checks[1]).to be == {type: :gRPC, timeout: nil, interval: 5.5, path: "/health"} expect(cluster.health_checks[2][:type]).to be == :TCP expect(cluster.health_checks[3][:type]).to be == :HTTP end - + it "parses legacy hash load balancer policy names" do cluster = subject.new(name: "myservice", type: :EDS, lb_policy: :RANDOM) - + expect(cluster.load_balancer_policy).to be == :RANDOM end - + it "parses protobuf cluster data" do protobuf = Async::GRPC::XDS::ResourceBuilder.cluster("myservice", load_balancer_policy: :least_request) cluster = subject.from_proto(protobuf) - + expect(cluster.name).to be == "myservice" expect(cluster.type).to be == :EDS expect(cluster.load_balancer_policy).to be == :LEAST_REQUEST expect(cluster).to be(:eds_cluster?) end - + it "parses protobuf health checks" do protobuf = Async::GRPC::XDS::ResourceBuilder.cluster("myservice") protobuf.health_checks << Envoy::Config::Core::V3::HealthCheck.new( @@ -78,14 +78,14 @@ interval: Google::Protobuf::Duration.new(seconds: 10), http_health_check: Envoy::Config::Core::V3::HealthCheck::HttpHealthCheck.new(path: "/healthz") ) - + cluster = subject.from_proto(protobuf) - + expect(cluster.health_checks).to be == [ {type: :HTTP, timeout: 1.5, interval: 10.0, path: "/healthz"} ] end - + it "parses protobuf gRPC and TCP health checks" do protobuf = Async::GRPC::XDS::ResourceBuilder.cluster("myservice") protobuf.health_checks << Envoy::Config::Core::V3::HealthCheck.new( @@ -94,48 +94,48 @@ protobuf.health_checks << Envoy::Config::Core::V3::HealthCheck.new( tcp_health_check: Envoy::Config::Core::V3::HealthCheck::TcpHealthCheck.new ) - + cluster = subject.from_proto(protobuf) - + expect(cluster.health_checks.map{|health_check| health_check[:type]}).to be == [:gRPC, :TCP] end - + it "parses object-style health check oneof values" do health_checker = Object.new health_checker.define_singleton_method(:http_health_check) do Struct.new(:path).new("/object") end - + check = Struct.new(:health_checker, :timeout, :interval) do def http_health_check Struct.new(:path).new("/object") end end.new(health_checker, 1, 2) - + cluster = subject.new(name: "myservice", health_checks: [check]) - + expect(cluster.health_checks).to be == [ {type: :HTTP, timeout: 1, interval: 2, path: "/object"} ] end - + it "parses object-style gRPC and TCP health check oneof values" do grpc_health_checker = Object.new grpc_health_checker.define_singleton_method(:grpc_health_check){Object.new} tcp_health_checker = Object.new tcp_health_checker.define_singleton_method(:tcp_health_check){Object.new} - + grpc_check = Struct.new(:health_checker, :timeout, :interval) do end.new(grpc_health_checker, nil, nil) tcp_check = Struct.new(:health_checker, :timeout, :interval) do end.new(tcp_health_checker, nil, nil) - + cluster = subject.new(name: "myservice", health_checks: [grpc_check, tcp_check]) - + expect(cluster.health_checks.map{|health_check| health_check[:type]}).to be == [:gRPC, :TCP] end end - + describe Async::GRPC::XDS::Resources::ClusterLoadAssignment do it "parses hash endpoint assignments" do assignment = subject.new( @@ -158,17 +158,17 @@ def http_health_check } ] ) - + expect(assignment.cluster_name).to be == "myservice" expect(assignment.endpoints.size).to be == 1 - + endpoint = assignment.endpoints.first expect(endpoint.address).to be == "127.0.0.1" expect(endpoint.port).to be == 50051 expect(endpoint).to be(:healthy?) expect(endpoint.uri).to be == "http://127.0.0.1:50051" end - + it "parses protobuf endpoint assignments" do protobuf = Async::GRPC::XDS::ResourceBuilder.cluster_load_assignment( "myservice", @@ -178,14 +178,14 @@ def http_health_check ] ) assignment = subject.from_proto(protobuf) - + expect(assignment.cluster_name).to be == "myservice" expect(assignment.endpoints.map(&:address)).to be == ["127.0.0.1", "127.0.0.2"] expect(assignment.endpoints.map(&:port)).to be == [50051, 50052] expect(assignment.endpoints.map(&:healthy?)).to be == [true, false] end end - + describe Async::GRPC::XDS::Resources::Endpoint do it "treats unknown endpoints as healthy" do endpoint = subject.new( @@ -198,15 +198,15 @@ def http_health_check } } ) - + expect(endpoint.health_status).to be == :UNKNOWN expect(endpoint).to be(:healthy?) end - + it "uses the configured endpoint scheme" do previous = ENV["XDS_ENDPOINT_SCHEME"] ENV["XDS_ENDPOINT_SCHEME"] = "https" - + endpoint = subject.new( endpoint: { address: { @@ -217,12 +217,12 @@ def http_health_check } } ) - + expect(endpoint.uri).to be == "https://example.com:443" ensure ENV["XDS_ENDPOINT_SCHEME"] = previous end - + it "parses degraded endpoints as unhealthy" do endpoint = subject.new( endpoint: { @@ -235,11 +235,11 @@ def http_health_check }, health_status: :DEGRADED ) - + expect(endpoint.health_status).to be == :DEGRADED expect(endpoint).not.to be(:healthy?) end - + it "falls back to unknown for unsupported health status values" do endpoint = subject.new( endpoint: { @@ -252,13 +252,12 @@ def http_health_check }, health_status: :DRAINING ) - + expect(endpoint.health_status).to be == :UNKNOWN end end - - - - - - \ No newline at end of file + + + + + From edd0c28676134c888e18a9baee531f6073f8c0e1 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 24 Jul 2026 12:28:27 +1200 Subject: [PATCH 03/10] Use explicit hash endpoint coercion. --- lib/async/grpc/xds/resource_builder.rb | 32 +++++++------------------ test/async/grpc/xds/resource_builder.rb | 19 ++++++++++----- 2 files changed, 21 insertions(+), 30 deletions(-) diff --git a/lib/async/grpc/xds/resource_builder.rb b/lib/async/grpc/xds/resource_builder.rb index bd2ef11..8610a7d 100644 --- a/lib/async/grpc/xds/resource_builder.rb +++ b/lib/async/grpc/xds/resource_builder.rb @@ -88,30 +88,14 @@ def self.load_balancer_endpoint(endpoint) end def self.normalize_endpoint(endpoint) - case endpoint - when Hash - { - addresses: normalize_addresses(endpoint), - hostname: endpoint[:hostname] || endpoint["hostname"], - healthy: endpoint.key?(:healthy) ? endpoint[:healthy] : endpoint.fetch("healthy", true) - } - else - if endpoint.respond_to?(:addresses) - { - addresses: endpoint.addresses.map{|address| normalize_address(address)}, - hostname: endpoint.respond_to?(:hostname) ? endpoint.hostname : nil, - healthy: endpoint.respond_to?(:healthy?) ? endpoint.healthy? : true - } - elsif endpoint.respond_to?(:address) && endpoint.respond_to?(:port) - { - addresses: [{address: endpoint.address, port: endpoint.port.to_i}], - hostname: endpoint.respond_to?(:hostname) ? endpoint.hostname : nil, - healthy: endpoint.respond_to?(:healthy?) ? endpoint.healthy? : true - } - else - raise ArgumentError, "Invalid endpoint: #{endpoint.inspect}" - end - end + value = Hash.try_convert(endpoint) + raise ArgumentError, "Invalid endpoint: #{endpoint.inspect}" unless value + + { + addresses: normalize_addresses(value), + hostname: value[:hostname] || value["hostname"], + healthy: value.key?(:healthy) ? value[:healthy] : value.fetch("healthy", true) + } end def self.normalize_addresses(endpoint) diff --git a/test/async/grpc/xds/resource_builder.rb b/test/async/grpc/xds/resource_builder.rb index 65bf082..9e035f3 100644 --- a/test/async/grpc/xds/resource_builder.rb +++ b/test/async/grpc/xds/resource_builder.rb @@ -59,17 +59,24 @@ expect(second.health_status).to be == :UNHEALTHY end - it "builds endpoint assignments from endpoint-like objects" do - endpoint = Struct.new(:address, :port, :hostname) do - def healthy? - false - end - end.new("127.0.0.3", "50053", "three") + it "builds endpoint assignments from hash-like objects" do + endpoint = Object.new + def endpoint.to_hash + { + addresses: [ + {address: "127.0.0.3", port: "50053"}, + {path: "/tmp/falcon.ipc"}, + ], + hostname: "three", + healthy: false, + } + end load_balancer_endpoint = subject.load_balancer_endpoint(endpoint) expect(load_balancer_endpoint.endpoint.address.socket_address.address).to be == "127.0.0.3" expect(load_balancer_endpoint.endpoint.address.socket_address.port_value).to be == 50053 + expect(load_balancer_endpoint.endpoint.additional_addresses.first.address.pipe.path).to be == "/tmp/falcon.ipc" expect(load_balancer_endpoint.endpoint.hostname).to be == "three" expect(load_balancer_endpoint.health_status).to be == :UNHEALTHY end From 1e8cf02cfbd17e5b93284f5cee5b76a4e9f7b7b4 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 24 Jul 2026 12:31:37 +1200 Subject: [PATCH 04/10] Narrow endpoint input to worker state shape. --- lib/async/grpc/xds/resource_builder.rb | 41 ++++++------------------- test/async/grpc/xds/control_plane.rb | 8 ++--- test/async/grpc/xds/resource_builder.rb | 26 ++-------------- test/async/grpc/xds/resources.rb | 5 ++- 4 files changed, 18 insertions(+), 62 deletions(-) diff --git a/lib/async/grpc/xds/resource_builder.rb b/lib/async/grpc/xds/resource_builder.rb index 8610a7d..d040e69 100644 --- a/lib/async/grpc/xds/resource_builder.rb +++ b/lib/async/grpc/xds/resource_builder.rb @@ -88,37 +88,16 @@ def self.load_balancer_endpoint(endpoint) end def self.normalize_endpoint(endpoint) - value = Hash.try_convert(endpoint) - raise ArgumentError, "Invalid endpoint: #{endpoint.inspect}" unless value - - { - addresses: normalize_addresses(value), - hostname: value[:hostname] || value["hostname"], - healthy: value.key?(:healthy) ? value[:healthy] : value.fetch("healthy", true) - } - end - - def self.normalize_addresses(endpoint) - addresses = if addresses = endpoint[:addresses] || endpoint["addresses"] - addresses.map{|address| normalize_address(address)} - else - [normalize_address(endpoint)] - end + raise ArgumentError, "Invalid endpoint: #{endpoint.inspect}" unless endpoint.is_a?(Hash) + addresses = endpoint.fetch(:addresses) raise ArgumentError, "An endpoint requires at least one address!" if addresses.empty? - addresses - end - - def self.normalize_address(address) - if path = address[:path] || address["path"] - {path: path} - else - { - address: address.fetch(:address){address.fetch("address")}, - port: address.fetch(:port){address.fetch("port")}.to_i, - } - end + { + addresses: addresses, + hostname: endpoint[:hostname], + healthy: endpoint.fetch(:healthy, true) + } end def self.build_address(address) @@ -130,14 +109,14 @@ def self.build_address(address) Envoy::Config::Core::V3::Address.new( socket_address: Envoy::Config::Core::V3::SocketAddress.new( protocol: Envoy::Config::Core::V3::SocketAddress::Protocol::TCP, - address: address[:address], - port_value: address[:port] + address: address.fetch(:address), + port_value: address.fetch(:port) ) ) end end - private_class_method :normalize_addresses, :normalize_address, :build_address + private_class_method :build_address def self.duration(seconds) whole_seconds = seconds.to_i diff --git a/test/async/grpc/xds/control_plane.rb b/test/async/grpc/xds/control_plane.rb index 99186b0..8dfec01 100644 --- a/test/async/grpc/xds/control_plane.rb +++ b/test/async/grpc/xds/control_plane.rb @@ -23,7 +23,7 @@ end it "publishes endpoint resources" do - control_plane.update_endpoints("myservice", [{address: "127.0.0.1", port: 50051}]) + control_plane.update_endpoints("myservice", [{addresses: [{address: "127.0.0.1", port: 50051}]}]) response = control_plane.response( Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, @@ -58,8 +58,8 @@ end it "increments versions when resources change" do - control_plane.update_endpoints("myservice", [{address: "127.0.0.1", port: 50051}]) - control_plane.update_endpoints("myservice", [{address: "127.0.0.2", port: 50052}]) + control_plane.update_endpoints("myservice", [{addresses: [{address: "127.0.0.1", port: 50051}]}]) + control_plane.update_endpoints("myservice", [{addresses: [{address: "127.0.0.2", port: 50052}]}]) expect(control_plane.version(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE)).to be == "2" end @@ -79,7 +79,7 @@ end it "removes endpoint resources" do - control_plane.update_endpoints("myservice", [{address: "127.0.0.1", port: 50051}]) + control_plane.update_endpoints("myservice", [{addresses: [{address: "127.0.0.1", port: 50051}]}]) control_plane.remove_endpoints("myservice") expect(control_plane.resources(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE)).to be(:empty?) diff --git a/test/async/grpc/xds/resource_builder.rb b/test/async/grpc/xds/resource_builder.rb index 9e035f3..904e011 100644 --- a/test/async/grpc/xds/resource_builder.rb +++ b/test/async/grpc/xds/resource_builder.rb @@ -37,8 +37,8 @@ assignment = subject.cluster_load_assignment( "myservice", [ - {address: "127.0.0.1", port: 50051, hostname: "one", healthy: true}, - {"address" => "127.0.0.2", "port" => "50052", "healthy" => false} + {addresses: [{address: "127.0.0.1", port: 50051}], hostname: "one", healthy: true}, + {addresses: [{address: "127.0.0.2", port: 50052}], healthy: false} ] ) @@ -59,28 +59,6 @@ expect(second.health_status).to be == :UNHEALTHY end - it "builds endpoint assignments from hash-like objects" do - endpoint = Object.new - def endpoint.to_hash - { - addresses: [ - {address: "127.0.0.3", port: "50053"}, - {path: "/tmp/falcon.ipc"}, - ], - hostname: "three", - healthy: false, - } - end - - load_balancer_endpoint = subject.load_balancer_endpoint(endpoint) - - expect(load_balancer_endpoint.endpoint.address.socket_address.address).to be == "127.0.0.3" - expect(load_balancer_endpoint.endpoint.address.socket_address.port_value).to be == 50053 - expect(load_balancer_endpoint.endpoint.additional_addresses.first.address.pipe.path).to be == "/tmp/falcon.ipc" - expect(load_balancer_endpoint.endpoint.hostname).to be == "three" - expect(load_balancer_endpoint.health_status).to be == :UNHEALTHY - end - it "builds grouped IP and Unix endpoint addresses" do load_balancer_endpoint = subject.load_balancer_endpoint({ addresses: [ diff --git a/test/async/grpc/xds/resources.rb b/test/async/grpc/xds/resources.rb index 09e13c0..c9192a0 100644 --- a/test/async/grpc/xds/resources.rb +++ b/test/async/grpc/xds/resources.rb @@ -173,8 +173,8 @@ def http_health_check protobuf = Async::GRPC::XDS::ResourceBuilder.cluster_load_assignment( "myservice", [ - {address: "127.0.0.1", port: 50051}, - {address: "127.0.0.2", port: 50052, healthy: false} + {addresses: [{address: "127.0.0.1", port: 50051}]}, + {addresses: [{address: "127.0.0.2", port: 50052}], healthy: false} ] ) assignment = subject.from_proto(protobuf) @@ -260,4 +260,3 @@ def http_health_check - From e1ed96e08a49a6ae7ebbe72fff2ac6bf4c8fa3b9 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 24 Jul 2026 12:35:27 +1200 Subject: [PATCH 05/10] Accept canonical endpoint metadata only. --- lib/async/grpc/xds/resource_builder.rb | 8 +++----- test/async/grpc/xds/resource_builder.rb | 3 +-- 2 files changed, 4 insertions(+), 7 deletions(-) diff --git a/lib/async/grpc/xds/resource_builder.rb b/lib/async/grpc/xds/resource_builder.rb index d040e69..dd6b4a0 100644 --- a/lib/async/grpc/xds/resource_builder.rb +++ b/lib/async/grpc/xds/resource_builder.rb @@ -46,9 +46,9 @@ def self.cluster(name, service_name: name, load_balancer_policy: :round_robin, c } case protocol - when :http1, "http1", "http/1.1" + when :http1 # Envoy uses HTTP/1 by default. - when :http2, "http2", "h2" + when :http2 options[:http2_protocol_options] = Envoy::Config::Core::V3::Http2ProtocolOptions.new else raise ArgumentError, "Unsupported upstream protocol: #{protocol.inspect}" @@ -80,8 +80,7 @@ def self.load_balancer_endpoint(endpoint) Envoy::Config::Endpoint::V3::Endpoint::AdditionalAddress.new( address: build_address(additional_address) ) - end, - hostname: endpoint[:hostname].to_s + end ), health_status: health_status_value(endpoint.fetch(:healthy, true)) ) @@ -95,7 +94,6 @@ def self.normalize_endpoint(endpoint) { addresses: addresses, - hostname: endpoint[:hostname], healthy: endpoint.fetch(:healthy, true) } end diff --git a/test/async/grpc/xds/resource_builder.rb b/test/async/grpc/xds/resource_builder.rb index 904e011..018d689 100644 --- a/test/async/grpc/xds/resource_builder.rb +++ b/test/async/grpc/xds/resource_builder.rb @@ -37,7 +37,7 @@ assignment = subject.cluster_load_assignment( "myservice", [ - {addresses: [{address: "127.0.0.1", port: 50051}], hostname: "one", healthy: true}, + {addresses: [{address: "127.0.0.1", port: 50051}], healthy: true}, {addresses: [{address: "127.0.0.2", port: 50052}], healthy: false} ] ) @@ -50,7 +50,6 @@ first = endpoints.first expect(first.endpoint.address.socket_address.address).to be == "127.0.0.1" expect(first.endpoint.address.socket_address.port_value).to be == 50051 - expect(first.endpoint.hostname).to be == "one" expect(first.health_status).to be == :HEALTHY second = endpoints.last From e77d53fe8575533fc473051344fd5bfdcf25d627 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 24 Jul 2026 12:39:32 +1200 Subject: [PATCH 06/10] Require normalized endpoint state. --- lib/async/grpc/xds/resource_builder.rb | 21 ++++++--------------- test/async/grpc/xds/control_plane.rb | 8 ++++---- test/async/grpc/xds/resource_builder.rb | 5 +++-- test/async/grpc/xds/resources.rb | 3 +-- 4 files changed, 14 insertions(+), 23 deletions(-) diff --git a/lib/async/grpc/xds/resource_builder.rb b/lib/async/grpc/xds/resource_builder.rb index dd6b4a0..ad4eecf 100644 --- a/lib/async/grpc/xds/resource_builder.rb +++ b/lib/async/grpc/xds/resource_builder.rb @@ -69,8 +69,11 @@ def self.cluster_load_assignment(cluster_name, endpoints) end def self.load_balancer_endpoint(endpoint) - endpoint = normalize_endpoint(endpoint) - addresses = endpoint.fetch(:addresses) + raise ArgumentError, "Invalid endpoint: #{endpoint.inspect}" unless endpoint.is_a?(Hash) + + addresses, healthy = endpoint.fetch_values(:addresses, :healthy) + raise ArgumentError, "An endpoint requires at least one address!" if addresses.empty? + address = addresses.first Envoy::Config::Endpoint::V3::LbEndpoint.new( @@ -82,22 +85,10 @@ def self.load_balancer_endpoint(endpoint) ) end ), - health_status: health_status_value(endpoint.fetch(:healthy, true)) + health_status: health_status_value(healthy) ) end - def self.normalize_endpoint(endpoint) - raise ArgumentError, "Invalid endpoint: #{endpoint.inspect}" unless endpoint.is_a?(Hash) - - addresses = endpoint.fetch(:addresses) - raise ArgumentError, "An endpoint requires at least one address!" if addresses.empty? - - { - addresses: addresses, - healthy: endpoint.fetch(:healthy, true) - } - end - def self.build_address(address) if path = address[:path] Envoy::Config::Core::V3::Address.new( diff --git a/test/async/grpc/xds/control_plane.rb b/test/async/grpc/xds/control_plane.rb index 8dfec01..f599f39 100644 --- a/test/async/grpc/xds/control_plane.rb +++ b/test/async/grpc/xds/control_plane.rb @@ -23,7 +23,7 @@ end it "publishes endpoint resources" do - control_plane.update_endpoints("myservice", [{addresses: [{address: "127.0.0.1", port: 50051}]}]) + control_plane.update_endpoints("myservice", [{addresses: [{address: "127.0.0.1", port: 50051}], healthy: true}]) response = control_plane.response( Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE, @@ -58,8 +58,8 @@ end it "increments versions when resources change" do - control_plane.update_endpoints("myservice", [{addresses: [{address: "127.0.0.1", port: 50051}]}]) - control_plane.update_endpoints("myservice", [{addresses: [{address: "127.0.0.2", port: 50052}]}]) + control_plane.update_endpoints("myservice", [{addresses: [{address: "127.0.0.1", port: 50051}], healthy: true}]) + control_plane.update_endpoints("myservice", [{addresses: [{address: "127.0.0.2", port: 50052}], healthy: true}]) expect(control_plane.version(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE)).to be == "2" end @@ -79,7 +79,7 @@ end it "removes endpoint resources" do - control_plane.update_endpoints("myservice", [{addresses: [{address: "127.0.0.1", port: 50051}]}]) + control_plane.update_endpoints("myservice", [{addresses: [{address: "127.0.0.1", port: 50051}], healthy: true}]) control_plane.remove_endpoints("myservice") expect(control_plane.resources(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE)).to be(:empty?) diff --git a/test/async/grpc/xds/resource_builder.rb b/test/async/grpc/xds/resource_builder.rb index 018d689..a547f79 100644 --- a/test/async/grpc/xds/resource_builder.rb +++ b/test/async/grpc/xds/resource_builder.rb @@ -63,7 +63,8 @@ addresses: [ {path: "/tmp/falcon.ipc"}, {address: "127.0.0.1", port: 9292}, - ] + ], + healthy: true, }) endpoint = load_balancer_endpoint.endpoint @@ -81,7 +82,7 @@ it "rejects endpoints without addresses" do expect do - subject.load_balancer_endpoint(addresses: []) + subject.load_balancer_endpoint(addresses: [], healthy: true) end.to raise_exception(ArgumentError) end diff --git a/test/async/grpc/xds/resources.rb b/test/async/grpc/xds/resources.rb index c9192a0..4b4ec2c 100644 --- a/test/async/grpc/xds/resources.rb +++ b/test/async/grpc/xds/resources.rb @@ -173,7 +173,7 @@ def http_health_check protobuf = Async::GRPC::XDS::ResourceBuilder.cluster_load_assignment( "myservice", [ - {addresses: [{address: "127.0.0.1", port: 50051}]}, + {addresses: [{address: "127.0.0.1", port: 50051}], healthy: true}, {addresses: [{address: "127.0.0.2", port: 50052}], healthy: false} ] ) @@ -259,4 +259,3 @@ def http_health_check - From c703e30389ef03251818d4eb614b0e42074a54ca Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 24 Jul 2026 12:43:26 +1200 Subject: [PATCH 07/10] Simplify endpoint address handling. --- lib/async/grpc/xds/resource_builder.rb | 6 ++---- test/async/grpc/xds/resource_builder.rb | 6 ------ 2 files changed, 2 insertions(+), 10 deletions(-) diff --git a/lib/async/grpc/xds/resource_builder.rb b/lib/async/grpc/xds/resource_builder.rb index ad4eecf..936eb14 100644 --- a/lib/async/grpc/xds/resource_builder.rb +++ b/lib/async/grpc/xds/resource_builder.rb @@ -69,17 +69,15 @@ def self.cluster_load_assignment(cluster_name, endpoints) end def self.load_balancer_endpoint(endpoint) - raise ArgumentError, "Invalid endpoint: #{endpoint.inspect}" unless endpoint.is_a?(Hash) - addresses, healthy = endpoint.fetch_values(:addresses, :healthy) raise ArgumentError, "An endpoint requires at least one address!" if addresses.empty? - address = addresses.first + address, *additional_addresses = addresses Envoy::Config::Endpoint::V3::LbEndpoint.new( endpoint: Envoy::Config::Endpoint::V3::Endpoint.new( address: build_address(address), - additional_addresses: addresses.drop(1).map do |additional_address| + additional_addresses: additional_addresses.map do |additional_address| Envoy::Config::Endpoint::V3::Endpoint::AdditionalAddress.new( address: build_address(additional_address) ) diff --git a/test/async/grpc/xds/resource_builder.rb b/test/async/grpc/xds/resource_builder.rb index a547f79..fb3f532 100644 --- a/test/async/grpc/xds/resource_builder.rb +++ b/test/async/grpc/xds/resource_builder.rb @@ -86,12 +86,6 @@ end.to raise_exception(ArgumentError) end - it "rejects invalid endpoint data" do - expect do - subject.load_balancer_endpoint(Object.new) - end.to raise_exception(ArgumentError) - end - it "maps load balancer policies" do expect(subject.load_balancer_policy_value(:round_robin)).to be == Envoy::Config::Cluster::V3::Cluster::LbPolicy::ROUND_ROBIN expect(subject.load_balancer_policy_value("LEAST_REQUEST")).to be == Envoy::Config::Cluster::V3::Cluster::LbPolicy::LEAST_REQUEST From 8a0e1585dd18ab03d6d1ecdc0f06ebcea0b8c344 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 24 Jul 2026 12:55:34 +1200 Subject: [PATCH 08/10] Update xDS endpoint fixtures. --- xds/test/async/grpc/xds/control_plane.rb | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/xds/test/async/grpc/xds/control_plane.rb b/xds/test/async/grpc/xds/control_plane.rb index 04e6370..43cbeb0 100644 --- a/xds/test/async/grpc/xds/control_plane.rb +++ b/xds/test/async/grpc/xds/control_plane.rb @@ -33,8 +33,8 @@ control_plane.update_endpoints( "myservice", [ - {address: "127.0.0.1", port: 50051}, - {address: "127.0.0.2", port: 50052, healthy: false} + {addresses: [{address: "127.0.0.1", port: 50051}], healthy: true}, + {addresses: [{address: "127.0.0.2", port: 50052}], healthy: false} ] ) @@ -49,15 +49,15 @@ end it "increments resource versions" do - control_plane.update_endpoints("myservice", [{address: "127.0.0.1", port: 50051}]) - control_plane.update_endpoints("myservice", [{address: "127.0.0.2", port: 50052}]) + control_plane.update_endpoints("myservice", [{addresses: [{address: "127.0.0.1", port: 50051}], healthy: true}]) + control_plane.update_endpoints("myservice", [{addresses: [{address: "127.0.0.2", port: 50052}], healthy: true}]) expect(control_plane.version(Async::GRPC::XDS::ControlPlane::ENDPOINT_TYPE)).to be == "2" end it "serves resources over ADS" do control_plane.update_cluster("myservice") - control_plane.update_endpoints("myservice", [{address: "127.0.0.1", port: 50051}]) + control_plane.update_endpoints("myservice", [{addresses: [{address: "127.0.0.1", port: 50051}], healthy: true}]) port = available_port endpoint = Async::HTTP::Endpoint.parse( From 22136b5176b48276bd48f785072034a3ebc11a00 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 24 Jul 2026 12:57:12 +1200 Subject: [PATCH 09/10] Document xDS resource builder changes. --- lib/async/grpc/xds/resource_builder.rb | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/lib/async/grpc/xds/resource_builder.rb b/lib/async/grpc/xds/resource_builder.rb index 936eb14..5b48fb9 100644 --- a/lib/async/grpc/xds/resource_builder.rb +++ b/lib/async/grpc/xds/resource_builder.rb @@ -31,6 +31,14 @@ def self.pack(resource) ) end + # Build an EDS cluster resource. + # @parameter name [String] The cluster name. + # @parameter service_name [String] The EDS service name. + # @parameter load_balancer_policy [Symbol] The Envoy load-balancing policy. + # @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 self.cluster(name, service_name: name, load_balancer_policy: :round_robin, connect_timeout: 5, protocol: :http2) options = { name: name.to_s, @@ -57,6 +65,10 @@ def self.cluster(name, service_name: name, load_balancer_policy: :round_robin, c Envoy::Config::Cluster::V3::Cluster.new(**options) end + # Build an EDS cluster load assignment from normalized endpoint state. + # @parameter cluster_name [String] The cluster name. + # @parameter endpoints [Array(Hash)] The endpoints, each containing `:addresses` and `:healthy`. + # @returns [Envoy::Config::Endpoint::V3::ClusterLoadAssignment] The generated load assignment. def self.cluster_load_assignment(cluster_name, endpoints) Envoy::Config::Endpoint::V3::ClusterLoadAssignment.new( cluster_name: cluster_name.to_s, @@ -68,6 +80,11 @@ def self.cluster_load_assignment(cluster_name, endpoints) ) end + # Build an Envoy load-balancer endpoint from normalized endpoint state. + # @parameter endpoint [Hash] The endpoint containing `:addresses` and `:healthy`. + # @returns [Envoy::Config::Endpoint::V3::LbEndpoint] The generated load-balancer endpoint. + # @raises [KeyError] If required endpoint state is missing. + # @raises [ArgumentError] If the endpoint has no addresses. def self.load_balancer_endpoint(endpoint) addresses, healthy = endpoint.fetch_values(:addresses, :healthy) raise ArgumentError, "An endpoint requires at least one address!" if addresses.empty? @@ -87,6 +104,11 @@ def self.load_balancer_endpoint(endpoint) ) end + # Build an Envoy address from a normalized IP or Unix address. + # @parameter address [Hash] An IP `:address` and `:port`, or a Unix `:path`. + # @returns [Envoy::Config::Core::V3::Address] The generated Envoy address. + # @raises [KeyError] If required IP address state is missing. + # @private def self.build_address(address) if path = address[:path] Envoy::Config::Core::V3::Address.new( From ed49f5b3edc875eb6879ac5e191df0515731cd18 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 24 Jul 2026 13:01:25 +1200 Subject: [PATCH 10/10] Remove trailing blank lines. --- test/async/grpc/xds/resources.rb | 3 --- 1 file changed, 3 deletions(-) diff --git a/test/async/grpc/xds/resources.rb b/test/async/grpc/xds/resources.rb index 4b4ec2c..24f606d 100644 --- a/test/async/grpc/xds/resources.rb +++ b/test/async/grpc/xds/resources.rb @@ -256,6 +256,3 @@ def http_health_check expect(endpoint.health_status).to be == :UNKNOWN end end - - -