retry failed nakadi requests and respect the rate limits (#862)

Signed-off-by: Belmin Bedak <belmin.bedak@zalando.de>
Co-authored-by: Belmin Bedak <belmin.bedak@zalando.de>

.github/workflows/ci.yaml

@@ -10,8 +10,8 @@ jobs:
   tests:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v5
-    - uses: actions/setup-go@v6.0.0
+    - uses: actions/checkout@v6.0.2
+    - uses: actions/setup-go@v6.2.0
       with:
         go-version: '^1.25'
     - run: go version

.github/workflows/codeql-analysis.yml

@@ -38,16 +38,16 @@ jobs:
 
     steps:
     - name: Checkout repository
-      uses: actions/checkout@v5
+      uses: actions/checkout@v6.0.2
 
     - name: setup go
-      uses: actions/setup-go@v6.0.0
+      uses: actions/setup-go@v6.2.0
       with:
         go-version: '1.25'
 
     # Initializes the CodeQL tools for scanning.
     - name: Initialize CodeQL
-      uses: github/codeql-action/init@v3
+      uses: github/codeql-action/init@v4
       with:
         languages: ${{ matrix.language }}
         # If you wish to specify custom queries, you can do so here or in a config file.
@@ -61,7 +61,7 @@ jobs:
     # Autobuild attempts to build any compiled languages  (C/C++, C#, or Java).
     # If this step fails, then you should remove it and run the build manually (see below)
     - name: Autobuild
-      uses: github/codeql-action/autobuild@v3
+      uses: github/codeql-action/autobuild@v4
 
     # ℹ️ Command-line programs to run using the OS shell.
     # 📚 See https://docs.github.com/en/actions/using-workflows/workflow-syntax-for-github-actions#jobsjob_idstepsrun
@@ -74,4 +74,4 @@ jobs:
     #   ./location_of_script_within_repo/buildscript.sh
 
     - name: Perform CodeQL Analysis
-      uses: github/codeql-action/analyze@v3
+      uses: github/codeql-action/analyze@v4

.github/workflows/gh-packages.yaml

@@ -25,15 +25,15 @@ jobs:
       packages: write # to push packages
     steps:
       - name: Checkout
-        uses: actions/checkout@ff7abcd0c3c05ccf6adc123a8cd1fd4fb30fb493
+        uses: actions/checkout@0c366fd6a839edf440554fa01a7085ccba70ac98
 
-      - uses: actions/setup-go@44694675825211faa026b3c33043df3e48a5fa00
+      - uses: actions/setup-go@a5f9b05d2d216f63e13859e0d847461041025775
         with:
           # https://www.npmjs.com/package/semver#caret-ranges-123-025-004
           go-version: '^1.25'
 
       - name: Login to Github Container Registry
-        uses: docker/login-action@184bdaa0721073962dff0199f1fb9940f07167d1
+        uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9
         with:
           registry: ${{ env.REGISTRY }}
           username: ${{ github.actor }}
@@ -47,21 +47,21 @@ jobs:
           make build.linux.amd64 build.linux.arm64
 
       - name: Set up QEMU
-        uses: docker/setup-qemu-action@29109295f81e9208d7d86ff1c6c12d2833863392
+        uses: docker/setup-qemu-action@c7c53464625b32c7a7e944ae62b3e17d2b600130
 
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@e468171a9de216ec08956ac3ada2f0791b6bd435
+        uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f
 
       - name: Login to GitHub Container Registry
         if: github.event_name != 'pull_request'
-        uses: docker/login-action@184bdaa0721073962dff0199f1fb9940f07167d1
+        uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9
         with:
           registry: ghcr.io
           username: ${{ github.actor }}
           password: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Docker meta
-        uses: docker/metadata-action@c1e51972afc2121e065aed6d45c65596fe445f3f
+        uses: docker/metadata-action@c299e40c65443455700f0fdfc63efafe5b349051
         id: meta
         with:
           images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}

.github/workflows/release-chart.yml

@@ -10,7 +10,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout
-        uses: actions/checkout@v5
+        uses: actions/checkout@v6.0.2
         with:
           fetch-depth: 0
 

docs/helm/templates/custom-metrics-apiservice.yaml

@@ -3,6 +3,13 @@ apiVersion: apiregistration.k8s.io/v1
 kind: APIService
 metadata:
   name: v1beta1.custom.metrics.k8s.io
+  {{- with .Values.customMetricsApi.annotations }}
+  annotations:
+    {{- range $k, $v := . }}
+      {{- $value := $v | quote }}
+      {{- printf "%s: %s" (tpl $k $) (tpl $value $) | nindent 4 }}
+    {{- end }}
+  {{- end }}
 spec:
   service:
     name: kube-metrics-adapter
@@ -13,3 +20,4 @@ spec:
   groupPriorityMinimum: 100
   versionPriority: 100
 {{- end}}
+""

docs/helm/templates/external-metrics-apiservice.yaml

@@ -3,6 +3,13 @@ apiVersion: apiregistration.k8s.io/v1
 kind: APIService
 metadata:
   name: v1beta1.external.metrics.k8s.io
+  {{- with .Values.externalMetricsApi.annotations }}
+  annotations:
+    {{- range $k, $v := . }}
+      {{- $value := $v | quote }}
+      {{- printf "%s: %s" (tpl $k $) (tpl $value $) | nindent 4 }}
+    {{- end }}
+  {{- end }}
 spec:
   service:
     name: kube-metrics-adapter

docs/helm/values.yaml

@@ -15,7 +15,19 @@ addDirectoryHeader:
 contentionProfiling:
 profiling:
 enableCustomMetricsApi: true
+
+# -- add annotations to the custom metrics apiservice resource, e.g.
+# cert-manager.io/inject-ca-from: {{.Release.Namespace}}/kube-metrics-adapter to inject the CA certificate form a cert-manager-issued cert used for the deployment
+customMetricsApi:
+  annotations: {}
+
 enableExternalMetricsApi: true
+
+# -- add annotations to the custom metrics apiservice resource, e.g.
+# cert-manager.io/inject-ca-from: {{.Release.Namespace}}/kube-metrics-adapter to inject the CA certificate form a cert-manager-issued cert used for the deployment
+externalMetricsApi:
+  annotations: {}
+
 credentialsDirectory:
 disregardIncompatibleHPAs:
 http2MaxStreamsPerConnection:

go.mod

@@ -1,32 +1,31 @@
 module github.com/zalando-incubator/kube-metrics-adapter
 
 require (
-	github.com/argoproj/argo-rollouts v1.8.3
-	github.com/aws/aws-sdk-go-v2 v1.39.0
-	github.com/aws/aws-sdk-go-v2/config v1.31.8
-	github.com/aws/aws-sdk-go-v2/service/sqs v1.42.5
+	github.com/argoproj/argo-rollouts v1.8.4
+	github.com/aws/aws-sdk-go-v2 v1.41.1
+	github.com/aws/aws-sdk-go-v2/config v1.32.7
+	github.com/aws/aws-sdk-go-v2/service/sqs v1.42.21
 	github.com/influxdata/influxdb-client-go v1.4.0
 	github.com/prometheus/client_golang v1.23.2
-	github.com/prometheus/common v0.66.1
-	github.com/sirupsen/logrus v1.9.3
-	github.com/spf13/cobra v1.10.1
+	github.com/prometheus/common v0.67.5
+	github.com/sirupsen/logrus v1.9.4
+	github.com/spf13/cobra v1.10.2
 	github.com/spyzhov/ajson v0.9.6
 	github.com/stretchr/testify v1.11.1
-	github.com/szuecs/routegroup-client v0.28.2
+	github.com/szuecs/routegroup-client v0.34.1
 	github.com/zalando-incubator/cluster-lifecycle-manager v0.0.0-20250912104308-4489de744ee3
-	golang.org/x/net v0.44.0
-	golang.org/x/oauth2 v0.31.0
-	golang.org/x/sync v0.17.0
-	k8s.io/api v0.34.1
-	k8s.io/apimachinery v0.34.1
-	k8s.io/apiserver v0.34.1
-	k8s.io/client-go v0.34.1
-	k8s.io/component-base v0.34.1
+	golang.org/x/oauth2 v0.35.0
+	golang.org/x/sync v0.19.0
+	k8s.io/api v0.34.4
+	k8s.io/apimachinery v0.34.4
+	k8s.io/apiserver v0.34.4
+	k8s.io/client-go v0.34.4
+	k8s.io/component-base v0.34.4
 	k8s.io/klog v1.0.0
-	k8s.io/kube-openapi v0.0.0-20250710124328-f3f2b991d03b
-	k8s.io/metrics v0.33.5
-	k8s.io/utils v0.0.0-20250604170112-4c0f3b243397
-	sigs.k8s.io/custom-metrics-apiserver v1.33.0
+	k8s.io/kube-openapi v0.0.0-20250910181357-589584f1c912
+	k8s.io/metrics v0.34.4
+	k8s.io/utils v0.0.0-20251002143259-bc988d571ff4
+	sigs.k8s.io/custom-metrics-apiserver v1.34.0
 )
 
 require (
@@ -40,23 +39,25 @@ require (
 	github.com/andybalholm/brotli v1.1.0 // indirect
 	github.com/antlr4-go/antlr/v4 v4.13.0 // indirect
 	github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect
-	github.com/aws/aws-sdk-go-v2/credentials v1.18.12 // indirect
-	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.7 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.7 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.7 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.1 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.7 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sso v1.29.3 // indirect
-	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.34.4 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sts v1.38.4 // indirect
-	github.com/aws/smithy-go v1.23.0 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.19.7 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.17 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.17 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.17 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.17 // indirect
+	github.com/aws/aws-sdk-go-v2/service/signin v1.0.5 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.30.9 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.13 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.41.6 // indirect
+	github.com/aws/smithy-go v1.24.0 // indirect
 	github.com/aymerick/douceur v0.2.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/blang/semver/v4 v4.0.0 // indirect
 	github.com/bytedance/sonic v1.11.9 // indirect
 	github.com/bytedance/sonic/loader v0.1.1 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
+	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudwego/base64x v0.1.4 // indirect
 	github.com/cloudwego/iasm v0.2.0 // indirect
@@ -64,7 +65,7 @@ require (
 	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/deepmap/oapi-codegen v1.16.3 // indirect
-	github.com/emicklei/go-restful/v3 v3.12.2 // indirect
+	github.com/emicklei/go-restful/v3 v3.13.0 // indirect
 	github.com/fatih/color v1.18.0 // indirect
 	github.com/fatih/structs v1.1.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
@@ -168,23 +169,25 @@ require (
 	go.opentelemetry.io/otel/sdk v1.34.0 // indirect
 	go.opentelemetry.io/otel/trace v1.35.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.5.0 // indirect
+	go.uber.org/automaxprocs v1.6.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.0 // indirect
-	go.yaml.in/yaml/v2 v2.4.2 // indirect
+	go.yaml.in/yaml/v2 v2.4.3 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/arch v0.8.0 // indirect
-	golang.org/x/crypto v0.42.0 // indirect
+	golang.org/x/crypto v0.46.0 // indirect
 	golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 // indirect
-	golang.org/x/mod v0.28.0 // indirect
-	golang.org/x/sys v0.36.0 // indirect
-	golang.org/x/term v0.35.0 // indirect
-	golang.org/x/text v0.29.0 // indirect
+	golang.org/x/mod v0.30.0 // indirect
+	golang.org/x/net v0.48.0 // indirect
+	golang.org/x/sys v0.39.0 // indirect
+	golang.org/x/term v0.38.0 // indirect
+	golang.org/x/text v0.32.0 // indirect
 	golang.org/x/time v0.11.0 // indirect
-	golang.org/x/tools v0.37.0 // indirect
+	golang.org/x/tools v0.39.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20250303144028-a0af3efb3deb // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20250303144028-a0af3efb3deb // indirect
 	google.golang.org/grpc v1.72.1 // indirect
-	google.golang.org/protobuf v1.36.8 // indirect
+	google.golang.org/protobuf v1.36.11 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
@@ -192,10 +195,10 @@ require (
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/apiextensions-apiserver v0.34.0 // indirect
-	k8s.io/code-generator v0.34.1 // indirect
+	k8s.io/code-generator v0.34.4 // indirect
 	k8s.io/gengo/v2 v2.0.0-20250604051438-85fd79dbfd9f // indirect
 	k8s.io/klog/v2 v2.130.1 // indirect
-	k8s.io/kms v0.34.1 // indirect
+	k8s.io/kms v0.34.4 // indirect
 	sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.31.2 // indirect
 	sigs.k8s.io/controller-tools v0.19.0 // indirect
 	sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8 // indirect

pkg/api/generated/openapi/zz_generated.openapi.go

@@ -268,7 +268,6 @@ func GetOpenAPIDefinitions(ref common.ReferenceCallback) map[string]common.OpenA
 		"k8s.io/api/core/v1.WeightedPodAffinityTerm":                               schema_k8sio_api_core_v1_WeightedPodAffinityTerm(ref),
 		"k8s.io/api/core/v1.WindowsSecurityContextOptions":                         schema_k8sio_api_core_v1_WindowsSecurityContextOptions(ref),
 		"k8s.io/apimachinery/pkg/api/resource.Quantity":                            schema_apimachinery_pkg_api_resource_Quantity(ref),
-		"k8s.io/apimachinery/pkg/api/resource.int64Amount":                         schema_apimachinery_pkg_api_resource_int64Amount(ref),
 		"k8s.io/apimachinery/pkg/apis/meta/v1.APIGroup":                            schema_pkg_apis_meta_v1_APIGroup(ref),
 		"k8s.io/apimachinery/pkg/apis/meta/v1.APIGroupList":                        schema_pkg_apis_meta_v1_APIGroupList(ref),
 		"k8s.io/apimachinery/pkg/apis/meta/v1.APIResource":                         schema_pkg_apis_meta_v1_APIResource(ref),
@@ -15167,15 +15166,12 @@ func schema_pkg_apis_meta_v1_InternalEvent(ref common.ReferenceCallback) common.
 					"Object": {
 						SchemaProps: spec.SchemaProps{
 							Description: "Object is:\n * If Type is Added or Modified: the new state of the object.\n * If Type is Deleted: the state of the object immediately before deletion.\n * If Type is Bookmark: the object (instance of a type being watched) where\n   only ResourceVersion field is set. On successful restart of watch from a\n   bookmark resourceVersion, client is guaranteed to not get repeat event\n   nor miss any events.\n * If Type is Error: *api.Status is recommended; other types may make sense\n   depending on context.",
-							Ref:         ref("k8s.io/apimachinery/pkg/runtime.Object"),
 						},
 					},
 				},
 				Required: []string{"Type", "Object"},
 			},
 		},
-		Dependencies: []string{
-			"k8s.io/apimachinery/pkg/runtime.Object"},
 	}
 }
 

pkg/collector/collector.go

@@ -3,6 +3,7 @@ package collector
 import (
 	"context"
 	"fmt"
+	"strings"
 	"time"
 
 	log "github.com/sirupsen/logrus"
@@ -182,6 +183,25 @@ type MetricTypeName struct {
 	Metric autoscalingv2.MetricIdentifier
 }
 
+func (m MetricTypeName) String() string {
+	str := fmt.Sprintf("%s/%s", m.Type, m.Metric.Name)
+	if m.Metric.Selector != nil && len(m.Metric.Selector.MatchLabels) > 0 {
+		str += " " + mapToString(m.Metric.Selector.MatchLabels)
+	}
+	return str
+}
+
+func mapToString(m map[string]string) string {
+	str := "{"
+	keyVals := make([]string, 0, len(m))
+	for k, v := range m {
+		keyVals = append(keyVals, fmt.Sprintf("%s=%s", k, v))
+	}
+	str += strings.Join(keyVals, ",")
+	str += "}"
+	return str
+}
+
 type CollectedMetric struct {
 	Type      autoscalingv2.MetricSourceType
 	Namespace string

pkg/collector/collector_test.go

@@ -30,6 +30,88 @@ func (c *mockCollector) Interval() time.Duration {
 	return 0
 }
 
+func TestMetricTypeName_String(t *testing.T) {
+	for _, tc := range []struct {
+		name          string
+		metricType    autoscalingv2.MetricSourceType
+		metricName    string
+		selector      *metav1.LabelSelector
+		expectedStr   string
+		shouldContain []string
+	}{
+		{
+			name:        "nil selector with PodsMetricSourceType",
+			metricType:  autoscalingv2.PodsMetricSourceType,
+			metricName:  "metric-name",
+			selector:    nil,
+			expectedStr: "Pods/metric-name",
+		},
+		{
+			name:        "nil selector with ObjectMetricSourceType",
+			metricType:  autoscalingv2.ObjectMetricSourceType,
+			metricName:  "metric-name",
+			selector:    nil,
+			expectedStr: "Object/metric-name",
+		},
+		{
+			name:        "nil selector with ExternalMetricSourceType",
+			metricType:  autoscalingv2.ExternalMetricSourceType,
+			metricName:  "metric-name",
+			selector:    nil,
+			expectedStr: "External/metric-name",
+		},
+		{
+			name:       "empty selector (non-nil but empty MatchLabels)",
+			metricType: autoscalingv2.ExternalMetricSourceType,
+			metricName: "metric-name",
+			selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{},
+			},
+			expectedStr: "External/metric-name",
+		},
+		{
+			name:       "selector with single MatchLabel",
+			metricType: autoscalingv2.ExternalMetricSourceType,
+			metricName: "metric-name",
+			selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{"key": "value"},
+			},
+			shouldContain: []string{"External/metric-name", "key=value"},
+		},
+		{
+			name:       "selector with multiple MatchLabels",
+			metricType: autoscalingv2.ExternalMetricSourceType,
+			metricName: "metric-name",
+			selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{"key1": "val1", "key2": "val2"},
+			},
+			shouldContain: []string{"External/metric-name", "key1=val1", "key2=val2"},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			mtn := MetricTypeName{
+				Type: tc.metricType,
+				Metric: autoscalingv2.MetricIdentifier{
+					Name:     tc.metricName,
+					Selector: tc.selector,
+				},
+			}
+
+			result := mtn.String()
+
+			if tc.expectedStr != "" {
+				require.Equal(t, tc.expectedStr, result)
+			}
+
+			if len(tc.shouldContain) > 0 {
+				for _, substring := range tc.shouldContain {
+					require.Contains(t, result, substring, "result should contain %q", substring)
+				}
+			}
+		})
+	}
+}
+
 func TestNewCollector(t *testing.T) {
 	for _, tc := range []struct {
 		msg               string

pkg/collector/fake_collector_test.go

@@ -1,9 +1,9 @@
 package collector
 
 import (
+	"context"
 	"time"
 
-	"golang.org/x/net/context"
 	autoscalingv2 "k8s.io/api/autoscaling/v2"
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/metrics/pkg/apis/custom_metrics"

pkg/collector/zmon_collector.go

@@ -1,13 +1,13 @@
 package collector
 
 import (
+	"context"
 	"fmt"
 	"strconv"
 	"strings"
 	"time"
 
 	"github.com/zalando-incubator/kube-metrics-adapter/pkg/zmon"
-	"golang.org/x/net/context"
 	autoscalingv2 "k8s.io/api/autoscaling/v2"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

pkg/controller/scheduledscaling/scaler.go

@@ -1,10 +1,10 @@
 package scheduledscaling
 
 import (
+	"context"
 	"fmt"
 	"time"
 
-	"golang.org/x/net/context"
 	autoscalingv1 "k8s.io/api/autoscaling/v1"
 	autoscalingv2 "k8s.io/api/autoscaling/v2"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"

pkg/controller/scheduledscaling/scheduled_scaling.go

@@ -1,6 +1,7 @@
 package scheduledscaling
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"math"
@@ -10,7 +11,6 @@ import (
 	v1 "github.com/zalando-incubator/kube-metrics-adapter/pkg/apis/zalando.org/v1"
 	zalandov1 "github.com/zalando-incubator/kube-metrics-adapter/pkg/client/clientset/versioned/typed/zalando.org/v1"
 	"github.com/zalando-incubator/kube-metrics-adapter/pkg/recorder"
-	"golang.org/x/net/context"
 	"golang.org/x/sync/errgroup"
 	autoscalingv2 "k8s.io/api/autoscaling/v2"
 	corev1 "k8s.io/api/core/v1"
@@ -57,7 +57,7 @@ var (
 type now func() time.Time
 
 type scalingScheduleStore interface {
-	List() []interface{}
+	List() []any
 }
 
 type Controller struct {
@@ -325,7 +325,7 @@ func highestActiveSchedule(hpa *autoscalingv2.HorizontalPodAutoscaler, activeSch
 			continue
 		}
 
-		target := int64(metric.Object.Target.AverageValue.MilliValue() / 1000)
+		target := float64(metric.Object.Target.AverageValue.MilliValue()) / 1000.0
 		if target == 0 {
 			continue
 		}
@@ -338,10 +338,10 @@ func highestActiveSchedule(hpa *autoscalingv2.HorizontalPodAutoscaler, activeSch
 			value = activeSchedules[scheduleName]
 		}
 
-		expected := int64(math.Ceil(float64(value) / float64(target)))
+		expected := int64(math.Ceil(float64(value) / target))
 		if expected > highestExpected {
 			highestExpected = expected
-			usageRatio = float64(value) / (float64(target) * float64(currentReplicas))
+			usageRatio = float64(value) / (target * float64(currentReplicas))
 			highestObject = metric.Object.DescribedObject
 		}
 	}

pkg/controller/scheduledscaling/scheduled_scaling_test.go

@@ -336,35 +336,42 @@ func TestAdjustScaling(t *testing.T) {
 		msg             string
 		currentReplicas int32
 		desiredReplicas int32
-		targetValue     int64
+		targetValue     string
 		scheduleTarget  int64
 	}{
 		{
 			msg:             "current less than 10%% below desired (target 10000)",
 			currentReplicas: 28, // 7.1% increase to desired
 			desiredReplicas: 31,
-			targetValue:     333, // 10000/333 ~= 31
+			targetValue:     "333", // 10000/333 ~= 31
 			scheduleTarget:  10000,
 		},
 		{
 			msg:             "current less than 10%% below desired",
 			currentReplicas: 95, // 5.3% increase to desired
 			desiredReplicas: 100,
-			targetValue:     10, // 1000/10 = 100
+			targetValue:     "10", // 1000/10 = 100
 			scheduleTarget:  1000,
 		},
 		{
 			msg:             "current more than 10%% below desired, no adjustment",
 			currentReplicas: 90, // 11% increase to desired
 			desiredReplicas: 90,
-			targetValue:     10, // 1000/10 = 100
+			targetValue:     "10", // 1000/10 = 100
 			scheduleTarget:  1000,
 		},
 		{
 			msg:             "invalid HPA should not do any adjustment",
 			currentReplicas: 95,
 			desiredReplicas: 95,
-			targetValue:     0, // this is treated as invalid in the test, thus the HPA is ingored and no adjustment happens.
+			targetValue:     "", // this is treated as invalid in the test, thus the HPA is ingored and no adjustment happens.
+			scheduleTarget:  1000,
+		},
+		{
+			msg:             "handle milivalue as targetValue",
+			currentReplicas: 90, // 5.8% increase to desired
+			desiredReplicas: 96,
+			targetValue:     "10500m", // 1000/10.5 ~= 96
 			scheduleTarget:  1000,
 		},
 	} {
@@ -444,8 +451,11 @@ func TestAdjustScaling(t *testing.T) {
 				},
 			}
 
-			if tc.targetValue != 0 {
-				hpa.Spec.Metrics[0].Object.Target.AverageValue = resource.NewQuantity(tc.targetValue, resource.DecimalSI)
+			if tc.targetValue != "" {
+				quantity, err := resource.ParseQuantity(tc.targetValue)
+				require.NoError(t, err)
+
+				hpa.Spec.Metrics[0].Object.Target.AverageValue = &quantity
 			}
 
 			hpa, err = kubeClient.AutoscalingV2().HorizontalPodAutoscalers("default").Create(context.Background(), hpa, metav1.CreateOptions{})

pkg/nakadi/nakadi.go

@@ -8,6 +8,10 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"strconv"
+	"time"
+
+	"github.com/cenkalti/backoff/v5"
 )
 
 // Nakadi defines an interface for talking to the Nakadi API.
@@ -95,26 +99,40 @@ func (c *Client) subscriptions(ctx context.Context, filter *SubscriptionFilter,
 		endpoint.RawQuery = q.Encode()
 	}
 
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint.String(), nil)
-	if err != nil {
-		return nil, fmt.Errorf("[nakadi subscriptions] failed to create request: %w", err)
+	op := func() ([]byte, error) {
+		req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint.String(), nil)
+		if err != nil {
+			return nil, fmt.Errorf("[nakadi subscriptions] failed to create request: %w", err)
+		}
+
+		resp, err := c.http.Do(req)
+		if err != nil {
+			return nil, fmt.Errorf("[nakadi subscriptions] failed to make request: %w", err)
+		}
+		defer resp.Body.Close()
+
+		b, err := io.ReadAll(resp.Body)
+		if err != nil {
+			return nil, err
+		}
+
+		if err := checkResponseStatus(resp, b); err != nil {
+			return nil, fmt.Errorf("[nakadi subscriptions] %w", err)
+		}
+
+		return b, nil
 	}
 
-	resp, err := c.http.Do(req)
-	if err != nil {
-		return nil, fmt.Errorf("[nakadi subscriptions] failed to make request: %w", err)
-	}
-	defer resp.Body.Close()
-
-	d, err := io.ReadAll(resp.Body)
+	d, err := backoff.Retry(
+		ctx,
+		op,
+		backoff.WithBackOff(exponentialBackoff()),
+		backoff.WithMaxTries(3),
+	)
 	if err != nil {
 		return nil, err
 	}
 
-	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("[nakadi subscriptions] unexpected response code: %d (%s)", resp.StatusCode, string(d))
-	}
-
 	var subscriptionsResp struct {
 		Items []struct {
 			ID string `json:"id"`
@@ -194,26 +212,40 @@ func (c *Client) stats(ctx context.Context, filter *SubscriptionFilter) ([]stats
 		q.Set("show_time_lag", "true")
 		endpoint.RawQuery = q.Encode()
 
-		req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint.String(), nil)
-		if err != nil {
-			return nil, fmt.Errorf("[nakadi stats] failed to create request: %w", err)
+		op := func() ([]byte, error) {
+			req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint.String(), nil)
+			if err != nil {
+				return nil, fmt.Errorf("[nakadi stats] failed to create request: %w", err)
+			}
+
+			resp, err := c.http.Do(req)
+			if err != nil {
+				return nil, fmt.Errorf("[nakadi stats] failed to make request: %w", err)
+			}
+			defer resp.Body.Close()
+
+			b, err := io.ReadAll(resp.Body)
+			if err != nil {
+				return nil, err
+			}
+
+			if err := checkResponseStatus(resp, b); err != nil {
+				return nil, fmt.Errorf("[nakadi stats] %w", err)
+			}
+
+			return b, nil
 		}
 
-		resp, err := c.http.Do(req)
-		if err != nil {
-			return nil, fmt.Errorf("[nakadi stats] failed to make request: %w", err)
-		}
-		defer resp.Body.Close()
-
-		d, err := io.ReadAll(resp.Body)
+		d, err := backoff.Retry(
+			ctx,
+			op,
+			backoff.WithBackOff(exponentialBackoff()),
+			backoff.WithMaxTries(3),
+		)
 		if err != nil {
 			return nil, err
 		}
 
-		if resp.StatusCode != http.StatusOK {
-			return nil, fmt.Errorf("[nakadi stats] unexpected response code: %d (%s)", resp.StatusCode, string(d))
-		}
-
 		var result statsResp
 		err = json.Unmarshal(d, &result)
 		if err != nil {
@@ -229,3 +261,35 @@ func (c *Client) stats(ctx context.Context, filter *SubscriptionFilter) ([]stats
 
 	return stats, nil
 }
+
+func checkResponseStatus(resp *http.Response, b []byte) error {
+	if resp.StatusCode == http.StatusTooManyRequests {
+		h := resp.Header.Get("Retry-After")
+		if h == "" {
+			return fmt.Errorf("unexpected response code: %d (%s)", resp.StatusCode, string(b))
+
+		}
+		sec, err := strconv.ParseInt(h, 10, 32)
+		if err != nil {
+			return backoff.Permanent(err)
+		}
+		return backoff.RetryAfter(int(sec))
+	}
+	if resp.StatusCode >= http.StatusBadRequest &&
+		resp.StatusCode < http.StatusInternalServerError {
+		return backoff.Permanent(
+			fmt.Errorf("non-retryable response code: %d (%s)", resp.StatusCode, string(b)),
+		)
+	}
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("unexpected response code: %d (%s)", resp.StatusCode, string(b))
+	}
+	return nil
+}
+
+func exponentialBackoff() backoff.BackOff {
+	b := backoff.NewExponentialBackOff()
+	b.MaxInterval = time.Second * 30
+	b.InitialInterval = time.Millisecond * 100
+	return b
+}

pkg/nakadi/nakadi_test.go

@@ -5,8 +5,10 @@ import (
 	"errors"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"testing"
 
+	"github.com/cenkalti/backoff/v5"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -210,11 +212,70 @@ func TestQuery(tt *testing.T) {
 
 			nakadiClient := NewNakadiClient(ts.URL, client)
 			consumerLagSeconds, err := nakadiClient.ConsumerLagSeconds(context.Background(), ti.subscriptionFilter)
-			assert.Equal(t, ti.err, err)
+			assertErrorMessage(t, ti.err, err)
 			assert.Equal(t, ti.consumerLagSeconds, consumerLagSeconds)
 			unconsumedEvents, err := nakadiClient.UnconsumedEvents(context.Background(), ti.subscriptionFilter)
-			assert.Equal(t, ti.err, err)
+			assertErrorMessage(t, ti.err, err)
 			assert.Equal(t, ti.unconsumedEvents, unconsumedEvents)
 		})
 	}
 }
+
+func Test_checkResponseStatus(tt *testing.T) {
+	for _, ti := range []struct {
+		msg      string
+		resp     *http.Response
+		errCheck func(error) bool
+	}{
+		{
+			msg:  "nil when 200",
+			resp: &http.Response{StatusCode: http.StatusOK},
+		},
+		{
+			msg:  "backoff.Permanent when 4xx",
+			resp: &http.Response{StatusCode: http.StatusBadRequest},
+			errCheck: func(err error) bool {
+				var permanentErr *backoff.PermanentError
+				return errors.As(err, &permanentErr)
+			},
+		},
+		{
+			msg:  "unexpected error when 422 without Retry-After header",
+			resp: &http.Response{StatusCode: http.StatusTooManyRequests},
+			errCheck: func(err error) bool {
+				return strings.Contains(err.Error(), "unexpected response code")
+			},
+		},
+		{
+			msg: "backoff.RetryAfter when 422 with Retry-After header",
+			resp: &http.Response{
+				StatusCode: http.StatusTooManyRequests,
+				Header: http.Header{
+					"Retry-After": []string{"120"},
+				},
+			},
+			errCheck: func(err error) bool {
+				var retryAfterErr *backoff.RetryAfterError
+				return errors.As(err, &retryAfterErr)
+			},
+		},
+	} {
+		tt.Run(ti.msg, func(t *testing.T) {
+			err := checkResponseStatus(ti.resp, nil)
+			if ti.errCheck == nil {
+				assert.NoError(t, err)
+			} else {
+				assert.True(t, ti.errCheck(err))
+			}
+		})
+	}
+}
+
+func assertErrorMessage(t *testing.T, expected, actual error) {
+	t.Helper()
+	if expected != nil {
+		assert.EqualError(t, actual, expected.Error())
+	} else {
+		assert.NoError(t, actual)
+	}
+}

pkg/provider/hpa.go

@@ -3,6 +3,7 @@ package provider
 import (
 	"context"
 	"errors"
+	"fmt"
 	"reflect"
 	"sync"
 	"time"
@@ -345,7 +346,7 @@ type CollectorScheduler struct {
 	sync.RWMutex
 }
 
-// NewCollectorScheudler initializes a new CollectorScheduler.
+// NewCollectorScheduler initializes a new CollectorScheduler.
 func NewCollectorScheduler(ctx context.Context, metricsc chan<- metricCollection) *CollectorScheduler {
 	return &CollectorScheduler{
 		ctx:        ctx,
@@ -375,14 +376,17 @@ func (t *CollectorScheduler) Add(resourceRef resourceReference, typeName collect
 	collectors[typeName] = cancel
 
 	// start runner for new collector
-	go collectorRunner(ctx, metricCollector, t.metricSink)
+	go collectorRunner(ctx, typeName, metricCollector, t.metricSink)
 }
 
 // collectorRunner runs a collector at the desirec interval. If the passed
 // context is canceled the collection will be stopped.
-func collectorRunner(ctx context.Context, collector collector.Collector, metricsc chan<- metricCollection) {
+func collectorRunner(ctx context.Context, typeName collector.MetricTypeName, collector collector.Collector, metricsc chan<- metricCollection) {
 	for {
 		values, err := collector.GetMetrics(ctx)
+		if err != nil {
+			err = fmt.Errorf("getting metrics for %s failed: %w", typeName, err)
+		}
 
 		metricsc <- metricCollection{
 			Values: values,

pkg/provider/metric_store_test.go

@@ -1,13 +1,13 @@
 package provider
 
 import (
+	"context"
 	"sort"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/require"
 	"github.com/zalando-incubator/kube-metrics-adapter/pkg/collector"
-	"golang.org/x/net/context"
 	autoscalingv2 "k8s.io/api/autoscaling/v2"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

pkg/server/start.go

@@ -70,6 +70,8 @@ func NewCommandStartAdapterServer(stopCh <-chan struct{}) *cobra.Command {
 		NakadiTokenName:                   "nakadi",
 		CredentialsDir:                    "/meta/credentials",
 		ExternalRPSMetricName:             "skipper_serve_host_duration_seconds_count",
+		KubeClientQPS:                     0.0, // 0.0 will use the 5.0 default in client-go
+		KubeClientBurst:                   0,   // 0 will use the 10 default in client-go
 	}
 
 	cmd := &cobra.Command{
@@ -153,6 +155,10 @@ func NewCommandStartAdapterServer(stopCh <-chan struct{}) *cobra.Command {
 		"The name of the metric that should be used to query prometheus for RPS per hostname.")
 	flags.BoolVar(&o.ExternalRPSMetrics, "external-rps-metrics", o.ExternalRPSMetrics, ""+
 		"whether to enable external RPS metric collector or not")
+	flags.Float64Var(&o.KubeClientQPS, "kube-client-qps", o.KubeClientQPS, ""+
+		"maximum queries per second (QPS) to the Kubernetes API server (increase for large clusters), defaults to 5")
+	flags.IntVar(&o.KubeClientBurst, "kube-client-burst", o.KubeClientBurst, ""+
+		"maximum burst for throttle to the Kubernetes API server (increase for large clusters, defaults to 10)")
 	return cmd
 }
 
@@ -199,6 +205,8 @@ func (o AdapterServerOptions) RunCustomMetricsAdapterServer(stopCh <-chan struct
 	}()
 
 	clientConfig.Timeout = defaultClientGOTimeout
+	clientConfig.QPS = float32(o.KubeClientQPS)
+	clientConfig.Burst = o.KubeClientBurst
 
 	client, err := kubernetes.NewForConfig(clientConfig)
 	if err != nil {
@@ -546,4 +554,8 @@ type AdapterServerOptions struct {
 	ExternalRPSMetrics bool
 	// Name of the Prometheus metric that stores RPS by hostname for external RPS metrics.
 	ExternalRPSMetricName string
+	// KubeClientQPS configures the maximum QPS to the Kubernetes API server
+	KubeClientQPS float64
+	// KubeClientBurst configures the maximum burst for throttle to the Kubernetes API server
+	KubeClientBurst int
 }