diff --git a/context/getting-started.md b/context/getting-started.md index 35e94f4..4d5fbd1 100644 --- a/context/getting-started.md +++ b/context/getting-started.md @@ -41,16 +41,18 @@ state = { Workers without endpoint state are ignored by the Envoy monitor. -Falcon cluster workers can register their concrete post-bind listener automatically: +Falcon server workers can register their concrete bound listener automatically: ``` ruby service "application" do - include Falcon::Environment::Cluster + include Falcon::Environment::Server include Async::Service::Supervisor::Envoy::Supervised end ``` -Falcon describes its bound server resource as a listener. The integration converts that listener into Envoy upstream endpoint state, including its name, scheme, supported protocols, and every concrete IP or Unix socket address. Addresses belonging to one listener remain grouped as one Envoy load-balancer endpoint. +`Falcon::Environment::Server` reports one listener shared by all of its workers, while `Falcon::Environment::Cluster` reports the listener bound independently by each worker. Falcon describes either bound server resource as a listener. The integration converts that listener into Envoy upstream endpoint state, including its name, scheme, supported protocols, and every concrete IP or Unix socket address. + +Addresses belonging to one listener remain grouped as one Envoy load-balancer endpoint. When several workers report the same shared listener, the monitor publishes it once and keeps it available while any reporting worker is healthy. ## Monitor Usage diff --git a/guides/getting-started/readme.md b/guides/getting-started/readme.md index 35e94f4..4d5fbd1 100644 --- a/guides/getting-started/readme.md +++ b/guides/getting-started/readme.md @@ -41,16 +41,18 @@ state = { Workers without endpoint state are ignored by the Envoy monitor. -Falcon cluster workers can register their concrete post-bind listener automatically: +Falcon server workers can register their concrete bound listener automatically: ``` ruby service "application" do - include Falcon::Environment::Cluster + include Falcon::Environment::Server include Async::Service::Supervisor::Envoy::Supervised end ``` -Falcon describes its bound server resource as a listener. The integration converts that listener into Envoy upstream endpoint state, including its name, scheme, supported protocols, and every concrete IP or Unix socket address. Addresses belonging to one listener remain grouped as one Envoy load-balancer endpoint. +`Falcon::Environment::Server` reports one listener shared by all of its workers, while `Falcon::Environment::Cluster` reports the listener bound independently by each worker. Falcon describes either bound server resource as a listener. The integration converts that listener into Envoy upstream endpoint state, including its name, scheme, supported protocols, and every concrete IP or Unix socket address. + +Addresses belonging to one listener remain grouped as one Envoy load-balancer endpoint. When several workers report the same shared listener, the monitor publishes it once and keeps it available while any reporting worker is healthy. ## Monitor Usage diff --git a/lib/async/service/supervisor/envoy/endpoint.rb b/lib/async/service/supervisor/envoy/endpoint.rb index accf1b1..dc6f51f 100644 --- a/lib/async/service/supervisor/envoy/endpoint.rb +++ b/lib/async/service/supervisor/envoy/endpoint.rb @@ -7,8 +7,23 @@ module Async module Service module Supervisor module Envoy - # Represents one upstream endpoint published to Envoy EDS. + # Represents an upstream endpoint published to Envoy EDS. class Endpoint + # Build an immutable endpoint. + # @parameter name [String] The upstream cluster name. + # @parameter scheme [String | Symbol] The upstream application scheme. + # @parameter protocols [Array(String)] The supported upstream HTTP protocol names. + # @parameter addresses [Array(Hash)] The grouped concrete addresses. + # @returns [Endpoint] The immutable endpoint. + def self.build(name:, scheme:, protocols:, addresses:) + new( + name.to_s, + scheme.to_sym, + protocols.map(&:to_s).uniq, + addresses, + ).tap(&:freeze) + end + # Wrap serialized endpoint state. # @parameter value [Endpoint | Hash] The value to wrap. # @returns [Endpoint] The endpoint value. @@ -17,39 +32,26 @@ def self.wrap(value) when self value when Hash - new(**value) + build(**value) else raise ArgumentError, "Invalid Envoy endpoint: #{value.inspect}" end end - # Normalize a concrete endpoint address. - # @parameter value [Hash] The address value. - # @returns [Hash] A normalized IP or Unix address. - def self.normalize_address(value) - if path = value[:path] - raise ArgumentError, "A Unix endpoint cannot specify an IP address or port!" if value[:address] || value[:port] - - {path: path.to_s}.freeze - elsif value[:address] && value[:port] - {address: value[:address].to_s, port: Integer(value[:port])}.freeze - else - raise ArgumentError, "An endpoint address requires either path, or address and port: #{value.inspect}" - end - end - # Initialize an endpoint. # @parameter name [String] The upstream cluster name. - # @parameter scheme [String | Symbol] The upstream application scheme. + # @parameter scheme [Symbol] The upstream application scheme. # @parameter protocols [Array(String)] The supported upstream HTTP protocol names. # @parameter addresses [Array(Hash)] The grouped concrete addresses. - def initialize(name:, scheme:, protocols:, addresses:) - @name = name.to_s - @scheme = scheme.to_sym - @protocols = protocols.map(&:to_s).uniq.freeze + def initialize(name, scheme, protocols, addresses) + @name = name + @scheme = scheme + @protocols = protocols raise ArgumentError, "An endpoint requires at least one protocol!" if @protocols.empty? - @addresses = addresses.map{|value| self.class.normalize_address(value)}.freeze + @addresses = addresses raise ArgumentError, "An endpoint requires at least one address!" if @addresses.empty? + + @hash = nil end # @attribute [String] The upstream cluster name. @@ -64,6 +66,39 @@ def initialize(name:, scheme:, protocols:, addresses:) # @attribute [Array(Hash)] The grouped concrete addresses. attr :addresses + # Freeze this endpoint and cache its value hash. + def freeze + return self if frozen? + + @name.freeze + @protocols.each(&:freeze).freeze + @addresses.each do |address| + address.each_value(&:freeze) + address.freeze + end.freeze + @hash = self.hash + + super + end + + # Compare this endpoint with another endpoint by value. + # @parameter other [Object] The object to compare. + # @returns [Boolean] Whether both endpoints have identical values. + def eql?(other) + other.instance_of?(self.class) && + @name == other.name && + @scheme == other.scheme && + @protocols == other.protocols && + @addresses == other.addresses + end + + alias == eql? + + # Compute the value hash used when grouping endpoints. + # @returns [Integer] The endpoint value hash. + def hash + @hash || [self.class, @name, @scheme, @protocols, @addresses].hash + end end end end diff --git a/lib/async/service/supervisor/envoy/endpoint_group.rb b/lib/async/service/supervisor/envoy/endpoint_group.rb new file mode 100644 index 0000000..1a9dc73 --- /dev/null +++ b/lib/async/service/supervisor/envoy/endpoint_group.rb @@ -0,0 +1,63 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +module Async + module Service + module Supervisor + module Envoy + # Groups the workers reporting a single upstream endpoint. + class EndpointGroup + # Initialize an endpoint group. + # @parameter endpoint [Endpoint] The endpoint shared by the workers. + def initialize(endpoint) + @endpoint = endpoint + @workers = {} + end + + # @attribute [Endpoint] The endpoint shared by the workers. + attr :endpoint + + # Add or update a worker report. + # @parameter worker [SupervisorController] The worker reporting the endpoint. + # @parameter healthy [Boolean] Whether the worker considers the endpoint healthy. + def add(worker, healthy:) + @workers[worker.id] = healthy + end + + # Remove a worker report. + # @parameter worker [SupervisorController] The worker to remove. + # @returns [Boolean | Nil] The removed health value, if present. + def remove(worker) + @workers.delete(worker.id) + end + + # Count the workers reporting this endpoint. + # @returns [Integer] The number of reporting workers. + def size + @workers.size + end + + # Determine whether any workers report this endpoint. + # @returns [Boolean] Whether the group has no workers. + def empty? + @workers.empty? + end + + # Determine the aggregate endpoint health. + # @returns [Boolean] Whether any reporting worker is healthy. + def healthy? + @workers.each_value.any? + end + + # Convert the group to an xDS endpoint description. + # @returns [Hash] The endpoint addresses and aggregate health. + def as_json + {addresses: @endpoint.addresses, healthy: healthy?} + end + end + end + end + end +end diff --git a/lib/async/service/supervisor/envoy/monitor.rb b/lib/async/service/supervisor/envoy/monitor.rb index fed8cc0..0d26f4f 100644 --- a/lib/async/service/supervisor/envoy/monitor.rb +++ b/lib/async/service/supervisor/envoy/monitor.rb @@ -10,6 +10,7 @@ require_relative "delegate" require_relative "endpoint" +require_relative "endpoint_group" module Async module Service @@ -108,6 +109,7 @@ def build_record(supervisor_controller, endpoint) { cluster: cluster.to_s, endpoint: endpoint, + worker: supervisor_controller, healthy: @delegate.healthy?(supervisor_controller, endpoint), } end @@ -146,9 +148,14 @@ def build_records_by_cluster def build_clusters(records_by_cluster = build_records_by_cluster) records_by_cluster.transform_values do |records| - records.map do |record| - {addresses: record[:endpoint].addresses, healthy: record[:healthy]} + groups = {} + + records.each do |record| + group = groups[record[:endpoint]] ||= EndpointGroup.new(record[:endpoint]) + group.add(record[:worker], healthy: record[:healthy]) end + + groups.each_value.map(&:as_json) end end diff --git a/lib/async/service/supervisor/envoy/supervised.rb b/lib/async/service/supervisor/envoy/supervised.rb index e1f6693..a7443c6 100644 --- a/lib/async/service/supervisor/envoy/supervised.rb +++ b/lib/async/service/supervisor/envoy/supervised.rb @@ -9,7 +9,7 @@ module Async module Service module Supervisor module Envoy - # Registers post-bind cluster listeners with the supervisor. + # Registers bound Falcon listeners with the supervisor. module Supervised include Async::Service::Supervisor::Supervised diff --git a/readme.md b/readme.md index 2c1d20e..be79fc7 100644 --- a/readme.md +++ b/readme.md @@ -24,6 +24,10 @@ Please see the [project documentation](https://socketry.github.io/async-service- Please see the [project releases](https://socketry.github.io/async-service-supervisor-envoy/releases/index) for all releases. +### Unreleased + + - Deduplicate shared listener endpoints reported by multiple supervised workers. + ### v0.0.1 - Register concrete Falcon cluster listeners as Envoy upstream endpoint state after binding. diff --git a/releases.md b/releases.md index 9bd3b6b..6191fc2 100644 --- a/releases.md +++ b/releases.md @@ -1,5 +1,9 @@ # Releases +## Unreleased + + - Deduplicate immutable endpoint values reported by multiple supervised workers and aggregate their health. + ## v0.0.1 - Register concrete Falcon cluster listeners as Envoy upstream endpoint state after binding. diff --git a/test/async/service/supervisor/envoy/endpoint.rb b/test/async/service/supervisor/envoy/endpoint.rb new file mode 100644 index 0000000..06d3898 --- /dev/null +++ b/test/async/service/supervisor/envoy/endpoint.rb @@ -0,0 +1,75 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "async/service/supervisor/envoy/endpoint" + +describe Async::Service::Supervisor::Envoy::Endpoint do + let(:attributes) do + { + name: "api", + scheme: "http", + protocols: ["h2"], + addresses: [{path: "/tmp/api.ipc"}], + } + end + + it "wraps endpoint values" do + endpoint = subject.wrap(attributes) + + expect(endpoint.name).to be == "api" + expect(endpoint.scheme).to be == :http + expect(endpoint.protocols).to be == ["h2"] + expect(endpoint.addresses).to be == [{path: "/tmp/api.ipc"}] + end + + it "returns endpoint instances unchanged" do + endpoint = subject.build(**attributes) + + expect(subject.wrap(endpoint)).to be_equal(endpoint) + end + + it "has immutable value semantics" do + endpoint = subject.build(**attributes) + equivalent = subject.build(**attributes) + + expect(endpoint).to be == equivalent + expect(endpoint.eql?(equivalent)).to be == true + expect(endpoint.hash).to be == equivalent.hash + expect({endpoint => true}[equivalent]).to be == true + + expect(endpoint.frozen?).to be == true + expect(endpoint.name.frozen?).to be == true + expect(endpoint.protocols.frozen?).to be == true + expect(endpoint.protocols.first.frozen?).to be == true + expect(endpoint.addresses.frozen?).to be == true + expect(endpoint.addresses.first.frozen?).to be == true + expect(endpoint.addresses.first[:path].frozen?).to be == true + expect(endpoint.freeze).to be_equal(endpoint) + end + + it "preserves address order in endpoint identity" do + addresses = [ + {address: "127.0.0.1", port: 9292}, + {path: "/tmp/api.ipc"}, + ] + + endpoint = subject.build(**attributes, addresses: addresses) + reordered = subject.build(**attributes, addresses: addresses.reverse) + + expect(endpoint).not.to be == reordered + end + + it "rejects invalid endpoint objects" do + expect do + subject.wrap(Object.new) + end.to raise_exception(ArgumentError) + end + + it "rejects endpoints without protocols" do + expect do + subject.build(**attributes, protocols: []) + end.to raise_exception(ArgumentError) + end +end diff --git a/test/async/service/supervisor/envoy/endpoint_group.rb b/test/async/service/supervisor/envoy/endpoint_group.rb new file mode 100644 index 0000000..7c3744a --- /dev/null +++ b/test/async/service/supervisor/envoy/endpoint_group.rb @@ -0,0 +1,54 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "async/service/supervisor/envoy/endpoint_group" +require "async/service/supervisor/envoy/endpoint" + +describe Async::Service::Supervisor::Envoy::EndpointGroup do + Worker = Struct.new(:id) + + let(:endpoint) do + Async::Service::Supervisor::Envoy::Endpoint.build( + name: "api", + scheme: "http", + protocols: ["h2"], + addresses: [{path: "/tmp/api.ipc"}], + ) + end + + let(:group) {subject.new(endpoint)} + + it "tracks workers precisely and aggregates health" do + first = Worker.new(1) + second = Worker.new(2) + + group.add(first, healthy: false) + group.add(second, healthy: true) + + expect(group.size).to be == 2 + expect(group.healthy?).to be == true + expect(group.as_json).to be == { + addresses: [{path: "/tmp/api.ipc"}], + healthy: true, + } + + group.remove(second) + + expect(group.size).to be == 1 + expect(group.healthy?).to be == false + + group.remove(first) + + expect(group.empty?).to be == true + end + + it "updates an existing worker report by worker ID" do + group.add(Worker.new(1), healthy: false) + group.add(Worker.new(1), healthy: true) + + expect(group.size).to be == 1 + expect(group.healthy?).to be == true + end +end diff --git a/test/async/service/supervisor/envoy/monitor.rb b/test/async/service/supervisor/envoy/monitor.rb index f5e3f00..62716d1 100644 --- a/test/async/service/supervisor/envoy/monitor.rb +++ b/test/async/service/supervisor/envoy/monitor.rb @@ -100,6 +100,36 @@ def endpoint_assignment(cluster) expect(assignment.endpoints.first.lb_endpoints).to be(:empty?) end + it "publishes a shared listener once while any worker remains" do + delegate = Class.new(Async::Service::Supervisor::Envoy::Delegate) do + def healthy?(supervisor_controller, endpoint) + supervisor_controller.state[:healthy] + end + end.new + + monitor = subject.new(delegate: delegate) + endpoint = {name: "myservice", scheme: "http", protocols: ["h2"], addresses: [{address: "127.0.0.1", port: 50051}]} + first = Controller.new(1, {endpoint: endpoint, healthy: false}) + second = Controller.new(2, {endpoint: endpoint, healthy: true}) + + monitor.register(first) + monitor.register(second) + + expect(monitor.as_json[:clusters]["myservice"]).to be == [ + {addresses: [{address: "127.0.0.1", port: 50051}], healthy: true} + ] + + monitor.remove(second) + + expect(monitor.as_json[:clusters]["myservice"]).to be == [ + {addresses: [{address: "127.0.0.1", port: 50051}], healthy: false} + ] + + monitor.remove(first) + + expect(monitor.as_json[:clusters]["myservice"]).to be == nil + end + it "groups workers by service name" do monitor.register(Controller.new(1, { endpoint: {name: "service-a", scheme: "http", protocols: ["h2"], addresses: [{address: "127.0.0.1", port: 50051}]} @@ -211,7 +241,7 @@ def healthy?(supervisor_controller, endpoint) delegate = Class.new(Async::Service::Supervisor::Envoy::Delegate) do def endpoint_list(supervisor_controller) [ - Async::Service::Supervisor::Envoy::Endpoint.new( + Async::Service::Supervisor::Envoy::Endpoint.build( name: "ignored", scheme: "http", protocols: ["http/1.1"], @@ -245,48 +275,6 @@ def healthy?(supervisor_controller, endpoint) } end - it "wraps endpoint values" do - endpoint = Async::Service::Supervisor::Envoy::Endpoint.wrap( - {name: "api", scheme: "http", protocols: ["h2"], addresses: [{path: "/tmp/api.ipc"}]} - ) - - expect(endpoint.name).to be == "api" - expect(endpoint.scheme).to be == :http - expect(endpoint.protocols).to be == ["h2"] - expect(endpoint.protocols.frozen?).to be == true - expect(endpoint.addresses).to be == [{path: "/tmp/api.ipc"}] - end - - it "returns endpoint instances unchanged" do - endpoint = Async::Service::Supervisor::Envoy::Endpoint.new( - name: "api", scheme: "http", protocols: ["h2"], addresses: [{path: "/tmp/api.ipc"}] - ) - - expect(Async::Service::Supervisor::Envoy::Endpoint.wrap(endpoint)).to be == endpoint - end - - it "rejects invalid endpoint objects" do - expect do - Async::Service::Supervisor::Envoy::Endpoint.wrap(Object.new) - end.to raise_exception(ArgumentError) - end - - it "rejects endpoints without protocols" do - expect do - Async::Service::Supervisor::Envoy::Endpoint.new( - name: "api", scheme: "http", protocols: [], addresses: [{path: "/tmp/api.ipc"}] - ) - end.to raise_exception(ArgumentError) - end - - it "rejects invalid endpoint addresses" do - expect do - Async::Service::Supervisor::Envoy::Endpoint.new( - name: "api", scheme: "http", protocols: ["h2"], addresses: [{}] - ) - end.to raise_exception(ArgumentError) - end - it "selects the preferred common endpoint protocol" do monitor.register(Controller.new(1, { endpoint: {name: "myservice", scheme: "http", protocols: ["h2", "http/1.1"], addresses: [{path: "/tmp/one.ipc"}]}