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 077f7974 Complete controller wiring and gateway registration (#992)
077f7974 is described below
commit 077f79748e0f4be3b8732dac801d85391504a6f7
Author: mfordjody <[email protected]>
AuthorDate: Fri Aug 7 21:39:59 2026 +0800
Complete controller wiring and gateway registration (#992)
* Complete controller wiring and gateway registration
Complete controller wiring and gateway registration v2
fix ci
fix ci
fix ci
* fix ci
---
.gitattributes | 13 +
.github/workflows/ci.yml | 17 ++
dubbod/discovery/cmd/app/cmd.go | 4 +
dubbod/discovery/pkg/activation/controller.go | 135 ++++++++++
dubbod/discovery/pkg/activation/demand.go | 55 ++++
dubbod/discovery/pkg/activation/demand_service.go | 100 ++++++++
.../pkg/activation/demand_service_test.go | 270 ++++++++++++++++++++
dubbod/discovery/pkg/activation/demand_test.go | 64 +++++
.../discovery/pkg/activation/demandpb/demand.pb.go | 283 +++++++++++++++++++++
.../discovery/pkg/activation/demandpb/demand.proto | 75 ++++++
.../pkg/activation/demandpb/demand_grpc.pb.go | 162 ++++++++++++
dubbod/discovery/pkg/activation/server.go | 106 ++++++++
dubbod/discovery/pkg/activation/server_test.go | 94 +++++++
dubbod/discovery/pkg/bootstrap/activation.go | 49 ++++
dubbod/discovery/pkg/bootstrap/options.go | 3 +
dubbod/discovery/pkg/bootstrap/server.go | 6 +
manifests/charts/dubbod/templates/clusterrole.yaml | 7 +
manifests/charts/dubbod/templates/deployment.yaml | 21 ++
manifests/charts/dubbod/templates/service.yaml | 55 ++++
manifests/charts/dubbod/values.yaml | 5 +
operator/pkg/apis/proto/values_types.proto | 14 +
operator/pkg/apis/values_types.pb.go | 213 ++++++++++------
operator/pkg/render/manifest_test.go | 88 +++++++
tools/make/common.mk | 10 +
tools/make/lint.mk | 30 +++
25 files changed, 1807 insertions(+), 72 deletions(-)
diff --git a/.gitattributes b/.gitattributes
new file mode 100644
index 00000000..ff76498f
--- /dev/null
+++ b/.gitattributes
@@ -0,0 +1,13 @@
+# Generated files. Marked so they collapse in review diffs and are excluded
+# from language statistics; the source of truth is the .proto / metadata.yaml
+# beside them.
+#
+# They are deliberately left mergeable. Forcing a conflict on every generated
+# file would block a pull request whenever two branches touch the same API,
+# which is exactly when merging is most routine. `make check-generate` is the
+# real guard: it regenerates everything and fails with a diff if the committed
+# output no longer matches, so a bad merge or a hand-edit is caught there
+# rather than becoming a compile error in a file nobody edited.
+*.pb.go linguist-generated=true
+*.gen.go linguist-generated=true
+kubernetes/customresourcedefinitions.gen.yaml linguist-generated=true
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 57d73e96..c0344dc3 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -35,6 +35,10 @@ concurrency:
env:
GOLANGCI_LINT_VERSION: v2.6.2
+ # Must match PROTOC_VERSION in tools/make/common.mk: protoc stamps its own
+ # version into every generated file, so a different one here rewrites them
+ # all and make check-generate can never pass.
+ PROTOC_VERSION: v33.0
jobs:
lint:
@@ -83,6 +87,19 @@ jobs:
- name: Check go.mod/go.sum are tidy
run: make check-tidy
+ # Generated sources are never merged line by line (see .gitattributes);
+ # they are regenerated. This proves the committed output still matches
+ # what the generators produce, so a hand-edited or stale generated file
+ # fails here with a diff instead of later as a compile error.
+ - name: Install protoc
+ uses: arduino/setup-protoc@v3
+ with:
+ version: ${{ env.PROTOC_VERSION }}
+ repo-token: ${{ secrets.GITHUB_TOKEN }}
+
+ - name: Check generated sources are up to date
+ run: make check-generate
+
- name: Check License Header
uses: apache/skywalking-eyes/[email protected]
diff --git a/dubbod/discovery/cmd/app/cmd.go b/dubbod/discovery/cmd/app/cmd.go
index 09e3048d..2147b0c8 100644
--- a/dubbod/discovery/cmd/app/cmd.go
+++ b/dubbod/discovery/cmd/app/cmd.go
@@ -145,6 +145,10 @@ func addFlags(c *cobra.Command) {
"managementAddr",
":26080",
"Management API HTTP address")
+ c.PersistentFlags().StringVar(&serverArgs.ServerOptions.ActivationAddr,
+ "activationAddr",
+ ":26030",
+ "KEDA external scaler gRPC address for on-demand activation;
empty disables it")
c.PersistentFlags().StringVar(&serverArgs.ServerOptions.HTTPSAddr,
"httpsAddr",
":26017",
diff --git a/dubbod/discovery/pkg/activation/controller.go
b/dubbod/discovery/pkg/activation/controller.go
new file mode 100644
index 00000000..1a4bb324
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/controller.go
@@ -0,0 +1,135 @@
+// 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 (
+ "time"
+
+ "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/log"
+ clientnetworking
"github.com/kdubbo/client-go/pkg/apis/networking/v1alpha3"
+ corev1 "k8s.io/api/core/v1"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ klabels "k8s.io/apimachinery/pkg/labels"
+ "k8s.io/apimachinery/pkg/types"
+)
+
+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.
+const resyncInterval = 30 * time.Second
+
+// Controller keeps ServiceActivationPolicy status in step with what the mesh
+// can actually do for that policy.
+//
+// It publishes status only. Replica counts belong to the autoscaler the policy
+// references, and writing them here would put two controllers on the same
+// field.
+type Controller struct {
+ policies kclient.Client[*clientnetworking.ServiceActivationPolicy]
+ services kclient.Client[*corev1.Service]
+ queue controllers.Queue
+
+ evaluator PolicyEvaluator
+}
+
+// NewController wires the policy watch to the live scaler and gateway state.
+func NewController(client kube.Client, streams StreamLookup, reporters
ReporterLookup) *Controller {
+ c := &Controller{
+ policies:
kclient.New[*clientnetworking.ServiceActivationPolicy](client),
+ services: kclient.New[*corev1.Service](client),
+ }
+ c.evaluator = PolicyEvaluator{
+ Services: c,
+ Streams: streams,
+ Reporters: reporters,
+ }
+
+ c.queue = controllers.NewQueue("service activation policy",
+ controllers.WithReconciler(c.Reconcile),
+ controllers.WithMaxAttempts(5))
+
+ c.policies.AddEventHandler(controllers.ObjectHandler(c.queue.AddObject))
+ // 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) {
+ for _, policy := range c.policies.List(o.GetNamespace(),
klabels.Everything()) {
+ c.queue.AddObject(policy)
+ }
+ }))
+
+ return c
+}
+
+// HasService satisfies ServiceLookup from the informer cache.
+func (c *Controller) HasService(namespace, name string) bool {
+ return c.services.Get(name, namespace) != nil
+}
+
+func (c *Controller) Run(stop <-chan struct{}) {
+ kube.WaitForCacheSync("activation controller", stop,
c.policies.HasSynced, c.services.HasSynced)
+
+ go c.resync(stop)
+
+ c.queue.Run(stop)
+ controllers.ShutdownAll(c.policies, c.services)
+}
+
+// resync re-queues every policy on a tick, picking up scaler and gateway
+// changes that Kubernetes never reports.
+func (c *Controller) resync(stop <-chan struct{}) {
+ ticker := time.NewTicker(resyncInterval)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-stop:
+ return
+ case <-ticker.C:
+ for _, policy := range
c.policies.List(metav1.NamespaceAll, klabels.Everything()) {
+ c.queue.AddObject(policy)
+ }
+ }
+ }
+}
+
+func (c *Controller) Reconcile(key types.NamespacedName) error {
+ policy := c.policies.Get(key.Name, key.Namespace)
+ if policy == nil {
+ // Deleted; nothing to publish.
+ return nil
+ }
+
+ conditions := c.evaluator.Evaluate(policy)
+ if SameConditions(policy.Status.GetConditions(), conditions) {
+ // Writing an unchanged status would feed the resync tick back
into
+ // itself and turn a quiet cluster into a steady write load.
+ return nil
+ }
+
+ updated := policy.DeepCopy()
+ updated.Status.Conditions = conditions
+ if _, err := c.policies.UpdateStatus(updated); err != nil {
+ return err
+ }
+
+ logger.Debugf("updated %s/%s: %s", key.Namespace, key.Name,
Summary(conditions))
+ return nil
+}
diff --git a/dubbod/discovery/pkg/activation/demand.go
b/dubbod/discovery/pkg/activation/demand.go
index 5c6e57eb..24d5d95f 100644
--- a/dubbod/discovery/pkg/activation/demand.go
+++ b/dubbod/discovery/pkg/activation/demand.go
@@ -100,6 +100,61 @@ func (r *Registry) Report(reporter string, target Target,
pending int64) {
notify(subscribers, total)
}
+// ReportSnapshot replaces everything a gateway previously reported.
+//
+// A target missing from the snapshot has drained at that gateway, so it must
be
+// cleared rather than left at its last value. Report cannot express that: it
+// only ever speaks about one target, and a gateway with nothing pending would
+// have no message to send.
+func (r *Registry) ReportSnapshot(reporter string, pending map[Target]int64) {
+ type notification struct {
+ subscribers []chan int64
+ total int64
+ }
+
+ r.mu.Lock()
+ now := r.now()
+ touched := map[Target]struct{}{}
+
+ // Drop this reporter from targets it no longer mentions.
+ for target, byReporter := range r.targets {
+ if _, ok := byReporter[reporter]; !ok {
+ continue
+ }
+ if _, still := pending[target]; still {
+ continue
+ }
+ delete(byReporter, reporter)
+ touched[target] = struct{}{}
+ }
+
+ for target, count := range pending {
+ if count < 0 {
+ count = 0
+ }
+ byReporter, ok := r.targets[target]
+ if !ok {
+ byReporter = map[string]report{}
+ r.targets[target] = byReporter
+ }
+ byReporter[reporter] = report{pending: count, received: now}
+ touched[target] = struct{}{}
+ }
+
+ notifications := make([]notification, 0, len(touched))
+ for target := range touched {
+ notifications = append(notifications, notification{
+ subscribers: r.snapshotSubscribersLocked(target),
+ total: r.totalLocked(target),
+ })
+ }
+ r.mu.Unlock()
+
+ for _, item := range notifications {
+ notify(item.subscribers, item.total)
+ }
+}
+
// Forget drops a gateway's reports, for a clean shutdown that should not wait
// out the TTL.
func (r *Registry) Forget(reporter string) {
diff --git a/dubbod/discovery/pkg/activation/demand_service.go
b/dubbod/discovery/pkg/activation/demand_service.go
new file mode 100644
index 00000000..e07b97e0
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/demand_service.go
@@ -0,0 +1,100 @@
+// 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 (
+ "errors"
+ "io"
+ "strings"
+
+
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/activation/demandpb"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
+)
+
+// DemandService receives the demand gateways broadcast to every control-plane
+// replica.
+type DemandService struct {
+ demandpb.UnimplementedActivationDemandServer
+
+ registry *Registry
+}
+
+func NewDemandService(registry *Registry) *DemandService {
+ return &DemandService{registry: registry}
+}
+
+// Report consumes one gateway's snapshot stream until it ends.
+//
+// The stream's lifetime is the gateway's liveness. On any exit the gateway is
+// forgotten immediately rather than left to age out, so a rolling gateway
+// update does not hold a workload scaled up for a full TTL after the old pod
+// is gone.
+func (s *DemandService) Report(stream demandpb.ActivationDemand_ReportServer)
error {
+ reporter := ""
+ var snapshots int64
+
+ defer func() {
+ if reporter != "" {
+ s.registry.Forget(reporter)
+ }
+ }()
+
+ for {
+ snapshot, err := stream.Recv()
+ if errors.Is(err, io.EOF) {
+ return
stream.SendAndClose(&demandpb.ReportSummary{Snapshots: snapshots})
+ }
+ if err != nil {
+ return err
+ }
+
+ name := strings.TrimSpace(snapshot.GetReporter())
+ if name == "" {
+ return status.Error(codes.InvalidArgument, "demand
snapshot is missing a reporter identity")
+ }
+ // Two identities on one stream would leave the first one's
demand
+ // behind with nothing refreshing it.
+ if reporter != "" && name != reporter {
+ return status.Errorf(codes.InvalidArgument,
+ "reporter changed mid-stream from %q to %q",
reporter, name)
+ }
+ reporter = name
+
+ pending, err := targetsOf(snapshot)
+ if err != nil {
+ return err
+ }
+ s.registry.ReportSnapshot(reporter, pending)
+ snapshots++
+ }
+}
+
+func targetsOf(snapshot *demandpb.DemandSnapshot) (map[Target]int64, error) {
+ pending := make(map[Target]int64, len(snapshot.GetTargets()))
+ for _, item := range snapshot.GetTargets() {
+ namespace := strings.TrimSpace(item.GetNamespace())
+ service := strings.TrimSpace(item.GetService())
+ if namespace == "" || service == "" {
+ return nil, status.Errorf(codes.InvalidArgument,
+ "demand target is missing a namespace or
service: %q/%q", namespace, service)
+ }
+ // Summed rather than overwritten: a malformed snapshot that
repeats a
+ // target must not silently discard one of the counts.
+ pending[Target{Namespace: namespace, Name: service}] +=
item.GetPending()
+ }
+ return pending, nil
+}
diff --git a/dubbod/discovery/pkg/activation/demand_service_test.go
b/dubbod/discovery/pkg/activation/demand_service_test.go
new file mode 100644
index 00000000..07227cf3
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/demand_service_test.go
@@ -0,0 +1,270 @@
+// 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 (
+ "context"
+ "net"
+ "testing"
+ "time"
+
+
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/activation/demandpb"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/credentials/insecure"
+ "google.golang.org/grpc/status"
+)
+
+var reviews = Target{Namespace: "app", Name: "reviews"}
+
+func startDemand(t *testing.T, registry *Registry)
demandpb.ActivationDemandClient {
+ t.Helper()
+
+ listener, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatal(err)
+ }
+ server := grpc.NewServer()
+ demandpb.RegisterActivationDemandServer(server,
NewDemandService(registry))
+ go func() { _ = server.Serve(listener) }()
+
+ connection, err := grpc.NewClient(listener.Addr().String(),
+ grpc.WithTransportCredentials(insecure.NewCredentials()))
+ if err != nil {
+ server.Stop()
+ t.Fatal(err)
+ }
+ t.Cleanup(func() {
+ _ = connection.Close()
+ server.Stop()
+ })
+ return demandpb.NewActivationDemandClient(connection)
+}
+
+func snapshot(reporter string, targets ...*demandpb.TargetDemand)
*demandpb.DemandSnapshot {
+ return &demandpb.DemandSnapshot{Reporter: reporter, Targets: targets}
+}
+
+func demandFor(target Target, pending int64) *demandpb.TargetDemand {
+ return &demandpb.TargetDemand{
+ Namespace: target.Namespace,
+ Service: target.Name,
+ Pending: pending,
+ }
+}
+
+func waitFor(t *testing.T, condition func() bool, message string) {
+ t.Helper()
+ deadline := time.Now().Add(2 * time.Second)
+ for time.Now().Before(deadline) {
+ if condition() {
+ return
+ }
+ time.Sleep(time.Millisecond)
+ }
+ t.Fatal(message)
+}
+
+func TestReportStreamFeedsTheRegistry(t *testing.T) {
+ registry := NewRegistry()
+ client := startDemand(t, registry)
+
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+
+ stream, err := client.Report(ctx)
+ if err != nil {
+ t.Fatalf("Report() error = %v", err)
+ }
+ if err := stream.Send(snapshot("gateway-a", demandFor(orders, 3),
demandFor(reviews, 1))); err != nil {
+ t.Fatalf("Send() error = %v", err)
+ }
+ waitFor(t, func() bool { return registry.Pending(orders) == 3 &&
registry.Pending(reviews) == 1 },
+ "registry did not pick up the first snapshot")
+
+ // A target dropped from the snapshot has drained: there is no separate
+ // clear message, so its absence has to be what clears it.
+ if err := stream.Send(snapshot("gateway-a", demandFor(orders, 2))); err
!= nil {
+ t.Fatalf("Send() error = %v", err)
+ }
+ waitFor(t, func() bool { return registry.Pending(orders) == 2 &&
registry.Pending(reviews) == 0 },
+ "omitted target was not cleared")
+
+ summary, err := stream.CloseAndRecv()
+ if err != nil {
+ t.Fatalf("CloseAndRecv() error = %v", err)
+ }
+ if summary.GetSnapshots() != 2 {
+ t.Fatalf("snapshots = %d, want 2", summary.GetSnapshots())
+ }
+}
+
+// The stream is the gateway's liveness. When it ends the demand has to go with
+// it, or a rolling gateway update holds the workload up for a whole TTL after
+// the old pod is gone.
+func TestClosingTheStreamForgetsTheGateway(t *testing.T) {
+ registry := NewRegistry()
+ client := startDemand(t, registry)
+
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+
+ stream, err := client.Report(ctx)
+ if err != nil {
+ t.Fatalf("Report() error = %v", err)
+ }
+ if err := stream.Send(snapshot("gateway-a", demandFor(orders, 5))); err
!= nil {
+ t.Fatalf("Send() error = %v", err)
+ }
+ waitFor(t, func() bool { return registry.Pending(orders) == 5 },
"snapshot was not recorded")
+
+ if _, err := stream.CloseAndRecv(); err != nil {
+ t.Fatalf("CloseAndRecv() error = %v", err)
+ }
+ waitFor(t, func() bool { return registry.Pending(orders) == 0 },
+ "demand survived the stream closing")
+}
+
+// Every replica gets the same broadcast, so two gateways reporting the same
+// target must add up rather than overwrite one another.
+func TestReportsFromSeveralGatewaysAccumulate(t *testing.T) {
+ registry := NewRegistry()
+ client := startDemand(t, registry)
+
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+
+ first, err := client.Report(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ second, err := client.Report(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ if err := first.Send(snapshot("gateway-a", demandFor(orders, 2))); err
!= nil {
+ t.Fatal(err)
+ }
+ if err := second.Send(snapshot("gateway-b", demandFor(orders, 3))); err
!= nil {
+ t.Fatal(err)
+ }
+ waitFor(t, func() bool { return registry.Pending(orders) == 5 },
+ "reports from two gateways did not accumulate")
+
+ if _, err := first.CloseAndRecv(); err != nil {
+ t.Fatal(err)
+ }
+ waitFor(t, func() bool { return registry.Pending(orders) == 3 },
+ "closing one gateway removed more than its own demand")
+}
+
+func TestReportRejectsMalformedSnapshots(t *testing.T) {
+ tests := []struct {
+ name string
+ send *demandpb.DemandSnapshot
+ }{
+ {name: "no reporter", send: snapshot("", demandFor(orders, 1))},
+ {name: "blank reporter", send: snapshot(" ",
demandFor(orders, 1))},
+ {
+ name: "target without namespace",
+ send: snapshot("gateway-a",
&demandpb.TargetDemand{Service: "orders", Pending: 1}),
+ },
+ {
+ name: "target without service",
+ send: snapshot("gateway-a",
&demandpb.TargetDemand{Namespace: "app", Pending: 1}),
+ },
+ }
+
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ client := startDemand(t, NewRegistry())
+ ctx, cancel :=
context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+
+ stream, err := client.Report(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Send may or may not observe the error depending on
when the
+ // server rejects it; CloseAndRecv is where it always
surfaces.
+ _ = stream.Send(test.send)
+ if _, err := stream.CloseAndRecv(); status.Code(err) !=
codes.InvalidArgument {
+ t.Fatalf("CloseAndRecv() error = %v, want
InvalidArgument", err)
+ }
+ })
+ }
+}
+
+// One stream is one gateway. Allowing a second identity would strand the
+// first one's demand with nothing left to refresh it.
+func TestReportRejectsAChangedReporter(t *testing.T) {
+ client := startDemand(t, NewRegistry())
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+
+ stream, err := client.Report(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := stream.Send(snapshot("gateway-a", demandFor(orders, 1))); err
!= nil {
+ t.Fatal(err)
+ }
+ _ = stream.Send(snapshot("gateway-b", demandFor(orders, 1)))
+
+ if _, err := stream.CloseAndRecv(); status.Code(err) !=
codes.InvalidArgument {
+ t.Fatalf("CloseAndRecv() error = %v, want InvalidArgument", err)
+ }
+}
+
+// Demand reported to this replica has to be visible to the KEDA stream this
+// replica is serving; that is the whole reason both live on one listener.
+func TestReportedDemandActivatesTheScalerOnTheSameReplica(t *testing.T) {
+ server := NewServer()
+
+ listener, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatal(err)
+ }
+ address := listener.Addr().String()
+ _ = listener.Close()
+
+ stop := make(chan struct{})
+ if err := server.Serve(address, stop); err != nil {
+ t.Fatalf("Serve() error = %v", err)
+ }
+ t.Cleanup(func() { close(stop) })
+
+ connection, err := grpc.NewClient(address,
grpc.WithTransportCredentials(insecure.NewCredentials()))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { _ = connection.Close() })
+
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+
+ stream, err :=
demandpb.NewActivationDemandClient(connection).Report(ctx)
+ if err != nil {
+ t.Fatalf("Report() error = %v", err)
+ }
+ if err := stream.Send(snapshot("gateway-a", demandFor(orders, 1))); err
!= nil {
+ t.Fatal(err)
+ }
+
+ waitFor(t, func() bool { return server.Registry().Pending(orders) == 1
},
+ "demand did not reach the registry")
+}
diff --git a/dubbod/discovery/pkg/activation/demand_test.go
b/dubbod/discovery/pkg/activation/demand_test.go
index 4b661c8b..9d5bac43 100644
--- a/dubbod/discovery/pkg/activation/demand_test.go
+++ b/dubbod/discovery/pkg/activation/demand_test.go
@@ -215,3 +215,67 @@ func receive(t *testing.T, updates <-chan int64) int64 {
return 0
}
}
+
+func TestReportSnapshotReplacesTheWholeReporterView(t *testing.T) {
+ registry := NewRegistry()
+ reviews := Target{Namespace: "app", Name: "reviews"}
+
+ registry.ReportSnapshot("gateway-a", map[Target]int64{orders: 3,
reviews: 2})
+ if got := registry.Pending(orders); got != 3 {
+ t.Fatalf("orders pending = %d, want 3", got)
+ }
+ if got := registry.Pending(reviews); got != 2 {
+ t.Fatalf("reviews pending = %d, want 2", got)
+ }
+
+ // A target absent from the new snapshot has drained; there is no
separate
+ // clear message, so its absence is what has to clear it.
+ registry.ReportSnapshot("gateway-a", map[Target]int64{orders: 1})
+ if got := registry.Pending(orders); got != 1 {
+ t.Fatalf("orders pending after replacement = %d, want 1", got)
+ }
+ if got := registry.Pending(reviews); got != 0 {
+ t.Fatalf("reviews pending after omission = %d, want 0", got)
+ }
+
+ // An empty snapshot means the gateway drained everything.
+ registry.ReportSnapshot("gateway-a", nil)
+ if got := registry.Pending(orders); got != 0 {
+ t.Fatalf("orders pending after empty snapshot = %d, want 0",
got)
+ }
+}
+
+// Snapshots replace only the reporting gateway's own view; a broadcast from
+// one gateway must never clear another's demand.
+func TestReportSnapshotLeavesOtherReportersAlone(t *testing.T) {
+ registry := NewRegistry()
+
+ registry.ReportSnapshot("gateway-a", map[Target]int64{orders: 2})
+ registry.ReportSnapshot("gateway-b", map[Target]int64{orders: 3})
+ if got := registry.Pending(orders); got != 5 {
+ t.Fatalf("pending = %d, want 5", got)
+ }
+
+ registry.ReportSnapshot("gateway-a", nil)
+ if got := registry.Pending(orders); got != 3 {
+ t.Fatalf("pending after gateway-a drained = %d, want 3", got)
+ }
+}
+
+func TestReportSnapshotNotifiesSubscribers(t *testing.T) {
+ registry := NewRegistry()
+ updates, cancel := registry.Subscribe(orders)
+ defer cancel()
+
+ registry.ReportSnapshot("gateway-a", map[Target]int64{orders: 4})
+ if got := receive(t, updates); got != 4 {
+ t.Fatalf("update = %d, want 4", got)
+ }
+
+ // Dropping the target must wake the subscriber too, or a KEDA stream
would
+ // never learn the workload can scale back down.
+ registry.ReportSnapshot("gateway-a", nil)
+ if got := receive(t, updates); got != 0 {
+ t.Fatalf("drain update = %d, want 0", got)
+ }
+}
diff --git a/dubbod/discovery/pkg/activation/demandpb/demand.pb.go
b/dubbod/discovery/pkg/activation/demandpb/demand.pb.go
new file mode 100644
index 00000000..31a945dc
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/demandpb/demand.pb.go
@@ -0,0 +1,283 @@
+// 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.
+//
+// Regenerate with:
+//
+// protoc -I . --go_out=. --go_opt=paths=source_relative \
+// --go-grpc_out=. --go-grpc_opt=paths=source_relative demand.proto
+
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// protoc-gen-go v1.36.11
+// protoc v6.33.0
+// source: demand.proto
+
+package demandpb
+
+import (
+ protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+ protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+ reflect "reflect"
+ sync "sync"
+ unsafe "unsafe"
+)
+
+const (
+ // Verify that this generated code is sufficiently up-to-date.
+ _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+ // Verify that runtime/protoimpl is sufficiently up-to-date.
+ _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+// DemandSnapshot is the complete set of targets one gateway is holding
+// requests for. It replaces that gateway's previous snapshot rather than
+// adding to it, which is what lets a lost or duplicated message converge on
+// the next send instead of skewing the total permanently.
+type DemandSnapshot struct {
+ state protoimpl.MessageState `protogen:"open.v1"`
+ // Stable identity of the reporting gateway, unique across replicas.
Pod name
+ // is the expected value; two gateways sharing an identity would
overwrite
+ // each other's counts.
+ Reporter string `protobuf:"bytes,1,opt,name=reporter,proto3"
json:"reporter,omitempty"`
+ // Targets with requests waiting. A target absent from a snapshot has no
+ // pending requests at that gateway; there is no separate clear message.
+ Targets []*TargetDemand
`protobuf:"bytes,2,rep,name=targets,proto3" json:"targets,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
+}
+
+func (x *DemandSnapshot) Reset() {
+ *x = DemandSnapshot{}
+ mi := &file_demand_proto_msgTypes[0]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+}
+
+func (x *DemandSnapshot) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*DemandSnapshot) ProtoMessage() {}
+
+func (x *DemandSnapshot) ProtoReflect() protoreflect.Message {
+ mi := &file_demand_proto_msgTypes[0]
+ 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 DemandSnapshot.ProtoReflect.Descriptor instead.
+func (*DemandSnapshot) Descriptor() ([]byte, []int) {
+ return file_demand_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *DemandSnapshot) GetReporter() string {
+ if x != nil {
+ return x.Reporter
+ }
+ return ""
+}
+
+func (x *DemandSnapshot) GetTargets() []*TargetDemand {
+ if x != nil {
+ return x.Targets
+ }
+ return nil
+}
+
+type TargetDemand struct {
+ state protoimpl.MessageState `protogen:"open.v1"`
+ // Namespace of the Service being activated.
+ Namespace string `protobuf:"bytes,1,opt,name=namespace,proto3"
json:"namespace,omitempty"`
+ // Name of the Service being activated.
+ Service string `protobuf:"bytes,2,opt,name=service,proto3"
json:"service,omitempty"`
+ // Requests this gateway is currently holding for the target.
+ Pending int64 `protobuf:"varint,3,opt,name=pending,proto3"
json:"pending,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
+}
+
+func (x *TargetDemand) Reset() {
+ *x = TargetDemand{}
+ mi := &file_demand_proto_msgTypes[1]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+}
+
+func (x *TargetDemand) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*TargetDemand) ProtoMessage() {}
+
+func (x *TargetDemand) ProtoReflect() protoreflect.Message {
+ mi := &file_demand_proto_msgTypes[1]
+ 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 TargetDemand.ProtoReflect.Descriptor instead.
+func (*TargetDemand) Descriptor() ([]byte, []int) {
+ return file_demand_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *TargetDemand) GetNamespace() string {
+ if x != nil {
+ return x.Namespace
+ }
+ return ""
+}
+
+func (x *TargetDemand) GetService() string {
+ if x != nil {
+ return x.Service
+ }
+ return ""
+}
+
+func (x *TargetDemand) GetPending() int64 {
+ if x != nil {
+ return x.Pending
+ }
+ return 0
+}
+
+// ReportSummary is returned once the stream closes. It exists so a gateway can
+// tell an orderly shutdown from a connection that was cut.
+type ReportSummary struct {
+ state protoimpl.MessageState `protogen:"open.v1"`
+ // Number of snapshots accepted on the stream.
+ Snapshots int64 `protobuf:"varint,1,opt,name=snapshots,proto3"
json:"snapshots,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
+}
+
+func (x *ReportSummary) Reset() {
+ *x = ReportSummary{}
+ mi := &file_demand_proto_msgTypes[2]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+}
+
+func (x *ReportSummary) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ReportSummary) ProtoMessage() {}
+
+func (x *ReportSummary) ProtoReflect() protoreflect.Message {
+ mi := &file_demand_proto_msgTypes[2]
+ 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 ReportSummary.ProtoReflect.Descriptor instead.
+func (*ReportSummary) Descriptor() ([]byte, []int) {
+ return file_demand_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *ReportSummary) GetSnapshots() int64 {
+ if x != nil {
+ return x.Snapshots
+ }
+ return 0
+}
+
+var File_demand_proto protoreflect.FileDescriptor
+
+const file_demand_proto_rawDesc = "" +
+ "\n" +
+ "\fdemand.proto\x12\x19dubbo.activation.v1alpha1\"o\n" +
+ "\x0eDemandSnapshot\x12\x1a\n" +
+ "\breporter\x18\x01 \x01(\tR\breporter\x12A\n" +
+ "\atargets\x18\x02
\x03(\v2'.dubbo.activation.v1alpha1.TargetDemandR\atargets\"`\n" +
+ "\fTargetDemand\x12\x1c\n" +
+ "\tnamespace\x18\x01 \x01(\tR\tnamespace\x12\x18\n" +
+ "\aservice\x18\x02 \x01(\tR\aservice\x12\x18\n" +
+ "\apending\x18\x03 \x01(\x03R\apending\"-\n" +
+ "\rReportSummary\x12\x1c\n" +
+ "\tsnapshots\x18\x01 \x01(\x03R\tsnapshots2u\n" +
+ "\x10ActivationDemand\x12a\n" +
+
"\x06Report\x12).dubbo.activation.v1alpha1.DemandSnapshot\x1a(.dubbo.activation.v1alpha1.ReportSummary\"\x00(\x01BMZKgithub.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/activation/demandpbb\x06proto3"
+
+var (
+ file_demand_proto_rawDescOnce sync.Once
+ file_demand_proto_rawDescData []byte
+)
+
+func file_demand_proto_rawDescGZIP() []byte {
+ file_demand_proto_rawDescOnce.Do(func() {
+ file_demand_proto_rawDescData =
protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_demand_proto_rawDesc),
len(file_demand_proto_rawDesc)))
+ })
+ return file_demand_proto_rawDescData
+}
+
+var file_demand_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
+var file_demand_proto_goTypes = []any{
+ (*DemandSnapshot)(nil), // 0: dubbo.activation.v1alpha1.DemandSnapshot
+ (*TargetDemand)(nil), // 1: dubbo.activation.v1alpha1.TargetDemand
+ (*ReportSummary)(nil), // 2: dubbo.activation.v1alpha1.ReportSummary
+}
+var file_demand_proto_depIdxs = []int32{
+ 1, // 0: dubbo.activation.v1alpha1.DemandSnapshot.targets:type_name ->
dubbo.activation.v1alpha1.TargetDemand
+ 0, // 1: dubbo.activation.v1alpha1.ActivationDemand.Report:input_type
-> dubbo.activation.v1alpha1.DemandSnapshot
+ 2, // 2: dubbo.activation.v1alpha1.ActivationDemand.Report:output_type
-> dubbo.activation.v1alpha1.ReportSummary
+ 2, // [2:3] is the sub-list for method output_type
+ 1, // [1:2] is the sub-list for method input_type
+ 1, // [1:1] is the sub-list for extension type_name
+ 1, // [1:1] is the sub-list for extension extendee
+ 0, // [0:1] is the sub-list for field type_name
+}
+
+func init() { file_demand_proto_init() }
+func file_demand_proto_init() {
+ if File_demand_proto != nil {
+ return
+ }
+ type x struct{}
+ out := protoimpl.TypeBuilder{
+ File: protoimpl.DescBuilder{
+ GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+ RawDescriptor:
unsafe.Slice(unsafe.StringData(file_demand_proto_rawDesc),
len(file_demand_proto_rawDesc)),
+ NumEnums: 0,
+ NumMessages: 3,
+ NumExtensions: 0,
+ NumServices: 1,
+ },
+ GoTypes: file_demand_proto_goTypes,
+ DependencyIndexes: file_demand_proto_depIdxs,
+ MessageInfos: file_demand_proto_msgTypes,
+ }.Build()
+ File_demand_proto = out.File
+ file_demand_proto_goTypes = nil
+ file_demand_proto_depIdxs = nil
+}
diff --git a/dubbod/discovery/pkg/activation/demandpb/demand.proto
b/dubbod/discovery/pkg/activation/demandpb/demand.proto
new file mode 100644
index 00000000..4f2604c6
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/demandpb/demand.proto
@@ -0,0 +1,75 @@
+// 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.
+//
+// Regenerate with:
+//
+// protoc -I . --go_out=. --go_opt=paths=source_relative \
+// --go-grpc_out=. --go-grpc_opt=paths=source_relative demand.proto
+
+syntax = "proto3";
+
+package dubbo.activation.v1alpha1;
+
+option go_package =
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/activation/demandpb";
+
+// ActivationDemand carries the requests a gateway is holding for Services that
+// have no endpoints, so the control plane can tell KEDA to start them.
+//
+// A gateway must report to every control-plane replica, not just one. KEDA
+// opens its activation stream against a single replica chosen by the Service,
+// and a report that only reached a different replica would leave those
requests
+// waiting for a scale-up nobody asked for. Reports are idempotent snapshots
+// precisely so broadcasting them is safe.
+service ActivationDemand {
+ // Report streams demand snapshots for the lifetime of the gateway.
+ //
+ // The stream itself is the liveness signal: when it ends, the control plane
+ // drops that gateway's demand immediately instead of waiting for it to age
+ // out, so a gateway that shuts down cleanly cannot hold a workload up.
+ rpc Report(stream DemandSnapshot) returns (ReportSummary) {}
+}
+
+// DemandSnapshot is the complete set of targets one gateway is holding
+// requests for. It replaces that gateway's previous snapshot rather than
+// adding to it, which is what lets a lost or duplicated message converge on
+// the next send instead of skewing the total permanently.
+message DemandSnapshot {
+ // Stable identity of the reporting gateway, unique across replicas. Pod name
+ // is the expected value; two gateways sharing an identity would overwrite
+ // each other's counts.
+ string reporter = 1;
+
+ // Targets with requests waiting. A target absent from a snapshot has no
+ // pending requests at that gateway; there is no separate clear message.
+ repeated TargetDemand targets = 2;
+}
+
+message TargetDemand {
+ // Namespace of the Service being activated.
+ string namespace = 1;
+
+ // Name of the Service being activated.
+ string service = 2;
+
+ // Requests this gateway is currently holding for the target.
+ int64 pending = 3;
+}
+
+// ReportSummary is returned once the stream closes. It exists so a gateway can
+// tell an orderly shutdown from a connection that was cut.
+message ReportSummary {
+ // Number of snapshots accepted on the stream.
+ int64 snapshots = 1;
+}
diff --git a/dubbod/discovery/pkg/activation/demandpb/demand_grpc.pb.go
b/dubbod/discovery/pkg/activation/demandpb/demand_grpc.pb.go
new file mode 100644
index 00000000..b08cfb1a
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/demandpb/demand_grpc.pb.go
@@ -0,0 +1,162 @@
+// 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.
+//
+// Regenerate with:
+//
+// protoc -I . --go_out=. --go_opt=paths=source_relative \
+// --go-grpc_out=. --go-grpc_opt=paths=source_relative demand.proto
+
+// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.6.2
+// - protoc v6.33.0
+// source: demand.proto
+
+package demandpb
+
+import (
+ context "context"
+ grpc "google.golang.org/grpc"
+ codes "google.golang.org/grpc/codes"
+ status "google.golang.org/grpc/status"
+)
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+// Requires gRPC-Go v1.64.0 or later.
+const _ = grpc.SupportPackageIsVersion9
+
+const (
+ ActivationDemand_Report_FullMethodName =
"/dubbo.activation.v1alpha1.ActivationDemand/Report"
+)
+
+// ActivationDemandClient is the client API for ActivationDemand service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please
refer to
https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
+//
+// ActivationDemand carries the requests a gateway is holding for Services that
+// have no endpoints, so the control plane can tell KEDA to start them.
+//
+// A gateway must report to every control-plane replica, not just one. KEDA
+// opens its activation stream against a single replica chosen by the Service,
+// and a report that only reached a different replica would leave those
requests
+// waiting for a scale-up nobody asked for. Reports are idempotent snapshots
+// precisely so broadcasting them is safe.
+type ActivationDemandClient interface {
+ // Report streams demand snapshots for the lifetime of the gateway.
+ //
+ // The stream itself is the liveness signal: when it ends, the control
plane
+ // drops that gateway's demand immediately instead of waiting for it to
age
+ // out, so a gateway that shuts down cleanly cannot hold a workload up.
+ Report(ctx context.Context, opts ...grpc.CallOption)
(grpc.ClientStreamingClient[DemandSnapshot, ReportSummary], error)
+}
+
+type activationDemandClient struct {
+ cc grpc.ClientConnInterface
+}
+
+func NewActivationDemandClient(cc grpc.ClientConnInterface)
ActivationDemandClient {
+ return &activationDemandClient{cc}
+}
+
+func (c *activationDemandClient) Report(ctx context.Context, opts
...grpc.CallOption) (grpc.ClientStreamingClient[DemandSnapshot, ReportSummary],
error) {
+ cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+ stream, err := c.cc.NewStream(ctx,
&ActivationDemand_ServiceDesc.Streams[0],
ActivationDemand_Report_FullMethodName, cOpts...)
+ if err != nil {
+ return nil, err
+ }
+ x := &grpc.GenericClientStream[DemandSnapshot,
ReportSummary]{ClientStream: stream}
+ return x, nil
+}
+
+// This type alias is provided for backwards compatibility with existing code
that references the prior non-generic stream type by name.
+type ActivationDemand_ReportClient =
grpc.ClientStreamingClient[DemandSnapshot, ReportSummary]
+
+// ActivationDemandServer is the server API for ActivationDemand service.
+// All implementations must embed UnimplementedActivationDemandServer
+// for forward compatibility.
+//
+// ActivationDemand carries the requests a gateway is holding for Services that
+// have no endpoints, so the control plane can tell KEDA to start them.
+//
+// A gateway must report to every control-plane replica, not just one. KEDA
+// opens its activation stream against a single replica chosen by the Service,
+// and a report that only reached a different replica would leave those
requests
+// waiting for a scale-up nobody asked for. Reports are idempotent snapshots
+// precisely so broadcasting them is safe.
+type ActivationDemandServer interface {
+ // Report streams demand snapshots for the lifetime of the gateway.
+ //
+ // The stream itself is the liveness signal: when it ends, the control
plane
+ // drops that gateway's demand immediately instead of waiting for it to
age
+ // out, so a gateway that shuts down cleanly cannot hold a workload up.
+ Report(grpc.ClientStreamingServer[DemandSnapshot, ReportSummary]) error
+ mustEmbedUnimplementedActivationDemandServer()
+}
+
+// UnimplementedActivationDemandServer must be embedded to have
+// forward compatible implementations.
+//
+// NOTE: this should be embedded by value instead of pointer to avoid a nil
+// pointer dereference when methods are called.
+type UnimplementedActivationDemandServer struct{}
+
+func (UnimplementedActivationDemandServer)
Report(grpc.ClientStreamingServer[DemandSnapshot, ReportSummary]) error {
+ return status.Error(codes.Unimplemented, "method Report not
implemented")
+}
+func (UnimplementedActivationDemandServer)
mustEmbedUnimplementedActivationDemandServer() {}
+func (UnimplementedActivationDemandServer) testEmbeddedByValue()
{}
+
+// UnsafeActivationDemandServer may be embedded to opt out of forward
compatibility for this service.
+// Use of this interface is not recommended, as added methods to
ActivationDemandServer will
+// result in compilation errors.
+type UnsafeActivationDemandServer interface {
+ mustEmbedUnimplementedActivationDemandServer()
+}
+
+func RegisterActivationDemandServer(s grpc.ServiceRegistrar, srv
ActivationDemandServer) {
+ // If the following call panics, it indicates
UnimplementedActivationDemandServer was
+ // embedded by pointer and is nil. This will cause panics if an
+ // unimplemented method is ever invoked, so we test this at
initialization
+ // time to prevent it from happening at runtime later due to I/O.
+ if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
+ t.testEmbeddedByValue()
+ }
+ s.RegisterService(&ActivationDemand_ServiceDesc, srv)
+}
+
+func _ActivationDemand_Report_Handler(srv interface{}, stream
grpc.ServerStream) error {
+ return
srv.(ActivationDemandServer).Report(&grpc.GenericServerStream[DemandSnapshot,
ReportSummary]{ServerStream: stream})
+}
+
+// This type alias is provided for backwards compatibility with existing code
that references the prior non-generic stream type by name.
+type ActivationDemand_ReportServer =
grpc.ClientStreamingServer[DemandSnapshot, ReportSummary]
+
+// ActivationDemand_ServiceDesc is the grpc.ServiceDesc for ActivationDemand
service.
+// It's only intended for direct use with grpc.RegisterService,
+// and not to be introspected or modified (even as a copy)
+var ActivationDemand_ServiceDesc = grpc.ServiceDesc{
+ ServiceName: "dubbo.activation.v1alpha1.ActivationDemand",
+ HandlerType: (*ActivationDemandServer)(nil),
+ Methods: []grpc.MethodDesc{},
+ Streams: []grpc.StreamDesc{
+ {
+ StreamName: "Report",
+ Handler: _ActivationDemand_Report_Handler,
+ ClientStreams: true,
+ },
+ },
+ Metadata: "demand.proto",
+}
diff --git a/dubbod/discovery/pkg/activation/server.go
b/dubbod/discovery/pkg/activation/server.go
new file mode 100644
index 00000000..6b8b0d1b
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/server.go
@@ -0,0 +1,106 @@
+// 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 (
+ "fmt"
+ "net"
+ "time"
+
+
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/activation/demandpb"
+
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/activation/externalscaler"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/keepalive"
+)
+
+// keepaliveOptions hold the StreamIsActive connections open through idle
+// periods. Those streams carry nothing between activations, and a middlebox
+// that drops idle connections would silently stop KEDA from ever hearing that
+// a request arrived.
+var keepaliveOptions = keepalive.ServerParameters{
+ Time: 30 * time.Second,
+ Timeout: 10 * time.Second,
+}
+
+// enforcementPolicy accepts the pings KEDA's client sends on an otherwise idle
+// activation stream. Without permitting them the server would close the very
+// connections it needs to keep.
+var enforcementPolicy = keepalive.EnforcementPolicy{
+ MinTime: 10 * time.Second,
+ PermitWithoutStream: true,
+}
+
+// Server is the KEDA-facing endpoint: a gRPC service KEDA dials for every
+// ScaledObject whose external trigger points at this control plane.
+type Server struct {
+ scaler *Scaler
+ registry *Registry
+ grpc *grpc.Server
+}
+
+// NewServer builds the activation endpoint and the demand registry behind it.
+func NewServer() *Server {
+ registry := NewRegistry()
+ scaler := NewScaler(registry)
+
+ server := grpc.NewServer(
+ grpc.KeepaliveParams(keepaliveOptions),
+ grpc.KeepaliveEnforcementPolicy(enforcementPolicy),
+ )
+ externalscaler.RegisterExternalScalerServer(server, scaler)
+ // Gateways report into the same registry the scaler reads, on the same
+ // listener: a gateway that can reach this replica can always be heard
by
+ // the KEDA stream this replica is serving.
+ demandpb.RegisterActivationDemandServer(server,
NewDemandService(registry))
+
+ return &Server{scaler: scaler, registry: registry, grpc: server}
+}
+
+// Scaler exposes the KEDA subscription state the policy controller reports.
+func (s *Server) Scaler() *Scaler { return s.scaler }
+
+// Registry exposes the demand store gateways report into.
+func (s *Server) Registry() *Registry { return s.registry }
+
+// Serve listens on addr until stop is closed. An empty address disables the
+// endpoint, which is how a cluster without KEDA installed runs unchanged.
+func (s *Server) Serve(addr string, stop <-chan struct{}) error {
+ if addr == "" {
+ logger.Info("activation scaler disabled; no listen address
configured")
+ return nil
+ }
+
+ listener, err := net.Listen("tcp", addr)
+ if err != nil {
+ return fmt.Errorf("unable to listen on activation scaler
socket: %v", err)
+ }
+
+ go func() {
+ logger.Infof("starting activation scaler at %s",
listener.Addr())
+ if err := s.grpc.Serve(listener); err != nil {
+ logger.Errorf("error serving activation scaler: %v",
err)
+ }
+ }()
+
+ go func() {
+ <-stop
+ // Graceful: an activation stream that is mid-send is the only
way KEDA
+ // learns about a waiting request, so it is worth letting it
finish.
+ s.grpc.GracefulStop()
+ }()
+
+ return nil
+}
diff --git a/dubbod/discovery/pkg/activation/server_test.go
b/dubbod/discovery/pkg/activation/server_test.go
new file mode 100644
index 00000000..62e81472
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/server_test.go
@@ -0,0 +1,94 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (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 (
+ "context"
+ "net"
+ "testing"
+ "time"
+
+
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/activation/externalscaler"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/credentials/insecure"
+)
+
+// An empty address is how a cluster without KEDA runs unchanged: no listener,
+// no error, and the rest of the control plane comes up as usual.
+func TestServeWithoutAddressIsANoOp(t *testing.T) {
+ server := NewServer()
+ stop := make(chan struct{})
+ defer close(stop)
+
+ if err := server.Serve("", stop); err != nil {
+ t.Fatalf("Serve() with no address error = %v", err)
+ }
+}
+
+func TestServeReportsAnUnusableAddress(t *testing.T) {
+ // Occupy a port, then ask the scaler for the same one.
+ taken, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer func() { _ = taken.Close() }()
+
+ server := NewServer()
+ stop := make(chan struct{})
+ defer close(stop)
+
+ if err := server.Serve(taken.Addr().String(), stop); err == nil {
+ t.Fatal("Serve() on an occupied address returned no error")
+ }
+}
+
+// The registry the gateways report into and the scaler KEDA dials have to be
+// the same one, or demand would be recorded where nothing reads it.
+func TestServerSharesOneRegistryBetweenReportsAndScaler(t *testing.T) {
+ server := NewServer()
+
+ listener, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatal(err)
+ }
+ address := listener.Addr().String()
+ _ = listener.Close()
+
+ stop := make(chan struct{})
+ if err := server.Serve(address, stop); err != nil {
+ t.Fatalf("Serve() error = %v", err)
+ }
+ t.Cleanup(func() { close(stop) })
+
+ connection, err := grpc.NewClient(address,
grpc.WithTransportCredentials(insecure.NewCredentials()))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { _ = connection.Close() })
+ client := externalscaler.NewExternalScalerClient(connection)
+
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+
+ server.Registry().Report("gateway-a", orders, 2)
+ active, err := client.IsActive(ctx, serviceRef())
+ if err != nil {
+ t.Fatalf("IsActive() error = %v", err)
+ }
+ if !active.GetResult() {
+ t.Fatal("IsActive() = false after demand was reported to the
server registry")
+ }
+}
diff --git a/dubbod/discovery/pkg/bootstrap/activation.go
b/dubbod/discovery/pkg/bootstrap/activation.go
new file mode 100644
index 00000000..08bd9b25
--- /dev/null
+++ b/dubbod/discovery/pkg/bootstrap/activation.go
@@ -0,0 +1,49 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package bootstrap
+
+import (
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/activation"
+ "github.com/apache/dubbo-kubernetes/pkg/log"
+)
+
+// initActivation starts the KEDA-facing scaler and the policy controller that
+// reports what it can actually do.
+//
+// Both are optional. Without a Kubernetes client there is nothing to watch,
and
+// without a listen address KEDA has nothing to dial; either way the rest of
the
+// control plane runs unchanged, because activation is a per-Service opt-in
+// rather than a mesh-wide dependency.
+func (s *Server) initActivation(args *DubboArgs) error {
+ s.activation = activation.NewServer()
+
+ if err := s.activation.Serve(args.ServerOptions.ActivationAddr,
s.internalStop); err != nil {
+ return err
+ }
+
+ if s.kubeClient == nil {
+ log.Info("activation policy controller disabled; no kube
client")
+ return nil
+ }
+
+ controller := activation.NewController(s.kubeClient,
s.activation.Scaler(), s.activation.Registry())
+ s.addStartFunc("activation policy controller", func(stop <-chan
struct{}) error {
+ go controller.Run(stop)
+ return nil
+ })
+
+ return nil
+}
diff --git a/dubbod/discovery/pkg/bootstrap/options.go
b/dubbod/discovery/pkg/bootstrap/options.go
index 5b3050fb..f5e9bbe5 100644
--- a/dubbod/discovery/pkg/bootstrap/options.go
+++ b/dubbod/discovery/pkg/bootstrap/options.go
@@ -57,6 +57,9 @@ type DiscoveryServerOptions struct {
HTTPSAddr string
GRPCAddr string
SecureGRPCAddr string
+ // ActivationAddr serves KEDA's external scaler contract. Empty
disables it,
+ // which is how a cluster without KEDA runs unchanged.
+ ActivationAddr string
TLSOptions TLSOptions
}
diff --git a/dubbod/discovery/pkg/bootstrap/server.go
b/dubbod/discovery/pkg/bootstrap/server.go
index 1db92cdb..34a206da 100644
--- a/dubbod/discovery/pkg/bootstrap/server.go
+++ b/dubbod/discovery/pkg/bootstrap/server.go
@@ -30,6 +30,7 @@ import (
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/status"
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/activation"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/features"
dubbogrpc "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/grpc"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/keycertbundle"
@@ -130,6 +131,7 @@ type Server struct {
proxylessGRPCWorkloadController *proxylessGRPCWorkloadController
proxylessGRPCRemoteControllers
*multicluster.Component[*proxylessGRPCClusterController]
statusManager *status.Manager
+ activation *activation.Server
}
type readinessFlags struct {
@@ -248,6 +250,10 @@ func NewServer(args *DubboArgs, initFuncs
...func(*Server)) (*Server, error) {
return nil, fmt.Errorf("error initializing Management Server:
%v", err)
}
+ if err := s.initActivation(args); err != nil {
+ return nil, fmt.Errorf("error initializing activation scaler:
%v", err)
+ }
+
// Initialize monitoring server
if err := s.initMonitor(args.ServerOptions.HTTPAddr); err != nil {
return nil, fmt.Errorf("error initializing monitoring: %v", err)
diff --git a/manifests/charts/dubbod/templates/clusterrole.yaml
b/manifests/charts/dubbod/templates/clusterrole.yaml
index 04339413..412f6c5c 100644
--- a/manifests/charts/dubbod/templates/clusterrole.yaml
+++ b/manifests/charts/dubbod/templates/clusterrole.yaml
@@ -24,6 +24,13 @@ rules:
- apiGroups: [ "networking.dubbo.apache.org" ]
verbs: [ "get", "watch", "list" ]
resources: [ "*" ]
+ # The activation controller publishes readiness back onto the policy. Without
+ # the status subresource the conditions are computed and then silently
+ # dropped, leaving an operator with no way to tell a working policy from a
+ # broken one.
+ - apiGroups: [ "networking.dubbo.apache.org" ]
+ verbs: [ "get", "update", "patch" ]
+ resources: [ "serviceactivationpolicies/status" ]
- apiGroups: [ "telemetry.dubbo.apache.org" ]
verbs: [ "get", "watch", "list" ]
resources: [ "*" ]
diff --git a/manifests/charts/dubbod/templates/deployment.yaml
b/manifests/charts/dubbod/templates/deployment.yaml
index f11dc0f4..88dcbe02 100644
--- a/manifests/charts/dubbod/templates/deployment.yaml
+++ b/manifests/charts/dubbod/templates/deployment.yaml
@@ -29,6 +29,13 @@
{{- $remoteAccess := $multicluster.remoteAccess | default dict }}
{{- $eastWestGateway := $multicluster.eastWestGateway | default dict }}
{{- $managementPort := int (coalesce $management.port $defaultManagement.port
26080) }}
+{{- $defaultActivation := $defaultGlobal.activation | default dict }}
+{{- $activation := $global.activation | default dict }}
+{{- /* coalesce skips zero, so an explicit 0 has to be read with hasKey or the
off switch silently falls back to the default. */ -}}
+{{- $activationPort := int (coalesce $defaultActivation.port 26030) }}
+{{- if hasKey $activation "port" }}
+{{- $activationPort = int $activation.port }}
+{{- end }}
{{- $replicaCount := int (coalesce .Values.replicaCount $defaults.replicaCount
1) }}
{{- $gateway := $global.gateway | default dict }}
{{- $defaultGateway := $defaultGlobal.gateway | default dict }}
@@ -98,6 +105,15 @@ spec:
- cluster.local
- --managementAddr
- ":{{ $managementPort }}"
+{{- if gt $activationPort 0 }}
+ - --activationAddr
+ - ":{{ $activationPort }}"
+{{- else }}
+ # Port 0 turns the KEDA scaler off; an empty address is how dubbod
+ # skips the listener entirely.
+ - --activationAddr
+ - ""
+{{- end }}
ports:
- containerPort: 8080
protocol: TCP
@@ -105,6 +121,11 @@ spec:
- containerPort: {{ $managementPort }}
protocol: TCP
name: management
+{{- if gt $activationPort 0 }}
+ - containerPort: {{ $activationPort }}
+ protocol: TCP
+ name: activation
+{{- end }}
- containerPort: 26010
protocol: TCP
name: grpc-xds
diff --git a/manifests/charts/dubbod/templates/service.yaml
b/manifests/charts/dubbod/templates/service.yaml
index 7aa1ef49..f638a74b 100644
--- a/manifests/charts/dubbod/templates/service.yaml
+++ b/manifests/charts/dubbod/templates/service.yaml
@@ -23,6 +23,13 @@
{{- $multicluster := $global.multicluster | default dict }}
{{- $remoteAccess := $multicluster.remoteAccess | default dict }}
{{- $managementPort := int (coalesce $management.port $defaultManagement.port
26080) }}
+{{- $defaultActivation := $defaultGlobal.activation | default dict }}
+{{- $activation := $global.activation | default dict }}
+{{- /* coalesce skips zero, so an explicit 0 has to be read with hasKey or the
off switch silently falls back to the default. */ -}}
+{{- $activationPort := int (coalesce $defaultActivation.port 26030) }}
+{{- if hasKey $activation "port" }}
+{{- $activationPort = int $activation.port }}
+{{- end }}
{{- $remoteAccessEnabled := coalesce $remoteAccess.enabled
$defaultRemoteAccess.enabled false }}
{{- $remoteAccessServiceType := coalesce $remoteAccess.serviceType
$defaultRemoteAccess.serviceType "LoadBalancer" }}
{{- $remoteAccessGRPCPort := int (coalesce $remoteAccess.grpcPort
$defaultRemoteAccess.grpcPort 26010) }}
@@ -72,6 +79,54 @@ spec:
protocol: TCP
selector:
app: dubbod
+{{- if gt $activationPort 0 }}
+---
+# Separate Service because KEDA dials this by name from its own namespace; it
+# has no reason to see the management API, and the two scale independently.
+# Load balanced on purpose: KEDA needs exactly one replica to answer.
+apiVersion: v1
+kind: Service
+metadata:
+ name: dubbod-activation
+ namespace: dubbo-system
+ labels:
+ app: dubbod
+spec:
+ ports:
+ - port: {{ $activationPort }}
+ name: grpc-activation
+ targetPort: {{ $activationPort }}
+ protocol: TCP
+ selector:
+ app: dubbod
+---
+# Headless companion, for gateways rather than KEDA.
+#
+# A gateway must report its pending requests to every replica: KEDA's
+# activation stream lands on one replica chosen by the load-balanced Service
+# above, and a report that reached only a different replica would leave those
+# requests waiting for a scale-up nobody asked for. Resolving this name yields
+# one A record per pod, which is how a gateway finds them all.
+apiVersion: v1
+kind: Service
+metadata:
+ name: dubbod-activation-replicas
+ namespace: dubbo-system
+ labels:
+ app: dubbod
+spec:
+ clusterIP: None
+ # Report to replicas that are still starting as well: a gateway holding a
+ # request cannot wait for readiness to settle before asking for a scale-up.
+ publishNotReadyAddresses: true
+ ports:
+ - port: {{ $activationPort }}
+ name: grpc-activation
+ targetPort: {{ $activationPort }}
+ protocol: TCP
+ selector:
+ app: dubbod
+{{- end }}
{{- if $remoteAccessEnabled }}
---
apiVersion: v1
diff --git a/manifests/charts/dubbod/values.yaml
b/manifests/charts/dubbod/values.yaml
index b5731aff..04358956 100644
--- a/manifests/charts/dubbod/values.yaml
+++ b/manifests/charts/dubbod/values.yaml
@@ -54,6 +54,11 @@ _internal_default_values_not_set:
# gateway.dubbo.apache.org/replicas annotation.
replicaCount: 2
+ activation:
+ # KEDA external scaler for on-demand activation. Set port to 0 to disable
+ # it; the rest of the control plane is unaffected either way.
+ port: 26030
+
configValidation: true
multicluster:
diff --git a/operator/pkg/apis/proto/values_types.proto
b/operator/pkg/apis/proto/values_types.proto
index bcbeec79..75038ac7 100644
--- a/operator/pkg/apis/proto/values_types.proto
+++ b/operator/pkg/apis/proto/values_types.proto
@@ -28,6 +28,15 @@ message ManagementConfig {
int64 port = 1;
}
+message ActivationConfig {
+ // gRPC port serving KEDA's external scaler contract. Zero disables it.
+ //
+ // Wrapped so an explicit 0 is distinguishable from an unset field: a bare
+ // int64 drops its zero value, which would make the off switch unreachable
+ // through the operator API.
+ google.protobuf.Int64Value port = 1;
+}
+
message ProxyConfig {
string clusterDomain = 1;
}
@@ -80,6 +89,11 @@ message GlobalConfig {
ProxylessConfig proxyless = 6;
GatewayConfig gateway = 7;
+
+ // Field numbers are part of the wire format, so two branches must never
+ // claim the same one: 7 already belongs to gateway, and reusing it would
+ // produce a proto that compiles while silently reinterpreting data.
+ ActivationConfig activation = 8;
}
message RemoteAccessConfig {
diff --git a/operator/pkg/apis/values_types.pb.go
b/operator/pkg/apis/values_types.pb.go
index 03c82c28..7aef78d3 100644
--- a/operator/pkg/apis/values_types.pb.go
+++ b/operator/pkg/apis/values_types.pb.go
@@ -82,6 +82,55 @@ func (x *ManagementConfig) GetPort() int64 {
return 0
}
+type ActivationConfig struct {
+ state protoimpl.MessageState `protogen:"open.v1"`
+ // gRPC port serving KEDA's external scaler contract. Zero disables it.
+ //
+ // Wrapped so an explicit 0 is distinguishable from an unset field: a
bare
+ // int64 drops its zero value, which would make the off switch
unreachable
+ // through the operator API.
+ Port *wrapperspb.Int64Value
`protobuf:"bytes,1,opt,name=port,proto3" json:"port,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
+}
+
+func (x *ActivationConfig) Reset() {
+ *x = ActivationConfig{}
+ mi := &file_values_types_proto_msgTypes[1]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+}
+
+func (x *ActivationConfig) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ActivationConfig) ProtoMessage() {}
+
+func (x *ActivationConfig) ProtoReflect() protoreflect.Message {
+ mi := &file_values_types_proto_msgTypes[1]
+ 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 ActivationConfig.ProtoReflect.Descriptor instead.
+func (*ActivationConfig) Descriptor() ([]byte, []int) {
+ return file_values_types_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *ActivationConfig) GetPort() *wrapperspb.Int64Value {
+ if x != nil {
+ return x.Port
+ }
+ return nil
+}
+
type ProxyConfig struct {
state protoimpl.MessageState `protogen:"open.v1"`
ClusterDomain string
`protobuf:"bytes,1,opt,name=clusterDomain,proto3"
json:"clusterDomain,omitempty"`
@@ -91,7 +140,7 @@ type ProxyConfig struct {
func (x *ProxyConfig) Reset() {
*x = ProxyConfig{}
- mi := &file_values_types_proto_msgTypes[1]
+ mi := &file_values_types_proto_msgTypes[2]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -103,7 +152,7 @@ func (x *ProxyConfig) String() string {
func (*ProxyConfig) ProtoMessage() {}
func (x *ProxyConfig) ProtoReflect() protoreflect.Message {
- mi := &file_values_types_proto_msgTypes[1]
+ mi := &file_values_types_proto_msgTypes[2]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -116,7 +165,7 @@ func (x *ProxyConfig) ProtoReflect() protoreflect.Message {
// Deprecated: Use ProxyConfig.ProtoReflect.Descriptor instead.
func (*ProxyConfig) Descriptor() ([]byte, []int) {
- return file_values_types_proto_rawDescGZIP(), []int{1}
+ return file_values_types_proto_rawDescGZIP(), []int{2}
}
func (x *ProxyConfig) GetClusterDomain() string {
@@ -135,7 +184,7 @@ type ProxylessConfig struct {
func (x *ProxylessConfig) Reset() {
*x = ProxylessConfig{}
- mi := &file_values_types_proto_msgTypes[2]
+ mi := &file_values_types_proto_msgTypes[3]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -147,7 +196,7 @@ func (x *ProxylessConfig) String() string {
func (*ProxylessConfig) ProtoMessage() {}
func (x *ProxylessConfig) ProtoReflect() protoreflect.Message {
- mi := &file_values_types_proto_msgTypes[2]
+ mi := &file_values_types_proto_msgTypes[3]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -160,7 +209,7 @@ func (x *ProxylessConfig) ProtoReflect()
protoreflect.Message {
// Deprecated: Use ProxylessConfig.ProtoReflect.Descriptor instead.
func (*ProxylessConfig) Descriptor() ([]byte, []int) {
- return file_values_types_proto_rawDescGZIP(), []int{2}
+ return file_values_types_proto_rawDescGZIP(), []int{3}
}
func (x *ProxylessConfig) GetCni() *MeshCNIConfig {
@@ -189,7 +238,7 @@ type MeshCNIConfig struct {
func (x *MeshCNIConfig) Reset() {
*x = MeshCNIConfig{}
- mi := &file_values_types_proto_msgTypes[3]
+ mi := &file_values_types_proto_msgTypes[4]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -201,7 +250,7 @@ func (x *MeshCNIConfig) String() string {
func (*MeshCNIConfig) ProtoMessage() {}
func (x *MeshCNIConfig) ProtoReflect() protoreflect.Message {
- mi := &file_values_types_proto_msgTypes[3]
+ mi := &file_values_types_proto_msgTypes[4]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -214,7 +263,7 @@ func (x *MeshCNIConfig) ProtoReflect() protoreflect.Message
{
// Deprecated: Use MeshCNIConfig.ProtoReflect.Descriptor instead.
func (*MeshCNIConfig) Descriptor() ([]byte, []int) {
- return file_values_types_proto_rawDescGZIP(), []int{3}
+ return file_values_types_proto_rawDescGZIP(), []int{4}
}
func (x *MeshCNIConfig) GetEnabled() bool {
@@ -305,7 +354,7 @@ type GatewayConfig struct {
func (x *GatewayConfig) Reset() {
*x = GatewayConfig{}
- mi := &file_values_types_proto_msgTypes[4]
+ mi := &file_values_types_proto_msgTypes[5]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -317,7 +366,7 @@ func (x *GatewayConfig) String() string {
func (*GatewayConfig) ProtoMessage() {}
func (x *GatewayConfig) 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 {
@@ -330,7 +379,7 @@ func (x *GatewayConfig) ProtoReflect() protoreflect.Message
{
// Deprecated: Use GatewayConfig.ProtoReflect.Descriptor instead.
func (*GatewayConfig) Descriptor() ([]byte, []int) {
- return file_values_types_proto_rawDescGZIP(), []int{4}
+ return file_values_types_proto_rawDescGZIP(), []int{5}
}
func (x *GatewayConfig) GetReplicaCount() *wrapperspb.Int32Value {
@@ -349,13 +398,17 @@ type GlobalConfig struct {
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
+ // Field numbers are part of the wire format, so two branches must never
+ // claim the same one: 7 already belongs to gateway, and reusing it
would
+ // produce a proto that compiles while silently reinterpreting data.
+ Activation *ActivationConfig
`protobuf:"bytes,8,opt,name=activation,proto3" json:"activation,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
}
func (x *GlobalConfig) Reset() {
*x = GlobalConfig{}
- mi := &file_values_types_proto_msgTypes[5]
+ mi := &file_values_types_proto_msgTypes[6]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -367,7 +420,7 @@ func (x *GlobalConfig) String() string {
func (*GlobalConfig) ProtoMessage() {}
func (x *GlobalConfig) 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 {
@@ -380,7 +433,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{5}
+ return file_values_types_proto_rawDescGZIP(), []int{6}
}
func (x *GlobalConfig) GetProxy() *ProxyConfig {
@@ -432,6 +485,13 @@ func (x *GlobalConfig) GetGateway() *GatewayConfig {
return nil
}
+func (x *GlobalConfig) GetActivation() *ActivationConfig {
+ if x != nil {
+ return x.Activation
+ }
+ return nil
+}
+
type RemoteAccessConfig struct {
state protoimpl.MessageState `protogen:"open.v1"`
Enabled bool
`protobuf:"varint,1,opt,name=enabled,proto3" json:"enabled,omitempty"`
@@ -446,7 +506,7 @@ type RemoteAccessConfig struct {
func (x *RemoteAccessConfig) Reset() {
*x = RemoteAccessConfig{}
- mi := &file_values_types_proto_msgTypes[6]
+ mi := &file_values_types_proto_msgTypes[7]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -458,7 +518,7 @@ func (x *RemoteAccessConfig) String() string {
func (*RemoteAccessConfig) ProtoMessage() {}
func (x *RemoteAccessConfig) 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 {
@@ -471,7 +531,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{6}
+ return file_values_types_proto_rawDescGZIP(), []int{7}
}
func (x *RemoteAccessConfig) GetEnabled() bool {
@@ -527,7 +587,7 @@ type EastWestGatewayEndpoint struct {
func (x *EastWestGatewayEndpoint) Reset() {
*x = EastWestGatewayEndpoint{}
- mi := &file_values_types_proto_msgTypes[7]
+ mi := &file_values_types_proto_msgTypes[8]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -539,7 +599,7 @@ func (x *EastWestGatewayEndpoint) String() string {
func (*EastWestGatewayEndpoint) ProtoMessage() {}
func (x *EastWestGatewayEndpoint) 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 {
@@ -552,7 +612,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{7}
+ return file_values_types_proto_rawDescGZIP(), []int{8}
}
func (x *EastWestGatewayEndpoint) GetClusterName() string {
@@ -591,7 +651,7 @@ type EastWestGatewayConfig struct {
func (x *EastWestGatewayConfig) Reset() {
*x = EastWestGatewayConfig{}
- mi := &file_values_types_proto_msgTypes[8]
+ mi := &file_values_types_proto_msgTypes[9]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -603,7 +663,7 @@ func (x *EastWestGatewayConfig) String() string {
func (*EastWestGatewayConfig) ProtoMessage() {}
func (x *EastWestGatewayConfig) 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 {
@@ -616,7 +676,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{8}
+ return file_values_types_proto_rawDescGZIP(), []int{9}
}
func (x *EastWestGatewayConfig) GetEnabled() bool {
@@ -678,7 +738,7 @@ type MulticlusterConfig struct {
func (x *MulticlusterConfig) Reset() {
*x = MulticlusterConfig{}
- 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 +750,7 @@ func (x *MulticlusterConfig) String() string {
func (*MulticlusterConfig) ProtoMessage() {}
func (x *MulticlusterConfig) 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 +763,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{9}
+ return file_values_types_proto_rawDescGZIP(), []int{10}
}
func (x *MulticlusterConfig) GetRemoteAccess() *RemoteAccessConfig {
@@ -736,7 +796,7 @@ type Values struct {
func (x *Values) Reset() {
*x = Values{}
- mi := &file_values_types_proto_msgTypes[10]
+ mi := &file_values_types_proto_msgTypes[11]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -748,7 +808,7 @@ func (x *Values) String() string {
func (*Values) ProtoMessage() {}
func (x *Values) 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 {
@@ -761,7 +821,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{10}
+ return file_values_types_proto_rawDescGZIP(), []int{11}
}
func (x *Values) GetGlobal() *GlobalConfig {
@@ -805,7 +865,7 @@ type IntOrString struct {
func (x *IntOrString) Reset() {
*x = IntOrString{}
- mi := &file_values_types_proto_msgTypes[11]
+ mi := &file_values_types_proto_msgTypes[12]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -817,7 +877,7 @@ func (x *IntOrString) String() string {
func (*IntOrString) ProtoMessage() {}
func (x *IntOrString) ProtoReflect() protoreflect.Message {
- mi := &file_values_types_proto_msgTypes[11]
+ mi := &file_values_types_proto_msgTypes[12]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -830,7 +890,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{11}
+ return file_values_types_proto_rawDescGZIP(), []int{12}
}
func (x *IntOrString) GetType() int64 {
@@ -860,7 +920,9 @@ const file_values_types_proto_rawDesc = "" +
"\n" +
"\x12values_types.proto\x12\x17dubbo.operator.v1alpha1\x1a\x1egoogle/protobuf/wrappers.proto\"&\n"
+
"\x10ManagementConfig\x12\x12\n" +
- "\x04port\x18\x01 \x01(\x03R\x04port\"3\n" +
+ "\x04port\x18\x01 \x01(\x03R\x04port\"C\n" +
+ "\x10ActivationConfig\x12/\n" +
+ "\x04port\x18\x01
\x01(\v2\x1b.google.protobuf.Int64ValueR\x04port\"3\n" +
"\vProxyConfig\x12$\n" +
"\rclusterDomain\x18\x01 \x01(\tR\rclusterDomain\"K\n" +
"\x0fProxylessConfig\x128\n" +
@@ -879,7 +941,7 @@ const file_values_types_proto_rawDesc = "" +
" \x01(\tR\tipsetPath\x12(\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" +
+ "\freplicaCount\x18\x01
\x01(\v2\x1b.google.protobuf.Int32ValueR\freplicaCount\"\x87\x04\n" +
"\fGlobalConfig\x12:\n" +
"\x05proxy\x18\x01
\x01(\v2$.dubbo.operator.v1alpha1.ProxyConfigR\x05proxy\x12\x1e\n" +
"\n" +
@@ -891,7 +953,10 @@ const file_values_types_proto_rawDesc = "" +
"\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\x12@\n" +
- "\agateway\x18\a
\x01(\v2&.dubbo.operator.v1alpha1.GatewayConfigR\agateway\"\xd4\x01\n" +
+ "\agateway\x18\a
\x01(\v2&.dubbo.operator.v1alpha1.GatewayConfigR\agateway\x12I\n" +
+ "\n" +
+ "activation\x18\b
\x01(\v2).dubbo.operator.v1alpha1.ActivationConfigR\n" +
+ "activation\"\xd4\x01\n" +
"\x12RemoteAccessConfig\x12\x18\n" +
"\aenabled\x18\x01 \x01(\bR\aenabled\x12 \n" +
"\vserviceType\x18\x02 \x01(\tR\vserviceType\x12\x18\n" +
@@ -939,43 +1004,47 @@ func file_values_types_proto_rawDescGZIP() []byte {
return file_values_types_proto_rawDescData
}
-var file_values_types_proto_msgTypes = make([]protoimpl.MessageInfo, 12)
+var file_values_types_proto_msgTypes = make([]protoimpl.MessageInfo, 13)
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
- (*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
+ (*ActivationConfig)(nil), // 1:
dubbo.operator.v1alpha1.ActivationConfig
+ (*ProxyConfig)(nil), // 2:
dubbo.operator.v1alpha1.ProxyConfig
+ (*ProxylessConfig)(nil), // 3:
dubbo.operator.v1alpha1.ProxylessConfig
+ (*MeshCNIConfig)(nil), // 4:
dubbo.operator.v1alpha1.MeshCNIConfig
+ (*GatewayConfig)(nil), // 5:
dubbo.operator.v1alpha1.GatewayConfig
+ (*GlobalConfig)(nil), // 6:
dubbo.operator.v1alpha1.GlobalConfig
+ (*RemoteAccessConfig)(nil), // 7:
dubbo.operator.v1alpha1.RemoteAccessConfig
+ (*EastWestGatewayEndpoint)(nil), // 8:
dubbo.operator.v1alpha1.EastWestGatewayEndpoint
+ (*EastWestGatewayConfig)(nil), // 9:
dubbo.operator.v1alpha1.EastWestGatewayConfig
+ (*MulticlusterConfig)(nil), // 10:
dubbo.operator.v1alpha1.MulticlusterConfig
+ (*Values)(nil), // 11: dubbo.operator.v1alpha1.Values
+ (*IntOrString)(nil), // 12:
dubbo.operator.v1alpha1.IntOrString
+ (*wrapperspb.Int64Value)(nil), // 13: google.protobuf.Int64Value
+ (*wrapperspb.Int32Value)(nil), // 14: google.protobuf.Int32Value
+ (*wrapperspb.StringValue)(nil), // 15: google.protobuf.StringValue
}
var file_values_types_proto_depIdxs = []int32{
- 3, // 0: dubbo.operator.v1alpha1.ProxylessConfig.cni:type_name ->
dubbo.operator.v1alpha1.MeshCNIConfig
- 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
+ 13, // 0: dubbo.operator.v1alpha1.ActivationConfig.port:type_name ->
google.protobuf.Int64Value
+ 4, // 1: dubbo.operator.v1alpha1.ProxylessConfig.cni:type_name ->
dubbo.operator.v1alpha1.MeshCNIConfig
+ 14, // 2: dubbo.operator.v1alpha1.GatewayConfig.replicaCount:type_name
-> google.protobuf.Int32Value
+ 2, // 3: dubbo.operator.v1alpha1.GlobalConfig.proxy:type_name ->
dubbo.operator.v1alpha1.ProxyConfig
+ 0, // 4: dubbo.operator.v1alpha1.GlobalConfig.management:type_name ->
dubbo.operator.v1alpha1.ManagementConfig
+ 10, // 5: dubbo.operator.v1alpha1.GlobalConfig.multicluster:type_name
-> dubbo.operator.v1alpha1.MulticlusterConfig
+ 3, // 6: dubbo.operator.v1alpha1.GlobalConfig.proxyless:type_name ->
dubbo.operator.v1alpha1.ProxylessConfig
+ 5, // 7: dubbo.operator.v1alpha1.GlobalConfig.gateway:type_name ->
dubbo.operator.v1alpha1.GatewayConfig
+ 1, // 8: dubbo.operator.v1alpha1.GlobalConfig.activation:type_name ->
dubbo.operator.v1alpha1.ActivationConfig
+ 8, // 9:
dubbo.operator.v1alpha1.EastWestGatewayConfig.gateways:type_name ->
dubbo.operator.v1alpha1.EastWestGatewayEndpoint
+ 7, // 10:
dubbo.operator.v1alpha1.MulticlusterConfig.remoteAccess:type_name ->
dubbo.operator.v1alpha1.RemoteAccessConfig
+ 9, // 11:
dubbo.operator.v1alpha1.MulticlusterConfig.eastWestGateway:type_name ->
dubbo.operator.v1alpha1.EastWestGatewayConfig
+ 6, // 12: dubbo.operator.v1alpha1.Values.global:type_name ->
dubbo.operator.v1alpha1.GlobalConfig
+ 14, // 13: dubbo.operator.v1alpha1.Values.replicaCount:type_name ->
google.protobuf.Int32Value
+ 14, // 14: dubbo.operator.v1alpha1.IntOrString.intVal:type_name ->
google.protobuf.Int32Value
+ 15, // 15: dubbo.operator.v1alpha1.IntOrString.strVal:type_name ->
google.protobuf.StringValue
+ 16, // [16:16] is the sub-list for method output_type
+ 16, // [16:16] is the sub-list for method input_type
+ 16, // [16:16] is the sub-list for extension type_name
+ 16, // [16:16] is the sub-list for extension extendee
+ 0, // [0:16] is the sub-list for field type_name
}
func init() { file_values_types_proto_init() }
@@ -989,7 +1058,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: 12,
+ NumMessages: 13,
NumExtensions: 0,
NumServices: 0,
},
diff --git a/operator/pkg/render/manifest_test.go
b/operator/pkg/render/manifest_test.go
index 3a42bf5a..81d5451d 100644
--- a/operator/pkg/render/manifest_test.go
+++ b/operator/pkg/render/manifest_test.go
@@ -159,6 +159,94 @@ func
TestTelemetryValidationWebhookDoesNotRequireRevisionLabel(t *testing.T) {
t.Fatal("Telemetry validation webhook not rendered")
}
+// The activation scaler is only reachable if the flag, the container port and
+// the Service KEDA dials all agree on one number. Each is written in a
+// different template, so a mismatch renders cleanly and fails only at runtime.
+func TestGenerateManifestWiresActivationScalerEndToEnd(t *testing.T) {
+ manifests, _, err := GenerateManifest(nil,
[]string{"values.global.activation.port=26031"}, 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)
+ }
+ execute := containers[0].(map[string]interface{})
+
+ args, _, _ := unstructured.NestedStringSlice(execute, "args")
+ if !hasArgValue(args, "--activationAddr", ":26031") {
+ t.Fatalf("args = %v, want --activationAddr :26031", args)
+ }
+
+ ports, ok, err := unstructured.NestedSlice(execute, "ports")
+ if err != nil || !ok {
+ t.Fatalf("container ports missing: ok=%v err=%v", ok, err)
+ }
+ if !hasContainerPort(ports, "activation", 26031) {
+ t.Fatal("dubbod deployment missing activation containerPort
26031")
+ }
+
+ service := findManifest(t, manifests, "Service", "dubbod-activation")
+ port, _, _ := unstructured.NestedInt64(findPort(t, service,
"grpc-activation"), "port")
+ if port != 26031 {
+ t.Fatalf("dubbod-activation port = %d, want 26031", port)
+ }
+
+ // Gateways broadcast their demand to every replica, so they need a name
+ // that resolves to all of them. A load-balanced Service would deliver
each
+ // report to one arbitrary replica, and KEDA's stream is on another.
+ replicas := findManifest(t, manifests, "Service",
"dubbod-activation-replicas")
+ clusterIP, _, _ := unstructured.NestedString(replicas.Object, "spec",
"clusterIP")
+ if clusterIP != "None" {
+ t.Fatalf("dubbod-activation-replicas clusterIP = %q, want
None", clusterIP)
+ }
+ notReady, _, _ := unstructured.NestedBool(replicas.Object, "spec",
"publishNotReadyAddresses")
+ if !notReady {
+ t.Fatal("dubbod-activation-replicas must publish not-ready
addresses so a held request is not blocked on readiness")
+ }
+}
+
+// Port zero is the documented off switch. It has to remove the listener, the
+// port and the Service together; leaving a Service behind would publish an
+// endpoint that refuses every connection.
+func TestGenerateManifestDisablesActivationAtPortZero(t *testing.T) {
+ manifests, _, err := GenerateManifest(nil,
[]string{"values.global.activation.port=0"}, nil, nil)
+ if err != nil {
+ t.Fatalf("GenerateManifest() error = %v", err)
+ }
+ if hasManifest(manifests, "Service", "dubbod-activation") {
+ t.Fatal("dubbod-activation Service rendered while activation is
disabled")
+ }
+ if hasManifest(manifests, "Service", "dubbod-activation-replicas") {
+ t.Fatal("dubbod-activation-replicas Service rendered while
activation is disabled")
+ }
+
+ deployment := findManifest(t, manifests, "Deployment", "dubbod")
+ containers, _, _ := unstructured.NestedSlice(deployment.Object, "spec",
"template", "spec", "containers")
+ execute := containers[0].(map[string]interface{})
+ args, _, _ := unstructured.NestedStringSlice(execute, "args")
+ if !hasArgValue(args, "--activationAddr", "") {
+ t.Fatalf("args = %v, want --activationAddr with an empty
value", args)
+ }
+ ports, _, _ := unstructured.NestedSlice(execute, "ports")
+ for _, raw := range ports {
+ if name, _, _ :=
unstructured.NestedString(raw.(map[string]interface{}), "name"); name ==
"activation" {
+ t.Fatal("activation containerPort rendered while
activation is disabled")
+ }
+ }
+}
+
+func hasArgValue(args []string, flag, value string) bool {
+ for i := 0; i < len(args)-1; i++ {
+ if args[i] == flag && args[i+1] == value {
+ return true
+ }
+ }
+ return false
+}
+
func TestGenerateManifestExposesDubbodPrometheusScrapeEndpoint(t *testing.T) {
manifests, _, err := GenerateManifest(nil, nil, nil, nil)
if err != nil {
diff --git a/tools/make/common.mk b/tools/make/common.mk
index 819df27e..92c51373 100644
--- a/tools/make/common.mk
+++ b/tools/make/common.mk
@@ -25,6 +25,16 @@ BIN_DIR ?= bin
GOTESTFLAGS ?= -race
GOLANGCI_LINT_VERSION ?= v2.6.2
+# Code generation. protoc-gen-go must match the google.golang.org/protobuf in
+# go.mod, and protoc must match the version already recorded in the generated
+# headers: a mismatch rewrites every file and turns `make check-generate` into
+# a permanent failure.
+PROTOC_VERSION ?= v33.0
+PROTOC_GEN_GO_VERSION ?= v1.36.11
+GOIMPORTS_VERSION ?= latest
+TOOL_BIN ?= $(CURDIR)/$(BIN_DIR)/tools
+OPERATOR_APIS_DIR := operator/pkg/apis
+
HUB ?= kdubbo
IMAGE_TAG ?= debug
IMAGE ?= $(HUB)/dubbod:$(IMAGE_TAG)
diff --git a/tools/make/lint.mk b/tools/make/lint.mk
index 6dae36a2..b689d4cf 100644
--- a/tools/make/lint.mk
+++ b/tools/make/lint.mk
@@ -71,3 +71,33 @@ check-clean-repo: ## Fail if the working tree is dirty.
.PHONY: check-tidy
check-tidy: tidy check-clean-repo ## go mod tidy, then fail if it changed
anything.
+
+.PHONY: generate
+generate: generate-proto generate-schema ## Re-run every code generator.
+
+.PHONY: generate-proto
+generate-proto: ## Regenerate the operator values API from its .proto.
+ @command -v protoc >/dev/null 2>&1 || { \
+ echo "protoc not found; install $(PROTOC_VERSION) from
https://github.com/protocolbuffers/protobuf/releases";; \
+ exit 1; \
+ }
+ @GOBIN=$(TOOL_BIN) go install
google.golang.org/protobuf/cmd/protoc-gen-go@$(PROTOC_GEN_GO_VERSION)
+ @cd $(OPERATOR_APIS_DIR) && PATH="$(TOOL_BIN):$$PATH" protoc \
+ --proto_path=proto \
+ --proto_path=$$(go list -f '{{ .Dir }}' -m k8s.io/api) \
+ --proto_path=$$(go list -f '{{ .Dir }}' -m k8s.io/apimachinery)
\
+ --go_out=. proto/values_types.proto
+ @# go_package is a full import path and protoc has no
paths=source_relative
+ @# here, so the output lands under a mirrored directory tree.
+ mv
$(OPERATOR_APIS_DIR)/dubbo.apache.org/dubbo/operator/pkg/apis/values_types.pb.go
\
+ $(OPERATOR_APIS_DIR)/values_types.pb.go
+ rm -rf $(OPERATOR_APIS_DIR)/dubbo.apache.org
+
+.PHONY: generate-schema
+generate-schema: ## Regenerate the resource schema from metadata.yaml.
+ @GOBIN=$(TOOL_BIN) go install
golang.org/x/tools/cmd/goimports@$(GOIMPORTS_VERSION)
+ @PATH="$(TOOL_BIN):$$PATH" go run
./pkg/config/schema/codegen/tools/collections.main.go
+
+.PHONY: check-generate
+check-generate: generate check-clean-repo ## Regenerate, then fail if anything
changed.
+ @echo "Generated sources are up to date."
