diff --git a/.github/workflows/test-control-plane.yaml b/.github/workflows/test-control-plane.yaml index 2ab1271..e3ffdf0 100644 --- a/.github/workflows/test-control-plane.yaml +++ b/.github/workflows/test-control-plane.yaml @@ -28,9 +28,11 @@ jobs: - name: CDS and EDS publish_clusters: "true" envoy_config: ./envoy.yaml + orca: "true" - name: EDS only publish_clusters: "false" envoy_config: ./envoy-eds.yaml + orca: "false" experimental: [false] @@ -47,4 +49,5 @@ jobs: RUBY_VERSION: ${{ matrix.ruby }} PUBLISH_CLUSTERS: ${{ matrix.discovery.publish_clusters }} ENVOY_CONFIG: ${{ matrix.discovery.envoy_config }} + ORCA: ${{ matrix.discovery.orca }} run: bundle exec bake test:integration name=control-plane diff --git a/bake/async/service/supervisor/envoy.rb b/bake/async/service/supervisor/envoy.rb index 005e315..0d7f723 100644 --- a/bake/async/service/supervisor/envoy.rb +++ b/bake/async/service/supervisor/envoy.rb @@ -22,6 +22,15 @@ def endpoints end end +# Get the latest ORCA load reports sampled by the running Envoy monitor. +def orca + data = envoy_status.fetch(:data) + + data.fetch(:orca) do + raise "The Async::Service::Supervisor::Envoy::Monitor is not configured for ORCA reporting." + end +end + private def supervisor_status diff --git a/integration/control-plane/backend.rb b/integration/control-plane/backend.rb index 1b24f60..68b2408 100644 --- a/integration/control-plane/backend.rb +++ b/integration/control-plane/backend.rb @@ -6,11 +6,19 @@ require "async" require "async/http/protocol/http1" require "async/service/supervisor/worker" +require "async/utilization" require "falcon/server" require "io/endpoint/generic" require "io/endpoint/host_endpoint" require "socket" +UTILIZATION_SCHEMA = { + connections_active: :u32, + connections_total: :u64, + requests_active: :u32, + requests_total: :u64, +}.freeze + backend_id = ENV.fetch("BACKEND_ID") backend_port = Integer(ENV.fetch("BACKEND_PORT")) service_name = ENV.fetch("SERVICE_NAME") @@ -33,13 +41,15 @@ Sync do supervisor_endpoint = IO::Endpoint::Generic.parse(ENV.fetch("SUPERVISOR_ENDPOINT")) http_endpoint = IO::Endpoint.tcp("0.0.0.0", backend_port) + utilization_registry = Async::Utilization::Registry.new middleware = Falcon::Server.middleware(rack_application, cache: false) server = Falcon::Server.new( middleware, http_endpoint, protocol: Async::HTTP::Protocol::HTTP1, - scheme: "http" + scheme: "http", + utilization_registry: utilization_registry ) state = { @@ -53,7 +63,9 @@ worker = Async::Service::Supervisor::Worker.new( endpoint: supervisor_endpoint, - state: state + state: state, + utilization_schema: UTILIZATION_SCHEMA, + utilization_registry: utilization_registry ) worker.run diff --git a/integration/control-plane/docker-compose.yaml b/integration/control-plane/docker-compose.yaml index 7392673..7348cd9 100644 --- a/integration/control-plane/docker-compose.yaml +++ b/integration/control-plane/docker-compose.yaml @@ -11,6 +11,10 @@ services: - XDS_BIND=http://0.0.0.0:18000 - SUPERVISOR_ENDPOINT=tcp://0.0.0.0:12000 - PUBLISH_CLUSTERS=${PUBLISH_CLUSTERS:-true} + - ORCA=${ORCA:-false} + - UTILIZATION_PATH=/tmp/async-service-supervisor/utilization.shm + volumes: + - utilization:/tmp/async-service-supervisor backend-a: build: @@ -25,6 +29,8 @@ services: - BACKEND_PORT=9292 - SERVICE_NAME=app-http1 - SUPERVISOR_ENDPOINT=tcp://supervisor:12000 + volumes: + - utilization:/tmp/async-service-supervisor depends_on: supervisor: condition: service_started @@ -42,6 +48,8 @@ services: - BACKEND_PORT=9293 - SERVICE_NAME=app-http1 - SUPERVISOR_ENDPOINT=tcp://supervisor:12000 + volumes: + - utilization:/tmp/async-service-supervisor depends_on: supervisor: condition: service_started @@ -74,5 +82,9 @@ services: - COVERAGE=${COVERAGE} - ENVOY_URI=http://envoy:10000 - ENVOY_ADMIN_URI=http://envoy:19000 + - ORCA=${ORCA:-false} depends_on: - envoy + +volumes: + utilization: diff --git a/integration/control-plane/supervisor.rb b/integration/control-plane/supervisor.rb index eac7a3f..2e6bf08 100644 --- a/integration/control-plane/supervisor.rb +++ b/integration/control-plane/supervisor.rb @@ -5,6 +5,7 @@ require "async" require "async/service/supervisor/server" +require "async/service/supervisor/utilization_monitor" require "async/service/supervisor/envoy" require "io/endpoint/generic" require "io/endpoint/host_endpoint" @@ -15,14 +16,25 @@ def endpoint(value) Sync do supervisor_endpoint = endpoint(ENV.fetch("SUPERVISOR_ENDPOINT")) + orca = ENV.fetch("ORCA", "false") == "true" + + utilization_monitor = if orca + Async::Service::Supervisor::UtilizationMonitor.new( + path: ENV.fetch("UTILIZATION_PATH", "utilization.shm"), + interval: 1 + ) + end + monitor = Async::Service::Supervisor::Envoy::Monitor.new( bind: ENV.fetch("XDS_BIND"), - publish_clusters: ENV.fetch("PUBLISH_CLUSTERS", "true") == "true" + publish_clusters: ENV.fetch("PUBLISH_CLUSTERS", "true") == "true", + orca: orca, + utilization_monitor: utilization_monitor ) server = Async::Service::Supervisor::Server.new( endpoint: supervisor_endpoint, - monitors: [monitor] + monitors: [utilization_monitor, monitor].compact ) server.run diff --git a/integration/control-plane/test/envoy.rb b/integration/control-plane/test/envoy.rb index 2d33f25..b257a4e 100644 --- a/integration/control-plane/test/envoy.rb +++ b/integration/control-plane/test/envoy.rb @@ -29,6 +29,25 @@ def eventually(timeout: 20, interval: 0.5) raise error || "Condition was not met within #{timeout} seconds!" end + def find_hash(value, &block) + case value + when Hash + return value if yield(value) + + value.each_value do |child| + if match = find_hash(child, &block) + return match + end + end + when Array + value.each do |child| + if match = find_hash(child, &block) + return match + end + end + end + end + it "routes requests through Envoy to supervised Falcon workers" do uri = envoy_uri @@ -68,4 +87,34 @@ def eventually(timeout: 20, interval: 0.5) expect(addresses.sort).to be == [9292, 9293] end + + if ENV.fetch("ORCA", "false") == "true" + it "reports ORCA load-balancing policy to Envoy" do + uri = admin_uri + "/config_dump" + + cluster = eventually do + if (response = Net::HTTP.get_response(uri)).code.to_i == 200 + config_dump = JSON.parse(response.body) + clusters_config = config_dump.fetch("configs").find do |config| + config["@type"]&.end_with?("envoy.admin.v3.ClustersConfigDump") + end + dynamic_clusters = clusters_config.fetch("dynamic_active_clusters", []) + cluster = dynamic_clusters.filter_map{|entry| entry["cluster"]}.find{|cluster| cluster["name"] == "app-http1"} + + if cluster + cluster_json = JSON.generate(cluster) + cluster if cluster_json.include?("client_side_weighted_round_robin") || cluster_json.include?("ClientSideWeightedRoundRobin") + end + end + end + + typed_extension_config = cluster.fetch("load_balancing_policy").fetch("policies").first.fetch("typed_extension_config") + typed_config = typed_extension_config.fetch("typed_config") + + expect(typed_extension_config.fetch("name")).to be == "envoy.load_balancing_policies.client_side_weighted_round_robin" + expect(typed_config.fetch("@type")).to be == "type.googleapis.com/envoy.extensions.load_balancing_policies.client_side_weighted_round_robin.v3.ClientSideWeightedRoundRobin" + expect(typed_config.fetch("enable_oob_load_report")).to be == true + expect(typed_config.fetch("oob_reporting_period")).to be == "1s" + end + end end diff --git a/lib/async/service/supervisor/envoy/monitor.rb b/lib/async/service/supervisor/envoy/monitor.rb index aba3eb7..98e7864 100644 --- a/lib/async/service/supervisor/envoy/monitor.rb +++ b/lib/async/service/supervisor/envoy/monitor.rb @@ -144,9 +144,13 @@ def run(parent: Async::Task.current) # @returns [Hash] The clusters and endpoint hashes. def as_json @mutex.synchronize do - { + data = { clusters: build_clusters } + + data[:orca] = build_orca if @orca + + data end end @@ -267,6 +271,14 @@ def build_clusters(records_by_cluster = build_records_by_cluster) end end + def build_orca + { + port: @orca_port, + authorities: @authorities.dup, + reports: @load_reports.transform_values(&:to_h), + } + end + def cluster_configuration(records) schemes = records.map{|record| record[:endpoint].scheme}.uniq protocols = records.map{|record| record[:endpoint].protocols} diff --git a/test/async/service/supervisor/envoy/monitor.rb b/test/async/service/supervisor/envoy/monitor.rb index 56166e9..bd58ded 100644 --- a/test/async/service/supervisor/envoy/monitor.rb +++ b/test/async/service/supervisor/envoy/monitor.rb @@ -151,6 +151,17 @@ def processor(samples = []) expect(report.cpu_utilization).to be == 0.5 expect(report.rps_fractional).to be == 2.0 expect(report.named_metrics).to be == {"orca.heartbeat" => 0.0} + expect(monitor.as_json[:orca]).to be == { + port: 18000, + authorities: {"worker-1" => 1}, + reports: { + "worker-1" => { + cpu_utilization: 0.5, + rps_fractional: 2.0, + named_metrics: {"orca.heartbeat" => 0.0}, + } + } + } expect(monitor.worker?("worker-1")).to be == true monitor.remove(controller) diff --git a/test/bake/async/service/supervisor/envoy.rb b/test/bake/async/service/supervisor/envoy.rb index c9220a0..44fc11c 100644 --- a/test/bake/async/service/supervisor/envoy.rb +++ b/test/bake/async/service/supervisor/envoy.rb @@ -17,6 +17,17 @@ "api" => [ {addresses: [{path: "/tmp/api.ipc"}], healthy: true} ] + }, + orca: { + port: 18000, + authorities: {"worker-1" => 1}, + reports: { + "worker-1" => { + cpu_utilization: 0.5, + rps_fractional: 2.0, + named_metrics: {"orca.heartbeat" => 0.0} + } + } } } } @@ -48,6 +59,17 @@ def invoke(name, status: self.status) "api" => [ {addresses: [{path: "/tmp/api.ipc"}], healthy: true} ] + }, + orca: { + port: 18000, + authorities: {"worker-1" => 1}, + reports: { + "worker-1" => { + cpu_utilization: 0.5, + rps_fractional: 2.0, + named_metrics: {"orca.heartbeat" => 0.0} + } + } } } } @@ -71,9 +93,38 @@ def invoke(name, status: self.status) ] end + it "returns ORCA load reports" do + expect(invoke("orca")).to be == { + port: 18000, + authorities: {"worker-1" => 1}, + reports: { + "worker-1" => { + cpu_utilization: 0.5, + rps_fractional: 2.0, + named_metrics: {"orca.heartbeat" => 0.0} + } + } + } + end + it "fails when the Envoy monitor is not running" do expect do invoke("status", status: []) end.to raise_exception(RuntimeError, message: be =~ /no .*envoy.*monitor/i) end + + it "fails when ORCA reporting is not configured" do + status = [ + { + type: "Async::Service::Supervisor::Envoy::Monitor", + data: { + clusters: {} + } + } + ] + + expect do + invoke("orca", status: status) + end.to raise_exception(RuntimeError, message: be =~ /ORCA reporting/) + end end