feat: support custom metrics (#2480)

Signed-off-by: ashing <[email protected]>

Author: ronething

go.mod

@@ -26,6 +26,8 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	pkgmetrics "github.com/apache/apisix-ingress-controller/pkg/metrics"
 )
 
 const UpdateChannelBufferSize = 1000
@@ -110,6 +112,8 @@ func (u *UpdateHandler) Start(ctx context.Context) error {
 		case <-ctx.Done():
 			return nil
 		case update := <-u.updateChannel:
+			// Decrement queue length after removing item from queue
+			pkgmetrics.DecStatusQueueLength()
 			u.log.Info("received a status update", "namespace", update.NamespacedName.Namespace,
 				"name", update.NamespacedName.Name)
 
@@ -137,4 +141,6 @@ type UpdateWriter struct {
 func (u *UpdateWriter) Update(update Update) {
 	u.wg.Wait()
 	u.updateChannel <- update
+	// Increment queue length after adding new item
+	pkgmetrics.IncStatusQueueLength()
 }

go.sum

@@ -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/controller/status/updater.go

@@ -43,6 +43,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 {
@@ -389,6 +390,8 @@ func (d *adcClient) sync(ctx context.Context, task Task) error {
 		return nil
 	}
 
+	var errs types.ADCExecutionErrors
+
 	// for global rules, we need to list all global rules and set it to the task resources
 	if slices.Contains(task.ResourceTypes, "global_rule") {
 		for _, config := range task.configs {
@@ -410,42 +413,90 @@ func (d *adcClient) sync(ctx context.Context, task Task) error {
 			task.Resources.GlobalRules = globalrule
 			log.Debugw("syncing resources global rules", zap.Any("globalRules", task.Resources.GlobalRules))
 
+			fileIOStart := time.Now()
 			syncFilePath, cleanup, err := prepareSyncFile(task.Resources)
 			if err != nil {
+				pkgmetrics.RecordFileIODuration("prepare_sync_file", "failure", time.Since(fileIOStart).Seconds())
 				return err
 			}
+			pkgmetrics.RecordFileIODuration("prepare_sync_file", "success", time.Since(fileIOStart).Seconds())
 			defer cleanup()
 
 			args := BuildADCExecuteArgs(syncFilePath, task.Labels, task.ResourceTypes)
 
-			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()
+
+			status := "success"
+			if err != nil {
+				status = "failure"
 				log.Errorw("failed to execute adc command", zap.Error(err), zap.Any("config", config))
-				return err
+
+				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)
 		}
 
 		return nil
 	}
 
+	// Record file I/O duration
+	fileIOStart := time.Now()
 	// every task resources is the same, so we can use the first config to prepare the sync file
 	syncFilePath, cleanup, err := prepareSyncFile(task.Resources)
 	if err != nil {
+		pkgmetrics.RecordFileIODuration("prepare_sync_file", "failure", time.Since(fileIOStart).Seconds())
 		return err
 	}
+	pkgmetrics.RecordFileIODuration("prepare_sync_file", "success", time.Since(fileIOStart).Seconds())
 	defer cleanup()
 
 	args := BuildADCExecuteArgs(syncFilePath, task.Labels, task.ResourceTypes)
 
-	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()
+
+		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
 	}

internal/manager/run.go

@@ -0,0 +1,114 @@
+// 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"},
+	)
+
+	// Status update channel queue length gauge
+	StatusUpdateQueueLength = prometheus.NewGauge(
+		prometheus.GaugeOpts{
+			Name: "apisix_ingress_status_update_queue_length",
+			Help: "Current length of the status update queue",
+		},
+	)
+
+	// File I/O operation duration histogram
+	FileIODuration = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Name:    "apisix_ingress_file_io_duration_seconds",
+			Help:    "Time spent on file I/O operations",
+			Buckets: prometheus.DefBuckets,
+		},
+		[]string{"operation", "status"},
+	)
+)
+
+// 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,
+		StatusUpdateQueueLength,
+		FileIODuration,
+	)
+}
+
+// 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()
+}
+
+// UpdateStatusQueueLength updates the status update queue length gauge
+func UpdateStatusQueueLength(length float64) {
+	StatusUpdateQueueLength.Set(length)
+}
+
+// IncStatusQueueLength increments the status update queue length gauge by 1
+func IncStatusQueueLength() {
+	StatusUpdateQueueLength.Inc()
+}
+
+// DecStatusQueueLength decrements the status update queue length gauge by 1
+func DecStatusQueueLength() {
+	StatusUpdateQueueLength.Dec()
+}
+
+// RecordFileIODuration records the duration of a file I/O operation
+func RecordFileIODuration(operation, status string, duration float64) {
+	FileIODuration.WithLabelValues(operation, status).Observe(duration)
+}

internal/provider/adc/adc.go

@@ -20,6 +20,7 @@ package apisix
 import (
 	"context"
 	"fmt"
+	"io"
 	"net"
 	"net/http"
 	"time"
@@ -96,6 +97,37 @@ spec:
 			err := s.DeleteResource("ApisixRoute", "default")
 			Expect(err).ShouldNot(HaveOccurred(), "deleting ApisixRoute")
 			Eventually(request).WithArguments("/headers").WithTimeout(8 * time.Second).ProbeEvery(time.Second).Should(Equal(http.StatusNotFound))
+
+			By("request /metrics endpoint from controller")
+
+			// Get the metrics service endpoint
+			metricsURL := s.GetMetricsEndpoint()
+
+			By("verify metrics content")
+			resp, err := http.Get(metricsURL)
+			Expect(err).ShouldNot(HaveOccurred(), "request metrics endpoint")
+			defer func() {
+				_ = resp.Body.Close()
+			}()
+
+			Expect(resp.StatusCode).Should(Equal(http.StatusOK))
+
+			body, err := io.ReadAll(resp.Body)
+			Expect(err).ShouldNot(HaveOccurred(), "read metrics response")
+
+			bodyStr := string(body)
+
+			// Verify prometheus format
+			Expect(resp.Header.Get("Content-Type")).Should(ContainSubstring("text/plain; version=0.0.4; charset=utf-8"))
+
+			// Verify specific metrics from metrics.go exist
+			Expect(bodyStr).Should(ContainSubstring("apisix_ingress_adc_sync_duration_seconds"))
+			Expect(bodyStr).Should(ContainSubstring("apisix_ingress_adc_sync_total"))
+			Expect(bodyStr).Should(ContainSubstring("apisix_ingress_status_update_queue_length"))
+			Expect(bodyStr).Should(ContainSubstring("apisix_ingress_file_io_duration_seconds"))
+
+			// Log metrics for debugging
+			fmt.Printf("Metrics endpoint response:\n%s\n", bodyStr)
 		})
 
 		It("Test plugins in ApisixRoute", func() {

pkg/metrics/metrics.go

@@ -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

test/e2e/apisix/route.go

@@ -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/framework/manifests/apisix.yaml

@@ -444,3 +444,12 @@ func NewClient(scheme, host string) *httpexpect.Expect {
 		),
 	})
 }
+
+func (s *Scaffold) GetMetricsEndpoint() string {
+	tunnel := k8s.NewTunnel(s.kubectlOptions, k8s.ResourceTypeService, "apisix-ingress-controller-manager-metrics-service", 8080, 8080)
+	if err := tunnel.ForwardPortE(s.t); err != nil {
+		return ""
+	}
+	s.addFinalizers(tunnel.Close)
+	return fmt.Sprintf("http://%s/metrics", tunnel.Endpoint())
+}