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 0226285e Complete production service activation path (#1006)
0226285e is described below
commit 0226285ea46613234bdfee4c4ea8e3c3a73f9736
Author: mfordjody <[email protected]>
AuthorDate: Sat Aug 8 21:37:24 2026 +0800
Complete production service activation path (#1006)
* feat: complete production service activation
* test: capture managed gateway crash logs
* ci: run activation e2e natively on arm64
---
.github/workflows/ci.yml | 50 ++++-
dubbod/discovery/cmd/app/grpc_outbound.go | 7 +
dubbod/discovery/cmd/app/grpc_outbound_test.go | 34 ++++
.../discovery/pkg/activation/cluster_readiness.go | 140 ++++++++++++++
.../pkg/activation/cluster_readiness_test.go | 62 +++++++
dubbod/discovery/pkg/activation/controller.go | 78 ++++++--
dubbod/discovery/pkg/activation/policy.go | 41 ++---
dubbod/discovery/pkg/activation/policy_test.go | 42 +++--
dubbod/discovery/pkg/bootstrap/activation.go | 2 +-
dubbod/discovery/pkg/bootstrap/server.go | 22 ++-
.../config/kube/gateway/deployment_controller.go | 76 ++++++--
.../kube/gateway/deployment_controller_test.go | 32 +++-
dubbod/discovery/pkg/model/push_context.go | 128 +++++++++----
dubbod/discovery/pkg/model/push_context_test.go | 49 +++++
dubbod/discovery/pkg/networking/grpcgen/rds.go | 36 ++--
.../discovery/pkg/networking/grpcgen/rds_test.go | 39 ++++
dubbod/discovery/pkg/xds/delta_test.go | 6 +
dubbod/discovery/pkg/xds/xdsgen.go | 5 +-
manifests/charts/dubbod/templates/clusterrole.yaml | 6 +
manifests/charts/dubbod/templates/deployment.yaml | 3 +
manifests/charts/dubbod/values.yaml | 4 +
samples/activation/README.md | 30 +--
samples/activation/activation-policy.yaml | 6 +-
samples/activation/scaledobject.yaml | 8 +-
tests/e2e/run.sh | 203 ++++++++++++++++++++-
.../testdata/eastwest-activation-scaledobject.yaml | 39 ++++
26 files changed, 993 insertions(+), 155 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index c0344dc3..79b5fdb0 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -228,6 +228,51 @@ jobs:
UPGRADE_FROM_VERSION: '0.4.3'
run: make test-e2e
+ activation-e2e:
+ name: Activation E2E (KEDA)
+ runs-on: ubuntu-24.04-arm
+ timeout-minutes: 45
+ if: github.repository == 'apache/dubbo-kubernetes'
+ env:
+ DOCKER_DEFAULT_PLATFORM: linux/arm64
+ steps:
+ - name: Checkout code
+ uses: actions/checkout@v4
+
+ - name: Verify native ARM64 runner
+ run: |
+ test "$(uname -m)" = "aarch64"
+ test "$(docker info --format '{{.Architecture}}')" = "aarch64"
+
+ # Pin the data plane source that implements request holding and demand
+ # reporting. A floating main branch could make a control-plane PR pass
+ # against behavior it did not declare.
+ - name: Checkout dxgate
+ uses: actions/checkout@v4
+ with:
+ repository: kdubbo/dxgate
+ ref: d683797f44c9ba4c6e1eb6cdb2da21460b00cbfd
+ path: .e2e/dxgate
+
+ - name: Build dxgate activation image
+ run: docker build -t kdubbo/dxgate:ga-e2e .e2e/dxgate
+
+ - name: Install Helm
+ run: tools/ci/install-helm.sh v3.16.4
+
+ - name: Install kind
+ run: tools/ci/install-kind.sh v0.30.0
+
+ - name: Run real KEDA activation
+ env:
+ ACTIVATION_E2E: '1'
+ CLUSTER_NAME: dubbo-activation-ga
+ DXGATE_IMAGE: kdubbo/dxgate:ga-e2e
+ IMAGE: kdubbo/dubbod:activation-ga-e2e
+ UPGRADE_FROM_IMAGE: kdubbo/dubbod:activation-ga-e2e
+ UPGRADE_FROM_VERSION: '0.4.3'
+ run: make test-e2e
+
required:
name: CI Required
runs-on: ubuntu-slim
@@ -243,6 +288,7 @@ jobs:
- build
- helm
- e2e
+ - activation-e2e
steps:
- name: Verify required jobs passed
env:
@@ -253,6 +299,7 @@ jobs:
BUILD_RESULT: ${{ needs.build.result }}
HELM_RESULT: ${{ needs.helm.result }}
E2E_RESULT: ${{ needs.e2e.result }}
+ ACTIVATION_E2E_RESULT: ${{ needs.activation-e2e.result }}
run: |
failed=0
for job_result in \
@@ -262,7 +309,8 @@ jobs:
"performance=${PERFORMANCE_RESULT}" \
"build=${BUILD_RESULT}" \
"helm=${HELM_RESULT}" \
- "e2e=${E2E_RESULT}"; do
+ "e2e=${E2E_RESULT}" \
+ "activation-e2e=${ACTIVATION_E2E_RESULT}"; do
job="${job_result%%=*}"
result="${job_result#*=}"
echo "${job}: ${result}"
diff --git a/dubbod/discovery/cmd/app/grpc_outbound.go
b/dubbod/discovery/cmd/app/grpc_outbound.go
index 57e191c9..6f1bde45 100644
--- a/dubbod/discovery/cmd/app/grpc_outbound.go
+++ b/dubbod/discovery/cmd/app/grpc_outbound.go
@@ -1011,6 +1011,13 @@ func runSampleRequest(ctx context.Context, adsClient
*sampleADSClient, clients *
}
continue
}
+ if resp.StatusCode >= http.StatusBadRequest {
+ return "", fmt.Errorf(
+ "upstream returned HTTP status %d: %s",
+ resp.StatusCode,
+ strings.TrimSpace(string(body)),
+ )
+ }
return strings.TrimSpace(string(body)), nil
}
return "", lastErr
diff --git a/dubbod/discovery/cmd/app/grpc_outbound_test.go
b/dubbod/discovery/cmd/app/grpc_outbound_test.go
index 2337413c..a871cdf9 100644
--- a/dubbod/discovery/cmd/app/grpc_outbound_test.go
+++ b/dubbod/discovery/cmd/app/grpc_outbound_test.go
@@ -351,6 +351,40 @@ func TestRunSampleRequestsRetriesConfiguredStatus(t
*testing.T) {
}
}
+func TestRunSampleRequestsRejectsUnsuccessfulStatus(t *testing.T) {
+ server := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, _ *http.Request) {
+ http.Error(w, "route not found", http.StatusNotFound)
+ }))
+ defer server.Close()
+
+ endpoint := endpointForServer(t, server)
+ cluster := "outbound|8080||payment.e2e.svc.cluster.local"
+ snapshot := xdsRouteSnapshot{
+ Host: "payment.e2e.svc.cluster.local",
+ Port: 8080,
+ Destinations: []xdsDestination{{
+ Cluster: cluster,
+ Host: "payment.e2e.svc.cluster.local",
+ Weight: 100,
+ Endpoints: []xdsEndpoint{endpoint},
+ }},
+ }
+ client := &sampleADSClient{
+ host: snapshot.Host,
+ port: snapshot.Port,
+ path: "/health",
+ route: map[string]uint32{cluster: 100},
+ endpoints: map[string][]xdsEndpoint{cluster: {endpoint}},
+ clusterTLS: map[string]*tlsv1.UpstreamTlsContext{},
+ requestHeaders: http.Header{},
+ }
+
+ _, err := runSampleRequestsWithOutput(context.Background(), client,
snapshot, 1, 0, time.Second, nil)
+ if err == nil || err.Error() != "upstream returned HTTP status 404:
route not found" {
+ t.Fatalf("runSampleRequestsWithOutput() error = %v", err)
+ }
+}
+
func TestRunSampleRequestsRetriesConnectionFailureOnNextEndpoint(t *testing.T)
{
goodServer := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, _ *http.Request) {
_, _ = w.Write([]byte("recovered"))
diff --git a/dubbod/discovery/pkg/activation/cluster_readiness.go
b/dubbod/discovery/pkg/activation/cluster_readiness.go
new file mode 100644
index 00000000..fd74ab03
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/cluster_readiness.go
@@ -0,0 +1,140 @@
+// 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 activation
+
+import (
+ "strings"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/features"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/gvr"
+ "github.com/apache/dubbo-kubernetes/pkg/kube"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/controllers"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/kclient"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/kubetypes"
+ clientnetworking
"github.com/kdubbo/client-go/pkg/apis/networking/v1alpha3"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+ klabels "k8s.io/apimachinery/pkg/labels"
+ "k8s.io/apimachinery/pkg/runtime/schema"
+ gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
+)
+
+var scaledObjectGVR = schema.GroupVersionResource{
+ Group: "keda.sh", Version: "v1alpha1", Resource: "scaledobjects",
+}
+
+// clusterReadiness derives policy status from objects all dubbod replicas see.
+// Process-local streams are deliberately not used: in HA, KEDA and gateways
+// can connect to different replicas, so no replica has a complete view.
+type clusterReadiness struct {
+ scaledObjects kclient.Informer[controllers.Object]
+ gateways kclient.Informer[*gatewayv1.Gateway]
+ gatewayClasses kclient.Informer[*gatewayv1.GatewayClass]
+}
+
+func newClusterReadiness(client kube.Client) *clusterReadiness {
+ filter := kclient.Filter{ObjectFilter: client.ObjectFilter()}
+ return &clusterReadiness{
+ scaledObjects: kclient.NewDelayedInformer[controllers.Object](
+ client, scaledObjectGVR, kubetypes.DynamicInformer,
filter,
+ ),
+ gateways: kclient.NewDelayedInformer[*gatewayv1.Gateway](
+ client, gvr.KubernetesGateway,
kubetypes.StandardInformer, filter,
+ ),
+ gatewayClasses:
kclient.NewDelayedInformer[*gatewayv1.GatewayClass](
+ client, gvr.GatewayClass, kubetypes.StandardInformer,
filter,
+ ),
+ }
+}
+
+func (r *clusterReadiness) AddEventHandlers(
+ scaledObjectChanged func(namespace, name string),
+ gatewayChanged func(namespace string),
+ gatewayClassChanged func(name string),
+) {
+ r.scaledObjects.AddEventHandler(controllers.ObjectHandler(func(object
controllers.Object) {
+ scaledObjectChanged(object.GetNamespace(), object.GetName())
+ }))
+ r.gateways.AddEventHandler(controllers.ObjectHandler(func(object
controllers.Object) {
+ gatewayChanged(object.GetNamespace())
+ }))
+ r.gatewayClasses.AddEventHandler(controllers.ObjectHandler(func(object
controllers.Object) {
+ gatewayClassChanged(object.GetName())
+ }))
+}
+
+func (r *clusterReadiness) HasSynced() bool {
+ return r.scaledObjects.HasSynced() &&
+ r.gateways.HasSynced() &&
+ r.gatewayClasses.HasSynced()
+}
+
+func (r *clusterReadiness) ShutdownHandlers() {
+ controllers.ShutdownAll(r.scaledObjects, r.gateways, r.gatewayClasses)
+}
+
+func (r *clusterReadiness) ScalerReady(policy
*clientnetworking.ServiceActivationPolicy) bool {
+ ref := policy.Spec.GetAutoscalerRef()
+ if ref == nil ||
+ !strings.EqualFold(ref.GetGroup(), scaledObjectGVR.Group) ||
+ !strings.EqualFold(ref.GetKind(), "ScaledObject") {
+ return false
+ }
+ object := r.scaledObjects.Get(ref.GetName(), policy.GetNamespace())
+ return scaledObjectReady(object)
+}
+
+func scaledObjectReady(object controllers.Object) bool {
+ scaledObject, ok := object.(*unstructured.Unstructured)
+ if !ok || scaledObject == nil {
+ return false
+ }
+ conditions, found, err := unstructured.NestedSlice(scaledObject.Object,
"status", "conditions")
+ if err != nil || !found {
+ return false
+ }
+ for _, item := range conditions {
+ condition, ok := item.(map[string]any)
+ if ok && condition["type"] == "Ready" && condition["status"] ==
conditionTrue {
+ return true
+ }
+ }
+ return false
+}
+
+func (r *clusterReadiness) ActivatorReady(policy
*clientnetworking.ServiceActivationPolicy) bool {
+ for _, gateway := range r.gateways.List(policy.GetNamespace(),
klabels.Everything()) {
+ class :=
r.gatewayClasses.Get(string(gateway.Spec.GatewayClassName), "")
+ if class == nil || string(class.Spec.ControllerName) !=
features.ManagedGatewayController {
+ continue
+ }
+ if gatewayProgrammed(gateway) {
+ return true
+ }
+ }
+ return false
+}
+
+func gatewayProgrammed(gateway *gatewayv1.Gateway) bool {
+ for _, condition := range gateway.Status.Conditions {
+ if condition.Type ==
string(gatewayv1.GatewayConditionProgrammed) &&
+ condition.Status == metav1.ConditionTrue &&
+ condition.ObservedGeneration == gateway.Generation {
+ return true
+ }
+ }
+ return false
+}
diff --git a/dubbod/discovery/pkg/activation/cluster_readiness_test.go
b/dubbod/discovery/pkg/activation/cluster_readiness_test.go
new file mode 100644
index 00000000..c72982b7
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/cluster_readiness_test.go
@@ -0,0 +1,62 @@
+// 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 activation
+
+import (
+ "testing"
+
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+ gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
+)
+
+func TestScaledObjectReady(t *testing.T) {
+ object := &unstructured.Unstructured{Object: map[string]any{
+ "status": map[string]any{
+ "conditions": []any{
+ map[string]any{"type": "Ready", "status":
"True"},
+ },
+ },
+ }}
+ if !scaledObjectReady(object) {
+ t.Fatal("ScaledObject Ready=True was not recognized")
+ }
+
+ object.Object["status"] = map[string]any{
+ "conditions": []any{map[string]any{"type": "Ready", "status":
"False"}},
+ }
+ if scaledObjectReady(object) {
+ t.Fatal("ScaledObject Ready=False was reported ready")
+ }
+}
+
+func TestGatewayProgrammedRequiresCurrentGeneration(t *testing.T) {
+ gateway := &gatewayv1.Gateway{
+ ObjectMeta: metav1.ObjectMeta{Generation: 4},
+ Status: gatewayv1.GatewayStatus{Conditions: []metav1.Condition{{
+ Type:
string(gatewayv1.GatewayConditionProgrammed),
+ Status: metav1.ConditionTrue,
+ ObservedGeneration: 3,
+ }}},
+ }
+ if gatewayProgrammed(gateway) {
+ t.Fatal("stale Programmed=True condition was reported ready")
+ }
+ gateway.Status.Conditions[0].ObservedGeneration = 4
+ if !gatewayProgrammed(gateway) {
+ t.Fatal("current Programmed=True condition was not recognized")
+ }
+}
diff --git a/dubbod/discovery/pkg/activation/controller.go
b/dubbod/discovery/pkg/activation/controller.go
index 1a4bb324..6eb5c364 100644
--- a/dubbod/discovery/pkg/activation/controller.go
+++ b/dubbod/discovery/pkg/activation/controller.go
@@ -16,6 +16,7 @@
package activation
import (
+ "strings"
"time"
"github.com/apache/dubbo-kubernetes/pkg/kube"
@@ -32,9 +33,8 @@ import (
var logger = log.RegisterScope("activation", "Service activation policies")
// resyncInterval re-evaluates every policy periodically. Two of the four
-// conditions are derived from live state that produces no Kubernetes event at
-// all — a KEDA stream opening, a gateway starting or stopping reports — so
-// without a tick they would stay stale until the policy itself changed.
+// conditions are also re-evaluated periodically as a safety net for missed
+// informer edges.
const resyncInterval = 30 * time.Second
// Controller keeps ServiceActivationPolicy status in step with what the mesh
@@ -49,25 +49,43 @@ type Controller struct {
queue controllers.Queue
evaluator PolicyEvaluator
+ readiness *clusterReadiness
}
-// NewController wires the policy watch to the live scaler and gateway state.
-func NewController(client kube.Client, streams StreamLookup, reporters
ReporterLookup) *Controller {
+// NewController wires policy status to cluster-visible autoscaler and Gateway
+// state, so every HA replica evaluates the same facts.
+func NewController(client kube.Client) *Controller {
+ readiness := newClusterReadiness(client)
c := &Controller{
- policies:
kclient.New[*clientnetworking.ServiceActivationPolicy](client),
- services: kclient.New[*corev1.Service](client),
+ policies:
kclient.New[*clientnetworking.ServiceActivationPolicy](client),
+ services: kclient.New[*corev1.Service](client),
+ readiness: readiness,
}
c.evaluator = PolicyEvaluator{
Services: c,
- Streams: streams,
- Reporters: reporters,
+ Scaler: readiness,
+ Activator: readiness,
}
c.queue = controllers.NewQueue("service activation policy",
controllers.WithReconciler(c.Reconcile),
controllers.WithMaxAttempts(5))
- c.policies.AddEventHandler(controllers.ObjectHandler(c.queue.AddObject))
+
c.policies.AddEventHandler(controllers.EventHandler[*clientnetworking.ServiceActivationPolicy]{
+ AddFunc: func(policy *clientnetworking.ServiceActivationPolicy)
{
+ c.queue.AddObject(policy)
+ },
+ UpdateFunc: func(oldPolicy, newPolicy
*clientnetworking.ServiceActivationPolicy) {
+ // Do not feed our status writes straight back into the
queue.
+ // ScaledObject and Gateway informers drive runtime
convergence.
+ if oldPolicy.GetGeneration() !=
newPolicy.GetGeneration() {
+ c.queue.AddObject(newPolicy)
+ }
+ },
+ DeleteFunc: func(policy
*clientnetworking.ServiceActivationPolicy) {
+ c.queue.AddObject(policy)
+ },
+ })
// A policy is only accepted once its target Service exists, so a
Service
// appearing later has to re-open the policies that were rejected for
it.
c.services.AddEventHandler(controllers.ObjectHandler(func(o
controllers.Object) {
@@ -75,6 +93,37 @@ func NewController(client kube.Client, streams StreamLookup,
reporters ReporterL
c.queue.AddObject(policy)
}
}))
+ readiness.AddEventHandlers(
+ func(namespace, name string) {
+ for _, policy := range c.policies.List(namespace,
klabels.Everything()) {
+ ref := policy.Spec.GetAutoscalerRef()
+ if ref != nil &&
+ ref.GetName() == name &&
+ strings.EqualFold(ref.GetGroup(),
scaledObjectGVR.Group) &&
+ strings.EqualFold(ref.GetKind(),
"ScaledObject") {
+ c.queue.AddObject(policy)
+ }
+ }
+ },
+ func(namespace string) {
+ for _, policy := range c.policies.List(namespace,
klabels.Everything()) {
+ c.queue.AddObject(policy)
+ }
+ },
+ func(className string) {
+ namespaces := map[string]struct{}{}
+ for _, gateway := range
readiness.gateways.List(metav1.NamespaceAll, klabels.Everything()) {
+ if string(gateway.Spec.GatewayClassName) ==
className {
+ namespaces[gateway.GetNamespace()] =
struct{}{}
+ }
+ }
+ for namespace := range namespaces {
+ for _, policy := range
c.policies.List(namespace, klabels.Everything()) {
+ c.queue.AddObject(policy)
+ }
+ }
+ },
+ )
return c
}
@@ -85,12 +134,19 @@ func (c *Controller) HasService(namespace, name string)
bool {
}
func (c *Controller) Run(stop <-chan struct{}) {
- kube.WaitForCacheSync("activation controller", stop,
c.policies.HasSynced, c.services.HasSynced)
+ kube.WaitForCacheSync(
+ "activation controller",
+ stop,
+ c.policies.HasSynced,
+ c.services.HasSynced,
+ c.readiness.HasSynced,
+ )
go c.resync(stop)
c.queue.Run(stop)
controllers.ShutdownAll(c.policies, c.services)
+ c.readiness.ShutdownHandlers()
}
// resync re-queues every policy on a tick, picking up scaler and gateway
diff --git a/dubbod/discovery/pkg/activation/policy.go
b/dubbod/discovery/pkg/activation/policy.go
index 5c2483c1..162074dc 100644
--- a/dubbod/discovery/pkg/activation/policy.go
+++ b/dubbod/discovery/pkg/activation/policy.go
@@ -37,12 +37,12 @@ const (
// and replayed at all.
ConditionEligible = "Eligible"
- // ConditionScalerReady covers KEDA: whether it is listening for
activation
- // on this target.
+ // ConditionScalerReady covers KEDA: whether the referenced
ScaledObject is
+ // ready to obtain activation metrics.
ConditionScalerReady = "ScalerReady"
- // ConditionActivatorReady covers the gateways: whether any of them is
in a
- // position to catch a request for this target.
+ // ConditionActivatorReady covers the data plane: whether a managed
Gateway
+ // in the target namespace is programmed to catch requests.
ConditionActivatorReady = "ActivatorReady"
)
@@ -57,23 +57,23 @@ type ServiceLookup interface {
HasService(namespace, name string) bool
}
-// StreamLookup reports whether KEDA holds an activation stream for a target.
-type StreamLookup interface {
- Subscribed(Target) bool
+// ScalerStatusLookup reports whether the policy's autoscaler is ready.
+type ScalerStatusLookup interface {
+ ScalerReady(*clientnetworking.ServiceActivationPolicy) bool
}
-// ReporterLookup reports how many gateways stand ready to hold requests for a
-// target.
-type ReporterLookup interface {
- Reporters(Target) int
+// ActivatorStatusLookup reports whether a managed gateway is programmed for
+// the policy's namespace.
+type ActivatorStatusLookup interface {
+ ActivatorReady(*clientnetworking.ServiceActivationPolicy) bool
}
// PolicyEvaluator turns a policy plus live state into the conditions published
// on its status.
type PolicyEvaluator struct {
Services ServiceLookup
- Streams StreamLookup
- Reporters ReporterLookup
+ Scaler ScalerStatusLookup
+ Activator ActivatorStatusLookup
}
// Evaluate returns the conditions for one policy, in a stable order so an
@@ -99,15 +99,14 @@ func (e PolicyEvaluator) Evaluate(policy
*clientnetworking.ServiceActivationPoli
return conditions
}
- target := targetOf(policy)
eligible, eligibleReason := eligible(spec)
conditions = append(conditions, condition(ConditionEligible, eligible,
eligibleReason, generation))
- scalerReady := e.Streams != nil && e.Streams.Subscribed(target)
+ scalerReady := e.Scaler != nil && e.Scaler.ScalerReady(policy)
conditions = append(conditions,
condition(ConditionScalerReady, scalerReady,
scalerReason(scalerReady), generation))
- activatorReady := e.Reporters != nil && e.Reporters.Reporters(target) > 0
+ activatorReady := e.Activator != nil &&
e.Activator.ActivatorReady(policy)
conditions = append(conditions,
condition(ConditionActivatorReady, activatorReady,
activatorReason(activatorReady), generation))
@@ -154,18 +153,16 @@ func eligible(spec *networking.ServiceActivationPolicy)
(bool, string) {
func scalerReason(ready bool) string {
if ready {
- return "ScalerSubscribed"
+ return "ScaledObjectReady"
}
- // The usual cause is a ScaledObject that does not point its external
- // trigger at this scaler, or points it at a different Service.
- return "ScalerNotSubscribed"
+ return "ScaledObjectNotReady"
}
func activatorReason(ready bool) string {
if ready {
- return "GatewayReporting"
+ return "GatewayProgrammed"
}
- return "NoGatewayReporting"
+ return "NoProgrammedGateway"
}
// targetOf resolves the Service a policy activates. The namespace comes from
diff --git a/dubbod/discovery/pkg/activation/policy_test.go
b/dubbod/discovery/pkg/activation/policy_test.go
index ab8ede0d..77d118bf 100644
--- a/dubbod/discovery/pkg/activation/policy_test.go
+++ b/dubbod/discovery/pkg/activation/policy_test.go
@@ -29,13 +29,15 @@ type services map[string]bool
func (s services) HasService(namespace, name string) bool { return
s[namespace+"/"+name] }
-type subscribed map[Target]bool
+type scalerStatus bool
-func (s subscribed) Subscribed(target Target) bool { return s[target] }
+func (s scalerStatus) ScalerReady(*clientnetworking.ServiceActivationPolicy)
bool { return bool(s) }
-type reporters map[Target]int
+type activatorStatus bool
-func (r reporters) Reporters(target Target) int { return r[target] }
+func (a activatorStatus)
ActivatorReady(*clientnetworking.ServiceActivationPolicy) bool {
+ return bool(a)
+}
// policy builds the kube wrapper from its parts. The generated spec is a proto
// message with an embedded mutex, so it is assembled in place rather than
@@ -83,8 +85,8 @@ func conditionsByType(t *testing.T, evaluator
PolicyEvaluator, p *clientnetworki
func TestEvaluateReportsEveryConditionTrueWhenThePathIsComplete(t *testing.T) {
evaluator := PolicyEvaluator{
Services: services{"app/orders": true},
- Streams: subscribed{orders: true},
- Reporters: reporters{orders: 2},
+ Scaler: scalerStatus(true),
+ Activator: activatorStatus(true),
}
got := conditionsByType(t, evaluator, validPolicy())
@@ -164,8 +166,8 @@ func TestEvaluateReportsMissingTargetService(t *testing.T) {
func TestEvaluateRejectsProtocolsThatCannotBeHeld(t *testing.T) {
evaluator := PolicyEvaluator{
Services: services{"app/orders": true},
- Streams: subscribed{orders: true},
- Reporters: reporters{orders: 1},
+ Scaler: scalerStatus(true),
+ Activator: activatorStatus(true),
}
unknown := policy(serviceTarget("orders"), autoscaler("orders"),
networking.ActivationProtocol(99))
@@ -198,27 +200,27 @@ func TestEvaluateAcceptsEveryActivatableProtocol(t
*testing.T) {
func TestEvaluateReportsScalerAndActivatorSeparately(t *testing.T) {
evaluator := PolicyEvaluator{
Services: services{"app/orders": true},
- Streams: subscribed{},
- Reporters: reporters{orders: 1},
+ Scaler: scalerStatus(false),
+ Activator: activatorStatus(true),
}
got := conditionsByType(t, evaluator, validPolicy())
- if want := "False/ScalerNotSubscribed"; got[ConditionScalerReady] !=
want {
+ if want := "False/ScaledObjectNotReady"; got[ConditionScalerReady] !=
want {
t.Fatalf("ScalerReady = %q, want %q",
got[ConditionScalerReady], want)
}
- if want := "True/GatewayReporting"; got[ConditionActivatorReady] !=
want {
+ if want := "True/GatewayProgrammed"; got[ConditionActivatorReady] !=
want {
t.Fatalf("ActivatorReady = %q, want %q",
got[ConditionActivatorReady], want)
}
evaluator = PolicyEvaluator{
Services: services{"app/orders": true},
- Streams: subscribed{orders: true},
- Reporters: reporters{},
+ Scaler: scalerStatus(true),
+ Activator: activatorStatus(false),
}
got = conditionsByType(t, evaluator, validPolicy())
- if want := "True/ScalerSubscribed"; got[ConditionScalerReady] != want {
+ if want := "True/ScaledObjectReady"; got[ConditionScalerReady] != want {
t.Fatalf("ScalerReady = %q, want %q",
got[ConditionScalerReady], want)
}
- if want := "False/NoGatewayReporting"; got[ConditionActivatorReady] !=
want {
+ if want := "False/NoProgrammedGateway"; got[ConditionActivatorReady] !=
want {
t.Fatalf("ActivatorReady = %q, want %q",
got[ConditionActivatorReady], want)
}
}
@@ -228,8 +230,8 @@ func TestEvaluateReportsScalerAndActivatorSeparately(t
*testing.T) {
func TestEvaluateIsStableAcrossCalls(t *testing.T) {
evaluator := PolicyEvaluator{
Services: services{"app/orders": true},
- Streams: subscribed{orders: true},
- Reporters: reporters{orders: 1},
+ Scaler: scalerStatus(true),
+ Activator: activatorStatus(true),
}
target := validPolicy()
@@ -240,8 +242,8 @@ func TestEvaluateIsStableAcrossCalls(t *testing.T) {
// A real change must still be detected.
changed := PolicyEvaluator{
Services: services{"app/orders": true},
- Streams: subscribed{},
- Reporters: reporters{orders: 1},
+ Scaler: scalerStatus(false),
+ Activator: activatorStatus(true),
}
if SameConditions(evaluator.Evaluate(target), changed.Evaluate(target))
{
t.Fatal("SameConditions() reported no change after the scaler
unsubscribed")
diff --git a/dubbod/discovery/pkg/bootstrap/activation.go
b/dubbod/discovery/pkg/bootstrap/activation.go
index 08bd9b25..bd341793 100644
--- a/dubbod/discovery/pkg/bootstrap/activation.go
+++ b/dubbod/discovery/pkg/bootstrap/activation.go
@@ -39,7 +39,7 @@ func (s *Server) initActivation(args *DubboArgs) error {
return nil
}
- controller := activation.NewController(s.kubeClient,
s.activation.Scaler(), s.activation.Registry())
+ controller := activation.NewController(s.kubeClient)
s.addStartFunc("activation policy controller", func(stop <-chan
struct{}) error {
go controller.Run(stop)
return nil
diff --git a/dubbod/discovery/pkg/bootstrap/server.go
b/dubbod/discovery/pkg/bootstrap/server.go
index 97cb738d..bcb49a48 100644
--- a/dubbod/discovery/pkg/bootstrap/server.go
+++ b/dubbod/discovery/pkg/bootstrap/server.go
@@ -535,6 +535,13 @@ func (s *Server) initRegistryEventHandlers() {
Namespace: cfg.Namespace,
}
+ if configKind == kind.ServiceActivationPolicy &&
+ event == model.EventUpdate &&
+ cfg.Generation == prev.Generation {
+ log.Debugf("ignoring status-only update for %s/%s/%s",
configKey.Kind, configKey.Namespace, configKey.Name)
+ return
+ }
+
// Log the config change
log.Infof("%s event for %s/%s/%s", event, configKey.Kind,
configKey.Namespace, configKey.Name)
@@ -553,11 +560,22 @@ func (s *Server) initRegistryEventHandlers() {
configKind == kind.ServiceActivationPolicy
// Trigger ConfigUpdate to push changes to all connected proxies
- s.XDSServer.ConfigUpdate(&model.PushRequest{
+ pushRequest := &model.PushRequest{
ConfigsUpdated: sets.New(configKey),
Reason:
model.NewReasonStats(model.DependentResource),
Full: needsFullPush,
- })
+ }
+ if configKind == kind.ServiceActivationPolicy {
+ pushRequest.Forced = true
+ pushRequest.ServiceActivationPolicyUpdates =
map[model.ConfigKey]model.ServiceActivationPolicyUpdate{
+ configKey: {
+ Config: cfg,
+ Previous: prev,
+ Deleted: event == model.EventDelete,
+ },
+ }
+ }
+ s.XDSServer.ConfigUpdate(pushRequest)
}
schemas := collections.Dubbo.All()
if features.EnableGatewayAPI {
diff --git a/dubbod/discovery/pkg/config/kube/gateway/deployment_controller.go
b/dubbod/discovery/pkg/config/kube/gateway/deployment_controller.go
index 60769004..727adfda 100644
--- a/dubbod/discovery/pkg/config/kube/gateway/deployment_controller.go
+++ b/dubbod/discovery/pkg/config/kube/gateway/deployment_controller.go
@@ -47,6 +47,7 @@ import (
"github.com/apache/dubbo-kubernetes/pkg/config/constants"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/gvr"
telemetryconfig
"github.com/apache/dubbo-kubernetes/pkg/config/telemetry"
+ "github.com/apache/dubbo-kubernetes/pkg/grpcxds"
"github.com/apache/dubbo-kubernetes/pkg/kube"
"github.com/apache/dubbo-kubernetes/pkg/kube/controllers"
"github.com/apache/dubbo-kubernetes/pkg/kube/inject"
@@ -457,9 +458,29 @@ func (d *DeploymentController) configureGateway(log
*dubbolog.Logger, gw gateway
}
}
- if err := d.cleanupLegacyGatewayResources(context.TODO(), log,
gw.Namespace, legacyName, defaultName); err != nil {
+ if err := d.cleanupLegacyGatewayResources(
+ context.TODO(),
+ log,
+ gw.Namespace,
+ legacyName,
+ defaultName,
+ gw.Name,
+ ); err != nil {
log.Warnf("failed cleaning up legacy dxgate resources %s/%s:
%v", gw.Namespace, legacyName, err)
}
+ if defaultName != defaultDxgateGatewayName {
+ if err := d.cleanupLegacyGatewayResources(
+ context.TODO(),
+ log,
+ gw.Namespace,
+ defaultDxgateGatewayName,
+ defaultName,
+ gw.Name,
+ ); err != nil {
+ log.Warnf("failed cleaning up fixed-name dxgate
resources %s/%s: %v",
+ gw.Namespace, defaultDxgateGatewayName, err)
+ }
+ }
log.Infof("gateway updated successfully")
return nil
@@ -718,12 +739,13 @@ func buildDxgateBootstrapConfig(xdsAddress string,
listenerNames []string, clust
}
func dxgateListenerNames(namespace, serviceName, domainSuffix string, ports
[]corev1.ServicePort) []string {
- if domainSuffix == "" {
- domainSuffix = constants.DefaultClusterLocalDomain
- }
out := make([]string, 0, len(ports))
for _, port := range ports {
- out = append(out, fmt.Sprintf("%s.%s.svc.%s:%d", serviceName,
namespace, domainSuffix, port.Port))
+ targetPort := port.TargetPort.IntValue()
+ if targetPort <= 0 {
+ targetPort = int(port.Port)
+ }
+ out = append(out, fmt.Sprintf("%s0.0.0.0:%d",
grpcxds.ServerListenerNamePrefix, targetPort))
}
sort.Strings(out)
return out
@@ -1258,7 +1280,14 @@ func (d *DeploymentController) canManage(gvr
schema.GroupVersionResource, name,
return managed, obj.GetResourceVersion()
}
-func (d *DeploymentController) cleanupLegacyGatewayResources(ctx
context.Context, log *dubbolog.Logger, namespace, legacyName, currentName
string) error {
+func (d *DeploymentController) cleanupLegacyGatewayResources(
+ ctx context.Context,
+ log *dubbolog.Logger,
+ namespace,
+ legacyName,
+ currentName,
+ gatewayName string,
+) error {
if d.client == nil || legacyName == "" || legacyName == currentName {
return nil
}
@@ -1267,14 +1296,14 @@ func (d *DeploymentController)
cleanupLegacyGatewayResources(ctx context.Context
for _, cleanup := range []struct {
kind string
name string
- fn func(context.Context, string, string) error
+ fn func(context.Context, string, string, string) error
}{
{kind: "ConfigMap", name: legacyName + "-bootstrap", fn:
d.deleteManagedConfigMap},
{kind: "ServiceAccount", name: legacyName, fn:
d.deleteManagedServiceAccount},
{kind: "Deployment", name: legacyName, fn:
d.deleteManagedDeployment},
{kind: "Service", name: legacyName, fn: d.deleteManagedService},
} {
- if err := cleanup.fn(ctx, namespace, cleanup.name); err != nil {
+ if err := cleanup.fn(ctx, namespace, cleanup.name,
gatewayName); err != nil {
errs = append(errs, fmt.Sprintf("%s/%s: %v",
cleanup.kind, cleanup.name, err))
}
}
@@ -1286,7 +1315,7 @@ func (d *DeploymentController)
cleanupLegacyGatewayResources(ctx context.Context
return nil
}
-func (d *DeploymentController) deleteManagedConfigMap(ctx context.Context,
namespace, name string) error {
+func (d *DeploymentController) deleteManagedConfigMap(ctx context.Context,
namespace, name, gatewayName string) error {
obj, err := d.client.Kube().CoreV1().ConfigMaps(namespace).Get(ctx,
name, metav1.GetOptions{})
if apierrors.IsNotFound(err) {
return nil
@@ -1294,7 +1323,7 @@ func (d *DeploymentController) deleteManagedConfigMap(ctx
context.Context, names
if err != nil {
return err
}
- if !isManagedGatewayResource(obj.Labels) {
+ if !isManagedGatewayResourceFor(obj.Labels, gatewayName) {
return nil
}
err = d.client.Kube().CoreV1().ConfigMaps(namespace).Delete(ctx, name,
metav1.DeleteOptions{})
@@ -1304,7 +1333,7 @@ func (d *DeploymentController) deleteManagedConfigMap(ctx
context.Context, names
return err
}
-func (d *DeploymentController) deleteManagedServiceAccount(ctx
context.Context, namespace, name string) error {
+func (d *DeploymentController) deleteManagedServiceAccount(ctx
context.Context, namespace, name, gatewayName string) error {
obj, err :=
d.client.Kube().CoreV1().ServiceAccounts(namespace).Get(ctx, name,
metav1.GetOptions{})
if apierrors.IsNotFound(err) {
return nil
@@ -1312,7 +1341,7 @@ func (d *DeploymentController)
deleteManagedServiceAccount(ctx context.Context,
if err != nil {
return err
}
- if !isManagedGatewayResource(obj.Labels) {
+ if !isManagedGatewayResourceFor(obj.Labels, gatewayName) {
return nil
}
err = d.client.Kube().CoreV1().ServiceAccounts(namespace).Delete(ctx,
name, metav1.DeleteOptions{})
@@ -1322,7 +1351,7 @@ func (d *DeploymentController)
deleteManagedServiceAccount(ctx context.Context,
return err
}
-func (d *DeploymentController) deleteManagedDeployment(ctx context.Context,
namespace, name string) error {
+func (d *DeploymentController) deleteManagedDeployment(ctx context.Context,
namespace, name, gatewayName string) error {
obj, err := d.client.Kube().AppsV1().Deployments(namespace).Get(ctx,
name, metav1.GetOptions{})
if apierrors.IsNotFound(err) {
return nil
@@ -1330,7 +1359,7 @@ func (d *DeploymentController)
deleteManagedDeployment(ctx context.Context, name
if err != nil {
return err
}
- if !isManagedGatewayResource(obj.Labels) {
+ if !isManagedGatewayResourceFor(obj.Labels, gatewayName) {
return nil
}
err = d.client.Kube().AppsV1().Deployments(namespace).Delete(ctx, name,
metav1.DeleteOptions{})
@@ -1340,7 +1369,7 @@ func (d *DeploymentController)
deleteManagedDeployment(ctx context.Context, name
return err
}
-func (d *DeploymentController) deleteManagedService(ctx context.Context,
namespace, name string) error {
+func (d *DeploymentController) deleteManagedService(ctx context.Context,
namespace, name, gatewayName string) error {
obj, err := d.client.Kube().CoreV1().Services(namespace).Get(ctx, name,
metav1.GetOptions{})
if apierrors.IsNotFound(err) {
return nil
@@ -1348,7 +1377,7 @@ func (d *DeploymentController) deleteManagedService(ctx
context.Context, namespa
if err != nil {
return err
}
- if !isManagedGatewayResource(obj.Labels) {
+ if !isManagedGatewayResourceFor(obj.Labels, gatewayName) {
return nil
}
err = d.client.Kube().CoreV1().Services(namespace).Delete(ctx, name,
metav1.DeleteOptions{})
@@ -1363,6 +1392,11 @@ func isManagedGatewayResource(labels map[string]string)
bool {
return ok
}
+func isManagedGatewayResourceFor(labels map[string]string, gatewayName string)
bool {
+ return isManagedGatewayResource(labels) &&
+ labels["gateway.networking.k8s.io/gateway-name"] == gatewayName
+}
+
func (d *DeploymentController) HandleTagChange(newTags any) {
for _, gw := range d.gateways.List(metav1.NamespaceAll,
klabels.Everything()) {
d.queue.AddObject(gw)
@@ -1384,8 +1418,14 @@ func IsManaged(gw *gateway.GatewaySpec) bool {
return false
}
-func getDefaultName(_ string, _ *gateway.GatewaySpec, _ bool) string {
- return defaultDxgateGatewayName
+func getDefaultName(name string, kgw *gateway.GatewaySpec, disableNameSuffix
bool) string {
+ // Keep the canonical Activator Service stable: proxyless cold EDS
points at
+ // this namespace-local name. Every other Gateway needs its own
resources or
+ // two Gateway reconciles overwrite the same Deployment, Service and
config.
+ if name == defaultDxgateGatewayName {
+ return defaultDxgateGatewayName
+ }
+ return getLegacyDefaultName(name, kgw, disableNameSuffix)
}
func getLegacyDefaultName(name string, kgw *gateway.GatewaySpec,
disableNameSuffix bool) string {
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 b7b0da8b..826f4701 100644
--- a/dubbod/discovery/pkg/config/kube/gateway/deployment_controller_test.go
+++ b/dubbod/discovery/pkg/config/kube/gateway/deployment_controller_test.go
@@ -441,6 +441,12 @@ func
TestDeploymentControllerBuildDxgateBootstrapConfigUsesXDSAddressAnnotation(
if cfg.ClusterID != "remote" {
t.Fatalf("clusterID = %q, want remote", cfg.ClusterID)
}
+ if diff := cmp.Diff(
+ []string{"xds.dubbo.apache.org/grpc/lds/inbound/0.0.0.0:15080"},
+ cfg.ListenerNames,
+ ); diff != "" {
+ t.Fatalf("listener names (-want +got):\n%s", diff)
+ }
}
func TestExtractServicePortsTargetsGRPCInbound(t *testing.T) {
@@ -612,13 +618,29 @@ func
TestObservabilityConfigForGatewayInvalidAccessLogAnnotationsFallBack(t *tes
}
}
-func TestGetDefaultNameUsesFixedDxgateGatewayName(t *testing.T) {
+func TestGetDefaultNameKeepsCanonicalActivatorAndIsolatesOtherGateways(t
*testing.T) {
spec := &gatewayv1.GatewaySpec{GatewayClassName: "dubbo"}
- if got := getDefaultName("httpbin-gateway", spec, false); got !=
"dxgate-gateway" {
- t.Fatalf("default name = %q, want dxgate-gateway", got)
+ if got := getDefaultName("dxgate-gateway", spec, false); got !=
"dxgate-gateway" {
+ t.Fatalf("canonical name = %q, want dxgate-gateway", got)
+ }
+ if got := getDefaultName("public", spec, false); got != "public-dubbo" {
+ t.Fatalf("derived name = %q, want public-dubbo", got)
+ }
+ if got := getDefaultName("private", spec, true); got != "private" {
+ t.Fatalf("name with suffix disabled = %q, want private", got)
+ }
+}
+
+func TestManagedGatewayResourceCleanupRequiresMatchingGateway(t *testing.T) {
+ labels := map[string]string{
+ "gateway.dubbo.apache.org/managed":
"dubbo.apache.org-gateway-controller",
+ "gateway.networking.k8s.io/gateway-name": "public",
+ }
+ if !isManagedGatewayResourceFor(labels, "public") {
+ t.Fatal("matching managed resource was not recognized")
}
- if got := getDefaultName("foo-gateway", spec, true); got !=
"dxgate-gateway" {
- t.Fatalf("default name with suffix disabled = %q, want
dxgate-gateway", got)
+ if isManagedGatewayResourceFor(labels, "dxgate-gateway") {
+ t.Fatal("cleanup would delete another Gateway's resources")
}
}
diff --git a/dubbod/discovery/pkg/model/push_context.go
b/dubbod/discovery/pkg/model/push_context.go
index 075769e7..b9bf9a9c 100644
--- a/dubbod/discovery/pkg/model/push_context.go
+++ b/dubbod/discovery/pkg/model/push_context.go
@@ -87,14 +87,21 @@ type PushContext struct {
}
type PushRequest struct {
- Reason ReasonStats
- ConfigsUpdated sets.Set[ConfigKey]
- AddressesUpdated sets.Set[string]
- Forced bool
- Full bool
- Push *PushContext
- Start time.Time
- Delta ResourceDelta
+ Reason ReasonStats
+ ConfigsUpdated sets.Set[ConfigKey]
+ ServiceActivationPolicyUpdates
map[ConfigKey]ServiceActivationPolicyUpdate
+ AddressesUpdated sets.Set[string]
+ Forced bool
+ Full bool
+ Push *PushContext
+ Start time.Time
+ Delta ResourceDelta
+}
+
+type ServiceActivationPolicyUpdate struct {
+ Config config.Config
+ Previous config.Config
+ Deleted bool
}
type XDSUpdater interface {
@@ -310,6 +317,18 @@ func (pr *PushRequest) Merge(other *PushRequest)
*PushRequest {
} else {
pr.AddressesUpdated.Merge(other.AddressesUpdated)
}
+ if len(other.ServiceActivationPolicyUpdates) > 0 {
+ if pr.ServiceActivationPolicyUpdates == nil {
+ pr.ServiceActivationPolicyUpdates =
make(map[ConfigKey]ServiceActivationPolicyUpdate)
+ }
+ for key, update := range other.ServiceActivationPolicyUpdates {
+ if existing, found :=
pr.ServiceActivationPolicyUpdates[key]; found &&
+ existing.Previous.Spec != nil {
+ update.Previous = existing.Previous
+ }
+ pr.ServiceActivationPolicyUpdates[key] = update
+ }
+ }
pr.Delta = mergeResourceDelta(pr.Delta, other.Delta)
@@ -358,6 +377,12 @@ func (pr *PushRequest) Copy() *PushRequest {
if pr.AddressesUpdated != nil {
out.AddressesUpdated = pr.AddressesUpdated.Copy()
}
+ if pr.ServiceActivationPolicyUpdates != nil {
+ out.ServiceActivationPolicyUpdates =
make(map[ConfigKey]ServiceActivationPolicyUpdate,
len(pr.ServiceActivationPolicyUpdates))
+ for key, update := range pr.ServiceActivationPolicyUpdates {
+ out.ServiceActivationPolicyUpdates[key] = update
+ }
+ }
out.Delta = copyResourceDelta(pr.Delta)
return &out
}
@@ -432,6 +457,7 @@ func (ps *PushContext) InitContext(env *Environment,
oldPushContext *PushContext
if pushReq == nil || oldPushContext == nil ||
!oldPushContext.InitDone.Load() || pushReq.Forced {
ps.createNewContext(env)
+ ps.applyServiceActivationPolicyUpdates(pushReq)
} else {
ps.updateContext(env, oldPushContext, pushReq)
}
@@ -609,13 +635,14 @@ func (ps *PushContext) createNewContext(env *Environment)
{
}
func (ps *PushContext) updateContext(env *Environment, oldPushContext
*PushContext, pushReq *PushRequest) {
- // Check if services have changed based on:
- // 1. ServiceEntry updates in ConfigsUpdated
- // 2. Address changes
- // 3. Actual service count changes from environment (for Kubernetes
Service changes)
- // servicesChanged := pushReq != nil &&
(HasConfigsOfKind(pushReq.ConfigsUpdated, kind.ServiceEntry) ||
- // len(pushReq.AddressesUpdated) > 0)
- servicesChanged := pushReq != nil && len(pushReq.AddressesUpdated) > 0
+ // A Service may appear after a policy in the same informer burst.
Rebuild
+ // the registry on the explicit Service event even when the total count
was
+ // already visible through env.Services(); otherwise one HA replica can
keep
+ // an activation RDS snapshot that permanently omits the new backend.
+ servicesChanged := pushReq != nil &&
+ (len(pushReq.AddressesUpdated) > 0 ||
+ HasConfigsOfKind(pushReq.ConfigsUpdated, kind.Service)
||
+ HasConfigsOfKind(pushReq.ConfigsUpdated,
kind.ServiceEntry))
// Also check if the actual number of services has changed
// This handles cases where Kubernetes Services are added/removed
without ServiceEntry updates
@@ -677,13 +704,8 @@ func (ps *PushContext) updateContext(env *Environment,
oldPushContext *PushConte
ps.faultInjectionIndex = oldPushContext.faultInjectionIndex
}
- serviceActivationPoliciesChanged := pushReq != nil &&
HasConfigsOfKind(pushReq.ConfigsUpdated, kind.ServiceActivationPolicy)
- if serviceActivationPoliciesChanged {
- log.Debugf("ServiceActivationPolicies changed, re-initializing
activation index")
- ps.initServiceActivationPolicies(env)
- } else {
- ps.serviceActivationIndex =
oldPushContext.serviceActivationIndex
- }
+ ps.serviceActivationIndex =
copyServiceActivationPolicyIndex(oldPushContext.serviceActivationIndex)
+ ps.applyServiceActivationPolicyUpdates(pushReq)
authnPoliciesChanged := pushReq != nil && (pushReq.Full ||
authPolicyKindsChanged(pushReq.ConfigsUpdated))
if authnPoliciesChanged || oldPushContext == nil ||
oldPushContext.AuthenticationPolicies == nil {
@@ -1122,22 +1144,60 @@ func (ps *PushContext)
initServiceActivationPolicies(env *Environment) {
if env == nil {
return
}
- for _, cfg := range
sortConfigByCreationTime(env.List(gvk.ServiceActivationPolicy, NamespaceAll)) {
- spec, ok := cfg.Spec.(*networking.ServiceActivationPolicy)
- if !ok || spec == nil || spec.GetTargetRef() == nil ||
spec.GetAutoscalerRef() == nil {
- continue
+ configs :=
sortConfigByCreationTime(env.List(gvk.ServiceActivationPolicy, NamespaceAll))
+ for _, cfg := range configs {
+ ps.upsertServiceActivationPolicy(cfg)
+ }
+ log.Debugf("activation policy index rebuilt: configs=%d services=%d",
len(configs), len(ps.serviceActivationIndex.services))
+}
+
+func copyServiceActivationPolicyIndex(in serviceActivationPolicyIndex)
serviceActivationPolicyIndex {
+ out := serviceActivationPolicyIndex{services: make(map[string][]string,
len(in.services))}
+ for key, accounts := range in.services {
+ out.services[key] = append([]string(nil), accounts...)
+ }
+ return out
+}
+
+func (ps *PushContext) applyServiceActivationPolicyUpdates(pushReq
*PushRequest) {
+ if pushReq == nil {
+ return
+ }
+ for _, update := range pushReq.ServiceActivationPolicyUpdates {
+ if update.Previous.Spec != nil {
+ ps.deleteServiceActivationPolicy(update.Previous)
}
- target := spec.GetTargetRef()
- if strings.TrimSpace(target.GetName()) == "" ||
- (target.GetKind() != "" &&
!strings.EqualFold(target.GetKind(), "Service")) ||
- strings.TrimSpace(target.GetGroup()) != "" ||
- strings.TrimSpace(spec.GetAutoscalerRef().GetName()) ==
"" ||
- len(spec.GetBackendServiceAccounts()) == 0 {
+ if update.Deleted {
+ ps.deleteServiceActivationPolicy(update.Config)
continue
}
-
ps.serviceActivationIndex.services[backendTLSPolicyServiceKey(cfg.Namespace,
target.GetName())] =
- append([]string(nil),
spec.GetBackendServiceAccounts()...)
+ ps.upsertServiceActivationPolicy(update.Config)
+ }
+}
+
+func (ps *PushContext) deleteServiceActivationPolicy(cfg config.Config) {
+ spec, ok := cfg.Spec.(*networking.ServiceActivationPolicy)
+ if ok && spec != nil && spec.GetTargetRef() != nil {
+ delete(ps.serviceActivationIndex.services,
+ backendTLSPolicyServiceKey(cfg.Namespace,
spec.GetTargetRef().GetName()))
+ }
+}
+
+func (ps *PushContext) upsertServiceActivationPolicy(cfg config.Config) {
+ spec, ok := cfg.Spec.(*networking.ServiceActivationPolicy)
+ if !ok || spec == nil || spec.GetTargetRef() == nil ||
spec.GetAutoscalerRef() == nil {
+ return
+ }
+ target := spec.GetTargetRef()
+ if strings.TrimSpace(target.GetName()) == "" ||
+ (target.GetKind() != "" && !strings.EqualFold(target.GetKind(),
"Service")) ||
+ strings.TrimSpace(target.GetGroup()) != "" ||
+ strings.TrimSpace(spec.GetAutoscalerRef().GetName()) == "" ||
+ len(spec.GetBackendServiceAccounts()) == 0 {
+ return
}
+
ps.serviceActivationIndex.services[backendTLSPolicyServiceKey(cfg.Namespace,
target.GetName())] =
+ append([]string(nil), spec.GetBackendServiceAccounts()...)
}
func (ps *PushContext) initServiceAccounts(env *Environment, services
[]*Service) {
diff --git a/dubbod/discovery/pkg/model/push_context_test.go
b/dubbod/discovery/pkg/model/push_context_test.go
index 02f95714..d22cdc56 100644
--- a/dubbod/discovery/pkg/model/push_context_test.go
+++ b/dubbod/discovery/pkg/model/push_context_test.go
@@ -20,8 +20,11 @@ import (
"testing"
"time"
+ "github.com/apache/dubbo-kubernetes/pkg/config"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/gvk"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/kind"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
+ networking "github.com/kdubbo/api/networking/v1alpha3"
)
func TestPushRequestCopyMergePreservesQueuedUpdates(t *testing.T) {
@@ -99,3 +102,49 @@ func TestPushContextStatusJSONEmptyObject(t *testing.T) {
t.Fatalf("StatusJSON() = %s, want {}", string(data))
}
}
+
+func TestApplyServiceActivationPolicyUpdates(t *testing.T) {
+ key := ConfigKey{Kind: kind.ServiceActivationPolicy, Name: "payment",
Namespace: "app"}
+ policy := config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.ServiceActivationPolicy,
+ Name: key.Name,
+ Namespace: key.Namespace,
+ },
+ Spec: &networking.ServiceActivationPolicy{
+ TargetRef: &networking.PolicyTargetReference{
+ Kind: "Service",
+ Name: "payment",
+ },
+ AutoscalerRef:
&networking.AutoscalerReference{Name: "payment"},
+ BackendServiceAccounts: []string{"payment"},
+ },
+ }
+ add := &PushRequest{
+ ServiceActivationPolicyUpdates:
map[ConfigKey]ServiceActivationPolicyUpdate{
+ key: {Config: policy},
+ },
+ }
+
+ push := NewPushContext()
+ push.applyServiceActivationPolicyUpdates(add)
+ if !push.ServiceActivationEnabled("app", "payment") {
+ t.Fatal("policy event did not add payment to activation index")
+ }
+
+ remove := &PushRequest{
+ ServiceActivationPolicyUpdates:
map[ConfigKey]ServiceActivationPolicyUpdate{
+ key: {Config: policy, Deleted: true},
+ },
+ }
+ merged := add.CopyMerge(remove)
+ if !merged.ServiceActivationPolicyUpdates[key].Deleted {
+ t.Fatal("later delete did not replace queued add")
+ }
+ next := NewPushContext()
+ next.serviceActivationIndex =
copyServiceActivationPolicyIndex(push.serviceActivationIndex)
+ next.applyServiceActivationPolicyUpdates(merged)
+ if next.ServiceActivationEnabled("app", "payment") {
+ t.Fatal("policy delete did not remove payment from activation
index")
+ }
+}
diff --git a/dubbod/discovery/pkg/networking/grpcgen/rds.go
b/dubbod/discovery/pkg/networking/grpcgen/rds.go
index b585fac2..7a7df64e 100644
--- a/dubbod/discovery/pkg/networking/grpcgen/rds.go
+++ b/dubbod/discovery/pkg/networking/grpcgen/rds.go
@@ -179,18 +179,24 @@ func buildHTTPRoute(node *model.Proxy, push
*model.PushContext, routeName string
// For regular service Pods, use NonForwardingAction to handle requests
directly
// Also check if this is a Gateway Pod by checking service name
(fallback for when node.Type is not Router)
isGatewayPod := false
+ isGatewayDataPort := parsedPort == 80
+ gatewayListenerPort := parsedPort
var gatewayName, gatewayNamespace string
- // Try to find Gateway Pod's service and extract Gateway name from
labels
+ // Resolve the listener's Service port from the generated Gateway
Service.
+ // dxgate binds the Service targetPort (15080 by default), while
Gateway API
+ // parentRefs and HTTPRoutes refer to the public listener port (for
example 80).
for _, st := range node.ServiceTargets {
- if
strings.Contains(strings.ToLower(st.Service.Attributes.Name), "gateway") {
+ if st.Service == nil {
+ continue
+ }
+ if name, ok :=
st.Service.Attributes.Labels["gateway.networking.k8s.io/gateway-name"]; ok {
isGatewayPod = true
- // Try to get Gateway name from service labels
- if len(st.Service.Attributes.Labels) > 0 {
- if name, ok :=
st.Service.Attributes.Labels["gateway.networking.k8s.io/gateway-name"]; ok {
- gatewayName = name
- gatewayNamespace =
st.Service.Attributes.Namespace
- break
- }
+ if st.Port.ServicePort != nil && st.Port.TargetPort ==
uint32(parsedPort) {
+ gatewayName = name
+ gatewayNamespace =
st.Service.Attributes.Namespace
+ gatewayListenerPort = st.Port.Port
+ isGatewayDataPort = true
+ break
}
}
}
@@ -200,10 +206,8 @@ func buildHTTPRoute(node *model.Proxy, push
*model.PushContext, routeName string
}
log.Infof("Gateway Pod inbound listener, routeName=%s, port=%d,
gateway=%s/%s", routeName, parsedPort, gatewayNamespace, gatewayName)
- // CRITICAL: Only apply HTTPRoute to the Gateway listener port
(80)
- // Other ports (26012, 26021, etc.) are service ports and
should not use HTTPRoute
- if parsedPort != 80 {
- log.Debugf("Gateway Pod inbound listener port %d is not
80, skipping HTTPRoute (this is a service port, not Gateway listener)",
parsedPort)
+ if !isGatewayDataPort {
+ log.Debugf("Gateway Pod inbound listener port %d is not
a Gateway Service targetPort, skipping HTTPRoute", parsedPort)
// Return empty route config for non-Gateway ports
return &route.RouteConfiguration{
Name: routeName,
@@ -236,8 +240,8 @@ func buildHTTPRoute(node *model.Proxy, push
*model.PushContext, routeName string
log.Debugf("Gateway Pod inbound listener, found %d HTTPRoute(s)
with wildcard match", len(allHTTPRoutes))
// Filter HTTPRoutes by parentRef to match this Gateway
- httpRoutes := filterHTTPRoutesByGateway(allHTTPRoutes,
gatewayName, gatewayNamespace, parsedPort)
- log.Debugf("Gateway Pod inbound listener, filtered to %d
HTTPRoute(s) matching gateway %s/%s port %d", len(httpRoutes),
gatewayNamespace, gatewayName, parsedPort)
+ httpRoutes := filterHTTPRoutesByGateway(allHTTPRoutes,
gatewayName, gatewayNamespace, gatewayListenerPort)
+ 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
// because Gateway Pods route traffic based on HTTPRoute
hostnames in the request
@@ -271,7 +275,7 @@ func buildHTTPRoute(node *model.Proxy, push
*model.PushContext, routeName string
}
}
- if routes :=
buildRoutesFromGatewayHTTPRoute(httpRoutes, host.Name("*"), parsedPort, nil);
len(routes) > 0 {
+ if routes :=
buildRoutesFromGatewayHTTPRoute(httpRoutes, host.Name("*"),
gatewayListenerPort, nil); len(routes) > 0 {
log.Infof("Gateway Pod inbound listener built
%d routes from HTTPRoute", len(routes))
outboundRoutes = routes
} else {
diff --git a/dubbod/discovery/pkg/networking/grpcgen/rds_test.go
b/dubbod/discovery/pkg/networking/grpcgen/rds_test.go
index e4075a05..d2b035c9 100644
--- a/dubbod/discovery/pkg/networking/grpcgen/rds_test.go
+++ b/dubbod/discovery/pkg/networking/grpcgen/rds_test.go
@@ -296,6 +296,45 @@ func
TestGatewayRDSRoutesActivationAuthorityToOriginalCluster(t *testing.T) {
}
}
+func TestGatewayInboundTargetPortIncludesActivationRoutes(t *testing.T) {
+ target := newRDSTestService("payment", "app",
"payment.app.svc.cluster.local", 8080)
+ activator := newRDSTestService(
+ model.ActivationGatewayServiceName,
+ "app",
+ "dxgate-gateway.app.svc.cluster.local",
+ 80,
+ )
+ activator.Attributes.Labels = map[string]string{
+ "gateway.networking.k8s.io/gateway-name": "dxgate-gateway",
+ }
+ push := newRDSTestPushContext(t, []config.Config{
+ newActivationPolicyConfig("payment", "app", "payment"),
+ }, []*model.Service{target, activator})
+ proxy := &model.Proxy{
+ ID: "dxgate-gateway.app",
+ Type: model.Router,
+ ConfigNamespace: "app",
+ ServiceTargets: []model.ServiceTarget{{
+ Service: activator,
+ Port: model.ServiceInstancePort{
+ ServicePort: activator.Ports[0],
+ TargetPort: 15080,
+ },
+ }},
+ }
+
+ rc := buildHTTPRoute(proxy, push, "15080")
+ if rc == nil {
+ t.Fatal("buildHTTPRoute() returned nil")
+ }
+ for _, virtualHost := range rc.GetVirtualHosts() {
+ if virtualHost.GetName() ==
"activation|payment.app.svc.cluster.local|8080" {
+ return
+ }
+ }
+ t.Fatalf("activation virtual host not found on Gateway targetPort: %v",
rc.GetVirtualHosts())
+}
+
func newRDSTestPushContext(t *testing.T, configs []config.Config, services
[]*model.Service) *model.PushContext {
t.Helper()
diff --git a/dubbod/discovery/pkg/xds/delta_test.go
b/dubbod/discovery/pkg/xds/delta_test.go
index 992060a0..84a6f4e6 100644
--- a/dubbod/discovery/pkg/xds/delta_test.go
+++ b/dubbod/discovery/pkg/xds/delta_test.go
@@ -80,6 +80,12 @@ func TestPushDeltaXdsAppliesRemovedResourcesToWatchedState(t
*testing.T) {
assertWatchedNames(t, con.proxy.GetWatchedResource(v1.ClusterType),
keptName, addedName)
}
+func TestShouldSetWatchedResourcesTracksRoutes(t *testing.T) {
+ if !shouldSetWatchedResources(&model.WatchedResource{TypeUrl:
v1.RouteType}) {
+ t.Fatal("Delta RDS responses must update watched resource
names")
+ }
+}
+
func newDeltaXDSTestServer(generator model.XdsResourceGenerator)
(*DiscoveryServer, *Connection, *fakeDeltaADSStream) {
push := model.NewPushContext()
push.PushVersion = "test-version"
diff --git a/dubbod/discovery/pkg/xds/xdsgen.go
b/dubbod/discovery/pkg/xds/xdsgen.go
index c10a14e4..d1c2c6e5 100644
--- a/dubbod/discovery/pkg/xds/xdsgen.go
+++ b/dubbod/discovery/pkg/xds/xdsgen.go
@@ -657,7 +657,10 @@ func shouldSetWatchedResources(w *model.WatchedResource)
bool {
if w == nil {
return false
}
- return xds.IsWildcardTypeURL(w.TypeUrl)
+ // RDS is not globally wildcard in xDS, but dxgate starts with a
wildcard
+ // Delta subscription and then ACKs without repeating concrete names.
Keep
+ // the names we actually sent so later policy pushes can rebuild those
routes.
+ return xds.IsWildcardTypeURL(w.TypeUrl) || w.TypeUrl == v1.RouteType
}
// extractRouteNamesFromLDS extracts route names referenced in LDS listener
resources
diff --git a/manifests/charts/dubbod/templates/clusterrole.yaml
b/manifests/charts/dubbod/templates/clusterrole.yaml
index 412f6c5c..751e0f2b 100644
--- a/manifests/charts/dubbod/templates/clusterrole.yaml
+++ b/manifests/charts/dubbod/templates/clusterrole.yaml
@@ -72,6 +72,12 @@ rules:
- apiGroups: ["coordination.k8s.io"]
resources: ["leases"]
verbs: ["get", "list", "watch", "create", "update", "patch"]
+ # ServiceActivationPolicy status follows the referenced KEDA object. This is
+ # read-only and optional: the delayed informer remains dormant when KEDA is
+ # not installed.
+ - apiGroups: ["keda.sh"]
+ resources: ["scaledobjects"]
+ verbs: ["get", "list", "watch"]
- apiGroups: ["gateway.networking.k8s.io"]
resources:
- gateways
diff --git a/manifests/charts/dubbod/templates/deployment.yaml
b/manifests/charts/dubbod/templates/deployment.yaml
index 88dcbe02..09e38679 100644
--- a/manifests/charts/dubbod/templates/deployment.yaml
+++ b/manifests/charts/dubbod/templates/deployment.yaml
@@ -39,6 +39,7 @@
{{- $replicaCount := int (coalesce .Values.replicaCount $defaults.replicaCount
1) }}
{{- $gateway := $global.gateway | default dict }}
{{- $defaultGateway := $defaultGlobal.gateway | default dict }}
+{{- $gatewayImage := coalesce $gateway.image $defaultGateway.image
"kdubbo/dxgate:latest" }}
{{- $gatewayReplicas := int (coalesce $gateway.replicaCount
$defaultGateway.replicaCount 2) }}
{{- $remoteAccessCertificateHosts := $remoteAccess.certificateHosts | default
$defaultRemoteAccess.certificateHosts | default (list) }}
{{- $eastWestGateways := $eastWestGateway.gateways | default
$defaultEastWestGateway.gateways | default (list) }}
@@ -140,6 +141,8 @@ spec:
value: default
- name: DUBBO_CERT_PROVIDER
value: dubbod
+ - name: DUBBO_DXGATE_IMAGE
+ value: {{ $gatewayImage | quote }}
{{- if gt (len $remoteAccessCertificateHosts) 0 }}
- name: DUBBOD_CUSTOM_HOST
value: {{ join "," $remoteAccessCertificateHosts | quote }}
diff --git a/manifests/charts/dubbod/values.yaml
b/manifests/charts/dubbod/values.yaml
index 04358956..c006466d 100644
--- a/manifests/charts/dubbod/values.yaml
+++ b/manifests/charts/dubbod/values.yaml
@@ -49,6 +49,10 @@ _internal_default_values_not_set:
port: 26080
gateway:
+ # Image used by managed Gateway deployments. Keep it independently
+ # configurable from dubbod/CNI so a gateway can be rolled forward or
+ # back without replacing the control plane.
+ image: "kdubbo/dxgate:latest"
# Default replica count for every managed dxgate Deployment. An
# individual Gateway overrides it with the
# gateway.dubbo.apache.org/replicas annotation.
diff --git a/samples/activation/README.md b/samples/activation/README.md
index 173ffff3..2dd23501 100644
--- a/samples/activation/README.md
+++ b/samples/activation/README.md
@@ -19,7 +19,7 @@
EDS 下发新端点 -> dxgate 放行被扣住的请求
```
-网关是唯一能做这件事的位置。出向是 proxyless 的，调用方进程内的 gRPC xDS client 直接拿 EDS
端点建连，端点为空时当场失败，请求路径上没有任何东西活得够久去触发扩容。而进到网关的请求已经在网关自己的 task 里，可以等。
+南北向请求直接经过 dxgate。东西向请求原本由 proxyless 调用方直接读 EDS；冷服务的 EDS 现在会临时改写为专用 Activator
`dxgate-gateway`，所以同一个扣流和上报机制也能接住服务间调用。后端就绪后，EDS 恢复真实端点，新请求直接访问后端。
网关只负责等和上报，副本数始终由 KEDA 写。这条边界是有意的：网关重启不会把某个工作负载留在没人要求过的副本数上。
@@ -30,7 +30,7 @@ helm repo add kedacore https://kedacore.github.io/charts
helm install keda kedacore/keda -n keda --create-namespace
```
-需要一个 Gateway API 的 Gateway 把流量导到 `payment`。`dubbod` 会为它拉起 dxgate，并注入
`DXGATE_ACTIVATION_CONTROL_PLANE`。
+需要一个名为 `dxgate-gateway` 的 Gateway 作为命名空间内的专用 Activator。`dubbod` 会为它拉起
dxgate，并注入 `DXGATE_ACTIVATION_CONTROL_PLANE`。其他 Gateway 使用各自派生的
Deployment/Service 名称，不会覆盖或清理 Activator 资源。
### 部署
@@ -60,8 +60,8 @@ time curl -s http://$GATEWAY/payment/healthz
请求会挂住几秒——那是冷启动——然后正常返回，不是 503。同一时刻看网关：
```bash
-kubectl -n dubbo-system exec deploy/dxgate -- \
- curl -s localhost:15021/metrics | grep dxgate_activation_requests_held
+kubectl -n activation port-forward deploy/dxgate-gateway 15021:26021
+curl -s localhost:15021/metrics | grep dxgate_activation_requests_held
# dxgate_activation_requests_held 1
```
@@ -73,7 +73,7 @@ kubectl -n dubbo-system exec deploy/dxgate -- \
kubectl -n activation get serviceactivationpolicy payment -o
jsonpath='{.status.conditions}' | jq
```
-四个 condition：`Accepted` 策略本身合法，`Eligible` 目标可被激活，`ScalerReady` KEDA
能拿到指标，`ActivatorReady` 有网关在上报。
+四个 condition：`Accepted` 策略本身合法，`Eligible` 目标可被激活，`ScalerReady` 引用的 KEDA
`ScaledObject` 已 Ready，`ActivatorReady` 同命名空间至少一个 Dubbo Gateway 已
Programmed。后两项读取 Kubernetes 共享状态，HA 副本不会因各自持有不同连接而互相覆盖。
### 三个组件各自负责什么
@@ -119,20 +119,26 @@ kubectl -n activation get serviceactivationpolicy payment
-o jsonpath='{.status.
`dxplane_connections_force_closed_total` 非零说明 25s 不够，调
`DUBBO_GRPC_INBOUND_TERMINATION_DRAIN_DURATION`。
-### 为什么 scaler 地址是 headless 的
+### 为什么网关上报和 KEDA 查询使用不同地址
```yaml
-scalerAddress: dubbod-activation-replicas.dubbo-system.svc.cluster.local:26030
+scalerAddress: dubbod-activation.dubbo-system.svc.cluster.local:26030
```
-不是 `dubbod-activation`。KEDA 查询会落到某一个副本上，而网关上报也只会到某一个副本——如果两边落到不同副本，KEDA 看到的
pending 永远是 0。所以网关向 headless 名解析出的**每一个**地址都上报一份，这样无论 KEDA 问到谁都能拿到数。
+ScaledObject 使用负载均衡的 `dubbod-activation`，无论 KEDA 落到哪个控制面副本都能拿到相同
pending。网关上报使用 headless 的
`dubbod-activation-replicas`，向解析出的**每一个**地址各上报一份，保证每个控制面副本都有相同数据。
同理，网关必须注入 `POD_NAME`：控制面按 reporter 身份聚合，两个网关副本共用一个身份会互相覆盖。这个变量由
`kube-gateway.yaml` 自动注入。
-### 东西向还不行
+### 东西向和 mTLS
-这个样例是南北向的：请求从网关进来。
+带 `ServiceActivationPolicy` 的冷服务不会收到空 EDS。`dubbod` 把端点临时改成同命名空间
`dxgate-gateway` 的地址；Activator RDS 再按原始 Host 路由到真实服务。扩容完成后只切 EDS，不切 CDS。
-服务之间的调用（东西向）目前不能缩到零。调用方是 proxyless 的，直接读 EDS
建连，中间没有网关。`dubbod/discovery/pkg/xds` 还不感知激活状态，缩到零时下发的是空端点列表，调用方直接失败。
+`backendServiceAccounts` 是生产必填项。CDS 的 `MatchSubjectAltNames` 始终包含后端身份和
Activator 身份，冷/热切换期间 SAN 集合保持不变，避免证书校验窗口。不要为了省配置使用通配 SAN。
-所以只被其他服务调用、不经过网关的服务，仍然用 `minReplicaCount: 1`。
+### 生产边界
+
+- 只支持 HTTP 和 unary gRPC；流式 RPC、长连接、启动时间超过调用方 deadline 的服务保持 `minReplicaCount:
1`。
+- Activator 和 dubbod 都至少两个副本，并配置 PodDisruptionBudget。控制面 pending
是内存状态；全部控制面同时重启时，在网关下一次上报前 KEDA 暂时读到 0。
+- `maxPendingRequests` 和网关全局 backlog 都要压测。满载时按 `failurePolicy` 快速失败，不承诺无限排队。
+- 监控 `dxgate_activation_requests_held`、请求 4xx/5xx、KEDA ScaledObject/HPA 条件、策略的
`ScalerReady`/`ActivatorReady`。告警必须覆盖“pending 持续上升但副本仍为 0”。
+- 升级先保持目标至少一个副本，升级 CRD/base、dubbod、dxgate 后确认两种 SAN 和 Activator RDS 已下发，再恢复
`minReplicaCount: 0`。
diff --git a/samples/activation/activation-policy.yaml
b/samples/activation/activation-policy.yaml
index 089c574d..6725abce 100644
--- a/samples/activation/activation-policy.yaml
+++ b/samples/activation/activation-policy.yaml
@@ -14,10 +14,8 @@
# limitations under the License.
# Declares that payment may be scaled to zero and that requests for it should
-# be held rather than failed. Without this the gateway still holds the request
-# — it cannot tell a cold Service from a dead one on its own — but the control
-# plane publishes no scaler metric for the target, so nothing scales up and the
-# request only ever times out.
+# be held rather than failed. Without this, EDS is not rewritten to the
+# Activator and the control plane publishes no scaler metric for the target.
apiVersion: networking.dubbo.apache.org/v1alpha3
kind: ServiceActivationPolicy
metadata:
diff --git a/samples/activation/scaledobject.yaml
b/samples/activation/scaledobject.yaml
index 789033eb..7357e35c 100644
--- a/samples/activation/scaledobject.yaml
+++ b/samples/activation/scaledobject.yaml
@@ -43,10 +43,10 @@ spec:
triggers:
- type: external
metadata:
- # Headless: one address per dubbod pod. The load-balanced Service would
- # send every query to one replica, which may not be the replica the
- # gateway reported to.
- scalerAddress:
dubbod-activation-replicas.dubbo-system.svc.cluster.local:26030
+ # Gateways report to every control-plane replica through the headless
+ # replicas Service. KEDA can therefore use the stable load-balanced
+ # Service and receive the same pending count from any replica.
+ scalerAddress: dubbod-activation.dubbo-system.svc.cluster.local:26030
service: payment
namespace: activation
# Pending requests per replica before KEDA adds another. Activation
from
diff --git a/tests/e2e/run.sh b/tests/e2e/run.sh
index d10a2a8a..042a7de9 100755
--- a/tests/e2e/run.sh
+++ b/tests/e2e/run.sh
@@ -18,7 +18,7 @@
# installs the base + dubbod Helm charts, and asserts that the control
# plane serves xDS state for workloads and ServiceEntry configuration.
#
-# Requirements: docker, kind, kubectl, helm.
+# Requirements: docker, kind, kubectl, helm, jq.
#
# Environment knobs:
# CLUSTER_NAME kind cluster name (default: dubbo-e2e)
@@ -31,6 +31,9 @@
# KEEP_CLUSTER set to 1 to keep the kind cluster after the run
# 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)
+# DXGATE_IMAGE prebuilt dxgate image used by managed Gateways
+# KEDA_VERSION pinned KEDA chart/app version (default: 2.20.2)
set -euo pipefail
@@ -48,6 +51,11 @@ UPGRADE_TMP_DIR=""
PREVIOUS_CHART=""
KIND="${KIND:-kind}"
KIND_NODE_IMAGE="${KIND_NODE_IMAGE:-}"
+ACTIVATION_E2E="${ACTIVATION_E2E:-0}"
+DXGATE_IMAGE="${DXGATE_IMAGE:-kdubbo/dxgate: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}"
log() { echo "--- $*"; }
@@ -57,6 +65,17 @@ fail() {
"${KUBECTL[@]}" get pods -A -o wide >&2 || true
echo "--- diagnostics: dubbod logs ---" >&2
"${KUBECTL[@]}" -n "${SYSTEM_NS}" logs deploy/dubbod --tail=100 >&2 || true
+ echo "--- diagnostics: managed gateway logs ---" >&2
+ local pod
+ while read -r pod; do
+ [[ -n "${pod}" ]] || continue
+ echo "--- ${APP_NS}/${pod} current ---" >&2
+ "${KUBECTL[@]}" -n "${APP_NS}" logs "${pod}" --all-containers --tail=100
>&2 || true
+ echo "--- ${APP_NS}/${pod} previous ---" >&2
+ "${KUBECTL[@]}" -n "${APP_NS}" logs "${pod}" --all-containers --previous
--tail=100 >&2 || true
+ done < <("${KUBECTL[@]}" -n "${APP_NS}" get pods \
+ -l gateway.networking.k8s.io/gateway-name \
+ -o jsonpath='{range .items[*]}{.metadata.name}{"\n"}{end}' 2>/dev/null ||
true)
exit 1
}
@@ -119,11 +138,33 @@ if ! "${KIND}" get clusters 2>/dev/null | grep -qx
"${CLUSTER_NAME}"; then
"${KIND}" "${KIND_CREATE_ARGS[@]}"
fi
+# A kept cluster is useful for debugging and repeated activation runs. Reset
+# release-scoped state so runtime-mutated webhook fields cannot conflict with
+# the next Helm install after a previous namespace was deleted.
+helm uninstall dubbod --kube-context "kind-${CLUSTER_NAME}" -n "${SYSTEM_NS}"
--ignore-not-found >/dev/null 2>&1 || true
+helm uninstall dubbo-base --kube-context "kind-${CLUSTER_NAME}" -n
"${SYSTEM_NS}" --ignore-not-found >/dev/null 2>&1 || true
+"${KUBECTL[@]}" delete namespace "${APP_NS}" "${SYSTEM_NS}" --ignore-not-found
--wait=true >/dev/null
+"${KUBECTL[@]}" delete
validatingwebhookconfiguration,mutatingwebhookconfiguration \
+ -l app=dubbod --ignore-not-found >/dev/null
+
log "loading ${IMAGE} into kind"
"${KIND}" load docker-image "${IMAGE}" --name "${CLUSTER_NAME}"
if [[ "${IMAGE}" != "${UPGRADE_FROM_IMAGE}" ]]; then
"${KIND}" load docker-image "${UPGRADE_FROM_IMAGE}" --name "${CLUSTER_NAME}"
fi
+if [[ "${ACTIVATION_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"
+ docker tag "${IMAGE}" "${ACTIVATION_CLIENT_IMAGE}"
+ log "loading activation data-plane images into kind"
+ "${KIND}" load docker-image \
+ "${DXGATE_IMAGE}" \
+ "${ACTIVATION_APP_IMAGE}" \
+ "${ACTIVATION_CLIENT_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
@@ -131,6 +172,20 @@ log "installing Gateway API CRDs"
GATEWAY_API_VERSION="${GATEWAY_API_VERSION:-v1.4.1}"
"${KUBECTL[@]}" apply --server-side -f
"https://github.com/kubernetes-sigs/gateway-api/releases/download/${GATEWAY_API_VERSION}/experimental-install.yaml";
+if [[ "${ACTIVATION_E2E}" == "1" ]]; then
+ log "installing KEDA ${KEDA_VERSION}"
+ helm repo add kedacore https://kedacore.github.io/charts --force-update
+ helm repo update kedacore
+ helm upgrade --install keda kedacore/keda \
+ --version "${KEDA_VERSION}" \
+ --kube-context "kind-${CLUSTER_NAME}" \
+ -n keda --create-namespace
+ "${KUBECTL[@]}" -n keda rollout status deploy/keda-operator --timeout=300s \
+ || fail "KEDA operator did not become ready"
+ "${KUBECTL[@]}" -n keda rollout status deploy/keda-admission-webhooks
--timeout=300s \
+ || fail "KEDA admission webhook did not become ready"
+fi
+
log "installing base chart (CRDs)"
helm upgrade --install dubbo-base "${ROOT}/manifests/charts/base" \
--kube-context "kind-${CLUSTER_NAME}" \
@@ -146,6 +201,7 @@ install_dubbod() {
-n "${SYSTEM_NS}" \
--set global.proxyless.cni.enabled=false \
--set-string global.proxyless.cni.image="${image}" \
+ --set-string global.gateway.image="${DXGATE_IMAGE}" \
--set replicaCount="${DUBBOD_REPLICAS}"
}
@@ -285,17 +341,156 @@ retry "Accepted condition on the policy"
check_policy_accepted
log "asserting a managed gateway is told where to report demand"
"${KUBECTL[@]}" -n "${APP_NS}" apply -f
"${ROOT}/tests/e2e/testdata/gateway.yaml" \
|| fail "Gateway was rejected"
-check_gateway_deployment() { "${KUBECTL[@]}" -n "${APP_NS}" get deploy
dxgate-gateway >/dev/null; }
+check_gateway_deployment() { "${KUBECTL[@]}" -n "${APP_NS}" get deploy
public-dubbo >/dev/null; }
retry "managed gateway deployment" check_gateway_deployment
-GATEWAY_ENV="$("${KUBECTL[@]}" -n "${APP_NS}" get deploy dxgate-gateway \
+GATEWAY_ENV="$("${KUBECTL[@]}" -n "${APP_NS}" get deploy public-dubbo \
-o
jsonpath='{.spec.template.spec.containers[0].env[?(@.name=="DXGATE_ACTIVATION_CONTROL_PLANE")].value}')"
[[ "${GATEWAY_ENV}" == dubbod-activation-replicas.* ]] \
|| fail "gateway reports to '${GATEWAY_ENV}', want the headless activation
Service"
# Reports are attributed per reporter; without a distinct identity two gateway
# replicas overwrite each other's counts instead of adding to them.
-"${KUBECTL[@]}" -n "${APP_NS}" get deploy dxgate-gateway \
+"${KUBECTL[@]}" -n "${APP_NS}" get deploy public-dubbo \
-o
jsonpath='{.spec.template.spec.containers[0].env[?(@.name=="POD_NAME")].valueFrom.fieldRef.fieldPath}'
\
| grep -qx metadata.name \
|| fail "gateway does not inject POD_NAME; demand reports would not be
attributable to a replica"
+if [[ "${ACTIVATION_E2E}" == "1" ]]; then
+ log "asserting multiple Gateways have isolated resources"
+ "${KUBECTL[@]}" -n "${APP_NS}" get deploy dxgate-gateway public-dubbo
>/dev/null \
+ || fail "canonical and public Gateway deployments do not coexist"
+ "${KUBECTL[@]}" -n "${APP_NS}" rollout status deploy/dxgate-gateway
--timeout=300s \
+ || fail "canonical Activator gateway did not become ready"
+ "${KUBECTL[@]}" -n "${APP_NS}" rollout status deploy/public-dubbo
--timeout=300s \
+ || 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"
+ "${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"
+
+ check_all_activators_have_payment_route() {
+ local pod pods
+ pods="$("${KUBECTL[@]}" -n "${APP_NS}" get pods \
+ -l gateway.networking.k8s.io/gateway-name=dxgate-gateway \
+ --field-selector=status.phase=Running \
+ -o jsonpath='{range .items[*]}{.metadata.name}{"\n"}{end}')"
+ [[ "$(wc -w <<<"${pods}")" -ge 2 ]] || return 1
+ while read -r pod; do
+ "${KUBECTL[@]}" get --raw \
+ "/api/v1/namespaces/${APP_NS}/pods/${pod}:26021/proxy/debug/config" \
+ | jq -e '.listeners[]
+ | select(.bind == "0.0.0.0:15080")
+ | .virtual_hosts[]
+ | select(.name == "activation|payment.e2e.svc.cluster.local|8080")' \
+ >/dev/null \
+ || return 1
+ done <<<"${pods}"
+ }
+ retry "all Activator replicas receive the payment activation route" \
+ check_all_activators_have_payment_route
+
+ check_payment_policy_runtime_ready() {
+ local conditions
+ conditions="$("${KUBECTL[@]}" -n "${APP_NS}" get serviceactivationpolicy
payment \
+ -o jsonpath='{range .status.conditions[*]}{.type}={.status}{"\n"}{end}')"
+ grep -qx 'ScalerReady=True' <<<"${conditions}" &&
+ grep -qx 'ActivatorReady=True' <<<"${conditions}"
+ }
+ retry "payment policy scaler and Activator readiness" \
+ check_payment_policy_runtime_ready
+
+ check_payment_scaled_to_zero() {
+ [[ "$("${KUBECTL[@]}" -n "${APP_NS}" get deploy payment -o
jsonpath='{.spec.replicas}')" == "0" ]]
+ }
+ retry "KEDA scale payment to zero" check_payment_scaled_to_zero
+
+ log "removing one control-plane and one Activator replica before cold demand"
+ "${KUBECTL[@]}" -n "${SYSTEM_NS}" delete pod \
+ "$("${KUBECTL[@]}" -n "${SYSTEM_NS}" get pod -l app=dubbod -o
jsonpath='{.items[0].metadata.name}')" \
+ --wait=false
+ "${KUBECTL[@]}" -n "${APP_NS}" delete pod \
+ "$("${KUBECTL[@]}" -n "${APP_NS}" get pod \
+ -l gateway.networking.k8s.io/gateway-name=dxgate-gateway \
+ -o jsonpath='{.items[0].metadata.name}')" \
+ --wait=false
+
+ "${KUBECTL[@]}" -n "${SYSTEM_NS}" rollout status deploy/dubbod
--timeout=180s \
+ || fail "control-plane replica did not recover"
+ "${KUBECTL[@]}" -n "${APP_NS}" rollout status deploy/dxgate-gateway
--timeout=180s \
+ || fail "Activator replica did not recover"
+ retry "all Activator replicas retain the payment route after failover" \
+ check_all_activators_have_payment_route
+ retry "payment policy remains runtime-ready after HA failover" \
+ check_payment_policy_runtime_ready
+
+ 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"
+ activation_metrics() {
+ local pod
+ while read -r pod; do
+ "${KUBECTL[@]}" get --raw \
+ "/api/v1/namespaces/${APP_NS}/pods/${pod}:26021/proxy/metrics"
+ done < <("${KUBECTL[@]}" -n "${APP_NS}" get pods \
+ -l gateway.networking.k8s.io/gateway-name=dxgate-gateway \
+ -o jsonpath='{range .items[*]}{.metadata.name}{"\n"}{end}')
+ }
+ activation_payment_requests_total() {
+ activation_metrics \
+ | awk '/^dxgate_http_route_requests_total\{/ &&
/cluster="outbound\|8080\|\|payment\.e2e\.svc\.cluster\.local"/ { total += $NF
} END { print total + 0 }'
+ }
+ # held_requests is an instantaneous gauge and can return to zero before a
+ # polling assertion observes it. The durable proof is the complete sequence:
+ # replicas were zero above, KEDA now scales from pending demand, the request
+ # succeeds, and the Activator route counter increases.
+ check_payment_scaled_from_zero() {
+ [[ "$("${KUBECTL[@]}" -n "${APP_NS}" get deploy payment -o
jsonpath='{.spec.replicas}')" -ge 1 ]]
+ }
+ retry "KEDA scale payment from zero" check_payment_scaled_from_zero
+ log "KEDA scaled payment from zero"
+ "${KUBECTL[@]}" -n "${APP_NS}" rollout status deploy/payment --timeout=180s \
+ || fail "payment did not become ready after KEDA activation"
+ "${KUBECTL[@]}" -n "${APP_NS}" wait \
+
--for=jsonpath='{.status.containerStatuses[?(@.name=="client")].state.terminated.exitCode}'=0
\
+ pod/payment-client --timeout=120s \
+ || fail "held proxyless request did not complete after automatic scale-up"
+ check_payment_client_success() {
+ "${KUBECTL[@]}" -n "${APP_NS}" logs payment-client -c client 2>/dev/null |
grep -qx payment-ok
+ }
+ retry "cold proxyless response is available in the pod log" \
+ check_payment_client_success
+ COLD_ACTIVATOR_REQUESTS="$(activation_payment_requests_total)"
+ log "cold request completed through Activator (route
requests=${COLD_ACTIVATOR_REQUESTS})"
+ [[ "${COLD_ACTIVATOR_REQUESTS}" -ge 1 ]] \
+ || fail "cold request completed without an Activator data-plane metric"
+
+ log "asserting EDS converges back to a direct hot path"
+ # The first ready endpoint and its EDS update are independent events. Keep
+ # one backend alive while polling the data plane, otherwise the short test
+ # cooldown can scale it back to zero before xDS convergence is observable.
+ "${KUBECTL[@]}" -n "${APP_NS}" patch scaledobject payment --type=merge \
+ -p '{"spec":{"minReplicaCount":1}}' >/dev/null
+ hot_request_bypasses_activator() {
+ 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
+ "${KUBECTL[@]}" -n "${APP_NS}" wait \
+
--for=jsonpath='{.status.containerStatuses[?(@.name=="client")].state.terminated.exitCode}'=0
\
+ pod/payment-client --timeout=30s >/dev/null || return 1
+ check_payment_client_success || return 1
+ after="$(activation_payment_requests_total)"
+ [[ "${after}" == "${before}" ]]
+ }
+ retry "hot proxyless request bypasses the Activator after EDS convergence" \
+ hot_request_bypasses_activator
+ log "hot proxyless request bypassed the Activator"
+
+ "${KUBECTL[@]}" -n "${APP_NS}" patch scaledobject payment --type=merge \
+ -p '{"spec":{"minReplicaCount":0}}' >/dev/null
+ retry "KEDA scale payment back to zero" check_payment_scaled_to_zero
+ log "real KEDA zero-to-one-to-zero activation passed"
+fi
+
log "e2e smoke test passed"
diff --git a/tests/e2e/testdata/eastwest-activation-scaledobject.yaml
b/tests/e2e/testdata/eastwest-activation-scaledobject.yaml
new file mode 100644
index 00000000..5828a6f4
--- /dev/null
+++ b/tests/e2e/testdata/eastwest-activation-scaledobject.yaml
@@ -0,0 +1,39 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: keda.sh/v1alpha1
+kind: ScaledObject
+metadata:
+ name: payment
+ namespace: e2e
+spec:
+ scaleTargetRef:
+ name: payment
+ minReplicaCount: 0
+ maxReplicaCount: 3
+ pollingInterval: 1
+ cooldownPeriod: 10
+ advanced:
+ horizontalPodAutoscalerConfig:
+ behavior:
+ scaleDown:
+ stabilizationWindowSeconds: 0
+ triggers:
+ - type: external
+ metadata:
+ scalerAddress: dubbod-activation.dubbo-system.svc.cluster.local:26030
+ service: payment
+ namespace: e2e
+ targetPendingRequests: "1"
