Support kpa/apa autoscaling workflow part I

.golangci.yml

@@ -41,4 +41,4 @@ linters:
     - typecheck
     - unconvert
 #    - unparam
-    - unused
+#    - unused

api/autoscaling/v1alpha1/podautoscaler_types.go

@@ -90,6 +90,8 @@ type MetricSource struct {
 	Endpoint string `json:"endpoint"`
 	// e.g. /api/metrics/cpu
 	Path string `json:"path"`
+	// e.g. kv_cache_utilization metrics
+	Name string `json:"metric"`
 }
 
 // PodAutoscalerStatus defines the observed state of PodAutoscaler

config/crd/bases/autoscaling.aibrix.ai_podautoscalers.yaml

@@ -57,11 +57,15 @@ spec:
                     endpoint:
                       description: e.g. service1.example.com
                       type: string
+                    metric:
+                      description: e.g. kv_cache_utilization metrics
+                      type: string
                     path:
                       description: e.g. /api/metrics/cpu
                       type: string
                   required:
                   - endpoint
+                  - metric
                   - path
                   type: object
                 type: array

pkg/controller/podautoscaler/algorithm/README.md

@@ -0,0 +1 @@
+placeholder

pkg/controller/podautoscaler/hpa_resources.go

@@ -17,7 +17,6 @@ limitations under the License.
 package podautoscaler
 
 import (
-	"context"
 	"fmt"
 	"math"
 	"strconv"
@@ -41,7 +40,7 @@ func getHPANameFromPa(pa *pav1.PodAutoscaler) string {
 }
 
 // MakeHPA creates an HPA resource from a PodAutoscaler resource.
-func MakeHPA(pa *pav1.PodAutoscaler, ctx context.Context) *autoscalingv2.HorizontalPodAutoscaler {
+func makeHPA(pa *pav1.PodAutoscaler) *autoscalingv2.HorizontalPodAutoscaler {
 	minReplicas, maxReplicas := pa.Spec.MinReplicas, pa.Spec.MaxReplicas
 	if maxReplicas == 0 {
 		maxReplicas = math.MaxInt32 // Set default to no upper limit if not specified

pkg/controller/podautoscaler/metrics/client.go

@@ -0,0 +1,109 @@
+/*
+Copyright 2024 The Aibrix Team.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package metrics
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+
+	"k8s.io/klog/v2"
+
+	autoscalingv2 "k8s.io/api/autoscaling/v2"
+	corev1 "k8s.io/api/core/v1"
+
+	"time"
+)
+
+const (
+	metricServerDefaultMetricWindow = time.Minute
+)
+
+// restMetricsClient is a client which supports fetching
+// metrics from the pod metrics prometheus API. In future,
+// it can fetch from the ai runtime api directly.
+type restMetricsClient struct {
+}
+
+func (r restMetricsClient) GetPodContainerMetric(ctx context.Context, metricName string, pod corev1.Pod, containerPort int) (PodMetricsInfo, time.Time, error) {
+	panic("not implemented")
+}
+
+func (r restMetricsClient) GetObjectMetric(ctx context.Context, metricName string, namespace string, objectRef *autoscalingv2.CrossVersionObjectReference, containerPort int) (PodMetricsInfo, time.Time, error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func GetMetricsFromPods(pods []corev1.Pod, metricName string, metricsPort int) ([]float64, error) {
+	metrics := make([]float64, 0, len(pods))
+
+	for _, pod := range pods {
+		// We should use the primary container port. In future, we can decide whether to use sidecar container's port
+		url := fmt.Sprintf("http://%s:%d/metrics", pod.Status.PodIP, metricsPort)
+
+		// scrape metrics
+		resp, err := http.Get(url)
+		if err != nil {
+			return nil, fmt.Errorf("failed to fetch metrics from pod %s: %v", pod.Name, err)
+		}
+		defer func() {
+			if err := resp.Body.Close(); err != nil {
+				// Handle the error here. For example, log it or take appropriate corrective action.
+				klog.InfoS("Error closing response body:", err)
+			}
+		}()
+		body, err := io.ReadAll(resp.Body)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read response from pod %s: %v", pod.Name, err)
+		}
+
+		metricValue, err := parseMetricFromBody(body, metricName)
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse metrics from pod %s: %v", pod.Name, err)
+		}
+
+		metrics = append(metrics, metricValue)
+	}
+
+	return metrics, nil
+}
+
+func parseMetricFromBody(body []byte, metricName string) (float64, error) {
+	lines := strings.Split(string(body), "\n")
+
+	for _, line := range lines {
+		if strings.Contains(line, metricName) {
+			// format is `http_requests_total 1234.56`
+			parts := strings.Fields(line)
+			if len(parts) < 2 {
+				return 0, fmt.Errorf("unexpected format for metric %s", metricName)
+			}
+
+			// parse to float64
+			value, err := strconv.ParseFloat(parts[len(parts)-1], 64)
+			if err != nil {
+				return 0, fmt.Errorf("failed to parse metric value for %s: %v", metricName, err)
+			}
+
+			return value, nil
+		}
+	}
+	return 0, fmt.Errorf("metrics %s not found", metricName)
+}

pkg/controller/podautoscaler/metrics/interface.go

@@ -0,0 +1,52 @@
+/*
+Copyright 2024 The Aibrix Team.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package metrics
+
+import (
+	"context"
+	"time"
+
+	v1 "k8s.io/api/core/v1"
+
+	autoscaling "k8s.io/api/autoscaling/v2"
+)
+
+// PodMetric contains pod metric value (the metric values are expected to be the metric as a milli-value)
+type PodMetric struct {
+	Timestamp       time.Time
+	Window          time.Duration
+	Value           int64
+	MetricsName     string
+	containerPort   int32
+	ScaleObjectName string
+}
+
+// PodMetricsInfo contains pod metrics as a map from pod names to PodMetricsInfo
+type PodMetricsInfo map[string]PodMetric
+
+// MetricsClient knows how to query a remote interface to retrieve container-level
+// resource metrics as well as pod-level arbitrary metrics
+type MetricsClient interface {
+	// GetPodContainerMetric gets the given resource metric (and an associated oldest timestamp)
+	// for the specified named container in specific pods in the given namespace and when
+	// the container is an empty string it returns the sum of all the container metrics.
+	GetPodContainerMetric(ctx context.Context, metricName string, pod v1.Pod, containerPort int) (PodMetricsInfo, time.Time, error)
+
+	// GetObjectMetric gets the given metric (and an associated timestamp) for the given
+	// object in the given namespace, it can be used to fetch any object metrics supports /scale interface
+	GetObjectMetric(ctx context.Context, metricName string, namespace string, objectRef *autoscaling.CrossVersionObjectReference, containerPort int) (PodMetricsInfo, time.Time, error)
+}

pkg/controller/podautoscaler/metrics/utilization.go

@@ -0,0 +1,64 @@
+/*
+Copyright 2024 The Aibrix Team.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package metrics
+
+import "fmt"
+
+// GetResourceUtilizationRatio takes in a set of metrics, a set of matching requests,
+// and a target utilization percentage, and calculates the ratio of
+// desired to actual utilization (returning that, the actual utilization, and the raw average value)
+func GetResourceUtilizationRatio(metrics PodMetricsInfo, requests map[string]int64, targetUtilization int32) (utilizationRatio float64, currentUtilization int32, rawAverageValue int64, err error) {
+	metricsTotal := int64(0)
+	requestsTotal := int64(0)
+	numEntries := 0
+
+	for podName, metric := range metrics {
+		request, hasRequest := requests[podName]
+		if !hasRequest {
+			// we check for missing requests elsewhere, so assuming missing requests == extraneous metrics
+			continue
+		}
+
+		metricsTotal += metric.Value
+		requestsTotal += request
+		numEntries++
+	}
+
+	// if the set of requests is completely disjoint from the set of metrics,
+	// then we could have an issue where the requests total is zero
+	if requestsTotal == 0 {
+		return 0, 0, 0, fmt.Errorf("no metrics returned matched known pods")
+	}
+
+	currentUtilization = int32((metricsTotal * 100) / requestsTotal)
+
+	return float64(currentUtilization) / float64(targetUtilization), currentUtilization, metricsTotal / int64(numEntries), nil
+}
+
+// GetMetricUsageRatio takes in a set of metrics and a target usage value,
+// and calculates the ratio of desired to actual usage
+// (returning that and the actual usage)
+func GetMetricUsageRatio(metrics PodMetricsInfo, targetUsage int64) (usageRatio float64, currentUsage int64) {
+	metricsTotal := int64(0)
+	for _, metric := range metrics {
+		metricsTotal += metric.Value
+	}
+
+	currentUsage = metricsTotal / int64(len(metrics))
+
+	return float64(currentUsage) / float64(targetUsage), currentUsage
+}