This is an automated email from the ASF dual-hosted git repository.
github-actions[bot] 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 277dc0b3 Validation of observable metrics (#1019)
277dc0b3 is described below
commit 277dc0b3a7d16f214e5b3d2b7024b601a2e96550
Author: mfordjody <[email protected]>
AuthorDate: Sat Aug 15 14:23:32 2026 +0800
Validation of observable metrics (#1019)
* fix: refresh Inherent runtime config on Telemetry changes
* feat: support required Telemetry standard metrics
* fix: make CNI upgrades converge
* test: validate application standard metrics end to end
* feat: automate single-container application metrics
---
cni/main.go | 6 +
cni/pkg/nodeagent/iptables.go | 36 +++-
cni/pkg/nodeagent/iptables_test.go | 3 +
dubbod/discovery/docker/dockerfile.dubbod | 4 +-
.../pkg/bootstrap/inherent_grpc_controller_test.go | 30 ++++
dubbod/discovery/pkg/bootstrap/server.go | 2 +
dubbod/discovery/pkg/bootstrap/server_test.go | 10 ++
go.mod | 14 +-
go.sum | 12 ++
manifests/charts/base/files/crd-all.gen.yaml | 7 +-
manifests/charts/dubbod/files/grpc-engine.yaml | 29 ----
.../charts/dubbod/templates/cni-daemonset.yaml | 3 +
manifests/charts/dubbod/templates/cni-rbac.yaml | 2 +-
pkg/config/validation/validators.go | 10 ++
pkg/config/validation/validators_test.go | 16 ++
pkg/kube/inject/inherent.go | 4 +
pkg/kube/inject/inherent_test.go | 122 ++++++--------
pkg/kube/inject/service_test.go | 174 +++----------------
pkg/kube/inject/webhook.go | 129 +++++---------
samples/addons/grafana.yaml | 187 +++++++++++++++++++++
tests/e2e/activationapp/go.mod | 3 +-
tests/e2e/activationapp/go.sum | 2 +
tests/e2e/activationapp/main.go | 53 ++++++
tests/e2e/testdata/telemetry-application.yaml | 49 ++++++
24 files changed, 542 insertions(+), 365 deletions(-)
diff --git a/cni/main.go b/cni/main.go
index be11f788..4d0cccbb 100644
--- a/cni/main.go
+++ b/cni/main.go
@@ -121,6 +121,12 @@ func runInstall(args []string) error {
IPSetPath: opts.IPSetPath,
StateDir: opts.StateDir,
}
+ // The cluster source reads the host kubeconfig written by Install.
Build it
+ // only after the first installation; InstallLoop repeats this
operation to
+ // refresh the projected service-account credentials.
+ if err := nodeagent.Install(ctx, opts); err != nil {
+ return err
+ }
cluster := clusterSource(opts, nodeName)
go func() {
_ = nodeagent.ReconcileLoop(ctx,
diff --git a/cni/pkg/nodeagent/iptables.go b/cni/pkg/nodeagent/iptables.go
index 200c3885..b4b609f8 100644
--- a/cni/pkg/nodeagent/iptables.go
+++ b/cni/pkg/nodeagent/iptables.go
@@ -26,10 +26,12 @@ import (
)
const (
- meshInboundChain = "DUBBO-GRPC-INBOUND"
- meshPodIPSet = "DUBBO-GRPC-INBOUND-PODS"
- meshExcludeIPSet = "DUBBO-GRPC-INBOUND-EXCLUDE"
- dxgateAdminPort = 26021
+ meshInboundChain = "DUBBO-GRPC-INBOUND"
+ meshPodIPSet = "DUBBO-GRPC-INBOUND-PODS"
+ meshExcludeIPSet = "DUBBO-GRPC-INBOUND-EXCLUDE"
+ legacyInboundChain = "DUBBO-XSERVER-INBOUND"
+ legacyPodIPSet = "DUBBO-XSERVER-PODS"
+ dxgateAdminPort = 26021
// dxproxyAdminPort carries the inbound sidecar's health, readiness and
// metrics endpoints. kubelet probes it from the host, so the fence has
to
// exempt it or every injected pod fails its readiness probe. Workloads
that
@@ -320,11 +322,16 @@ func (m *IPTablesRuleManager) ensureBase(ctx
context.Context) error {
// Reject only new inbound connections. Established replies to outbound
// control-plane traffic still target a managed Pod and must not be
fenced.
rejectOtherTCP := []string{"-m", "set", "--match-set", meshPodIPSet,
"dst", "-p", "tcp", "-m", "conntrack", "--ctstate", "NEW", "-j", "REJECT"}
+ // Older releases installed an unconditional reject before the allow
rules.
+ // Remove it during every reconciliation so upgrades converge without
+ // requiring an operator to flush the chain manually.
+ legacyRejectOtherTCP := []string{"-m", "set", "--match-set",
meshPodIPSet, "dst", "-p", "tcp", "-j", "REJECT"}
m.deleteRepeated(ctx, allowExcluded...)
m.deleteRepeated(ctx, allowGRPCInbound...)
m.deleteRepeated(ctx, allowDxgateAdmin...)
m.deleteRepeated(ctx, allowDxproxyAdmin...)
m.deleteRepeated(ctx, rejectOtherTCP...)
+ m.deleteRepeated(ctx, legacyRejectOtherTCP...)
if err := m.appendRule(ctx, allowExcluded...); err != nil {
return err
}
@@ -337,7 +344,26 @@ func (m *IPTablesRuleManager) ensureBase(ctx
context.Context) error {
if err := m.appendRule(ctx, allowDxproxyAdmin...); err != nil {
return err
}
- return m.appendRule(ctx, rejectOtherTCP...)
+ if err := m.appendRule(ctx, rejectOtherTCP...); err != nil {
+ return err
+ }
+ m.cleanupLegacyBase(ctx)
+ return nil
+}
+
+func (m *IPTablesRuleManager) cleanupLegacyBase(ctx context.Context) {
+ for _, chain := range []string{"FORWARD", "OUTPUT"} {
+ for i := 0; i < 20; i++ {
+ if err := m.run(ctx, "-w", "-t", "filter", "-D", chain,
"-j", legacyInboundChain); err != nil {
+ break
+ }
+ }
+ }
+ // Migration cleanup is best effort: absence means the node is already
+ // clean, while the new chain is fully installed before this runs.
+ _ = m.run(ctx, "-w", "-t", "filter", "-F", legacyInboundChain)
+ _ = m.run(ctx, "-w", "-t", "filter", "-X", legacyInboundChain)
+ _ = m.runIPSet(ctx, "destroy", legacyPodIPSet)
}
func (m *IPTablesRuleManager) deleteRepeated(ctx context.Context, args
...string) {
diff --git a/cni/pkg/nodeagent/iptables_test.go
b/cni/pkg/nodeagent/iptables_test.go
index db002649..f56166b7 100644
--- a/cni/pkg/nodeagent/iptables_test.go
+++ b/cni/pkg/nodeagent/iptables_test.go
@@ -41,6 +41,9 @@ func TestIPTablesRuleManagerAddsGRPCInboundBoundaryRules(t
*testing.T) {
"-N DUBBO-GRPC-INBOUND",
"-I FORWARD 1 -j DUBBO-GRPC-INBOUND",
"-I OUTPUT 1 -j DUBBO-GRPC-INBOUND",
+ "-D DUBBO-GRPC-INBOUND -m set --match-set
DUBBO-GRPC-INBOUND-PODS dst -p tcp -j REJECT",
+ "-D OUTPUT -j DUBBO-XSERVER-INBOUND",
+ "ipset destroy DUBBO-XSERVER-PODS",
"-A DUBBO-GRPC-INBOUND -m set --match-set
DUBBO-GRPC-INBOUND-EXCLUDE dst,dst -p tcp -j RETURN",
"-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",
diff --git a/dubbod/discovery/docker/dockerfile.dubbod
b/dubbod/discovery/docker/dockerfile.dubbod
index 2bf37f69..6fa392c5 100644
--- a/dubbod/discovery/docker/dockerfile.dubbod
+++ b/dubbod/discovery/docker/dockerfile.dubbod
@@ -49,7 +49,9 @@ RUN --mount=type=cache,target=/go/pkg/mod \
-o /out/dubbo-cni \
./cni/main.go
-FROM scratch
+FROM alpine:3.22
+
+RUN apk add --no-cache ipset iptables
COPY --from=builder /out/dubbod /usr/local/bin/dubbod
COPY --from=builder /out/dubbo-cni /usr/local/bin/dubbo-cni
diff --git a/dubbod/discovery/pkg/bootstrap/inherent_grpc_controller_test.go
b/dubbod/discovery/pkg/bootstrap/inherent_grpc_controller_test.go
index 164e6ebe..c965b697 100644
--- a/dubbod/discovery/pkg/bootstrap/inherent_grpc_controller_test.go
+++ b/dubbod/discovery/pkg/bootstrap/inherent_grpc_controller_test.go
@@ -251,6 +251,36 @@ func TestBuildRuntimeConfigJSON(t *testing.T) {
}
}
+func TestInherentTelemetrySerializesAllStandardMetrics(t *testing.T) {
+ metrics := []telemetryapi.StandardMetric{
+ telemetryapi.StandardMetric_REQUEST_COUNT,
+ telemetryapi.StandardMetric_REQUEST_DURATION,
+ telemetryapi.StandardMetric_REQUEST_SIZE,
+ telemetryapi.StandardMetric_RESPONSE_SIZE,
+ }
+ effective := telemetryconfig.EffectiveTracing{
+ MetricsConfigured: true,
+ MetricProviders: []string{telemetryconfig.PrometheusProvider},
+ MetricRules: make([]telemetryconfig.MetricRule, 0,
len(metrics)),
+ }
+ for _, metric := range metrics {
+ effective.MetricRules = append(effective.MetricRules,
telemetryconfig.MetricRule{
+ Metric: metric,
+ Scope: telemetryapi.MetricScope_CLIENT_AND_SERVER,
+ })
+ }
+
+ got := inherentGRPCTelemetryConfig(effective)
+ if got == nil || got.Metrics == nil || len(got.Metrics.Rules) !=
len(metrics) {
+ t.Fatalf("telemetry metrics = %#v", got)
+ }
+ for i, metric := range metrics {
+ if got.Metrics.Rules[i].Metric != metric.String() {
+ t.Fatalf("rule[%d].metric = %q, want %q", i,
got.Metrics.Rules[i].Metric, metric)
+ }
+ }
+}
+
func TestResolveInherentTelemetryForWorkload(t *testing.T) {
env, _ := newInherentRuntimeTestEnvironment(t, []config.Config{{
Meta: config.Meta{
diff --git a/dubbod/discovery/pkg/bootstrap/server.go
b/dubbod/discovery/pkg/bootstrap/server.go
index 482e1a81..22fc5cd5 100644
--- a/dubbod/discovery/pkg/bootstrap/server.go
+++ b/dubbod/discovery/pkg/bootstrap/server.go
@@ -622,6 +622,8 @@ func configKindForSchemaIdentifier(schemaID string)
(kind.Kind, bool) {
return kind.DxgateService, true
case "ServiceActivationPolicy":
return kind.ServiceActivationPolicy, true
+ case "Telemetry":
+ return kind.Telemetry, true
default:
return 0, false
}
diff --git a/dubbod/discovery/pkg/bootstrap/server_test.go
b/dubbod/discovery/pkg/bootstrap/server_test.go
index a04ae89c..94f9aa77 100644
--- a/dubbod/discovery/pkg/bootstrap/server_test.go
+++ b/dubbod/discovery/pkg/bootstrap/server_test.go
@@ -30,3 +30,13 @@ func
TestConfigKindForSchemaIdentifierIncludesDxgateService(t *testing.T) {
t.Fatalf("kind = %v, want %v", got, kind.DxgateService)
}
}
+
+func TestConfigKindForSchemaIdentifierIncludesTelemetry(t *testing.T) {
+ got, found := configKindForSchemaIdentifier("Telemetry")
+ if !found {
+ t.Fatal("Telemetry schema identifier was not mapped")
+ }
+ if got != kind.Telemetry {
+ t.Fatalf("kind = %v, want %v", got, kind.Telemetry)
+ }
+}
diff --git a/go.mod b/go.mod
index 44e1549a..0d80c333 100644
--- a/go.mod
+++ b/go.mod
@@ -17,10 +17,6 @@ module github.com/apache/dubbo-kubernetes
go 1.25.12
-replace github.com/kdubbo/api => ../api
-
-replace github.com/kdubbo/client-go => ../client-go
-
require (
k8s.io/apimachinery v0.34.1
k8s.io/client-go v0.34.1
@@ -45,9 +41,9 @@ require (
github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
github.com/hashicorp/go-multierror v1.1.1
github.com/hashicorp/golang-lru/v2 v2.0.7
- github.com/kdubbo/api v0.0.0-20260811085311-7752d1da2bcb
- github.com/kdubbo/client-go v0.0.0-20260809042806-04a31db14165
- github.com/kdubbo/xds-api v0.0.0-20260809042456-0d57cc43a21a
+ github.com/kdubbo/api v0.0.0-20260814141555-d9b670d33d9f
+ github.com/kdubbo/client-go v0.0.0-20260814141742-c761bafa7cf3
+ github.com/kdubbo/xds-api v0.0.0-20260814172110-c45be7c324a3
github.com/prometheus/client_golang v1.23.2
github.com/prometheus/client_model v0.6.2
github.com/spf13/cobra v1.10.2
@@ -146,7 +142,7 @@ require (
golang.org/x/oauth2 v0.36.0 // indirect
golang.org/x/sync v0.22.0 // indirect
golang.org/x/text v0.40.0 // indirect
- google.golang.org/genproto/googleapis/api
v0.0.0-20260414002931-afd174a4e478 // indirect
+ google.golang.org/genproto/googleapis/api
v0.0.0-20260420184626-e10c466a9529 // indirect
google.golang.org/genproto/googleapis/rpc
v0.0.0-20260427160629-7cedc36a6bc4 // indirect
gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect
gopkg.in/inf.v0 v0.9.1 // indirect
@@ -158,3 +154,5 @@ require (
sigs.k8s.io/randfill v1.0.0 // indirect
sigs.k8s.io/structured-merge-diff/v6 v6.3.0 // indirect
)
+
+replace google.golang.org/genproto => google.golang.org/genproto
v0.0.0-20260427160629-7cedc36a6bc4
diff --git a/go.sum b/go.sum
index bab6d2cf..fea71123 100644
--- a/go.sum
+++ b/go.sum
@@ -154,8 +154,16 @@ github.com/inconshreveable/mousetrap v1.1.0
h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2
github.com/inconshreveable/mousetrap v1.1.0/go.mod
h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
github.com/json-iterator/go v1.1.12
h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
github.com/json-iterator/go v1.1.12/go.mod
h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
+github.com/kdubbo/api v0.0.0-20260814141555-d9b670d33d9f
h1:T1jlRZ+ulFTUGziUnTjMvNEtiknaHQr4vQYviz4U/kQ=
+github.com/kdubbo/api v0.0.0-20260814141555-d9b670d33d9f/go.mod
h1:8BtJiIovg7QCPsCxXcw3gDf922VcvYq5ihOSvj49Rq8=
+github.com/kdubbo/client-go v0.0.0-20260814141742-c761bafa7cf3
h1:vTpsq7IAuUqKz05NoCWpDu0MY5SdjpH5NXZpj+l4I9Q=
+github.com/kdubbo/client-go v0.0.0-20260814141742-c761bafa7cf3/go.mod
h1:wCARoHJuh9ccSRQYlGXzPyMfUMYavnRot05dKh1N4KY=
github.com/kdubbo/xds-api v0.0.0-20260809042456-0d57cc43a21a
h1:stTvOOGy4r6DoxquO928mB9AGGoFBvjVHxqAGIGiSkI=
github.com/kdubbo/xds-api v0.0.0-20260809042456-0d57cc43a21a/go.mod
h1:o2HDUgL1ntaDbWomZ4cD2tt8jBamuG2qRtjXOa1zZ0Q=
+github.com/kdubbo/xds-api v0.0.0-20260814141925-82c6d4ca43c1
h1:/PaNI7QypbLp32sqRd54iu1DMs2XmxHKAaVSjhcLF/4=
+github.com/kdubbo/xds-api v0.0.0-20260814141925-82c6d4ca43c1/go.mod
h1:o2HDUgL1ntaDbWomZ4cD2tt8jBamuG2qRtjXOa1zZ0Q=
+github.com/kdubbo/xds-api v0.0.0-20260814172110-c45be7c324a3
h1:ypir1ZNYdAKOuWokpObAdUtCDXLD3yqP1AOqCURe3WU=
+github.com/kdubbo/xds-api v0.0.0-20260814172110-c45be7c324a3/go.mod
h1:o2HDUgL1ntaDbWomZ4cD2tt8jBamuG2qRtjXOa1zZ0Q=
github.com/kisielk/errcheck v1.5.0/go.mod
h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
github.com/kisielk/gotool v1.0.0/go.mod
h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
github.com/klauspost/compress v1.18.6
h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao=
@@ -360,8 +368,12 @@ google.golang.org/appengine v1.4.0/go.mod
h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7
google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod
h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod
h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc=
google.golang.org/genproto v0.0.0-20200423170343-7949de9c1215/go.mod
h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
+google.golang.org/genproto v0.0.0-20260427160629-7cedc36a6bc4
h1:2iMJZntwvmfgtse+s744JY7v7PgEdSBuFYXucvpOHNM=
+google.golang.org/genproto v0.0.0-20260427160629-7cedc36a6bc4/go.mod
h1:v14kaaboYyXQ1Gsu489Q+Hg/oN4B33mWtuOhF1HCeXA=
google.golang.org/genproto/googleapis/api v0.0.0-20260414002931-afd174a4e478
h1:yQugLulqltosq0B/f8l4w9VryjV+N/5gcW0jQ3N8Qec=
google.golang.org/genproto/googleapis/api
v0.0.0-20260414002931-afd174a4e478/go.mod
h1:C6ADNqOxbgdUUeRTU+LCHDPB9ttAMCTff6auwCVa4uc=
+google.golang.org/genproto/googleapis/api v0.0.0-20260420184626-e10c466a9529
h1:zUWMZsvo/IJcD1t6MNCPO/azZTwz0TvwCBqr5aifoVY=
+google.golang.org/genproto/googleapis/api
v0.0.0-20260420184626-e10c466a9529/go.mod
h1:a5OGAgyRr4lqco7AG9hQM9Fwh0N2ZV4grR0eXFEsXQg=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260427160629-7cedc36a6bc4
h1:tEkOQcXgF6dH1G+MVKZrfpYvozGrzb91k6ha7jireSM=
google.golang.org/genproto/googleapis/rpc
v0.0.0-20260427160629-7cedc36a6bc4/go.mod
h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
google.golang.org/grpc v1.19.0/go.mod
h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
diff --git a/manifests/charts/base/files/crd-all.gen.yaml
b/manifests/charts/base/files/crd-all.gen.yaml
index 6ca35631..95713abb 100644
--- a/manifests/charts/base/files/crd-all.gen.yaml
+++ b/manifests/charts/base/files/crd-all.gen.yaml
@@ -1626,10 +1626,13 @@ spec:
description: |-
REQUIRED.
- Valid Options: REQUEST_COUNT
+ Valid Options: REQUEST_COUNT, REQUEST_DURATION,
REQUEST_SIZE, RESPONSE_SIZE
enum:
- STANDARD_METRIC_UNSPECIFIED
- REQUEST_COUNT
+ - REQUEST_DURATION
+ - REQUEST_SIZE
+ - RESPONSE_SIZE
type: string
scope:
description: |-
@@ -1655,7 +1658,7 @@ spec:
- REMOVE
type: string
type: object
- description: Tag overrides keyed by metric tag
name.
+ description: Tag overrides keyed by standard label
name.
type: object
type: object
type: array
diff --git a/manifests/charts/dubbod/files/grpc-engine.yaml
b/manifests/charts/dubbod/files/grpc-engine.yaml
index 36922e31..57606c0f 100644
--- a/manifests/charts/dubbod/files/grpc-engine.yaml
+++ b/manifests/charts/dubbod/files/grpc-engine.yaml
@@ -103,35 +103,6 @@ spec:
- name: dubbo-xds
mountPath: /etc/dubbo/proxy
readOnly: true
-{{- $appPort := 80 }}
-{{- if and (gt (len .Spec.Containers) 0) (gt (len ((index .Spec.Containers
0).Ports)) 0) }}
-{{- $appPort = int ((index ((index .Spec.Containers 0).Ports)
0).ContainerPort) }}
-{{- end }}
- - name: dubbo-grpc-inbound
- image: {{ .ProxyImage | quote }}
- imagePullPolicy: IfNotPresent
- args:
- - grpc-inbound
- - --listen
- - :15080
- - --upstream
- - 127.0.0.1:{{ $appPort }}
- ports:
- - name: grpc-inbound
- containerPort: 15080
- protocol: TCP
- # Probe the listener the sidecar actually owns. periodSeconds x
- # failureThreshold must stay below the sidecar's termination drain delay
- # (5s by default), so kubelet withdraws the endpoint promptly.
- readinessProbe:
- tcpSocket:
- port: 15080
- periodSeconds: 2
- failureThreshold: 2
- volumeMounts:
- - name: dubbo-xds
- mountPath: /etc/dubbo/proxy
- readOnly: true
volumes:
- name: dubbo-xds
secret:
diff --git a/manifests/charts/dubbod/templates/cni-daemonset.yaml
b/manifests/charts/dubbod/templates/cni-daemonset.yaml
index 15dcbb85..6ca76ccc 100644
--- a/manifests/charts/dubbod/templates/cni-daemonset.yaml
+++ b/manifests/charts/dubbod/templates/cni-daemonset.yaml
@@ -81,6 +81,9 @@ spec:
capabilities:
drop:
- ALL
+ add:
+ - NET_ADMIN
+ - NET_RAW
volumeMounts:
- name: cni-bin-dir
mountPath: /opt/cni/bin
diff --git a/manifests/charts/dubbod/templates/cni-rbac.yaml
b/manifests/charts/dubbod/templates/cni-rbac.yaml
index d51e7d2e..5e6291c9 100644
--- a/manifests/charts/dubbod/templates/cni-rbac.yaml
+++ b/manifests/charts/dubbod/templates/cni-rbac.yaml
@@ -26,7 +26,7 @@ metadata:
rules:
- apiGroups: [""]
resources: ["pods"]
- verbs: ["get"]
+ verbs: ["get", "list"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
diff --git a/pkg/config/validation/validators.go
b/pkg/config/validation/validators.go
index ce82afb2..750bde73 100644
--- a/pkg/config/validation/validators.go
+++ b/pkg/config/validation/validators.go
@@ -574,6 +574,13 @@ func validateWorkloadAddress(field, value string) error {
return validateServiceEntryHost(field, value)
}
+var telemetryStandardLabels = map[string]struct{}{
+ "reporter": {},
+ "grpc_service": {},
+ "grpc_method": {},
+ "grpc_response_status": {},
+}
+
// ValidateTelemetry checks that a Telemetry resource is well-formed.
var ValidateTelemetry = RegisterValidateFunc("ValidateTelemetry",
func(cfg config.Config) (Warning, error) {
@@ -627,6 +634,9 @@ var ValidateTelemetry =
RegisterValidateFunc("ValidateTelemetry",
for name, override := range rule.GetTags() {
if strings.TrimSpace(name) == "" {
v = appendValidation(v,
fmt.Errorf("metrics[%d].rules[%d].tags contains an empty name", i, j))
+ } else if _, found :=
telemetryStandardLabels[name]; !found {
+ v = appendValidation(v,
fmt.Errorf(
+
"metrics[%d].rules[%d].tags[%q] is not a standard label", i, j, name))
}
if override == nil {
v = appendValidation(v,
fmt.Errorf("metrics[%d].rules[%d].tags[%q] must not be null", i, j, name))
diff --git a/pkg/config/validation/validators_test.go
b/pkg/config/validation/validators_test.go
index 30212f26..e71b6ee9 100644
--- a/pkg/config/validation/validators_test.go
+++ b/pkg/config/validation/validators_test.go
@@ -626,6 +626,9 @@ func TestValidateTelemetry(t *testing.T) {
Metric:
telemetry.StandardMetric_REQUEST_COUNT,
Scope:
telemetry.MetricScope_CLIENT_AND_SERVER,
Tags: map[string]*telemetry.TagOverride{
+ "reporter":
{Action: telemetry.TagOverride_REMOVE},
+ "grpc_service":
{Action: telemetry.TagOverride_REMOVE},
+ "grpc_method":
{Action: telemetry.TagOverride_REMOVE},
"grpc_response_status":
{Action: telemetry.TagOverride_REMOVE},
},
}},
@@ -639,6 +642,19 @@ func TestValidateTelemetry(t *testing.T) {
}}},
wantErr: true,
},
+ {
+ name: "unknown metrics standard label",
+ spec: &telemetry.Telemetry{Metrics:
[]*telemetry.Metrics{{
+ Rules: []*telemetry.MetricRule{{
+ Metric:
telemetry.StandardMetric_REQUEST_COUNT,
+ Scope: telemetry.MetricScope_CLIENT,
+ Tags: map[string]*telemetry.TagOverride{
+ "pod": {Action:
telemetry.TagOverride_REMOVE},
+ },
+ }},
+ }}},
+ wantErr: true,
+ },
{
name: "metrics rule without scope",
spec: &telemetry.Telemetry{Metrics:
[]*telemetry.Metrics{{
diff --git a/pkg/kube/inject/inherent.go b/pkg/kube/inject/inherent.go
index 226a170a..c8315c50 100644
--- a/pkg/kube/inject/inherent.go
+++ b/pkg/kube/inject/inherent.go
@@ -36,6 +36,10 @@ const (
InherentGRPCBootstrapPath = InherentXDSMountPath +
"/" + InherentGRPCBootstrapFileName
InherentXDSAddressEnvName = "XDS_ADDRESS"
InherentGRPCConfigEnvName = "DUBBO_GRPC_XDS_CONFIG"
+ InherentGRPCMetricsAddressEnvName =
"DUBBO_GRPC_METRICS_ADDRESS"
+ InherentGRPCMetricsAddress = ":9090"
+ InherentGRPCMetricsPortName = "metrics"
+ InherentGRPCMetricsPort = 9090
InherentGRPCKeepaliveEnvName = "DUBBO_GRPC_KEEPALIVE"
InherentGRPCKeepaliveTimeEnv = "GRPC_KEEPALIVE_INTERVAL"
InherentGRPCKeepaliveTimeoutEnv = "GRPC_KEEPALIVE_TIMEOUT"
diff --git a/pkg/kube/inject/inherent_test.go b/pkg/kube/inject/inherent_test.go
index 2dc08fc6..d4a68eb2 100644
--- a/pkg/kube/inject/inherent_test.go
+++ b/pkg/kube/inject/inherent_test.go
@@ -21,7 +21,6 @@ import (
"runtime"
"strings"
"testing"
- "time"
telemetryconfig
"github.com/apache/dubbo-kubernetes/pkg/config/telemetry"
meshv1alpha1 "github.com/kdubbo/api/mesh/v1alpha1"
@@ -85,18 +84,17 @@ func
TestInstallerGRPCEngineTemplateInjectsDirectXDSConnection(t *testing.T) {
t.Fatalf("RunTemplate() failed: %v", err)
}
- if len(injectedPod.Spec.Containers) != 2 {
- t.Fatalf("template containers = %d, want app overlay plus
grpc-inbound", len(injectedPod.Spec.Containers))
+ if len(injectedPod.Spec.Containers) != 1 {
+ t.Fatalf("template containers = %d, want application overlay
only", len(injectedPod.Spec.Containers))
}
if err := postProcessPod(mergedPod, *injectedPod, req); err != nil {
t.Fatalf("postProcessPod() failed: %v", err)
}
- if len(mergedPod.Spec.Containers) != 2 {
- t.Fatalf("containers = %d, want application container plus
grpc-inbound", len(mergedPod.Spec.Containers))
+ if len(mergedPod.Spec.Containers) != 1 {
+ t.Fatalf("containers = %d, want original application container
only", len(mergedPod.Spec.Containers))
}
assertDirectXDSConnection(t, mergedPod, "app",
InherentGRPCSecretNameForMeta(pod.ObjectMeta))
- assertGRPCInboundContainer(t, mergedPod)
}
func TestInstallerGRPCEngineTemplateUsesGenerateNameForDeploymentPods(t
*testing.T) {
@@ -155,11 +153,10 @@ func
TestInstallerGRPCEngineTemplateUsesGenerateNameForDeploymentPods(t *testing
if err := postProcessPod(mergedPod, *injectedPod, req); err != nil {
t.Fatalf("postProcessPod() failed: %v", err)
}
- if len(mergedPod.Spec.Containers) != 2 {
- t.Fatalf("containers = %d, want original nginx container plus
grpc-inbound", len(mergedPod.Spec.Containers))
+ if len(mergedPod.Spec.Containers) != 1 {
+ t.Fatalf("containers = %d, want original application container
only", len(mergedPod.Spec.Containers))
}
assertDirectXDSConnection(t, mergedPod, "nginx",
InherentGRPCSecretNameForMeta(pod.ObjectMeta))
- assertGRPCInboundContainer(t, mergedPod)
if got := mergedPod.Spec.Volumes[0].Secret.SecretName; got ==
InherentGRPCSecretName("") {
t.Fatalf("secret name = %q, want generateName-based secret",
got)
}
@@ -178,6 +175,21 @@ func assertDirectXDSConnection(t *testing.T, pod
*corev1.Pod, containerName, sec
if !hasEnv(container.Env, InherentGRPCConfigEnvName,
InherentGRPCConfigPath) {
t.Fatalf("%s env missing", InherentGRPCConfigEnvName)
}
+ if !hasEnv(container.Env, InherentGRPCMetricsAddressEnvName,
InherentGRPCMetricsAddress) {
+ t.Fatalf("%s env missing", InherentGRPCMetricsAddressEnvName)
+ }
+ foundMetricsPort := false
+ for _, port := range container.Ports {
+ if port.Name == InherentGRPCMetricsPortName &&
port.ContainerPort == InherentGRPCMetricsPort {
+ foundMetricsPort = true
+ }
+ }
+ if !foundMetricsPort {
+ t.Fatalf("metrics port %d missing", InherentGRPCMetricsPort)
+ }
+ if got := pod.Annotations["prometheus.io/scrape"]; got != "true" {
+ t.Fatalf("prometheus scrape annotation = %q, want true", got)
+ }
if !hasEnv(container.Env, InherentXDSAddressEnvName,
"dubbod.dubbo-system.svc:26012") {
t.Fatalf("%s env missing", InherentXDSAddressEnvName)
}
@@ -244,53 +256,6 @@ func assertNoArgs(t *testing.T, pod *corev1.Pod) {
}
}
-// inherentDrainDelay mirrors the sidecar's default termination drain delay.
-// The readiness probe must detect termination inside this window.
-const inherentDrainDelay = 5 * time.Second
-
-func assertGRPCInboundContainer(t *testing.T, pod *corev1.Pod) {
- t.Helper()
- container := FindContainer(InherentGRPCInboundContainerName,
pod.Spec.Containers)
- if container == nil {
- t.Fatalf("%s container missing",
InherentGRPCInboundContainerName)
- }
- if container.Image != "kdubbo/dubbod:debug" {
- t.Fatalf("grpc-inbound image = %q, want kdubbo/dubbod:debug",
container.Image)
- }
- wantArgs := []string{"grpc-inbound", "--listen", ":15080",
"--upstream", "127.0.0.1:80"}
- if strings.Join(container.Args, ",") != strings.Join(wantArgs, ",") {
- t.Fatalf("grpc-inbound args = %v, want %v", container.Args,
wantArgs)
- }
- if !hasMount(container.VolumeMounts, InherentXDSVolumeName,
InherentXDSMountPath, true) {
- t.Fatalf("grpc-inbound inherent xds mount missing")
- }
- assertDrainReadinessProbe(t, container)
-}
-
-// assertDrainReadinessProbe checks the probe that withdraws a terminating pod
-// from its EndpointSlice. Without it the sidecar's drain delay is inert:
kubelet
-// never observes the listener closing, so the endpoint is still published
-// after the data-plane port is gone.
-func assertDrainReadinessProbe(t *testing.T, container *corev1.Container) {
- t.Helper()
- probe := container.ReadinessProbe
- if probe == nil || probe.TCPSocket == nil {
- t.Fatalf("grpc-inbound readiness probe missing")
- }
- if probe.TCPSocket.Port.IntValue() != InherentGRPCInboundPort {
- t.Fatalf("grpc-inbound readiness probe = %v, want TCP port %d",
- probe.TCPSocket.Port, InherentGRPCInboundPort)
- }
- // The probe has to fail before the sidecar stops accepting, otherwise
the
- // endpoint is withdrawn only after the listener is already gone.
- if detection :=
time.Duration(probe.PeriodSeconds*probe.FailureThreshold) * time.Second;
detection >= inherentDrainDelay {
- t.Fatalf("readiness detection window = %v, want less than the
%v drain delay", detection, inherentDrainDelay)
- }
- if !hasContainerPort(container.Ports, InherentGRPCInboundPort) {
- t.Fatalf("grpc-inbound port %d not declared",
InherentGRPCInboundPort)
- }
-}
-
func TestGetProxyImageUsesTopLevelImage(t *testing.T) {
values := map[string]any{
"image": "kdubbo/dubbod:test",
@@ -300,15 +265,6 @@ func TestGetProxyImageUsesTopLevelImage(t *testing.T) {
}
}
-func hasContainerPort(ports []corev1.ContainerPort, want int) bool {
- for _, port := range ports {
- if int(port.ContainerPort) == want {
- return true
- }
- }
- return false
-}
-
func TestAddApplicationContainerConfigInjectsInherentGRPCContract(t
*testing.T) {
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
@@ -586,14 +542,6 @@ func TestEnsureInherentGRPCTemplateAnnotation(t
*testing.T) {
}
}
-func TestEnsureInherentManagedLabel(t *testing.T) {
- pod := &corev1.Pod{}
- ensureInherentManagedLabel(pod)
- if got := pod.Labels[InherentManagedLabel]; got !=
InherentManagedLabelValue {
- t.Fatalf("managed label = %q, want %q", got,
InherentManagedLabelValue)
- }
-}
-
func TestInherentGRPCSecretNameFitsKubernetesLengthLimit(t *testing.T) {
name :=
InherentGRPCSecretName("grpc-provider-012345678901234567890123456789012345678901234567890123")
if len(name) > 63 {
@@ -712,6 +660,34 @@ func TestInstallerGRPCEngineTemplateInjectsTelemetryEnv(t
*testing.T) {
if err != nil {
t.Fatalf("RunTemplate() failed: %v", err)
}
+ if err := postProcessPod(mergedPod, corev1.Pod{}, newParams(tracing));
err != nil {
+ t.Fatalf("postProcessPod() failed: %v", err)
+ }
+ if got := envValue(mergedPod, InherentGRPCMetricsAddressEnvName); got
!= InherentGRPCMetricsAddress {
+ t.Fatalf("%s = %q, want %q", InherentGRPCMetricsAddressEnvName,
got, InherentGRPCMetricsAddress)
+ }
+ app := FindContainer("app", mergedPod.Spec.Containers)
+ if app == nil {
+ t.Fatal("app container not found")
+ }
+ foundMetricsPort := false
+ for _, port := range app.Ports {
+ if port.Name == InherentGRPCMetricsPortName &&
port.ContainerPort == InherentGRPCMetricsPort {
+ foundMetricsPort = true
+ }
+ }
+ if !foundMetricsPort {
+ t.Fatalf("application metrics port %d missing",
InherentGRPCMetricsPort)
+ }
+ for key, want := range map[string]string{
+ "prometheus.io/scrape": "true",
+ "prometheus.io/path": "/metrics",
+ "prometheus.io/port": "9090",
+ } {
+ if got := mergedPod.Annotations[key]; got != want {
+ t.Fatalf("annotation %s = %q, want %q", key, got, want)
+ }
+ }
if got, want := envValue(mergedPod, "OTEL_EXPORTER_OTLP_ENDPOINT"),
"http://tracing.dubbo-system.svc:4317";; got != want {
t.Fatalf("OTEL_EXPORTER_OTLP_ENDPOINT = %q, want %q", got, want)
}
diff --git a/pkg/kube/inject/service_test.go b/pkg/kube/inject/service_test.go
index d5546a4e..a8a58eee 100644
--- a/pkg/kube/inject/service_test.go
+++ b/pkg/kube/inject/service_test.go
@@ -1,23 +1,12 @@
// Licensed to the Apache Software Foundation (ASF) under one or more
-// contributor license agreements. See the NOTICE file distributed with
+// 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.
+// The ASF licenses this file to You under the Apache License, Version 2.0.
package inject
import (
"encoding/json"
- "strings"
"testing"
"github.com/apache/dubbo-kubernetes/pkg/kube"
@@ -28,105 +17,24 @@ import (
"k8s.io/apimachinery/pkg/util/intstr"
)
-func TestRewriteInherentServiceTargetPortsRoutesTCPServiceToGRPCInbound(t
*testing.T) {
- svc := &corev1.Service{
- ObjectMeta: metav1.ObjectMeta{Name: "nginx", Namespace: "app"},
+func TestInjectServiceDoesNotRewriteApplicationTargetPort(t *testing.T) {
+ service := corev1.Service{
+ ObjectMeta: metav1.ObjectMeta{Name: "payment", Namespace:
"app"},
Spec: corev1.ServiceSpec{
- Selector: map[string]string{"app": "nginx"},
- Ports: []corev1.ServicePort{
- {Name: "http", Port: 80, TargetPort:
intstr.FromInt(80), Protocol: corev1.ProtocolTCP},
- {Name: "metrics", Port: 9090, TargetPort:
intstr.FromString("metrics"), Protocol: corev1.ProtocolTCP},
- },
+ Selector: map[string]string{"app": "payment"},
+ Ports: []corev1.ServicePort{{
+ Name: "grpc",
+ Port: 80,
+ TargetPort: intstr.FromInt(8080),
+ Protocol: corev1.ProtocolTCP,
+ }},
},
}
-
- if !rewriteInherentServiceTargetPorts(svc) {
- t.Fatalf("rewriteInherentServiceTargetPorts() = false, want
true")
- }
- for _, port := range svc.Spec.Ports {
- if got := port.TargetPort.IntVal; got !=
InherentGRPCInboundPort {
- t.Fatalf("port %s targetPort = %d, want %d", port.Name,
got, InherentGRPCInboundPort)
- }
- }
-}
-
-func TestRewriteInherentServiceTargetPortsSkipsUnsafeServices(t *testing.T) {
- cases := []struct {
- name string
- svc corev1.Service
- }{
- {
- name: "selectorless",
- svc: corev1.Service{
- ObjectMeta: metav1.ObjectMeta{Name: "manual",
Namespace: "app"},
- Spec: corev1.ServiceSpec{
- Ports: []corev1.ServicePort{{Name:
"http", Port: 80, TargetPort: intstr.FromInt(80)}},
- },
- },
- },
- {
- name: "external-name",
- svc: corev1.Service{
- ObjectMeta: metav1.ObjectMeta{Name: "external",
Namespace: "app"},
- Spec: corev1.ServiceSpec{
- Type:
corev1.ServiceTypeExternalName,
- ExternalName: "example.com",
- Selector: map[string]string{"app":
"external"},
- Ports:
[]corev1.ServicePort{{Name: "http", Port: 80, TargetPort: intstr.FromInt(80)}},
- },
- },
- },
- }
-
- for _, tt := range cases {
- t.Run(tt.name, func(t *testing.T) {
- svc := tt.svc
- if rewriteInherentServiceTargetPorts(&svc) {
- t.Fatalf("rewriteInherentServiceTargetPorts() =
true, want false")
- }
- if got := svc.Spec.Ports[0].TargetPort.IntVal; got !=
80 {
- t.Fatalf("targetPort = %d, want 80", got)
- }
- })
- }
-}
-
-func TestRewriteInherentServiceTargetPortsSkipsNonTCPPorts(t *testing.T) {
- svc := &corev1.Service{
- ObjectMeta: metav1.ObjectMeta{Name: "dns", Namespace: "app"},
- Spec: corev1.ServiceSpec{
- Selector: map[string]string{"app": "dns"},
- Ports: []corev1.ServicePort{
- {Name: "dns", Port: 53, TargetPort:
intstr.FromInt(53), Protocol: corev1.ProtocolUDP},
- },
- },
- }
-
- if rewriteInherentServiceTargetPorts(svc) {
- t.Fatalf("rewriteInherentServiceTargetPorts() = true, want
false")
- }
- if got := svc.Spec.Ports[0].TargetPort.IntVal; got != 53 {
- t.Fatalf("targetPort = %d, want 53", got)
- }
-}
-
-func TestInjectServiceCreatesGRPCInboundTargetPortPatch(t *testing.T) {
- svc := corev1.Service{
- ObjectMeta: metav1.ObjectMeta{Name: "nginx", Namespace: "app"},
- Spec: corev1.ServiceSpec{
- Selector: map[string]string{"app": "nginx"},
- Ports: []corev1.ServicePort{
- {Name: "http", Port: 80, TargetPort:
intstr.FromInt(80), Protocol: corev1.ProtocolTCP},
- },
- },
- }
- raw, err := json.Marshal(svc)
+ raw, err := json.Marshal(service)
if err != nil {
- t.Fatalf("json.Marshal() failed: %v", err)
+ t.Fatal(err)
}
-
- wh := &Webhook{Config: &Config{Policy: InjectionPolicyEnabled}}
- resp := wh.injectService(&kube.AdmissionReview{
+ response := (&Webhook{}).injectService(&kube.AdmissionReview{
Request: &kube.AdmissionRequest{
UID: types.UID("test"),
Kind: metav1.GroupVersionKind{Version: "v1", Kind:
"Service"},
@@ -135,54 +43,10 @@ func TestInjectServiceCreatesGRPCInboundTargetPortPatch(t
*testing.T) {
Object: runtime.RawExtension{Raw: raw},
},
}, "/inject")
-
- if !resp.Allowed {
- t.Fatalf("Allowed = false, want true")
- }
- if resp.PatchType == nil || *resp.PatchType != "JSONPatch" {
- t.Fatalf("PatchType = %v, want JSONPatch", resp.PatchType)
- }
- if !strings.Contains(string(resp.Patch), `"value":15080`) {
- t.Fatalf("patch = %s, want targetPort 15080",
string(resp.Patch))
- }
-}
-
-func TestInjectServiceSkipsInherentOptOutService(t *testing.T) {
- svc := corev1.Service{
- ObjectMeta: metav1.ObjectMeta{
- Name: "dxgate-gateway",
- Namespace: "app",
- Labels: map[string]string{
- "inherent.dubbo.apache.org/inject": "false",
- },
- },
- Spec: corev1.ServiceSpec{
- Selector: map[string]string{"app.kubernetes.io/name":
"dxgate"},
- Ports: []corev1.ServicePort{
- {Name: "http", Port: 80, TargetPort:
intstr.FromString("http"), Protocol: corev1.ProtocolTCP},
- },
- },
- }
- raw, err := json.Marshal(svc)
- if err != nil {
- t.Fatalf("json.Marshal() failed: %v", err)
- }
-
- wh := &Webhook{Config: &Config{Policy: InjectionPolicyEnabled}}
- resp := wh.injectService(&kube.AdmissionReview{
- Request: &kube.AdmissionRequest{
- UID: types.UID("test"),
- Kind: metav1.GroupVersionKind{Version: "v1", Kind:
"Service"},
- Namespace: "app",
- Operation: kube.Create,
- Object: runtime.RawExtension{Raw: raw},
- },
- }, "/inject")
-
- if !resp.Allowed {
- t.Fatalf("Allowed = false, want true")
+ if !response.Allowed {
+ t.Fatal("service admission rejected")
}
- if len(resp.Patch) != 0 {
- t.Fatalf("patch = %s, want no patch", string(resp.Patch))
+ if len(response.Patch) != 0 {
+ t.Fatalf("service patch = %s, want no targetPort rewrite",
response.Patch)
}
}
diff --git a/pkg/kube/inject/webhook.go b/pkg/kube/inject/webhook.go
index fc4257d4..508ece0b 100644
--- a/pkg/kube/inject/webhook.go
+++ b/pkg/kube/inject/webhook.go
@@ -23,6 +23,7 @@ import (
"net/http"
"os"
+ "strconv"
"strings"
"sync"
"text/template"
@@ -47,7 +48,6 @@ import (
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/serializer"
"k8s.io/apimachinery/pkg/types"
- "k8s.io/apimachinery/pkg/util/intstr"
"k8s.io/apimachinery/pkg/util/mergepatch"
"k8s.io/apimachinery/pkg/util/strategicpatch"
"sigs.k8s.io/yaml"
@@ -323,52 +323,7 @@ func (wh *Webhook) injectPod(ar *kube.AdmissionReview,
path string) *kube.Admiss
}
func (wh *Webhook) injectService(ar *kube.AdmissionReview, path string)
*kube.AdmissionResponse {
- log := webhookLog.WithLabels("path", path)
- req := ar.Request
- var svc corev1.Service
- if err := json.Unmarshal(req.Object.Raw, &svc); err != nil {
- log.Errorf("Could not unmarshal raw service object: %v %s",
err, string(req.Object.Raw))
- return toAdmissionResponse(err)
- }
- if svc.Namespace == "" {
- svc.Namespace = req.Namespace
- }
-
- log = log.WithLabels("service", svc.Namespace+"/"+svc.Name)
- log.Infof("Process inherent service request")
-
- wh.mu.RLock()
- required := injectRequired(IgnoredNamespaces.UnsortedList(), wh.Config,
&corev1.PodSpec{}, svc.ObjectMeta)
- wh.mu.RUnlock()
- if !required {
- log.Infof("Skipping service due to policy check")
- return &kube.AdmissionResponse{Allowed: true}
- }
-
- originalService, err := json.Marshal(svc)
- if err != nil {
- return toAdmissionResponse(err)
- }
- if !rewriteInherentServiceTargetPorts(&svc) {
- return &kube.AdmissionResponse{Allowed: true}
- }
-
- patchBytes, err := createServicePatch(&svc, originalService)
- if err != nil {
- log.Errorf("Service injection failed: %v", err)
- return toAdmissionResponse(err)
- }
-
- log.Infof("Service injection successfully, patch size: %d bytes",
len(patchBytes))
- reviewResponse := kube.AdmissionResponse{
- Allowed: true,
- Patch: patchBytes,
- PatchType: func() *string {
- pt := "JSONPatch"
- return &pt
- }(),
- }
- return &reviewResponse
+ return &kube.AdmissionResponse{Allowed: true}
}
func (wh *Webhook) Run(stop <-chan struct{}) {
@@ -454,12 +409,9 @@ func postProcessPod(pod *corev1.Pod, injectedPod
corev1.Pod, req InjectionParame
if shouldInjectInherentGRPC(req) {
// Add Inherent gRPC env and shared bootstrap/cert volume to
application containers.
ensureInherentGRPCTemplateAnnotation(pod)
- ensureInherentManagedLabel(pod)
if err := addApplicationContainerConfig(pod, req); err != nil {
return err
}
- // Must run after the sidecar is merged in: the rewrite reads
the port it forwards to.
- RewriteAppProbes(pod)
}
if err := reorderPod(pod, req); err != nil {
@@ -543,6 +495,19 @@ func addApplicationContainerConfig(pod *corev1.Pod, req
InjectionParameters) err
pod.Spec.Volumes = append(pod.Spec.Volumes, desiredVolume)
}
+ if len(pod.Spec.Containers) > 0 {
+ application := &pod.Spec.Containers[0]
+ application.Env = ensureEnvVar(application.Env, corev1.EnvVar{
+ Name: InherentGRPCMetricsAddressEnvName,
+ Value: InherentGRPCMetricsAddress,
+ })
+ application.Ports = ensureContainerPort(application.Ports,
corev1.ContainerPort{
+ Name: InherentGRPCMetricsPortName,
+ ContainerPort: InherentGRPCMetricsPort,
+ Protocol: corev1.ProtocolTCP,
+ })
+ ensureApplicationMetricsAnnotations(pod)
+ }
for i := range pod.Spec.Containers {
container := &pod.Spec.Containers[i]
if container.Name == "dubbo-proxy" || container.Name ==
"dubbo-validation" {
@@ -668,6 +633,30 @@ func addApplicationContainerConfig(pod *corev1.Pod, req
InjectionParameters) err
return nil
}
+func ensureContainerPort(ports []corev1.ContainerPort, desired
corev1.ContainerPort) []corev1.ContainerPort {
+ for _, port := range ports {
+ if port.Name == desired.Name || port.ContainerPort ==
desired.ContainerPort {
+ return ports
+ }
+ }
+ return append(ports, desired)
+}
+
+func ensureApplicationMetricsAnnotations(pod *corev1.Pod) {
+ if pod.Annotations == nil {
+ pod.Annotations = map[string]string{}
+ }
+ for key, value := range map[string]string{
+ "prometheus.io/scrape": "true",
+ "prometheus.io/path": "/metrics",
+ "prometheus.io/port": strconv.Itoa(InherentGRPCMetricsPort),
+ } {
+ if _, found := pod.Annotations[key]; !found {
+ pod.Annotations[key] = value
+ }
+ }
+}
+
func ensureInherentGRPCTemplateAnnotation(pod *corev1.Pod) {
if pod.Annotations == nil {
pod.Annotations = map[string]string{}
@@ -685,13 +674,6 @@ func ensureInherentGRPCTemplateAnnotation(pod *corev1.Pod)
{
pod.Annotations[InherentInjectTemplatesAnnoName] = templates + "," +
InherentGRPCTemplateName
}
-func ensureInherentManagedLabel(pod *corev1.Pod) {
- if pod.Labels == nil {
- pod.Labels = map[string]string{}
- }
- pod.Labels[InherentManagedLabel] = InherentManagedLabelValue
-}
-
func ensureEnvVar(envs []corev1.EnvVar, desired corev1.EnvVar) []corev1.EnvVar
{
for _, env := range envs {
if env.Name == desired.Name {
@@ -714,7 +696,6 @@ func reorderPod(pod *corev1.Pod, req InjectionParameters)
error {
// Proxy container should be last to ensure `kubectl exec` and similar
commands
// continue to default to the user's container
pod.Spec.Containers = modifyContainers(pod.Spec.Containers,
ProxyContainerName, MoveLast)
- pod.Spec.Containers = modifyContainers(pod.Spec.Containers,
InherentGRPCInboundContainerName, MoveLast)
return nil
}
@@ -730,38 +711,6 @@ func createPatch(pod *corev1.Pod, original []byte)
([]byte, error) {
return json.Marshal(p)
}
-func createServicePatch(svc *corev1.Service, original []byte) ([]byte, error) {
- reinjected, err := json.Marshal(svc)
- if err != nil {
- return nil, err
- }
- p, err := jsonpatch.CreatePatch(original, reinjected)
- if err != nil {
- return nil, err
- }
- return json.Marshal(p)
-}
-
-func rewriteInherentServiceTargetPorts(svc *corev1.Service) bool {
- if svc.Spec.Type == corev1.ServiceTypeExternalName ||
len(svc.Spec.Selector) == 0 {
- return false
- }
-
- changed := false
- for i := range svc.Spec.Ports {
- port := &svc.Spec.Ports[i]
- if port.Protocol != "" && port.Protocol != corev1.ProtocolTCP {
- continue
- }
- if port.TargetPort.Type == intstr.Int && port.TargetPort.IntVal
== InherentGRPCInboundPort {
- continue
- }
- port.TargetPort = intstr.FromInt(InherentGRPCInboundPort)
- changed = true
- }
- return changed
-}
-
func applyOverlayYAML(target *corev1.Pod, overlayYAML []byte) (*corev1.Pod,
error) {
currentJSON, err := json.Marshal(target)
if err != nil {
diff --git a/samples/addons/grafana.yaml b/samples/addons/grafana.yaml
index 98651b8c..a1a5f2e8 100644
--- a/samples/addons/grafana.yaml
+++ b/samples/addons/grafana.yaml
@@ -826,6 +826,193 @@ data:
"version": 1,
"weekStart": ""
}
+ application-standard-metrics.json: |
+ {
+ "annotations": {
+ "list": []
+ },
+ "editable": true,
+ "graphTooltip": 0,
+ "id": null,
+ "links": [],
+ "panels": [
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "prometheus"
+ },
+ "fieldConfig": {
+ "defaults": {
+ "unit": "reqps"
+ },
+ "overrides": []
+ },
+ "gridPos": {
+ "h": 8,
+ "w": 12,
+ "x": 0,
+ "y": 0
+ },
+ "id": 1,
+ "options": {
+ "legend": {
+ "displayMode": "list",
+ "placement": "bottom",
+ "showLegend": true
+ },
+ "tooltip": {
+ "mode": "multi",
+ "sort": "none"
+ }
+ },
+ "targets": [
+ {
+ "expr": "sum by (reporter)
(rate(dubbo_inherent_requests_total[1m]))",
+ "legendFormat": "{{reporter}}",
+ "refId": "A"
+ }
+ ],
+ "title": "Request Rate",
+ "type": "timeseries"
+ },
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "prometheus"
+ },
+ "fieldConfig": {
+ "defaults": {
+ "unit": "s"
+ },
+ "overrides": []
+ },
+ "gridPos": {
+ "h": 8,
+ "w": 12,
+ "x": 12,
+ "y": 0
+ },
+ "id": 2,
+ "options": {
+ "legend": {
+ "displayMode": "list",
+ "placement": "bottom",
+ "showLegend": true
+ },
+ "tooltip": {
+ "mode": "multi",
+ "sort": "none"
+ }
+ },
+ "targets": [
+ {
+ "expr": "histogram_quantile(0.95, sum by (le, reporter)
(rate(dubbo_inherent_request_duration_seconds_bucket[5m])))",
+ "legendFormat": "{{reporter}}",
+ "refId": "A"
+ }
+ ],
+ "title": "Request Duration P95",
+ "type": "timeseries"
+ },
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "prometheus"
+ },
+ "fieldConfig": {
+ "defaults": {
+ "unit": "bytes"
+ },
+ "overrides": []
+ },
+ "gridPos": {
+ "h": 8,
+ "w": 12,
+ "x": 0,
+ "y": 8
+ },
+ "id": 3,
+ "options": {
+ "legend": {
+ "displayMode": "list",
+ "placement": "bottom",
+ "showLegend": true
+ },
+ "tooltip": {
+ "mode": "multi",
+ "sort": "none"
+ }
+ },
+ "targets": [
+ {
+ "expr": "sum by (reporter)
(rate(dubbo_inherent_request_size_bytes_sum[5m])) / sum by (reporter)
(rate(dubbo_inherent_request_size_bytes_count[5m]))",
+ "legendFormat": "{{reporter}}",
+ "refId": "A"
+ }
+ ],
+ "title": "Average Request Size",
+ "type": "timeseries"
+ },
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "prometheus"
+ },
+ "fieldConfig": {
+ "defaults": {
+ "unit": "bytes"
+ },
+ "overrides": []
+ },
+ "gridPos": {
+ "h": 8,
+ "w": 12,
+ "x": 12,
+ "y": 8
+ },
+ "id": 4,
+ "options": {
+ "legend": {
+ "displayMode": "list",
+ "placement": "bottom",
+ "showLegend": true
+ },
+ "tooltip": {
+ "mode": "multi",
+ "sort": "none"
+ }
+ },
+ "targets": [
+ {
+ "expr": "sum by (reporter)
(rate(dubbo_inherent_response_size_bytes_sum[5m])) / sum by (reporter)
(rate(dubbo_inherent_response_size_bytes_count[5m]))",
+ "legendFormat": "{{reporter}}",
+ "refId": "A"
+ }
+ ],
+ "title": "Average Response Size",
+ "type": "timeseries"
+ }
+ ],
+ "refresh": "15s",
+ "schemaVersion": 39,
+ "tags": [
+ "dubbo",
+ "application",
+ "observability"
+ ],
+ "templating": {
+ "list": []
+ },
+ "time": {
+ "from": "now-30m",
+ "to": "now"
+ },
+ "timezone": "browser",
+ "title": "Application Standard Metrics",
+ "uid": "dubbo-application-standard-metrics",
+ "version": 1,
+ "weekStart": ""
+ }
---
apiVersion: v1
kind: Service
diff --git a/tests/e2e/activationapp/go.mod b/tests/e2e/activationapp/go.mod
index 8585ec70..87f18ab0 100644
--- a/tests/e2e/activationapp/go.mod
+++ b/tests/e2e/activationapp/go.mod
@@ -18,13 +18,14 @@ module dubbo.apache.org/activation-e2e
go 1.25.0
require (
+ github.com/kdubbo/xds-api v0.0.0-20260814172110-c45be7c324a3
golang.org/x/net v0.53.0
google.golang.org/grpc v1.82.1
+ google.golang.org/protobuf v1.36.11
)
require (
golang.org/x/sys v0.43.0 // indirect
golang.org/x/text v0.36.0 // indirect
google.golang.org/genproto/googleapis/rpc
v0.0.0-20260414002931-afd174a4e478 // indirect
- google.golang.org/protobuf v1.36.11 // indirect
)
diff --git a/tests/e2e/activationapp/go.sum b/tests/e2e/activationapp/go.sum
index 7b006a0a..c8161548 100644
--- a/tests/e2e/activationapp/go.sum
+++ b/tests/e2e/activationapp/go.sum
@@ -10,6 +10,8 @@ github.com/google/go-cmp v0.7.0
h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
github.com/google/go-cmp v0.7.0/go.mod
h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod
h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/kdubbo/xds-api v0.0.0-20260814172110-c45be7c324a3
h1:ypir1ZNYdAKOuWokpObAdUtCDXLD3yqP1AOqCURe3WU=
+github.com/kdubbo/xds-api v0.0.0-20260814172110-c45be7c324a3/go.mod
h1:o2HDUgL1ntaDbWomZ4cD2tt8jBamuG2qRtjXOa1zZ0Q=
go.opentelemetry.io/auto/sdk v1.2.1
h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
go.opentelemetry.io/auto/sdk v1.2.1/go.mod
h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
go.opentelemetry.io/otel v1.43.0
h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I=
diff --git a/tests/e2e/activationapp/main.go b/tests/e2e/activationapp/main.go
index 3eb7988b..0e965c22 100644
--- a/tests/e2e/activationapp/main.go
+++ b/tests/e2e/activationapp/main.go
@@ -16,16 +16,21 @@
package main
import (
+ "context"
"log"
+ "net"
"net/http"
"os"
"time"
+ runtimeapplication "github.com/kdubbo/xds-api/grpc/application"
"golang.org/x/net/http2"
"golang.org/x/net/http2/h2c"
"google.golang.org/grpc"
+ "google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/health"
healthpb "google.golang.org/grpc/health/grpc_health_v1"
+ "google.golang.org/protobuf/types/known/emptypb"
)
func main() {
@@ -35,6 +40,8 @@ func main() {
switch os.Args[1] {
case "server":
runServer()
+ case "telemetry":
+ runTelemetryApplication()
case "sleep":
time.Sleep(5 * time.Second)
default:
@@ -58,3 +65,49 @@ func runServer() {
log.Printf("SERVING address=:8080")
log.Fatal(http.ListenAndServe(":8080", handler))
}
+
+func runTelemetryApplication() {
+ listener, err := net.Listen("tcp", ":8080")
+ if err != nil {
+ log.Fatal(err)
+ }
+ grpcServer, err := runtimeapplication.NewServer()
+ if err != nil {
+ log.Fatal(err)
+ }
+ checker := health.NewServer()
+ checker.SetServingStatus("", healthpb.HealthCheckResponse_SERVING)
+ checker.SetServingStatus("telemetry-load",
healthpb.HealthCheckResponse_SERVING)
+ healthpb.RegisterHealthServer(grpcServer, checker)
+ go func() {
+ log.Printf("SERVING grpc=:8080")
+ log.Fatal(grpcServer.Serve(listener))
+ }()
+
+ connection, err := runtimeapplication.NewClient(
+ "passthrough:///127.0.0.1:8080",
+ grpc.WithTransportCredentials(insecure.NewCredentials()),
+ )
+ if err != nil {
+ log.Fatal(err)
+ }
+ defer connection.Close()
+
+ client := healthpb.NewHealthClient(connection)
+ ticker := time.NewTicker(time.Second)
+ defer ticker.Stop()
+ for range ticker.C {
+ ctx, cancel := context.WithTimeout(context.Background(),
time.Second)
+ _, successErr := client.Check(ctx,
&healthpb.HealthCheckRequest{Service: "telemetry-load"})
+ // The unknown method supplies a stable non-OK status for label
and
+ // aggregation verification without opening another connection.
+ failureErr := connection.Invoke(ctx,
"/telemetry.v1.Probe/Missing", &emptypb.Empty{}, &emptypb.Empty{})
+ cancel()
+ if successErr != nil {
+ log.Printf("health request failed: %v", successErr)
+ }
+ if failureErr == nil {
+ log.Printf("unknown method unexpectedly succeeded")
+ }
+ }
+}
diff --git a/tests/e2e/testdata/telemetry-application.yaml
b/tests/e2e/testdata/telemetry-application.yaml
new file mode 100644
index 00000000..1e1a5ec2
--- /dev/null
+++ b/tests/e2e/testdata/telemetry-application.yaml
@@ -0,0 +1,49 @@
+# 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.
+
+apiVersion: v1
+kind: Namespace
+metadata:
+ name: telemetry-e2e
+---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+ name: telemetry-application
+ namespace: telemetry-e2e
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+ name: telemetry-application
+ namespace: telemetry-e2e
+spec:
+ replicas: 1
+ selector:
+ matchLabels:
+ app.kubernetes.io/name: telemetry-application
+ template:
+ metadata:
+ labels:
+ app.kubernetes.io/name: telemetry-application
+ inherent.dubbo.apache.org/inject: "true"
+ annotations:
+ inject.dubbo.apache.org/templates: grpc-engine
+ spec:
+ serviceAccountName: telemetry-application
+ containers:
+ - name: application
+ image: kdubbo/activation-e2e:telemetry-auto-r2
+ imagePullPolicy: IfNotPresent
+ args:
+ - telemetry
+ ports:
+ - name: grpc
+ containerPort: 8080
+ readinessProbe:
+ httpGet:
+ path: /healthz
+ port: metrics
+ periodSeconds: 2
