This is an automated email from the ASF dual-hosted git repository.
github-actions[bot] pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/dubbo-kubernetes.git
The following commit(s) were added to refs/heads/master by this push:
new 1560a5d3 Completed the first task; the initial serverless setup is
taking shape (#990)
1560a5d3 is described below
commit 1560a5d3353c4848a778b991aa57ce61eb501d5c
Author: mfordjody <[email protected]>
AuthorDate: Fri Aug 7 06:29:06 2026 +0800
Completed the first task; the initial serverless setup is taking shape
(#990)
---
README.md | 2 +
cni/pkg/nodeagent/iptables.go | 11 ++
cni/pkg/nodeagent/iptables_test.go | 1 +
dubboctl/cmd/analyze.go | 146 ++++++++++++++++-
dubboctl/cmd/analyze_test.go | 172 ++++++++++++++++++++
.../config/kube/gateway/deployment_controller.go | 37 +++++
.../kube/gateway/deployment_controller_test.go | 113 +++++++++++++
dubbod/discovery/pkg/features/dubbo.go | 4 +
manifests/charts/dubbod/files/grpc-engine.yaml | 14 ++
manifests/charts/dubbod/files/kube-gateway.yaml | 50 +++++-
manifests/charts/dubbod/templates/clusterrole.yaml | 14 ++
manifests/charts/dubbod/templates/deployment.yaml | 23 +++
manifests/charts/dubbod/values.yaml | 15 +-
operator/pkg/apis/proto/values_types.proto | 13 ++
operator/pkg/apis/values_types.pb.go | 177 +++++++++++++++------
operator/pkg/render/manifest_test.go | 102 ++++++++++++
pkg/kube/inject/proxyless.go | 8 +-
pkg/kube/inject/proxyless_test.go | 39 +++++
samples/autoscaling/README.md | 77 +++++++++
samples/autoscaling/kafka-consumer.yaml | 59 +++++++
samples/autoscaling/scaledobject.yaml | 51 ++++++
21 files changed, 1069 insertions(+), 59 deletions(-)
diff --git a/README.md b/README.md
index 1f469895..4459cc06 100644
--- a/README.md
+++ b/README.md
@@ -10,7 +10,9 @@ Dubbo service mesh enables workloads to join natively,
receive policies from the
> [!WARNING]
> The current version is in the **Alpha** phase.
+>
> Releases `0.4.6–0.4.9` will be in the **Beta** phase.
+>
> Release `0.5.0` will be the first **RC** version.
Dubbo’s control plane provides an abstraction layer over the underlying
cluster management platform.
diff --git a/cni/pkg/nodeagent/iptables.go b/cni/pkg/nodeagent/iptables.go
index 4c162b07..701acd84 100644
--- a/cni/pkg/nodeagent/iptables.go
+++ b/cni/pkg/nodeagent/iptables.go
@@ -28,6 +28,12 @@ const (
meshPodIPSet = "DUBBO-GRPC-INBOUND-PODS"
meshExcludeIPSet = "DUBBO-GRPC-INBOUND-EXCLUDE"
dxgateAdminPort = 26021
+ // dxplaneAdminPort carries the inbound sidecar's health, readiness and
+ // metrics endpoints. kubelet probes it from the host, so the fence has
to
+ // exempt it or every injected pod fails its readiness probe. Workloads
that
+ // move the admin listener elsewhere exempt it with the
+ // proxyless.dubbo.apache.org/excludeInboundPorts annotation instead.
+ dxplaneAdminPort = 15020
)
type CommandRunner interface {
@@ -163,10 +169,12 @@ func (m *IPTablesRuleManager) ensureBase(ctx
context.Context) error {
allowExcluded := []string{"-m", "set", "--match-set", meshExcludeIPSet,
"dst,dst", "-p", "tcp", "-j", "RETURN"}
allowGRPCInbound := []string{"-m", "set", "--match-set", meshPodIPSet,
"dst", "-p", "tcp", "--dport", fmt.Sprint(m.grpcInboundPort), "-j", "RETURN"}
allowDxgateAdmin := []string{"-m", "set", "--match-set", meshPodIPSet,
"dst", "-p", "tcp", "--dport", fmt.Sprint(dxgateAdminPort), "-j", "RETURN"}
+ allowDxplaneAdmin := []string{"-m", "set", "--match-set", meshPodIPSet,
"dst", "-p", "tcp", "--dport", fmt.Sprint(dxplaneAdminPort), "-j", "RETURN"}
rejectOtherTCP := []string{"-m", "set", "--match-set", meshPodIPSet,
"dst", "-p", "tcp", "-j", "REJECT"}
m.deleteRepeated(ctx, allowExcluded...)
m.deleteRepeated(ctx, allowGRPCInbound...)
m.deleteRepeated(ctx, allowDxgateAdmin...)
+ m.deleteRepeated(ctx, allowDxplaneAdmin...)
m.deleteRepeated(ctx, rejectOtherTCP...)
if err := m.appendRule(ctx, allowExcluded...); err != nil {
return err
@@ -177,6 +185,9 @@ func (m *IPTablesRuleManager) ensureBase(ctx
context.Context) error {
if err := m.appendRule(ctx, allowDxgateAdmin...); err != nil {
return err
}
+ if err := m.appendRule(ctx, allowDxplaneAdmin...); err != nil {
+ return err
+ }
return m.appendRule(ctx, rejectOtherTCP...)
}
diff --git a/cni/pkg/nodeagent/iptables_test.go
b/cni/pkg/nodeagent/iptables_test.go
index dac96d27..e1121391 100644
--- a/cni/pkg/nodeagent/iptables_test.go
+++ b/cni/pkg/nodeagent/iptables_test.go
@@ -43,6 +43,7 @@ func TestIPTablesRuleManagerAddsGRPCInboundBoundaryRules(t
*testing.T) {
"-A DUBBO-GRPC-INBOUND -m set --match-set
DUBBO-GRPC-INBOUND-EXCLUDE dst,dst -p tcp -j RETURN",
"-A DUBBO-GRPC-INBOUND -m set --match-set
DUBBO-GRPC-INBOUND-PODS dst -p tcp --dport 15080 -j RETURN",
"-A DUBBO-GRPC-INBOUND -m set --match-set
DUBBO-GRPC-INBOUND-PODS dst -p tcp --dport 26021 -j RETURN",
+ "-A DUBBO-GRPC-INBOUND -m set --match-set
DUBBO-GRPC-INBOUND-PODS dst -p tcp --dport 15020 -j RETURN",
"-A DUBBO-GRPC-INBOUND -m set --match-set
DUBBO-GRPC-INBOUND-PODS dst -p tcp -j REJECT",
"ipset add DUBBO-GRPC-INBOUND-PODS 10.244.0.12 -exist",
"ipset add DUBBO-GRPC-INBOUND-EXCLUDE 10.244.0.12,tcp:9090
-exist",
diff --git a/dubboctl/cmd/analyze.go b/dubboctl/cmd/analyze.go
index dc1af1fc..24c6a63b 100644
--- a/dubboctl/cmd/analyze.go
+++ b/dubboctl/cmd/analyze.go
@@ -23,9 +23,12 @@ import (
"strings"
"github.com/apache/dubbo-kubernetes/dubboctl/pkg/cli"
+ "github.com/apache/dubbo-kubernetes/pkg/config/constants"
"github.com/apache/dubbo-kubernetes/pkg/kube"
"github.com/spf13/cobra"
+ appsv1 "k8s.io/api/apps/v1"
corev1 "k8s.io/api/core/v1"
+ policyv1 "k8s.io/api/policy/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
)
@@ -54,7 +57,12 @@ func AnalyzeCmd(ctx cli.Context) *cobra.Command {
Short: "Analyze Dubbo mesh configuration and report potential
issues",
Long: `Analyze inspects HTTPRoutes, security policies and
circuit breaker policies in the
cluster and reports references to missing services, selectors that match no
-workloads, and authentication configurations that are not actually enforced.`,
+workloads, and authentication configurations that are not actually enforced.
+
+It also checks that the control plane and every gateway survive a node drain:
+more than one replica, a PodDisruptionBudget, and replicas placed on different
+nodes. The control plane is checked even when the analysis is scoped to a
+single application namespace.`,
Example: ` # Analyze the default namespace
dubboctl analyze
@@ -105,9 +113,145 @@ func runAnalyzers(ctx context.Context, client
kube.CLIClient, namespace string)
msgs = append(msgs, analyzeHTTPRoutes(ctx, client, namespace,
services.Items)...)
msgs = append(msgs, analyzeSecurityPolicies(ctx, client, namespace,
pods.Items)...)
msgs = append(msgs, analyzeCircuitBreakerPolicies(ctx, client,
namespace, services.Items)...)
+ msgs = append(msgs, collectHighAvailability(ctx, client, namespace)...)
return msgs, nil
}
+// collectHighAvailability gathers the workloads whose availability the whole
+// mesh depends on. The control plane is always inspected in the system
+// namespace: a request scoped to one application namespace still fails in the
+// same way when dubbod goes down.
+func collectHighAvailability(ctx context.Context, client kube.CLIClient,
namespace string) []analyzeMessage {
+ namespaces := []string{namespace}
+ if namespace != metav1.NamespaceAll && namespace !=
constants.DubboSystemNamespace {
+ namespaces = append(namespaces, constants.DubboSystemNamespace)
+ }
+
+ msgs := []analyzeMessage{}
+ for _, ns := range namespaces {
+ deployments, err :=
client.Kube().AppsV1().Deployments(ns).List(ctx, metav1.ListOptions{})
+ if err != nil {
+ msgs = append(msgs, analyzeMessage{levelWarning,
"Deployment",
+ fmt.Sprintf("failed to list deployments in %s:
%v", ns, err)})
+ continue
+ }
+ budgets, err :=
client.Kube().PolicyV1().PodDisruptionBudgets(ns).List(ctx,
metav1.ListOptions{})
+ if err != nil {
+ msgs = append(msgs, analyzeMessage{levelWarning,
"PodDisruptionBudget",
+ fmt.Sprintf("failed to list pod disruption
budgets in %s: %v", ns, err)})
+ continue
+ }
+ pods, err := client.Kube().CoreV1().Pods(ns).List(ctx,
metav1.ListOptions{})
+ if err != nil {
+ msgs = append(msgs, analyzeMessage{levelWarning, "Pod",
+ fmt.Sprintf("failed to list pods in %s: %v",
ns, err)})
+ continue
+ }
+ msgs = append(msgs, analyzeHighAvailability(deployments.Items,
budgets.Items, pods.Items)...)
+ }
+ return msgs
+}
+
+// meshCriticalRole names a workload whose loss takes the mesh, or a whole
+// gateway's traffic, with it.
+func meshCriticalRole(deployment appsv1.Deployment) string {
+ labels := deployment.Spec.Template.Labels
+ switch {
+ case labels["app"] == "dubbod":
+ return "control plane"
+ case labels["app.kubernetes.io/name"] == "dxgate":
+ return "gateway"
+ default:
+ return ""
+ }
+}
+
+// analyzeHighAvailability reports mesh-critical workloads that a single node
+// drain, upgrade or eviction can take offline. It checks three things that all
+// have to hold together: more than one replica, a disruption budget so
+// voluntary evictions cannot remove them all at once, and pods that actually
+// landed on different nodes.
+func analyzeHighAvailability(deployments []appsv1.Deployment, budgets
[]policyv1.PodDisruptionBudget, pods []corev1.Pod) []analyzeMessage {
+ msgs := []analyzeMessage{}
+ for _, deployment := range deployments {
+ role := meshCriticalRole(deployment)
+ if role == "" {
+ continue
+ }
+ resource := fmt.Sprintf("Deployment %s/%s",
deployment.Namespace, deployment.Name)
+
+ replicas := int32(1)
+ if deployment.Spec.Replicas != nil {
+ replicas = *deployment.Spec.Replicas
+ }
+ if replicas == 0 {
+ // Deliberately scaled to zero; availability is not the
question.
+ continue
+ }
+ if replicas < 2 {
+ msgs = append(msgs, analyzeMessage{levelWarning,
resource,
+ fmt.Sprintf("%s runs a single replica: any node
drain, upgrade or eviction takes it offline", role)})
+ continue
+ }
+
+ selector, err :=
metav1.LabelSelectorAsSelector(deployment.Spec.Selector)
+ if err != nil {
+ continue
+ }
+ if !hasDisruptionBudget(budgets, deployment.Namespace,
deployment.Spec.Template.Labels) {
+ msgs = append(msgs, analyzeMessage{levelWarning,
resource,
+ fmt.Sprintf("%s has %d replicas but no
PodDisruptionBudget: a node drain can evict them all at once", role, replicas)})
+ }
+ if node, single := singleNode(pods, deployment.Namespace,
selector); single {
+ msgs = append(msgs, analyzeMessage{levelWarning,
resource,
+ fmt.Sprintf("all %s replicas are scheduled on
node %s: losing it takes the %s down despite the replica count", role, node,
role)})
+ }
+ }
+ return msgs
+}
+
+func hasDisruptionBudget(budgets []policyv1.PodDisruptionBudget, namespace
string, podLabels map[string]string) bool {
+ for _, budget := range budgets {
+ if budget.Namespace != namespace || budget.Spec.Selector == nil
{
+ continue
+ }
+ selector, err :=
metav1.LabelSelectorAsSelector(budget.Spec.Selector)
+ if err != nil || selector.Empty() {
+ continue
+ }
+ if selector.Matches(labels.Set(podLabels)) {
+ return true
+ }
+ }
+ return false
+}
+
+// singleNode reports the node every running replica shares, if they share one.
+// Fewer than two running pods is not evidence of bad placement, so it is not
+// reported.
+func singleNode(pods []corev1.Pod, namespace string, selector labels.Selector)
(string, bool) {
+ node := ""
+ count := 0
+ for _, pod := range pods {
+ if pod.Namespace != namespace || pod.DeletionTimestamp != nil {
+ continue
+ }
+ if pod.Status.Phase != corev1.PodRunning || pod.Spec.NodeName
== "" {
+ continue
+ }
+ if !selector.Matches(labels.Set(pod.Labels)) {
+ continue
+ }
+ if node == "" {
+ node = pod.Spec.NodeName
+ } else if pod.Spec.NodeName != node {
+ return "", false
+ }
+ count++
+ }
+ return node, count > 1
+}
+
// analyzeHTTPRoutes reports backendRefs pointing at services or ports that do
not exist.
func analyzeHTTPRoutes(ctx context.Context, client kube.CLIClient, namespace
string, services []corev1.Service) []analyzeMessage {
msgs := []analyzeMessage{}
diff --git a/dubboctl/cmd/analyze_test.go b/dubboctl/cmd/analyze_test.go
new file mode 100644
index 00000000..3f1024e1
--- /dev/null
+++ b/dubboctl/cmd/analyze_test.go
@@ -0,0 +1,172 @@
+// 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 cmd
+
+import (
+ "strings"
+ "testing"
+
+ appsv1 "k8s.io/api/apps/v1"
+ corev1 "k8s.io/api/core/v1"
+ policyv1 "k8s.io/api/policy/v1"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func controlPlaneDeployment(replicas int32) appsv1.Deployment {
+ return appsv1.Deployment{
+ ObjectMeta: metav1.ObjectMeta{Name: "dubbod", Namespace:
"dubbo-system"},
+ Spec: appsv1.DeploymentSpec{
+ Replicas: &replicas,
+ Selector: &metav1.LabelSelector{MatchLabels:
map[string]string{"app": "dubbod"}},
+ Template: corev1.PodTemplateSpec{
+ ObjectMeta: metav1.ObjectMeta{Labels:
map[string]string{"app": "dubbod"}},
+ },
+ },
+ }
+}
+
+func controlPlaneBudget() policyv1.PodDisruptionBudget {
+ return policyv1.PodDisruptionBudget{
+ ObjectMeta: metav1.ObjectMeta{Name: "dubbod", Namespace:
"dubbo-system"},
+ Spec: policyv1.PodDisruptionBudgetSpec{
+ Selector: &metav1.LabelSelector{MatchLabels:
map[string]string{"app": "dubbod"}},
+ },
+ }
+}
+
+func runningPod(name, node string, labels map[string]string) corev1.Pod {
+ return corev1.Pod{
+ ObjectMeta: metav1.ObjectMeta{Name: name, Namespace:
"dubbo-system", Labels: labels},
+ Spec: corev1.PodSpec{NodeName: node},
+ Status: corev1.PodStatus{Phase: corev1.PodRunning},
+ }
+}
+
+func messagesContain(msgs []analyzeMessage, substring string) bool {
+ for _, msg := range msgs {
+ if strings.Contains(msg.Message, substring) {
+ return true
+ }
+ }
+ return false
+}
+
+func TestAnalyzeHighAvailabilityFlagsSingleReplicaControlPlane(t *testing.T) {
+ msgs := analyzeHighAvailability(
+ []appsv1.Deployment{controlPlaneDeployment(1)},
+ nil,
+ []corev1.Pod{runningPod("dubbod-a", "node-1",
map[string]string{"app": "dubbod"})},
+ )
+ if len(msgs) != 1 {
+ t.Fatalf("messages = %d, want 1:\n%v", len(msgs), msgs)
+ }
+ if !messagesContain(msgs, "control plane runs a single replica") {
+ t.Fatalf("unexpected message: %v", msgs)
+ }
+}
+
+func TestAnalyzeHighAvailabilityFlagsMissingDisruptionBudget(t *testing.T) {
+ msgs := analyzeHighAvailability(
+ []appsv1.Deployment{controlPlaneDeployment(2)},
+ nil,
+ []corev1.Pod{
+ runningPod("dubbod-a", "node-1",
map[string]string{"app": "dubbod"}),
+ runningPod("dubbod-b", "node-2",
map[string]string{"app": "dubbod"}),
+ },
+ )
+ if !messagesContain(msgs, "no PodDisruptionBudget") {
+ t.Fatalf("missing disruption budget not reported: %v", msgs)
+ }
+}
+
+// Replica count alone proves nothing: two pods on one node still go down
+// together, which is exactly what the topology spread is meant to prevent.
+func TestAnalyzeHighAvailabilityFlagsReplicasOnOneNode(t *testing.T) {
+ msgs := analyzeHighAvailability(
+ []appsv1.Deployment{controlPlaneDeployment(2)},
+ []policyv1.PodDisruptionBudget{controlPlaneBudget()},
+ []corev1.Pod{
+ runningPod("dubbod-a", "node-1",
map[string]string{"app": "dubbod"}),
+ runningPod("dubbod-b", "node-1",
map[string]string{"app": "dubbod"}),
+ },
+ )
+ if len(msgs) != 1 {
+ t.Fatalf("messages = %d, want 1:\n%v", len(msgs), msgs)
+ }
+ if !messagesContain(msgs, "scheduled on node node-1") {
+ t.Fatalf("unexpected message: %v", msgs)
+ }
+}
+
+func TestAnalyzeHighAvailabilityAcceptsHealthyControlPlane(t *testing.T) {
+ msgs := analyzeHighAvailability(
+ []appsv1.Deployment{controlPlaneDeployment(2)},
+ []policyv1.PodDisruptionBudget{controlPlaneBudget()},
+ []corev1.Pod{
+ runningPod("dubbod-a", "node-1",
map[string]string{"app": "dubbod"}),
+ runningPod("dubbod-b", "node-2",
map[string]string{"app": "dubbod"}),
+ },
+ )
+ if len(msgs) != 0 {
+ t.Fatalf("healthy control plane reported %d messages:\n%v",
len(msgs), msgs)
+ }
+}
+
+func TestAnalyzeHighAvailabilityIgnoresScaledDownAndUnrelatedWorkloads(t
*testing.T) {
+ zero := int32(0)
+ scaledDown := controlPlaneDeployment(1)
+ scaledDown.Spec.Replicas = &zero
+
+ unrelated := appsv1.Deployment{
+ ObjectMeta: metav1.ObjectMeta{Name: "httpbin", Namespace:
"backend"},
+ Spec: appsv1.DeploymentSpec{
+ Selector: &metav1.LabelSelector{MatchLabels:
map[string]string{"app": "httpbin"}},
+ Template: corev1.PodTemplateSpec{
+ ObjectMeta: metav1.ObjectMeta{Labels:
map[string]string{"app": "httpbin"}},
+ },
+ },
+ }
+
+ msgs := analyzeHighAvailability([]appsv1.Deployment{scaledDown,
unrelated}, nil, nil)
+ if len(msgs) != 0 {
+ t.Fatalf("expected no messages, got:\n%v", msgs)
+ }
+}
+
+func TestAnalyzeHighAvailabilityFlagsSingleReplicaGateway(t *testing.T) {
+ replicas := int32(1)
+ gateway := appsv1.Deployment{
+ ObjectMeta: metav1.ObjectMeta{Name: "public-dubbo", Namespace:
"app"},
+ Spec: appsv1.DeploymentSpec{
+ Replicas: &replicas,
+ Selector: &metav1.LabelSelector{MatchLabels:
map[string]string{
+ "app.kubernetes.io/name": "dxgate",
+ "app.kubernetes.io/instance": "public-dubbo",
+ }},
+ Template: corev1.PodTemplateSpec{
+ ObjectMeta: metav1.ObjectMeta{Labels:
map[string]string{
+ "app.kubernetes.io/name": "dxgate",
+ "app.kubernetes.io/instance":
"public-dubbo",
+ }},
+ },
+ },
+ }
+
+ msgs := analyzeHighAvailability([]appsv1.Deployment{gateway}, nil, nil)
+ if !messagesContain(msgs, "gateway runs a single replica") {
+ t.Fatalf("single replica gateway not reported: %v", msgs)
+ }
+}
diff --git a/dubbod/discovery/pkg/config/kube/gateway/deployment_controller.go
b/dubbod/discovery/pkg/config/kube/gateway/deployment_controller.go
index e8bc0bba..7889f56c 100644
--- a/dubbod/discovery/pkg/config/kube/gateway/deployment_controller.go
+++ b/dubbod/discovery/pkg/config/kube/gateway/deployment_controller.go
@@ -82,8 +82,15 @@ const (
xdsAddressAnnotation = "gateway.dubbo.apache.org/xds-address"
accessLogAnnotation = "gateway.dubbo.apache.org/access-log"
accessLogFormatAnnotation =
"gateway.dubbo.apache.org/access-log-format"
+ replicasAnnotation = "gateway.dubbo.apache.org/replicas"
)
+// defaultGatewayReplicas keeps a gateway serving while one replica is drained,
+// upgraded or evicted. A single replica turns any of those into an outage for
+// every route the gateway fronts. The installed default comes from
+// features.DxgateReplicas; this is the floor used when that value is unusable.
+const defaultGatewayReplicas = 2
+
var builtinClasses = getBuiltinClasses()
var classInfos = getClassInfos()
@@ -405,6 +412,7 @@ func (d *DeploymentController) configureGateway(log
*dubbolog.Logger, gw gateway
ServiceAccount: defaultName,
Ports: ports,
ServiceType: serviceType,
+ Replicas: gatewayReplicas(gw),
Revision: d.revision,
ControllerLabel: gi.controllerLabel,
BootstrapConfig: bootstrapConfig,
@@ -460,6 +468,7 @@ type TemplateInput struct {
ServiceAccount string
Ports []corev1.ServicePort
ServiceType corev1.ServiceType
+ Replicas int32
Revision string
ControllerLabel string
@@ -1410,6 +1419,34 @@ func gatewayServiceNodePort(gw gateway.Gateway) int32 {
return int32(port)
}
+// gatewayReplicas reports how many dxgate pods to run. Zero is allowed so a
+// gateway can be scaled down deliberately, which is why this does not reuse
+// positiveIntAnnotation.
+func gatewayReplicas(gw gateway.Gateway) int32 {
+ return gatewayReplicasWithDefault(gw, installedGatewayReplicas())
+}
+
+func gatewayReplicasWithDefault(gw gateway.Gateway, fallback int32) int32 {
+ if gw.Annotations == nil || gw.Annotations[replicasAnnotation] == "" {
+ return fallback
+ }
+ replicas, err := strconv.Atoi(gw.Annotations[replicasAnnotation])
+ if err != nil || replicas < 0 {
+ return fallback
+ }
+ return int32(replicas)
+}
+
+// installedGatewayReplicas is the mesh-wide default. A negative setting is
+// meaningless and a zero one would make every gateway default to serving no
+// traffic, so both fall back to the highly available floor.
+func installedGatewayReplicas() int32 {
+ if features.DxgateReplicas < 1 {
+ return defaultGatewayReplicas
+ }
+ return int32(features.DxgateReplicas)
+}
+
func positiveIntAnnotation(gw gateway.Gateway, name string) (int, bool) {
if gw.Annotations == nil || gw.Annotations[name] == "" {
return 0, false
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 ffcfc15e..1101ec89 100644
--- a/dubbod/discovery/pkg/config/kube/gateway/deployment_controller_test.go
+++ b/dubbod/discovery/pkg/config/kube/gateway/deployment_controller_test.go
@@ -501,6 +501,47 @@ func TestServiceTypeForGatewayUsesAnnotation(t *testing.T)
{
}
}
+func TestGatewayReplicas(t *testing.T) {
+ tests := []struct {
+ name string
+ annotation string
+ fallback int32
+ want int32
+ }{
+ {name: "unset takes the installed default", fallback: 2, want:
2},
+ {name: "installed default is honored", fallback: 5, want: 5},
+ {name: "explicit count wins", annotation: "4", fallback: 2,
want: 4},
+ {name: "zero scales the gateway down deliberately", annotation:
"0", fallback: 2, want: 0},
+ {name: "negative falls back to the default", annotation: "-1",
fallback: 2, want: 2},
+ {name: "garbage falls back to the default", annotation: "many",
fallback: 2, want: 2},
+ }
+
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ meta := metav1.ObjectMeta{}
+ if test.annotation != "" {
+ meta.Annotations =
map[string]string{replicasAnnotation: test.annotation}
+ }
+ got :=
gatewayReplicasWithDefault(gatewayv1.Gateway{ObjectMeta: meta}, test.fallback)
+ if got != test.want {
+ t.Fatalf("gatewayReplicasWithDefault() = %d,
want %d", got, test.want)
+ }
+ })
+ }
+}
+
+// TestInstalledGatewayReplicasRejectsUnusableSettings guards the mesh-wide
+// default: zero would make every gateway default to serving no traffic at all,
+// which is never what an operator means by "the default".
+func TestInstalledGatewayReplicasRejectsUnusableSettings(t *testing.T) {
+ if got := installedGatewayReplicas(); got < 1 {
+ t.Fatalf("installedGatewayReplicas() = %d, want at least 1",
got)
+ }
+ if got := gatewayReplicas(gatewayv1.Gateway{}); got < 1 {
+ t.Fatalf("gatewayReplicas() with no annotation = %d, want at
least 1", got)
+ }
+}
+
func TestObservabilityConfigForGatewayDefaults(t *testing.T) {
cfg := resolveGatewayObservability(gatewayv1.Gateway{
ObjectMeta: metav1.ObjectMeta{Name: "public", Namespace: "app"},
@@ -702,6 +743,78 @@ func TestKubeGatewayTemplateRendersDxgateResources(t
*testing.T) {
}
}
+// TestKubeGatewayTemplateRendersHighAvailabilityResources covers the path a
+// real reconcile takes: gatewayReplicas defaults to 2, which must produce a
+// disruption budget and spread the pods. At one replica the budget has to be
+// omitted, or minAvailable 1 blocks every voluntary eviction of that pod.
+func TestKubeGatewayTemplateRendersHighAvailabilityResources(t *testing.T) {
+ templatePath := filepath.Join("..", "..", "..", "..", "..", "..",
"manifests", "charts", "dubbod", "files", "kube-gateway.yaml")
+ raw, err := os.ReadFile(templatePath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ templates, err := inject.ParseTemplates(inject.RawTemplates{"gateway":
string(raw)})
+ if err != nil {
+ t.Fatal(err)
+ }
+ controller := &DeploymentController{
+ injectConfig: func() inject.Config {
+ return inject.Config{Templates: templates}
+ },
+ }
+ input := TemplateInput{
+ Gateway: &gatewayv1.Gateway{
+ ObjectMeta: metav1.ObjectMeta{Name: "public",
Namespace: "app"},
+ },
+ DeploymentName: "public-dubbo",
+ ServiceAccount: "public-dubbo",
+ Ports: []corev1.ServicePort{{Name: "http", Port:
80, TargetPort: intstr.FromInt(15080)}},
+ ServiceType: corev1.ServiceTypeLoadBalancer,
+ Replicas: gatewayReplicas(gatewayv1.Gateway{}),
+ Revision: "default",
+ BootstrapConfig: "{}\n",
+ BootstrapConfigHash: "abc123",
+ DxgateImage: "kdubbo/dxgate:test",
+ }
+
+ rendered, err := controller.render("gateway", input)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(rendered) != 5 {
+ t.Fatalf("rendered %d resources, want 5 including the
disruption budget", len(rendered))
+ }
+ joined := strings.Join(rendered, "\n---\n")
+ for _, want := range []string{
+ "replicas: 2",
+ "kind: PodDisruptionBudget",
+ "minAvailable: 1",
+ "podAntiAffinity",
+ "topologySpreadConstraints",
+ "whenUnsatisfiable: ScheduleAnyway",
+ } {
+ if !strings.Contains(joined, want) {
+ t.Fatalf("high availability output missing %q:\n%s",
want, joined)
+ }
+ }
+
+ input.Replicas = 1
+ rendered, err = controller.render("gateway", input)
+ 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))
+ }
+ joined = strings.Join(rendered, "\n---\n")
+ if strings.Contains(joined, "kind: PodDisruptionBudget") {
+ t.Fatalf("disruption budget rendered for a single
replica:\n%s", joined)
+ }
+ if strings.Contains(joined, "topologySpreadConstraints") {
+ t.Fatalf("topology spread rendered for a single replica:\n%s",
joined)
+ }
+}
+
func TestResolveGatewayObservabilityTelemetryHierarchy(t *testing.T) {
resources := []telemetryconfig.Resource{
{
diff --git a/dubbod/discovery/pkg/features/dubbo.go
b/dubbod/discovery/pkg/features/dubbo.go
index 8dfa92d1..c455dc8d 100644
--- a/dubbod/discovery/pkg/features/dubbo.go
+++ b/dubbod/discovery/pkg/features/dubbo.go
@@ -63,6 +63,10 @@ var (
"Name of the default GatewayClass").Get()
DxgateImage = env.Register("DUBBO_DXGATE_IMAGE", "kdubbo/dxgate:latest",
"Container image used for managed Dubbo Gateway API data-plane
deployments").Get()
+ DxgateReplicas = env.Register("DUBBO_DXGATE_REPLICAS", 2,
+ "Default replica count for managed Dubbo Gateway API data-plane
deployments. Two or more keeps a"+
+ " gateway serving while one replica is drained;
individual gateways override it with the"+
+ " gateway.dubbo.apache.org/replicas annotation").Get()
StatusMaxWorkers = env.Register("DUBBO_STATUS_MAX_WORKERS", 100, "The
maximum number of workers"+
" for status update").Get()
)
diff --git a/manifests/charts/dubbod/files/grpc-engine.yaml
b/manifests/charts/dubbod/files/grpc-engine.yaml
index 8a4d9120..f714a8a3 100644
--- a/manifests/charts/dubbod/files/grpc-engine.yaml
+++ b/manifests/charts/dubbod/files/grpc-engine.yaml
@@ -120,6 +120,20 @@ spec:
- name: grpc-inbound
containerPort: 15080
protocol: TCP
+ - name: dxplane-admin
+ containerPort: 15020
+ protocol: TCP
+ # Failing readiness is what withdraws this pod from its EndpointSlice while
+ # the listener is still up, so callers whose EDS has not caught up keep
+ # succeeding through the drain. periodSeconds x failureThreshold must stay
+ # below the sidecar's termination drain delay (5s by default), or the pod
+ # closes its listener before kubelet has noticed it is going away.
+ readinessProbe:
+ httpGet:
+ path: /readyz
+ port: 15020
+ periodSeconds: 2
+ failureThreshold: 2
volumeMounts:
- name: dubbo-xds
mountPath: /etc/dubbo/proxy
diff --git a/manifests/charts/dubbod/files/kube-gateway.yaml
b/manifests/charts/dubbod/files/kube-gateway.yaml
index f89a2b9e..23b44ad8 100644
--- a/manifests/charts/dubbod/files/kube-gateway.yaml
+++ b/manifests/charts/dubbod/files/kube-gateway.yaml
@@ -50,7 +50,7 @@ metadata:
app.kubernetes.io/managed-by: dubbod
gateway.networking.k8s.io/gateway-name: {{ .Gateway.Name }}
spec:
- replicas: 1
+ replicas: {{ .Replicas }}
selector:
matchLabels:
app.kubernetes.io/name: dxgate
@@ -73,6 +73,35 @@ spec:
prometheus.io/port: "26021"
spec:
serviceAccountName: {{ .ServiceAccount }}
+{{- if gt (int .Replicas) 1 }}
+ affinity:
+ podAntiAffinity:
+ preferredDuringSchedulingIgnoredDuringExecution:
+ - weight: 100
+ podAffinityTerm:
+ topologyKey: kubernetes.io/hostname
+ labelSelector:
+ matchLabels:
+ app.kubernetes.io/name: dxgate
+ app.kubernetes.io/instance: {{ .DeploymentName }}
+ # Advisory spread: a single-node or single-zone cluster still schedules
+ # every replica rather than leaving the gateway short.
+ topologySpreadConstraints:
+ - maxSkew: 1
+ topologyKey: topology.kubernetes.io/zone
+ whenUnsatisfiable: ScheduleAnyway
+ labelSelector:
+ matchLabels:
+ app.kubernetes.io/name: dxgate
+ app.kubernetes.io/instance: {{ .DeploymentName }}
+ - maxSkew: 1
+ topologyKey: kubernetes.io/hostname
+ whenUnsatisfiable: ScheduleAnyway
+ labelSelector:
+ matchLabels:
+ app.kubernetes.io/name: dxgate
+ app.kubernetes.io/instance: {{ .DeploymentName }}
+{{- end }}
containers:
- name: dxgate
image: {{ .DxgateImage }}
@@ -188,3 +217,22 @@ spec:
{{- end }}
protocol: {{ .Protocol | default "TCP" }}
{{- end }}
+{{- if gt (int .Replicas) 1 }}
+---
+apiVersion: policy/v1
+kind: PodDisruptionBudget
+metadata:
+ name: {{ .DeploymentName }}
+ 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 }}
+spec:
+ minAvailable: 1
+ selector:
+ matchLabels:
+ app.kubernetes.io/name: dxgate
+ app.kubernetes.io/instance: {{ .DeploymentName }}
+{{- end }}
diff --git a/manifests/charts/dubbod/templates/clusterrole.yaml
b/manifests/charts/dubbod/templates/clusterrole.yaml
index 16f4c5c9..04339413 100644
--- a/manifests/charts/dubbod/templates/clusterrole.yaml
+++ b/manifests/charts/dubbod/templates/clusterrole.yaml
@@ -111,6 +111,20 @@ rules:
- update
- patch
- delete
+ # The gateway controller renders a PodDisruptionBudget alongside each
+ # multi-replica dxgate Deployment. Without this the budget is generated but
+ # never applied, and the apply failure is the only symptom.
+ - apiGroups: ["policy"]
+ resources:
+ - poddisruptionbudgets
+ verbs:
+ - get
+ - list
+ - watch
+ - create
+ - update
+ - patch
+ - delete
- apiGroups: [""]
resources:
- services
diff --git a/manifests/charts/dubbod/templates/deployment.yaml
b/manifests/charts/dubbod/templates/deployment.yaml
index 69fae644..f11dc0f4 100644
--- a/manifests/charts/dubbod/templates/deployment.yaml
+++ b/manifests/charts/dubbod/templates/deployment.yaml
@@ -30,6 +30,9 @@
{{- $eastWestGateway := $multicluster.eastWestGateway | default dict }}
{{- $managementPort := int (coalesce $management.port $defaultManagement.port
26080) }}
{{- $replicaCount := int (coalesce .Values.replicaCount $defaults.replicaCount
1) }}
+{{- $gateway := $global.gateway | default dict }}
+{{- $defaultGateway := $defaultGlobal.gateway | default dict }}
+{{- $gatewayReplicas := int (coalesce $gateway.replicaCount
$defaultGateway.replicaCount 2) }}
{{- $remoteAccessCertificateHosts := $remoteAccess.certificateHosts | default
$defaultRemoteAccess.certificateHosts | default (list) }}
{{- $eastWestGateways := $eastWestGateway.gateways | default
$defaultEastWestGateway.gateways | default (list) }}
{{- $eastWestGatewayEntries := list }}
@@ -67,6 +70,24 @@ spec:
labelSelector:
matchLabels:
app: dubbod
+{{- if gt $replicaCount 1 }}
+ # ScheduleAnyway keeps the spread advisory: a single-node or single-zone
+ # cluster still schedules every replica instead of leaving the control
+ # plane one pod short.
+ topologySpreadConstraints:
+ - maxSkew: 1
+ topologyKey: topology.kubernetes.io/zone
+ whenUnsatisfiable: ScheduleAnyway
+ labelSelector:
+ matchLabels:
+ app: dubbod
+ - maxSkew: 1
+ topologyKey: kubernetes.io/hostname
+ whenUnsatisfiable: ScheduleAnyway
+ labelSelector:
+ matchLabels:
+ app: dubbod
+{{- end }}
containers:
- name: execute
image: {{ $image | quote }}
@@ -106,6 +127,8 @@ spec:
- name: DUBBO_EASTWEST_GATEWAYS
value: {{ join "," $eastWestGatewayEntries | quote }}
{{- end }}
+ - name: DUBBO_DXGATE_REPLICAS
+ value: {{ $gatewayReplicas | quote }}
- name: CLUSTER_ID
value: "Kubernetes"
- name: POD_NAME
diff --git a/manifests/charts/dubbod/values.yaml
b/manifests/charts/dubbod/values.yaml
index a0dc0ebf..b5731aff 100644
--- a/manifests/charts/dubbod/values.yaml
+++ b/manifests/charts/dubbod/values.yaml
@@ -16,9 +16,12 @@
_internal_default_values_not_set:
revision: ""
- # Number of dubbod replicas. Set >= 2 for a highly available control plane;
- # a PodDisruptionBudget (minAvailable: 1) is created automatically then.
- replicaCount: 1
+ # Number of dubbod replicas. Two by default: a single replica makes every
+ # node drain, upgrade or eviction a full control-plane outage, during which
+ # no workload can be injected and no xDS update is served. At >= 2 a
+ # PodDisruptionBudget (minAvailable: 1) and topology spread are added
+ # automatically. Set to 1 only for local development.
+ replicaCount: 2
global:
proxy:
@@ -45,6 +48,12 @@ _internal_default_values_not_set:
management:
port: 26080
+ gateway:
+ # Default replica count for every managed dxgate Deployment. An
+ # individual Gateway overrides it with the
+ # gateway.dubbo.apache.org/replicas annotation.
+ replicaCount: 2
+
configValidation: true
multicluster:
diff --git a/operator/pkg/apis/proto/values_types.proto
b/operator/pkg/apis/proto/values_types.proto
index d008ae6a..bcbeec79 100644
--- a/operator/pkg/apis/proto/values_types.proto
+++ b/operator/pkg/apis/proto/values_types.proto
@@ -60,6 +60,12 @@ message MeshCNIConfig {
string refreshInterval = 11;
}
+message GatewayConfig {
+ // Default replica count for managed dxgate deployments. Individual Gateways
+ // override it with the gateway.dubbo.apache.org/replicas annotation.
+ google.protobuf.Int32Value replicaCount = 1;
+}
+
message GlobalConfig {
ProxyConfig proxy = 1;
@@ -72,6 +78,8 @@ message GlobalConfig {
MulticlusterConfig multicluster = 5;
ProxylessConfig proxyless = 6;
+
+ GatewayConfig gateway = 7;
}
message RemoteAccessConfig {
@@ -124,6 +132,11 @@ message Values {
// Revision for rendered control plane resources.
string revision = 2;
+
+ // Number of dubbod replicas. Two by default; a PodDisruptionBudget and
+ // topology spread are added automatically above one. Wrapped so an explicit
+ // 1 is distinguishable from an unset field.
+ google.protobuf.Int32Value replicaCount = 3;
}
// IntOrString is a type that can hold an int32 or a string. When used in
diff --git a/operator/pkg/apis/values_types.pb.go
b/operator/pkg/apis/values_types.pb.go
index a20ca413..03c82c28 100644
--- a/operator/pkg/apis/values_types.pb.go
+++ b/operator/pkg/apis/values_types.pb.go
@@ -294,6 +294,52 @@ func (x *MeshCNIConfig) GetRefreshInterval() string {
return ""
}
+type GatewayConfig struct {
+ state protoimpl.MessageState `protogen:"open.v1"`
+ // Default replica count for managed dxgate deployments. Individual
Gateways
+ // override it with the gateway.dubbo.apache.org/replicas annotation.
+ ReplicaCount *wrapperspb.Int32Value
`protobuf:"bytes,1,opt,name=replicaCount,proto3" json:"replicaCount,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
+}
+
+func (x *GatewayConfig) Reset() {
+ *x = GatewayConfig{}
+ mi := &file_values_types_proto_msgTypes[4]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+}
+
+func (x *GatewayConfig) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*GatewayConfig) ProtoMessage() {}
+
+func (x *GatewayConfig) ProtoReflect() protoreflect.Message {
+ mi := &file_values_types_proto_msgTypes[4]
+ if x != nil {
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ if ms.LoadMessageInfo() == nil {
+ ms.StoreMessageInfo(mi)
+ }
+ return ms
+ }
+ return mi.MessageOf(x)
+}
+
+// Deprecated: Use GatewayConfig.ProtoReflect.Descriptor instead.
+func (*GatewayConfig) Descriptor() ([]byte, []int) {
+ return file_values_types_proto_rawDescGZIP(), []int{4}
+}
+
+func (x *GatewayConfig) GetReplicaCount() *wrapperspb.Int32Value {
+ if x != nil {
+ return x.ReplicaCount
+ }
+ return nil
+}
+
type GlobalConfig struct {
state protoimpl.MessageState `protogen:"open.v1"`
Proxy *ProxyConfig
`protobuf:"bytes,1,opt,name=proxy,proto3" json:"proxy,omitempty"`
@@ -302,13 +348,14 @@ type GlobalConfig struct {
ConfigValidation bool
`protobuf:"varint,4,opt,name=configValidation,proto3"
json:"configValidation,omitempty"`
Multicluster *MulticlusterConfig
`protobuf:"bytes,5,opt,name=multicluster,proto3" json:"multicluster,omitempty"`
Proxyless *ProxylessConfig
`protobuf:"bytes,6,opt,name=proxyless,proto3" json:"proxyless,omitempty"`
+ Gateway *GatewayConfig
`protobuf:"bytes,7,opt,name=gateway,proto3" json:"gateway,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *GlobalConfig) Reset() {
*x = GlobalConfig{}
- mi := &file_values_types_proto_msgTypes[4]
+ mi := &file_values_types_proto_msgTypes[5]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -320,7 +367,7 @@ func (x *GlobalConfig) String() string {
func (*GlobalConfig) ProtoMessage() {}
func (x *GlobalConfig) ProtoReflect() protoreflect.Message {
- mi := &file_values_types_proto_msgTypes[4]
+ mi := &file_values_types_proto_msgTypes[5]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -333,7 +380,7 @@ func (x *GlobalConfig) ProtoReflect() protoreflect.Message {
// Deprecated: Use GlobalConfig.ProtoReflect.Descriptor instead.
func (*GlobalConfig) Descriptor() ([]byte, []int) {
- return file_values_types_proto_rawDescGZIP(), []int{4}
+ return file_values_types_proto_rawDescGZIP(), []int{5}
}
func (x *GlobalConfig) GetProxy() *ProxyConfig {
@@ -378,6 +425,13 @@ func (x *GlobalConfig) GetProxyless() *ProxylessConfig {
return nil
}
+func (x *GlobalConfig) GetGateway() *GatewayConfig {
+ if x != nil {
+ return x.Gateway
+ }
+ return nil
+}
+
type RemoteAccessConfig struct {
state protoimpl.MessageState `protogen:"open.v1"`
Enabled bool
`protobuf:"varint,1,opt,name=enabled,proto3" json:"enabled,omitempty"`
@@ -392,7 +446,7 @@ type RemoteAccessConfig struct {
func (x *RemoteAccessConfig) Reset() {
*x = RemoteAccessConfig{}
- mi := &file_values_types_proto_msgTypes[5]
+ mi := &file_values_types_proto_msgTypes[6]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -404,7 +458,7 @@ func (x *RemoteAccessConfig) String() string {
func (*RemoteAccessConfig) ProtoMessage() {}
func (x *RemoteAccessConfig) ProtoReflect() protoreflect.Message {
- mi := &file_values_types_proto_msgTypes[5]
+ mi := &file_values_types_proto_msgTypes[6]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -417,7 +471,7 @@ func (x *RemoteAccessConfig) ProtoReflect()
protoreflect.Message {
// Deprecated: Use RemoteAccessConfig.ProtoReflect.Descriptor instead.
func (*RemoteAccessConfig) Descriptor() ([]byte, []int) {
- return file_values_types_proto_rawDescGZIP(), []int{5}
+ return file_values_types_proto_rawDescGZIP(), []int{6}
}
func (x *RemoteAccessConfig) GetEnabled() bool {
@@ -473,7 +527,7 @@ type EastWestGatewayEndpoint struct {
func (x *EastWestGatewayEndpoint) Reset() {
*x = EastWestGatewayEndpoint{}
- mi := &file_values_types_proto_msgTypes[6]
+ mi := &file_values_types_proto_msgTypes[7]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -485,7 +539,7 @@ func (x *EastWestGatewayEndpoint) String() string {
func (*EastWestGatewayEndpoint) ProtoMessage() {}
func (x *EastWestGatewayEndpoint) ProtoReflect() protoreflect.Message {
- mi := &file_values_types_proto_msgTypes[6]
+ mi := &file_values_types_proto_msgTypes[7]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -498,7 +552,7 @@ func (x *EastWestGatewayEndpoint) ProtoReflect()
protoreflect.Message {
// Deprecated: Use EastWestGatewayEndpoint.ProtoReflect.Descriptor instead.
func (*EastWestGatewayEndpoint) Descriptor() ([]byte, []int) {
- return file_values_types_proto_rawDescGZIP(), []int{6}
+ return file_values_types_proto_rawDescGZIP(), []int{7}
}
func (x *EastWestGatewayEndpoint) GetClusterName() string {
@@ -537,7 +591,7 @@ type EastWestGatewayConfig struct {
func (x *EastWestGatewayConfig) Reset() {
*x = EastWestGatewayConfig{}
- mi := &file_values_types_proto_msgTypes[7]
+ mi := &file_values_types_proto_msgTypes[8]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -549,7 +603,7 @@ func (x *EastWestGatewayConfig) String() string {
func (*EastWestGatewayConfig) ProtoMessage() {}
func (x *EastWestGatewayConfig) ProtoReflect() protoreflect.Message {
- mi := &file_values_types_proto_msgTypes[7]
+ mi := &file_values_types_proto_msgTypes[8]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -562,7 +616,7 @@ func (x *EastWestGatewayConfig) ProtoReflect()
protoreflect.Message {
// Deprecated: Use EastWestGatewayConfig.ProtoReflect.Descriptor instead.
func (*EastWestGatewayConfig) Descriptor() ([]byte, []int) {
- return file_values_types_proto_rawDescGZIP(), []int{7}
+ return file_values_types_proto_rawDescGZIP(), []int{8}
}
func (x *EastWestGatewayConfig) GetEnabled() bool {
@@ -624,7 +678,7 @@ type MulticlusterConfig struct {
func (x *MulticlusterConfig) Reset() {
*x = MulticlusterConfig{}
- mi := &file_values_types_proto_msgTypes[8]
+ mi := &file_values_types_proto_msgTypes[9]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -636,7 +690,7 @@ func (x *MulticlusterConfig) String() string {
func (*MulticlusterConfig) ProtoMessage() {}
func (x *MulticlusterConfig) ProtoReflect() protoreflect.Message {
- mi := &file_values_types_proto_msgTypes[8]
+ mi := &file_values_types_proto_msgTypes[9]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -649,7 +703,7 @@ func (x *MulticlusterConfig) ProtoReflect()
protoreflect.Message {
// Deprecated: Use MulticlusterConfig.ProtoReflect.Descriptor instead.
func (*MulticlusterConfig) Descriptor() ([]byte, []int) {
- return file_values_types_proto_rawDescGZIP(), []int{8}
+ return file_values_types_proto_rawDescGZIP(), []int{9}
}
func (x *MulticlusterConfig) GetRemoteAccess() *RemoteAccessConfig {
@@ -671,14 +725,18 @@ type Values struct {
// Global configuration for dubbo components.
Global *GlobalConfig `protobuf:"bytes,1,opt,name=global,proto3"
json:"global,omitempty"`
// Revision for rendered control plane resources.
- Revision string `protobuf:"bytes,2,opt,name=revision,proto3"
json:"revision,omitempty"`
+ Revision string `protobuf:"bytes,2,opt,name=revision,proto3"
json:"revision,omitempty"`
+ // Number of dubbod replicas. Two by default; a PodDisruptionBudget and
+ // topology spread are added automatically above one. Wrapped so an
explicit
+ // 1 is distinguishable from an unset field.
+ ReplicaCount *wrapperspb.Int32Value
`protobuf:"bytes,3,opt,name=replicaCount,proto3" json:"replicaCount,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *Values) Reset() {
*x = Values{}
- mi := &file_values_types_proto_msgTypes[9]
+ mi := &file_values_types_proto_msgTypes[10]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -690,7 +748,7 @@ func (x *Values) String() string {
func (*Values) ProtoMessage() {}
func (x *Values) ProtoReflect() protoreflect.Message {
- mi := &file_values_types_proto_msgTypes[9]
+ mi := &file_values_types_proto_msgTypes[10]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -703,7 +761,7 @@ func (x *Values) ProtoReflect() protoreflect.Message {
// Deprecated: Use Values.ProtoReflect.Descriptor instead.
func (*Values) Descriptor() ([]byte, []int) {
- return file_values_types_proto_rawDescGZIP(), []int{9}
+ return file_values_types_proto_rawDescGZIP(), []int{10}
}
func (x *Values) GetGlobal() *GlobalConfig {
@@ -720,6 +778,13 @@ func (x *Values) GetRevision() string {
return ""
}
+func (x *Values) GetReplicaCount() *wrapperspb.Int32Value {
+ if x != nil {
+ return x.ReplicaCount
+ }
+ return nil
+}
+
// IntOrString is a type that can hold an int32 or a string. When used in
// JSON or YAML marshalling and unmarshalling, it produces or consumes the
// inner type. This allows you to have, for example, a JSON field that can
@@ -740,7 +805,7 @@ type IntOrString struct {
func (x *IntOrString) Reset() {
*x = IntOrString{}
- mi := &file_values_types_proto_msgTypes[10]
+ mi := &file_values_types_proto_msgTypes[11]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -752,7 +817,7 @@ func (x *IntOrString) String() string {
func (*IntOrString) ProtoMessage() {}
func (x *IntOrString) ProtoReflect() protoreflect.Message {
- mi := &file_values_types_proto_msgTypes[10]
+ mi := &file_values_types_proto_msgTypes[11]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -765,7 +830,7 @@ func (x *IntOrString) ProtoReflect() protoreflect.Message {
// Deprecated: Use IntOrString.ProtoReflect.Descriptor instead.
func (*IntOrString) Descriptor() ([]byte, []int) {
- return file_values_types_proto_rawDescGZIP(), []int{10}
+ return file_values_types_proto_rawDescGZIP(), []int{11}
}
func (x *IntOrString) GetType() int64 {
@@ -812,7 +877,9 @@ const file_values_types_proto_rawDesc = "" +
"\fiptablesPath\x18\t \x01(\tR\fiptablesPath\x12\x1c\n" +
"\tipsetPath\x18\n" +
" \x01(\tR\tipsetPath\x12(\n" +
- "\x0frefreshInterval\x18\v \x01(\tR\x0frefreshInterval\"\xfa\x02\n" +
+ "\x0frefreshInterval\x18\v \x01(\tR\x0frefreshInterval\"P\n" +
+ "\rGatewayConfig\x12?\n" +
+ "\freplicaCount\x18\x01
\x01(\v2\x1b.google.protobuf.Int32ValueR\freplicaCount\"\xbc\x03\n" +
"\fGlobalConfig\x12:\n" +
"\x05proxy\x18\x01
\x01(\v2$.dubbo.operator.v1alpha1.ProxyConfigR\x05proxy\x12\x1e\n" +
"\n" +
@@ -823,7 +890,8 @@ const file_values_types_proto_rawDesc = "" +
"management\x12*\n" +
"\x10configValidation\x18\x04 \x01(\bR\x10configValidation\x12O\n" +
"\fmulticluster\x18\x05
\x01(\v2+.dubbo.operator.v1alpha1.MulticlusterConfigR\fmulticluster\x12F\n" +
- "\tproxyless\x18\x06
\x01(\v2(.dubbo.operator.v1alpha1.ProxylessConfigR\tproxyless\"\xd4\x01\n" +
+ "\tproxyless\x18\x06
\x01(\v2(.dubbo.operator.v1alpha1.ProxylessConfigR\tproxyless\x12@\n" +
+ "\agateway\x18\a
\x01(\v2&.dubbo.operator.v1alpha1.GatewayConfigR\agateway\"\xd4\x01\n" +
"\x12RemoteAccessConfig\x12\x18\n" +
"\aenabled\x18\x01 \x01(\bR\aenabled\x12 \n" +
"\vserviceType\x18\x02 \x01(\tR\vserviceType\x12\x18\n" +
@@ -849,10 +917,11 @@ const file_values_types_proto_rawDesc = "" +
"xdsAddress\"\xbf\x01\n" +
"\x12MulticlusterConfig\x12O\n" +
"\fremoteAccess\x18\x01
\x01(\v2+.dubbo.operator.v1alpha1.RemoteAccessConfigR\fremoteAccess\x12X\n" +
- "\x0feastWestGateway\x18\x02
\x01(\v2..dubbo.operator.v1alpha1.EastWestGatewayConfigR\x0feastWestGateway\"c\n"
+
+ "\x0feastWestGateway\x18\x02
\x01(\v2..dubbo.operator.v1alpha1.EastWestGatewayConfigR\x0feastWestGateway\"\xa4\x01\n"
+
"\x06Values\x12=\n" +
"\x06global\x18\x01
\x01(\v2%.dubbo.operator.v1alpha1.GlobalConfigR\x06global\x12\x1a\n" +
- "\brevision\x18\x02 \x01(\tR\brevision\"\x8c\x01\n" +
+ "\brevision\x18\x02 \x01(\tR\brevision\x12?\n" +
+ "\freplicaCount\x18\x03
\x01(\v2\x1b.google.protobuf.Int32ValueR\freplicaCount\"\x8c\x01\n" +
"\vIntOrString\x12\x12\n" +
"\x04type\x18\x01 \x01(\x03R\x04type\x123\n" +
"\x06intVal\x18\x02
\x01(\v2\x1b.google.protobuf.Int32ValueR\x06intVal\x124\n" +
@@ -870,39 +939,43 @@ func file_values_types_proto_rawDescGZIP() []byte {
return file_values_types_proto_rawDescData
}
-var file_values_types_proto_msgTypes = make([]protoimpl.MessageInfo, 11)
+var file_values_types_proto_msgTypes = make([]protoimpl.MessageInfo, 12)
var file_values_types_proto_goTypes = []any{
(*ManagementConfig)(nil), // 0:
dubbo.operator.v1alpha1.ManagementConfig
(*ProxyConfig)(nil), // 1:
dubbo.operator.v1alpha1.ProxyConfig
(*ProxylessConfig)(nil), // 2:
dubbo.operator.v1alpha1.ProxylessConfig
(*MeshCNIConfig)(nil), // 3:
dubbo.operator.v1alpha1.MeshCNIConfig
- (*GlobalConfig)(nil), // 4:
dubbo.operator.v1alpha1.GlobalConfig
- (*RemoteAccessConfig)(nil), // 5:
dubbo.operator.v1alpha1.RemoteAccessConfig
- (*EastWestGatewayEndpoint)(nil), // 6:
dubbo.operator.v1alpha1.EastWestGatewayEndpoint
- (*EastWestGatewayConfig)(nil), // 7:
dubbo.operator.v1alpha1.EastWestGatewayConfig
- (*MulticlusterConfig)(nil), // 8:
dubbo.operator.v1alpha1.MulticlusterConfig
- (*Values)(nil), // 9: dubbo.operator.v1alpha1.Values
- (*IntOrString)(nil), // 10:
dubbo.operator.v1alpha1.IntOrString
- (*wrapperspb.Int32Value)(nil), // 11: google.protobuf.Int32Value
- (*wrapperspb.StringValue)(nil), // 12: google.protobuf.StringValue
+ (*GatewayConfig)(nil), // 4:
dubbo.operator.v1alpha1.GatewayConfig
+ (*GlobalConfig)(nil), // 5:
dubbo.operator.v1alpha1.GlobalConfig
+ (*RemoteAccessConfig)(nil), // 6:
dubbo.operator.v1alpha1.RemoteAccessConfig
+ (*EastWestGatewayEndpoint)(nil), // 7:
dubbo.operator.v1alpha1.EastWestGatewayEndpoint
+ (*EastWestGatewayConfig)(nil), // 8:
dubbo.operator.v1alpha1.EastWestGatewayConfig
+ (*MulticlusterConfig)(nil), // 9:
dubbo.operator.v1alpha1.MulticlusterConfig
+ (*Values)(nil), // 10: dubbo.operator.v1alpha1.Values
+ (*IntOrString)(nil), // 11:
dubbo.operator.v1alpha1.IntOrString
+ (*wrapperspb.Int32Value)(nil), // 12: google.protobuf.Int32Value
+ (*wrapperspb.StringValue)(nil), // 13: google.protobuf.StringValue
}
var file_values_types_proto_depIdxs = []int32{
3, // 0: dubbo.operator.v1alpha1.ProxylessConfig.cni:type_name ->
dubbo.operator.v1alpha1.MeshCNIConfig
- 1, // 1: dubbo.operator.v1alpha1.GlobalConfig.proxy:type_name ->
dubbo.operator.v1alpha1.ProxyConfig
- 0, // 2: dubbo.operator.v1alpha1.GlobalConfig.management:type_name ->
dubbo.operator.v1alpha1.ManagementConfig
- 8, // 3: dubbo.operator.v1alpha1.GlobalConfig.multicluster:type_name
-> dubbo.operator.v1alpha1.MulticlusterConfig
- 2, // 4: dubbo.operator.v1alpha1.GlobalConfig.proxyless:type_name ->
dubbo.operator.v1alpha1.ProxylessConfig
- 6, // 5:
dubbo.operator.v1alpha1.EastWestGatewayConfig.gateways:type_name ->
dubbo.operator.v1alpha1.EastWestGatewayEndpoint
- 5, // 6:
dubbo.operator.v1alpha1.MulticlusterConfig.remoteAccess:type_name ->
dubbo.operator.v1alpha1.RemoteAccessConfig
- 7, // 7:
dubbo.operator.v1alpha1.MulticlusterConfig.eastWestGateway:type_name ->
dubbo.operator.v1alpha1.EastWestGatewayConfig
- 4, // 8: dubbo.operator.v1alpha1.Values.global:type_name ->
dubbo.operator.v1alpha1.GlobalConfig
- 11, // 9: dubbo.operator.v1alpha1.IntOrString.intVal:type_name ->
google.protobuf.Int32Value
- 12, // 10: dubbo.operator.v1alpha1.IntOrString.strVal:type_name ->
google.protobuf.StringValue
- 11, // [11:11] is the sub-list for method output_type
- 11, // [11:11] is the sub-list for method input_type
- 11, // [11:11] is the sub-list for extension type_name
- 11, // [11:11] is the sub-list for extension extendee
- 0, // [0:11] is the sub-list for field type_name
+ 12, // 1: dubbo.operator.v1alpha1.GatewayConfig.replicaCount:type_name
-> google.protobuf.Int32Value
+ 1, // 2: dubbo.operator.v1alpha1.GlobalConfig.proxy:type_name ->
dubbo.operator.v1alpha1.ProxyConfig
+ 0, // 3: dubbo.operator.v1alpha1.GlobalConfig.management:type_name ->
dubbo.operator.v1alpha1.ManagementConfig
+ 9, // 4: dubbo.operator.v1alpha1.GlobalConfig.multicluster:type_name
-> dubbo.operator.v1alpha1.MulticlusterConfig
+ 2, // 5: dubbo.operator.v1alpha1.GlobalConfig.proxyless:type_name ->
dubbo.operator.v1alpha1.ProxylessConfig
+ 4, // 6: dubbo.operator.v1alpha1.GlobalConfig.gateway:type_name ->
dubbo.operator.v1alpha1.GatewayConfig
+ 7, // 7:
dubbo.operator.v1alpha1.EastWestGatewayConfig.gateways:type_name ->
dubbo.operator.v1alpha1.EastWestGatewayEndpoint
+ 6, // 8:
dubbo.operator.v1alpha1.MulticlusterConfig.remoteAccess:type_name ->
dubbo.operator.v1alpha1.RemoteAccessConfig
+ 8, // 9:
dubbo.operator.v1alpha1.MulticlusterConfig.eastWestGateway:type_name ->
dubbo.operator.v1alpha1.EastWestGatewayConfig
+ 5, // 10: dubbo.operator.v1alpha1.Values.global:type_name ->
dubbo.operator.v1alpha1.GlobalConfig
+ 12, // 11: dubbo.operator.v1alpha1.Values.replicaCount:type_name ->
google.protobuf.Int32Value
+ 12, // 12: dubbo.operator.v1alpha1.IntOrString.intVal:type_name ->
google.protobuf.Int32Value
+ 13, // 13: dubbo.operator.v1alpha1.IntOrString.strVal:type_name ->
google.protobuf.StringValue
+ 14, // [14:14] is the sub-list for method output_type
+ 14, // [14:14] is the sub-list for method input_type
+ 14, // [14:14] is the sub-list for extension type_name
+ 14, // [14:14] is the sub-list for extension extendee
+ 0, // [0:14] is the sub-list for field type_name
}
func init() { file_values_types_proto_init() }
@@ -916,7 +989,7 @@ func file_values_types_proto_init() {
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor:
unsafe.Slice(unsafe.StringData(file_values_types_proto_rawDesc),
len(file_values_types_proto_rawDesc)),
NumEnums: 0,
- NumMessages: 11,
+ NumMessages: 12,
NumExtensions: 0,
NumServices: 0,
},
diff --git a/operator/pkg/render/manifest_test.go
b/operator/pkg/render/manifest_test.go
index 445c1be5..3a42bf5a 100644
--- a/operator/pkg/render/manifest_test.go
+++ b/operator/pkg/render/manifest_test.go
@@ -33,6 +33,108 @@ func TestGenerateManifestUsesEmbeddedInstallerAssets(t
*testing.T) {
}
}
+// TestGenerateManifestDefaultsToHighlyAvailableControlPlane pins the default
+// posture: a single dubbod makes every node drain or upgrade a control-plane
+// outage, during which no pod can be injected and no xDS update is served.
+func TestGenerateManifestDefaultsToHighlyAvailableControlPlane(t *testing.T) {
+ manifests, _, err := GenerateManifest(nil, nil, nil, nil)
+ if err != nil {
+ t.Fatalf("GenerateManifest() error = %v", err)
+ }
+
+ deployment := findManifest(t, manifests, "Deployment", "dubbod")
+ replicas, ok, err := unstructured.NestedInt64(deployment.Object,
"spec", "replicas")
+ if err != nil || !ok {
+ t.Fatalf("replicas missing: ok=%v err=%v", ok, err)
+ }
+ if replicas < 2 {
+ t.Fatalf("dubbod replicas = %d, want at least 2", replicas)
+ }
+
+ // The budget only protects the control plane if it is actually
rendered at
+ // the default replica count.
+ budget := findManifest(t, manifests, "PodDisruptionBudget", "dubbod")
+ minAvailable, ok, err := unstructured.NestedInt64(budget.Object,
"spec", "minAvailable")
+ if err != nil || !ok {
+ t.Fatalf("minAvailable missing: ok=%v err=%v", ok, err)
+ }
+ if minAvailable != 1 {
+ t.Fatalf("dubbod PodDisruptionBudget minAvailable = %d, want
1", minAvailable)
+ }
+
+ constraints, ok, err := unstructured.NestedSlice(deployment.Object,
"spec", "template", "spec", "topologySpreadConstraints")
+ if err != nil || !ok || len(constraints) == 0 {
+ t.Fatalf("topologySpreadConstraints missing: ok=%v err=%v", ok,
err)
+ }
+ for _, raw := range constraints {
+ constraint := raw.(map[string]interface{})
+ // Spreading must stay advisory, or a single-node cluster
leaves the
+ // second replica permanently Pending.
+ if policy, _, _ := unstructured.NestedString(constraint,
"whenUnsatisfiable"); policy != "ScheduleAnyway" {
+ t.Fatalf("whenUnsatisfiable = %q, want ScheduleAnyway",
policy)
+ }
+ }
+}
+
+// TestGenerateManifestSingleReplicaOmitsDisruptionBudget guards the other
+// direction: a budget of minAvailable 1 over a single replica blocks every
+// voluntary eviction, so node drains hang instead of proceeding.
+func TestGenerateManifestSingleReplicaOmitsDisruptionBudget(t *testing.T) {
+ manifests, _, err := GenerateManifest(nil,
[]string{"values.replicaCount=1"}, nil, nil)
+ if err != nil {
+ t.Fatalf("GenerateManifest() error = %v", err)
+ }
+ for _, set := range manifests {
+ for _, item := range set.Manifests {
+ if item.GetKind() == "PodDisruptionBudget" &&
item.GetName() == "dubbod" {
+ t.Fatal("PodDisruptionBudget rendered for a
single-replica control plane")
+ }
+ }
+ }
+}
+
+// TestGenerateManifestPassesGatewayReplicaDefault walks the whole chain the
+// gateway replica default travels: chart values, the operator API schema, and
+// the environment variable dubbod reads at runtime. A break anywhere in it is
+// silent, because the controller simply falls back to its own constant.
+func TestGenerateManifestPassesGatewayReplicaDefault(t *testing.T) {
+ for _, test := range []struct {
+ name string
+ set []string
+ want string
+ }{
+ {name: "default", want: "2"},
+ {name: "override", set:
[]string{"values.global.gateway.replicaCount=3"}, want: "3"},
+ } {
+ t.Run(test.name, func(t *testing.T) {
+ manifests, _, err := GenerateManifest(nil, test.set,
nil, nil)
+ if err != nil {
+ t.Fatalf("GenerateManifest() error = %v", err)
+ }
+ deployment := findManifest(t, manifests, "Deployment",
"dubbod")
+ containers, ok, err :=
unstructured.NestedSlice(deployment.Object, "spec", "template", "spec",
"containers")
+ if err != nil || !ok || len(containers) == 0 {
+ t.Fatalf("deployment containers missing: ok=%v
err=%v", ok, err)
+ }
+ env, ok, err :=
unstructured.NestedSlice(containers[0].(map[string]interface{}), "env")
+ if err != nil || !ok {
+ t.Fatalf("container env missing: ok=%v err=%v",
ok, err)
+ }
+ for _, raw := range env {
+ entry := raw.(map[string]interface{})
+ if name, _, _ :=
unstructured.NestedString(entry, "name"); name != "DUBBO_DXGATE_REPLICAS" {
+ continue
+ }
+ if value, _, _ :=
unstructured.NestedString(entry, "value"); value != test.want {
+ t.Fatalf("DUBBO_DXGATE_REPLICAS = %q,
want %q", value, test.want)
+ }
+ return
+ }
+ t.Fatal("DUBBO_DXGATE_REPLICAS not rendered")
+ })
+ }
+}
+
func TestTelemetryValidationWebhookDoesNotRequireRevisionLabel(t *testing.T) {
manifests, _, err := GenerateManifest(nil, nil, nil, nil)
if err != nil {
diff --git a/pkg/kube/inject/proxyless.go b/pkg/kube/inject/proxyless.go
index 4b9a6e7f..2468d3c6 100644
--- a/pkg/kube/inject/proxyless.go
+++ b/pkg/kube/inject/proxyless.go
@@ -47,8 +47,12 @@ const (
ProxylessGRPCConfigPath = ProxylessXDSMountPath +
"/" + ProxylessGRPCConfigFileName
ProxylessGRPCInboundContainerName = "dubbo-grpc-inbound"
ProxylessGRPCInboundPort = 15080
- ProxylessManagedLabel =
"proxyless.dubbo.apache.org/managed"
- ProxylessManagedLabelValue = "true"
+ // ProxylessGRPCInboundAdminPort serves the inbound sidecar's health,
+ // readiness and metrics endpoints. It carries no mesh traffic, so the
node
+ // fence exempts it and kubelet can probe it directly.
+ ProxylessGRPCInboundAdminPort = 15020
+ ProxylessManagedLabel = "proxyless.dubbo.apache.org/managed"
+ ProxylessManagedLabelValue = "true"
)
// ProxylessExcludeInboundPortsAnnotation lists inbound ports that stay
diff --git a/pkg/kube/inject/proxyless_test.go
b/pkg/kube/inject/proxyless_test.go
index a1c0b91e..82917e84 100644
--- a/pkg/kube/inject/proxyless_test.go
+++ b/pkg/kube/inject/proxyless_test.go
@@ -21,6 +21,7 @@ import (
"runtime"
"strings"
"testing"
+ "time"
telemetryconfig
"github.com/apache/dubbo-kubernetes/pkg/config/telemetry"
meshv1alpha1 "github.com/kdubbo/api/mesh/v1alpha1"
@@ -243,6 +244,10 @@ func assertNoArgs(t *testing.T, pod *corev1.Pod) {
}
}
+// proxylessDrainDelay mirrors the sidecar's default termination drain delay.
+// The readiness probe must detect termination inside this window.
+const proxylessDrainDelay = 5 * time.Second
+
func assertGRPCInboundContainer(t *testing.T, pod *corev1.Pod) {
t.Helper()
container := FindContainer(ProxylessGRPCInboundContainerName,
pod.Spec.Containers)
@@ -259,6 +264,40 @@ func assertGRPCInboundContainer(t *testing.T, pod
*corev1.Pod) {
if !hasMount(container.VolumeMounts, ProxylessXDSVolumeName,
ProxylessXDSMountPath, true) {
t.Fatalf("grpc-inbound proxyless xds mount missing")
}
+ assertDrainReadinessProbe(t, container)
+}
+
+// assertDrainReadinessProbe checks the probe that withdraws a terminating pod
+// from its EndpointSlice. Without it the sidecar's drain delay is inert:
kubelet
+// never observes the 503, so the endpoint is still published when the listener
+// closes.
+func assertDrainReadinessProbe(t *testing.T, container *corev1.Container) {
+ t.Helper()
+ probe := container.ReadinessProbe
+ if probe == nil || probe.HTTPGet == nil {
+ t.Fatalf("grpc-inbound readiness probe missing")
+ }
+ if probe.HTTPGet.Path != "/readyz" || probe.HTTPGet.Port.IntValue() !=
ProxylessGRPCInboundAdminPort {
+ t.Fatalf("grpc-inbound readiness probe = %s:%v, want
/readyz:%d",
+ probe.HTTPGet.Path, probe.HTTPGet.Port,
ProxylessGRPCInboundAdminPort)
+ }
+ // The probe has to fail before the sidecar stops accepting, otherwise
the
+ // endpoint is withdrawn only after the listener is already gone.
+ if detection :=
time.Duration(probe.PeriodSeconds*probe.FailureThreshold) * time.Second;
detection >= proxylessDrainDelay {
+ t.Fatalf("readiness detection window = %v, want less than the
%v drain delay", detection, proxylessDrainDelay)
+ }
+ if !hasContainerPort(container.Ports, ProxylessGRPCInboundAdminPort) {
+ t.Fatalf("grpc-inbound admin port %d not declared",
ProxylessGRPCInboundAdminPort)
+ }
+}
+
+func hasContainerPort(ports []corev1.ContainerPort, want int) bool {
+ for _, port := range ports {
+ if int(port.ContainerPort) == want {
+ return true
+ }
+ }
+ return false
}
func TestAddApplicationContainerConfigInjectsProxylessGRPCContract(t
*testing.T) {
diff --git a/samples/autoscaling/README.md b/samples/autoscaling/README.md
new file mode 100644
index 00000000..99c6d672
--- /dev/null
+++ b/samples/autoscaling/README.md
@@ -0,0 +1,77 @@
+## autoscaling example
+
+用 KEDA 把队列消费者从 0 弹到 N，同时保持网格的 mTLS 和 xDS 不变。
+
+KEDA 不是本项目的依赖，chart 不安装也不托管它。这里的 `ScaledObject` 是原生 KEDA 资源，
+脱离网格照样有效。
+
+### 前置
+
+```bash
+helm repo add kedacore https://kedacore.github.io/charts
+helm install keda kedacore/keda -n keda --create-namespace
+```
+
+### 部署
+
+```bash
+kubectl create ns autoscaling
+kubectl label ns autoscaling dubbo-injection=enabled
+kubectl apply -f samples/autoscaling/kafka-consumer.yaml
+kubectl apply -f samples/autoscaling/scaledobject.yaml
+
+kubectl -n autoscaling get scaledobject order-consumer
+kubectl -n autoscaling get deploy order-consumer -w
+```
+
+topic 空闲时副本数降到 0，有积压时自动拉起。
+
+### 哪些工作负载可以缩到零
+
+| 类型 | 策略 | 原因 |
+| --- | --- | --- |
+| 队列消费者、定时任务 | 0↔N | 没有入向调用方，副本消失不会让谁的请求落空 |
+| 启动快的无状态 Unary 服务 | 1↔N | 东西向同步激活还没有，缩到 0 后调用方直接失败 |
+| 启动慢的 Java 服务 | 1↔N + 预热 | 冷启动几分钟，弹性来不及 |
+| Streaming、长连接 | 不缩到 0 | 缩容会切断进行中的流 |
+| StatefulSet、强状态服务 | 不缩到 0 | 副本身份和存储不是可丢弃的 |
+
+### 为什么 HTTP/gRPC 服务现在不能缩到零
+
+出向是 proxyless 的：调用方进程内的 gRPC xDS client 直接拿 EDS 端点建连。副本归零时
+dubbod 下发一个空的 CLA（`dubbod/discovery/pkg/xds/endpoints/endpoint_builder.go`），
+调用方立即失败。请求路径上没有任何组件能扣住这个请求去触发扩容 —— sidecar 网格里由
+sidecar 或 activator 承担的那个位置，在这里是空的。
+
+所以对被调用的服务用 `minReplicaCount: 1`。等 dxgate 的 Activator 模式落地后，
+北南向的 Unary 请求才能按需激活。
+
+### 副本数只能有一个归属
+
+`ScaledObject` 一旦生效，`Deployment.spec.replicas` 就交给 KEDA 了。两边都写会让
+KEDA 和 Deployment 控制器来回改同一个数字。`kafka-consumer.yaml` 里因此没有 `replicas` 字段。
+
+同理，不要给受 KEDA 管的 Deployment 再挂一个 HPA。
+
+### 缩容时的排空
+
+被缩掉的 pod 会走注入的 dxplane 排空流程：先 5s 只失败 readiness（让 EndpointSlice 摘掉），
+再关监听并等在途连接最多 25s。`terminationGracePeriodSeconds` 必须大于两者之和，
+样例里设的是 40s。设成默认 30s 也够，但没有余量。
+
+排空是否超预算，看 `dxplane_connections_force_closed_total`：非零说明 25s 不够，
+调 `DUBBO_GRPC_INBOUND_TERMINATION_DRAIN_DURATION`。
+
+### 冷启动的代价
+
+每个新副本都要重新取证书、建 xDS 连接、等首次 CDS/EDS 收敛，这笔开销在 sidecar 方案里
+由常驻 sidecar 摊销掉，proxyless 下每个副本重付。上生产前先量一下从 Pod Running 到
+能收发流量的时间，再决定 `cooldownPeriod` 和 `stabilizationWindowSeconds`。
+
+### 检查高可用配置
+
+```bash
+dubboctl analyze -n autoscaling
+```
+
+会报出单副本的控制面和网关、缺 PodDisruptionBudget、以及副本全落在同一个节点上的情况。
diff --git a/samples/autoscaling/kafka-consumer.yaml
b/samples/autoscaling/kafka-consumer.yaml
new file mode 100644
index 00000000..d24c51d5
--- /dev/null
+++ b/samples/autoscaling/kafka-consumer.yaml
@@ -0,0 +1,59 @@
+# 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.
+
+# A queue consumer: it pulls work from Kafka and calls other mesh services, but
+# nothing calls it. That is what makes it safe to scale to zero — no caller can
+# be left holding a request against an endpoint that no longer exists.
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+ name: order-consumer
+ namespace: autoscaling
+ labels:
+ app: order-consumer
+spec:
+ # No replicas field: KEDA owns the replica count through the ScaledObject.
+ # Setting it here makes two controllers fight over the same number.
+ selector:
+ matchLabels:
+ app: order-consumer
+ template:
+ metadata:
+ labels:
+ app: order-consumer
+ spec:
+ # The inbound sidecar drains in two phases (5s endpoint withdrawal, then
+ # up to 25s for in-flight connections). Anything below 30s here lets
+ # kubelet send SIGKILL mid-drain.
+ terminationGracePeriodSeconds: 40
+ containers:
+ - name: consumer
+ # Replace with your own consumer image.
+ image: kdubbo/order-consumer:latest
+ imagePullPolicy: IfNotPresent
+ env:
+ - name: KAFKA_BOOTSTRAP_SERVERS
+ value: kafka.kafka.svc.cluster.local:9092
+ - name: KAFKA_TOPIC
+ value: orders
+ - name: KAFKA_CONSUMER_GROUP
+ value: order-consumer
+ resources:
+ requests:
+ cpu: 100m
+ memory: 128Mi
+ limits:
+ cpu: 500m
+ memory: 512Mi
diff --git a/samples/autoscaling/scaledobject.yaml
b/samples/autoscaling/scaledobject.yaml
new file mode 100644
index 00000000..3009413d
--- /dev/null
+++ b/samples/autoscaling/scaledobject.yaml
@@ -0,0 +1,51 @@
+# 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.
+
+# KEDA owns the replica count. The mesh does not install, manage or wrap KEDA;
+# these are stock KEDA resources and stay valid without the mesh.
+apiVersion: keda.sh/v1alpha1
+kind: ScaledObject
+metadata:
+ name: order-consumer
+ namespace: autoscaling
+spec:
+ scaleTargetRef:
+ name: order-consumer
+ # Zero is safe here because the consumer has no inbound callers. An
+ # HTTP or gRPC service reached over the mesh must use minReplicaCount: 1
+ # until synchronous activation exists.
+ minReplicaCount: 0
+ maxReplicaCount: 20
+ # Long enough that a burst of short gaps in the topic does not scale the
+ # consumer down into a cold start on the next message.
+ cooldownPeriod: 300
+ pollingInterval: 15
+ advanced:
+ horizontalPodAutoscalerConfig:
+ behavior:
+ scaleDown:
+ # Cold start costs a certificate issue plus an xDS convergence per
+ # replica, so scale down slowly and scale up freely.
+ stabilizationWindowSeconds: 300
+ triggers:
+ - type: kafka
+ metadata:
+ bootstrapServers: kafka.kafka.svc.cluster.local:9092
+ consumerGroup: order-consumer
+ topic: orders
+ lagThreshold: "100"
+ # Required for 0 -> 1: without it KEDA cannot distinguish an idle
+ # topic from an unreachable one and refuses to activate.
+ activationLagThreshold: "1"
