Fairness

api/aperture/policy/language/v1/flowcontrol.proto

@@ -455,6 +455,9 @@ message Scheduler {
   // Key for a flow label that can be used to provide workloads for this request.
   // If this parameter is not provided, the workloads for the flow will be determined by the matched workload's name in the policy.
   string workload_label_key = 10;
+
+  // Key for a flow label that is used to enforce fairness amongst requests in a workload. If not specified, requests within a workload of the same priority are admitted in a FIFO manner.
+  string fairness_label_key = 11;
 }
 
 // _AIMD Load Scheduler_ uses a Gradient Controller to throttle the token rate based on the deviation of the signal from the setpoint.

api/buf.lock

@@ -9,8 +9,8 @@ deps:
   - remote: buf.build
     owner: envoyproxy
     repository: envoy
-    commit: 39f01805b8754956a78bc77bda4c3761
-    digest: shake256:a7e7c7675618056879597d6205e394e85b63b71fd1766818042e6fd9b5c4fd623bede2784a9c635dced61fb40a22610812f164ed40c7a0cecbe50c0ecc76952d
+    commit: 10791366aeb94287b5f8b62d1be8ec2e
+    digest: shake256:193299dfe4b65518327fccde2953e05969f23919eaabd68e8ad73bf3a88465620749671b1db7f76de02d38cba6d9a4b19ad947f7ab987de353c3333e683452ca
   - remote: buf.build
     owner: envoyproxy
     repository: protoc-gen-validate

blueprints/gen/jsonschema/_definitions.json

@@ -3595,6 +3595,10 @@
           "description": "This field allows you to override the default HTTP status code (`503 Service Unavailable`) that is returned when a request is denied.\n\n",
           "x-go-tag-default": "ServiceUnavailable"
         },
+        "fairness_label_key": {
+          "description": "Key for a flow label that is used to enforce fairness amongst requests in a workload. If not specified, requests within a workload of the same priority are admitted in a FIFO manner.",
+          "type": "string"
+        },
         "priority_label_key": {
           "description": "Key for a flow label that can be used to override the default priority for this flow.\nThe value associated with this key must be a valid number. Higher numbers means higher priority.\nIf this parameter is not provided, the priority for the flow will be determined by the matched workload's priority.",
           "type": "string"

blueprints/gen/v1/scheduler.libsonnet

@@ -19,6 +19,12 @@
   withDeniedResponseStatusCodeMixin(denied_response_status_code):: {
     denied_response_status_code+: denied_response_status_code,
   },
+  withFairnessLabelKey(fairness_label_key):: {
+    fairness_label_key: fairness_label_key,
+  },
+  withFairnessLabelKeyMixin(fairness_label_key):: {
+    fairness_label_key+: fairness_label_key,
+  },
   withPriorityLabelKey(priority_label_key):: {
     priority_label_key: priority_label_key,
   },

dashboards/grafana/dashboards/summary/scheduler/rows-fn.libsonnet

@@ -193,26 +193,4 @@ function(datasourceName, policyName, componentID, extraFilters={})
     ),
   ];
 
-  local legendFormat = '{{ instance }} - {{ policy_name }}';
-  local row10 = [
-    barGaugePanel(
-      'WFQ Scheduler Flows',
-      datasourceName,
-      'avg(wfq_flows_total{%(filters)s})' % { filters: stringFilters },
-      x=0,
-      h=6,
-      w=12,
-      legendFormat=legendFormat
-    ),
-    barGaugePanel(
-      'WFQ Scheduler Heap Requests',
-      datasourceName,
-      'avg(wfq_requests_total{%(filters)s})' % { filters: stringFilters },
-      x=12,
-      h=6,
-      w=12,
-      legendFormat=legendFormat
-    ),
-  ];
-
-  [row1, row2, row3, row4, row5, row6, row7, row8, row9, row10]
+  [row1, row2, row3, row4, row5, row6, row7, row8, row9]

docs/content/assets/openapiv2/aperture-controller.swagger.yaml

@@ -3890,6 +3890,9 @@ definitions:
                     This field allows you to override the default HTTP status code (`503 Service Unavailable`) that is returned when a request is denied.
 
                 x-go-tag-default: ServiceUnavailable
+            fairness_label_key:
+                description: Key for a flow label that is used to enforce fairness amongst requests in a workload. If not specified, requests within a workload of the same priority are admitted in a FIFO manner.
+                type: string
             priority_label_key:
                 description: |-
                     Key for a flow label that can be used to override the default priority for this flow.

docs/content/assets/openapiv2/aperture.swagger.yaml

@@ -4728,6 +4728,9 @@ definitions:
                     This field allows you to override the default HTTP status code (`503 Service Unavailable`) that is returned when a request is denied.
 
                 x-go-tag-default: ServiceUnavailable
+            fairness_label_key:
+                description: Key for a flow label that is used to enforce fairness amongst requests in a workload. If not specified, requests within a workload of the same priority are admitted in a FIFO manner.
+                type: string
             priority_label_key:
                 description: |-
                     Key for a flow label that can be used to override the default priority for this flow.

docs/content/reference/configuration/spec.md

@@ -195,7 +195,7 @@
 
 <!-- vale on -->
 
 Desired Load multiplier is the ratio of desired token rate to the incoming token
 rate.
 
 </dd>
@@ -470,7 +470,7 @@
 
 <!-- vale on -->
 
 Desired Load multiplier is the ratio of desired token rate to the incoming token
 rate.
 
 </dd>
@@ -730,7 +730,7 @@
 
 <!-- vale on -->
 
 Desired Load multiplier is the ratio of desired token rate to the incoming token
 rate.
 
 </dd>
@@ -1042,7 +1042,7 @@
 Signals are mapped to Boolean values as follows:
 
 - Zero is treated as false.
 - Any non-zero is treated as true.
 - Invalid inputs are considered unknown.
 
   :::note
@@ -1717,7 +1717,7 @@
 
 A signal can also have a special **Invalid** value. It's usually used to
 communicate that signal does not have a meaningful value at the moment, for
 example, [PromQL](#prom-q-l) emits such a value if it cannot execute a query.
 Components know when their input signals are invalid and can act accordingly.
 They can either propagate the invalid signal, by making their output itself
 invalid (for example, [ArithmeticCombinator](#arithmetic-combinator)) or use
@@ -4219,9 +4219,9 @@
 $$
 
 Note the additional `optimize` signal, that can be used to "nudge" the
 controller into desired idle state.
 
 The output can be _optionally_ clamped to desired range using `max` and `min`
 input.
 
 <dl>
@@ -4385,7 +4385,7 @@
 
 <!-- vale on -->
 
 Computed desired value of the control variable.
 
 </dd>
 </dl>
@@ -4574,7 +4574,7 @@
 <!-- vale on -->
 
 Resets the holder output to the current input signal when reset signal is valid
 and non-zero.
 
 </dd>
 </dl>
@@ -5008,7 +5008,7 @@
 
 <!-- vale on -->
 
 Resets the integrator output to zero when reset signal is valid and non-zero.
 Reset also resets the max and min constraints.
 
 </dd>
@@ -6415,7 +6415,7 @@
 It takes a signal and a setpoint as inputs and uses the PID algorithm to compute
 proportional (proportional to the magnitude of error), integral (proportional to
 accumulation of error), and derivative (proportional to how fast the signal is
 changing) terms which are summed up to get a desired output.
 
 <!-- vale off -->
 
@@ -6502,7 +6502,7 @@
 
 <!-- vale on -->
 
 The desired setpoint for the signal
 
 </dd>
 <dt>signal</dt>
@@ -6653,7 +6653,7 @@
 - The parameter must cover whole segment.
 - Additionally, path template can end with `/*` wildcard to match arbitrary
   number of trailing segments (0 or more).
 - Multiple consecutive `/` are ignored, as well as trailing `/`.
 - Parametrized path segments must come after static segments.
 - `*`, if present, must come last.
 - Most specific template "wins" (`/foo` over `/{}` and `/{}` over `/*`).
@@ -6846,7 +6846,7 @@
 
 <!-- vale on -->
 
 The number of replicas that are desired.
 
 </dd>
 </dl>
@@ -7594,7 +7594,7 @@
 
 <!-- vale on -->
 
 Desired Load multiplier is the ratio of desired token rate to the incoming token
 rate.
 
 </dd>
@@ -8805,6 +8805,20 @@
 This field allows you to override the default HTTP status code
 (`503 Service Unavailable`) that is returned when a request is denied.
 
+</dd>
+<dt>fairness_label_key</dt>
+<dd>
+
+<!-- vale off -->
+
+(string)
+
+<!-- vale on -->
+
+Key for a flow label that is used to enforce fairness amongst requests in a
+workload. If not specified, requests within a workload of the same priority are
+admitted in a FIFO manner.
+
 </dd>
 <dt>priority_label_key</dt>
 <dd>
@@ -8895,7 +8909,7 @@
 <!-- vale on -->
 
 Workload defines a class of flows that preferably have similar properties such
 as response latency and desired priority.
 
 <dl>
 <dt>label_matcher</dt>
@@ -9137,7 +9151,7 @@
 The _Signal Generator_ component generates a smooth and continuous signal by
 following a sequence of specified steps. Each step has two parameters:
 
 - `target_output`: The desired output value at the end of the step.
 - `duration`: The time it takes for the signal to change linearly from the
   previous step's `target_output` to the current step's `target_output`.
 

docs/content/reference/observability/prometheus-metrics/agent.md

@@ -254,8 +254,6 @@ This document describes the Prometheus metrics generated by Aperture Agents.
 
 | Name                                | Type    | Labels                                                                                                                           | Unit            | Description                                                                       |
 | ----------------------------------- | ------- | -------------------------------------------------------------------------------------------------------------------------------- | --------------- | --------------------------------------------------------------------------------- |
-| wfq_flows_total                     | Gauge   | agent_group, instance, job, process_uuid, policy_name, policy_hash, component_id                                                 | count (no unit) | A gauge that tracks the number of flows in the WFQScheduler                       |
-| wfq_requests_total                  | Gauge   | agent_group, instance, job, process_uuid, policy_name, policy_hash, component_id                                                 | count (no unit) | A gauge that tracks the number of queued requests in the WFQScheduler             |
 | token_bucket_lm_ratio               | Gauge   | agent_group, instance, job, process_uuid, policy_name, policy_hash, component_id                                                 | percentage      | A gauge that tracks the load multiplier                                           |
 | token_bucket_fill_rate              | Gauge   | agent_group, instance, job, process_uuid, policy_name, policy_hash, component_id                                                 | tokens/s        | A gauge that tracks the fill rate of token bucket                                 |
 | token_bucket_capacity_total         | Gauge   | agent_group, instance, job, process_uuid, policy_name, policy_hash, component_id                                                 | count (no unit) | A gauge that tracks the capacity of token bucket                                  |

docs/gen/policy/policy.yaml

@@ -3804,6 +3804,9 @@ definitions:
                     This field allows you to override the default HTTP status code (`503 Service Unavailable`) that is returned when a request is denied.
 
                 x-go-tag-default: ServiceUnavailable
+            fairness_label_key:
+                description: Key for a flow label that is used to enforce fairness amongst requests in a workload. If not specified, requests within a workload of the same priority are admitted in a FIFO manner.
+                type: string
             priority_label_key:
                 description: |-
                     Key for a flow label that can be used to override the default priority for this flow.

go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/buger/jsonparser v1.1.1
 	github.com/buraksezer/olric v0.0.0-00010101000000-000000000000
 	github.com/cenkalti/backoff/v4 v4.2.1
+	github.com/cespare/xxhash v1.1.0
 	github.com/charmbracelet/bubbletea v0.25.0
 	github.com/clarketm/json v1.17.1
 	github.com/containerd/cgroups v1.1.0

pkg/metrics/schema.go

@@ -120,10 +120,6 @@ const (
 	// RejectedTokensMetricName - total work measured in tokens of all rejected requests.
 	RejectedTokensMetricName = "rejected_tokens_total"
 
-	// WFQFlowsMetricName - weighted fair queuing number of flows gauge.
-	WFQFlowsMetricName = "wfq_flows_total"
-	// WFQRequestsMetricName - weighted fair queuing number of requests gauge.
-	WFQRequestsMetricName = "wfq_requests_total"
 	// TokenBucketLMMetricName - a gauge that tracks the load multiplier.
 	TokenBucketLMMetricName = "token_bucket_lm_ratio"
 	// TokenBucketFillRateMetricName - a gauge that tracks the fill rate of token bucket.

pkg/policies/flowcontrol/actuators/workload-scheduler/scheduler.go

@@ -40,9 +40,6 @@ type Factory struct {
 	registry status.Registry
 
 	// WFQ Metrics.
-	wfqFlowsGaugeVec    *prometheus.GaugeVec
-	wfqRequestsGaugeVec *prometheus.GaugeVec
-
 	incomingTokensCounterVec *prometheus.CounterVec
 	acceptedTokensCounterVec *prometheus.CounterVec
 	rejectedTokensCounterVec *prometheus.CounterVec
@@ -69,20 +66,6 @@ func newFactory(
 		registry: reg,
 	}
 
-	wsFactory.wfqFlowsGaugeVec = prometheus.NewGaugeVec(
-		prometheus.GaugeOpts{
-			Name: metrics.WFQFlowsMetricName,
-			Help: "A gauge that tracks the number of flows in the WFQScheduler",
-		},
-		MetricLabelKeys,
-	)
-	wsFactory.wfqRequestsGaugeVec = prometheus.NewGaugeVec(
-		prometheus.GaugeOpts{
-			Name: metrics.WFQRequestsMetricName,
-			Help: "A gauge that tracks the number of queued requests in the WFQScheduler",
-		},
-		MetricLabelKeys,
-	)
 	wsFactory.incomingTokensCounterVec = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Name: metrics.IncomingTokensMetricName,
@@ -170,15 +153,7 @@ func newFactory(
 		OnStart: func(_ context.Context) error {
 			var merr error
 
-			err := prometheusRegistry.Register(wsFactory.wfqFlowsGaugeVec)
-			if err != nil {
-				merr = multierr.Append(merr, err)
-			}
-			err = prometheusRegistry.Register(wsFactory.wfqRequestsGaugeVec)
-			if err != nil {
-				merr = multierr.Append(merr, err)
-			}
-			err = prometheusRegistry.Register(wsFactory.incomingTokensCounterVec)
+			err := prometheusRegistry.Register(wsFactory.incomingTokensCounterVec)
 			if err != nil {
 				merr = multierr.Append(merr, err)
 			}
@@ -220,14 +195,6 @@ func newFactory(
 		OnStop: func(_ context.Context) error {
 			var merr error
 
-			if !prometheusRegistry.Unregister(wsFactory.wfqFlowsGaugeVec) {
-				err := fmt.Errorf("failed to unregister wfq_flows metric")
-				merr = multierr.Append(merr, err)
-			}
-			if !prometheusRegistry.Unregister(wsFactory.wfqRequestsGaugeVec) {
-				err := fmt.Errorf("failed to unregister wfq_requests metric")
-				merr = multierr.Append(merr, err)
-			}
 			if !prometheusRegistry.Unregister(wsFactory.incomingTokensCounterVec) {
 				err := fmt.Errorf("failed to unregister incoming_tokens_total metric")
 				merr = multierr.Append(merr, err)
@@ -303,16 +270,6 @@ type SchedulerMetrics struct {
 
 // NewSchedulerMetrics creates a new SchedulerMetrics instance.
 func (wsFactory *Factory) NewSchedulerMetrics(metricLabels prometheus.Labels) (*SchedulerMetrics, error) {
-	wfqFlowsGauge, err := wsFactory.wfqFlowsGaugeVec.GetMetricWith(metricLabels)
-	if err != nil {
-		return nil, fmt.Errorf("%w: failed to get wfq flows gauge", err)
-	}
-
-	wfqRequestsGauge, err := wsFactory.wfqRequestsGaugeVec.GetMetricWith(metricLabels)
-	if err != nil {
-		return nil, fmt.Errorf("%w: failed to get wfq requests gauge", err)
-	}
-
 	incomingTokensCounter, err := wsFactory.incomingTokensCounterVec.GetMetricWith(metricLabels)
 	if err != nil {
 		return nil, err
@@ -329,8 +286,6 @@ func (wsFactory *Factory) NewSchedulerMetrics(metricLabels prometheus.Labels) (*
 	}
 
 	wfqMetrics := &scheduler.WFQMetrics{
-		FlowsGauge:                     wfqFlowsGauge,
-		HeapRequestsGauge:              wfqRequestsGauge,
 		IncomingTokensCounter:          incomingTokensCounter,
 		AcceptedTokensCounter:          acceptedTokensCounter,
 		RejectedTokensCounter:          rejectedTokensCounter,
@@ -351,16 +306,7 @@ func (wsFactory *Factory) NewSchedulerMetrics(metricLabels prometheus.Labels) (*
 func (sm *SchedulerMetrics) Delete() error {
 	var merr error
 
-	// Remove metrics from metric vectors
-	deleted := sm.wsFactory.wfqFlowsGaugeVec.Delete(sm.metricLabels)
-	if !deleted {
-		merr = multierr.Append(merr, errors.New("failed to delete wfq_flows gauge from its metric vector"))
-	}
-	deleted = sm.wsFactory.wfqRequestsGaugeVec.Delete(sm.metricLabels)
-	if !deleted {
-		merr = multierr.Append(merr, errors.New("failed to delete wfq_requests gauge from its metric vector"))
-	}
-	deleted = sm.wsFactory.incomingTokensCounterVec.Delete(sm.metricLabels)
+	deleted := sm.wsFactory.incomingTokensCounterVec.Delete(sm.metricLabels)
 	if !deleted {
 		merr = multierr.Append(merr, errors.New("failed to delete incoming_tokens_total counter from its metric vector"))
 	}
@@ -615,7 +561,14 @@ func (s *Scheduler) Decide(ctx context.Context, labels labels.Labels) (*flowcont
 		reqCtx = timeoutCtx
 	}
 
-	req := scheduler.NewRequest(matchedWorkloadLabel, tokens, invPriority)
+	var fairnessLabel string
+	if s.proto.FairnessLabelKey != "" {
+		if val, ok := labels.Get(s.proto.FairnessLabelKey); ok {
+			fairnessLabel = val
+		}
+	}
+
+	req := scheduler.NewRequest(matchedWorkloadLabel, fairnessLabel, tokens, invPriority)
 
 	accepted, remaining, current, reqID := s.scheduler.Schedule(reqCtx, req)
 

pkg/scheduler/request.go

@@ -2,14 +2,16 @@ package scheduler
 
 // Request is metadata for request in a flow that is to be allowed or dropped based on controlled delay and queue limits.
 type Request struct {
+	WorkloadLabel string  // for identifying workload
 	FairnessLabel string  // for enforcing fairness
 	Tokens        float64 // tokens (e.g. expected latency or complexity) for this request
 	InvPriority   float64 // larger values represent higher priority
 }
 
 // NewRequest calculates the inverse priority and returns a new Request.
-func NewRequest(fairnessLabel string, tokens float64, invPriority float64) *Request {
+func NewRequest(workloadLabel, fairnessLabel string, tokens float64, invPriority float64) *Request {
 	return &Request{
+		WorkloadLabel: workloadLabel,
 		FairnessLabel: fairnessLabel,
 		Tokens:        tokens,
 		InvPriority:   invPriority,