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 e4fe0b73 feat: add mesh-native AI service routing (#1007)
e4fe0b73 is described below
commit e4fe0b7338564c947efc7d916d95928c29a62671
Author: mfordjody <[email protected]>
AuthorDate: Sun Aug 9 11:27:28 2026 +0800
feat: add mesh-native AI service routing (#1007)
---
.github/workflows/ci.yml | 3 +-
dubbod/discovery/pkg/bootstrap/server.go | 59 +--
dubbod/discovery/pkg/bootstrap/server_test.go | 32 ++
.../pkg/config/kube/crdclient/types.gen.go | 51 +++
.../kube/gateway/deployment_controller_test.go | 77 ++--
dubbod/discovery/pkg/model/push_context.go | 39 ++
.../pkg/networking/grpcgen/agent_config.go | 430 +++++++++++++++++++++
dubbod/discovery/pkg/networking/grpcgen/rds.go | 52 ++-
.../discovery/pkg/networking/grpcgen/rds_test.go | 170 ++++++++
go.mod | 6 +-
go.sum | 12 +-
manifests/charts/base/files/crd-all.gen.yaml | 365 +++++++++++++++++
manifests/charts/dubbod/files/kube-gateway.yaml | 36 +-
manifests/charts/dubbod/templates/clusterrole.yaml | 12 +
.../dubbod/templates/configmap-injector.yaml | 17 +
pkg/config/schema/collections/collections.gen.go | 19 +
pkg/config/schema/gvk/resources.gen.go | 7 +
pkg/config/schema/gvr/resources.gen.go | 3 +
pkg/config/schema/kind/resources.gen.go | 5 +
pkg/config/schema/kubeclient/resources.gen.go | 13 +
pkg/config/schema/kubetypes/resources.gen.go | 4 +
pkg/config/schema/metadata.yaml | 10 +
pkg/config/validation/validators.go | 151 ++++++++
pkg/config/validation/validators_test.go | 106 +++++
samples/ai-mesh/README.md | 18 +
samples/ai-mesh/backends.yaml | 119 ++++++
samples/ai-mesh/gateway.yaml | 13 +
samples/ai-mesh/routes.yaml | 86 +++++
samples/ai-mesh/services.yaml | 80 ++++
tests/e2e/agentmock/Dockerfile | 9 +
tests/e2e/agentmock/main.go | 94 +++++
tests/e2e/run.sh | 134 ++++++-
32 files changed, 2132 insertions(+), 100 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 79b5fdb0..ecc8143e 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -251,7 +251,7 @@ jobs:
uses: actions/checkout@v4
with:
repository: kdubbo/dxgate
- ref: d683797f44c9ba4c6e1eb6cdb2da21460b00cbfd
+ ref: c8e423501f103c7faa08a0cee7a43680a5054138
path: .e2e/dxgate
- name: Build dxgate activation image
@@ -266,6 +266,7 @@ jobs:
- name: Run real KEDA activation
env:
ACTIVATION_E2E: '1'
+ AI_MESH_E2E: '1'
CLUSTER_NAME: dubbo-activation-ga
DXGATE_IMAGE: kdubbo/dxgate:ga-e2e
IMAGE: kdubbo/dubbod:activation-ga-e2e
diff --git a/dubbod/discovery/pkg/bootstrap/server.go
b/dubbod/discovery/pkg/bootstrap/server.go
index bcb49a48..a6a2faf5 100644
--- a/dubbod/discovery/pkg/bootstrap/server.go
+++ b/dubbod/discovery/pkg/bootstrap/server.go
@@ -500,31 +500,8 @@ func (s *Server) initRegistryEventHandlers() {
log.Infof("processing config change, schema identifier=%s,
GVK=%v, name=%s/%s, event=%s",
schemaID, cfg.GroupVersionKind, cfg.Namespace,
cfg.Name, event)
- var configKind kind.Kind
- switch schemaID {
- case "AuthorizationPolicy":
- configKind = kind.AuthorizationPolicy
- case "PeerAuthentication":
- configKind = kind.PeerAuthentication
- case "RequestAuthentication":
- configKind = kind.RequestAuthentication
- case "ReferenceGrant":
- configKind = kind.ReferenceGrant
- case "GatewayClass":
- configKind = kind.GatewayClass
- case "KubernetesGateway":
- configKind = kind.KubernetesGateway
- case "HTTPRoute":
- configKind = kind.HTTPRoute
- case "BackendTLSPolicy":
- configKind = kind.BackendTLSPolicy
- case "CircuitBreakerPolicy":
- configKind = kind.CircuitBreakerPolicy
- case "FaultInjectionPolicy":
- configKind = kind.FaultInjectionPolicy
- case "ServiceActivationPolicy":
- configKind = kind.ServiceActivationPolicy
- default:
+ configKind, found := configKindForSchemaIdentifier(schemaID)
+ if !found {
log.Debugf("unknown schema identifier %s for %v,
skipping", schemaID, cfg.GroupVersionKind)
return
}
@@ -557,6 +534,7 @@ func (s *Server) initRegistryEventHandlers() {
configKind == kind.ReferenceGrant ||
configKind == kind.CircuitBreakerPolicy ||
configKind == kind.FaultInjectionPolicy ||
+ configKind == kind.DxgateService ||
configKind == kind.ServiceActivationPolicy
// Trigger ConfigUpdate to push changes to all connected proxies
@@ -618,6 +596,37 @@ func (s *Server) initReadinessProbes() {
}
}
+func configKindForSchemaIdentifier(schemaID string) (kind.Kind, bool) {
+ switch schemaID {
+ case "AuthorizationPolicy":
+ return kind.AuthorizationPolicy, true
+ case "PeerAuthentication":
+ return kind.PeerAuthentication, true
+ case "RequestAuthentication":
+ return kind.RequestAuthentication, true
+ case "ReferenceGrant":
+ return kind.ReferenceGrant, true
+ case "GatewayClass":
+ return kind.GatewayClass, true
+ case "KubernetesGateway":
+ return kind.KubernetesGateway, true
+ case "HTTPRoute":
+ return kind.HTTPRoute, true
+ case "BackendTLSPolicy":
+ return kind.BackendTLSPolicy, true
+ case "CircuitBreakerPolicy":
+ return kind.CircuitBreakerPolicy, true
+ case "FaultInjectionPolicy":
+ return kind.FaultInjectionPolicy, true
+ case "DxgateService":
+ return kind.DxgateService, true
+ case "ServiceActivationPolicy":
+ return kind.ServiceActivationPolicy, true
+ default:
+ return 0, false
+ }
+}
+
func (s *Server) initMulticluster(args *DubboArgs) {
if s.kubeClient == nil {
return
diff --git a/dubbod/discovery/pkg/bootstrap/server_test.go
b/dubbod/discovery/pkg/bootstrap/server_test.go
new file mode 100644
index 00000000..a04ae89c
--- /dev/null
+++ b/dubbod/discovery/pkg/bootstrap/server_test.go
@@ -0,0 +1,32 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package bootstrap
+
+import (
+ "testing"
+
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/kind"
+)
+
+func TestConfigKindForSchemaIdentifierIncludesDxgateService(t *testing.T) {
+ got, found := configKindForSchemaIdentifier("DxgateService")
+ if !found {
+ t.Fatal("DxgateService schema identifier was not mapped")
+ }
+ if got != kind.DxgateService {
+ t.Fatalf("kind = %v, want %v", got, kind.DxgateService)
+ }
+}
diff --git a/dubbod/discovery/pkg/config/kube/crdclient/types.gen.go
b/dubbod/discovery/pkg/config/kube/crdclient/types.gen.go
index 6eaf61e3..731cc4db 100755
--- a/dubbod/discovery/pkg/config/kube/crdclient/types.gen.go
+++ b/dubbod/discovery/pkg/config/kube/crdclient/types.gen.go
@@ -50,6 +50,11 @@ func create(c kube.Client, cfg config.Config, objMeta
metav1.ObjectMeta) (metav1
ObjectMeta: objMeta,
Spec:
*(cfg.Spec.(*githubcomkdubboapinetworkingv1alpha3.CircuitBreakerPolicy)),
}, metav1.CreateOptions{})
+ case gvk.DxgateService:
+ return
c.Dubbo().NetworkingV1alpha3().DxgateServices(cfg.Namespace).Create(context.TODO(),
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.DxgateService{
+ ObjectMeta: objMeta,
+ Spec:
*(cfg.Spec.(*githubcomkdubboapinetworkingv1alpha3.DxgateService)),
+ }, metav1.CreateOptions{})
case gvk.FaultInjectionPolicy:
return
c.Dubbo().NetworkingV1alpha3().FaultInjectionPolicies(cfg.Namespace).Create(context.TODO(),
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.FaultInjectionPolicy{
ObjectMeta: objMeta,
@@ -127,6 +132,11 @@ func update(c kube.Client, cfg config.Config, objMeta
metav1.ObjectMeta) (metav1
ObjectMeta: objMeta,
Spec:
*(cfg.Spec.(*githubcomkdubboapinetworkingv1alpha3.CircuitBreakerPolicy)),
}, metav1.UpdateOptions{})
+ case gvk.DxgateService:
+ return
c.Dubbo().NetworkingV1alpha3().DxgateServices(cfg.Namespace).Update(context.TODO(),
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.DxgateService{
+ ObjectMeta: objMeta,
+ Spec:
*(cfg.Spec.(*githubcomkdubboapinetworkingv1alpha3.DxgateService)),
+ }, metav1.UpdateOptions{})
case gvk.FaultInjectionPolicy:
return
c.Dubbo().NetworkingV1alpha3().FaultInjectionPolicies(cfg.Namespace).Update(context.TODO(),
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.FaultInjectionPolicy{
ObjectMeta: objMeta,
@@ -204,6 +214,11 @@ func updateStatus(c kube.Client, cfg config.Config,
objMeta metav1.ObjectMeta) (
ObjectMeta: objMeta,
Status:
*(cfg.Status.(*githubcomkdubboapimetav1alpha1.DubboStatus)),
}, metav1.UpdateOptions{})
+ case gvk.DxgateService:
+ return
c.Dubbo().NetworkingV1alpha3().DxgateServices(cfg.Namespace).UpdateStatus(context.TODO(),
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.DxgateService{
+ ObjectMeta: objMeta,
+ Status:
*(cfg.Status.(*githubcomkdubboapimetav1alpha1.DubboStatus)),
+ }, metav1.UpdateOptions{})
case gvk.FaultInjectionPolicy:
return
c.Dubbo().NetworkingV1alpha3().FaultInjectionPolicies(cfg.Namespace).UpdateStatus(context.TODO(),
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.FaultInjectionPolicy{
ObjectMeta: objMeta,
@@ -309,6 +324,21 @@ func patch(c kube.Client, orig config.Config, origMeta
metav1.ObjectMeta, mod co
}
return
c.Dubbo().NetworkingV1alpha3().CircuitBreakerPolicies(orig.Namespace).
Patch(context.TODO(), orig.Name, typ, patchBytes,
metav1.PatchOptions{FieldManager: "pilot-discovery"})
+ case gvk.DxgateService:
+ oldRes :=
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.DxgateService{
+ ObjectMeta: origMeta,
+ Spec:
*(orig.Spec.(*githubcomkdubboapinetworkingv1alpha3.DxgateService)),
+ }
+ modRes :=
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.DxgateService{
+ ObjectMeta: modMeta,
+ Spec:
*(mod.Spec.(*githubcomkdubboapinetworkingv1alpha3.DxgateService)),
+ }
+ patchBytes, err := genPatchBytes(oldRes, modRes, typ)
+ if err != nil {
+ return nil, err
+ }
+ return
c.Dubbo().NetworkingV1alpha3().DxgateServices(orig.Namespace).
+ Patch(context.TODO(), orig.Name, typ, patchBytes,
metav1.PatchOptions{FieldManager: "pilot-discovery"})
case gvk.FaultInjectionPolicy:
oldRes :=
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.FaultInjectionPolicy{
ObjectMeta: origMeta,
@@ -491,6 +521,8 @@ func delete(c kube.Client, typ config.GroupVersionKind,
name, namespace string,
return
c.GatewayAPI().GatewayV1().BackendTLSPolicies(namespace).Delete(context.TODO(),
name, deleteOptions)
case gvk.CircuitBreakerPolicy:
return
c.Dubbo().NetworkingV1alpha3().CircuitBreakerPolicies(namespace).Delete(context.TODO(),
name, deleteOptions)
+ case gvk.DxgateService:
+ return
c.Dubbo().NetworkingV1alpha3().DxgateServices(namespace).Delete(context.TODO(),
name, deleteOptions)
case gvk.FaultInjectionPolicy:
return
c.Dubbo().NetworkingV1alpha3().FaultInjectionPolicies(namespace).Delete(context.TODO(),
name, deleteOptions)
case gvk.GatewayClass:
@@ -648,6 +680,25 @@ var translationMap = map[config.GroupVersionKind]func(r
runtime.Object) config.C
Spec: &obj.Spec,
}
},
+ gvk.DxgateService: func(r runtime.Object) config.Config {
+ obj :=
r.(*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.DxgateService)
+ return config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.DxgateService,
+ Name: obj.Name,
+ Namespace: obj.Namespace,
+ Labels: obj.Labels,
+ Annotations: obj.Annotations,
+ ResourceVersion: obj.ResourceVersion,
+ CreationTimestamp: obj.CreationTimestamp.Time,
+ OwnerReferences: obj.OwnerReferences,
+ UID: string(obj.UID),
+ Generation: obj.Generation,
+ },
+ Spec: &obj.Spec,
+ Status: &obj.Status,
+ }
+ },
gvk.EndpointSlice: func(r runtime.Object) config.Config {
obj := r.(*k8sioapidiscoveryv1.EndpointSlice)
return config.Config{
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 826f4701..cd259594 100644
--- a/dubbod/discovery/pkg/config/kube/gateway/deployment_controller_test.go
+++ b/dubbod/discovery/pkg/config/kube/gateway/deployment_controller_test.go
@@ -691,8 +691,8 @@ func TestKubeGatewayTemplateRendersDxgateResources(t
*testing.T) {
if err != nil {
t.Fatal(err)
}
- if len(rendered) != 4 {
- t.Fatalf("expected 4 rendered resources, got %d", len(rendered))
+ if len(rendered) != 6 {
+ t.Fatalf("expected 6 rendered resources, got %d", len(rendered))
}
for i, doc := range rendered {
var obj map[string]any
@@ -706,25 +706,30 @@ func TestKubeGatewayTemplateRendersDxgateResources(t
*testing.T) {
if !strings.Contains(rendered[0], "bootstrap.json") ||
!strings.Contains(rendered[0], `"xds_address":
"http://dubbod.dubbo-system.svc:26010"`) {
t.Fatalf("configmap did not render dxgate bootstrap:\n%s",
rendered[0])
}
- if !strings.Contains(rendered[2], "image: kdubbo/dxgate:test") {
- t.Fatalf("deployment did not render dxgate image:\n%s",
rendered[2])
+ if !strings.Contains(rendered[1], "automountServiceAccountToken: true")
||
+ !strings.Contains(rendered[2], "resources: [\"secrets\"]") ||
+ !strings.Contains(rendered[3], "kind: RoleBinding") {
+ t.Fatalf("credential Secret RBAC not rendered:\n%s",
strings.Join(rendered[1:4], "\n---\n"))
}
- if !strings.Contains(rendered[2], "DXGATE_BOOTSTRAP") ||
strings.Contains(rendered[2], "DXGATE_STATIC_CONFIG") {
- t.Fatalf("deployment did not switch from static config to
bootstrap:\n%s", rendered[2])
+ if !strings.Contains(rendered[4], "image: kdubbo/dxgate:test") {
+ t.Fatalf("deployment did not render dxgate image:\n%s",
rendered[4])
}
- if !strings.Contains(rendered[2], `proxyless.dubbo.apache.org/inject:
"true"`) {
- t.Fatalf("deployment pod template did not enable proxyless
injection for dxgate mTLS certs:\n%s", rendered[2])
+ if !strings.Contains(rendered[4], "DXGATE_BOOTSTRAP") ||
strings.Contains(rendered[4], "DXGATE_STATIC_CONFIG") {
+ t.Fatalf("deployment did not switch from static config to
bootstrap:\n%s", rendered[4])
}
- if !strings.Contains(rendered[2], "inject.dubbo.apache.org/templates:
grpc-engine") {
- t.Fatalf("deployment pod template did not request grpc-engine
injection:\n%s", rendered[2])
+ if !strings.Contains(rendered[4], `proxyless.dubbo.apache.org/inject:
"true"`) {
+ t.Fatalf("deployment pod template did not enable proxyless
injection for dxgate mTLS certs:\n%s", rendered[4])
}
- if !strings.Contains(rendered[2], `prometheus.io/scrape: "true"`) ||
- !strings.Contains(rendered[2], "prometheus.io/path: /metrics")
||
- !strings.Contains(rendered[2], `prometheus.io/port: "26021"`) {
- t.Fatalf("deployment pod template did not render prometheus
scrape annotations:\n%s", rendered[2])
+ if !strings.Contains(rendered[4], "inject.dubbo.apache.org/templates:
grpc-engine") {
+ t.Fatalf("deployment pod template did not request grpc-engine
injection:\n%s", rendered[4])
}
- if strings.Contains(rendered[2], "DXGATE_OTEL_ENDPOINT") {
- t.Fatalf("deployment rendered OTEL endpoint without
annotation:\n%s", rendered[2])
+ if !strings.Contains(rendered[4], `prometheus.io/scrape: "true"`) ||
+ !strings.Contains(rendered[4], "prometheus.io/path: /metrics")
||
+ !strings.Contains(rendered[4], `prometheus.io/port: "26021"`) {
+ t.Fatalf("deployment pod template did not render prometheus
scrape annotations:\n%s", rendered[4])
+ }
+ if strings.Contains(rendered[4], "DXGATE_OTEL_ENDPOINT") {
+ t.Fatalf("deployment rendered OTEL endpoint without
annotation:\n%s", rendered[4])
}
for _, want := range []string{
"DXGATE_GATEWAY_NAME",
@@ -737,21 +742,21 @@ func TestKubeGatewayTemplateRendersDxgateResources(t
*testing.T) {
"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[4], want) {
+ t.Fatalf("deployment did not render observability env
%q:\n%s", want, rendered[4])
}
}
- 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[4], "app.kubernetes.io/instance:
public-dubbo") {
+ t.Fatalf("deployment did not render stable dxgate instance
label:\n%s", rendered[4])
}
- if !strings.Contains(rendered[3], "targetPort: 15080") {
- t.Fatalf("service did not target grpc-inbound port:\n%s",
rendered[3])
+ if !strings.Contains(rendered[5], "targetPort: 15080") {
+ t.Fatalf("service did not target grpc-inbound port:\n%s",
rendered[5])
}
- 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])
+ if !strings.Contains(rendered[5], `proxyless.dubbo.apache.org/inject:
"false"`) {
+ t.Fatalf("service did not opt out of proxyless targetPort
rewriting:\n%s", rendered[5])
}
- if !strings.Contains(rendered[3], "app.kubernetes.io/instance:
public-dubbo") {
- t.Fatalf("service did not render stable dxgate instance
selector:\n%s", rendered[3])
+ if !strings.Contains(rendered[5], "app.kubernetes.io/instance:
public-dubbo") {
+ t.Fatalf("service did not render stable dxgate instance
selector:\n%s", rendered[5])
}
input.OtelEndpoint = "http://tracing.dubbo-system.svc:4317";
@@ -759,9 +764,9 @@ func TestKubeGatewayTemplateRendersDxgateResources(t
*testing.T) {
if err != nil {
t.Fatal(err)
}
- if !strings.Contains(rendered[2], "DXGATE_OTEL_ENDPOINT") ||
- !strings.Contains(rendered[2], `value:
"http://tracing.dubbo-system.svc:4317"`) {
- t.Fatalf("deployment did not render dxgate OTEL endpoint:\n%s",
rendered[2])
+ if !strings.Contains(rendered[4], "DXGATE_OTEL_ENDPOINT") ||
+ !strings.Contains(rendered[4], `value:
"http://tracing.dubbo-system.svc:4317"`) {
+ t.Fatalf("deployment did not render dxgate OTEL endpoint:\n%s",
rendered[4])
}
}
@@ -803,8 +808,8 @@ func
TestKubeGatewayTemplateRendersHighAvailabilityResources(t *testing.T) {
if err != nil {
t.Fatal(err)
}
- if len(rendered) != 5 {
- t.Fatalf("rendered %d resources, want 5 including the
disruption budget", len(rendered))
+ if len(rendered) != 7 {
+ t.Fatalf("rendered %d resources, want 7 including credential
RBAC and disruption budget", len(rendered))
}
joined := strings.Join(rendered, "\n---\n")
for _, want := range []string{
@@ -825,8 +830,8 @@ func
TestKubeGatewayTemplateRendersHighAvailabilityResources(t *testing.T) {
if err != nil {
t.Fatal(err)
}
- if len(rendered) != 4 {
- t.Fatalf("rendered %d resources at one replica, want 4 without
a disruption budget", len(rendered))
+ if len(rendered) != 6 {
+ t.Fatalf("rendered %d resources at one replica, want 6 without
a disruption budget", len(rendered))
}
joined = strings.Join(rendered, "\n---\n")
if strings.Contains(joined, "kind: PodDisruptionBudget") {
@@ -934,7 +939,7 @@ func TestKubeGatewayTemplateRendersActivationEnv(t
*testing.T) {
if err != nil {
t.Fatal(err)
}
- deployment := rendered[2]
+ deployment := rendered[4]
if !strings.Contains(deployment, "DXGATE_ACTIVATION_CONTROL_PLANE") ||
!strings.Contains(deployment,
"dubbod-activation-replicas.dubbo-system.svc.cluster.local:26030") {
t.Fatalf("deployment did not render the activation control
plane:\n%s", deployment)
@@ -954,7 +959,7 @@ func TestKubeGatewayTemplateRendersActivationEnv(t
*testing.T) {
if err != nil {
t.Fatal(err)
}
- if strings.Contains(off[2], "DXGATE_ACTIVATION") {
- t.Fatalf("activation env leaked into a gateway with activation
disabled:\n%s", off[2])
+ if strings.Contains(off[4], "DXGATE_ACTIVATION") {
+ t.Fatalf("activation env leaked into a gateway with activation
disabled:\n%s", off[4])
}
}
diff --git a/dubbod/discovery/pkg/model/push_context.go
b/dubbod/discovery/pkg/model/push_context.go
index b9bf9a9c..72742d38 100644
--- a/dubbod/discovery/pkg/model/push_context.go
+++ b/dubbod/discovery/pkg/model/push_context.go
@@ -75,6 +75,7 @@ type PushContext struct {
ServiceIndex serviceIndex
virtualServiceIndex virtualServiceIndex
httpRouteIndex httpRouteIndex
+ dxgateServiceIndex dxgateServiceIndex
backendTLSPolicyIndex backendTLSPolicyIndex
faultInjectionIndex faultInjectionPolicyIndex
serviceActivationIndex serviceActivationPolicyIndex
@@ -165,6 +166,10 @@ type httpRouteIndex struct {
hostToRoutes map[host.Name][]config.Config
}
+type dxgateServiceIndex struct {
+ byNamespace map[string]map[string]config.Config
+}
+
type backendTLSPolicyIndex struct {
serviceTLS map[string]BackendTLSSettings
}
@@ -193,6 +198,7 @@ func NewPushContext() *PushContext {
return &PushContext{
ServiceIndex: newServiceIndex(),
virtualServiceIndex: newVirtualServiceIndex(),
+ dxgateServiceIndex: dxgateServiceIndex{byNamespace:
map[string]map[string]config.Config{}},
backendTLSPolicyIndex: backendTLSPolicyIndex{serviceTLS:
map[string]BackendTLSSettings{}},
serviceActivationIndex: serviceActivationPolicyIndex{
services: map[string][]string{},
@@ -628,6 +634,7 @@ func (ps *PushContext) createNewContext(env *Environment) {
// Initialize Kubernetes Gateway API resources if the controller is
enabled.
ps.initKubernetesGateways(env)
ps.initHTTPRoutes(env)
+ ps.initDxgateServices(env)
ps.initBackendTLSPolicies(env)
ps.initFaultInjectionPolicies(env)
ps.initServiceActivationPolicies(env)
@@ -687,6 +694,13 @@ func (ps *PushContext) updateContext(env *Environment,
oldPushContext *PushConte
ps.httpRouteIndex = oldPushContext.httpRouteIndex
}
+ dxgateServicesChanged := pushReq != nil &&
HasConfigsOfKind(pushReq.ConfigsUpdated, kind.DxgateService)
+ if dxgateServicesChanged {
+ ps.initDxgateServices(env)
+ } else {
+ ps.dxgateServiceIndex = oldPushContext.dxgateServiceIndex
+ }
+
backendTLSPoliciesChanged := pushReq != nil &&
HasConfigsOfKind(pushReq.ConfigsUpdated, kind.BackendTLSPolicy)
if backendTLSPoliciesChanged {
log.Debugf("BackendTLSPolicies changed, re-initializing
BackendTLSPolicy index")
@@ -866,6 +880,31 @@ func (ps *PushContext) initHTTPRoutes(env *Environment) {
}
}
+func (ps *PushContext) initDxgateServices(env *Environment) {
+ services := sortConfigByCreationTime(env.List(gvk.DxgateService,
NamespaceAll))
+ index := make(map[string]map[string]config.Config)
+ for _, service := range services {
+ if index[service.Namespace] == nil {
+ index[service.Namespace] =
make(map[string]config.Config)
+ }
+ index[service.Namespace][service.Name] = service
+ }
+ ps.dxgateServiceIndex = dxgateServiceIndex{byNamespace: index}
+}
+
+// DxgateService resolves the mesh-native backend named by an HTTPRoute.
+func (ps *PushContext) DxgateService(namespace, name string) (config.Config,
bool) {
+ if ps == nil {
+ return config.Config{}, false
+ }
+ services := ps.dxgateServiceIndex.byNamespace[namespace]
+ if services == nil {
+ return config.Config{}, false
+ }
+ service, found := services[name]
+ return service, found
+}
+
// HTTPRouteForHost returns HTTPRoutes that match the given hostname.
func (ps *PushContext) HTTPRouteForHost(hostname host.Name) []config.Config {
var routes []config.Config
diff --git a/dubbod/discovery/pkg/networking/grpcgen/agent_config.go
b/dubbod/discovery/pkg/networking/grpcgen/agent_config.go
new file mode 100644
index 00000000..86847fd3
--- /dev/null
+++ b/dubbod/discovery/pkg/networking/grpcgen/agent_config.go
@@ -0,0 +1,430 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package grpcgen
+
+import (
+ "fmt"
+ "sort"
+ "strings"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+ "github.com/apache/dubbo-kubernetes/pkg/config"
+ networking "github.com/kdubbo/api/networking/v1alpha3"
+ route "github.com/kdubbo/xds-api/route/v1"
+ gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
+)
+
+const (
+ dxgateServiceGroup = "networking.dubbo.apache.org"
+ dxgateServiceKind = "DxgateService"
+)
+
+func isDxgateServiceBackend(ref gatewayv1.HTTPBackendRef) bool {
+ group := ""
+ if ref.Group != nil {
+ group = string(*ref.Group)
+ }
+ kind := "Service"
+ if ref.Kind != nil {
+ kind = string(*ref.Kind)
+ }
+ return group == dxgateServiceGroup && kind == dxgateServiceKind
+}
+
+func ruleUsesDxgateService(rule gatewayv1.HTTPRouteRule) bool {
+ for _, ref := range rule.BackendRefs {
+ if isDxgateServiceBackend(ref) {
+ return true
+ }
+ }
+ return false
+}
+
+func buildAgentConfig(push *model.PushContext, httpRoutes []config.Config)
*route.AgentConfig {
+ if push == nil {
+ return nil
+ }
+ out := &route.AgentConfig{}
+ providers := map[string]bool{}
+ backends := map[string]bool{}
+ policies := map[string]bool{}
+
+ for _, routeConfig := range httpRoutes {
+ spec, ok := routeConfig.Spec.(*gatewayv1.HTTPRouteSpec)
+ if !ok {
+ continue
+ }
+ for ruleIndex, rule := range spec.Rules {
+ if !ruleUsesDxgateService(rule) {
+ continue
+ }
+ if len(rule.BackendRefs) == 0 {
+ continue
+ }
+
+ var protocol route.AgentProtocol
+ var weighted []*route.WeightedBackend
+ valid := true
+ for _, ref := range rule.BackendRefs {
+ if !isDxgateServiceBackend(ref) {
+ log.Warnf("HTTPRoute %s/%s rule[%d]
mixes Service and DxgateService backends", routeConfig.Namespace,
routeConfig.Name, ruleIndex)
+ valid = false
+ break
+ }
+ namespace := routeConfig.Namespace
+ if ref.Namespace != nil {
+ namespace = string(*ref.Namespace)
+ }
+ if namespace != routeConfig.Namespace {
+ log.Warnf("HTTPRoute %s/%s rule[%d]
cross-namespace DxgateService reference %s/%s is not supported",
routeConfig.Namespace, routeConfig.Name, ruleIndex, namespace, ref.Name)
+ valid = false
+ break
+ }
+ serviceConfig, found :=
push.DxgateService(namespace, string(ref.Name))
+ if !found {
+ log.Warnf("HTTPRoute %s/%s rule[%d]
references missing DxgateService %s/%s", routeConfig.Namespace,
routeConfig.Name, ruleIndex, namespace, ref.Name)
+ valid = false
+ break
+ }
+ service, ok :=
serviceConfig.Spec.(*networking.DxgateService)
+ if !ok {
+ valid = false
+ break
+ }
+ compiled, serviceProtocol, err :=
compileDxgateService(serviceConfig, service)
+ if err != nil {
+ log.Warnf("DxgateService %s/%s cannot
be compiled: %v", namespace, ref.Name, err)
+ valid = false
+ break
+ }
+ if protocol !=
route.AgentProtocol_AGENT_PROTOCOL_UNSPECIFIED && protocol != serviceProtocol {
+ log.Warnf("HTTPRoute %s/%s rule[%d]
mixes DxgateService protocols", routeConfig.Namespace, routeConfig.Name,
ruleIndex)
+ valid = false
+ break
+ }
+ protocol = serviceProtocol
+ if compiled.provider != nil &&
!providers[compiled.provider.Name] {
+ out.Providers = append(out.Providers,
compiled.provider)
+ providers[compiled.provider.Name] = true
+ }
+ if compiled.policy != nil &&
!policies[compiled.policy.Name] {
+ out.Policies = append(out.Policies,
compiled.policy)
+ policies[compiled.policy.Name] = true
+ }
+ weight := uint32(1)
+ if ref.Weight != nil && *ref.Weight > 0 {
+ weight = uint32(*ref.Weight)
+ }
+ for _, backend := range compiled.backends {
+ if !backends[backend.Name] {
+ out.Backends =
append(out.Backends, backend)
+ backends[backend.Name] = true
+ }
+ weighted = append(weighted,
&route.WeightedBackend{Name: backend.Name, Weight: weight})
+ }
+ }
+ if !valid || len(weighted) == 0 {
+ continue
+ }
+ out.AgentRoutes = append(out.AgentRoutes,
&route.AgentRoute{
+ Name: fmt.Sprintf("%s/%s/%d",
routeConfig.Namespace, routeConfig.Name, ruleIndex),
+ Protocol: protocol,
+ Matches:
compileAgentMatches(spec.Hostnames, rule.Matches),
+ WeightedBackends: weighted,
+ Policies: policyNames(weighted,
out.Backends),
+ Rewrite:
compileAgentRewrite(rule.Filters),
+ })
+ }
+ }
+ if len(out.AgentRoutes) == 0 {
+ return nil
+ }
+ sort.Slice(out.Providers, func(i, j int) bool { return
out.Providers[i].Name < out.Providers[j].Name })
+ sort.Slice(out.Backends, func(i, j int) bool { return
out.Backends[i].Name < out.Backends[j].Name })
+ sort.Slice(out.Policies, func(i, j int) bool { return
out.Policies[i].Name < out.Policies[j].Name })
+ sort.Slice(out.AgentRoutes, func(i, j int) bool { return
out.AgentRoutes[i].Name < out.AgentRoutes[j].Name })
+ return out
+}
+
+type compiledDxgateService struct {
+ provider *route.AgentProvider
+ backends []*route.AgentBackend
+ policy *route.AgentPolicy
+}
+
+func compileDxgateService(cfg config.Config, service
*networking.DxgateService) (compiledDxgateService, route.AgentProtocol, error) {
+ prefix := fmt.Sprintf("%s.%s", cfg.Name, cfg.Namespace)
+ policyName := prefix + ".policy"
+ policy := compileAgentPolicy(policyName, cfg.Namespace,
service.GetPolicies())
+ policyRefs := []string(nil)
+ if policy != nil {
+ policyRefs = []string{policyName}
+ }
+
+ switch {
+ case service.GetAi() != nil:
+ ai := service.GetAi()
+ if ai.GetProvider() == nil {
+ return compiledDxgateService{}, 0, fmt.Errorf("provider
is not set")
+ }
+ providerName := prefix + ".provider"
+ provider := &route.AgentProvider{
+ Name: providerName,
+ BaseUrl: ai.GetEndpoint(),
+ Credential: compileSecretReference(cfg.Namespace,
ai.GetProvider().GetCredential()),
+ RequestHeaders:
compileHeaderValues(ai.GetProvider().GetRequestHeaders()),
+ }
+ switch {
+ case ai.GetProvider().GetOpenai() != nil:
+ provider.Kind = route.AgentProviderKind_OPENAI
+ case ai.GetProvider().GetAnthropic() != nil:
+ provider.Kind = route.AgentProviderKind_ANTHROPIC
+ default:
+ return compiledDxgateService{}, 0, fmt.Errorf("provider
is not set")
+ }
+ backend := &route.AgentBackend{
+ Name: prefix,
+ Backend: &route.AgentBackend_Llm{Llm: &route.LLMBackend{
+ Provider: providerName,
+ Models: append([]string(nil),
ai.GetModels()...),
+ ModelRewrites:
cloneStringMap(ai.GetModelRewrites()),
+ }},
+ Policies: policyRefs,
+ }
+ return compiledDxgateService{provider: provider, backends:
[]*route.AgentBackend{backend}, policy: policy}, route.AgentProtocol_LLM, nil
+ case service.GetMcp() != nil:
+ backends := make([]*route.AgentBackend, 0,
len(service.GetMcp().GetTargets()))
+ for _, target := range service.GetMcp().GetTargets() {
+ static := target.GetStatic()
+ if static == nil || static.GetBackendRef() == nil {
+ continue
+ }
+ namespace := static.GetBackendRef().GetNamespace()
+ if namespace == "" {
+ namespace = cfg.Namespace
+ }
+ backends = append(backends, &route.AgentBackend{
+ Name: fmt.Sprintf("%s.%s", prefix,
target.GetName()),
+ Backend: &route.AgentBackend_Mcp{Mcp:
&route.MCPBackend{
+ Endpoint:
serviceEndpoint(static.GetBackendRef().GetName(), namespace, static.GetPort(),
static.GetPath()),
+ Tools: append([]string(nil),
target.GetTools()...),
+ }},
+ Policies: policyRefs,
+ })
+ }
+ return compiledDxgateService{backends: backends, policy:
policy}, route.AgentProtocol_MCP, nil
+ case service.GetA2A() != nil:
+ a2a := service.GetA2A()
+ endpoint := ""
+ if ref := a2a.GetBackendRef(); ref != nil {
+ namespace := ref.GetNamespace()
+ if namespace == "" {
+ namespace = cfg.Namespace
+ }
+ endpoint = serviceEndpoint(ref.GetName(), namespace,
a2a.GetPort(), a2a.GetPath())
+ } else {
+ endpoint = hostEndpoint(a2a.GetHost(), a2a.GetPort(),
a2a.GetPath())
+ }
+ backend := &route.AgentBackend{
+ Name: prefix,
+ Backend: &route.AgentBackend_A2A{A2A: &route.A2ABackend{
+ Endpoint: endpoint,
+ Agent: a2a.GetAgent(),
+ }},
+ Policies: policyRefs,
+ }
+ return compiledDxgateService{backends:
[]*route.AgentBackend{backend}, policy: policy}, route.AgentProtocol_A2A, nil
+ default:
+ return compiledDxgateService{}, 0, fmt.Errorf("service type is
not set")
+ }
+}
+
+func serviceEndpoint(name, namespace string, port uint32, path string) string {
+ return hostEndpoint(fmt.Sprintf("%s.%s.svc.cluster.local", name,
namespace), port, path)
+}
+
+func hostEndpoint(host string, port uint32, path string) string {
+ return fmt.Sprintf("http://%s:%d%s";, host, port,
normalizeEndpointPath(path))
+}
+
+func normalizeEndpointPath(path string) string {
+ if path == "" || path == "/" {
+ return ""
+ }
+ return "/" + strings.Trim(path, "/")
+}
+
+func compileAgentMatches(hostnames []gatewayv1.Hostname, matches
[]gatewayv1.HTTPRouteMatch) []*route.AgentRouteMatch {
+ hosts := make([]string, 0, len(hostnames))
+ for _, hostname := range hostnames {
+ hosts = append(hosts, string(hostname))
+ }
+ if len(hosts) == 0 {
+ hosts = []string{""}
+ }
+ if len(matches) == 0 {
+ matches = []gatewayv1.HTTPRouteMatch{{}}
+ }
+ out := make([]*route.AgentRouteMatch, 0, len(hosts)*len(matches))
+ for _, hostname := range hosts {
+ for _, match := range matches {
+ compiled := &route.AgentRouteMatch{
+ Host: hostname,
+ Path: compileAgentPathMatch(match.Path),
+ Headers: make([]*route.AgentHeaderMatch, 0,
len(match.Headers)),
+ }
+ if match.Method != nil {
+ compiled.Method = string(*match.Method)
+ }
+ for _, header := range match.Headers {
+ compiled.Headers = append(compiled.Headers,
&route.AgentHeaderMatch{
+ Name: string(header.Name), Value:
header.Value,
+ })
+ }
+ out = append(out, compiled)
+ }
+ }
+ return out
+}
+
+func compileAgentPathMatch(match *gatewayv1.HTTPPathMatch)
*route.AgentPathMatch {
+ if match == nil || match.Value == nil {
+ return &route.AgentPathMatch{Match:
&route.AgentPathMatch_Prefix{Prefix: "/"}}
+ }
+ if match.Type != nil && *match.Type == gatewayv1.PathMatchExact {
+ return &route.AgentPathMatch{Match:
&route.AgentPathMatch_Exact{Exact: *match.Value}}
+ }
+ return &route.AgentPathMatch{Match:
&route.AgentPathMatch_Prefix{Prefix: *match.Value}}
+}
+
+func compileAgentRewrite(filters []gatewayv1.HTTPRouteFilter)
*route.PathRewrite {
+ for _, filter := range filters {
+ if filter.Type != gatewayv1.HTTPRouteFilterURLRewrite ||
filter.URLRewrite == nil || filter.URLRewrite.Path == nil {
+ continue
+ }
+ if filter.URLRewrite.Path.Type ==
gatewayv1.PrefixMatchHTTPPathModifier &&
filter.URLRewrite.Path.ReplacePrefixMatch != nil {
+ return &route.PathRewrite{ReplacePrefixMatch:
*filter.URLRewrite.Path.ReplacePrefixMatch}
+ }
+ }
+ return nil
+}
+
+func compileAgentPolicy(name, namespace string, in
*networking.DxgateServicePolicies) *route.AgentPolicy {
+ if in == nil {
+ return nil
+ }
+ out := &route.AgentPolicy{
+ Name: name,
+ Timeout: in.GetTimeout(),
+ MaxBodyBytes: uint64(max(in.GetMaxBodyBytes(), 0)),
+ RequestHeaders: compileHeaderTransform(in.GetRequestHeaders()),
+ ResponseHeaders:
compileHeaderTransform(in.GetResponseHeaders()),
+ }
+ if auth := in.GetAuth(); auth != nil {
+ out.Auth = &route.ClientAuthPolicy{
+ Header: auth.GetHeader(),
+ SecretRef: compileSecretReference(namespace,
auth.GetSecretRef()),
+ }
+ }
+ if limit := in.GetRateLimit(); limit != nil {
+ out.RateLimit = &route.AgentRateLimitPolicy{
+ Requests: limit.GetRequests(), Window:
limit.GetWindow(),
+ Key: compileRateLimitKey(limit.GetKey()), Header:
limit.GetHeader(),
+ }
+ }
+ if limit := in.GetTokenLimit(); limit != nil {
+ out.TokenLimit = &route.AgentTokenLimitPolicy{
+ Tokens: limit.GetTokens(), Window: limit.GetWindow(),
+ Key: compileRateLimitKey(limit.GetKey()), Header:
limit.GetHeader(),
+ }
+ }
+ if retry := in.GetRetry(); retry != nil {
+ out.Retry = &route.AgentRetryPolicy{
+ Attempts: retry.GetAttempts(), StatusCodes:
append([]uint32(nil), retry.GetStatusCodes()...),
+ }
+ }
+ return out
+}
+
+func compileSecretReference(namespace string, ref
*networking.SecretKeyReference) *route.SecretKeyReference {
+ if ref == nil {
+ return nil
+ }
+ return &route.SecretKeyReference{Namespace: namespace, Name:
ref.GetName(), Key: ref.GetKey()}
+}
+
+func compileHeaderTransform(in *networking.HeaderTransform)
*route.AgentHeaderTransform {
+ if in == nil {
+ return nil
+ }
+ return &route.AgentHeaderTransform{
+ Add: compileHeaderValues(in.GetAdd()),
+ Remove: append([]string(nil), in.GetRemove()...),
+ }
+}
+
+func compileHeaderValues(in []*networking.HeaderValue)
[]*route.AgentHeaderValue {
+ out := make([]*route.AgentHeaderValue, 0, len(in))
+ for _, value := range in {
+ if value != nil {
+ out = append(out, &route.AgentHeaderValue{Name:
value.GetName(), Value: value.GetValue()})
+ }
+ }
+ return out
+}
+
+func compileRateLimitKey(in networking.RateLimitKey) route.AgentRateLimitKey {
+ switch in {
+ case networking.RateLimitKey_BACKEND:
+ return route.AgentRateLimitKey_BACKEND
+ case networking.RateLimitKey_HEADER:
+ return route.AgentRateLimitKey_HEADER
+ default:
+ return route.AgentRateLimitKey_ROUTE
+ }
+}
+
+func policyNames(weighted []*route.WeightedBackend, backends
[]*route.AgentBackend) []string {
+ seen := map[string]bool{}
+ var out []string
+ for _, selected := range weighted {
+ for _, backend := range backends {
+ if backend.Name != selected.Name {
+ continue
+ }
+ for _, policy := range backend.Policies {
+ if !seen[policy] {
+ out = append(out, policy)
+ seen[policy] = true
+ }
+ }
+ }
+ }
+ sort.Strings(out)
+ return out
+}
+
+func cloneStringMap(in map[string]string) map[string]string {
+ if len(in) == 0 {
+ return nil
+ }
+ out := make(map[string]string, len(in))
+ for key, value := range in {
+ out[key] = value
+ }
+ return out
+}
diff --git a/dubbod/discovery/pkg/networking/grpcgen/rds.go
b/dubbod/discovery/pkg/networking/grpcgen/rds.go
index 7a7df64e..9a718dd8 100644
--- a/dubbod/discovery/pkg/networking/grpcgen/rds.go
+++ b/dubbod/discovery/pkg/networking/grpcgen/rds.go
@@ -241,6 +241,7 @@ func buildHTTPRoute(node *model.Proxy, push
*model.PushContext, routeName string
// Filter HTTPRoutes by parentRef to match this Gateway
httpRoutes := filterHTTPRoutesByGateway(allHTTPRoutes,
gatewayName, gatewayNamespace, gatewayListenerPort)
+ agentConfig := buildAgentConfig(push, httpRoutes)
log.Debugf("Gateway Pod inbound listener, filtered to %d
HTTPRoute(s) matching gateway %s/%s listener port %d", len(httpRoutes),
gatewayNamespace, gatewayName, gatewayListenerPort)
// For Gateway Pod, we also need to collect HTTPRoutes with
specific hostnames
@@ -313,6 +314,7 @@ func buildHTTPRoute(node *model.Proxy, push
*model.PushContext, routeName string
return &route.RouteConfiguration{
Name: routeName,
VirtualHosts: virtualHosts,
+ AgentConfig: agentConfig,
}
}
@@ -429,6 +431,13 @@ func buildRoutesFromGatewayHTTPRoute(httpRoutes
[]config.Config, hostName host.N
log.Debugf("HTTPRoute %s/%s rule[%d] has no
backendRefs, skipping", hrConfig.Namespace, hrConfig.Name, ruleIdx)
continue
}
+ if ruleUsesDxgateService(rule) {
+ // Mesh-native LLM/MCP/A2A backends are carried
in AgentConfig.
+ // Emitting them as ordinary clusters would
fabricate a
+ // Kubernetes Service with the same name and
bypass protocol
+ // translation.
+ continue
+ }
// Build weighted clusters from backendRefs
weights := make([]*route.WeightedCluster_ClusterWeight,
0, len(rule.BackendRefs))
@@ -483,9 +492,6 @@ func buildRoutesFromGatewayHTTPRoute(httpRoutes
[]config.Config, hostName host.N
weightedClusters.TotalWeight =
wrapperspb.UInt32(totalWeight)
}
- // Build route match from HTTPRoute matches
- routeMatch :=
buildRouteMatchFromHTTPRouteMatches(rule.Matches)
-
routeAction := &route.RouteAction{
ClusterSpecifier:
&route.RouteAction_WeightedClusters{
WeightedClusters: weightedClusters,
@@ -499,16 +505,18 @@ func buildRoutesFromGatewayHTTPRoute(httpRoutes
[]config.Config, hostName host.N
routeAction.RetryPolicy =
gatewayAPIRetryPolicy(rule.Retry, rule.Timeouts)
routeAction.FaultPolicy = faultPolicy
- builtRoute := &route.Route{
- Match: routeMatch,
- Action: &route.Route_Route{
- Route: routeAction,
- },
+ routeMatches :=
buildRouteMatchesFromHTTPRouteMatches(rule.Matches)
+ for _, routeMatch := range routeMatches {
+ allRoutes = append(allRoutes, &route.Route{
+ Match: routeMatch,
+ Action: &route.Route_Route{
+ Route: routeAction,
+ },
+ })
}
- log.Infof("HTTPRoute %s/%s rule[%d] -> built route with
%d clusters, totalWeight=%d",
- hrConfig.Namespace, hrConfig.Name, ruleIdx,
len(weights), totalWeight)
- allRoutes = append(allRoutes, builtRoute)
+ log.Infof("HTTPRoute %s/%s rule[%d] -> built %d routes
with %d clusters, totalWeight=%d",
+ hrConfig.Namespace, hrConfig.Name, ruleIdx,
len(routeMatches), len(weights), totalWeight)
}
}
@@ -701,19 +709,21 @@ func isServiceParentRef(parentRef
sigsk8siogatewayapiapisv1.ParentReference) boo
return parentRef.Group == nil || string(*parentRef.Group) == ""
}
-// buildRouteMatchFromHTTPRouteMatches converts Gateway API HTTPRouteMatch to
XDS RouteMatch
-func buildRouteMatchFromHTTPRouteMatches(matches
[]sigsk8siogatewayapiapisv1.HTTPRouteMatch) *route.RouteMatch {
+// buildRouteMatchesFromHTTPRouteMatches preserves Gateway API OR semantics:
+// every HTTPRouteMatch in a rule becomes an independent xDS route.
+func buildRouteMatchesFromHTTPRouteMatches(matches
[]sigsk8siogatewayapiapisv1.HTTPRouteMatch) []*route.RouteMatch {
if len(matches) == 0 {
- // No matches means match all
- return &route.RouteMatch{
- PathSpecifier: &route.RouteMatch_Prefix{
- Prefix: "/",
- },
- }
+ matches = []sigsk8siogatewayapiapisv1.HTTPRouteMatch{{}}
}
- // For now, we'll use the first match. In a full implementation, we
might need to merge multiple matches.
- match := matches[0]
+ out := make([]*route.RouteMatch, 0, len(matches))
+ for _, match := range matches {
+ out = append(out, buildRouteMatchFromHTTPRouteMatch(match))
+ }
+ return out
+}
+
+func buildRouteMatchFromHTTPRouteMatch(match
sigsk8siogatewayapiapisv1.HTTPRouteMatch) *route.RouteMatch {
routeMatch := &route.RouteMatch{}
// Handle path match
diff --git a/dubbod/discovery/pkg/networking/grpcgen/rds_test.go
b/dubbod/discovery/pkg/networking/grpcgen/rds_test.go
index d2b035c9..babdd1dd 100644
--- a/dubbod/discovery/pkg/networking/grpcgen/rds_test.go
+++ b/dubbod/discovery/pkg/networking/grpcgen/rds_test.go
@@ -17,6 +17,7 @@ package grpcgen
import (
"reflect"
+ "strings"
"testing"
"time"
@@ -37,6 +38,26 @@ import (
gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
)
+func TestBuildRouteMatchesPreservesHTTPRouteORSemantics(t *testing.T) {
+ pathType := gatewayv1.PathMatchPathPrefix
+ users := "/users"
+ orders := "/orders"
+ matches :=
buildRouteMatchesFromHTTPRouteMatches([]gatewayv1.HTTPRouteMatch{
+ {Path: &gatewayv1.HTTPPathMatch{Type: &pathType, Value:
&users}},
+ {Path: &gatewayv1.HTTPPathMatch{Type: &pathType, Value:
&orders}},
+ })
+
+ if len(matches) != 2 {
+ t.Fatalf("matches = %d, want 2 independent xDS routes",
len(matches))
+ }
+ if got := matches[0].GetPrefix(); got != users {
+ t.Fatalf("first prefix = %q, want %q", got, users)
+ }
+ if got := matches[1].GetPrefix(); got != orders {
+ t.Fatalf("second prefix = %q, want %q", got, orders)
+ }
+}
+
func TestBuildHTTPRouteProxylessOutboundIgnoresGatewayAttachedHTTPRoute(t
*testing.T) {
push := newRDSTestPushContext(t, []config.Config{
newWildcardHTTPRouteConfig("httpbin", "default", 8000),
@@ -335,6 +356,108 @@ func
TestGatewayInboundTargetPortIncludesActivationRoutes(t *testing.T) {
t.Fatalf("activation virtual host not found on Gateway targetPort: %v",
rc.GetVirtualHosts())
}
+func TestBuildAgentConfigCompilesDxgateServiceHTTPRoute(t *testing.T) {
+ routeConfig := newDxgateHTTPRouteConfig("anthropic", "app",
"/anthropic", "/v1/chat/completions")
+ serviceConfig := config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.DxgateService,
+ Name: "anthropic",
+ Namespace: "app",
+ },
+ Spec: &networking.DxgateService{
+ Service: &networking.DxgateService_Ai{Ai:
&networking.AIService{
+ Provider: &networking.AIProvider{
+ Provider:
&networking.AIProvider_Anthropic{Anthropic: &networking.AnthropicProvider{
+ Model: "claude-test",
+ }},
+ Credential:
&networking.SecretKeyReference{Name: "anthropic", Key: "api-key"},
+ },
+ Models: []string{"claude-test"},
+ Endpoint: "http://anthropic-mock.app.svc:8080";,
+ }},
+ Policies: &networking.DxgateServicePolicies{
+ Timeout: durationpb.New(5 * time.Second),
+ Retry: &networking.RetryPolicy{Attempts: 2,
StatusCodes: []uint32{503}},
+ },
+ },
+ }
+ push := newRDSTestPushContext(t, []config.Config{routeConfig,
serviceConfig}, nil)
+
+ got := buildAgentConfig(push, []config.Config{routeConfig})
+ if got == nil {
+ t.Fatal("AgentConfig = nil")
+ }
+ if len(got.GetProviders()) != 1 || got.GetProviders()[0].GetKind() !=
route.AgentProviderKind_ANTHROPIC {
+ t.Fatalf("providers = %v, want one Anthropic provider",
got.GetProviders())
+ }
+ if credential := got.GetProviders()[0].GetCredential();
credential.GetNamespace() != "app" ||
+ credential.GetName() != "anthropic" || credential.GetKey() !=
"api-key" {
+ t.Fatalf("credential = %v", credential)
+ }
+ if len(got.GetBackends()) != 1 || got.GetBackends()[0].GetLlm() == nil {
+ t.Fatalf("backends = %v, want one LLM backend",
got.GetBackends())
+ }
+ if len(got.GetAgentRoutes()) != 1 {
+ t.Fatalf("agent routes = %d, want 1", len(got.GetAgentRoutes()))
+ }
+ agentRoute := got.GetAgentRoutes()[0]
+ if agentRoute.GetProtocol() != route.AgentProtocol_LLM {
+ t.Fatalf("protocol = %v, want LLM", agentRoute.GetProtocol())
+ }
+ if got := agentRoute.GetMatches()[0].GetPath().GetPrefix(); got !=
"/anthropic" {
+ t.Fatalf("path prefix = %q, want /anthropic", got)
+ }
+ if got := agentRoute.GetRewrite().GetReplacePrefixMatch(); got !=
"/v1/chat/completions" {
+ t.Fatalf("rewrite prefix = %q, want /v1/chat/completions", got)
+ }
+ if got := got.GetPolicies()[0].GetRetry().GetAttempts(); got != 2 {
+ t.Fatalf("retry attempts = %d, want 2", got)
+ }
+}
+
+func TestBuildAgentConfigExpandsMCPFederationTargets(t *testing.T) {
+ routeConfig := newDxgateHTTPRouteConfig("company-tools", "app", "/mcp",
"")
+ serviceConfig := config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.DxgateService,
+ Name: "company-tools",
+ Namespace: "app",
+ },
+ Spec: &networking.DxgateService{
+ Service: &networking.DxgateService_Mcp{Mcp:
&networking.MCPService{
+ Targets: []*networking.MCPTarget{
+ {
+ Name: "orders",
+ Static:
&networking.StaticBackend{
+ BackendRef:
&networking.BackendReference{Name: "orders-tools"},
+ Port: 8080,
+ },
+ },
+ {
+ Name: "tickets",
+ Static:
&networking.StaticBackend{
+ BackendRef:
&networking.BackendReference{Name: "tickets-tools"},
+ Port: 8081,
+ },
+ },
+ },
+ }},
+ },
+ }
+ push := newRDSTestPushContext(t, []config.Config{routeConfig,
serviceConfig}, nil)
+
+ got := buildAgentConfig(push, []config.Config{routeConfig})
+ if got == nil || len(got.GetBackends()) != 2 {
+ t.Fatalf("backends = %v, want 2 MCP targets", got.GetBackends())
+ }
+ if len(got.GetAgentRoutes()) != 1 ||
len(got.GetAgentRoutes()[0].GetWeightedBackends()) != 2 {
+ t.Fatalf("agent routes = %v, want one federated route",
got.GetAgentRoutes())
+ }
+ if endpoint := got.GetBackends()[0].GetMcp().GetEndpoint();
!strings.Contains(endpoint, ".app.svc.cluster.local:") {
+ t.Fatalf("MCP endpoint = %q, want in-cluster Service DNS",
endpoint)
+ }
+}
+
func newRDSTestPushContext(t *testing.T, configs []config.Config, services
[]*model.Service) *model.PushContext {
t.Helper()
@@ -358,6 +481,53 @@ func newRDSTestPushContext(t *testing.T, configs
[]config.Config, services []*mo
return push
}
+func newDxgateHTTPRouteConfig(backendName, namespace, path, rewrite string)
config.Config {
+ group := gatewayv1.Group(dxgateServiceGroup)
+ kind := gatewayv1.Kind(dxgateServiceKind)
+ pathType := gatewayv1.PathMatchPathPrefix
+ method := gatewayv1.HTTPMethodPost
+ weight := int32(100)
+ rule := gatewayv1.HTTPRouteRule{
+ Matches: []gatewayv1.HTTPRouteMatch{{
+ Path: &gatewayv1.HTTPPathMatch{Type: &pathType,
Value: &path},
+ Method: &method,
+ }},
+ BackendRefs: []gatewayv1.HTTPBackendRef{{
+ BackendRef: gatewayv1.BackendRef{
+ BackendObjectReference:
gatewayv1.BackendObjectReference{
+ Group: &group,
+ Kind: &kind,
+ Name:
gatewayv1.ObjectName(backendName),
+ },
+ Weight: &weight,
+ },
+ }},
+ }
+ if rewrite != "" {
+ modifier := gatewayv1.PrefixMatchHTTPPathModifier
+ rule.Filters = []gatewayv1.HTTPRouteFilter{{
+ Type: gatewayv1.HTTPRouteFilterURLRewrite,
+ URLRewrite: &gatewayv1.HTTPURLRewriteFilter{
+ Path: &gatewayv1.HTTPPathModifier{
+ Type: modifier,
+ ReplacePrefixMatch: &rewrite,
+ },
+ },
+ }}
+ }
+ return config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.HTTPRoute,
+ Name: backendName,
+ Namespace: namespace,
+ Domain: "cluster.local",
+ },
+ Spec: &gatewayv1.HTTPRouteSpec{
+ Rules: []gatewayv1.HTTPRouteRule{rule},
+ },
+ }
+}
+
func newWildcardHTTPRouteConfig(backendName, backendNamespace string,
backendPort int32) config.Config {
port := backendPort
weight := int32(1)
diff --git a/go.mod b/go.mod
index ad044fb3..d44a074f 100644
--- a/go.mod
+++ b/go.mod
@@ -52,9 +52,9 @@ require (
github.com/hashicorp/go-multierror v1.1.1
github.com/hashicorp/golang-lru/v2 v2.0.7
github.com/heroku/color v0.0.6
- github.com/kdubbo/api v0.0.0-20260808071541-7709b6f3f491
- github.com/kdubbo/client-go v0.0.0-20260807013041-7abbf3125711
- github.com/kdubbo/xds-api v0.0.0-20260808063945-44451fbf2da1
+ github.com/kdubbo/api v0.0.0-20260808154749-d91601e19406
+ github.com/kdubbo/client-go v0.0.0-20260808154904-3acbf91635a1
+ github.com/kdubbo/xds-api v0.0.0-20260808155041-b6b8371fcc89
github.com/moby/moby/client v0.4.1
github.com/moby/term v0.5.2
github.com/ory/viper v1.7.5
diff --git a/go.sum b/go.sum
index 1c07d989..5e956c1c 100644
--- a/go.sum
+++ b/go.sum
@@ -368,12 +368,12 @@ github.com/jtolds/gls v4.20.0+incompatible/go.mod
h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfV
github.com/julienschmidt/httprouter v1.2.0/go.mod
h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51
h1:Z9n2FFNUXsshfwJMBgNA0RU6/i7WVaAegv3PtuIHPMs=
github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51/go.mod
h1:CzGEWj7cYgsdH8dAjBGEr58BoE7ScuLd+fwFZ44+/x8=
-github.com/kdubbo/api v0.0.0-20260808071541-7709b6f3f491
h1:KukdrR5lEbLUqh45vTPJnGLTusT9dCoro/LkiSvF+dk=
-github.com/kdubbo/api v0.0.0-20260808071541-7709b6f3f491/go.mod
h1:8BtJiIovg7QCPsCxXcw3gDf922VcvYq5ihOSvj49Rq8=
-github.com/kdubbo/client-go v0.0.0-20260807013041-7abbf3125711
h1:ofQz1xUC5DZyh+3lVPIFdipjKxe6jQ9uQbNMYoLInaI=
-github.com/kdubbo/client-go v0.0.0-20260807013041-7abbf3125711/go.mod
h1:ogHgHroSROD3HwYPBKS8c2dpcfd4cZxpwZrzTSCXjXs=
-github.com/kdubbo/xds-api v0.0.0-20260808063945-44451fbf2da1
h1:s/M95tB0/0Pf0icg9qCiK7zEWFsUt0F+8r3MYP3bUCQ=
-github.com/kdubbo/xds-api v0.0.0-20260808063945-44451fbf2da1/go.mod
h1:o2HDUgL1ntaDbWomZ4cD2tt8jBamuG2qRtjXOa1zZ0Q=
+github.com/kdubbo/api v0.0.0-20260808154749-d91601e19406
h1:HlD3yE/A+aHwXNIMydXMtdCTJyLukk8Znv+fo2pX8NA=
+github.com/kdubbo/api v0.0.0-20260808154749-d91601e19406/go.mod
h1:8BtJiIovg7QCPsCxXcw3gDf922VcvYq5ihOSvj49Rq8=
+github.com/kdubbo/client-go v0.0.0-20260808154904-3acbf91635a1
h1:ykfiDq3lqlBJp8r30xJIJC+E2amT0GQVQep62+lhdzo=
+github.com/kdubbo/client-go v0.0.0-20260808154904-3acbf91635a1/go.mod
h1:euRf0fpvc8PfHmIwb3Oj/b42aJDhMVX7PHAYzSZjpU8=
+github.com/kdubbo/xds-api v0.0.0-20260808155041-b6b8371fcc89
h1:HVDXIqVzomm6eHAaitmaB3JGPd7w7BhSg5ns1ObLvzo=
+github.com/kdubbo/xds-api v0.0.0-20260808155041-b6b8371fcc89/go.mod
h1:o2HDUgL1ntaDbWomZ4cD2tt8jBamuG2qRtjXOa1zZ0Q=
github.com/kevinburke/ssh_config v1.2.0
h1:x584FjTGwHzMwvHx18PXxbBVzfnxogHaAReU4gf13a4=
github.com/kevinburke/ssh_config v1.2.0/go.mod
h1:CT57kijsi8u/K/BOFA39wgDQJ9CxiF4nAY/ojJ6r6mM=
github.com/kisielk/errcheck v1.5.0/go.mod
h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
diff --git a/manifests/charts/base/files/crd-all.gen.yaml
b/manifests/charts/base/files/crd-all.gen.yaml
index 34c42e30..21f52ae6 100644
--- a/manifests/charts/base/files/crd-all.gen.yaml
+++ b/manifests/charts/base/files/crd-all.gen.yaml
@@ -152,6 +152,371 @@ spec:
---
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
+metadata:
+ annotations:
+ "helm.sh/resource-policy": keep
+ labels:
+ app: dubbo
+ chart: dubbo
+ dubbo: networking
+ heritage: Tiller
+ release: dubbo
+ name: dxgateservices.networking.dubbo.apache.org
+spec:
+ group: networking.dubbo.apache.org
+ names:
+ categories:
+ - dubbo
+ - networking
+ kind: DxgateService
+ listKind: DxgateServiceList
+ plural: dxgateservices
+ shortNames:
+ - dxsvc
+ singular: dxgateservice
+ scope: Namespaced
+ versions:
+ - additionalPrinterColumns:
+ - description: CreationTimestamp is a timestamp representing the server
time when
+ this object was created.
+ jsonPath: .metadata.creationTimestamp
+ name: Age
+ type: date
+ name: v1alpha3
+ schema:
+ openAPIV3Schema:
+ properties:
+ spec:
+ description: 'Mesh-native LLM, MCP, or A2A backend referenced by
Gateway
+ API routes. See more details at: '
+ oneOf:
+ - not:
+ anyOf:
+ - required:
+ - ai
+ - required:
+ - mcp
+ - required:
+ - a2a
+ - required:
+ - ai
+ - required:
+ - mcp
+ - required:
+ - a2a
+ properties:
+ a2a:
+ properties:
+ agent:
+ type: string
+ backendRef:
+ description: Prefer backend_ref for an in-cluster agent.
+ properties:
+ name:
+ description: Kubernetes Service name.
+ type: string
+ namespace:
+ description: Defaults to the DxgateService namespace.
+ type: string
+ required:
+ - name
+ type: object
+ host:
+ type: string
+ path:
+ type: string
+ port:
+ maximum: 4294967295
+ minimum: 0
+ type: integer
+ type: object
+ ai:
+ properties:
+ endpoint:
+ description: Optional backend URL.
+ type: string
+ modelRewrites:
+ additionalProperties:
+ type: string
+ description: Client model alias to upstream model name.
+ type: object
+ models:
+ description: Models accepted by this backend.
+ items:
+ type: string
+ type: array
+ provider:
+ oneOf:
+ - not:
+ anyOf:
+ - required:
+ - openai
+ - required:
+ - anthropic
+ - required:
+ - openai
+ - required:
+ - anthropic
+ properties:
+ anthropic:
+ properties:
+ model:
+ description: Default upstream model.
+ type: string
+ type: object
+ credential:
+ description: Provider API key.
+ properties:
+ key:
+ type: string
+ name:
+ type: string
+ required:
+ - name
+ - key
+ type: object
+ openai:
+ type: object
+ requestHeaders:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ required:
+ - name
+ - value
+ type: object
+ type: array
+ type: object
+ routes:
+ additionalProperties:
+ description: |2-
+
+
+ Valid Options: COMPLETIONS, RESPONSES, EMBEDDINGS,
PASSTHROUGH
+ enum:
+ - AI_OPERATION_UNSPECIFIED
+ - COMPLETIONS
+ - RESPONSES
+ - EMBEDDINGS
+ - PASSTHROUGH
+ type: string
+ description: OpenAI-compatible operations explicitly
enabled for
+ this service.
+ type: object
+ required:
+ - provider
+ type: object
+ mcp:
+ properties:
+ targets:
+ items:
+ properties:
+ name:
+ type: string
+ static:
+ properties:
+ backendRef:
+ properties:
+ name:
+ description: Kubernetes Service name.
+ type: string
+ namespace:
+ description: Defaults to the DxgateService
namespace.
+ type: string
+ required:
+ - name
+ type: object
+ path:
+ type: string
+ port:
+ maximum: 4294967295
+ minimum: 0
+ type: integer
+ required:
+ - backendRef
+ - port
+ type: object
+ tools:
+ items:
+ type: string
+ type: array
+ required:
+ - name
+ - static
+ type: object
+ type: array
+ required:
+ - targets
+ type: object
+ policies:
+ description: Policies applied to every HTTPRoute reference to
this
+ service.
+ properties:
+ auth:
+ description: Client authentication at the gateway.
+ properties:
+ header:
+ type: string
+ secretRef:
+ properties:
+ key:
+ type: string
+ name:
+ type: string
+ required:
+ - name
+ - key
+ type: object
+ required:
+ - secretRef
+ type: object
+ maxBodyBytes:
+ format: int64
+ type: integer
+ rateLimit:
+ properties:
+ header:
+ type: string
+ key:
+ description: |2-
+
+
+ Valid Options: ROUTE, BACKEND, HEADER
+ enum:
+ - RATE_LIMIT_KEY_UNSPECIFIED
+ - ROUTE
+ - BACKEND
+ - HEADER
+ type: string
+ requests:
+ maximum: 4294967295
+ minimum: 0
+ type: integer
+ window:
+ type: string
+ x-kubernetes-validations:
+ - message: must be a valid duration greater than 1ms
+ rule: duration(self) >= duration('1ms')
+ required:
+ - requests
+ - window
+ type: object
+ requestHeaders:
+ properties:
+ add:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ required:
+ - name
+ - value
+ type: object
+ type: array
+ remove:
+ items:
+ type: string
+ type: array
+ type: object
+ responseHeaders:
+ properties:
+ add:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ required:
+ - name
+ - value
+ type: object
+ type: array
+ remove:
+ items:
+ type: string
+ type: array
+ type: object
+ retry:
+ properties:
+ attempts:
+ maximum: 4294967295
+ minimum: 0
+ type: integer
+ statusCodes:
+ items:
+ maximum: 4294967295
+ minimum: 0
+ type: integer
+ type: array
+ type: object
+ timeout:
+ type: string
+ x-kubernetes-validations:
+ - message: must be a valid duration greater than 1ms
+ rule: duration(self) >= duration('1ms')
+ tokenLimit:
+ properties:
+ header:
+ type: string
+ key:
+ description: |2-
+
+
+ Valid Options: ROUTE, BACKEND, HEADER
+ enum:
+ - RATE_LIMIT_KEY_UNSPECIFIED
+ - ROUTE
+ - BACKEND
+ - HEADER
+ type: string
+ tokens:
+ minimum: 0
+ type: integer
+ window:
+ type: string
+ x-kubernetes-validations:
+ - message: must be a valid duration greater than 1ms
+ rule: duration(self) >= duration('1ms')
+ required:
+ - tokens
+ - window
+ type: object
+ type: object
+ type: object
+ status:
+ properties:
+ conditions:
+ items:
+ properties:
+ observedGeneration:
+ anyOf:
+ - type: integer
+ - type: string
+ x-kubernetes-int-or-string: true
+ reason:
+ type: string
+ status:
+ type: string
+ type:
+ type: string
+ type: object
+ type: array
+ type: object
+ x-kubernetes-preserve-unknown-fields: true
+ type: object
+ served: true
+ storage: true
+ subresources:
+ status: {}
+---
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
metadata:
annotations:
"helm.sh/resource-policy": keep
diff --git a/manifests/charts/dubbod/files/kube-gateway.yaml
b/manifests/charts/dubbod/files/kube-gateway.yaml
index 3e3ab51c..c2beb796 100644
--- a/manifests/charts/dubbod/files/kube-gateway.yaml
+++ b/manifests/charts/dubbod/files/kube-gateway.yaml
@@ -37,7 +37,41 @@ metadata:
app.kubernetes.io/instance: {{ .DeploymentName }}
app.kubernetes.io/managed-by: dubbod
gateway.networking.k8s.io/gateway-name: {{ .Gateway.Name }}
-automountServiceAccountToken: false
+automountServiceAccountToken: true
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: Role
+metadata:
+ name: {{ .ServiceAccount }}-credentials
+ namespace: {{ .Gateway.Namespace }}
+ labels:
+ app.kubernetes.io/name: dxgate
+ app.kubernetes.io/instance: {{ .DeploymentName }}
+ app.kubernetes.io/managed-by: dubbod
+ gateway.networking.k8s.io/gateway-name: {{ .Gateway.Name }}
+rules:
+- apiGroups: [""]
+ resources: ["secrets"]
+ verbs: ["get"]
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+ name: {{ .ServiceAccount }}-credentials
+ namespace: {{ .Gateway.Namespace }}
+ labels:
+ app.kubernetes.io/name: dxgate
+ app.kubernetes.io/instance: {{ .DeploymentName }}
+ app.kubernetes.io/managed-by: dubbod
+ gateway.networking.k8s.io/gateway-name: {{ .Gateway.Name }}
+roleRef:
+ apiGroup: rbac.authorization.k8s.io
+ kind: Role
+ name: {{ .ServiceAccount }}-credentials
+subjects:
+- kind: ServiceAccount
+ name: {{ .ServiceAccount }}
+ namespace: {{ .Gateway.Namespace }}
---
apiVersion: apps/v1
kind: Deployment
diff --git a/manifests/charts/dubbod/templates/clusterrole.yaml
b/manifests/charts/dubbod/templates/clusterrole.yaml
index 751e0f2b..80b23939 100644
--- a/manifests/charts/dubbod/templates/clusterrole.yaml
+++ b/manifests/charts/dubbod/templates/clusterrole.yaml
@@ -113,6 +113,18 @@ rules:
- update
- patch
- delete
+ - apiGroups: ["rbac.authorization.k8s.io"]
+ resources:
+ - roles
+ - rolebindings
+ verbs:
+ - get
+ - list
+ - watch
+ - create
+ - update
+ - patch
+ - delete
- apiGroups: ["apps"]
resources:
- deployments
diff --git a/manifests/charts/dubbod/templates/configmap-injector.yaml
b/manifests/charts/dubbod/templates/configmap-injector.yaml
index 80a9b1b3..0d4a9f1a 100644
--- a/manifests/charts/dubbod/templates/configmap-injector.yaml
+++ b/manifests/charts/dubbod/templates/configmap-injector.yaml
@@ -13,6 +13,22 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+{{- $defaults := .Values._internal_default_values_not_set | default dict }}
+{{- $defaultGlobal := $defaults.global | default dict }}
+{{- $defaultProxyless := $defaultGlobal.proxyless | default dict }}
+{{- $defaultCNI := $defaultProxyless.cni | default dict }}
+{{- $global := .Values.global | default dict }}
+{{- $proxyless := $global.proxyless | default dict }}
+{{- $cni := $proxyless.cni | default dict }}
+{{- $image := coalesce $cni.image $defaultCNI.image (printf
"ghcr.io/apache/dubbo-kubernetes/dubbod:%s" .Chart.AppVersion) }}
+{{- $injectorValues := deepCopy (omit .Values
"_internal_default_values_not_set") }}
+{{- $injectorGlobal := get $injectorValues "global" | default dict }}
+{{- $_ := set $injectorValues "global" $injectorGlobal }}
+{{- $injectorProxyless := get $injectorGlobal "proxyless" | default dict }}
+{{- $_ = set $injectorGlobal "proxyless" $injectorProxyless }}
+{{- $injectorCNI := get $injectorProxyless "cni" | default dict }}
+{{- $_ = set $injectorProxyless "cni" $injectorCNI }}
+{{- $_ = set $injectorCNI "image" $image }}
apiVersion: v1
kind: ConfigMap
metadata:
@@ -31,3 +47,4 @@ data:
gateway: |
{{ .Files.Get "files/kube-gateway.yaml" | trim | indent 8 }}
values: |-
+{{ $injectorValues | toYaml | indent 4 }}
diff --git a/pkg/config/schema/collections/collections.gen.go
b/pkg/config/schema/collections/collections.gen.go
index d1a08f35..9d82b78a 100755
--- a/pkg/config/schema/collections/collections.gen.go
+++ b/pkg/config/schema/collections/collections.gen.go
@@ -137,6 +137,21 @@ var (
ValidateProto: validation.EmptyValidate,
}.MustBuild()
+ DxgateService = resource.Builder{
+ Identifier: "DxgateService",
+ Group: "networking.dubbo.apache.org",
+ Kind: "DxgateService",
+ Plural: "dxgateservices",
+ Version: "v1alpha3",
+ Proto: "dubbo.networking.v1alpha3.DxgateService",
StatusProto: "dubbo.meta.v1alpha1.DubboStatus",
+ ReflectType:
reflect.TypeOf(&githubcomkdubboapinetworkingv1alpha3.DxgateService{}).Elem(),
StatusType:
reflect.TypeOf(&githubcomkdubboapimetav1alpha1.DubboStatus{}).Elem(),
+ ProtoPackage: "github.com/kdubbo/api/networking/v1alpha3",
StatusPackage: "github.com/kdubbo/api/meta/v1alpha1",
+ ClusterScoped: false,
+ Synthetic: false,
+ Builtin: false,
+ ValidateProto: validation.ValidateDxgateService,
+ }.MustBuild()
+
EndpointSlice = resource.Builder{
Identifier: "EndpointSlice",
Group: "discovery.k8s.io",
@@ -548,6 +563,7 @@ var (
MustAdd(CustomResourceDefinition).
MustAdd(DaemonSet).
MustAdd(Deployment).
+ MustAdd(DxgateService).
MustAdd(EndpointSlice).
MustAdd(Endpoints).
MustAdd(FaultInjectionPolicy).
@@ -607,6 +623,7 @@ var (
Dubbo = collection.NewSchemasBuilder().
MustAdd(AuthorizationPolicy).
MustAdd(CircuitBreakerPolicy).
+ MustAdd(DxgateService).
MustAdd(FaultInjectionPolicy).
MustAdd(PeerAuthentication).
MustAdd(RequestAuthentication).
@@ -621,6 +638,7 @@ var (
MustAdd(AuthorizationPolicy).
MustAdd(BackendTLSPolicy).
MustAdd(CircuitBreakerPolicy).
+ MustAdd(DxgateService).
MustAdd(FaultInjectionPolicy).
MustAdd(GatewayClass).
MustAdd(HTTPRoute).
@@ -639,6 +657,7 @@ var (
MustAdd(AuthorizationPolicy).
MustAdd(BackendTLSPolicy).
MustAdd(CircuitBreakerPolicy).
+ MustAdd(DxgateService).
MustAdd(FaultInjectionPolicy).
MustAdd(GatewayClass).
MustAdd(HTTPRoute).
diff --git a/pkg/config/schema/gvk/resources.gen.go
b/pkg/config/schema/gvk/resources.gen.go
index 90b93612..b094ac5c 100755
--- a/pkg/config/schema/gvk/resources.gen.go
+++ b/pkg/config/schema/gvk/resources.gen.go
@@ -18,6 +18,7 @@ var (
CustomResourceDefinition = config.GroupVersionKind{Group:
"apiextensions.k8s.io", Version: "v1", Kind: "CustomResourceDefinition"}
DaemonSet = config.GroupVersionKind{Group: "apps",
Version: "v1", Kind: "DaemonSet"}
Deployment = config.GroupVersionKind{Group: "apps",
Version: "v1", Kind: "Deployment"}
+ DxgateService = config.GroupVersionKind{Group:
"networking.dubbo.apache.org", Version: "v1alpha3", Kind: "DxgateService"}
EndpointSlice = config.GroupVersionKind{Group:
"discovery.k8s.io", Version: "v1", Kind: "EndpointSlice"}
Endpoints = config.GroupVersionKind{Group: "",
Version: "v1", Kind: "Endpoints"}
FaultInjectionPolicy = config.GroupVersionKind{Group:
"networking.dubbo.apache.org", Version: "v1alpha3", Kind:
"FaultInjectionPolicy"}
@@ -69,6 +70,8 @@ func ToGVR(g config.GroupVersionKind)
(schema.GroupVersionResource, bool) {
return gvr.DaemonSet, true
case Deployment:
return gvr.Deployment, true
+ case DxgateService:
+ return gvr.DxgateService, true
case EndpointSlice:
return gvr.EndpointSlice, true
case Endpoints:
@@ -150,6 +153,8 @@ func MustToKind(g config.GroupVersionKind) kind.Kind {
return kind.DaemonSet
case Deployment:
return kind.Deployment
+ case DxgateService:
+ return kind.DxgateService
case EndpointSlice:
return kind.EndpointSlice
case Endpoints:
@@ -234,6 +239,8 @@ func FromGVR(g schema.GroupVersionResource)
(config.GroupVersionKind, bool) {
return DaemonSet, true
case gvr.Deployment:
return Deployment, true
+ case gvr.DxgateService:
+ return DxgateService, true
case gvr.EndpointSlice:
return EndpointSlice, true
case gvr.Endpoints:
diff --git a/pkg/config/schema/gvr/resources.gen.go
b/pkg/config/schema/gvr/resources.gen.go
index 79d7ab4f..b71d600f 100755
--- a/pkg/config/schema/gvr/resources.gen.go
+++ b/pkg/config/schema/gvr/resources.gen.go
@@ -13,6 +13,7 @@ var (
CustomResourceDefinition = schema.GroupVersionResource{Group:
"apiextensions.k8s.io", Version: "v1", Resource: "customresourcedefinitions"}
DaemonSet = schema.GroupVersionResource{Group:
"apps", Version: "v1", Resource: "daemonsets"}
Deployment = schema.GroupVersionResource{Group:
"apps", Version: "v1", Resource: "deployments"}
+ DxgateService = schema.GroupVersionResource{Group:
"networking.dubbo.apache.org", Version: "v1alpha3", Resource: "dxgateservices"}
EndpointSlice = schema.GroupVersionResource{Group:
"discovery.k8s.io", Version: "v1", Resource: "endpointslices"}
Endpoints = schema.GroupVersionResource{Group: "",
Version: "v1", Resource: "endpoints"}
FaultInjectionPolicy = schema.GroupVersionResource{Group:
"networking.dubbo.apache.org", Version: "v1alpha3", Resource:
"faultinjectionpolicies"}
@@ -63,6 +64,8 @@ func IsClusterScoped(g schema.GroupVersionResource) bool {
return false
case Deployment:
return false
+ case DxgateService:
+ return false
case EndpointSlice:
return false
case Endpoints:
diff --git a/pkg/config/schema/kind/resources.gen.go
b/pkg/config/schema/kind/resources.gen.go
index 2eeffb3e..01ade2a7 100755
--- a/pkg/config/schema/kind/resources.gen.go
+++ b/pkg/config/schema/kind/resources.gen.go
@@ -13,6 +13,7 @@ const (
DNSName
DaemonSet
Deployment
+ DxgateService
EndpointSlice
Endpoints
FaultInjectionPolicy
@@ -61,6 +62,8 @@ func (k Kind) String() string {
return "DaemonSet"
case Deployment:
return "Deployment"
+ case DxgateService:
+ return "DxgateService"
case EndpointSlice:
return "EndpointSlice"
case Endpoints:
@@ -138,6 +141,8 @@ func FromString(s string) Kind {
return DaemonSet
case "Deployment":
return Deployment
+ case "DxgateService":
+ return DxgateService
case "EndpointSlice":
return EndpointSlice
case "Endpoints":
diff --git a/pkg/config/schema/kubeclient/resources.gen.go
b/pkg/config/schema/kubeclient/resources.gen.go
index 217a0ec5..61e5a3ab 100755
--- a/pkg/config/schema/kubeclient/resources.gen.go
+++ b/pkg/config/schema/kubeclient/resources.gen.go
@@ -48,6 +48,8 @@ func GetWriteClient[T runtime.Object](c ClientGetter,
namespace string) ktypes.W
return
c.Kube().AppsV1().DaemonSets(namespace).(ktypes.WriteAPI[T])
case *k8sioapiappsv1.Deployment:
return
c.Kube().AppsV1().Deployments(namespace).(ktypes.WriteAPI[T])
+ case
*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.DxgateService:
+ return
c.Dubbo().NetworkingV1alpha3().DxgateServices(namespace).(ktypes.WriteAPI[T])
case *k8sioapidiscoveryv1.EndpointSlice:
return
c.Kube().DiscoveryV1().EndpointSlices(namespace).(ktypes.WriteAPI[T])
case *k8sioapicorev1.Endpoints:
@@ -119,6 +121,8 @@ func GetClient[T, TL runtime.Object](c ClientGetter,
namespace string) ktypes.Re
return
c.Kube().AppsV1().DaemonSets(namespace).(ktypes.ReadWriteAPI[T, TL])
case *k8sioapiappsv1.Deployment:
return
c.Kube().AppsV1().Deployments(namespace).(ktypes.ReadWriteAPI[T, TL])
+ case
*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.DxgateService:
+ return
c.Dubbo().NetworkingV1alpha3().DxgateServices(namespace).(ktypes.ReadWriteAPI[T,
TL])
case *k8sioapidiscoveryv1.EndpointSlice:
return
c.Kube().DiscoveryV1().EndpointSlices(namespace).(ktypes.ReadWriteAPI[T, TL])
case *k8sioapicorev1.Endpoints:
@@ -190,6 +194,8 @@ func gvrToObject(g schema.GroupVersionResource)
runtime.Object {
return &k8sioapiappsv1.DaemonSet{}
case gvr.Deployment:
return &k8sioapiappsv1.Deployment{}
+ case gvr.DxgateService:
+ return
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.DxgateService{}
case gvr.EndpointSlice:
return &k8sioapidiscoveryv1.EndpointSlice{}
case gvr.Endpoints:
@@ -299,6 +305,13 @@ func getInformerFiltered(c ClientGetter, opts
ktypes.InformerOptions, g schema.G
w = func(options metav1.ListOptions) (watch.Interface, error) {
return
c.Kube().AppsV1().Deployments(opts.Namespace).Watch(context.Background(),
options)
}
+ case gvr.DxgateService:
+ l = func(options metav1.ListOptions) (runtime.Object, error) {
+ return
c.Dubbo().NetworkingV1alpha3().DxgateServices(opts.Namespace).List(context.Background(),
options)
+ }
+ w = func(options metav1.ListOptions) (watch.Interface, error) {
+ return
c.Dubbo().NetworkingV1alpha3().DxgateServices(opts.Namespace).Watch(context.Background(),
options)
+ }
case gvr.EndpointSlice:
l = func(options metav1.ListOptions) (runtime.Object, error) {
return
c.Kube().DiscoveryV1().EndpointSlices(opts.Namespace).List(context.Background(),
options)
diff --git a/pkg/config/schema/kubetypes/resources.gen.go
b/pkg/config/schema/kubetypes/resources.gen.go
index 00efb091..ea73932a 100755
--- a/pkg/config/schema/kubetypes/resources.gen.go
+++ b/pkg/config/schema/kubetypes/resources.gen.go
@@ -44,6 +44,10 @@ func getGvk(obj any) (config.GroupVersionKind, bool) {
return gvk.DaemonSet, true
case *k8sioapiappsv1.Deployment:
return gvk.Deployment, true
+ case *githubcomkdubboapinetworkingv1alpha3.DxgateService:
+ return gvk.DxgateService, true
+ case
*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.DxgateService:
+ return gvk.DxgateService, true
case *k8sioapidiscoveryv1.EndpointSlice:
return gvk.EndpointSlice, true
case *k8sioapicorev1.Endpoints:
diff --git a/pkg/config/schema/metadata.yaml b/pkg/config/schema/metadata.yaml
index 7a03f977..c238b616 100644
--- a/pkg/config/schema/metadata.yaml
+++ b/pkg/config/schema/metadata.yaml
@@ -240,6 +240,16 @@ resources:
statusProto: "dubbo.meta.v1alpha1.DubboStatus"
statusProtoPackage: "github.com/kdubbo/api/meta/v1alpha1"
+ - kind: DxgateService
+ plural: "dxgateservices"
+ group: "networking.dubbo.apache.org"
+ version: "v1alpha3"
+ proto: "dubbo.networking.v1alpha3.DxgateService"
+ protoPackage: "github.com/kdubbo/api/networking/v1alpha3"
+ validate: "validation.ValidateDxgateService"
+ statusProto: "dubbo.meta.v1alpha1.DubboStatus"
+ statusProtoPackage: "github.com/kdubbo/api/meta/v1alpha1"
+
- kind: ServiceActivationPolicy
plural: "serviceactivationpolicies"
group: "networking.dubbo.apache.org"
diff --git a/pkg/config/validation/validators.go
b/pkg/config/validation/validators.go
index b5b61739..8d5b050a 100644
--- a/pkg/config/validation/validators.go
+++ b/pkg/config/validation/validators.go
@@ -20,6 +20,7 @@ import (
"fmt"
"math"
"net"
+ "net/url"
"strings"
"time"
@@ -263,6 +264,156 @@ var ValidateFaultInjectionPolicy =
RegisterValidateFunc("ValidateFaultInjectionP
return v.Unwrap()
})
+// ValidateDxgateService checks that a mesh-native LLM, MCP, or A2A backend can
+// be compiled into one unambiguous data-plane configuration.
+var ValidateDxgateService = RegisterValidateFunc("ValidateDxgateService",
+ func(cfg config.Config) (Warning, error) {
+ spec, ok := cfg.Spec.(*networking.DxgateService)
+ if !ok {
+ return nil, fmt.Errorf("cannot cast to DxgateService")
+ }
+ v := Validation{}
+ switch {
+ case spec.GetAi() != nil:
+ ai := spec.GetAi()
+ if ai.GetProvider() == nil ||
ai.GetProvider().GetProvider() == nil {
+ v = appendValidation(v, fmt.Errorf("ai.provider
must select openai or anthropic"))
+ }
+ if endpoint := ai.GetEndpoint(); endpoint != "" {
+ parsed, err := url.ParseRequestURI(endpoint)
+ if err != nil || (parsed.Scheme != "http" &&
parsed.Scheme != "https") || parsed.Host == "" {
+ v = appendValidation(v,
fmt.Errorf("ai.endpoint %q must be an absolute HTTP(S) URL", endpoint))
+ }
+ }
+ for i, model := range ai.GetModels() {
+ if strings.TrimSpace(model) == "" {
+ v = appendValidation(v,
fmt.Errorf("ai.models[%d] must not be empty", i))
+ }
+ }
+ for path := range ai.GetRoutes() {
+ if !strings.HasPrefix(path, "/") {
+ v = appendValidation(v,
fmt.Errorf("ai.routes key %q must start with /", path))
+ }
+ }
+ if credential := ai.GetProvider().GetCredential();
credential != nil {
+ v = appendValidation(v,
validateSecretKeyReference("ai.provider.credential", credential))
+ }
+ case spec.GetMcp() != nil:
+ targets := spec.GetMcp().GetTargets()
+ if len(targets) == 0 {
+ v = appendValidation(v, fmt.Errorf("mcp.targets
must not be empty"))
+ }
+ names := make(map[string]struct{}, len(targets))
+ for i, target := range targets {
+ if target == nil {
+ v = appendValidation(v,
fmt.Errorf("mcp.targets[%d] must not be null", i))
+ continue
+ }
+ if target.GetName() == "" {
+ v = appendValidation(v,
fmt.Errorf("mcp.targets[%d].name must not be empty", i))
+ } else if _, found := names[target.GetName()];
found {
+ v = appendValidation(v,
fmt.Errorf("mcp.targets[%d].name %q is duplicated", i, target.GetName()))
+ } else {
+ names[target.GetName()] = struct{}{}
+ }
+ static := target.GetStatic()
+ if static == nil {
+ v = appendValidation(v,
fmt.Errorf("mcp.targets[%d].static must be set", i))
+ continue
+ }
+ v = appendValidation(v,
+
validateBackendReference(fmt.Sprintf("mcp.targets[%d].static.backendRef", i),
static.GetBackendRef()),
+
validateDxgatePort(fmt.Sprintf("mcp.targets[%d].static.port", i),
static.GetPort()),
+
validateOptionalPath(fmt.Sprintf("mcp.targets[%d].static.path", i),
static.GetPath()),
+ )
+ }
+ case spec.GetA2A() != nil:
+ a2a := spec.GetA2A()
+ hasRef := a2a.GetBackendRef() != nil
+ hasHost := strings.TrimSpace(a2a.GetHost()) != ""
+ if hasRef == hasHost {
+ v = appendValidation(v, fmt.Errorf("a2a must
set exactly one of backendRef or host"))
+ }
+ if hasRef {
+ v = appendValidation(v,
validateBackendReference("a2a.backendRef", a2a.GetBackendRef()))
+ }
+ v = appendValidation(v,
+ validateDxgatePort("a2a.port", a2a.GetPort()),
+ validateOptionalPath("a2a.path", a2a.GetPath()),
+ )
+ default:
+ v = appendValidation(v, fmt.Errorf("exactly one of ai,
mcp, or a2a must be set"))
+ }
+
+ if policies := spec.GetPolicies(); policies != nil {
+ if auth := policies.GetAuth(); auth != nil {
+ v = appendValidation(v,
validateSecretKeyReference("policies.auth.secretRef", auth.GetSecretRef()))
+ }
+ if rate := policies.GetRateLimit(); rate != nil {
+ if rate.GetRequests() == 0 {
+ v = appendValidation(v,
fmt.Errorf("policies.rateLimit.requests must be greater than zero"))
+ }
+ v = appendValidation(v,
validatePositiveDuration("policies.rateLimit.window", rate.GetWindow()))
+ }
+ if tokens := policies.GetTokenLimit(); tokens != nil {
+ if tokens.GetTokens() == 0 {
+ v = appendValidation(v,
fmt.Errorf("policies.tokenLimit.tokens must be greater than zero"))
+ }
+ v = appendValidation(v,
validatePositiveDuration("policies.tokenLimit.window", tokens.GetWindow()))
+ }
+ v = appendValidation(v,
validatePositiveDuration("policies.timeout", policies.GetTimeout()))
+ if retry := policies.GetRetry(); retry != nil {
+ if retry.GetAttempts() == 0 {
+ v = appendValidation(v,
fmt.Errorf("policies.retry.attempts must be greater than zero"))
+ }
+ for i, status := range retry.GetStatusCodes() {
+ if status < 400 || status > 599 {
+ v = appendValidation(v,
fmt.Errorf("policies.retry.statusCodes[%d] must be in range [400, 599], got
%d", i, status))
+ }
+ }
+ }
+ if policies.GetMaxBodyBytes() < 0 {
+ v = appendValidation(v,
fmt.Errorf("policies.maxBodyBytes must not be negative"))
+ }
+ }
+ return v.Unwrap()
+ })
+
+func validateSecretKeyReference(field string, ref
*networking.SecretKeyReference) error {
+ if ref == nil {
+ return fmt.Errorf("%s must be set", field)
+ }
+ var errs error
+ if strings.TrimSpace(ref.GetName()) == "" {
+ errs = AppendErrors(errs, fmt.Errorf("%s.name must not be
empty", field))
+ }
+ if strings.TrimSpace(ref.GetKey()) == "" {
+ errs = AppendErrors(errs, fmt.Errorf("%s.key must not be
empty", field))
+ }
+ return errs
+}
+
+func validateBackendReference(field string, ref *networking.BackendReference)
error {
+ if ref == nil || strings.TrimSpace(ref.GetName()) == "" {
+ return fmt.Errorf("%s.name must not be empty", field)
+ }
+ return nil
+}
+
+func validateDxgatePort(field string, port uint32) error {
+ if port == 0 || port > 65535 {
+ return fmt.Errorf("%s must be in range [1, 65535], got %d",
field, port)
+ }
+ return nil
+}
+
+func validateOptionalPath(field, path string) error {
+ if path != "" && !strings.HasPrefix(path, "/") {
+ return fmt.Errorf("%s must start with /", field)
+ }
+ return nil
+}
+
// ValidateServiceEntry checks that a ServiceEntry can be converted into
services and endpoints.
var ValidateServiceEntry = RegisterValidateFunc("ValidateServiceEntry",
func(cfg config.Config) (Warning, error) {
spec, ok := cfg.Spec.(*networking.ServiceEntry)
diff --git a/pkg/config/validation/validators_test.go
b/pkg/config/validation/validators_test.go
index 08dce1af..34cdbd71 100644
--- a/pkg/config/validation/validators_test.go
+++ b/pkg/config/validation/validators_test.go
@@ -382,6 +382,112 @@ func TestValidateFaultInjectionPolicy(t *testing.T) {
}
}
+func TestValidateDxgateService(t *testing.T) {
+ openAI := func() *networking.DxgateService {
+ return &networking.DxgateService{
+ Service: &networking.DxgateService_Ai{Ai:
&networking.AIService{
+ Provider: &networking.AIProvider{
+ Provider:
&networking.AIProvider_Openai{Openai: &networking.OpenAIProvider{}},
+ },
+ Models: []string{"gpt-test"},
+ Endpoint: "http://llm.default.svc:8080/v1";,
+ }},
+ }
+ }
+ mcp := func() *networking.DxgateService {
+ return &networking.DxgateService{
+ Service: &networking.DxgateService_Mcp{Mcp:
&networking.MCPService{
+ Targets: []*networking.MCPTarget{{
+ Name: "tools",
+ Static: &networking.StaticBackend{
+ BackendRef:
&networking.BackendReference{Name: "tools"},
+ Port: 8080,
+ },
+ }},
+ }},
+ }
+ }
+ a2a := func() *networking.DxgateService {
+ return &networking.DxgateService{
+ Service: &networking.DxgateService_A2A{A2A:
&networking.A2AService{
+ BackendRef: &networking.BackendReference{Name:
"review-agent"},
+ Port: 9090,
+ }},
+ }
+ }
+ cases := []struct {
+ name string
+ spec *networking.DxgateService
+ wantErr bool
+ }{
+ {name: "openai", spec: openAI()},
+ {name: "mcp", spec: mcp()},
+ {name: "a2a", spec: a2a()},
+ {name: "missing service", spec: &networking.DxgateService{},
wantErr: true},
+ {
+ name: "ai provider missing",
+ spec: &networking.DxgateService{
+ Service: &networking.DxgateService_Ai{Ai:
&networking.AIService{}},
+ },
+ wantErr: true,
+ },
+ {
+ name: "bad ai endpoint",
+ spec: func() *networking.DxgateService {
+ s := openAI()
+ s.GetAi().Endpoint = "llm:8080"
+ return s
+ }(),
+ wantErr: true,
+ },
+ {
+ name: "empty mcp targets",
+ spec: &networking.DxgateService{
+ Service: &networking.DxgateService_Mcp{Mcp:
&networking.MCPService{}},
+ },
+ wantErr: true,
+ },
+ {
+ name: "duplicate mcp target",
+ spec: func() *networking.DxgateService {
+ s := mcp()
+ s.GetMcp().Targets = append(s.GetMcp().Targets,
s.GetMcp().Targets[0])
+ return s
+ }(),
+ wantErr: true,
+ },
+ {
+ name: "a2a ambiguous target",
+ spec: func() *networking.DxgateService {
+ s := a2a()
+ s.GetA2A().Host = "agent.example.com"
+ return s
+ }(),
+ wantErr: true,
+ },
+ {
+ name: "bad policy",
+ spec: func() *networking.DxgateService {
+ s := openAI()
+ s.Policies = &networking.DxgateServicePolicies{
+ Retry:
&networking.RetryPolicy{Attempts: 0, StatusCodes: []uint32{200}},
+ Timeout: durationpb.New(-time.Second),
+ }
+ return s
+ }(),
+ wantErr: true,
+ },
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ _, err := ValidateDxgateService(makeConfig(tc.spec))
+ if (err != nil) != tc.wantErr {
+ t.Fatalf("got err=%v, wantErr=%v", err,
tc.wantErr)
+ }
+ })
+ }
+}
+
func TestValidateServiceEntry(t *testing.T) {
valid := func() *networking.ServiceEntry {
return &networking.ServiceEntry{
diff --git a/samples/ai-mesh/README.md b/samples/ai-mesh/README.md
new file mode 100644
index 00000000..5fca59e4
--- /dev/null
+++ b/samples/ai-mesh/README.md
@@ -0,0 +1,18 @@
+# Mesh-native AI services
+
+This sample uses one API: `networking.dubbo.apache.org/v1alpha3`
+`DxgateService`. Ordinary HTTP backends remain core Kubernetes `Service`
+objects. `dubbod` compiles both kinds of `HTTPRoute` backend into RDS and
+delivers it to dxgate over xDS.
+
+```bash
+kubectl create namespace ai-mesh
+kubectl -n ai-mesh apply -f gateway.yaml
+kubectl -n ai-mesh apply -f backends.yaml
+kubectl -n ai-mesh apply -f services.yaml
+kubectl -n ai-mesh apply -f routes.yaml
+```
+
+The sample image `kdubbo/agent-mock:latest` is the no-key E2E fixture in
+`tests/e2e/agentmock`; replace its Services and Secret values with production
+backends and credentials.
diff --git a/samples/ai-mesh/backends.yaml b/samples/ai-mesh/backends.yaml
new file mode 100644
index 00000000..ec4e6fb9
--- /dev/null
+++ b/samples/ai-mesh/backends.yaml
@@ -0,0 +1,119 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+ name: agent-mock
+spec:
+ replicas: 1
+ selector:
+ matchLabels:
+ app: agent-mock
+ template:
+ metadata:
+ labels:
+ app: agent-mock
+ spec:
+ containers:
+ - name: http
+ image: kdubbo/agent-mock:latest
+ imagePullPolicy: IfNotPresent
+ env:
+ - {name: MOCK_MODE, value: http}
+ - {name: PORT, value: "8080"}
+ ports:
+ - {name: http, containerPort: 8080}
+ - name: openai
+ image: kdubbo/agent-mock:latest
+ imagePullPolicy: IfNotPresent
+ env:
+ - {name: MOCK_MODE, value: openai}
+ - {name: PORT, value: "8081"}
+ ports:
+ - {name: openai, containerPort: 8081}
+ - name: anthropic
+ image: kdubbo/agent-mock:latest
+ imagePullPolicy: IfNotPresent
+ env:
+ - {name: MOCK_MODE, value: anthropic}
+ - {name: PORT, value: "8082"}
+ ports:
+ - {name: anthropic, containerPort: 8082}
+ - name: mcp-search
+ image: kdubbo/agent-mock:latest
+ imagePullPolicy: IfNotPresent
+ env:
+ - {name: MOCK_MODE, value: mcp}
+ - {name: MOCK_NAME, value: search}
+ - {name: PORT, value: "8083"}
+ ports:
+ - {name: mcp-search, containerPort: 8083}
+ - name: mcp-calendar
+ image: kdubbo/agent-mock:latest
+ imagePullPolicy: IfNotPresent
+ env:
+ - {name: MOCK_MODE, value: mcp}
+ - {name: MOCK_NAME, value: calendar}
+ - {name: PORT, value: "8084"}
+ ports:
+ - {name: mcp-calendar, containerPort: 8084}
+ - name: a2a
+ image: kdubbo/agent-mock:latest
+ imagePullPolicy: IfNotPresent
+ env:
+ - {name: MOCK_MODE, value: a2a}
+ - {name: PORT, value: "8085"}
+ ports:
+ - {name: a2a, containerPort: 8085}
+---
+apiVersion: v1
+kind: Service
+metadata:
+ name: mock-http
+spec:
+ selector: {app: agent-mock}
+ ports:
+ - {name: http, port: 8080, targetPort: http}
+---
+apiVersion: v1
+kind: Service
+metadata:
+ name: mock-openai
+spec:
+ selector: {app: agent-mock}
+ ports:
+ - {name: http, port: 8081, targetPort: openai}
+---
+apiVersion: v1
+kind: Service
+metadata:
+ name: mock-anthropic
+spec:
+ selector: {app: agent-mock}
+ ports:
+ - {name: http, port: 8082, targetPort: anthropic}
+---
+apiVersion: v1
+kind: Service
+metadata:
+ name: mock-mcp-search
+spec:
+ selector: {app: agent-mock}
+ ports:
+ - {name: http, port: 8083, targetPort: mcp-search}
+---
+apiVersion: v1
+kind: Service
+metadata:
+ name: mock-mcp-calendar
+spec:
+ selector: {app: agent-mock}
+ ports:
+ - {name: http, port: 8084, targetPort: mcp-calendar}
+---
+apiVersion: v1
+kind: Service
+metadata:
+ name: mock-a2a
+spec:
+ selector: {app: agent-mock}
+ ports:
+ - {name: http, port: 8085, targetPort: a2a}
diff --git a/samples/ai-mesh/gateway.yaml b/samples/ai-mesh/gateway.yaml
new file mode 100644
index 00000000..86ea738f
--- /dev/null
+++ b/samples/ai-mesh/gateway.yaml
@@ -0,0 +1,13 @@
+apiVersion: gateway.networking.k8s.io/v1
+kind: Gateway
+metadata:
+ name: public
+spec:
+ gatewayClassName: dubbo
+ listeners:
+ - name: http
+ protocol: HTTP
+ port: 80
+ allowedRoutes:
+ namespaces:
+ from: Same
diff --git a/samples/ai-mesh/routes.yaml b/samples/ai-mesh/routes.yaml
new file mode 100644
index 00000000..ad2dd293
--- /dev/null
+++ b/samples/ai-mesh/routes.yaml
@@ -0,0 +1,86 @@
+apiVersion: gateway.networking.k8s.io/v1
+kind: HTTPRoute
+metadata:
+ name: ordinary-http
+spec:
+ parentRefs:
+ - {name: public, sectionName: http}
+ rules:
+ - matches:
+ - path: {type: PathPrefix, value: /users}
+ - path: {type: PathPrefix, value: /orders}
+ backendRefs:
+ - {name: mock-http, port: 8080}
+---
+apiVersion: gateway.networking.k8s.io/v1
+kind: HTTPRoute
+metadata:
+ name: openai
+spec:
+ parentRefs:
+ - {name: public, sectionName: http}
+ rules:
+ - matches:
+ - path: {type: PathPrefix, value: /openai}
+ filters:
+ - type: URLRewrite
+ urlRewrite:
+ path:
+ type: ReplacePrefixMatch
+ replacePrefixMatch: /v1
+ backendRefs:
+ - group: networking.dubbo.apache.org
+ kind: DxgateService
+ name: openai
+---
+apiVersion: gateway.networking.k8s.io/v1
+kind: HTTPRoute
+metadata:
+ name: anthropic
+spec:
+ parentRefs:
+ - {name: public, sectionName: http}
+ rules:
+ - matches:
+ - path: {type: PathPrefix, value: /anthropic}
+ filters:
+ - type: URLRewrite
+ urlRewrite:
+ path:
+ type: ReplacePrefixMatch
+ replacePrefixMatch: /v1
+ backendRefs:
+ - group: networking.dubbo.apache.org
+ kind: DxgateService
+ name: anthropic
+---
+apiVersion: gateway.networking.k8s.io/v1
+kind: HTTPRoute
+metadata:
+ name: tools
+spec:
+ parentRefs:
+ - {name: public, sectionName: http}
+ rules:
+ - matches:
+ - path: {type: Exact, value: /mcp}
+ backendRefs:
+ - group: networking.dubbo.apache.org
+ kind: DxgateService
+ name: tools
+---
+apiVersion: gateway.networking.k8s.io/v1
+kind: HTTPRoute
+metadata:
+ name: planner
+spec:
+ parentRefs:
+ - {name: public, sectionName: http}
+ rules:
+ - matches:
+ - path: {type: Exact, value: /a2a}
+ - path: {type: Exact, value: /.well-known/agent-card.json}
+ backendRefs:
+ - group: networking.dubbo.apache.org
+ kind: DxgateService
+ name: planner
diff --git a/samples/ai-mesh/services.yaml b/samples/ai-mesh/services.yaml
new file mode 100644
index 00000000..48aabb9b
--- /dev/null
+++ b/samples/ai-mesh/services.yaml
@@ -0,0 +1,80 @@
+apiVersion: v1
+kind: Secret
+metadata:
+ name: agent-credentials
+type: Opaque
+stringData:
+ openai: mock-openai-key
+ anthropic: mock-anthropic-key
+ client: mock-client-key
+---
+apiVersion: networking.dubbo.apache.org/v1alpha3
+kind: DxgateService
+metadata:
+ name: openai
+spec:
+ ai:
+ endpoint: http://mock-openai:8081/v1
+ provider:
+ openai: {}
+ credential:
+ name: agent-credentials
+ key: openai
+ models: [gpt-mock]
+ routes:
+ /v1/chat/completions: COMPLETIONS
+ policies:
+ auth:
+ header: x-client-key
+ secretRef:
+ name: agent-credentials
+ key: client
+ timeout: 5s
+ retry:
+ attempts: 2
+ statusCodes: [502, 503, 504]
+---
+apiVersion: networking.dubbo.apache.org/v1alpha3
+kind: DxgateService
+metadata:
+ name: anthropic
+spec:
+ ai:
+ endpoint: http://mock-anthropic:8082
+ provider:
+ anthropic:
+ model: claude-mock
+ credential:
+ name: agent-credentials
+ key: anthropic
+ models: [claude-mock]
+ routes:
+ /v1/chat/completions: COMPLETIONS
+---
+apiVersion: networking.dubbo.apache.org/v1alpha3
+kind: DxgateService
+metadata:
+ name: tools
+spec:
+ mcp:
+ targets:
+ - name: search
+ static:
+ backendRef: {name: mock-mcp-search}
+ port: 8083
+ tools: [search]
+ - name: calendar
+ static:
+ backendRef: {name: mock-mcp-calendar}
+ port: 8084
+ tools: [calendar]
+---
+apiVersion: networking.dubbo.apache.org/v1alpha3
+kind: DxgateService
+metadata:
+ name: planner
+spec:
+ a2a:
+ backendRef: {name: mock-a2a}
+ port: 8085
+ agent: planner
diff --git a/tests/e2e/agentmock/Dockerfile b/tests/e2e/agentmock/Dockerfile
new file mode 100644
index 00000000..d9c11d3b
--- /dev/null
+++ b/tests/e2e/agentmock/Dockerfile
@@ -0,0 +1,9 @@
+FROM golang:1.25-alpine AS builder
+WORKDIR /src
+COPY main.go .
+RUN CGO_ENABLED=0 go build -trimpath -ldflags="-s -w" -o /agentmock main.go
+
+FROM scratch
+COPY --from=builder /agentmock /agentmock
+USER 65532:65532
+ENTRYPOINT ["/agentmock"]
diff --git a/tests/e2e/agentmock/main.go b/tests/e2e/agentmock/main.go
new file mode 100644
index 00000000..c7a91265
--- /dev/null
+++ b/tests/e2e/agentmock/main.go
@@ -0,0 +1,94 @@
+// 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.
+
+package main
+
+import (
+ "encoding/json"
+ "fmt"
+ "log"
+ "net/http"
+ "os"
+ "time"
+)
+
+type request struct {
+ ID any `json:"id"`
+ Method string `json:"method"`
+ Model string `json:"model"`
+ Params map[string]any `json:"params"`
+}
+
+func main() {
+ mode := env("MOCK_MODE", "http")
+ name := env("MOCK_NAME", mode)
+ port := env("PORT", "8080")
+ mux := http.NewServeMux()
+ mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request)
{
+ w.WriteHeader(http.StatusOK)
+ })
+ mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+ switch mode {
+ case "http":
+ writeJSON(w, map[string]any{"service": "http", "path":
r.URL.Path})
+ case "openai":
+ var in request
+ _ = json.NewDecoder(r.Body).Decode(&in)
+ writeJSON(w, map[string]any{
+ "id": "chatcmpl-mock", "object":
"chat.completion", "model": in.Model,
+ "provider_authorization":
r.Header.Get("Authorization"),
+ "choices":
[]any{map[string]any{"index": 0, "message": map[string]any{"role": "assistant",
"content": "openai-mock"}, "finish_reason": "stop"}},
+ "usage":
map[string]any{"prompt_tokens": 2, "completion_tokens": 3, "total_tokens": 5},
+ })
+ case "anthropic":
+ var in request
+ _ = json.NewDecoder(r.Body).Decode(&in)
+ writeJSON(w, map[string]any{
+ "id": "msg-mock", "type": "message", "role":
"assistant", "model": in.Model,
+ "content": []any{map[string]any{"type":
"text", "text": "anthropic-mock"}},
+ "stop_reason": "end_turn",
+ "usage": map[string]any{"input_tokens":
4, "output_tokens": 6},
+ })
+ case "mcp":
+ var in request
+ _ = json.NewDecoder(r.Body).Decode(&in)
+ result := map[string]any{}
+ switch in.Method {
+ case "initialize":
+ result = map[string]any{"protocolVersion":
"2025-03-26", "serverInfo": map[string]any{"name": name, "version": "1.0.0"},
"capabilities": map[string]any{"tools": map[string]any{}}}
+ case "tools/list":
+ result = map[string]any{"tools":
[]any{map[string]any{"name": name, "description": name + " mock tool",
"inputSchema": map[string]any{"type": "object"}}}}
+ case "tools/call":
+ result = map[string]any{"content":
[]any{map[string]any{"type": "text", "text": name + "-ok"}}}
+ }
+ writeJSON(w, map[string]any{"jsonrpc": "2.0", "id":
in.ID, "result": result})
+ case "a2a":
+ if r.Method == http.MethodGet {
+ writeJSON(w, map[string]any{"name": "planner",
"version": "1.0.0", "url": "http://mock-a2a:8080/a2a"})
+ return
+ }
+ var in request
+ _ = json.NewDecoder(r.Body).Decode(&in)
+ writeJSON(w, map[string]any{"jsonrpc": "2.0", "id":
in.ID, "result": map[string]any{"id": "task-mock", "status":
map[string]any{"state": "completed"}}})
+ default:
+ http.Error(w, fmt.Sprintf("unknown MOCK_MODE %q",
mode), http.StatusInternalServerError)
+ }
+ })
+ server := &http.Server{Addr: ":" + port, Handler: mux,
ReadHeaderTimeout: 5 * time.Second}
+ log.Printf("agent mock mode=%s name=%s addr=%s", mode, name,
server.Addr)
+ log.Fatal(server.ListenAndServe())
+}
+
+func env(name, fallback string) string {
+ if value := os.Getenv(name); value != "" {
+ return value
+ }
+ return fallback
+}
+
+func writeJSON(w http.ResponseWriter, value any) {
+ w.Header().Set("Content-Type", "application/json")
+ _ = json.NewEncoder(w).Encode(value)
+}
diff --git a/tests/e2e/run.sh b/tests/e2e/run.sh
index 042a7de9..efb0d795 100755
--- a/tests/e2e/run.sh
+++ b/tests/e2e/run.sh
@@ -32,6 +32,7 @@
# KIND path to the kind binary (default: kind)
# KIND_NODE_IMAGE kind node image override (default: kind release
default)
# ACTIVATION_E2E install KEDA and run real scale-to-zero E2E (default: 0)
+# AI_MESH_E2E run no-key HTTP/LLM/MCP/A2A E2E (default: 0)
# DXGATE_IMAGE prebuilt dxgate image used by managed Gateways
# KEDA_VERSION pinned KEDA chart/app version (default: 2.20.2)
@@ -52,7 +53,9 @@ PREVIOUS_CHART=""
KIND="${KIND:-kind}"
KIND_NODE_IMAGE="${KIND_NODE_IMAGE:-}"
ACTIVATION_E2E="${ACTIVATION_E2E:-0}"
+AI_MESH_E2E="${AI_MESH_E2E:-0}"
DXGATE_IMAGE="${DXGATE_IMAGE:-kdubbo/dxgate:latest}"
+AGENT_MOCK_IMAGE="${AGENT_MOCK_IMAGE:-kdubbo/agent-mock:latest}"
ACTIVATION_APP_IMAGE="${ACTIVATION_APP_IMAGE:-kdubbo/activation-e2e:latest}"
ACTIVATION_CLIENT_IMAGE="${ACTIVATION_CLIENT_IMAGE:-kdubbo/activation-client:latest}"
KEDA_VERSION="${KEDA_VERSION:-2.20.2}"
@@ -81,6 +84,7 @@ fail() {
cleanup() {
if [[ -n "${PF_PID:-}" ]]; then kill "${PF_PID}" 2>/dev/null || true; fi
+ if [[ -n "${AI_PF_PID:-}" ]]; then kill "${AI_PF_PID}" 2>/dev/null || true;
fi
if [[ "${KEEP_CLUSTER:-0}" != "1" ]]; then
"${KIND}" delete cluster --name "${CLUSTER_NAME}" || true
fi
@@ -90,6 +94,14 @@ cleanup() {
}
trap cleanup EXIT
+apply_activation_fixture() {
+ sed \
+ -e "s#kdubbo/activation-e2e:latest#${ACTIVATION_APP_IMAGE}#g" \
+ -e "s#kdubbo/activation-client:latest#${ACTIVATION_CLIENT_IMAGE}#g" \
+ "$1" \
+ | "${KUBECTL[@]}" apply -f -
+}
+
prepare_previous_chart() {
if [[ -n "${UPGRADE_FROM_CHART}" ]]; then
[[ -e "${UPGRADE_FROM_CHART}" ]] || fail "previous chart not found:
${UPGRADE_FROM_CHART}"
@@ -152,11 +164,18 @@ log "loading ${IMAGE} into kind"
if [[ "${IMAGE}" != "${UPGRADE_FROM_IMAGE}" ]]; then
"${KIND}" load docker-image "${UPGRADE_FROM_IMAGE}" --name "${CLUSTER_NAME}"
fi
-if [[ "${ACTIVATION_E2E}" == "1" ]]; then
+if [[ "${ACTIVATION_E2E}" == "1" || "${AI_MESH_E2E}" == "1" ]]; then
docker image inspect "${DXGATE_IMAGE}" >/dev/null 2>&1 \
- || fail "ACTIVATION_E2E requires prebuilt ${DXGATE_IMAGE}"
- log "building ${ACTIVATION_APP_IMAGE}"
- docker build -t "${ACTIVATION_APP_IMAGE}" "${ROOT}/tests/e2e/activationapp"
+ || fail "data-plane E2E requires prebuilt ${DXGATE_IMAGE}"
+fi
+if [[ "${ACTIVATION_E2E}" == "1" ]]; then
+ if [[ "${SKIP_BUILD:-0}" != "1" ]]; then
+ log "building ${ACTIVATION_APP_IMAGE}"
+ docker build -t "${ACTIVATION_APP_IMAGE}" "${ROOT}/tests/e2e/activationapp"
+ else
+ docker image inspect "${ACTIVATION_APP_IMAGE}" >/dev/null 2>&1 \
+ || fail "SKIP_BUILD=1 requires prebuilt ${ACTIVATION_APP_IMAGE}"
+ fi
docker tag "${IMAGE}" "${ACTIVATION_CLIENT_IMAGE}"
log "loading activation data-plane images into kind"
"${KIND}" load docker-image \
@@ -165,6 +184,20 @@ if [[ "${ACTIVATION_E2E}" == "1" ]]; then
"${ACTIVATION_CLIENT_IMAGE}" \
--name "${CLUSTER_NAME}"
fi
+if [[ "${AI_MESH_E2E}" == "1" ]]; then
+ if [[ "${SKIP_BUILD:-0}" != "1" ]]; then
+ log "building ${AGENT_MOCK_IMAGE}"
+ docker build -t "${AGENT_MOCK_IMAGE}" "${ROOT}/tests/e2e/agentmock"
+ else
+ docker image inspect "${AGENT_MOCK_IMAGE}" >/dev/null 2>&1 \
+ || fail "SKIP_BUILD=1 requires prebuilt ${AGENT_MOCK_IMAGE}"
+ fi
+ log "loading AI mesh data-plane images into kind"
+ "${KIND}" load docker-image \
+ "${DXGATE_IMAGE}" \
+ "${AGENT_MOCK_IMAGE}" \
+ --name "${CLUSTER_NAME}"
+fi
log "installing Gateway API CRDs"
# Pin to the sigs.k8s.io/gateway-api version in go.mod. HTTPRoute retry is an
@@ -354,6 +387,93 @@ GATEWAY_ENV="$("${KUBECTL[@]}" -n "${APP_NS}" get deploy
public-dubbo \
| grep -qx metadata.name \
|| fail "gateway does not inject POD_NAME; demand reports would not be
attributable to a replica"
+if [[ "${AI_MESH_E2E}" == "1" ]]; then
+ log "applying mesh-native HTTP, OpenAI, Anthropic, MCP, and A2A sample"
+ sed "s#kdubbo/agent-mock:latest#${AGENT_MOCK_IMAGE}#g" \
+ "${ROOT}/samples/ai-mesh/backends.yaml" \
+ | "${KUBECTL[@]}" -n "${APP_NS}" apply -f -
+ "${KUBECTL[@]}" -n "${APP_NS}" apply -f
"${ROOT}/samples/ai-mesh/services.yaml"
+ "${KUBECTL[@]}" -n "${APP_NS}" apply -f "${ROOT}/samples/ai-mesh/routes.yaml"
+ "${KUBECTL[@]}" -n "${APP_NS}" rollout status deploy/agent-mock
--timeout=300s \
+ || fail "agent mock deployment did not become ready"
+ "${KUBECTL[@]}" -n "${APP_NS}" rollout status deploy/public-dubbo
--timeout=300s \
+ || fail "mesh gateway deployment did not become ready"
+
+ "${KUBECTL[@]}" get crd dxgateservices.networking.dubbo.apache.org
>/dev/null \
+ || fail "DxgateService CRD is missing"
+ "${KUBECTL[@]}" -n "${APP_NS}" get role public-dubbo-credentials >/dev/null \
+ || fail "dxgate credential Role is missing"
+ "${KUBECTL[@]}" -n "${APP_NS}" get rolebinding public-dubbo-credentials
>/dev/null \
+ || fail "dxgate credential RoleBinding is missing"
+ "${KUBECTL[@]}" auth can-i get secret/agent-credentials \
+ --as="system:serviceaccount:${APP_NS}:public-dubbo" -n "${APP_NS}" \
+ | grep -qx yes || fail "dxgate ServiceAccount cannot resolve referenced
Secret"
+
+ log "port-forwarding mesh gateway"
+ "${KUBECTL[@]}" -n "${APP_NS}" port-forward svc/public-dubbo 18081:80
>/dev/null 2>&1 &
+ AI_PF_PID=$!
+ ai_get() { curl -sf --max-time 10 "http://127.0.0.1:18081$1";; }
+ ai_post() {
+ local path="$1" body="$2"
+ curl -sf --max-time 10 -H 'content-type: application/json' \
+ -d "${body}" "http://127.0.0.1:18081${path}";
+ }
+ ai_openai() {
+ curl -sf --max-time 10 -H 'content-type: application/json' \
+ -H 'x-client-key: mock-client-key' \
+ -d '{"model":"gpt-mock","messages":[{"role":"user","content":"ping"}]}' \
+ http://127.0.0.1:18081/openai/chat/completions
+ }
+
+ retry "ordinary /users Service route" ai_get /users
+ [[ "$(ai_get /users | jq -r .path)" == "/users" ]] \
+ || fail "ordinary /users route returned the wrong backend response"
+ [[ "$(ai_get /orders | jq -r .path)" == "/orders" ]] \
+ || fail "ordinary /orders route returned the wrong backend response"
+ retry "OpenAI DxgateService route and Secret resolution" ai_openai
+ OPENAI_RESPONSE="$(ai_openai)"
+ [[ "$(jq -r .choices[0].message.content <<<"${OPENAI_RESPONSE}")" ==
"openai-mock" ]] \
+ || fail "OpenAI mock response was not proxied"
+ [[ "$(jq -r .provider_authorization <<<"${OPENAI_RESPONSE}")" == "Bearer
mock-openai-key" ]] \
+ || fail "OpenAI provider credential was not resolved from the Secret"
+
+ ANTHROPIC_RESPONSE="$(ai_post /anthropic/chat/completions \
+ '{"model":"claude-mock","messages":[{"role":"user","content":"ping"}]}')"
+ [[ "$(jq -r .choices[0].message.content <<<"${ANTHROPIC_RESPONSE}")" ==
"anthropic-mock" ]] \
+ || fail "Anthropic native response was not translated to OpenAI"
+
+ MCP_RESPONSE="$(ai_post /mcp
'{"jsonrpc":"2.0","id":1,"method":"tools/list"}')"
+ [[ "$(jq -r '[.result.tools[].name] | sort | join(",")'
<<<"${MCP_RESPONSE}")" == "calendar,search" ]] \
+ || fail "MCP tools/list was not federated across both targets"
+ [[ "$(ai_post /mcp
'{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"search","arguments":{}}}'
\
+ | jq -r .result.content[0].text)" == "search-ok" ]] \
+ || fail "MCP tools/call did not reach the selected target"
+
+ [[ "$(ai_get /.well-known/agent-card.json | jq -r .name)" == "planner" ]] \
+ || fail "A2A Agent Card was not proxied"
+ [[ "$(ai_post /a2a
'{"jsonrpc":"2.0","id":3,"method":"message/send","params":{}}' \
+ | jq -r .result.status.state)" == "completed" ]] \
+ || fail "A2A task request did not complete"
+
+ check_agent_config_programmed() {
+ local pod
+ pod="$("${KUBECTL[@]}" -n "${APP_NS}" get pods \
+ -l gateway.networking.k8s.io/gateway-name=public \
+ -o jsonpath='{.items[0].metadata.name}')"
+ "${KUBECTL[@]}" get --raw \
+ "/api/v1/namespaces/${APP_NS}/pods/${pod}:26021/proxy/debug/config" \
+ | jq -e '
+ ([.providers[].name] | length) == 2 and
+ ([.backends[] | select(.type == "llm")] | length) == 2 and
+ ([.backends[] | select(.type == "mcp")] | length) == 2 and
+ ([.backends[] | select(.type == "a2a")] | length) == 1 and
+ ([.routes[] | .protocol] | sort | join(",")) == "a2a,llm,llm,mcp"
+ ' >/dev/null
+ }
+ retry "compiled AgentConfig visible in dxgate /debug/config"
check_agent_config_programmed
+ log "mesh-native HTTP, LLM, MCP, and A2A E2E passed"
+fi
+
if [[ "${ACTIVATION_E2E}" == "1" ]]; then
log "asserting multiple Gateways have isolated resources"
"${KUBECTL[@]}" -n "${APP_NS}" get deploy dxgate-gateway public-dubbo
>/dev/null \
@@ -364,7 +484,7 @@ if [[ "${ACTIVATION_E2E}" == "1" ]]; then
|| fail "second managed gateway did not become ready"
log "deploying the proxyless activation target and KEDA ScaledObject"
- "${KUBECTL[@]}" apply -f
"${ROOT}/tests/e2e/testdata/eastwest-activation.yaml"
+ apply_activation_fixture
"${ROOT}/tests/e2e/testdata/eastwest-activation.yaml"
"${KUBECTL[@]}" apply -f
"${ROOT}/tests/e2e/testdata/eastwest-activation-scaledobject.yaml"
"${KUBECTL[@]}" -n "${APP_NS}" wait --for=condition=Ready
scaledobject/payment --timeout=180s \
|| fail "KEDA ScaledObject did not become ready"
@@ -426,7 +546,7 @@ if [[ "${ACTIVATION_E2E}" == "1" ]]; then
log "sending one proxyless request while payment is at zero"
"${KUBECTL[@]}" -n "${APP_NS}" delete pod payment-client --ignore-not-found
- "${KUBECTL[@]}" apply -f
"${ROOT}/tests/e2e/testdata/eastwest-activation-client.yaml"
+ apply_activation_fixture
"${ROOT}/tests/e2e/testdata/eastwest-activation-client.yaml"
activation_metrics() {
local pod
while read -r pod; do
@@ -475,7 +595,7 @@ if [[ "${ACTIVATION_E2E}" == "1" ]]; then
local before after
before="$(activation_payment_requests_total)"
"${KUBECTL[@]}" -n "${APP_NS}" delete pod payment-client
--ignore-not-found --wait=true >/dev/null
- "${KUBECTL[@]}" apply -f
"${ROOT}/tests/e2e/testdata/eastwest-activation-client.yaml" >/dev/null
+ apply_activation_fixture
"${ROOT}/tests/e2e/testdata/eastwest-activation-client.yaml" >/dev/null
"${KUBECTL[@]}" -n "${APP_NS}" wait \
--for=jsonpath='{.status.containerStatuses[?(@.name=="client")].state.terminated.exitCode}'=0
\
pod/payment-client --timeout=30s >/dev/null || return 1
