feat: support custom metrics

Makefile

@@ -145,6 +145,10 @@ conformance-test:
 conformance-test-standalone:
 	go test -v ./test/conformance/apisix -tags=conformance -timeout 60m
 
+.PHONY: load-test
+load-test:
+	go test -v ./test/e2e/load-test -test.timeout=$(TEST_TIMEOUT) -v -ginkgo.v
+
 .PHONY: lint
 lint: sort-import golangci-lint ## Run golangci-lint linter
 	$(GOLANGCI_LINT) run

go.mod

@@ -17,6 +17,7 @@ require (
 	github.com/onsi/ginkgo/v2 v2.20.0
 	github.com/onsi/gomega v1.34.1
 	github.com/pkg/errors v0.9.1
+	github.com/prometheus/client_golang v1.19.1
 	github.com/samber/lo v1.47.0
 	github.com/spf13/cobra v1.8.1
 	github.com/stretchr/testify v1.10.0
@@ -148,7 +149,6 @@ require (
 	github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/pquerna/otp v1.4.0 // indirect
-	github.com/prometheus/client_golang v1.19.1 // indirect
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.55.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect

internal/controller/status/updater.go

@@ -28,7 +28,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-const UpdateChannelBufferSize = 1000
+const UpdateChannelBufferSize = 5000
 
 type Update struct {
 	NamespacedName types.NamespacedName
@@ -136,5 +136,6 @@ type UpdateWriter struct {
 
 func (u *UpdateWriter) Update(update Update) {
 	u.wg.Wait()
+	// len(u.updateChannel)
 	u.updateChannel <- update
 }

internal/manager/run.go

@@ -42,6 +42,7 @@ import (
 	"github.com/apache/apisix-ingress-controller/internal/controller/config"
 	"github.com/apache/apisix-ingress-controller/internal/controller/status"
 	"github.com/apache/apisix-ingress-controller/internal/provider/adc"
+	_ "github.com/apache/apisix-ingress-controller/pkg/metrics"
 )
 
 var (

internal/provider/adc/adc.go

@@ -42,6 +42,7 @@ import (
 	"github.com/apache/apisix-ingress-controller/internal/provider/adc/translator"
 	"github.com/apache/apisix-ingress-controller/internal/types"
 	"github.com/apache/apisix-ingress-controller/internal/utils"
+	pkgmetrics "github.com/apache/apisix-ingress-controller/pkg/metrics"
 )
 
 type adcConfig struct {
@@ -311,9 +312,13 @@ func (d *adcClient) Start(ctx context.Context) error {
 	for {
 		select {
 		case <-ticker.C:
+			start := time.Now()
+			log.Infof("adcClient start sync, %v", start)
 			if err := d.Sync(ctx); err != nil {
 				log.Error(err)
 			}
+			duration := time.Since(start)
+			log.Infof("adcClient sync duration: %v, now: %v", duration, time.Now())
 		case <-ctx.Done():
 			return nil
 		}
@@ -390,14 +395,35 @@ func (d *adcClient) sync(ctx context.Context, task Task) error {
 
 	var errs types.ADCExecutionErrors
 	for _, config := range task.configs {
-		if err := d.executor.Execute(ctx, d.BackendMode, config, args); err != nil {
+		// Record sync duration for each config
+		startTime := time.Now()
+		resourceType := strings.Join(task.ResourceTypes, ",")
+		if resourceType == "" {
+			resourceType = "all"
+		}
+
+		err := d.executor.Execute(ctx, d.BackendMode, config, args)
+		duration := time.Since(startTime).Seconds()
+		log.Infof("synced %s in %f seconds, service list length: %d", config.Name, duration, len(task.Resources.Services))
+
+		status := "success"
+		if err != nil {
+			status = "failure"
 			log.Errorw("failed to execute adc command", zap.Error(err), zap.Any("config", config))
+
 			var execErr types.ADCExecutionError
 			if errors.As(err, &execErr) {
 				errs.Errors = append(errs.Errors, execErr)
+				pkgmetrics.RecordExecutionError(config.Name, execErr.Name)
+			} else {
+				pkgmetrics.RecordExecutionError(config.Name, "unknown")
 			}
 		}
+
+		// Record metrics
+		pkgmetrics.RecordSyncDuration(config.Name, resourceType, status, duration)
 	}
+
 	if len(errs.Errors) > 0 {
 		return errs
 	}

pkg/metrics/metrics.go

@@ -0,0 +1,89 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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 (
+	"github.com/prometheus/client_golang/prometheus"
+	"sigs.k8s.io/controller-runtime/pkg/metrics"
+)
+
+var (
+	// ADC sync operation duration histogram
+	ADCSyncDuration = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Name:    "apisix_ingress_adc_sync_duration_seconds",
+			Help:    "Time spent on ADC sync operations",
+			Buckets: prometheus.DefBuckets,
+		},
+		[]string{"config_name", "resource_type", "status"},
+	)
+
+	// ADC sync operation counter
+	ADCSyncTotal = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "apisix_ingress_adc_sync_total",
+			Help: "Total number of ADC sync operations",
+		},
+		[]string{"config_name", "resource_type", "status"},
+	)
+
+	// ADC execution errors counter
+	ADCExecutionErrors = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "apisix_ingress_adc_execution_errors_total",
+			Help: "Total number of ADC execution errors",
+		},
+		[]string{"config_name", "error_type"},
+	)
+
+	// Resource sync gauge
+	ResourceSyncGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "apisix_ingress_resources_synced",
+			Help: "Number of resources currently synced",
+		},
+		[]string{"config_name", "resource_type"},
+	)
+)
+
+// init registers all metrics with the global prometheus registry
+func init() {
+	// Register metrics with controller-runtime's metrics registry
+	metrics.Registry.MustRegister(
+		ADCSyncDuration,
+		ADCSyncTotal,
+		ADCExecutionErrors,
+		ResourceSyncGauge,
+	)
+}
+
+// RecordSyncDuration records the duration of an ADC sync operation
+func RecordSyncDuration(configName, resourceType, status string, duration float64) {
+	ADCSyncDuration.WithLabelValues(configName, resourceType, status).Observe(duration)
+	ADCSyncTotal.WithLabelValues(configName, resourceType, status).Inc()
+}
+
+// RecordExecutionError records an ADC execution error
+func RecordExecutionError(configName, errorType string) {
+	ADCExecutionErrors.WithLabelValues(configName, errorType).Inc()
+}
+
+// UpdateResourceGauge updates the resource sync gauge
+func UpdateResourceGauge(configName, resourceType string, count float64) {
+	ResourceSyncGauge.WithLabelValues(configName, resourceType).Set(count)
+}

test/e2e/framework/apisix_consts.go

@@ -27,7 +27,8 @@ import (
 )
 
 var (
-	ProviderType = cmp.Or(os.Getenv("PROVIDER_TYPE"), "apisix")
+	ProviderType       = cmp.Or(os.Getenv("PROVIDER_TYPE"), "apisix-standalone")
+	ProviderSyncPeriod = cmp.Or(os.Getenv("PROVIDER_SYNC_PERIOD"), "5s")
 )
 
 var (

test/e2e/framework/manifests/apisix.yaml

@@ -90,6 +90,9 @@ spec:
             - name: admin
               containerPort: 9180
               protocol: TCP
+            - name: control
+              containerPort: 9090
+              protocol: TCP
           volumeMounts:
             - name: config-writable
               mountPath: /usr/local/apisix/conf
@@ -130,3 +133,18 @@ spec:
   selector:
     app.kubernetes.io/name: apisix 
   type: {{ .ServiceType | default "NodePort" }}
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: apisix-control-api
+  labels:
+    app.kubernetes.io/name: apisix-control-api
+spec:
+  ports:
+    - port: 9090
+      name: control
+      protocol: TCP
+      targetPort: 9090
+  selector:
+    app.kubernetes.io/name: apisix

test/e2e/framework/manifests/ingress.yaml

@@ -346,10 +346,10 @@ metadata:
   namespace: {{ .Namespace }}
 spec:
   ports:
-  - name: https
-    port: 8443
+  - name: metrics
+    port: 8080
     protocol: TCP
-    targetPort: 8443
+    targetPort: 8080
   selector:
     control-plane: controller-manager
 ---
@@ -399,19 +399,16 @@ spec:
           initialDelaySeconds: 15
           periodSeconds: 20
         name: manager
+        ports:
+        - name: metrics
+          containerPort: 8080
+          protocol: TCP
         readinessProbe:
           httpGet:
             path: /readyz
             port: 8081
           initialDelaySeconds: 5
           periodSeconds: 10
-        resources:
-          limits:
-            cpu: 500m
-            memory: 128Mi
-          requests:
-            cpu: 10m
-            memory: 64Mi
         securityContext:
           allowPrivilegeEscalation: false
           capabilities:

test/e2e/load-test/e2e_test.go

@@ -0,0 +1,43 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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 load_test
+
+import (
+	"fmt"
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"github.com/apache/apisix-ingress-controller/test/e2e/framework"
+	"github.com/apache/apisix-ingress-controller/test/e2e/scaffold"
+)
+
+// Run long-term-stability tests using Ginkgo runner.
+func TestLongTermStability(t *testing.T) {
+	RegisterFailHandler(Fail)
+	var f = framework.NewFramework()
+	_ = f
+
+	scaffold.NewDeployer = func(s *scaffold.Scaffold) scaffold.Deployer {
+		return scaffold.NewAPISIXDeployer(s)
+	}
+
+	_, _ = fmt.Fprintf(GinkgoWriter, "Starting load-test suite\n")
+	RunSpecs(t, "long-term-stability suite")
+}