This is an automated email from the ASF dual-hosted git repository.
mfordjody pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/dubbo-kubernetes.git
The following commit(s) were added to refs/heads/master by this push:
new d47c48d7 Supplement and improve observable construction (#931)
d47c48d7 is described below
commit d47c48d7fa012740d876371ff9f2628c5dc0dda5
Author: mfordjody <[email protected]>
AuthorDate: Tue Jun 30 07:57:00 2026 +0800
Supplement and improve observable construction (#931)
---
cli/cmd/dashboard.go | 22 ++++
cli/cmd/dashboard_test.go | 29 +++++
.../config/kube/gateway/deployment_controller.go | 76 ++++++++++-
.../kube/gateway/deployment_controller_test.go | 108 +++++++++++++++-
manifests/charts/dubbod/files/kube-gateway.yaml | 10 ++
manifests/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(-)
diff --git a/cli/cmd/dashboard.go b/cli/cmd/dashboard.go
index 2e062f06..c2b0847f 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 a087463a..b44a963f 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 6b9362ce..85d0cd41 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 da3cef09..6d273c81 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 d0962904..6c10353a 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 5e525070..4f0afd13 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 9829b8b5..bad4eec5 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 56cc67fc..ea8af9f6 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 5114d30c..97f474ef 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 35603243..6f6bf3be 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 91bf05d2..c3554114 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 406eebd8..b9a56ef6 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 00000000..cb22d666
--- /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 17f1d459..4301ab58 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: {}
