This is an automated email from the ASF dual-hosted git repository.
github-actions[bot] pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/dubbo-kubernetes.git
The following commit(s) were added to refs/heads/master by this push:
new 328ac70d Complete the registration of core activation resources (#991)
328ac70d is described below
commit 328ac70db2f0b9e829b145b14d80b331a1597449
Author: mfordjody <[email protected]>
AuthorDate: Fri Aug 7 15:16:14 2026 +0800
Complete the registration of core activation resources (#991)
---
dubbod/discovery/pkg/activation/demand.go | 226 +++++++++
dubbod/discovery/pkg/activation/demand_test.go | 217 +++++++++
.../activation/externalscaler/externalscaler.pb.go | 512 +++++++++++++++++++++
.../activation/externalscaler/externalscaler.proto | 75 +++
.../externalscaler/externalscaler_grpc.pb.go | 315 +++++++++++++
dubbod/discovery/pkg/activation/grpc_test.go | 147 ++++++
dubbod/discovery/pkg/activation/policy.go | 220 +++++++++
dubbod/discovery/pkg/activation/policy_test.go | 284 ++++++++++++
dubbod/discovery/pkg/activation/scaler.go | 262 +++++++++++
dubbod/discovery/pkg/activation/scaler_test.go | 306 ++++++++++++
.../pkg/config/kube/crdclient/types.gen.go | 51 ++
go.mod | 4 +-
go.sum | 8 +-
pkg/config/schema/collections/collections.gen.go | 19 +
pkg/config/schema/gvk/resources.gen.go | 7 +
pkg/config/schema/gvr/resources.gen.go | 3 +
pkg/config/schema/kind/resources.gen.go | 5 +
pkg/config/schema/kubeclient/resources.gen.go | 13 +
pkg/config/schema/kubetypes/resources.gen.go | 4 +
pkg/config/schema/metadata.yaml | 9 +
20 files changed, 2681 insertions(+), 6 deletions(-)
diff --git a/dubbod/discovery/pkg/activation/demand.go
b/dubbod/discovery/pkg/activation/demand.go
new file mode 100644
index 00000000..5c6e57eb
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/demand.go
@@ -0,0 +1,226 @@
+// 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 (
+ "sync"
+ "time"
+)
+
+// Target is the Service whose requests are being held while it scales up.
+type Target struct {
+ Namespace string
+ Name string
+}
+
+// DemandSource reports how many requests are waiting for a target to get
+// endpoints. It is the seam between the gateways that hold the requests and
+// the KEDA scaler that reports them, so the scaler can be built and tested
+// without depending on how demand arrives.
+type DemandSource interface {
+ // Pending returns the number of requests currently held for the target.
+ Pending(Target) int64
+
+ // Subscribe delivers the pending count whenever it changes. The
channel is
+ // closed when cancel is called. Implementations must not block on it: a
+ // slow subscriber may miss intermediate values but must still converge
on
+ // the latest one, because a missed edge would leave a workload scaled
to
+ // zero with requests waiting on it.
+ Subscribe(Target) (updates <-chan int64, cancel func())
+}
+
+// reporterTTL bounds how long one gateway's report is trusted. A gateway that
+// dies mid-activation stops refreshing, and its demand has to expire on its
+// own; otherwise the target stays scaled up forever with nothing waiting on
it.
+const reporterTTL = 30 * time.Second
+
+// Registry aggregates pending counts reported by gateways.
+//
+// Reports are absolute counts per reporter, not deltas: a gateway that
+// restarts, or whose report is lost, converges on the next report instead of
+// leaving the total permanently skewed.
+type Registry struct {
+ mu sync.Mutex
+ // Per target, the last count each reporter published and when.
+ targets map[Target]map[string]report
+ // Per target, the live subscribers.
+ subscribers map[Target]map[int]chan int64
+ nextID int
+
+ // now is swappable so expiry can be tested without sleeping.
+ now func() time.Time
+ ttl time.Duration
+}
+
+type report struct {
+ pending int64
+ received time.Time
+}
+
+func NewRegistry() *Registry {
+ return &Registry{
+ targets: map[Target]map[string]report{},
+ subscribers: map[Target]map[int]chan int64{},
+ now: time.Now,
+ ttl: reporterTTL,
+ }
+}
+
+// Report records the requests one gateway is currently holding for a target.
+// It is called on every refresh, including with zero, which is how a gateway
+// says it has drained.
+func (r *Registry) Report(reporter string, target Target, pending int64) {
+ if pending < 0 {
+ pending = 0
+ }
+ r.mu.Lock()
+ byReporter, ok := r.targets[target]
+ if !ok {
+ byReporter = map[string]report{}
+ r.targets[target] = byReporter
+ }
+ byReporter[reporter] = report{pending: pending, received: r.now()}
+ total := r.totalLocked(target)
+ subscribers := r.snapshotSubscribersLocked(target)
+ r.mu.Unlock()
+
+ notify(subscribers, total)
+}
+
+// Forget drops a gateway's reports, for a clean shutdown that should not wait
+// out the TTL.
+func (r *Registry) Forget(reporter string) {
+ type notification struct {
+ subscribers []chan int64
+ total int64
+ }
+
+ r.mu.Lock()
+ var notifications []notification
+ for target, byReporter := range r.targets {
+ if _, ok := byReporter[reporter]; !ok {
+ continue
+ }
+ delete(byReporter, reporter)
+ notifications = append(notifications, notification{
+ subscribers: r.snapshotSubscribersLocked(target),
+ total: r.totalLocked(target),
+ })
+ }
+ r.mu.Unlock()
+
+ for _, item := range notifications {
+ notify(item.subscribers, item.total)
+ }
+}
+
+func (r *Registry) Pending(target Target) int64 {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ return r.totalLocked(target)
+}
+
+// Reporters counts the gateways currently holding requests for a target, or
+// standing by to. Zero means nothing would catch a request for it, which is
+// what the policy controller reports rather than letting a policy look ready
+// while no gateway can act on it.
+func (r *Registry) Reporters(target Target) int {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ // Expires stale reporters as a side effect, so a gateway that vanished
does
+ // not keep a policy looking healthy.
+ r.totalLocked(target)
+ return len(r.targets[target])
+}
+
+func (r *Registry) Subscribe(target Target) (<-chan int64, func()) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+
+ // Buffered by one so a notifier never blocks and a subscriber that is
busy
+ // still finds the latest value waiting for it.
+ updates := make(chan int64, 1)
+ id := r.nextID
+ r.nextID++
+ if r.subscribers[target] == nil {
+ r.subscribers[target] = map[int]chan int64{}
+ }
+ r.subscribers[target][id] = updates
+
+ cancel := func() {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ if channels, ok := r.subscribers[target]; ok {
+ if channel, ok := channels[id]; ok {
+ delete(channels, id)
+ close(channel)
+ }
+ if len(channels) == 0 {
+ delete(r.subscribers, target)
+ }
+ }
+ }
+ return updates, cancel
+}
+
+// totalLocked sums the live reports for a target, dropping expired reporters
as
+// it goes so a vanished gateway cannot hold a workload up indefinitely.
+func (r *Registry) totalLocked(target Target) int64 {
+ byReporter, ok := r.targets[target]
+ if !ok {
+ return 0
+ }
+ cutoff := r.now().Add(-r.ttl)
+ var total int64
+ for reporter, entry := range byReporter {
+ if entry.received.Before(cutoff) {
+ delete(byReporter, reporter)
+ continue
+ }
+ total += entry.pending
+ }
+ if len(byReporter) == 0 {
+ delete(r.targets, target)
+ }
+ return total
+}
+
+func (r *Registry) snapshotSubscribersLocked(target Target) []chan int64 {
+ channels := r.subscribers[target]
+ if len(channels) == 0 {
+ return nil
+ }
+ out := make([]chan int64, 0, len(channels))
+ for _, channel := range channels {
+ out = append(out, channel)
+ }
+ return out
+}
+
+// notify replaces any value a subscriber has not read yet. Only the latest
+// count matters, and dropping a stale one keeps the notifier non-blocking.
+func notify(subscribers []chan int64, total int64) {
+ for _, channel := range subscribers {
+ select {
+ case <-channel:
+ default:
+ }
+ select {
+ case channel <- total:
+ default:
+ }
+ }
+}
diff --git a/dubbod/discovery/pkg/activation/demand_test.go
b/dubbod/discovery/pkg/activation/demand_test.go
new file mode 100644
index 00000000..4b661c8b
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/demand_test.go
@@ -0,0 +1,217 @@
+// 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 (
+ "sync"
+ "testing"
+ "time"
+)
+
+var orders = Target{Namespace: "app", Name: "orders"}
+
+func TestRegistrySumsReportersAndTreatsReportsAsAbsolute(t *testing.T) {
+ registry := NewRegistry()
+
+ registry.Report("gateway-a", orders, 3)
+ registry.Report("gateway-b", orders, 2)
+ if got := registry.Pending(orders); got != 5 {
+ t.Fatalf("pending = %d, want 5", got)
+ }
+
+ // Reports are absolute, not deltas: a gateway restating its count must
+ // replace its own contribution rather than add to it.
+ registry.Report("gateway-a", orders, 1)
+ if got := registry.Pending(orders); got != 3 {
+ t.Fatalf("pending after restatement = %d, want 3", got)
+ }
+
+ registry.Report("gateway-a", orders, 0)
+ registry.Report("gateway-b", orders, 0)
+ if got := registry.Pending(orders); got != 0 {
+ t.Fatalf("pending after drain = %d, want 0", got)
+ }
+}
+
+func TestRegistryIsolatesTargets(t *testing.T) {
+ registry := NewRegistry()
+ reviews := Target{Namespace: "app", Name: "reviews"}
+ otherNamespace := Target{Namespace: "staging", Name: "orders"}
+
+ registry.Report("gateway-a", orders, 4)
+ if got := registry.Pending(reviews); got != 0 {
+ t.Fatalf("unrelated service pending = %d, want 0", got)
+ }
+ // Same Service name in another namespace is a different workload.
+ if got := registry.Pending(otherNamespace); got != 0 {
+ t.Fatalf("same name in another namespace pending = %d, want 0",
got)
+ }
+}
+
+// A gateway that dies mid-activation stops refreshing. Its demand has to age
+// out, or the target stays scaled up forever with nothing waiting on it.
+func TestRegistryExpiresStaleReporters(t *testing.T) {
+ registry := NewRegistry()
+ now := time.Unix(0, 0)
+ registry.now = func() time.Time { return now }
+
+ registry.Report("gateway-a", orders, 5)
+ registry.Report("gateway-b", orders, 1)
+
+ now = now.Add(reporterTTL / 2)
+ registry.Report("gateway-b", orders, 1)
+
+ // gateway-a has not refreshed for longer than the TTL; gateway-b has.
+ now = now.Add(reporterTTL/2 + time.Second)
+ if got := registry.Pending(orders); got != 1 {
+ t.Fatalf("pending after gateway-a expired = %d, want 1", got)
+ }
+
+ now = now.Add(reporterTTL + time.Second)
+ if got := registry.Pending(orders); got != 0 {
+ t.Fatalf("pending after all reporters expired = %d, want 0",
got)
+ }
+}
+
+func TestRegistryForgetDropsAReporterImmediately(t *testing.T) {
+ registry := NewRegistry()
+ registry.Report("gateway-a", orders, 3)
+ registry.Report("gateway-b", orders, 2)
+
+ registry.Forget("gateway-a")
+ if got := registry.Pending(orders); got != 2 {
+ t.Fatalf("pending after forget = %d, want 2", got)
+ }
+}
+
+func TestRegistryReportersCountsLiveGateways(t *testing.T) {
+ registry := NewRegistry()
+ if got := registry.Reporters(orders); got != 0 {
+ t.Fatalf("reporters with no gateway = %d, want 0", got)
+ }
+
+ registry.Report("gateway-a", orders, 0)
+ registry.Report("gateway-b", orders, 0)
+ // A gateway holding zero requests is still standing by to catch one.
+ if got := registry.Reporters(orders); got != 2 {
+ t.Fatalf("reporters = %d, want 2", got)
+ }
+
+ registry.Forget("gateway-b")
+ if got := registry.Reporters(orders); got != 1 {
+ t.Fatalf("reporters after forget = %d, want 1", got)
+ }
+}
+
+func TestRegistryNotifiesSubscribers(t *testing.T) {
+ registry := NewRegistry()
+ updates, cancel := registry.Subscribe(orders)
+ defer cancel()
+
+ registry.Report("gateway-a", orders, 2)
+ if got := receive(t, updates); got != 2 {
+ t.Fatalf("update = %d, want 2", got)
+ }
+
+ registry.Report("gateway-a", orders, 0)
+ if got := receive(t, updates); got != 0 {
+ t.Fatalf("drain update = %d, want 0", got)
+ }
+}
+
+// Only the newest count matters. A subscriber that was busy must not be handed
+// a stale value that says zero while requests are waiting.
+func TestRegistrySubscriberSeesLatestCountAfterMissingUpdates(t *testing.T) {
+ registry := NewRegistry()
+ updates, cancel := registry.Subscribe(orders)
+ defer cancel()
+
+ registry.Report("gateway-a", orders, 1)
+ registry.Report("gateway-a", orders, 7)
+ registry.Report("gateway-a", orders, 4)
+
+ if got := receive(t, updates); got != 4 {
+ t.Fatalf("update = %d, want the latest count 4", got)
+ }
+}
+
+func TestRegistryCancelStopsDelivery(t *testing.T) {
+ registry := NewRegistry()
+ updates, cancel := registry.Subscribe(orders)
+
+ cancel()
+ if _, open := <-updates; open {
+ t.Fatal("channel still open after cancel")
+ }
+
+ // Reporting after cancel must not panic on the closed channel.
+ registry.Report("gateway-a", orders, 1)
+}
+
+// Reporting must never block on a subscriber, or one stuck gateway stream
+// would stall every other gateway's reports.
+func TestRegistryReportDoesNotBlockOnIdleSubscriber(t *testing.T) {
+ registry := NewRegistry()
+ _, cancel := registry.Subscribe(orders)
+ defer cancel()
+
+ done := make(chan struct{})
+ go func() {
+ defer close(done)
+ for i := 0; i < 100; i++ {
+ registry.Report("gateway-a", orders, int64(i))
+ }
+ }()
+
+ select {
+ case <-done:
+ case <-time.After(5 * time.Second):
+ t.Fatal("Report blocked on a subscriber that never read")
+ }
+}
+
+func TestRegistryConcurrentReportsAndReads(t *testing.T) {
+ registry := NewRegistry()
+ var wait sync.WaitGroup
+ for reporter := 0; reporter < 8; reporter++ {
+ wait.Add(2)
+ name := "gateway-" + string(rune('a'+reporter))
+ go func() {
+ defer wait.Done()
+ for i := 0; i < 200; i++ {
+ registry.Report(name, orders, int64(i%5))
+ }
+ }()
+ go func() {
+ defer wait.Done()
+ for i := 0; i < 200; i++ {
+ registry.Pending(orders)
+ }
+ }()
+ }
+ wait.Wait()
+}
+
+func receive(t *testing.T, updates <-chan int64) int64 {
+ t.Helper()
+ select {
+ case value := <-updates:
+ return value
+ case <-time.After(2 * time.Second):
+ t.Fatal("no update delivered")
+ return 0
+ }
+}
diff --git
a/dubbod/discovery/pkg/activation/externalscaler/externalscaler.pb.go
b/dubbod/discovery/pkg/activation/externalscaler/externalscaler.pb.go
new file mode 100644
index 00000000..82f9a04b
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/externalscaler/externalscaler.pb.go
@@ -0,0 +1,512 @@
+// Vendored from github.com/kedacore/keda
+// pkg/scalers/externalscaler/externalscaler.proto (Apache-2.0).
+//
+// This is the contract KEDA dials when a ScaledObject declares an "external"
+// or "external-push" trigger. Regenerate with:
+//
+// protoc -I . --go_out=. --go_opt=paths=source_relative \
+// --go-grpc_out=. --go-grpc_opt=paths=source_relative externalscaler.proto
+
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// protoc-gen-go v1.36.11
+// protoc v6.33.0
+// source: externalscaler.proto
+
+package externalscaler
+
+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)
+)
+
+type ScaledObjectRef struct {
+ state protoimpl.MessageState `protogen:"open.v1"`
+ Name string
`protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
+ Namespace string
`protobuf:"bytes,2,opt,name=namespace,proto3" json:"namespace,omitempty"`
+ ScalerMetadata map[string]string
`protobuf:"bytes,3,rep,name=scalerMetadata,proto3"
json:"scalerMetadata,omitempty" protobuf_key:"bytes,1,opt,name=key"
protobuf_val:"bytes,2,opt,name=value"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
+}
+
+func (x *ScaledObjectRef) Reset() {
+ *x = ScaledObjectRef{}
+ mi := &file_externalscaler_proto_msgTypes[0]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+}
+
+func (x *ScaledObjectRef) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ScaledObjectRef) ProtoMessage() {}
+
+func (x *ScaledObjectRef) ProtoReflect() protoreflect.Message {
+ mi := &file_externalscaler_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 ScaledObjectRef.ProtoReflect.Descriptor instead.
+func (*ScaledObjectRef) Descriptor() ([]byte, []int) {
+ return file_externalscaler_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *ScaledObjectRef) GetName() string {
+ if x != nil {
+ return x.Name
+ }
+ return ""
+}
+
+func (x *ScaledObjectRef) GetNamespace() string {
+ if x != nil {
+ return x.Namespace
+ }
+ return ""
+}
+
+func (x *ScaledObjectRef) GetScalerMetadata() map[string]string {
+ if x != nil {
+ return x.ScalerMetadata
+ }
+ return nil
+}
+
+type IsActiveResponse struct {
+ state protoimpl.MessageState `protogen:"open.v1"`
+ Result bool
`protobuf:"varint,1,opt,name=result,proto3" json:"result,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
+}
+
+func (x *IsActiveResponse) Reset() {
+ *x = IsActiveResponse{}
+ mi := &file_externalscaler_proto_msgTypes[1]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+}
+
+func (x *IsActiveResponse) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*IsActiveResponse) ProtoMessage() {}
+
+func (x *IsActiveResponse) ProtoReflect() protoreflect.Message {
+ mi := &file_externalscaler_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 IsActiveResponse.ProtoReflect.Descriptor instead.
+func (*IsActiveResponse) Descriptor() ([]byte, []int) {
+ return file_externalscaler_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *IsActiveResponse) GetResult() bool {
+ if x != nil {
+ return x.Result
+ }
+ return false
+}
+
+type GetMetricSpecResponse struct {
+ state protoimpl.MessageState `protogen:"open.v1"`
+ MetricSpecs []*MetricSpec
`protobuf:"bytes,1,rep,name=metricSpecs,proto3" json:"metricSpecs,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
+}
+
+func (x *GetMetricSpecResponse) Reset() {
+ *x = GetMetricSpecResponse{}
+ mi := &file_externalscaler_proto_msgTypes[2]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+}
+
+func (x *GetMetricSpecResponse) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*GetMetricSpecResponse) ProtoMessage() {}
+
+func (x *GetMetricSpecResponse) ProtoReflect() protoreflect.Message {
+ mi := &file_externalscaler_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 GetMetricSpecResponse.ProtoReflect.Descriptor instead.
+func (*GetMetricSpecResponse) Descriptor() ([]byte, []int) {
+ return file_externalscaler_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *GetMetricSpecResponse) GetMetricSpecs() []*MetricSpec {
+ if x != nil {
+ return x.MetricSpecs
+ }
+ return nil
+}
+
+type MetricSpec struct {
+ state protoimpl.MessageState `protogen:"open.v1"`
+ MetricName string
`protobuf:"bytes,1,opt,name=metricName,proto3" json:"metricName,omitempty"`
+ // deprecated, use targetSizeFloat instead
+ TargetSize int64 `protobuf:"varint,2,opt,name=targetSize,proto3"
json:"targetSize,omitempty"`
+ TargetSizeFloat float64
`protobuf:"fixed64,3,opt,name=targetSizeFloat,proto3"
json:"targetSizeFloat,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
+}
+
+func (x *MetricSpec) Reset() {
+ *x = MetricSpec{}
+ mi := &file_externalscaler_proto_msgTypes[3]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+}
+
+func (x *MetricSpec) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*MetricSpec) ProtoMessage() {}
+
+func (x *MetricSpec) ProtoReflect() protoreflect.Message {
+ mi := &file_externalscaler_proto_msgTypes[3]
+ 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 MetricSpec.ProtoReflect.Descriptor instead.
+func (*MetricSpec) Descriptor() ([]byte, []int) {
+ return file_externalscaler_proto_rawDescGZIP(), []int{3}
+}
+
+func (x *MetricSpec) GetMetricName() string {
+ if x != nil {
+ return x.MetricName
+ }
+ return ""
+}
+
+func (x *MetricSpec) GetTargetSize() int64 {
+ if x != nil {
+ return x.TargetSize
+ }
+ return 0
+}
+
+func (x *MetricSpec) GetTargetSizeFloat() float64 {
+ if x != nil {
+ return x.TargetSizeFloat
+ }
+ return 0
+}
+
+type GetMetricsRequest struct {
+ state protoimpl.MessageState `protogen:"open.v1"`
+ ScaledObjectRef *ScaledObjectRef
`protobuf:"bytes,1,opt,name=scaledObjectRef,proto3"
json:"scaledObjectRef,omitempty"`
+ MetricName string
`protobuf:"bytes,2,opt,name=metricName,proto3" json:"metricName,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
+}
+
+func (x *GetMetricsRequest) Reset() {
+ *x = GetMetricsRequest{}
+ mi := &file_externalscaler_proto_msgTypes[4]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+}
+
+func (x *GetMetricsRequest) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*GetMetricsRequest) ProtoMessage() {}
+
+func (x *GetMetricsRequest) ProtoReflect() protoreflect.Message {
+ mi := &file_externalscaler_proto_msgTypes[4]
+ if x != nil {
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ if ms.LoadMessageInfo() == nil {
+ ms.StoreMessageInfo(mi)
+ }
+ return ms
+ }
+ return mi.MessageOf(x)
+}
+
+// Deprecated: Use GetMetricsRequest.ProtoReflect.Descriptor instead.
+func (*GetMetricsRequest) Descriptor() ([]byte, []int) {
+ return file_externalscaler_proto_rawDescGZIP(), []int{4}
+}
+
+func (x *GetMetricsRequest) GetScaledObjectRef() *ScaledObjectRef {
+ if x != nil {
+ return x.ScaledObjectRef
+ }
+ return nil
+}
+
+func (x *GetMetricsRequest) GetMetricName() string {
+ if x != nil {
+ return x.MetricName
+ }
+ return ""
+}
+
+type GetMetricsResponse struct {
+ state protoimpl.MessageState `protogen:"open.v1"`
+ MetricValues []*MetricValue
`protobuf:"bytes,1,rep,name=metricValues,proto3" json:"metricValues,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
+}
+
+func (x *GetMetricsResponse) Reset() {
+ *x = GetMetricsResponse{}
+ mi := &file_externalscaler_proto_msgTypes[5]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+}
+
+func (x *GetMetricsResponse) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*GetMetricsResponse) ProtoMessage() {}
+
+func (x *GetMetricsResponse) ProtoReflect() protoreflect.Message {
+ mi := &file_externalscaler_proto_msgTypes[5]
+ 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 GetMetricsResponse.ProtoReflect.Descriptor instead.
+func (*GetMetricsResponse) Descriptor() ([]byte, []int) {
+ return file_externalscaler_proto_rawDescGZIP(), []int{5}
+}
+
+func (x *GetMetricsResponse) GetMetricValues() []*MetricValue {
+ if x != nil {
+ return x.MetricValues
+ }
+ return nil
+}
+
+type MetricValue struct {
+ state protoimpl.MessageState `protogen:"open.v1"`
+ MetricName string
`protobuf:"bytes,1,opt,name=metricName,proto3" json:"metricName,omitempty"`
+ // deprecated, use metricValueFloat instead
+ MetricValue int64
`protobuf:"varint,2,opt,name=metricValue,proto3" json:"metricValue,omitempty"`
+ MetricValueFloat float64
`protobuf:"fixed64,3,opt,name=metricValueFloat,proto3"
json:"metricValueFloat,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
+}
+
+func (x *MetricValue) Reset() {
+ *x = MetricValue{}
+ mi := &file_externalscaler_proto_msgTypes[6]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+}
+
+func (x *MetricValue) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*MetricValue) ProtoMessage() {}
+
+func (x *MetricValue) ProtoReflect() protoreflect.Message {
+ mi := &file_externalscaler_proto_msgTypes[6]
+ 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 MetricValue.ProtoReflect.Descriptor instead.
+func (*MetricValue) Descriptor() ([]byte, []int) {
+ return file_externalscaler_proto_rawDescGZIP(), []int{6}
+}
+
+func (x *MetricValue) GetMetricName() string {
+ if x != nil {
+ return x.MetricName
+ }
+ return ""
+}
+
+func (x *MetricValue) GetMetricValue() int64 {
+ if x != nil {
+ return x.MetricValue
+ }
+ return 0
+}
+
+func (x *MetricValue) GetMetricValueFloat() float64 {
+ if x != nil {
+ return x.MetricValueFloat
+ }
+ return 0
+}
+
+var File_externalscaler_proto protoreflect.FileDescriptor
+
+const file_externalscaler_proto_rawDesc = "" +
+ "\n" +
+ "\x14externalscaler.proto\x12\x0eexternalscaler\"\xe3\x01\n" +
+ "\x0fScaledObjectRef\x12\x12\n" +
+ "\x04name\x18\x01 \x01(\tR\x04name\x12\x1c\n" +
+ "\tnamespace\x18\x02 \x01(\tR\tnamespace\x12[\n" +
+ "\x0escalerMetadata\x18\x03
\x03(\v23.externalscaler.ScaledObjectRef.ScalerMetadataEntryR\x0escalerMetadata\x1aA\n"
+
+ "\x13ScalerMetadataEntry\x12\x10\n" +
+ "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
+ "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"*\n" +
+ "\x10IsActiveResponse\x12\x16\n" +
+ "\x06result\x18\x01 \x01(\bR\x06result\"U\n" +
+ "\x15GetMetricSpecResponse\x12<\n" +
+ "\vmetricSpecs\x18\x01
\x03(\v2\x1a.externalscaler.MetricSpecR\vmetricSpecs\"v\n" +
+ "\n" +
+ "MetricSpec\x12\x1e\n" +
+ "\n" +
+ "metricName\x18\x01 \x01(\tR\n" +
+ "metricName\x12\x1e\n" +
+ "\n" +
+ "targetSize\x18\x02 \x01(\x03R\n" +
+ "targetSize\x12(\n" +
+ "\x0ftargetSizeFloat\x18\x03 \x01(\x01R\x0ftargetSizeFloat\"~\n" +
+ "\x11GetMetricsRequest\x12I\n" +
+ "\x0fscaledObjectRef\x18\x01
\x01(\v2\x1f.externalscaler.ScaledObjectRefR\x0fscaledObjectRef\x12\x1e\n" +
+ "\n" +
+ "metricName\x18\x02 \x01(\tR\n" +
+ "metricName\"U\n" +
+ "\x12GetMetricsResponse\x12?\n" +
+ "\fmetricValues\x18\x01
\x03(\v2\x1b.externalscaler.MetricValueR\fmetricValues\"{\n" +
+ "\vMetricValue\x12\x1e\n" +
+ "\n" +
+ "metricName\x18\x01 \x01(\tR\n" +
+ "metricName\x12 \n" +
+ "\vmetricValue\x18\x02 \x01(\x03R\vmetricValue\x12*\n" +
+ "\x10metricValueFloat\x18\x03
\x01(\x01R\x10metricValueFloat2\xcc\x03\n" +
+ "\x0eExternalScaler\x12O\n" +
+ "\bIsActive\x12\x1f.externalscaler.ScaledObjectRef\x1a
.externalscaler.IsActiveResponse\"\x00\x12W\n" +
+ "\x0eStreamIsActive\x12\x1f.externalscaler.ScaledObjectRef\x1a
.externalscaler.IsActiveResponse\"\x000\x01\x12Y\n" +
+
"\rGetMetricSpec\x12\x1f.externalscaler.ScaledObjectRef\x1a%.externalscaler.GetMetricSpecResponse\"\x00\x12U\n"
+
+ "\n" +
+
"GetMetrics\x12!.externalscaler.GetMetricsRequest\x1a\".externalscaler.GetMetricsResponse\"\x00\x12^\n"
+
+
"\x10StreamMetricSpec\x12\x1f.externalscaler.ScaledObjectRef\x1a%.externalscaler.GetMetricSpecResponse\"\x000\x01B\x12Z\x10.;externalscalerb\x06proto3"
+
+var (
+ file_externalscaler_proto_rawDescOnce sync.Once
+ file_externalscaler_proto_rawDescData []byte
+)
+
+func file_externalscaler_proto_rawDescGZIP() []byte {
+ file_externalscaler_proto_rawDescOnce.Do(func() {
+ file_externalscaler_proto_rawDescData =
protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_externalscaler_proto_rawDesc),
len(file_externalscaler_proto_rawDesc)))
+ })
+ return file_externalscaler_proto_rawDescData
+}
+
+var file_externalscaler_proto_msgTypes = make([]protoimpl.MessageInfo, 8)
+var file_externalscaler_proto_goTypes = []any{
+ (*ScaledObjectRef)(nil), // 0: externalscaler.ScaledObjectRef
+ (*IsActiveResponse)(nil), // 1: externalscaler.IsActiveResponse
+ (*GetMetricSpecResponse)(nil), // 2:
externalscaler.GetMetricSpecResponse
+ (*MetricSpec)(nil), // 3: externalscaler.MetricSpec
+ (*GetMetricsRequest)(nil), // 4: externalscaler.GetMetricsRequest
+ (*GetMetricsResponse)(nil), // 5: externalscaler.GetMetricsResponse
+ (*MetricValue)(nil), // 6: externalscaler.MetricValue
+ nil, // 7:
externalscaler.ScaledObjectRef.ScalerMetadataEntry
+}
+var file_externalscaler_proto_depIdxs = []int32{
+ 7, // 0: externalscaler.ScaledObjectRef.scalerMetadata:type_name ->
externalscaler.ScaledObjectRef.ScalerMetadataEntry
+ 3, // 1: externalscaler.GetMetricSpecResponse.metricSpecs:type_name ->
externalscaler.MetricSpec
+ 0, // 2: externalscaler.GetMetricsRequest.scaledObjectRef:type_name ->
externalscaler.ScaledObjectRef
+ 6, // 3: externalscaler.GetMetricsResponse.metricValues:type_name ->
externalscaler.MetricValue
+ 0, // 4: externalscaler.ExternalScaler.IsActive:input_type ->
externalscaler.ScaledObjectRef
+ 0, // 5: externalscaler.ExternalScaler.StreamIsActive:input_type ->
externalscaler.ScaledObjectRef
+ 0, // 6: externalscaler.ExternalScaler.GetMetricSpec:input_type ->
externalscaler.ScaledObjectRef
+ 4, // 7: externalscaler.ExternalScaler.GetMetrics:input_type ->
externalscaler.GetMetricsRequest
+ 0, // 8: externalscaler.ExternalScaler.StreamMetricSpec:input_type ->
externalscaler.ScaledObjectRef
+ 1, // 9: externalscaler.ExternalScaler.IsActive:output_type ->
externalscaler.IsActiveResponse
+ 1, // 10: externalscaler.ExternalScaler.StreamIsActive:output_type ->
externalscaler.IsActiveResponse
+ 2, // 11: externalscaler.ExternalScaler.GetMetricSpec:output_type ->
externalscaler.GetMetricSpecResponse
+ 5, // 12: externalscaler.ExternalScaler.GetMetrics:output_type ->
externalscaler.GetMetricsResponse
+ 2, // 13: externalscaler.ExternalScaler.StreamMetricSpec:output_type ->
externalscaler.GetMetricSpecResponse
+ 9, // [9:14] is the sub-list for method output_type
+ 4, // [4:9] is the sub-list for method input_type
+ 4, // [4:4] is the sub-list for extension type_name
+ 4, // [4:4] is the sub-list for extension extendee
+ 0, // [0:4] is the sub-list for field type_name
+}
+
+func init() { file_externalscaler_proto_init() }
+func file_externalscaler_proto_init() {
+ if File_externalscaler_proto != nil {
+ return
+ }
+ type x struct{}
+ out := protoimpl.TypeBuilder{
+ File: protoimpl.DescBuilder{
+ GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+ RawDescriptor:
unsafe.Slice(unsafe.StringData(file_externalscaler_proto_rawDesc),
len(file_externalscaler_proto_rawDesc)),
+ NumEnums: 0,
+ NumMessages: 8,
+ NumExtensions: 0,
+ NumServices: 1,
+ },
+ GoTypes: file_externalscaler_proto_goTypes,
+ DependencyIndexes: file_externalscaler_proto_depIdxs,
+ MessageInfos: file_externalscaler_proto_msgTypes,
+ }.Build()
+ File_externalscaler_proto = out.File
+ file_externalscaler_proto_goTypes = nil
+ file_externalscaler_proto_depIdxs = nil
+}
diff --git
a/dubbod/discovery/pkg/activation/externalscaler/externalscaler.proto
b/dubbod/discovery/pkg/activation/externalscaler/externalscaler.proto
new file mode 100644
index 00000000..ae546a6c
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/externalscaler/externalscaler.proto
@@ -0,0 +1,75 @@
+// Vendored from github.com/kedacore/keda
+// pkg/scalers/externalscaler/externalscaler.proto (Apache-2.0).
+//
+// This is the contract KEDA dials when a ScaledObject declares an "external"
+// or "external-push" trigger. Regenerate with:
+//
+// protoc -I . --go_out=. --go_opt=paths=source_relative \
+// --go-grpc_out=. --go-grpc_opt=paths=source_relative externalscaler.proto
+
+syntax = "proto3";
+
+package externalscaler;
+option go_package = ".;externalscaler";
+
+service ExternalScaler {
+ rpc IsActive(ScaledObjectRef) returns (IsActiveResponse) {}
+ rpc StreamIsActive(ScaledObjectRef) returns (stream IsActiveResponse) {}
+ rpc GetMetricSpec(ScaledObjectRef) returns (GetMetricSpecResponse) {}
+ rpc GetMetrics(GetMetricsRequest) returns (GetMetricsResponse) {}
+
+ // Optional. When implemented, the scaler pushes updated metric specs
+ // whenever HPA target values change. KEDA updates the cached specs and
+ // syncs the HPA accordingly. If unimplemented (returns Unimplemented),
+ // KEDA silently falls back to the existing pull-based behavior.
+ //
+ // Servers implementing this RPC must follow two rules:
+ // 1. Send the current metric specs immediately when the stream is
+ // opened (like StreamIsActive), not only when values change.
+ // 2. Keep GetMetricSpec returning the same current values as the
+ // latest streamed update. KEDA may rebuild its internal scaler
+ // state (e.g. after a scaler error) and falls back to
+ // GetMetricSpec until the next streamed update; if the two
+ // diverge, HPA targets can temporarily revert to stale values.
+ rpc StreamMetricSpec(ScaledObjectRef) returns (stream
GetMetricSpecResponse) {}
+}
+
+message ScaledObjectRef {
+ string name = 1;
+ string namespace = 2;
+ map<string, string> scalerMetadata = 3;
+}
+
+message IsActiveResponse {
+ bool result = 1;
+}
+
+message GetMetricSpecResponse {
+ repeated MetricSpec metricSpecs = 1;
+}
+
+message MetricSpec {
+ string metricName = 1;
+
+ // deprecated, use targetSizeFloat instead
+ int64 targetSize = 2;
+ double targetSizeFloat = 3;
+}
+
+message GetMetricsRequest {
+ ScaledObjectRef scaledObjectRef = 1;
+ string metricName = 2;
+}
+
+message GetMetricsResponse {
+ repeated MetricValue metricValues = 1;
+}
+
+message MetricValue {
+ string metricName = 1;
+
+ // deprecated, use metricValueFloat instead
+ int64 metricValue = 2;
+
+ double metricValueFloat = 3;
+}
\ No newline at end of file
diff --git
a/dubbod/discovery/pkg/activation/externalscaler/externalscaler_grpc.pb.go
b/dubbod/discovery/pkg/activation/externalscaler/externalscaler_grpc.pb.go
new file mode 100644
index 00000000..b59333dd
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/externalscaler/externalscaler_grpc.pb.go
@@ -0,0 +1,315 @@
+// Vendored from github.com/kedacore/keda
+// pkg/scalers/externalscaler/externalscaler.proto (Apache-2.0).
+//
+// This is the contract KEDA dials when a ScaledObject declares an "external"
+// or "external-push" trigger. Regenerate with:
+//
+// protoc -I . --go_out=. --go_opt=paths=source_relative \
+// --go-grpc_out=. --go-grpc_opt=paths=source_relative externalscaler.proto
+
+// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.6.2
+// - protoc v6.33.0
+// source: externalscaler.proto
+
+package externalscaler
+
+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 (
+ ExternalScaler_IsActive_FullMethodName =
"/externalscaler.ExternalScaler/IsActive"
+ ExternalScaler_StreamIsActive_FullMethodName =
"/externalscaler.ExternalScaler/StreamIsActive"
+ ExternalScaler_GetMetricSpec_FullMethodName =
"/externalscaler.ExternalScaler/GetMetricSpec"
+ ExternalScaler_GetMetrics_FullMethodName =
"/externalscaler.ExternalScaler/GetMetrics"
+ ExternalScaler_StreamMetricSpec_FullMethodName =
"/externalscaler.ExternalScaler/StreamMetricSpec"
+)
+
+// ExternalScalerClient is the client API for ExternalScaler 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.
+type ExternalScalerClient interface {
+ IsActive(ctx context.Context, in *ScaledObjectRef, opts
...grpc.CallOption) (*IsActiveResponse, error)
+ StreamIsActive(ctx context.Context, in *ScaledObjectRef, opts
...grpc.CallOption) (grpc.ServerStreamingClient[IsActiveResponse], error)
+ GetMetricSpec(ctx context.Context, in *ScaledObjectRef, opts
...grpc.CallOption) (*GetMetricSpecResponse, error)
+ GetMetrics(ctx context.Context, in *GetMetricsRequest, opts
...grpc.CallOption) (*GetMetricsResponse, error)
+ // Optional. When implemented, the scaler pushes updated metric specs
+ // whenever HPA target values change. KEDA updates the cached specs and
+ // syncs the HPA accordingly. If unimplemented (returns Unimplemented),
+ // KEDA silently falls back to the existing pull-based behavior.
+ //
+ // Servers implementing this RPC must follow two rules:
+ // 1. Send the current metric specs immediately when the stream is
+ // opened (like StreamIsActive), not only when values change.
+ // 2. Keep GetMetricSpec returning the same current values as the
+ // latest streamed update. KEDA may rebuild its internal scaler
+ // state (e.g. after a scaler error) and falls back to
+ // GetMetricSpec until the next streamed update; if the two
+ // diverge, HPA targets can temporarily revert to stale values.
+ StreamMetricSpec(ctx context.Context, in *ScaledObjectRef, opts
...grpc.CallOption) (grpc.ServerStreamingClient[GetMetricSpecResponse], error)
+}
+
+type externalScalerClient struct {
+ cc grpc.ClientConnInterface
+}
+
+func NewExternalScalerClient(cc grpc.ClientConnInterface) ExternalScalerClient
{
+ return &externalScalerClient{cc}
+}
+
+func (c *externalScalerClient) IsActive(ctx context.Context, in
*ScaledObjectRef, opts ...grpc.CallOption) (*IsActiveResponse, error) {
+ cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+ out := new(IsActiveResponse)
+ err := c.cc.Invoke(ctx, ExternalScaler_IsActive_FullMethodName, in,
out, cOpts...)
+ if err != nil {
+ return nil, err
+ }
+ return out, nil
+}
+
+func (c *externalScalerClient) StreamIsActive(ctx context.Context, in
*ScaledObjectRef, opts ...grpc.CallOption)
(grpc.ServerStreamingClient[IsActiveResponse], error) {
+ cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+ stream, err := c.cc.NewStream(ctx,
&ExternalScaler_ServiceDesc.Streams[0],
ExternalScaler_StreamIsActive_FullMethodName, cOpts...)
+ if err != nil {
+ return nil, err
+ }
+ x := &grpc.GenericClientStream[ScaledObjectRef,
IsActiveResponse]{ClientStream: stream}
+ if err := x.ClientStream.SendMsg(in); err != nil {
+ return nil, err
+ }
+ if err := x.ClientStream.CloseSend(); err != nil {
+ return nil, err
+ }
+ 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 ExternalScaler_StreamIsActiveClient =
grpc.ServerStreamingClient[IsActiveResponse]
+
+func (c *externalScalerClient) GetMetricSpec(ctx context.Context, in
*ScaledObjectRef, opts ...grpc.CallOption) (*GetMetricSpecResponse, error) {
+ cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+ out := new(GetMetricSpecResponse)
+ err := c.cc.Invoke(ctx, ExternalScaler_GetMetricSpec_FullMethodName,
in, out, cOpts...)
+ if err != nil {
+ return nil, err
+ }
+ return out, nil
+}
+
+func (c *externalScalerClient) GetMetrics(ctx context.Context, in
*GetMetricsRequest, opts ...grpc.CallOption) (*GetMetricsResponse, error) {
+ cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+ out := new(GetMetricsResponse)
+ err := c.cc.Invoke(ctx, ExternalScaler_GetMetrics_FullMethodName, in,
out, cOpts...)
+ if err != nil {
+ return nil, err
+ }
+ return out, nil
+}
+
+func (c *externalScalerClient) StreamMetricSpec(ctx context.Context, in
*ScaledObjectRef, opts ...grpc.CallOption)
(grpc.ServerStreamingClient[GetMetricSpecResponse], error) {
+ cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+ stream, err := c.cc.NewStream(ctx,
&ExternalScaler_ServiceDesc.Streams[1],
ExternalScaler_StreamMetricSpec_FullMethodName, cOpts...)
+ if err != nil {
+ return nil, err
+ }
+ x := &grpc.GenericClientStream[ScaledObjectRef,
GetMetricSpecResponse]{ClientStream: stream}
+ if err := x.ClientStream.SendMsg(in); err != nil {
+ return nil, err
+ }
+ if err := x.ClientStream.CloseSend(); err != nil {
+ return nil, err
+ }
+ 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 ExternalScaler_StreamMetricSpecClient =
grpc.ServerStreamingClient[GetMetricSpecResponse]
+
+// ExternalScalerServer is the server API for ExternalScaler service.
+// All implementations must embed UnimplementedExternalScalerServer
+// for forward compatibility.
+type ExternalScalerServer interface {
+ IsActive(context.Context, *ScaledObjectRef) (*IsActiveResponse, error)
+ StreamIsActive(*ScaledObjectRef,
grpc.ServerStreamingServer[IsActiveResponse]) error
+ GetMetricSpec(context.Context, *ScaledObjectRef)
(*GetMetricSpecResponse, error)
+ GetMetrics(context.Context, *GetMetricsRequest) (*GetMetricsResponse,
error)
+ // Optional. When implemented, the scaler pushes updated metric specs
+ // whenever HPA target values change. KEDA updates the cached specs and
+ // syncs the HPA accordingly. If unimplemented (returns Unimplemented),
+ // KEDA silently falls back to the existing pull-based behavior.
+ //
+ // Servers implementing this RPC must follow two rules:
+ // 1. Send the current metric specs immediately when the stream is
+ // opened (like StreamIsActive), not only when values change.
+ // 2. Keep GetMetricSpec returning the same current values as the
+ // latest streamed update. KEDA may rebuild its internal scaler
+ // state (e.g. after a scaler error) and falls back to
+ // GetMetricSpec until the next streamed update; if the two
+ // diverge, HPA targets can temporarily revert to stale values.
+ StreamMetricSpec(*ScaledObjectRef,
grpc.ServerStreamingServer[GetMetricSpecResponse]) error
+ mustEmbedUnimplementedExternalScalerServer()
+}
+
+// UnimplementedExternalScalerServer 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 UnimplementedExternalScalerServer struct{}
+
+func (UnimplementedExternalScalerServer) IsActive(context.Context,
*ScaledObjectRef) (*IsActiveResponse, error) {
+ return nil, status.Error(codes.Unimplemented, "method IsActive not
implemented")
+}
+func (UnimplementedExternalScalerServer) StreamIsActive(*ScaledObjectRef,
grpc.ServerStreamingServer[IsActiveResponse]) error {
+ return status.Error(codes.Unimplemented, "method StreamIsActive not
implemented")
+}
+func (UnimplementedExternalScalerServer) GetMetricSpec(context.Context,
*ScaledObjectRef) (*GetMetricSpecResponse, error) {
+ return nil, status.Error(codes.Unimplemented, "method GetMetricSpec not
implemented")
+}
+func (UnimplementedExternalScalerServer) GetMetrics(context.Context,
*GetMetricsRequest) (*GetMetricsResponse, error) {
+ return nil, status.Error(codes.Unimplemented, "method GetMetrics not
implemented")
+}
+func (UnimplementedExternalScalerServer) StreamMetricSpec(*ScaledObjectRef,
grpc.ServerStreamingServer[GetMetricSpecResponse]) error {
+ return status.Error(codes.Unimplemented, "method StreamMetricSpec not
implemented")
+}
+func (UnimplementedExternalScalerServer)
mustEmbedUnimplementedExternalScalerServer() {}
+func (UnimplementedExternalScalerServer) testEmbeddedByValue()
{}
+
+// UnsafeExternalScalerServer may be embedded to opt out of forward
compatibility for this service.
+// Use of this interface is not recommended, as added methods to
ExternalScalerServer will
+// result in compilation errors.
+type UnsafeExternalScalerServer interface {
+ mustEmbedUnimplementedExternalScalerServer()
+}
+
+func RegisterExternalScalerServer(s grpc.ServiceRegistrar, srv
ExternalScalerServer) {
+ // If the following call panics, it indicates
UnimplementedExternalScalerServer 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(&ExternalScaler_ServiceDesc, srv)
+}
+
+func _ExternalScaler_IsActive_Handler(srv interface{}, ctx context.Context,
dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor)
(interface{}, error) {
+ in := new(ScaledObjectRef)
+ if err := dec(in); err != nil {
+ return nil, err
+ }
+ if interceptor == nil {
+ return srv.(ExternalScalerServer).IsActive(ctx, in)
+ }
+ info := &grpc.UnaryServerInfo{
+ Server: srv,
+ FullMethod: ExternalScaler_IsActive_FullMethodName,
+ }
+ handler := func(ctx context.Context, req interface{}) (interface{},
error) {
+ return srv.(ExternalScalerServer).IsActive(ctx,
req.(*ScaledObjectRef))
+ }
+ return interceptor(ctx, in, info, handler)
+}
+
+func _ExternalScaler_StreamIsActive_Handler(srv interface{}, stream
grpc.ServerStream) error {
+ m := new(ScaledObjectRef)
+ if err := stream.RecvMsg(m); err != nil {
+ return err
+ }
+ return srv.(ExternalScalerServer).StreamIsActive(m,
&grpc.GenericServerStream[ScaledObjectRef, IsActiveResponse]{ServerStream:
stream})
+}
+
+// This type alias is provided for backwards compatibility with existing code
that references the prior non-generic stream type by name.
+type ExternalScaler_StreamIsActiveServer =
grpc.ServerStreamingServer[IsActiveResponse]
+
+func _ExternalScaler_GetMetricSpec_Handler(srv interface{}, ctx
context.Context, dec func(interface{}) error, interceptor
grpc.UnaryServerInterceptor) (interface{}, error) {
+ in := new(ScaledObjectRef)
+ if err := dec(in); err != nil {
+ return nil, err
+ }
+ if interceptor == nil {
+ return srv.(ExternalScalerServer).GetMetricSpec(ctx, in)
+ }
+ info := &grpc.UnaryServerInfo{
+ Server: srv,
+ FullMethod: ExternalScaler_GetMetricSpec_FullMethodName,
+ }
+ handler := func(ctx context.Context, req interface{}) (interface{},
error) {
+ return srv.(ExternalScalerServer).GetMetricSpec(ctx,
req.(*ScaledObjectRef))
+ }
+ return interceptor(ctx, in, info, handler)
+}
+
+func _ExternalScaler_GetMetrics_Handler(srv interface{}, ctx context.Context,
dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor)
(interface{}, error) {
+ in := new(GetMetricsRequest)
+ if err := dec(in); err != nil {
+ return nil, err
+ }
+ if interceptor == nil {
+ return srv.(ExternalScalerServer).GetMetrics(ctx, in)
+ }
+ info := &grpc.UnaryServerInfo{
+ Server: srv,
+ FullMethod: ExternalScaler_GetMetrics_FullMethodName,
+ }
+ handler := func(ctx context.Context, req interface{}) (interface{},
error) {
+ return srv.(ExternalScalerServer).GetMetrics(ctx,
req.(*GetMetricsRequest))
+ }
+ return interceptor(ctx, in, info, handler)
+}
+
+func _ExternalScaler_StreamMetricSpec_Handler(srv interface{}, stream
grpc.ServerStream) error {
+ m := new(ScaledObjectRef)
+ if err := stream.RecvMsg(m); err != nil {
+ return err
+ }
+ return srv.(ExternalScalerServer).StreamMetricSpec(m,
&grpc.GenericServerStream[ScaledObjectRef, GetMetricSpecResponse]{ServerStream:
stream})
+}
+
+// This type alias is provided for backwards compatibility with existing code
that references the prior non-generic stream type by name.
+type ExternalScaler_StreamMetricSpecServer =
grpc.ServerStreamingServer[GetMetricSpecResponse]
+
+// ExternalScaler_ServiceDesc is the grpc.ServiceDesc for ExternalScaler
service.
+// It's only intended for direct use with grpc.RegisterService,
+// and not to be introspected or modified (even as a copy)
+var ExternalScaler_ServiceDesc = grpc.ServiceDesc{
+ ServiceName: "externalscaler.ExternalScaler",
+ HandlerType: (*ExternalScalerServer)(nil),
+ Methods: []grpc.MethodDesc{
+ {
+ MethodName: "IsActive",
+ Handler: _ExternalScaler_IsActive_Handler,
+ },
+ {
+ MethodName: "GetMetricSpec",
+ Handler: _ExternalScaler_GetMetricSpec_Handler,
+ },
+ {
+ MethodName: "GetMetrics",
+ Handler: _ExternalScaler_GetMetrics_Handler,
+ },
+ },
+ Streams: []grpc.StreamDesc{
+ {
+ StreamName: "StreamIsActive",
+ Handler: _ExternalScaler_StreamIsActive_Handler,
+ ServerStreams: true,
+ },
+ {
+ StreamName: "StreamMetricSpec",
+ Handler: _ExternalScaler_StreamMetricSpec_Handler,
+ ServerStreams: true,
+ },
+ },
+ Metadata: "externalscaler.proto",
+}
diff --git a/dubbod/discovery/pkg/activation/grpc_test.go
b/dubbod/discovery/pkg/activation/grpc_test.go
new file mode 100644
index 00000000..8329cb61
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/grpc_test.go
@@ -0,0 +1,147 @@
+// 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/codes"
+ "google.golang.org/grpc/credentials/insecure"
+ "google.golang.org/grpc/status"
+)
+
+// startScaler serves the scaler over a real gRPC connection, so the wire
+// contract is exercised the way KEDA drives it rather than through a stub.
+func startScaler(t *testing.T, registry *Registry)
externalscaler.ExternalScalerClient {
+ t.Helper()
+
+ listener, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatal(err)
+ }
+ server := grpc.NewServer()
+ externalscaler.RegisterExternalScalerServer(server, NewScaler(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 externalscaler.NewExternalScalerClient(connection)
+}
+
+func TestScalerOverGRPCDrivesActivation(t *testing.T) {
+ registry := NewRegistry()
+ client := startScaler(t, registry)
+
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+
+ stream, err := client.StreamIsActive(ctx, serviceRef())
+ if err != nil {
+ t.Fatalf("StreamIsActive() error = %v", err)
+ }
+
+ first, err := stream.Recv()
+ if err != nil {
+ t.Fatalf("first Recv() error = %v", err)
+ }
+ if first.GetResult() {
+ t.Fatal("first streamed value = true, want false with no
demand")
+ }
+
+ registry.Report("gateway-a", orders, 4)
+ next, err := stream.Recv()
+ if err != nil {
+ t.Fatalf("Recv() after demand error = %v", err)
+ }
+ if !next.GetResult() {
+ t.Fatal("streamed value after demand = false, want true")
+ }
+
+ // The polling path has to agree with the pushed one; KEDA falls back
to it
+ // whenever it rebuilds its scaler state.
+ active, err := client.IsActive(ctx, serviceRef())
+ if err != nil {
+ t.Fatalf("IsActive() error = %v", err)
+ }
+ if !active.GetResult() {
+ t.Fatal("IsActive() = false while the stream reported active")
+ }
+
+ spec, err := client.GetMetricSpec(ctx, serviceRef())
+ if err != nil {
+ t.Fatalf("GetMetricSpec() error = %v", err)
+ }
+ metrics, err := client.GetMetrics(ctx,
&externalscaler.GetMetricsRequest{
+ ScaledObjectRef: serviceRef(),
+ MetricName: spec.GetMetricSpecs()[0].GetMetricName(),
+ })
+ if err != nil {
+ t.Fatalf("GetMetrics() error = %v", err)
+ }
+ if got := metrics.GetMetricValues()[0].GetMetricValueFloat(); got != 4 {
+ t.Fatalf("metricValueFloat = %v, want 4", got)
+ }
+}
+
+func TestScalerOverGRPCReportsInvalidTriggerAsStatusError(t *testing.T) {
+ client := startScaler(t, NewRegistry())
+
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+
+ if _, err := client.IsActive(ctx, ref(map[string]string{}));
status.Code(err) != codes.InvalidArgument {
+ t.Fatalf("IsActive() error = %v, want InvalidArgument", err)
+ }
+
+ stream, err := client.StreamIsActive(ctx, ref(map[string]string{}))
+ if err != nil {
+ t.Fatalf("StreamIsActive() error = %v", err)
+ }
+ if _, err := stream.Recv(); status.Code(err) != codes.InvalidArgument {
+ t.Fatalf("Recv() error = %v, want InvalidArgument", err)
+ }
+}
+
+// StreamMetricSpec is optional. KEDA falls back to polling GetMetricSpec when
+// it is unimplemented, so returning Unimplemented has to be the actual
+// behavior rather than a crash or an empty stream.
+func TestStreamMetricSpecIsUnimplemented(t *testing.T) {
+ client := startScaler(t, NewRegistry())
+
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+
+ stream, err := client.StreamMetricSpec(ctx, serviceRef())
+ if err != nil {
+ t.Fatalf("StreamMetricSpec() error = %v", err)
+ }
+ if _, err := stream.Recv(); status.Code(err) != codes.Unimplemented {
+ t.Fatalf("Recv() error = %v, want Unimplemented", err)
+ }
+}
diff --git a/dubbod/discovery/pkg/activation/policy.go
b/dubbod/discovery/pkg/activation/policy.go
new file mode 100644
index 00000000..5c2483c1
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/policy.go
@@ -0,0 +1,220 @@
+// 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"
+ "strings"
+
+ metav1alpha1 "github.com/kdubbo/api/meta/v1alpha1"
+ networking "github.com/kdubbo/api/networking/v1alpha3"
+ clientnetworking
"github.com/kdubbo/client-go/pkg/apis/networking/v1alpha3"
+)
+
+// Condition types reported on a ServiceActivationPolicy. Each answers a
+// different question, and all four have to hold before a request will actually
+// be held and replayed. Collapsing them into one would leave an operator
+// guessing which half of the path is broken.
+const (
+ // ConditionAccepted covers the policy itself: well formed, and
pointing at
+ // objects that exist.
+ ConditionAccepted = "Accepted"
+
+ // ConditionEligible covers the target: whether its protocols can be
held
+ // and replayed at all.
+ ConditionEligible = "Eligible"
+
+ // ConditionScalerReady covers KEDA: whether it is listening for
activation
+ // on this target.
+ ConditionScalerReady = "ScalerReady"
+
+ // ConditionActivatorReady covers the gateways: whether any of them is
in a
+ // position to catch a request for this target.
+ ConditionActivatorReady = "ActivatorReady"
+)
+
+const (
+ conditionTrue = "True"
+ conditionFalse = "False"
+)
+
+// ServiceLookup reports whether the target Service exists. It is an interface
+// so policy evaluation can be tested without a cluster.
+type ServiceLookup interface {
+ HasService(namespace, name string) bool
+}
+
+// StreamLookup reports whether KEDA holds an activation stream for a target.
+type StreamLookup interface {
+ Subscribed(Target) bool
+}
+
+// ReporterLookup reports how many gateways stand ready to hold requests for a
+// target.
+type ReporterLookup interface {
+ Reporters(Target) int
+}
+
+// PolicyEvaluator turns a policy plus live state into the conditions published
+// on its status.
+type PolicyEvaluator struct {
+ Services ServiceLookup
+ Streams StreamLookup
+ Reporters ReporterLookup
+}
+
+// Evaluate returns the conditions for one policy, in a stable order so an
+// unchanged policy does not produce a status update on every resync.
+func (e PolicyEvaluator) Evaluate(policy
*clientnetworking.ServiceActivationPolicy) []*metav1alpha1.DubboCondition {
+ generation := policy.GetGeneration()
+ spec := &policy.Spec
+
+ accepted, reason := e.accepted(policy.GetNamespace(), spec)
+ conditions := []*metav1alpha1.DubboCondition{
+ condition(ConditionAccepted, accepted, reason, generation),
+ }
+
+ // The remaining conditions describe a path that only exists once the
policy
+ // is accepted. Reporting them against an unresolved target would invent
+ // answers about a Service that may not be the intended one.
+ if !accepted {
+ conditions = append(conditions,
+ condition(ConditionEligible, false,
"PolicyNotAccepted", generation),
+ condition(ConditionScalerReady, false,
"PolicyNotAccepted", generation),
+ condition(ConditionActivatorReady, false,
"PolicyNotAccepted", generation),
+ )
+ return conditions
+ }
+
+ target := targetOf(policy)
+ eligible, eligibleReason := eligible(spec)
+ conditions = append(conditions, condition(ConditionEligible, eligible,
eligibleReason, generation))
+
+ scalerReady := e.Streams != nil && e.Streams.Subscribed(target)
+ conditions = append(conditions,
+ condition(ConditionScalerReady, scalerReady,
scalerReason(scalerReady), generation))
+
+ activatorReady := e.Reporters != nil && e.Reporters.Reporters(target) > 0
+ conditions = append(conditions,
+ condition(ConditionActivatorReady, activatorReady,
activatorReason(activatorReady), generation))
+
+ return conditions
+}
+
+func (e PolicyEvaluator) accepted(namespace string, spec
*networking.ServiceActivationPolicy) (bool, string) {
+ target := spec.GetTargetRef()
+ if target == nil || strings.TrimSpace(target.GetName()) == "" {
+ return false, "TargetRefMissing"
+ }
+ // Only Services have endpoints to wait on; anything else would leave
the
+ // gateway holding requests for something that never becomes routable.
+ if kind := target.GetKind(); kind != "" && kind != "Service" {
+ return false, "TargetKindUnsupported"
+ }
+ if group := target.GetGroup(); group != "" {
+ return false, "TargetGroupUnsupported"
+ }
+ if autoscaler := spec.GetAutoscalerRef(); autoscaler == nil ||
strings.TrimSpace(autoscaler.GetName()) == "" {
+ return false, "AutoscalerRefMissing"
+ }
+ if e.Services != nil && !e.Services.HasService(namespace,
target.GetName()) {
+ return false, "TargetServiceNotFound"
+ }
+ return true, "Accepted"
+}
+
+// eligible rejects protocols that cannot survive being held. A stream cannot
be
+// replayed once the backend is up, so holding one only delays the failure.
+func eligible(spec *networking.ServiceActivationPolicy) (bool, string) {
+ for _, protocol := range spec.GetProtocols() {
+ switch protocol {
+ case
networking.ActivationProtocol_ACTIVATION_PROTOCOL_UNSPECIFIED,
+ networking.ActivationProtocol_HTTP,
+ networking.ActivationProtocol_GRPC_UNARY,
+ networking.ActivationProtocol_TRIPLE_UNARY:
+ default:
+ return false, "ProtocolNotActivatable"
+ }
+ }
+ return true, "Eligible"
+}
+
+func scalerReason(ready bool) string {
+ if ready {
+ return "ScalerSubscribed"
+ }
+ // The usual cause is a ScaledObject that does not point its external
+ // trigger at this scaler, or points it at a different Service.
+ return "ScalerNotSubscribed"
+}
+
+func activatorReason(ready bool) string {
+ if ready {
+ return "GatewayReporting"
+ }
+ return "NoGatewayReporting"
+}
+
+// targetOf resolves the Service a policy activates. The namespace comes from
+// the policy, so a policy can never reach across namespaces into a Service it
+// does not own.
+func targetOf(policy *clientnetworking.ServiceActivationPolicy) Target {
+ return Target{
+ Namespace: policy.GetNamespace(),
+ Name: policy.Spec.GetTargetRef().GetName(),
+ }
+}
+
+func condition(conditionType string, ok bool, reason string, generation int64)
*metav1alpha1.DubboCondition {
+ value := conditionFalse
+ if ok {
+ value = conditionTrue
+ }
+ return &metav1alpha1.DubboCondition{
+ Type: conditionType,
+ Status: value,
+ Reason: reason,
+ ObservedGeneration: generation,
+ }
+}
+
+// SameConditions reports whether two condition sets carry the same
information,
+// so an unchanged policy is not written back on every resync. Status writes
are
+// not free: a resync storm across every policy is a self-inflicted load spike
+// on the API server.
+func SameConditions(a, b []*metav1alpha1.DubboCondition) bool {
+ if len(a) != len(b) {
+ return false
+ }
+ for i := range a {
+ if a[i].GetType() != b[i].GetType() ||
+ a[i].GetStatus() != b[i].GetStatus() ||
+ a[i].GetReason() != b[i].GetReason() ||
+ a[i].GetObservedGeneration() !=
b[i].GetObservedGeneration() {
+ return false
+ }
+ }
+ return true
+}
+
+// Summary renders the conditions for logs and dubboctl output.
+func Summary(conditions []*metav1alpha1.DubboCondition) string {
+ parts := make([]string, 0, len(conditions))
+ for _, item := range conditions {
+ parts = append(parts, fmt.Sprintf("%s=%s(%s)", item.GetType(),
item.GetStatus(), item.GetReason()))
+ }
+ return strings.Join(parts, " ")
+}
diff --git a/dubbod/discovery/pkg/activation/policy_test.go
b/dubbod/discovery/pkg/activation/policy_test.go
new file mode 100644
index 00000000..ab8ede0d
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/policy_test.go
@@ -0,0 +1,284 @@
+// 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"
+ "testing"
+ "time"
+
+ networking "github.com/kdubbo/api/networking/v1alpha3"
+ clientnetworking
"github.com/kdubbo/client-go/pkg/apis/networking/v1alpha3"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+type services map[string]bool
+
+func (s services) HasService(namespace, name string) bool { return
s[namespace+"/"+name] }
+
+type subscribed map[Target]bool
+
+func (s subscribed) Subscribed(target Target) bool { return s[target] }
+
+type reporters map[Target]int
+
+func (r reporters) Reporters(target Target) int { return r[target] }
+
+// policy builds the kube wrapper from its parts. The generated spec is a proto
+// message with an embedded mutex, so it is assembled in place rather than
+// copied in from a variable.
+func policy(
+ target *networking.PolicyTargetReference,
+ autoscaler *networking.AutoscalerReference,
+ protocols ...networking.ActivationProtocol,
+) *clientnetworking.ServiceActivationPolicy {
+ return &clientnetworking.ServiceActivationPolicy{
+ ObjectMeta: metav1.ObjectMeta{Name: "orders", Namespace: "app",
Generation: 7},
+ Spec: networking.ServiceActivationPolicy{
+ TargetRef: target,
+ AutoscalerRef: autoscaler,
+ Protocols: protocols,
+ },
+ }
+}
+
+func serviceTarget(name string) *networking.PolicyTargetReference {
+ return &networking.PolicyTargetReference{Kind: "Service", Name: name}
+}
+
+func autoscaler(name string) *networking.AutoscalerReference {
+ return &networking.AutoscalerReference{Name: name}
+}
+
+func validPolicy() *clientnetworking.ServiceActivationPolicy {
+ return policy(serviceTarget("orders"), autoscaler("orders"))
+}
+
+func conditionsByType(t *testing.T, evaluator PolicyEvaluator, p
*clientnetworking.ServiceActivationPolicy) map[string]string {
+ t.Helper()
+ out := map[string]string{}
+ for _, item := range evaluator.Evaluate(p) {
+ if item.GetObservedGeneration() != p.GetGeneration() {
+ t.Fatalf("%s observedGeneration = %d, want %d",
+ item.GetType(), item.GetObservedGeneration(),
p.GetGeneration())
+ }
+ out[item.GetType()] = item.GetStatus() + "/" + item.GetReason()
+ }
+ return out
+}
+
+func TestEvaluateReportsEveryConditionTrueWhenThePathIsComplete(t *testing.T) {
+ evaluator := PolicyEvaluator{
+ Services: services{"app/orders": true},
+ Streams: subscribed{orders: true},
+ Reporters: reporters{orders: 2},
+ }
+
+ got := conditionsByType(t, evaluator, validPolicy())
+ for _, conditionType := range []string{
+ ConditionAccepted, ConditionEligible, ConditionScalerReady,
ConditionActivatorReady,
+ } {
+ if status := got[conditionType]; status[:4] != "True" {
+ t.Fatalf("%s = %q, want True", conditionType, status)
+ }
+ }
+}
+
+func TestEvaluateRejectsMalformedPolicies(t *testing.T) {
+ evaluator := PolicyEvaluator{Services: services{"app/orders": true}}
+
+ tests := []struct {
+ name string
+ target *networking.PolicyTargetReference
+ autoscaler *networking.AutoscalerReference
+ reason string
+ }{
+ {
+ name: "no target",
+ autoscaler: autoscaler("orders"),
+ reason: "TargetRefMissing",
+ },
+ {
+ name: "blank target name",
+ target: serviceTarget(" "),
+ autoscaler: autoscaler("orders"),
+ reason: "TargetRefMissing",
+ },
+ {
+ // Only Services have endpoints to wait on. Anything
else leaves the
+ // gateway holding requests for something that never
becomes routable.
+ name: "non-Service target",
+ target: &networking.PolicyTargetReference{Kind:
"Deployment", Name: "orders"},
+ autoscaler: autoscaler("orders"),
+ reason: "TargetKindUnsupported",
+ },
+ {
+ name: "no autoscaler",
+ target: serviceTarget("orders"),
+ reason: "AutoscalerRefMissing",
+ },
+ }
+
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ got := conditionsByType(t, evaluator,
policy(test.target, test.autoscaler))
+ if want := "False/" + test.reason;
got[ConditionAccepted] != want {
+ t.Fatalf("Accepted = %q, want %q",
got[ConditionAccepted], want)
+ }
+ // The rest of the path does not exist yet; reporting
on it would
+ // invent answers about a target that was never
resolved.
+ for _, conditionType := range []string{
+ ConditionEligible, ConditionScalerReady,
ConditionActivatorReady,
+ } {
+ if want := "False/PolicyNotAccepted";
got[conditionType] != want {
+ t.Fatalf("%s = %q, want %q",
conditionType, got[conditionType], want)
+ }
+ }
+ })
+ }
+}
+
+func TestEvaluateReportsMissingTargetService(t *testing.T) {
+ evaluator := PolicyEvaluator{Services: services{}}
+ got := conditionsByType(t, evaluator, validPolicy())
+ if want := "False/TargetServiceNotFound"; got[ConditionAccepted] !=
want {
+ t.Fatalf("Accepted = %q, want %q", got[ConditionAccepted], want)
+ }
+}
+
+// A stream cannot be replayed once the backend is up, so a policy naming a
+// streaming protocol must not look ready.
+func TestEvaluateRejectsProtocolsThatCannotBeHeld(t *testing.T) {
+ evaluator := PolicyEvaluator{
+ Services: services{"app/orders": true},
+ Streams: subscribed{orders: true},
+ Reporters: reporters{orders: 1},
+ }
+
+ unknown := policy(serviceTarget("orders"), autoscaler("orders"),
networking.ActivationProtocol(99))
+
+ got := conditionsByType(t, evaluator, unknown)
+ if got[ConditionAccepted][:4] != "True" {
+ t.Fatalf("Accepted = %q, want True", got[ConditionAccepted])
+ }
+ if want := "False/ProtocolNotActivatable"; got[ConditionEligible] !=
want {
+ t.Fatalf("Eligible = %q, want %q", got[ConditionEligible], want)
+ }
+}
+
+func TestEvaluateAcceptsEveryActivatableProtocol(t *testing.T) {
+ evaluator := PolicyEvaluator{Services: services{"app/orders": true}}
+ all := policy(serviceTarget("orders"), autoscaler("orders"),
+ networking.ActivationProtocol_HTTP,
+ networking.ActivationProtocol_GRPC_UNARY,
+ networking.ActivationProtocol_TRIPLE_UNARY,
+ )
+
+ got := conditionsByType(t, evaluator, all)
+ if want := "True/Eligible"; got[ConditionEligible] != want {
+ t.Fatalf("Eligible = %q, want %q", got[ConditionEligible], want)
+ }
+}
+
+// A ScaledObject that never points its trigger here looks identical to a
+// working one until the first request is dropped, so it has to be reported.
+func TestEvaluateReportsScalerAndActivatorSeparately(t *testing.T) {
+ evaluator := PolicyEvaluator{
+ Services: services{"app/orders": true},
+ Streams: subscribed{},
+ Reporters: reporters{orders: 1},
+ }
+ got := conditionsByType(t, evaluator, validPolicy())
+ if want := "False/ScalerNotSubscribed"; got[ConditionScalerReady] !=
want {
+ t.Fatalf("ScalerReady = %q, want %q",
got[ConditionScalerReady], want)
+ }
+ if want := "True/GatewayReporting"; got[ConditionActivatorReady] !=
want {
+ t.Fatalf("ActivatorReady = %q, want %q",
got[ConditionActivatorReady], want)
+ }
+
+ evaluator = PolicyEvaluator{
+ Services: services{"app/orders": true},
+ Streams: subscribed{orders: true},
+ Reporters: reporters{},
+ }
+ got = conditionsByType(t, evaluator, validPolicy())
+ if want := "True/ScalerSubscribed"; got[ConditionScalerReady] != want {
+ t.Fatalf("ScalerReady = %q, want %q",
got[ConditionScalerReady], want)
+ }
+ if want := "False/NoGatewayReporting"; got[ConditionActivatorReady] !=
want {
+ t.Fatalf("ActivatorReady = %q, want %q",
got[ConditionActivatorReady], want)
+ }
+}
+
+// Evaluate feeds a change detector, so an unchanged policy must produce an
+// identical result; otherwise every resync writes status back to the API
server.
+func TestEvaluateIsStableAcrossCalls(t *testing.T) {
+ evaluator := PolicyEvaluator{
+ Services: services{"app/orders": true},
+ Streams: subscribed{orders: true},
+ Reporters: reporters{orders: 1},
+ }
+ target := validPolicy()
+
+ if !SameConditions(evaluator.Evaluate(target),
evaluator.Evaluate(target)) {
+ t.Fatal("Evaluate() produced different conditions for an
unchanged policy")
+ }
+
+ // A real change must still be detected.
+ changed := PolicyEvaluator{
+ Services: services{"app/orders": true},
+ Streams: subscribed{},
+ Reporters: reporters{orders: 1},
+ }
+ if SameConditions(evaluator.Evaluate(target), changed.Evaluate(target))
{
+ t.Fatal("SameConditions() reported no change after the scaler
unsubscribed")
+ }
+}
+
+// ScalerReady is derived from live KEDA streams, so the tracking has to follow
+// the stream's lifetime exactly: a policy must stop looking ready the moment
+// KEDA stops listening.
+func TestScalerTracksSubscriptionForTheController(t *testing.T) {
+ scaler := NewScaler(NewRegistry())
+ if scaler.Subscribed(orders) {
+ t.Fatal("Subscribed() = true before any stream opened")
+ }
+
+ ctx, cancel := context.WithCancel(context.Background())
+ stream := newFakeStream(ctx)
+ done := make(chan struct{})
+ go func() {
+ defer close(done)
+ _ = scaler.StreamIsActive(serviceRef(), stream)
+ }()
+
+ // The first send happens after the subscription is registered, so
receiving
+ // it means tracking is in place.
+ stream.next(t)
+ if !scaler.Subscribed(orders) {
+ t.Fatal("Subscribed() = false while a stream is open")
+ }
+
+ cancel()
+ select {
+ case <-done:
+ case <-time.After(2 * time.Second):
+ t.Fatal("StreamIsActive did not return after the context was
canceled")
+ }
+ if scaler.Subscribed(orders) {
+ t.Fatal("Subscribed() = true after the stream closed")
+ }
+}
diff --git a/dubbod/discovery/pkg/activation/scaler.go
b/dubbod/discovery/pkg/activation/scaler.go
new file mode 100644
index 00000000..bfb0092b
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/scaler.go
@@ -0,0 +1,262 @@
+// 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 serves KEDA's external scaler contract for Services that
+// are scaled to zero.
+//
+// KEDA owns the replica count. This package only answers "is anything waiting
+// on this Service, and how much", so KEDA can take a Service from zero to one
+// when a request arrives for it. Nothing here writes replicas, which is what
+// keeps a Service from being driven by two controllers at once.
+package activation
+
+import (
+ "context"
+ "fmt"
+ "strconv"
+ "strings"
+ "sync"
+
+
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/activation/externalscaler"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
+)
+
+const (
+ // serviceMetadataKey names the Service a ScaledObject trigger
activates.
+ // It is required: guessing it from the ScaledObject name would silently
+ // activate the wrong workload when the two do not match.
+ serviceMetadataKey = "service"
+
+ // namespaceMetadataKey overrides the namespace, for the rare
ScaledObject
+ // that lives apart from the Service it scales. Defaults to the
+ // ScaledObject's own namespace.
+ namespaceMetadataKey = "namespace"
+
+ // targetPendingMetadataKey is the pending-request count HPA aims to
keep
+ // per replica once the workload is past zero.
+ targetPendingMetadataKey = "targetPendingRequests"
+
+ defaultTargetPendingRequests = 1.0
+
+ // metricPrefix keeps this scaler's metric distinguishable from the
other
+ // triggers on the same ScaledObject.
+ metricPrefix = "dubbo-activation"
+)
+
+// Scaler implements KEDA's ExternalScaler over a DemandSource.
+type Scaler struct {
+ externalscaler.UnimplementedExternalScalerServer
+
+ demand DemandSource
+
+ // streams counts the open StreamIsActive calls per target. A target
with
+ // none is one KEDA is not listening to, which is the difference
between a
+ // policy that will activate and one that only looks like it will.
+ streamsMu sync.Mutex
+ streams map[Target]int
+}
+
+func NewScaler(demand DemandSource) *Scaler {
+ return &Scaler{
+ demand: demand,
+ streams: map[Target]int{},
+ }
+}
+
+// Subscribed reports whether KEDA currently holds an activation stream for the
+// target. The policy controller surfaces this as a status condition, because
+// otherwise a misconfigured ScaledObject looks identical to a working one
until
+// the first request is dropped.
+func (s *Scaler) Subscribed(target Target) bool {
+ s.streamsMu.Lock()
+ defer s.streamsMu.Unlock()
+ return s.streams[target] > 0
+}
+
+func (s *Scaler) streamOpened(target Target) {
+ s.streamsMu.Lock()
+ defer s.streamsMu.Unlock()
+ s.streams[target]++
+}
+
+func (s *Scaler) streamClosed(target Target) {
+ s.streamsMu.Lock()
+ defer s.streamsMu.Unlock()
+ if s.streams[target] <= 1 {
+ delete(s.streams, target)
+ return
+ }
+ s.streams[target]--
+}
+
+// trigger is one ScaledObject's external trigger, resolved to what this scaler
+// needs to answer for it.
+type trigger struct {
+ target Target
+ metricName string
+ targetPending float64
+}
+
+func (s *Scaler) IsActive(_ context.Context, ref
*externalscaler.ScaledObjectRef) (*externalscaler.IsActiveResponse, error) {
+ parsed, err := parseTrigger(ref)
+ if err != nil {
+ return nil, err
+ }
+ return &externalscaler.IsActiveResponse{
+ Result: s.demand.Pending(parsed.target) > 0,
+ }, nil
+}
+
+// StreamIsActive pushes activation as it happens, so a request held at the
+// gateway does not wait out KEDA's polling interval before the workload is
+// even asked to start.
+func (s *Scaler) StreamIsActive(ref *externalscaler.ScaledObjectRef, stream
externalscaler.ExternalScaler_StreamIsActiveServer) error {
+ parsed, err := parseTrigger(ref)
+ if err != nil {
+ return err
+ }
+
+ s.streamOpened(parsed.target)
+ defer s.streamClosed(parsed.target)
+
+ // Subscribe before the first read, or demand arriving in between would
be
+ // reported by neither the initial send nor an update.
+ updates, cancel := s.demand.Subscribe(parsed.target)
+ defer cancel()
+
+ // KEDA expects the current state as soon as the stream opens, not only
on
+ // the next change.
+ if err := stream.Send(&externalscaler.IsActiveResponse{
+ Result: s.demand.Pending(parsed.target) > 0,
+ }); err != nil {
+ return err
+ }
+
+ ctx := stream.Context()
+ for {
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case pending, ok := <-updates:
+ if !ok {
+ return nil
+ }
+ if err :=
stream.Send(&externalscaler.IsActiveResponse{Result: pending > 0}); err != nil {
+ return err
+ }
+ }
+ }
+}
+
+func (s *Scaler) GetMetricSpec(_ context.Context, ref
*externalscaler.ScaledObjectRef) (*externalscaler.GetMetricSpecResponse, error)
{
+ parsed, err := parseTrigger(ref)
+ if err != nil {
+ return nil, err
+ }
+ return &externalscaler.GetMetricSpecResponse{
+ MetricSpecs: []*externalscaler.MetricSpec{{
+ MetricName: parsed.metricName,
+ TargetSizeFloat: parsed.targetPending,
+ // TargetSize is deprecated upstream but still read by
older KEDA
+ // releases, so both are set and kept consistent.
+ TargetSize: int64(parsed.targetPending),
+ }},
+ }, nil
+}
+
+func (s *Scaler) GetMetrics(_ context.Context, request
*externalscaler.GetMetricsRequest) (*externalscaler.GetMetricsResponse, error) {
+ parsed, err := parseTrigger(request.GetScaledObjectRef())
+ if err != nil {
+ return nil, err
+ }
+ // KEDA echoes back the name from GetMetricSpec. A mismatch means the
two
+ // calls disagree about what is being measured, which would otherwise
show
+ // up as a workload that scales on the wrong signal.
+ if name := request.GetMetricName(); name != parsed.metricName {
+ return nil, status.Errorf(codes.InvalidArgument,
+ "unknown metric %q for %s/%s, expected %q",
+ name, parsed.target.Namespace, parsed.target.Name,
parsed.metricName)
+ }
+
+ pending := float64(s.demand.Pending(parsed.target))
+ return &externalscaler.GetMetricsResponse{
+ MetricValues: []*externalscaler.MetricValue{{
+ MetricName: parsed.metricName,
+ MetricValueFloat: pending,
+ MetricValue: int64(pending),
+ }},
+ }, nil
+}
+
+func parseTrigger(ref *externalscaler.ScaledObjectRef) (trigger, error) {
+ if ref == nil {
+ return trigger{}, status.Error(codes.InvalidArgument, "missing
scaled object reference")
+ }
+ metadata := ref.GetScalerMetadata()
+
+ service := strings.TrimSpace(metadata[serviceMetadataKey])
+ if service == "" {
+ return trigger{}, status.Errorf(codes.InvalidArgument,
+ "scaled object %s/%s: trigger metadata %q is required",
+ ref.GetNamespace(), ref.GetName(), serviceMetadataKey)
+ }
+
+ namespace := strings.TrimSpace(metadata[namespaceMetadataKey])
+ if namespace == "" {
+ namespace = strings.TrimSpace(ref.GetNamespace())
+ }
+ if namespace == "" {
+ return trigger{}, status.Errorf(codes.InvalidArgument,
+ "scaled object %s: namespace is unknown, set trigger
metadata %q",
+ ref.GetName(), namespaceMetadataKey)
+ }
+
+ targetPending, err :=
parseTargetPending(metadata[targetPendingMetadataKey])
+ if err != nil {
+ return trigger{}, status.Errorf(codes.InvalidArgument,
+ "scaled object %s/%s: %v", ref.GetNamespace(),
ref.GetName(), err)
+ }
+
+ target := Target{Namespace: namespace, Name: service}
+ return trigger{
+ target: target,
+ metricName: metricName(target),
+ targetPending: targetPending,
+ }, nil
+}
+
+func parseTargetPending(raw string) (float64, error) {
+ raw = strings.TrimSpace(raw)
+ if raw == "" {
+ return defaultTargetPendingRequests, nil
+ }
+ value, err := strconv.ParseFloat(raw, 64)
+ if err != nil {
+ return 0, fmt.Errorf("trigger metadata %q is not a number: %q",
targetPendingMetadataKey, raw)
+ }
+ // Zero or negative would make HPA divide by it and demand an unbounded
+ // replica count from a single held request.
+ if value <= 0 {
+ return 0, fmt.Errorf("trigger metadata %q must be greater than
zero, got %q", targetPendingMetadataKey, raw)
+ }
+ return value, nil
+}
+
+// metricName is derived from the target rather than the ScaledObject so two
+// triggers pointing at the same Service report the same series.
+func metricName(target Target) string {
+ return fmt.Sprintf("%s-%s-%s", metricPrefix, target.Namespace,
target.Name)
+}
diff --git a/dubbod/discovery/pkg/activation/scaler_test.go
b/dubbod/discovery/pkg/activation/scaler_test.go
new file mode 100644
index 00000000..03ed79c0
--- /dev/null
+++ b/dubbod/discovery/pkg/activation/scaler_test.go
@@ -0,0 +1,306 @@
+// 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"
+ "testing"
+ "time"
+
+
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/activation/externalscaler"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/metadata"
+ "google.golang.org/grpc/status"
+)
+
+func ref(scalerMetadata map[string]string) *externalscaler.ScaledObjectRef {
+ return &externalscaler.ScaledObjectRef{
+ Name: "orders",
+ Namespace: "app",
+ ScalerMetadata: scalerMetadata,
+ }
+}
+
+func serviceRef() *externalscaler.ScaledObjectRef {
+ return ref(map[string]string{serviceMetadataKey: "orders"})
+}
+
+func TestIsActiveReflectsPendingDemand(t *testing.T) {
+ registry := NewRegistry()
+ scaler := NewScaler(registry)
+
+ response, err := scaler.IsActive(context.Background(), serviceRef())
+ if err != nil {
+ t.Fatalf("IsActive() error = %v", err)
+ }
+ if response.GetResult() {
+ t.Fatal("IsActive() = true with no pending requests")
+ }
+
+ registry.Report("gateway-a", orders, 1)
+ response, err = scaler.IsActive(context.Background(), serviceRef())
+ if err != nil {
+ t.Fatalf("IsActive() error = %v", err)
+ }
+ if !response.GetResult() {
+ t.Fatal("IsActive() = false while a request is held")
+ }
+}
+
+// Guessing the target from the ScaledObject name would silently activate the
+// wrong workload whenever the two differ, so the metadata is required.
+func TestTriggerMetadataIsValidated(t *testing.T) {
+ scaler := NewScaler(NewRegistry())
+
+ tests := []struct {
+ name string
+ ref *externalscaler.ScaledObjectRef
+ }{
+ {name: "nil reference", ref: nil},
+ {name: "missing service", ref: ref(map[string]string{})},
+ {name: "blank service", ref:
ref(map[string]string{serviceMetadataKey: " "})},
+ {
+ name: "unparsable target",
+ ref: ref(map[string]string{
+ serviceMetadataKey: "orders",
+ targetPendingMetadataKey: "many",
+ }),
+ },
+ {
+ name: "zero target would divide by zero in HPA",
+ ref: ref(map[string]string{
+ serviceMetadataKey: "orders",
+ targetPendingMetadataKey: "0",
+ }),
+ },
+ {
+ name: "negative target",
+ ref: ref(map[string]string{
+ serviceMetadataKey: "orders",
+ targetPendingMetadataKey: "-1",
+ }),
+ },
+ }
+
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ if _, err := scaler.IsActive(context.Background(),
test.ref); status.Code(err) != codes.InvalidArgument {
+ t.Fatalf("IsActive() error = %v, want
InvalidArgument", err)
+ }
+ })
+ }
+}
+
+func TestNamespaceDefaultsToScaledObjectAndCanBeOverridden(t *testing.T) {
+ registry := NewRegistry()
+ scaler := NewScaler(registry)
+
+ // Demand recorded in the ScaledObject's own namespace is picked up
without
+ // any namespace metadata.
+ registry.Report("gateway-a", orders, 1)
+ response, err := scaler.IsActive(context.Background(), serviceRef())
+ if err != nil {
+ t.Fatalf("IsActive() error = %v", err)
+ }
+ if !response.GetResult() {
+ t.Fatal("IsActive() = false for a target in the ScaledObject
namespace")
+ }
+
+ // With an override, the same ScaledObject must look elsewhere and see
none.
+ overridden := ref(map[string]string{
+ serviceMetadataKey: "orders",
+ namespaceMetadataKey: "staging",
+ })
+ response, err = scaler.IsActive(context.Background(), overridden)
+ if err != nil {
+ t.Fatalf("IsActive() with override error = %v", err)
+ }
+ if response.GetResult() {
+ t.Fatal("IsActive() = true for a namespace with no demand")
+ }
+}
+
+func TestGetMetricSpecAdvertisesTargetAndName(t *testing.T) {
+ scaler := NewScaler(NewRegistry())
+
+ spec, err := scaler.GetMetricSpec(context.Background(), serviceRef())
+ if err != nil {
+ t.Fatalf("GetMetricSpec() error = %v", err)
+ }
+ if len(spec.GetMetricSpecs()) != 1 {
+ t.Fatalf("metric specs = %d, want 1",
len(spec.GetMetricSpecs()))
+ }
+ got := spec.GetMetricSpecs()[0]
+ if want := "dubbo-activation-app-orders"; got.GetMetricName() != want {
+ t.Fatalf("metric name = %q, want %q", got.GetMetricName(), want)
+ }
+ if got.GetTargetSizeFloat() != defaultTargetPendingRequests {
+ t.Fatalf("targetSizeFloat = %v, want %v",
got.GetTargetSizeFloat(), defaultTargetPendingRequests)
+ }
+ // Older KEDA releases still read the deprecated integer field, so it
must
+ // agree with the float rather than being left at zero.
+ if got.GetTargetSize() != int64(defaultTargetPendingRequests) {
+ t.Fatalf("targetSize = %d, want %d", got.GetTargetSize(),
int64(defaultTargetPendingRequests))
+ }
+
+ custom := ref(map[string]string{
+ serviceMetadataKey: "orders",
+ targetPendingMetadataKey: "5",
+ })
+ spec, err = scaler.GetMetricSpec(context.Background(), custom)
+ if err != nil {
+ t.Fatalf("GetMetricSpec() with custom target error = %v", err)
+ }
+ if got := spec.GetMetricSpecs()[0].GetTargetSizeFloat(); got != 5 {
+ t.Fatalf("targetSizeFloat = %v, want 5", got)
+ }
+}
+
+func TestGetMetricsReportsPendingUnderTheAdvertisedName(t *testing.T) {
+ registry := NewRegistry()
+ scaler := NewScaler(registry)
+ registry.Report("gateway-a", orders, 3)
+
+ spec, err := scaler.GetMetricSpec(context.Background(), serviceRef())
+ if err != nil {
+ t.Fatalf("GetMetricSpec() error = %v", err)
+ }
+ name := spec.GetMetricSpecs()[0].GetMetricName()
+
+ metrics, err := scaler.GetMetrics(context.Background(),
&externalscaler.GetMetricsRequest{
+ ScaledObjectRef: serviceRef(),
+ MetricName: name,
+ })
+ if err != nil {
+ t.Fatalf("GetMetrics() error = %v", err)
+ }
+ if len(metrics.GetMetricValues()) != 1 {
+ t.Fatalf("metric values = %d, want 1",
len(metrics.GetMetricValues()))
+ }
+ value := metrics.GetMetricValues()[0]
+ if value.GetMetricName() != name {
+ t.Fatalf("metric name = %q, want %q", value.GetMetricName(),
name)
+ }
+ if value.GetMetricValueFloat() != 3 {
+ t.Fatalf("metricValueFloat = %v, want 3",
value.GetMetricValueFloat())
+ }
+ if value.GetMetricValue() != 3 {
+ t.Fatalf("metricValue = %d, want 3", value.GetMetricValue())
+ }
+}
+
+// A name GetMetricSpec never advertised means the two calls disagree about
what
+// is being measured; scaling on that would use the wrong signal.
+func TestGetMetricsRejectsUnknownMetricName(t *testing.T) {
+ scaler := NewScaler(NewRegistry())
+
+ _, err := scaler.GetMetrics(context.Background(),
&externalscaler.GetMetricsRequest{
+ ScaledObjectRef: serviceRef(),
+ MetricName: "some-other-metric",
+ })
+ if status.Code(err) != codes.InvalidArgument {
+ t.Fatalf("GetMetrics() error = %v, want InvalidArgument", err)
+ }
+}
+
+// KEDA expects the current state as soon as the stream opens, not only on the
+// next change; otherwise a workload with requests already waiting stays at
zero.
+func TestStreamIsActiveSendsCurrentStateImmediately(t *testing.T) {
+ registry := NewRegistry()
+ scaler := NewScaler(registry)
+ registry.Report("gateway-a", orders, 2)
+
+ stream := newFakeStream(context.Background())
+ go func() { _ = scaler.StreamIsActive(serviceRef(), stream) }()
+
+ if got := stream.next(t); !got {
+ t.Fatal("first streamed value = false, want true for existing
demand")
+ }
+}
+
+func TestStreamIsActivePushesTransitions(t *testing.T) {
+ registry := NewRegistry()
+ scaler := NewScaler(registry)
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ stream := newFakeStream(ctx)
+ errCh := make(chan error, 1)
+ go func() { errCh <- scaler.StreamIsActive(serviceRef(), stream) }()
+
+ if got := stream.next(t); got {
+ t.Fatal("first streamed value = true, want false with no
demand")
+ }
+
+ registry.Report("gateway-a", orders, 1)
+ if got := stream.next(t); !got {
+ t.Fatal("streamed value after demand = false, want true")
+ }
+
+ registry.Report("gateway-a", orders, 0)
+ if got := stream.next(t); got {
+ t.Fatal("streamed value after drain = true, want false")
+ }
+
+ cancel()
+ select {
+ case <-errCh:
+ case <-time.After(2 * time.Second):
+ t.Fatal("StreamIsActive did not return after the stream context
was canceled")
+ }
+}
+
+func TestStreamIsActiveRejectsInvalidTrigger(t *testing.T) {
+ scaler := NewScaler(NewRegistry())
+ err := scaler.StreamIsActive(ref(map[string]string{}),
newFakeStream(context.Background()))
+ if status.Code(err) != codes.InvalidArgument {
+ t.Fatalf("StreamIsActive() error = %v, want InvalidArgument",
err)
+ }
+}
+
+// fakeStream stands in for the gRPC server stream. Sends are buffered so the
+// scaler is never blocked by the test's read pace.
+type fakeStream struct {
+ ctx context.Context
+ messages chan bool
+}
+
+func newFakeStream(ctx context.Context) *fakeStream {
+ return &fakeStream{ctx: ctx, messages: make(chan bool, 16)}
+}
+
+func (s *fakeStream) next(t *testing.T) bool {
+ t.Helper()
+ select {
+ case value := <-s.messages:
+ return value
+ case <-time.After(2 * time.Second):
+ t.Fatal("no message streamed")
+ return false
+ }
+}
+
+func (s *fakeStream) Send(response *externalscaler.IsActiveResponse) error {
+ s.messages <- response.GetResult()
+ return nil
+}
+
+func (s *fakeStream) Context() context.Context { return s.ctx }
+func (s *fakeStream) SetHeader(metadata.MD) error { return nil }
+func (s *fakeStream) SendHeader(metadata.MD) error { return nil }
+func (s *fakeStream) SetTrailer(metadata.MD) {}
+func (s *fakeStream) SendMsg(any) error { return nil }
+func (s *fakeStream) RecvMsg(any) error { return nil }
diff --git a/dubbod/discovery/pkg/config/kube/crdclient/types.gen.go
b/dubbod/discovery/pkg/config/kube/crdclient/types.gen.go
index 44ba1f33..6eaf61e3 100755
--- a/dubbod/discovery/pkg/config/kube/crdclient/types.gen.go
+++ b/dubbod/discovery/pkg/config/kube/crdclient/types.gen.go
@@ -85,6 +85,11 @@ func create(c kube.Client, cfg config.Config, objMeta
metav1.ObjectMeta) (metav1
ObjectMeta: objMeta,
Spec:
*(cfg.Spec.(*githubcomkdubboapisecurityv1alpha3.RequestAuthentication)),
}, metav1.CreateOptions{})
+ case gvk.ServiceActivationPolicy:
+ return
c.Dubbo().NetworkingV1alpha3().ServiceActivationPolicies(cfg.Namespace).Create(context.TODO(),
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceActivationPolicy{
+ ObjectMeta: objMeta,
+ Spec:
*(cfg.Spec.(*githubcomkdubboapinetworkingv1alpha3.ServiceActivationPolicy)),
+ }, metav1.CreateOptions{})
case gvk.ServiceEntry:
return
c.Dubbo().NetworkingV1alpha3().ServiceEntries(cfg.Namespace).Create(context.TODO(),
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceEntry{
ObjectMeta: objMeta,
@@ -157,6 +162,11 @@ func update(c kube.Client, cfg config.Config, objMeta
metav1.ObjectMeta) (metav1
ObjectMeta: objMeta,
Spec:
*(cfg.Spec.(*githubcomkdubboapisecurityv1alpha3.RequestAuthentication)),
}, metav1.UpdateOptions{})
+ case gvk.ServiceActivationPolicy:
+ return
c.Dubbo().NetworkingV1alpha3().ServiceActivationPolicies(cfg.Namespace).Update(context.TODO(),
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceActivationPolicy{
+ ObjectMeta: objMeta,
+ Spec:
*(cfg.Spec.(*githubcomkdubboapinetworkingv1alpha3.ServiceActivationPolicy)),
+ }, metav1.UpdateOptions{})
case gvk.ServiceEntry:
return
c.Dubbo().NetworkingV1alpha3().ServiceEntries(cfg.Namespace).Update(context.TODO(),
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceEntry{
ObjectMeta: objMeta,
@@ -224,6 +234,11 @@ func updateStatus(c kube.Client, cfg config.Config,
objMeta metav1.ObjectMeta) (
ObjectMeta: objMeta,
Status:
*(cfg.Status.(*githubcomkdubboapimetav1alpha1.DubboStatus)),
}, metav1.UpdateOptions{})
+ case gvk.ServiceActivationPolicy:
+ return
c.Dubbo().NetworkingV1alpha3().ServiceActivationPolicies(cfg.Namespace).UpdateStatus(context.TODO(),
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceActivationPolicy{
+ ObjectMeta: objMeta,
+ Status:
*(cfg.Status.(*githubcomkdubboapimetav1alpha1.DubboStatus)),
+ }, metav1.UpdateOptions{})
case gvk.ServiceEntry:
return
c.Dubbo().NetworkingV1alpha3().ServiceEntries(cfg.Namespace).UpdateStatus(context.TODO(),
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceEntry{
ObjectMeta: objMeta,
@@ -399,6 +414,21 @@ func patch(c kube.Client, orig config.Config, origMeta
metav1.ObjectMeta, mod co
}
return
c.Dubbo().SecurityV1alpha3().RequestAuthentications(orig.Namespace).
Patch(context.TODO(), orig.Name, typ, patchBytes,
metav1.PatchOptions{FieldManager: "pilot-discovery"})
+ case gvk.ServiceActivationPolicy:
+ oldRes :=
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceActivationPolicy{
+ ObjectMeta: origMeta,
+ Spec:
*(orig.Spec.(*githubcomkdubboapinetworkingv1alpha3.ServiceActivationPolicy)),
+ }
+ modRes :=
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceActivationPolicy{
+ ObjectMeta: modMeta,
+ Spec:
*(mod.Spec.(*githubcomkdubboapinetworkingv1alpha3.ServiceActivationPolicy)),
+ }
+ patchBytes, err := genPatchBytes(oldRes, modRes, typ)
+ if err != nil {
+ return nil, err
+ }
+ return
c.Dubbo().NetworkingV1alpha3().ServiceActivationPolicies(orig.Namespace).
+ Patch(context.TODO(), orig.Name, typ, patchBytes,
metav1.PatchOptions{FieldManager: "pilot-discovery"})
case gvk.ServiceEntry:
oldRes :=
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceEntry{
ObjectMeta: origMeta,
@@ -475,6 +505,8 @@ func delete(c kube.Client, typ config.GroupVersionKind,
name, namespace string,
return
c.GatewayAPI().GatewayV1beta1().ReferenceGrants(namespace).Delete(context.TODO(),
name, deleteOptions)
case gvk.RequestAuthentication:
return
c.Dubbo().SecurityV1alpha3().RequestAuthentications(namespace).Delete(context.TODO(),
name, deleteOptions)
+ case gvk.ServiceActivationPolicy:
+ return
c.Dubbo().NetworkingV1alpha3().ServiceActivationPolicies(namespace).Delete(context.TODO(),
name, deleteOptions)
case gvk.ServiceEntry:
return
c.Dubbo().NetworkingV1alpha3().ServiceEntries(namespace).Delete(context.TODO(),
name, deleteOptions)
case gvk.Telemetry:
@@ -967,6 +999,25 @@ var translationMap = map[config.GroupVersionKind]func(r
runtime.Object) config.C
Spec: obj,
}
},
+ gvk.ServiceActivationPolicy: func(r runtime.Object) config.Config {
+ obj :=
r.(*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceActivationPolicy)
+ return config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.ServiceActivationPolicy,
+ Name: obj.Name,
+ Namespace: obj.Namespace,
+ Labels: obj.Labels,
+ Annotations: obj.Annotations,
+ ResourceVersion: obj.ResourceVersion,
+ CreationTimestamp: obj.CreationTimestamp.Time,
+ OwnerReferences: obj.OwnerReferences,
+ UID: string(obj.UID),
+ Generation: obj.Generation,
+ },
+ Spec: &obj.Spec,
+ Status: &obj.Status,
+ }
+ },
gvk.ServiceEntry: func(r runtime.Object) config.Config {
obj :=
r.(*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceEntry)
return config.Config{
diff --git a/go.mod b/go.mod
index c10abf4c..4be1eb32 100644
--- a/go.mod
+++ b/go.mod
@@ -52,8 +52,8 @@ require (
github.com/hashicorp/go-multierror v1.1.1
github.com/hashicorp/golang-lru/v2 v2.0.7
github.com/heroku/color v0.0.6
- github.com/kdubbo/api v0.0.0-20260728161804-a5971782efe0
- github.com/kdubbo/client-go v0.0.0-20260729004545-0427ad75f167
+ github.com/kdubbo/api v0.0.0-20260806182421-754eac4e05d4
+ github.com/kdubbo/client-go v0.0.0-20260807013041-7abbf3125711
github.com/kdubbo/xds-api v0.0.0-20260728161804-af6dbc11367a
github.com/moby/moby/client v0.4.1
github.com/moby/term v0.5.2
diff --git a/go.sum b/go.sum
index 26b0dbb9..bbaa16fe 100644
--- a/go.sum
+++ b/go.sum
@@ -368,10 +368,10 @@ github.com/jtolds/gls v4.20.0+incompatible/go.mod
h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfV
github.com/julienschmidt/httprouter v1.2.0/go.mod
h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51
h1:Z9n2FFNUXsshfwJMBgNA0RU6/i7WVaAegv3PtuIHPMs=
github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51/go.mod
h1:CzGEWj7cYgsdH8dAjBGEr58BoE7ScuLd+fwFZ44+/x8=
-github.com/kdubbo/api v0.0.0-20260728161804-a5971782efe0
h1:t4bS0hQfyiu3QdsL+frOw/KbOjmvQERPTaziWHhLc3U=
-github.com/kdubbo/api v0.0.0-20260728161804-a5971782efe0/go.mod
h1:8BtJiIovg7QCPsCxXcw3gDf922VcvYq5ihOSvj49Rq8=
-github.com/kdubbo/client-go v0.0.0-20260729004545-0427ad75f167
h1:j5nY/UzzGztfLEPrXbNvohGO53PVaRhoIK1GrtrikOs=
-github.com/kdubbo/client-go v0.0.0-20260729004545-0427ad75f167/go.mod
h1:/wrQJoD+yhTTBi/5sC8rHLK+62mk4vOnLaYZ1Sf+DOQ=
+github.com/kdubbo/api v0.0.0-20260806182421-754eac4e05d4
h1:zewrXkOD5XpbL6+QWLAbxsiVFSvpAivN5UQOoMNXBpk=
+github.com/kdubbo/api v0.0.0-20260806182421-754eac4e05d4/go.mod
h1:8BtJiIovg7QCPsCxXcw3gDf922VcvYq5ihOSvj49Rq8=
+github.com/kdubbo/client-go v0.0.0-20260807013041-7abbf3125711
h1:ofQz1xUC5DZyh+3lVPIFdipjKxe6jQ9uQbNMYoLInaI=
+github.com/kdubbo/client-go v0.0.0-20260807013041-7abbf3125711/go.mod
h1:ogHgHroSROD3HwYPBKS8c2dpcfd4cZxpwZrzTSCXjXs=
github.com/kdubbo/xds-api v0.0.0-20260728161804-af6dbc11367a
h1:WfpZeq43xfNY+6icInQxNSNDNLsbDj3QebyvLD9OmL8=
github.com/kdubbo/xds-api v0.0.0-20260728161804-af6dbc11367a/go.mod
h1:o2HDUgL1ntaDbWomZ4cD2tt8jBamuG2qRtjXOa1zZ0Q=
github.com/kevinburke/ssh_config v1.2.0
h1:x584FjTGwHzMwvHx18PXxbBVzfnxogHaAReU4gf13a4=
diff --git a/pkg/config/schema/collections/collections.gen.go
b/pkg/config/schema/collections/collections.gen.go
index 6eb97852..d1a08f35 100755
--- a/pkg/config/schema/collections/collections.gen.go
+++ b/pkg/config/schema/collections/collections.gen.go
@@ -449,6 +449,21 @@ var (
ValidateProto: validation.EmptyValidate,
}.MustBuild()
+ ServiceActivationPolicy = resource.Builder{
+ Identifier: "ServiceActivationPolicy",
+ Group: "networking.dubbo.apache.org",
+ Kind: "ServiceActivationPolicy",
+ Plural: "serviceactivationpolicies",
+ Version: "v1alpha3",
+ Proto:
"dubbo.networking.v1alpha3.ServiceActivationPolicy", StatusProto:
"dubbo.meta.v1alpha1.DubboStatus",
+ ReflectType:
reflect.TypeOf(&githubcomkdubboapinetworkingv1alpha3.ServiceActivationPolicy{}).Elem(),
StatusType:
reflect.TypeOf(&githubcomkdubboapimetav1alpha1.DubboStatus{}).Elem(),
+ ProtoPackage: "github.com/kdubbo/api/networking/v1alpha3",
StatusPackage: "github.com/kdubbo/api/meta/v1alpha1",
+ ClusterScoped: false,
+ Synthetic: false,
+ Builtin: false,
+ ValidateProto: validation.EmptyValidate,
+ }.MustBuild()
+
ServiceEntry = resource.Builder{
Identifier: "ServiceEntry",
Group: "networking.dubbo.apache.org",
@@ -553,6 +568,7 @@ var (
MustAdd(Secret).
MustAdd(Service).
MustAdd(ServiceAccount).
+ MustAdd(ServiceActivationPolicy).
MustAdd(ServiceEntry).
MustAdd(StatefulSet).
MustAdd(Telemetry).
@@ -594,6 +610,7 @@ var (
MustAdd(FaultInjectionPolicy).
MustAdd(PeerAuthentication).
MustAdd(RequestAuthentication).
+ MustAdd(ServiceActivationPolicy).
MustAdd(ServiceEntry).
MustAdd(Telemetry).
MustAdd(WorkloadEntry).
@@ -611,6 +628,7 @@ var (
MustAdd(PeerAuthentication).
MustAdd(ReferenceGrant).
MustAdd(RequestAuthentication).
+ MustAdd(ServiceActivationPolicy).
MustAdd(ServiceEntry).
MustAdd(Telemetry).
MustAdd(WorkloadEntry).
@@ -628,6 +646,7 @@ var (
MustAdd(PeerAuthentication).
MustAdd(ReferenceGrant).
MustAdd(RequestAuthentication).
+ MustAdd(ServiceActivationPolicy).
MustAdd(ServiceEntry).
MustAdd(Telemetry).
MustAdd(WorkloadEntry).
diff --git a/pkg/config/schema/gvk/resources.gen.go
b/pkg/config/schema/gvk/resources.gen.go
index 27a3ccbb..90b93612 100755
--- a/pkg/config/schema/gvk/resources.gen.go
+++ b/pkg/config/schema/gvk/resources.gen.go
@@ -42,6 +42,7 @@ var (
Secret = config.GroupVersionKind{Group: "",
Version: "v1", Kind: "Secret"}
Service = config.GroupVersionKind{Group: "",
Version: "v1", Kind: "Service"}
ServiceAccount = config.GroupVersionKind{Group: "",
Version: "v1", Kind: "ServiceAccount"}
+ ServiceActivationPolicy = config.GroupVersionKind{Group:
"networking.dubbo.apache.org", Version: "v1alpha3", Kind:
"ServiceActivationPolicy"}
ServiceEntry = config.GroupVersionKind{Group:
"networking.dubbo.apache.org", Version: "v1alpha3", Kind: "ServiceEntry"}
StatefulSet = config.GroupVersionKind{Group: "apps",
Version: "v1", Kind: "StatefulSet"}
Telemetry = config.GroupVersionKind{Group:
"telemetry.dubbo.apache.org", Version: "v1alpha1", Kind: "Telemetry"}
@@ -116,6 +117,8 @@ func ToGVR(g config.GroupVersionKind)
(schema.GroupVersionResource, bool) {
return gvr.Service, true
case ServiceAccount:
return gvr.ServiceAccount, true
+ case ServiceActivationPolicy:
+ return gvr.ServiceActivationPolicy, true
case ServiceEntry:
return gvr.ServiceEntry, true
case StatefulSet:
@@ -187,6 +190,8 @@ func MustToKind(g config.GroupVersionKind) kind.Kind {
return kind.Service
case ServiceAccount:
return kind.ServiceAccount
+ case ServiceActivationPolicy:
+ return kind.ServiceActivationPolicy
case ServiceEntry:
return kind.ServiceEntry
case StatefulSet:
@@ -269,6 +274,8 @@ func FromGVR(g schema.GroupVersionResource)
(config.GroupVersionKind, bool) {
return Service, true
case gvr.ServiceAccount:
return ServiceAccount, true
+ case gvr.ServiceActivationPolicy:
+ return ServiceActivationPolicy, true
case gvr.ServiceEntry:
return ServiceEntry, true
case gvr.StatefulSet:
diff --git a/pkg/config/schema/gvr/resources.gen.go
b/pkg/config/schema/gvr/resources.gen.go
index e9188777..79d7ab4f 100755
--- a/pkg/config/schema/gvr/resources.gen.go
+++ b/pkg/config/schema/gvr/resources.gen.go
@@ -37,6 +37,7 @@ var (
Secret = schema.GroupVersionResource{Group: "",
Version: "v1", Resource: "secrets"}
Service = schema.GroupVersionResource{Group: "",
Version: "v1", Resource: "services"}
ServiceAccount = schema.GroupVersionResource{Group: "",
Version: "v1", Resource: "serviceaccounts"}
+ ServiceActivationPolicy = schema.GroupVersionResource{Group:
"networking.dubbo.apache.org", Version: "v1alpha3", Resource:
"serviceactivationpolicies"}
ServiceEntry = schema.GroupVersionResource{Group:
"networking.dubbo.apache.org", Version: "v1alpha3", Resource: "serviceentries"}
StatefulSet = schema.GroupVersionResource{Group:
"apps", Version: "v1", Resource: "statefulsets"}
Telemetry = schema.GroupVersionResource{Group:
"telemetry.dubbo.apache.org", Version: "v1alpha1", Resource: "telemetries"}
@@ -108,6 +109,8 @@ func IsClusterScoped(g schema.GroupVersionResource) bool {
return false
case ServiceAccount:
return false
+ case ServiceActivationPolicy:
+ return false
case ServiceEntry:
return false
case StatefulSet:
diff --git a/pkg/config/schema/kind/resources.gen.go
b/pkg/config/schema/kind/resources.gen.go
index 5e4e84b3..2eeffb3e 100755
--- a/pkg/config/schema/kind/resources.gen.go
+++ b/pkg/config/schema/kind/resources.gen.go
@@ -33,6 +33,7 @@ const (
Secret
Service
ServiceAccount
+ ServiceActivationPolicy
ServiceEntry
StatefulSet
Telemetry
@@ -100,6 +101,8 @@ func (k Kind) String() string {
return "Service"
case ServiceAccount:
return "ServiceAccount"
+ case ServiceActivationPolicy:
+ return "ServiceActivationPolicy"
case ServiceEntry:
return "ServiceEntry"
case StatefulSet:
@@ -175,6 +178,8 @@ func FromString(s string) Kind {
return Service
case "ServiceAccount":
return ServiceAccount
+ case "ServiceActivationPolicy":
+ return ServiceActivationPolicy
case "ServiceEntry":
return ServiceEntry
case "StatefulSet":
diff --git a/pkg/config/schema/kubeclient/resources.gen.go
b/pkg/config/schema/kubeclient/resources.gen.go
index c8ca80d3..217a0ec5 100755
--- a/pkg/config/schema/kubeclient/resources.gen.go
+++ b/pkg/config/schema/kubeclient/resources.gen.go
@@ -86,6 +86,8 @@ func GetWriteClient[T runtime.Object](c ClientGetter,
namespace string) ktypes.W
return
c.Kube().CoreV1().Services(namespace).(ktypes.WriteAPI[T])
case *k8sioapicorev1.ServiceAccount:
return
c.Kube().CoreV1().ServiceAccounts(namespace).(ktypes.WriteAPI[T])
+ case
*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceActivationPolicy:
+ return
c.Dubbo().NetworkingV1alpha3().ServiceActivationPolicies(namespace).(ktypes.WriteAPI[T])
case
*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceEntry:
return
c.Dubbo().NetworkingV1alpha3().ServiceEntries(namespace).(ktypes.WriteAPI[T])
case *k8sioapiappsv1.StatefulSet:
@@ -155,6 +157,8 @@ func GetClient[T, TL runtime.Object](c ClientGetter,
namespace string) ktypes.Re
return
c.Kube().CoreV1().Services(namespace).(ktypes.ReadWriteAPI[T, TL])
case *k8sioapicorev1.ServiceAccount:
return
c.Kube().CoreV1().ServiceAccounts(namespace).(ktypes.ReadWriteAPI[T, TL])
+ case
*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceActivationPolicy:
+ return
c.Dubbo().NetworkingV1alpha3().ServiceActivationPolicies(namespace).(ktypes.ReadWriteAPI[T,
TL])
case
*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceEntry:
return
c.Dubbo().NetworkingV1alpha3().ServiceEntries(namespace).(ktypes.ReadWriteAPI[T,
TL])
case *k8sioapiappsv1.StatefulSet:
@@ -224,6 +228,8 @@ func gvrToObject(g schema.GroupVersionResource)
runtime.Object {
return &k8sioapicorev1.Service{}
case gvr.ServiceAccount:
return &k8sioapicorev1.ServiceAccount{}
+ case gvr.ServiceActivationPolicy:
+ return
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceActivationPolicy{}
case gvr.ServiceEntry:
return
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceEntry{}
case gvr.StatefulSet:
@@ -426,6 +432,13 @@ func getInformerFiltered(c ClientGetter, opts
ktypes.InformerOptions, g schema.G
w = func(options metav1.ListOptions) (watch.Interface, error) {
return
c.Kube().CoreV1().ServiceAccounts(opts.Namespace).Watch(context.Background(),
options)
}
+ case gvr.ServiceActivationPolicy:
+ l = func(options metav1.ListOptions) (runtime.Object, error) {
+ return
c.Dubbo().NetworkingV1alpha3().ServiceActivationPolicies(opts.Namespace).List(context.Background(),
options)
+ }
+ w = func(options metav1.ListOptions) (watch.Interface, error) {
+ return
c.Dubbo().NetworkingV1alpha3().ServiceActivationPolicies(opts.Namespace).Watch(context.Background(),
options)
+ }
case gvr.ServiceEntry:
l = func(options metav1.ListOptions) (runtime.Object, error) {
return
c.Dubbo().NetworkingV1alpha3().ServiceEntries(opts.Namespace).List(context.Background(),
options)
diff --git a/pkg/config/schema/kubetypes/resources.gen.go
b/pkg/config/schema/kubetypes/resources.gen.go
index 544ef07a..00efb091 100755
--- a/pkg/config/schema/kubetypes/resources.gen.go
+++ b/pkg/config/schema/kubetypes/resources.gen.go
@@ -90,6 +90,10 @@ func getGvk(obj any) (config.GroupVersionKind, bool) {
return gvk.Service, true
case *k8sioapicorev1.ServiceAccount:
return gvk.ServiceAccount, true
+ case *githubcomkdubboapinetworkingv1alpha3.ServiceActivationPolicy:
+ return gvk.ServiceActivationPolicy, true
+ case
*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceActivationPolicy:
+ return gvk.ServiceActivationPolicy, true
case *githubcomkdubboapinetworkingv1alpha3.ServiceEntry:
return gvk.ServiceEntry, true
case
*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceEntry:
diff --git a/pkg/config/schema/metadata.yaml b/pkg/config/schema/metadata.yaml
index a81a3356..7a03f977 100644
--- a/pkg/config/schema/metadata.yaml
+++ b/pkg/config/schema/metadata.yaml
@@ -240,6 +240,15 @@ resources:
statusProto: "dubbo.meta.v1alpha1.DubboStatus"
statusProtoPackage: "github.com/kdubbo/api/meta/v1alpha1"
+ - kind: ServiceActivationPolicy
+ plural: "serviceactivationpolicies"
+ group: "networking.dubbo.apache.org"
+ version: "v1alpha3"
+ proto: "dubbo.networking.v1alpha3.ServiceActivationPolicy"
+ protoPackage: "github.com/kdubbo/api/networking/v1alpha3"
+ statusProto: "dubbo.meta.v1alpha1.DubboStatus"
+ statusProtoPackage: "github.com/kdubbo/api/meta/v1alpha1"
+
- kind: ServiceEntry
plural: "serviceentries"
group: "networking.dubbo.apache.org"
