diff --git a/lib/async/grpc/xds/resource_builder.rb b/lib/async/grpc/xds/resource_builder.rb index c94f0c2..5b48fb9 100644 --- a/lib/async/grpc/xds/resource_builder.rb +++ b/lib/async/grpc/xds/resource_builder.rb @@ -31,8 +31,16 @@ 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( + # 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, type: Envoy::Config::Cluster::V3::Cluster::DiscoveryType::EDS, eds_cluster_config: Envoy::Config::Cluster::V3::Cluster::EdsClusterConfig.new( @@ -43,10 +51,24 @@ 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 + # Envoy uses HTTP/1 by default. + when :http2 + 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 + # 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, @@ -57,55 +79,61 @@ 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) - endpoint = normalize_endpoint(endpoint) - + addresses, healthy = endpoint.fetch_values(:addresses, :healthy) + raise ArgumentError, "An endpoint requires at least one address!" if addresses.empty? + + address, *additional_addresses = addresses + 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: additional_addresses.map do |additional_address| + Envoy::Config::Endpoint::V3::Endpoint::AdditionalAddress.new( + address: build_address(additional_address) ) - ), - hostname: endpoint[:hostname].to_s + end ), - health_status: health_status_value(endpoint.fetch(:healthy, true)) + health_status: health_status_value(healthy) ) end - - 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, - hostname: endpoint[:hostname] || endpoint["hostname"], - healthy: endpoint.key?(:healthy) ? endpoint[:healthy] : endpoint.fetch("healthy", true) - } + + # 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( + pipe: Envoy::Config::Core::V3::Pipe.new(path: path) + ) else - if endpoint.respond_to?(:address) && endpoint.respond_to?(:port) - { - 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 + Envoy::Config::Core::V3::Address.new( + socket_address: Envoy::Config::Core::V3::SocketAddress.new( + protocol: Envoy::Config::Core::V3::SocketAddress::Protocol::TCP, + address: address.fetch(:address), + port_value: address.fetch(:port) + ) + ) end end - + + private_class_method :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" @@ -118,7 +146,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 @@ -135,4 +163,3 @@ def self.health_status_value(healthy) end end end - \ No newline at end of file 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/control_plane.rb b/test/async/grpc/xds/control_plane.rb index 99186b0..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", [{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", [{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 @@ -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}], 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 c7b8f36..fb3f532 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) @@ -31,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}], healthy: true}, + {addresses: [{address: "127.0.0.2", port: 50052}], healthy: false} ] ) @@ -44,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 @@ -53,24 +58,31 @@ 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") - - load_balancer_endpoint = subject.load_balancer_endpoint(endpoint) + 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}, + ], + healthy: true, + }) + endpoint = 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.hostname).to be == "three" - expect(load_balancer_endpoint.health_status).to be == :UNHEALTHY + 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 invalid endpoint data" do + it "rejects endpoints without addresses" do expect do - subject.load_balancer_endpoint(Object.new) + 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 f526266..24f606d 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,34 +158,34 @@ 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", [ - {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} ] ) 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,7 @@ def http_health_check }, health_status: :DRAINING ) - + expect(endpoint.health_status).to be == :UNKNOWN end end - - - - - - \ No newline at end of file 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(