From 5a197b18853a413ffdaf7cfe9df653788a9a5769 Mon Sep 17 00:00:00 2001 From: mfordjody <11638005@qq.com> Date: Tue, 30 Jun 2026 07:51:40 +0800 Subject: [PATCH] Supplement and improve observable construction --- cli/cmd/dashboard.go | 22 +++ cli/cmd/dashboard_test.go | 29 ++++ .../kube/gateway/deployment_controller.go | 76 ++++++++- .../gateway/deployment_controller_test.go | 108 ++++++++++++- .../charts/dubbod/files/kube-gateway.yaml | 10 ++ .../charts/dubbod/templates/clusterrole.yaml | 1 + pkg/cni/iptables.go | 6 + pkg/cni/iptables_test.go | 1 + pkg/log/log_test.go | 4 +- pkg/monitoring/monitoring.go | 129 +++++++++++----- pkg/monitoring/monitoring_test.go | 84 +++++++++- samples/addons/grafana.yaml | 144 ++++++++++++++++++ samples/addons/opentelemetry.yaml | 138 +++++++++++++++++ samples/addons/prometheus.yaml | 63 ++++++++ 14 files changed, 761 insertions(+), 54 deletions(-) create mode 100644 samples/addons/opentelemetry.yaml diff --git a/cli/cmd/dashboard.go b/cli/cmd/dashboard.go index 2e062f064..c2b0847f3 100644 --- a/cli/cmd/dashboard.go +++ b/cli/cmd/dashboard.go @@ -102,6 +102,12 @@ func DashboardCmd(ctx cli.Context) *cobra.Command { } if dashboardHasTracing(files) { _, err = fmt.Fprintf(cmd.OutOrStdout(), "tracing: kubectl -n %s port-forward svc/tracing 16686:16686\n", dashboardNamespace) + if err != nil { + return err + } + } + if dashboardHasOpenTelemetry(files) { + _, err = fmt.Fprintf(cmd.OutOrStdout(), "opentelemetry: kubectl -n %s port-forward svc/opentelemetry-collector 4317:4317\n", dashboardNamespace) } return err }, @@ -137,6 +143,10 @@ func dashboardManifestFiles(path string) ([]string, error) { if _, err := os.Stat(tracing); err == nil { files = append(files, tracing) } + opentelemetry := filepath.Join(path, "opentelemetry.yaml") + if _, err := os.Stat(opentelemetry); err == nil { + files = append(files, opentelemetry) + } return files, nil } @@ -149,11 +159,23 @@ func dashboardHasTracing(files []string) bool { return false } +func dashboardHasOpenTelemetry(files []string) bool { + for _, file := range files { + if filepath.Base(file) == "opentelemetry.yaml" { + return true + } + } + return false +} + func dashboardWaitDeploymentNames(files []string) []string { names := []string{"prometheus", "grafana"} if dashboardHasTracing(files) { names = append(names, "tracing") } + if dashboardHasOpenTelemetry(files) { + names = append(names, "opentelemetry-collector") + } return names } diff --git a/cli/cmd/dashboard_test.go b/cli/cmd/dashboard_test.go index a087463ae..b44a963fc 100644 --- a/cli/cmd/dashboard_test.go +++ b/cli/cmd/dashboard_test.go @@ -58,6 +58,35 @@ func TestDashboardManifestFilesIncludesTracingWhenPresent(t *testing.T) { } } +func TestDashboardManifestFilesIncludesOpenTelemetryWhenPresent(t *testing.T) { + dir := t.TempDir() + for _, name := range []string{"prometheus.yaml", "grafana.yaml", "tracing.yaml", "opentelemetry.yaml"} { + if err := os.WriteFile(filepath.Join(dir, name), []byte("apiVersion: v1\nkind: ConfigMap\nmetadata:\n name: "+name+"\n"), 0o600); err != nil { + t.Fatal(err) + } + } + + got, err := dashboardManifestFiles(dir) + if err != nil { + t.Fatalf("dashboardManifestFiles() returned error: %v", err) + } + want := []string{ + filepath.Join(dir, "prometheus.yaml"), + filepath.Join(dir, "grafana.yaml"), + filepath.Join(dir, "tracing.yaml"), + filepath.Join(dir, "opentelemetry.yaml"), + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("dashboardManifestFiles() = %v, want %v", got, want) + } + if !dashboardHasOpenTelemetry(got) { + t.Fatalf("dashboardHasOpenTelemetry() = false, want true") + } + if names := dashboardWaitDeploymentNames(got); !reflect.DeepEqual(names, []string{"prometheus", "grafana", "tracing", "opentelemetry-collector"}) { + t.Fatalf("dashboardWaitDeploymentNames() = %v", names) + } +} + func TestReadManifestObjects(t *testing.T) { file := filepath.Join(t.TempDir(), "manifest.yaml") manifest := strings.Join([]string{ diff --git a/dubbod/discovery/pkg/config/kube/gateway/deployment_controller.go b/dubbod/discovery/pkg/config/kube/gateway/deployment_controller.go index 6b9362cec..85d0cd416 100644 --- a/dubbod/discovery/pkg/config/kube/gateway/deployment_controller.go +++ b/dubbod/discovery/pkg/config/kube/gateway/deployment_controller.go @@ -23,6 +23,7 @@ import ( "encoding/hex" "encoding/json" "fmt" + "math" "sort" "strconv" "strings" @@ -80,6 +81,10 @@ const ( serviceNodePortAnnotation = "gateway.dubbo.apache.org/node-port" xdsAddressAnnotation = "gateway.dubbo.apache.org/xds-address" otelEndpointAnnotation = "gateway.dubbo.apache.org/otel-endpoint" + otelSamplingAnnotation = "gateway.dubbo.apache.org/otel-sampling-percentage" + otelServiceNameAnnotation = "gateway.dubbo.apache.org/otel-service-name" + accessLogAnnotation = "gateway.dubbo.apache.org/access-log" + accessLogFormatAnnotation = "gateway.dubbo.apache.org/access-log-format" ) var builtinClasses = getBuiltinClasses() @@ -387,6 +392,7 @@ func (d *DeploymentController) configureGateway(log *dubbolog.Logger, gw gateway return err } + observability := observabilityConfigForGateway(gw) input := TemplateInput{ Gateway: &gw, GatewayClass: string(gw.Spec.GatewayClassName), @@ -402,7 +408,11 @@ func (d *DeploymentController) configureGateway(log *dubbolog.Logger, gw gateway SystemNamespace: d.systemNamespace, ClusterID: string(d.clusterID), DomainSuffix: d.domainSuffix(), - OtelEndpoint: strings.TrimSpace(gw.Annotations[otelEndpointAnnotation]), + OtelEndpoint: observability.OtelEndpoint, + OtelServiceName: observability.OtelServiceName, + OtelSampling: observability.OtelSampling, + AccessLog: observability.AccessLog, + AccessLogFormat: observability.AccessLogFormat, } log.Infof("desired dxgate deployment=%s/%s gatewayClass=%s serviceType=%s ports=%s image=%s", @@ -456,6 +466,65 @@ type TemplateInput struct { ClusterID string DomainSuffix string OtelEndpoint string + OtelServiceName string + OtelSampling string + AccessLog string + AccessLogFormat string +} + +type gatewayObservabilityConfig struct { + OtelEndpoint string + OtelServiceName string + OtelSampling string + AccessLog string + AccessLogFormat string +} + +func observabilityConfigForGateway(gw gateway.Gateway) gatewayObservabilityConfig { + return gatewayObservabilityConfig{ + OtelEndpoint: strings.TrimSpace(gw.Annotations[otelEndpointAnnotation]), + OtelServiceName: gatewayOtelServiceName(gw), + OtelSampling: gatewayOtelSampling(gw), + AccessLog: gatewayAccessLog(gw), + AccessLogFormat: gatewayAccessLogFormat(gw), + } +} + +func gatewayOtelServiceName(gw gateway.Gateway) string { + if value := strings.TrimSpace(gw.Annotations[otelServiceNameAnnotation]); value != "" { + return value + } + return fmt.Sprintf("dxgate.%s.%s", gw.Namespace, gw.Name) +} + +func gatewayOtelSampling(gw gateway.Gateway) string { + value := strings.TrimSpace(gw.Annotations[otelSamplingAnnotation]) + if value == "" { + return "100" + } + parsed, err := strconv.ParseFloat(value, 64) + if err != nil || math.IsNaN(parsed) || math.IsInf(parsed, 0) || parsed < 0 || parsed > 100 { + return "100" + } + return strconv.FormatFloat(parsed, 'f', -1, 64) +} + +func gatewayAccessLog(gw gateway.Gateway) string { + switch strings.ToLower(strings.TrimSpace(gw.Annotations[accessLogAnnotation])) { + case "false", "0", "no", "off": + return "false" + default: + return "true" + } +} + +func gatewayAccessLogFormat(gw gateway.Gateway) string { + switch strings.ToLower(strings.TrimSpace(gw.Annotations[accessLogFormatAnnotation])) { + case "json": + return "json" + default: + return "text" + } } type dxgateBootstrapConfig struct { @@ -1366,10 +1435,7 @@ func gatewayServiceTargetPort(gw gateway.Gateway) intstr.IntOrString { if port, ok := positiveIntAnnotation(gw, serviceTargetPortAnnotation); ok { return intstr.FromInt(port) } - if gw.Annotations[eastWestGatewayAnnotation] == "true" { - return intstr.FromInt(inject.ProxylessGRPCInboundPort) - } - return intstr.FromString("http") + return intstr.FromInt(inject.ProxylessGRPCInboundPort) } func gatewayServiceNodePort(gw gateway.Gateway) int32 { diff --git a/dubbod/discovery/pkg/config/kube/gateway/deployment_controller_test.go b/dubbod/discovery/pkg/config/kube/gateway/deployment_controller_test.go index da3cef09e..6d273c81e 100644 --- a/dubbod/discovery/pkg/config/kube/gateway/deployment_controller_test.go +++ b/dubbod/discovery/pkg/config/kube/gateway/deployment_controller_test.go @@ -440,7 +440,7 @@ func TestDeploymentControllerBuildDxgateBootstrapConfigUsesXDSAddressAnnotation( } } -func TestExtractServicePortsTargetsDxgateContainerPorts(t *testing.T) { +func TestExtractServicePortsTargetsGRPCInbound(t *testing.T) { gw := gatewayv1.Gateway{ Spec: gatewayv1.GatewaySpec{ Listeners: []gatewayv1.Listener{ @@ -454,7 +454,7 @@ func TestExtractServicePortsTargetsDxgateContainerPorts(t *testing.T) { if len(ports) != 1 { t.Fatalf("expected only the http listener port, got %#v", ports) } - if ports[0].Name != "http" || ports[0].Port != 8080 || ports[0].TargetPort.String() != "http" { + if ports[0].Name != "http" || ports[0].Port != 8080 || ports[0].TargetPort.IntValue() != 15080 { t.Fatalf("unexpected http service port: %#v", ports[0]) } } @@ -498,6 +498,85 @@ func TestServiceTypeForGatewayUsesAnnotation(t *testing.T) { } } +func TestObservabilityConfigForGatewayDefaults(t *testing.T) { + cfg := observabilityConfigForGateway(gatewayv1.Gateway{ + ObjectMeta: metav1.ObjectMeta{Name: "public", Namespace: "app"}, + }) + + if cfg.OtelEndpoint != "" { + t.Fatalf("otel endpoint = %q, want empty", cfg.OtelEndpoint) + } + if cfg.OtelServiceName != "dxgate.app.public" { + t.Fatalf("otel service name = %q", cfg.OtelServiceName) + } + if cfg.OtelSampling != "100" { + t.Fatalf("otel sampling = %q", cfg.OtelSampling) + } + if cfg.AccessLog != "true" { + t.Fatalf("access log = %q", cfg.AccessLog) + } + if cfg.AccessLogFormat != "text" { + t.Fatalf("access log format = %q", cfg.AccessLogFormat) + } +} + +func TestObservabilityConfigForGatewayAnnotations(t *testing.T) { + cfg := observabilityConfigForGateway(gatewayv1.Gateway{ + ObjectMeta: metav1.ObjectMeta{ + Name: "public", + Namespace: "app", + Annotations: map[string]string{ + otelEndpointAnnotation: " http://otel-collector.dubbo-system.svc:4317 ", + otelServiceNameAnnotation: "edge-gateway", + otelSamplingAnnotation: "12.5", + accessLogAnnotation: "false", + accessLogFormatAnnotation: "json", + }, + }, + }) + + if cfg.OtelEndpoint != "http://otel-collector.dubbo-system.svc:4317" { + t.Fatalf("otel endpoint = %q", cfg.OtelEndpoint) + } + if cfg.OtelServiceName != "edge-gateway" { + t.Fatalf("otel service name = %q", cfg.OtelServiceName) + } + if cfg.OtelSampling != "12.5" { + t.Fatalf("otel sampling = %q", cfg.OtelSampling) + } + if cfg.AccessLog != "false" { + t.Fatalf("access log = %q", cfg.AccessLog) + } + if cfg.AccessLogFormat != "json" { + t.Fatalf("access log format = %q", cfg.AccessLogFormat) + } +} + +func TestObservabilityConfigForGatewayInvalidSamplingFallsBack(t *testing.T) { + for _, value := range []string{"abc", "-1", "101", "NaN", "+Inf"} { + cfg := observabilityConfigForGateway(gatewayv1.Gateway{ + ObjectMeta: metav1.ObjectMeta{ + Name: "public", + Namespace: "app", + Annotations: map[string]string{ + otelSamplingAnnotation: value, + accessLogAnnotation: "maybe", + accessLogFormatAnnotation: "yaml", + }, + }, + }) + if cfg.OtelSampling != "100" { + t.Fatalf("otel sampling for %q = %q, want 100", value, cfg.OtelSampling) + } + if cfg.AccessLog != "true" { + t.Fatalf("access log for %q = %q, want true", value, cfg.AccessLog) + } + if cfg.AccessLogFormat != "text" { + t.Fatalf("access log format for %q = %q, want text", value, cfg.AccessLogFormat) + } + } +} + func TestGetDefaultNameUsesFixedDxgateGatewayName(t *testing.T) { spec := &gatewayv1.GatewaySpec{GatewayClassName: "dubbo"} if got := getDefaultName("httpbin-gateway", spec, false); got != "dxgate-gateway" { @@ -537,7 +616,7 @@ func TestKubeGatewayTemplateRendersDxgateResources(t *testing.T) { }, DeploymentName: "public-dubbo", ServiceAccount: "public-dubbo", - Ports: []corev1.ServicePort{{Name: "http", Port: 80, TargetPort: intstr.FromString("http")}}, + Ports: []corev1.ServicePort{{Name: "http", Port: 80, TargetPort: intstr.FromInt(15080)}}, ServiceType: corev1.ServiceTypeLoadBalancer, Revision: "default", BootstrapConfig: "{\n \"xds_address\": \"http://dubbod.dubbo-system.svc:26010\",\n \"listener_names\": [\n \"public-dubbo.app.svc.cluster.local:80\"\n ],\n \"cluster_id\": \"Kubernetes\",\n \"dns_domain\": \"cluster.local\"\n}\n", @@ -546,6 +625,10 @@ func TestKubeGatewayTemplateRendersDxgateResources(t *testing.T) { SystemNamespace: "dubbo-system", ClusterID: "Kubernetes", DomainSuffix: "cluster.local", + OtelServiceName: "dxgate.app.public", + OtelSampling: "100", + AccessLog: "true", + AccessLogFormat: "text", } rendered, err := controller.render("gateway", input) if err != nil { @@ -586,11 +669,26 @@ func TestKubeGatewayTemplateRendersDxgateResources(t *testing.T) { if strings.Contains(rendered[2], "DXGATE_OTEL_ENDPOINT") { t.Fatalf("deployment rendered OTEL endpoint without annotation:\n%s", rendered[2]) } + for _, want := range []string{ + "DXGATE_GATEWAY_NAME", + `value: "public"`, + "DXGATE_OTEL_SERVICE_NAME", + `value: "dxgate.app.public"`, + "DXGATE_OTEL_SAMPLING_PERCENTAGE", + `value: "100"`, + "DXGATE_ACCESS_LOG", + "DXGATE_ACCESS_LOG_FORMAT", + `value: "text"`, + } { + if !strings.Contains(rendered[2], want) { + t.Fatalf("deployment did not render observability env %q:\n%s", want, rendered[2]) + } + } if !strings.Contains(rendered[2], "app.kubernetes.io/instance: public-dubbo") { t.Fatalf("deployment did not render stable dxgate instance label:\n%s", rendered[2]) } - if !strings.Contains(rendered[3], "targetPort: http") { - t.Fatalf("service did not target dxgate http port:\n%s", rendered[3]) + if !strings.Contains(rendered[3], "targetPort: 15080") { + t.Fatalf("service did not target grpc-inbound port:\n%s", rendered[3]) } if !strings.Contains(rendered[3], `proxyless.dubbo.apache.org/inject: "false"`) { t.Fatalf("service did not opt out of proxyless targetPort rewriting:\n%s", rendered[3]) diff --git a/manifests/charts/dubbod/files/kube-gateway.yaml b/manifests/charts/dubbod/files/kube-gateway.yaml index d09629049..6c10353a9 100644 --- a/manifests/charts/dubbod/files/kube-gateway.yaml +++ b/manifests/charts/dubbod/files/kube-gateway.yaml @@ -84,10 +84,20 @@ spec: value: 0.0.0.0:8080 - name: DXGATE_ADMIN_ADDR value: 0.0.0.0:26021 + - name: DXGATE_GATEWAY_NAME + value: {{ .Gateway.Name | quote }} + - name: DXGATE_OTEL_SERVICE_NAME + value: {{ .OtelServiceName | quote }} + - name: DXGATE_OTEL_SAMPLING_PERCENTAGE + value: {{ .OtelSampling | quote }} {{- if .OtelEndpoint }} - name: DXGATE_OTEL_ENDPOINT value: {{ .OtelEndpoint | quote }} {{- end }} + - name: DXGATE_ACCESS_LOG + value: {{ .AccessLog | quote }} + - name: DXGATE_ACCESS_LOG_FORMAT + value: {{ .AccessLogFormat | quote }} - name: POD_NAME valueFrom: fieldRef: diff --git a/manifests/charts/dubbod/templates/clusterrole.yaml b/manifests/charts/dubbod/templates/clusterrole.yaml index 5e5250701..4f0afd13d 100644 --- a/manifests/charts/dubbod/templates/clusterrole.yaml +++ b/manifests/charts/dubbod/templates/clusterrole.yaml @@ -62,6 +62,7 @@ rules: - gateways - gatewayclasses - httproutes + - backendtlspolicies verbs: - get - list diff --git a/pkg/cni/iptables.go b/pkg/cni/iptables.go index 9829b8b5c..bad4eec52 100644 --- a/pkg/cni/iptables.go +++ b/pkg/cni/iptables.go @@ -26,6 +26,7 @@ import ( const ( meshInboundChain = "DUBBO-GRPC-INBOUND" meshPodIPSet = "DUBBO-GRPC-INBOUND-PODS" + dxgateAdminPort = 26021 ) type CommandRunner interface { @@ -98,12 +99,17 @@ func (m *IPTablesRuleManager) ensureBase(ctx context.Context) error { } } allowGRPCInbound := []string{"-m", "set", "--match-set", meshPodIPSet, "dst", "-p", "tcp", "--dport", fmt.Sprint(m.grpcInboundPort), "-j", "RETURN"} + allowDxgateAdmin := []string{"-m", "set", "--match-set", meshPodIPSet, "dst", "-p", "tcp", "--dport", fmt.Sprint(dxgateAdminPort), "-j", "RETURN"} rejectOtherTCP := []string{"-m", "set", "--match-set", meshPodIPSet, "dst", "-p", "tcp", "-j", "REJECT"} m.deleteRepeated(ctx, allowGRPCInbound...) + m.deleteRepeated(ctx, allowDxgateAdmin...) m.deleteRepeated(ctx, rejectOtherTCP...) if err := m.appendRule(ctx, allowGRPCInbound...); err != nil { return err } + if err := m.appendRule(ctx, allowDxgateAdmin...); err != nil { + return err + } return m.appendRule(ctx, rejectOtherTCP...) } diff --git a/pkg/cni/iptables_test.go b/pkg/cni/iptables_test.go index 56cc67fcd..ea8af9f66 100644 --- a/pkg/cni/iptables_test.go +++ b/pkg/cni/iptables_test.go @@ -40,6 +40,7 @@ func TestIPTablesRuleManagerAddsGRPCInboundBoundaryRules(t *testing.T) { "-I FORWARD 1 -j DUBBO-GRPC-INBOUND", "-I OUTPUT 1 -j DUBBO-GRPC-INBOUND", "-A DUBBO-GRPC-INBOUND -m set --match-set DUBBO-GRPC-INBOUND-PODS dst -p tcp --dport 15080 -j RETURN", + "-A DUBBO-GRPC-INBOUND -m set --match-set DUBBO-GRPC-INBOUND-PODS dst -p tcp --dport 26021 -j RETURN", "-A DUBBO-GRPC-INBOUND -m set --match-set DUBBO-GRPC-INBOUND-PODS dst -p tcp -j REJECT", "ipset add DUBBO-GRPC-INBOUND-PODS 10.244.0.12 -exist", } { diff --git a/pkg/log/log_test.go b/pkg/log/log_test.go index 5114d30cf..97f474ef4 100644 --- a/pkg/log/log_test.go +++ b/pkg/log/log_test.go @@ -123,8 +123,8 @@ func TestLogMessagesMetricRecordsLevelAndScope(t *testing.T) { labels[label.GetName()] = label.GetValue() } if labels["level"] == "warn" && labels["scope"] == "log-metric-test" { - if metric.GetGauge().GetValue() < 1 { - t.Fatalf("dubbod_log_messages_total = %v, want >= 1", metric.GetGauge().GetValue()) + if metric.GetCounter().GetValue() < 1 { + t.Fatalf("dubbod_log_messages_total = %v, want >= 1", metric.GetCounter().GetValue()) } return } diff --git a/pkg/monitoring/monitoring.go b/pkg/monitoring/monitoring.go index 356032434..6f6bf3beb 100644 --- a/pkg/monitoring/monitoring.go +++ b/pkg/monitoring/monitoring.go @@ -138,7 +138,8 @@ type metric struct { description string valueType prometheus.ValueType labelNames []string - vec *prometheus.GaugeVec + counterVec *prometheus.CounterVec + gaugeVec *prometheus.GaugeVec vecOnce sync.Once counter prometheus.Counter gauge prometheus.Gauge @@ -160,36 +161,20 @@ func newMetric(name, description string, valueType prometheus.ValueType, opts .. // If labels are predefined, create the vector immediately if len(o.labels) > 0 { if valueType == prometheus.CounterValue { - m.vec = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + m.counterVec = prometheus.NewCounterVec(prometheus.CounterOpts{ Name: name, Help: description, }, o.labels) - } else { - m.vec = prometheus.NewGaugeVec(prometheus.GaugeOpts{ - Name: name, - Help: description, - }, o.labels) - } - registryLock.Lock() - registry.MustRegister(m.vec) - registryLock.Unlock() - } else { - // No predefined labels, create scalar metric - if valueType == prometheus.CounterValue { - m.counter = prometheus.NewCounter(prometheus.CounterOpts{ - Name: name, - Help: description, - }) registryLock.Lock() - registry.MustRegister(m.counter) + registry.MustRegister(m.counterVec) registryLock.Unlock() } else { - m.gauge = prometheus.NewGauge(prometheus.GaugeOpts{ + m.gaugeVec = prometheus.NewGaugeVec(prometheus.GaugeOpts{ Name: name, Help: description, - }) + }, o.labels) registryLock.Lock() - registry.MustRegister(m.gauge) + registry.MustRegister(m.gaugeVec) registryLock.Unlock() } } @@ -210,7 +195,32 @@ func (m *metric) Name() string { } func (m *metric) Record(value float64) { + if m.counterVec != nil || m.gaugeVec != nil { + return + } + m.vecOnce.Do(func() { + if m.valueType == prometheus.CounterValue { + m.counter = prometheus.NewCounter(prometheus.CounterOpts{ + Name: m.name, + Help: m.description, + }) + registryLock.Lock() + registry.MustRegister(m.counter) + registryLock.Unlock() + } else { + m.gauge = prometheus.NewGauge(prometheus.GaugeOpts{ + Name: m.name, + Help: m.description, + }) + registryLock.Lock() + registry.MustRegister(m.gauge) + registryLock.Unlock() + } + }) if m.counter != nil { + if value < 0 { + return + } m.counter.Add(value) } else if m.gauge != nil { if value >= 0 { @@ -227,14 +237,25 @@ func (m *metric) RecordInt(value int64) { func (m *metric) With(labelValues ...LabelValue) Metric { // If vector already exists (predefined labels), use it directly - if m.vec != nil { + if m.counterVec != nil { + labels := make(prometheus.Labels) + for _, lv := range labelValues { + labels[lv.Name] = lv.Value + } + return &labeledMetric{ + parent: m, + counter: m.counterVec.With(labels), + labels: labels, + } + } + if m.gaugeVec != nil { labels := make(prometheus.Labels) for _, lv := range labelValues { labels[lv.Name] = lv.Value } return &labeledMetric{ parent: m, - gauge: m.vec.With(labels), + gauge: m.gaugeVec.With(labels), labels: labels, } } @@ -247,20 +268,22 @@ func (m *metric) With(labelValues ...LabelValue) Metric { } if m.valueType == prometheus.CounterValue { - m.vec = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + m.counterVec = prometheus.NewCounterVec(prometheus.CounterOpts{ Name: m.name, Help: m.description, }, labelNames) + registryLock.Lock() + registry.MustRegister(m.counterVec) + registryLock.Unlock() } else { - m.vec = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + m.gaugeVec = prometheus.NewGaugeVec(prometheus.GaugeOpts{ Name: m.name, Help: m.description, }, labelNames) + registryLock.Lock() + registry.MustRegister(m.gaugeVec) + registryLock.Unlock() } - - registryLock.Lock() - registry.MustRegister(m.vec) - registryLock.Unlock() }) labels := make(prometheus.Labels) @@ -269,16 +292,32 @@ func (m *metric) With(labelValues ...LabelValue) Metric { } return &labeledMetric{ - parent: m, - gauge: m.vec.With(labels), - labels: labels, + parent: m, + counter: m.counterWith(labels), + gauge: m.gaugeWith(labels), + labels: labels, } } type labeledMetric struct { - parent *metric - gauge prometheus.Gauge - labels prometheus.Labels + parent *metric + counter prometheus.Counter + gauge prometheus.Gauge + labels prometheus.Labels +} + +func (m *metric) counterWith(labels prometheus.Labels) prometheus.Counter { + if m.counterVec == nil { + return nil + } + return m.counterVec.With(labels) +} + +func (m *metric) gaugeWith(labels prometheus.Labels) prometheus.Gauge { + if m.gaugeVec == nil { + return nil + } + return m.gaugeVec.With(labels) } func (lm *labeledMetric) Increment() { @@ -295,7 +334,10 @@ func (lm *labeledMetric) Name() string { func (lm *labeledMetric) Record(value float64) { if lm.parent.valueType == prometheus.CounterValue { - lm.gauge.Add(value) + if lm.counter == nil || value < 0 { + return + } + lm.counter.Add(value) } else { if value >= 0 { lm.gauge.Set(value) @@ -320,7 +362,18 @@ func (lm *labeledMetric) With(labelValues ...LabelValue) Metric { return &labeledMetric{ parent: lm.parent, - gauge: lm.parent.vec.With(newLabels), + counter: func() prometheus.Counter { + if lm.parent.counterVec == nil { + return nil + } + return lm.parent.counterVec.With(newLabels) + }(), + gauge: func() prometheus.Gauge { + if lm.parent.gaugeVec == nil { + return nil + } + return lm.parent.gaugeVec.With(newLabels) + }(), labels: newLabels, } } diff --git a/pkg/monitoring/monitoring_test.go b/pkg/monitoring/monitoring_test.go index 91bf05d21..c35541148 100644 --- a/pkg/monitoring/monitoring_test.go +++ b/pkg/monitoring/monitoring_test.go @@ -20,8 +20,54 @@ import ( "fmt" "testing" "time" + + dto "github.com/prometheus/client_model/go" ) +func TestNewSumWithoutLabelsExposesCounter(t *testing.T) { + name := fmt.Sprintf("test_sum_without_labels_total_%d", time.Now().UnixNano()) + metric := NewSum(name, "Test scalar counter.") + + metric.Increment() + metric.RecordInt(2) + + family := findMetricFamily(t, name) + if family.GetType() != dto.MetricType_COUNTER { + t.Fatalf("metric type = %s, want COUNTER", family.GetType()) + } + if got := family.GetMetric()[0].GetCounter().GetValue(); got != 3 { + t.Fatalf("counter value = %v, want 3", got) + } +} + +func TestNewSumWithPredeclaredLabelsExposesCounterVec(t *testing.T) { + phaseTag := CreateLabel("phase") + name := fmt.Sprintf("test_sum_with_labels_total_%d", time.Now().UnixNano()) + metric := NewSum(name, "Test labeled counter.", WithLabels("phase")) + + metric.With(phaseTag.Value("push")).Increment() + metric.With(phaseTag.Value("send")).RecordInt(2) + + assertCounterSeries(t, name, map[string]float64{ + "push": 1, + "send": 2, + }) +} + +func TestNewSumWithDynamicLabelsExposesCounterVec(t *testing.T) { + typeTag := CreateLabel("type") + name := fmt.Sprintf("test_sum_with_dynamic_labels_total_%d", time.Now().UnixNano()) + metric := NewSum(name, "Test dynamically labeled counter.") + + metric.With(typeTag.Value("cds")).Increment() + metric.With(typeTag.Value("eds")).RecordInt(4) + + assertCounterSeries(t, name, map[string]float64{ + "cds": 1, + "eds": 4, + }) +} + func TestDistributionWithLabelsRegistersOnce(t *testing.T) { phaseTag := CreateLabel("phase") name := fmt.Sprintf("test_distribution_with_labels_seconds_%d", time.Now().UnixNano()) @@ -35,17 +81,47 @@ func TestDistributionWithLabelsRegistersOnce(t *testing.T) { metric.With(phaseTag.Value("push")).Record(0.2) metric.With(phaseTag.Value("send")).Record(0.4) + family := findMetricFamily(t, name) + if got := len(family.GetMetric()); got != 2 { + t.Fatalf("metric series = %d, want 2", got) + } +} + +func findMetricFamily(t *testing.T, name string) *dto.MetricFamily { + t.Helper() families, err := GetRegistry().Gather() if err != nil { t.Fatalf("gather metrics: %v", err) } for _, family := range families { if family.GetName() == name { - if got := len(family.GetMetric()); got != 2 { - t.Fatalf("metric series = %d, want 2", got) - } - return + return family } } t.Fatalf("%s not gathered", name) + return nil +} + +func assertCounterSeries(t *testing.T, name string, want map[string]float64) { + t.Helper() + family := findMetricFamily(t, name) + if family.GetType() != dto.MetricType_COUNTER { + t.Fatalf("metric type = %s, want COUNTER", family.GetType()) + } + got := map[string]float64{} + for _, metric := range family.GetMetric() { + for _, label := range metric.GetLabel() { + if label.GetName() == "phase" || label.GetName() == "type" { + got[label.GetValue()] = metric.GetCounter().GetValue() + } + } + } + if len(got) != len(want) { + t.Fatalf("counter series = %#v, want %#v", got, want) + } + for label, value := range want { + if got[label] != value { + t.Fatalf("counter series %s = %v, want %v", label, got[label], value) + } + } } diff --git a/samples/addons/grafana.yaml b/samples/addons/grafana.yaml index 406eebd84..b9a56ef69 100644 --- a/samples/addons/grafana.yaml +++ b/samples/addons/grafana.yaml @@ -961,6 +961,150 @@ data: ], "title": "HTTP Upstream P95", "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "fieldConfig": { + "defaults": { + "custom": { + "drawStyle": "line", + "fillOpacity": 10, + "lineWidth": 1, + "showPoints": "never" + }, + "unit": "ms" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 8, + "x": 0, + "y": 49 + }, + "id": 23, + "options": { + "legend": { + "calcs": [ + "lastNotNull" + ], + "displayMode": "table", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "targets": [ + { + "expr": "histogram_quantile(0.99, sum by (le, route, cluster) (rate(dxgate_http_route_latency_ms_bucket[5m])))", + "legendFormat": "{{route}} / {{cluster}}", + "refId": "A" + } + ], + "title": "HTTP Upstream P99", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "fieldConfig": { + "defaults": { + "custom": { + "drawStyle": "line", + "fillOpacity": 10, + "lineWidth": 1, + "showPoints": "never" + }, + "unit": "ops" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 8, + "x": 8, + "y": 49 + }, + "id": 24, + "options": { + "legend": { + "calcs": [ + "lastNotNull" + ], + "displayMode": "table", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "targets": [ + { + "expr": "sum by (route, cluster, status_code) (rate(dxgate_http_route_requests_total{status_code=~\"5..\"}[5m]))", + "legendFormat": "{{route}} / {{cluster}} / {{status_code}}", + "refId": "A" + } + ], + "title": "HTTP 5xx Rate", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "fieldConfig": { + "defaults": { + "custom": { + "drawStyle": "line", + "fillOpacity": 10, + "lineWidth": 1, + "showPoints": "never" + }, + "unit": "ops" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 8, + "x": 16, + "y": 49 + }, + "id": 25, + "options": { + "legend": { + "calcs": [ + "lastNotNull" + ], + "displayMode": "table", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "targets": [ + { + "expr": "sum by (route, cluster, status_code) (rate(dxgate_http_route_requests_total{status_code=~\"401|403|429\"}[5m]))", + "legendFormat": "{{route}} / {{cluster}} / {{status_code}}", + "refId": "A" + } + ], + "title": "Policy Rejection Rate", + "type": "timeseries" } ], "refresh": "15s", diff --git a/samples/addons/opentelemetry.yaml b/samples/addons/opentelemetry.yaml new file mode 100644 index 000000000..cb22d6660 --- /dev/null +++ b/samples/addons/opentelemetry.yaml @@ -0,0 +1,138 @@ +# 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. + +apiVersion: v1 +kind: ServiceAccount +metadata: + name: opentelemetry-collector + namespace: dubbo-system + labels: + app.kubernetes.io/name: opentelemetry-collector + app.kubernetes.io/part-of: dubbo-observability +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: opentelemetry-collector + namespace: dubbo-system + labels: + app.kubernetes.io/name: opentelemetry-collector + app.kubernetes.io/part-of: dubbo-observability +data: + collector.yaml: | + receivers: + otlp: + protocols: + grpc: + endpoint: 0.0.0.0:4317 + http: + endpoint: 0.0.0.0:4318 + + processors: + batch: {} + + exporters: + debug: + verbosity: basic + + service: + pipelines: + traces: + receivers: [otlp] + processors: [batch] + exporters: [debug] + logs: + receivers: [otlp] + processors: [batch] + exporters: [debug] +--- +apiVersion: v1 +kind: Service +metadata: + name: opentelemetry-collector + namespace: dubbo-system + labels: + app.kubernetes.io/name: opentelemetry-collector + app.kubernetes.io/part-of: dubbo-observability +spec: + selector: + app.kubernetes.io/name: opentelemetry-collector + ports: + - name: grpc-otlp + port: 4317 + targetPort: grpc-otlp + protocol: TCP + - name: http-otlp + port: 4318 + targetPort: http-otlp + protocol: TCP +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: opentelemetry-collector + namespace: dubbo-system + labels: + app.kubernetes.io/name: opentelemetry-collector + app.kubernetes.io/part-of: dubbo-observability +spec: + replicas: 1 + selector: + matchLabels: + app.kubernetes.io/name: opentelemetry-collector + template: + metadata: + labels: + app.kubernetes.io/name: opentelemetry-collector + app.kubernetes.io/part-of: dubbo-observability + spec: + serviceAccountName: opentelemetry-collector + containers: + - name: collector + image: otel/opentelemetry-collector-contrib:0.128.0 + imagePullPolicy: IfNotPresent + args: + - --config=/etc/otelcol/collector.yaml + ports: + - name: grpc-otlp + containerPort: 4317 + protocol: TCP + - name: http-otlp + containerPort: 4318 + protocol: TCP + readinessProbe: + tcpSocket: + port: grpc-otlp + initialDelaySeconds: 5 + periodSeconds: 10 + livenessProbe: + tcpSocket: + port: grpc-otlp + initialDelaySeconds: 20 + periodSeconds: 15 + resources: + requests: + cpu: 100m + memory: 128Mi + limits: + memory: 512Mi + volumeMounts: + - name: config + mountPath: /etc/otelcol + readOnly: true + volumes: + - name: config + configMap: + name: opentelemetry-collector diff --git a/samples/addons/prometheus.yaml b/samples/addons/prometheus.yaml index 17f1d4597..4301ab58f 100644 --- a/samples/addons/prometheus.yaml +++ b/samples/addons/prometheus.yaml @@ -64,6 +64,9 @@ data: scrape_interval: 15s evaluation_interval: 15s + rule_files: + - /etc/prometheus/rules/*.yaml + scrape_configs: - job_name: prometheus static_configs: @@ -103,6 +106,60 @@ data: target_label: app --- apiVersion: v1 +kind: ConfigMap +metadata: + name: prometheus-rules + namespace: dubbo-system + labels: + app.kubernetes.io/name: prometheus + app.kubernetes.io/part-of: dubbo-observability +data: + dubbo-alerts.yaml: | + groups: + - name: dubbo-observability + rules: + - alert: Dxgate5xxRateHigh + expr: sum by (namespace, gateway, route, cluster) (rate(dxgate_http_route_requests_total{status_code=~"5.."}[5m])) > 0 + for: 5m + labels: + severity: warning + annotations: + summary: dxgate 5xx responses are increasing + description: dxgate is returning 5xx responses for one or more Gateway routes. + - alert: DxgateP95LatencyHigh + expr: histogram_quantile(0.95, sum by (le, namespace, gateway, route, cluster) (rate(dxgate_http_route_latency_ms_bucket[5m]))) > 1000 + for: 10m + labels: + severity: warning + annotations: + summary: dxgate P95 latency is above 1s + description: Gateway upstream latency is above the sample threshold. + - alert: DxgateNotReady + expr: up{app="dxgate"} == 0 + for: 2m + labels: + severity: critical + annotations: + summary: dxgate metrics target is down + description: Prometheus cannot scrape a dxgate pod or the pod is not ready. + - alert: DubbodXDSRejects + expr: sum(rate(dubbod_total_xds_rejects[5m])) > 0 + for: 5m + labels: + severity: warning + annotations: + summary: xDS rejects are increasing + description: One or more data-plane clients rejected xDS resources from dubbod. + - alert: DubbodSDSError + expr: sum(rate(dubbod_sds_certificate_errors_total[5m])) > 0 + for: 5m + labels: + severity: warning + annotations: + summary: SDS certificate errors are increasing + description: Workload certificate fetch or delivery is failing. +--- +apiVersion: v1 kind: Service metadata: name: prometheus @@ -173,11 +230,17 @@ spec: - name: config mountPath: /etc/prometheus readOnly: true + - name: rules + mountPath: /etc/prometheus/rules + readOnly: true - name: storage mountPath: /prometheus volumes: - name: config configMap: name: prometheus + - name: rules + configMap: + name: prometheus-rules - name: storage emptyDir: {}