Exercise ORCA in control plane integration

.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

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

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

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:

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

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

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}

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)

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