This is an automated email from the ASF dual-hosted git repository.
mfordjody pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/dubbo-kubernetes.git
The following commit(s) were added to refs/heads/master by this push:
new 0b60a259 feat(discovery): add ServiceEntry registry (#953)
0b60a259 is described below
commit 0b60a25939faa007f723f2bbce553dc25153842a
Author: mfordjody <[email protected]>
AuthorDate: Mon Jul 13 19:30:03 2026 +0800
feat(discovery): add ServiceEntry registry (#953)
* Update ci infra v2
* Validate unlabeled telemetry resources
* feat(discovery): add ServiceEntry registry
---
.../discovery/pkg/bootstrap/service_controller.go | 6 +
.../pkg/config/kube/crdclient/types.gen.go | 102 +++++++
.../pkg/serviceregistry/aggregate/controller.go | 34 ++-
.../serviceregistry/aggregate/controller_test.go | 69 +++++
.../pkg/serviceregistry/provider/provider.go | 2 +
.../pkg/serviceregistry/serviceentry/controller.go | 311 +++++++++++++++++++
.../serviceentry/controller_test.go | 197 ++++++++++++
go.mod | 4 +-
go.sum | 8 +-
manifests/charts/base/files/crd-all.gen.yaml | 335 +++++++++++++++++++++
pkg/config/protocol/instance.go | 12 +
pkg/config/schema/collections/collections.gen.go | 38 +++
pkg/config/schema/gvk/resources.gen.go | 14 +
pkg/config/schema/gvr/resources.gen.go | 6 +
pkg/config/schema/kind/resources.gen.go | 10 +
pkg/config/schema/kubeclient/resources.gen.go | 26 ++
pkg/config/schema/kubetypes/resources.gen.go | 8 +
pkg/config/schema/metadata.yaml | 18 ++
pkg/config/validation/validators.go | 124 ++++++++
pkg/config/validation/validators_test.go | 62 ++++
20 files changed, 1377 insertions(+), 9 deletions(-)
diff --git a/dubbod/discovery/pkg/bootstrap/service_controller.go
b/dubbod/discovery/pkg/bootstrap/service_controller.go
index 6a555e8f..caf2915e 100644
--- a/dubbod/discovery/pkg/bootstrap/service_controller.go
+++ b/dubbod/discovery/pkg/bootstrap/service_controller.go
@@ -24,6 +24,7 @@ import (
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/serviceregistry/aggregate"
kubecontroller
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/serviceregistry/kube/controller"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/serviceregistry/provider"
+
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/serviceregistry/serviceentry"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
)
@@ -52,6 +53,11 @@ func (s *Server) initServiceControllers(args *DubboArgs)
error {
return fmt.Errorf("service registry %s is not
supported", r)
}
}
+
serviceControllers.AddRegistryAndRun(serviceentry.NewController(serviceentry.Options{
+ ConfigController: s.configController,
+ XDSUpdater: s.XDSServer,
+ ClusterID: s.clusterID,
+ }), s.internalStop)
s.addStartFunc("service controllers", func(stop <-chan struct{}) error {
go serviceControllers.Run(stop)
diff --git a/dubbod/discovery/pkg/config/kube/crdclient/types.gen.go
b/dubbod/discovery/pkg/config/kube/crdclient/types.gen.go
index ef1064fc..818efced 100755
--- a/dubbod/discovery/pkg/config/kube/crdclient/types.gen.go
+++ b/dubbod/discovery/pkg/config/kube/crdclient/types.gen.go
@@ -80,11 +80,21 @@ func create(c kube.Client, cfg config.Config, objMeta
metav1.ObjectMeta) (metav1
ObjectMeta: objMeta,
Spec:
*(cfg.Spec.(*githubcomkdubboapisecurityv1alpha3.RequestAuthentication)),
}, metav1.CreateOptions{})
+ case gvk.ServiceEntry:
+ return
c.Dubbo().NetworkingV1alpha3().ServiceEntries(cfg.Namespace).Create(context.TODO(),
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceEntry{
+ ObjectMeta: objMeta,
+ Spec:
*(cfg.Spec.(*githubcomkdubboapinetworkingv1alpha3.ServiceEntry)),
+ }, metav1.CreateOptions{})
case gvk.Telemetry:
return
c.Dubbo().TelemetryV1alpha1().Telemetries(cfg.Namespace).Create(context.TODO(),
&apigithubcomapachedubbokubernetesapitelemetryv1alpha1.Telemetry{
ObjectMeta: objMeta,
Spec:
*(cfg.Spec.(*githubcomkdubboapitelemetryv1alpha1.Telemetry)),
}, metav1.CreateOptions{})
+ case gvk.WorkloadEntry:
+ return
c.Dubbo().NetworkingV1alpha3().WorkloadEntries(cfg.Namespace).Create(context.TODO(),
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.WorkloadEntry{
+ ObjectMeta: objMeta,
+ Spec:
*(cfg.Spec.(*githubcomkdubboapinetworkingv1alpha3.WorkloadEntry)),
+ }, metav1.CreateOptions{})
default:
return nil, fmt.Errorf("unsupported type: %v",
cfg.GroupVersionKind)
}
@@ -137,11 +147,21 @@ func update(c kube.Client, cfg config.Config, objMeta
metav1.ObjectMeta) (metav1
ObjectMeta: objMeta,
Spec:
*(cfg.Spec.(*githubcomkdubboapisecurityv1alpha3.RequestAuthentication)),
}, metav1.UpdateOptions{})
+ case gvk.ServiceEntry:
+ return
c.Dubbo().NetworkingV1alpha3().ServiceEntries(cfg.Namespace).Update(context.TODO(),
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceEntry{
+ ObjectMeta: objMeta,
+ Spec:
*(cfg.Spec.(*githubcomkdubboapinetworkingv1alpha3.ServiceEntry)),
+ }, metav1.UpdateOptions{})
case gvk.Telemetry:
return
c.Dubbo().TelemetryV1alpha1().Telemetries(cfg.Namespace).Update(context.TODO(),
&apigithubcomapachedubbokubernetesapitelemetryv1alpha1.Telemetry{
ObjectMeta: objMeta,
Spec:
*(cfg.Spec.(*githubcomkdubboapitelemetryv1alpha1.Telemetry)),
}, metav1.UpdateOptions{})
+ case gvk.WorkloadEntry:
+ return
c.Dubbo().NetworkingV1alpha3().WorkloadEntries(cfg.Namespace).Update(context.TODO(),
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.WorkloadEntry{
+ ObjectMeta: objMeta,
+ Spec:
*(cfg.Spec.(*githubcomkdubboapinetworkingv1alpha3.WorkloadEntry)),
+ }, metav1.UpdateOptions{})
default:
return nil, fmt.Errorf("unsupported type: %v",
cfg.GroupVersionKind)
}
@@ -189,11 +209,21 @@ func updateStatus(c kube.Client, cfg config.Config,
objMeta metav1.ObjectMeta) (
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,
+ Status:
*(cfg.Status.(*githubcomkdubboapimetav1alpha1.DubboStatus)),
+ }, metav1.UpdateOptions{})
case gvk.Telemetry:
return
c.Dubbo().TelemetryV1alpha1().Telemetries(cfg.Namespace).UpdateStatus(context.TODO(),
&apigithubcomapachedubbokubernetesapitelemetryv1alpha1.Telemetry{
ObjectMeta: objMeta,
Status:
*(cfg.Status.(*githubcomkdubboapimetav1alpha1.DubboStatus)),
}, metav1.UpdateOptions{})
+ case gvk.WorkloadEntry:
+ return
c.Dubbo().NetworkingV1alpha3().WorkloadEntries(cfg.Namespace).UpdateStatus(context.TODO(),
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.WorkloadEntry{
+ ObjectMeta: objMeta,
+ Status:
*(cfg.Status.(*githubcomkdubboapimetav1alpha1.DubboStatus)),
+ }, metav1.UpdateOptions{})
default:
return nil, fmt.Errorf("unsupported type: %v",
cfg.GroupVersionKind)
}
@@ -339,6 +369,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.ServiceEntry:
+ oldRes :=
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceEntry{
+ ObjectMeta: origMeta,
+ Spec:
*(orig.Spec.(*githubcomkdubboapinetworkingv1alpha3.ServiceEntry)),
+ }
+ modRes :=
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceEntry{
+ ObjectMeta: modMeta,
+ Spec:
*(mod.Spec.(*githubcomkdubboapinetworkingv1alpha3.ServiceEntry)),
+ }
+ patchBytes, err := genPatchBytes(oldRes, modRes, typ)
+ if err != nil {
+ return nil, err
+ }
+ return
c.Dubbo().NetworkingV1alpha3().ServiceEntries(orig.Namespace).
+ Patch(context.TODO(), orig.Name, typ, patchBytes,
metav1.PatchOptions{FieldManager: "pilot-discovery"})
case gvk.Telemetry:
oldRes :=
&apigithubcomapachedubbokubernetesapitelemetryv1alpha1.Telemetry{
ObjectMeta: origMeta,
@@ -354,6 +399,21 @@ func patch(c kube.Client, orig config.Config, origMeta
metav1.ObjectMeta, mod co
}
return
c.Dubbo().TelemetryV1alpha1().Telemetries(orig.Namespace).
Patch(context.TODO(), orig.Name, typ, patchBytes,
metav1.PatchOptions{FieldManager: "pilot-discovery"})
+ case gvk.WorkloadEntry:
+ oldRes :=
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.WorkloadEntry{
+ ObjectMeta: origMeta,
+ Spec:
*(orig.Spec.(*githubcomkdubboapinetworkingv1alpha3.WorkloadEntry)),
+ }
+ modRes :=
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.WorkloadEntry{
+ ObjectMeta: modMeta,
+ Spec:
*(mod.Spec.(*githubcomkdubboapinetworkingv1alpha3.WorkloadEntry)),
+ }
+ patchBytes, err := genPatchBytes(oldRes, modRes, typ)
+ if err != nil {
+ return nil, err
+ }
+ return
c.Dubbo().NetworkingV1alpha3().WorkloadEntries(orig.Namespace).
+ Patch(context.TODO(), orig.Name, typ, patchBytes,
metav1.PatchOptions{FieldManager: "pilot-discovery"})
default:
return nil, fmt.Errorf("unsupported type: %v",
orig.GroupVersionKind)
}
@@ -383,8 +443,12 @@ 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.ServiceEntry:
+ return
c.Dubbo().NetworkingV1alpha3().ServiceEntries(namespace).Delete(context.TODO(),
name, deleteOptions)
case gvk.Telemetry:
return
c.Dubbo().TelemetryV1alpha1().Telemetries(namespace).Delete(context.TODO(),
name, deleteOptions)
+ case gvk.WorkloadEntry:
+ return
c.Dubbo().NetworkingV1alpha3().WorkloadEntries(namespace).Delete(context.TODO(),
name, deleteOptions)
default:
return fmt.Errorf("unsupported type: %v", typ)
}
@@ -852,6 +916,25 @@ var translationMap = map[config.GroupVersionKind]func(r
runtime.Object) config.C
Spec: obj,
}
},
+ gvk.ServiceEntry: func(r runtime.Object) config.Config {
+ obj :=
r.(*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceEntry)
+ return config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.ServiceEntry,
+ 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.StatefulSet: func(r runtime.Object) config.Config {
obj := r.(*k8sioapiappsv1.StatefulSet)
return config.Config{
@@ -907,4 +990,23 @@ var translationMap = map[config.GroupVersionKind]func(r
runtime.Object) config.C
Spec: obj,
}
},
+ gvk.WorkloadEntry: func(r runtime.Object) config.Config {
+ obj :=
r.(*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.WorkloadEntry)
+ return config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.WorkloadEntry,
+ 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,
+ }
+ },
}
diff --git a/dubbod/discovery/pkg/serviceregistry/aggregate/controller.go
b/dubbod/discovery/pkg/serviceregistry/aggregate/controller.go
index d4fb226a..1a1765b0 100644
--- a/dubbod/discovery/pkg/serviceregistry/aggregate/controller.go
+++ b/dubbod/discovery/pkg/serviceregistry/aggregate/controller.go
@@ -100,8 +100,16 @@ func (c *Controller) Services() []*model.Service {
for _, r := range c.GetRegistries() {
svcs := r.Services()
if r.Provider() != provider.Kubernetes {
- index += len(svcs)
- services = append(services, svcs...)
+ for _, s := range svcs {
+ if previous, found := smap[s.Hostname]; found
&& r.Provider() == provider.External {
+ services[previous] =
services[previous].DeepCopy()
+ decorateService(services[previous], s)
+ continue
+ }
+ smap[s.Hostname] = index
+ index++
+ services = append(services, s)
+ }
} else {
for _, s := range svcs {
previous, ok := smap[s.Hostname]
@@ -138,7 +146,13 @@ func (c *Controller) GetService(hostname host.Name)
*model.Service {
continue
}
if r.Provider() != provider.Kubernetes {
- return service
+ if out == nil {
+ return service
+ }
+ if r.Provider() == provider.External {
+ decorateService(out, service)
+ }
+ return out
}
if out == nil {
out = service.DeepCopy()
@@ -150,6 +164,20 @@ func (c *Controller) GetService(hostname host.Name)
*model.Service {
return out
}
+func decorateService(dst, src *model.Service) {
+ accounts := make(map[string]struct{},
len(dst.ServiceAccounts)+len(src.ServiceAccounts))
+ for _, account := range dst.ServiceAccounts {
+ accounts[account] = struct{}{}
+ }
+ for _, account := range src.ServiceAccounts {
+ if _, found := accounts[account]; found {
+ continue
+ }
+ dst.ServiceAccounts = append(dst.ServiceAccounts, account)
+ accounts[account] = struct{}{}
+ }
+}
+
func (c *Controller) GetRegistries() []serviceregistry.Instance {
c.storeLock.RLock()
defer c.storeLock.RUnlock()
diff --git a/dubbod/discovery/pkg/serviceregistry/aggregate/controller_test.go
b/dubbod/discovery/pkg/serviceregistry/aggregate/controller_test.go
new file mode 100644
index 00000000..4ed65b7e
--- /dev/null
+++ b/dubbod/discovery/pkg/serviceregistry/aggregate/controller_test.go
@@ -0,0 +1,69 @@
+//
+// 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 aggregate
+
+import (
+ "testing"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/serviceregistry/provider"
+ "github.com/apache/dubbo-kubernetes/pkg/cluster"
+ "github.com/apache/dubbo-kubernetes/pkg/config/host"
+)
+
+func TestServiceEntryDecoratesKubernetesService(t *testing.T) {
+ hostname := host.Name("reviews.bookinfo.svc.cluster.local")
+ kubeService := &model.Service{Hostname: hostname, ServiceAccounts:
[]string{"spiffe://cluster.local/ns/bookinfo/sa/default"}}
+ externalService := &model.Service{Hostname: hostname, ServiceAccounts:
[]string{
+ "spiffe://cluster.local/ns/bookinfo/sa/default",
+ "spiffe://cluster.local/ns/bookinfo/sa/reviews",
+ }}
+ controller := NewController(Options{})
+ controller.addRegistry(&staticRegistry{provider: provider.Kubernetes,
services: []*model.Service{kubeService}}, nil)
+ controller.addRegistry(&staticRegistry{provider: provider.External,
services: []*model.Service{externalService}}, nil)
+
+ services := controller.Services()
+ if len(services) != 1 {
+ t.Fatalf("got %d services, want one merged service",
len(services))
+ }
+ if got := len(services[0].ServiceAccounts); got != 2 {
+ t.Fatalf("got %d service accounts, want 2", got)
+ }
+ if got := controller.GetService(hostname); got == nil ||
len(got.ServiceAccounts) != 2 {
+ t.Fatalf("GetService() = %#v, want decorated Kubernetes
service", got)
+ }
+}
+
+type staticRegistry struct {
+ provider provider.ID
+ services []*model.Service
+}
+
+func (*staticRegistry) Run(stop <-chan struct{})
{ <-stop }
+func (*staticRegistry) HasSynced() bool
{ return true }
+func (s *staticRegistry) Services() []*model.Service
{ return s.services }
+func (s *staticRegistry) Provider() provider.ID
{ return s.provider }
+func (*staticRegistry) Cluster() cluster.ID
{ return "cluster-1" }
+func (*staticRegistry) GetProxyServiceTargets(*model.Proxy)
[]model.ServiceTarget { return nil }
+func (s *staticRegistry) GetService(name host.Name) *model.Service {
+ for _, service := range s.services {
+ if service.Hostname == name {
+ return service
+ }
+ }
+ return nil
+}
diff --git a/dubbod/discovery/pkg/serviceregistry/provider/provider.go
b/dubbod/discovery/pkg/serviceregistry/provider/provider.go
index f484271a..10be63d0 100644
--- a/dubbod/discovery/pkg/serviceregistry/provider/provider.go
+++ b/dubbod/discovery/pkg/serviceregistry/provider/provider.go
@@ -22,6 +22,8 @@ type ID string
const (
// Kubernetes is a service registry backed by k8s API server
Kubernetes ID = "Kubernetes"
+ // External is a service registry backed by declarative ServiceEntry
resources.
+ External ID = "External"
)
func (id ID) String() string {
diff --git a/dubbod/discovery/pkg/serviceregistry/serviceentry/controller.go
b/dubbod/discovery/pkg/serviceregistry/serviceentry/controller.go
new file mode 100644
index 00000000..cd454339
--- /dev/null
+++ b/dubbod/discovery/pkg/serviceregistry/serviceentry/controller.go
@@ -0,0 +1,311 @@
+//
+// 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 serviceentry
+
+import (
+ "sort"
+ "strconv"
+ "sync"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/serviceregistry/provider"
+ "github.com/apache/dubbo-kubernetes/pkg/cluster"
+ "github.com/apache/dubbo-kubernetes/pkg/config"
+ "github.com/apache/dubbo-kubernetes/pkg/config/constants"
+ "github.com/apache/dubbo-kubernetes/pkg/config/host"
+ "github.com/apache/dubbo-kubernetes/pkg/config/labels"
+ "github.com/apache/dubbo-kubernetes/pkg/config/protocol"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/gvk"
+ "github.com/apache/dubbo-kubernetes/pkg/config/visibility"
+ "github.com/apache/dubbo-kubernetes/pkg/slices"
+ "github.com/apache/dubbo-kubernetes/pkg/util/sets"
+ networking "github.com/kdubbo/api/networking/v1alpha3"
+)
+
+type Options struct {
+ ConfigController model.ConfigStoreController
+ XDSUpdater model.XDSUpdater
+ ClusterID cluster.ID
+}
+
+type Controller struct {
+ mu sync.RWMutex
+ reconcileMu sync.Mutex
+ configController model.ConfigStoreController
+ xdsUpdater model.XDSUpdater
+ clusterID cluster.ID
+ services map[host.Name]*model.Service
+ endpoints map[string][]*model.DubboEndpoint
+}
+
+func NewController(opts Options) *Controller {
+ c := &Controller{
+ configController: opts.ConfigController,
+ xdsUpdater: opts.XDSUpdater,
+ clusterID: opts.ClusterID,
+ services: make(map[host.Name]*model.Service),
+ endpoints: make(map[string][]*model.DubboEndpoint),
+ }
+ opts.ConfigController.RegisterEventHandler(gvk.ServiceEntry,
c.onConfigEvent)
+ opts.ConfigController.RegisterEventHandler(gvk.WorkloadEntry,
c.onConfigEvent)
+ return c
+}
+
+func (c *Controller) onConfigEvent(_, _ config.Config, _ model.Event) {
+ c.reconcile()
+}
+
+func (c *Controller) reconcile() {
+ c.reconcileMu.Lock()
+ defer c.reconcileMu.Unlock()
+
+ entries := c.configController.List(gvk.ServiceEntry, model.NamespaceAll)
+ workloads := c.configController.List(gvk.WorkloadEntry,
model.NamespaceAll)
+ sort.Slice(entries, func(i, j int) bool {
+ if
entries[i].CreationTimestamp.Equal(entries[j].CreationTimestamp) {
+ if entries[i].Namespace == entries[j].Namespace {
+ return entries[i].Name < entries[j].Name
+ }
+ return entries[i].Namespace < entries[j].Namespace
+ }
+ return
entries[i].CreationTimestamp.Before(entries[j].CreationTimestamp)
+ })
+
+ services := make(map[host.Name]*model.Service)
+ endpoints := make(map[string][]*model.DubboEndpoint)
+ for _, cfg := range entries {
+ entry, ok := cfg.Spec.(*networking.ServiceEntry)
+ if !ok {
+ continue
+ }
+ selected := selectWorkloads(cfg.Namespace, entry, workloads)
+ for _, hostname := range entry.GetHosts() {
+ h := host.Name(hostname)
+ if _, found := services[h]; found {
+ continue
+ }
+ svc := c.convertService(cfg, entry, h)
+ services[h] = svc
+ endpoints[svc.Key()] = buildEndpoints(cfg, entry, svc,
selected)
+ }
+ }
+
+ c.mu.Lock()
+ previous := c.services
+ previousEndpoints := c.endpoints
+ c.services = services
+ c.endpoints = endpoints
+ c.mu.Unlock()
+
+ if c.xdsUpdater == nil {
+ return
+ }
+ shard := model.ShardKeyFromRegistry(c)
+ for hostname, svc := range services {
+ oldService := previous[hostname]
+ if oldService == nil {
+ c.xdsUpdater.ServiceUpdate(shard, string(hostname),
svc.Attributes.Namespace, model.EventAdd)
+ c.xdsUpdater.EDSUpdate(shard, string(hostname),
svc.Attributes.Namespace, endpoints[svc.Key()])
+ continue
+ }
+ if oldService.Attributes.Namespace != svc.Attributes.Namespace {
+ c.xdsUpdater.ServiceUpdate(shard, string(hostname),
oldService.Attributes.Namespace, model.EventDelete)
+ c.xdsUpdater.EDSUpdate(shard, string(hostname),
oldService.Attributes.Namespace, nil)
+ c.xdsUpdater.ServiceUpdate(shard, string(hostname),
svc.Attributes.Namespace, model.EventAdd)
+ c.xdsUpdater.EDSUpdate(shard, string(hostname),
svc.Attributes.Namespace, endpoints[svc.Key()])
+ continue
+ }
+ if !oldService.Equals(svc) {
+ c.xdsUpdater.ServiceUpdate(shard, string(hostname),
svc.Attributes.Namespace, model.EventUpdate)
+ }
+ if !slices.EqualFunc(previousEndpoints[oldService.Key()],
endpoints[svc.Key()], func(a, b *model.DubboEndpoint) bool {
+ return a.Equals(b)
+ }) {
+ c.xdsUpdater.EDSUpdate(shard, string(hostname),
svc.Attributes.Namespace, endpoints[svc.Key()])
+ }
+ }
+ for hostname, svc := range previous {
+ if _, found := services[hostname]; found {
+ continue
+ }
+ c.xdsUpdater.ServiceUpdate(shard, string(hostname),
svc.Attributes.Namespace, model.EventDelete)
+ c.xdsUpdater.EDSUpdate(shard, string(hostname),
svc.Attributes.Namespace, nil)
+ }
+}
+
+func (c *Controller) convertService(cfg config.Config, entry
*networking.ServiceEntry, hostname host.Name) *model.Service {
+ addresses := append([]string(nil), entry.GetAddresses()...)
+ defaultAddress := constants.UnspecifiedIP
+ if len(addresses) > 0 {
+ defaultAddress = addresses[0]
+ } else {
+ addresses = []string{defaultAddress}
+ }
+ ports := make(model.PortList, 0, len(entry.GetPorts()))
+ for _, port := range entry.GetPorts() {
+ ports = append(ports, &model.Port{Name: port.GetName(), Port:
int(port.GetNumber()), Protocol: protocol.Parse(port.GetProtocol())})
+ }
+ exportTo := sets.New[visibility.Instance]()
+ for _, namespace := range entry.GetExportTo() {
+ exportTo.Insert(visibility.Instance(namespace))
+ }
+ service := &model.Service{
+ Hostname: hostname,
+ Ports: ports,
+ ServiceAccounts: append([]string(nil),
entry.GetSubjectAltNames()...),
+ ClusterVIPs: model.AddressMap{Addresses:
map[cluster.ID][]string{
+ c.clusterID: addresses,
+ }},
+ CreationTime: cfg.CreationTimestamp,
+ DefaultAddress: defaultAddress,
+ ResourceVersion: cfg.ResourceVersion,
+ Resolution: convertResolution(entry.GetResolution()),
+ MeshExternal: entry.GetLocation() ==
networking.ServiceEntry_MESH_EXTERNAL,
+ Attributes: model.ServiceAttributes{
+ Name: cfg.Name,
+ Namespace: cfg.Namespace,
+ ExportTo: exportTo,
+ ServiceRegistry: provider.External,
+ },
+ }
+ service.Attributes.ObjectName = cfg.Name
+ return service
+}
+
+func convertResolution(resolution networking.ServiceEntry_Resolution)
model.Resolution {
+ switch resolution {
+ case networking.ServiceEntry_STATIC:
+ return model.ClientSideLB
+ case networking.ServiceEntry_DNS:
+ return model.DNSLB
+ case networking.ServiceEntry_DNS_ROUND_ROBIN:
+ return model.DNSRoundRobinLB
+ default:
+ return model.Passthrough
+ }
+}
+
+func selectWorkloads(namespace string, entry *networking.ServiceEntry, configs
[]config.Config) []namedWorkload {
+ selector := entry.GetWorkloadSelector()
+ if selector == nil {
+ out := make([]namedWorkload, 0, len(entry.GetEndpoints()))
+ for i, workload := range entry.GetEndpoints() {
+ out = append(out, namedWorkload{name: entryName(i),
workload: workload})
+ }
+ return out
+ }
+ wanted := labels.Instance(selector.GetMatchLabels())
+ out := make([]namedWorkload, 0)
+ for _, cfg := range configs {
+ if cfg.Namespace != namespace {
+ continue
+ }
+ workload, ok := cfg.Spec.(*networking.WorkloadEntry)
+ if ok && wanted.Match(labels.Instance(workload.GetLabels())) {
+ out = append(out, namedWorkload{name: cfg.Name,
workload: workload})
+ }
+ }
+ sort.Slice(out, func(i, j int) bool { return out[i].name < out[j].name
})
+ return out
+}
+
+type namedWorkload struct {
+ name string
+ workload *networking.WorkloadEntry
+}
+
+func entryName(index int) string {
+ return "inline-" + strconv.Itoa(index)
+}
+
+func buildEndpoints(cfg config.Config, entry *networking.ServiceEntry, svc
*model.Service, workloads []namedWorkload) []*model.DubboEndpoint {
+ if len(workloads) == 0 && (entry.GetResolution() ==
networking.ServiceEntry_DNS || entry.GetResolution() ==
networking.ServiceEntry_DNS_ROUND_ROBIN) {
+ workloads = []namedWorkload{{name: cfg.Name, workload:
&networking.WorkloadEntry{Address: string(svc.Hostname)}}}
+ }
+ result := make([]*model.DubboEndpoint, 0,
len(workloads)*len(entry.GetPorts()))
+ for _, named := range workloads {
+ for _, port := range entry.GetPorts() {
+ targetPort := named.workload.GetPorts()[port.GetName()]
+ if targetPort == 0 {
+ targetPort = port.GetTargetPort()
+ }
+ if targetPort == 0 {
+ targetPort = port.GetNumber()
+ }
+ result = append(result, &model.DubboEndpoint{
+ ServiceAccount:
named.workload.GetServiceAccount(),
+ Addresses:
[]string{named.workload.GetAddress()},
+ ServicePortName: port.GetName(),
+ Labels:
labels.Instance(named.workload.GetLabels()),
+ HealthStatus: model.Healthy,
+ EndpointPort: targetPort,
+ WorkloadName: named.name,
+ Namespace: cfg.Namespace,
+ })
+ }
+ }
+ return result
+}
+
+func (c *Controller) Services() []*model.Service {
+ c.mu.RLock()
+ defer c.mu.RUnlock()
+ out := make([]*model.Service, 0, len(c.services))
+ for _, svc := range c.services {
+ out = append(out, svc)
+ }
+ sort.Slice(out, func(i, j int) bool { return out[i].Hostname <
out[j].Hostname })
+ return out
+}
+
+func (c *Controller) GetService(hostname host.Name) *model.Service {
+ c.mu.RLock()
+ defer c.mu.RUnlock()
+ return c.services[hostname]
+}
+
+func (c *Controller) GetProxyServiceTargets(proxy *model.Proxy)
[]model.ServiceTarget {
+ if proxy == nil {
+ return nil
+ }
+ addresses := sets.New(proxy.IPAddresses...)
+ c.mu.RLock()
+ defer c.mu.RUnlock()
+ var out []model.ServiceTarget
+ for _, svc := range c.services {
+ for _, endpoint := range c.endpoints[svc.Key()] {
+ if !addresses.Contains(endpoint.FirstAddressOrNil()) {
+ continue
+ }
+ for _, port := range svc.Ports {
+ if port.Name == endpoint.ServicePortName {
+ out = append(out,
model.ServiceTarget{Service: svc, Port: model.ServiceInstancePort{ServicePort:
port, TargetPort: endpoint.EndpointPort}})
+ }
+ }
+ }
+ }
+ return out
+}
+
+func (c *Controller) Provider() provider.ID { return provider.External }
+func (c *Controller) Cluster() cluster.ID { return c.clusterID }
+func (c *Controller) HasSynced() bool { return
c.configController.HasSynced() }
+
+func (c *Controller) Run(stop <-chan struct{}) {
+ c.reconcile()
+ <-stop
+}
diff --git
a/dubbod/discovery/pkg/serviceregistry/serviceentry/controller_test.go
b/dubbod/discovery/pkg/serviceregistry/serviceentry/controller_test.go
new file mode 100644
index 00000000..7c822608
--- /dev/null
+++ b/dubbod/discovery/pkg/serviceregistry/serviceentry/controller_test.go
@@ -0,0 +1,197 @@
+//
+// 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 serviceentry
+
+import (
+ "testing"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/config/memory"
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+ "github.com/apache/dubbo-kubernetes/pkg/cluster"
+ "github.com/apache/dubbo-kubernetes/pkg/config"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/collections"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/gvk"
+ networking "github.com/kdubbo/api/networking/v1alpha3"
+ typev1alpha3 "github.com/kdubbo/api/type/v1alpha3"
+)
+
+func TestControllerReconcilesServiceEntryAndWorkloadEntry(t *testing.T) {
+ configs := newFakeConfigController()
+ updater := &fakeXDSUpdater{}
+ controller := NewController(Options{
+ ConfigController: configs,
+ XDSUpdater: updater,
+ ClusterID: cluster.ID("cluster-1"),
+ })
+
+ workload := config.Config{Meta: config.Meta{
+ GroupVersionKind: gvk.WorkloadEntry,
+ Name: "reviews-vm",
+ Namespace: "bookinfo",
+ }, Spec: &networking.WorkloadEntry{
+ Address: "10.0.0.10",
+ Ports: map[string]uint32{"grpc": 16000},
+ Labels: map[string]string{"app": "reviews"},
+ ServiceAccount: "reviews",
+ }}
+ configs.create(t, workload)
+ entry := config.Config{Meta: config.Meta{
+ GroupVersionKind: gvk.ServiceEntry,
+ Name: "reviews-external",
+ Namespace: "bookinfo",
+ }, Spec: &networking.ServiceEntry{
+ Hosts: []string{"reviews.example.com"},
+ Addresses: []string{"240.0.0.1"},
+ Location: networking.ServiceEntry_MESH_INTERNAL,
+ Resolution: networking.ServiceEntry_STATIC,
+ Ports: []*networking.ServicePort{{
+ Name: "grpc", Number: 50051, Protocol: "GRPC",
TargetPort: 15000,
+ }},
+ WorkloadSelector: &typev1alpha3.WorkloadSelector{MatchLabels:
map[string]string{"app": "reviews"}},
+ }}
+ configs.create(t, entry)
+
+ service := controller.GetService("reviews.example.com")
+ if service == nil {
+ t.Fatal("service was not created")
+ }
+ if service.DefaultAddress != "240.0.0.1" || service.Resolution !=
model.ClientSideLB || service.MeshExternal {
+ t.Fatalf("unexpected service: address=%s resolution=%v
external=%v", service.DefaultAddress, service.Resolution, service.MeshExternal)
+ }
+ last := updater.lastEDS(t)
+ if len(last.endpoints) != 1 {
+ t.Fatalf("got %d endpoints, want 1", len(last.endpoints))
+ }
+ endpoint := last.endpoints[0]
+ if endpoint.FirstAddressOrNil() != "10.0.0.10" || endpoint.EndpointPort
!= 16000 || endpoint.ServiceAccount != "reviews" {
+ t.Fatalf("unexpected endpoint: address=%s port=%d
serviceAccount=%s", endpoint.FirstAddressOrNil(), endpoint.EndpointPort,
endpoint.ServiceAccount)
+ }
+
+ workload.Spec.(*networking.WorkloadEntry).Address = "10.0.0.11"
+ configs.update(t, workload)
+ if got := updater.lastEDS(t).endpoints[0].FirstAddressOrNil(); got !=
"10.0.0.11" {
+ t.Fatalf("updated endpoint address = %s, want 10.0.0.11", got)
+ }
+ if got := len(updater.serviceEvents); got != 1 {
+ t.Fatalf("endpoint-only update emitted %d service events, want
1 total", got)
+ }
+ edsUpdates := len(updater.edsEvents)
+ configs.create(t, config.Config{Meta: config.Meta{
+ GroupVersionKind: gvk.WorkloadEntry,
+ Name: "unrelated",
+ Namespace: "bookinfo",
+ }, Spec: &networking.WorkloadEntry{Address: "10.0.0.20", Labels:
map[string]string{"app": "other"}}})
+ if got := len(updater.edsEvents); got != edsUpdates {
+ t.Fatalf("unrelated workload emitted an EDS update: got %d
updates, want %d", got, edsUpdates)
+ }
+
+ configs.delete(t, gvk.ServiceEntry, entry.Name, entry.Namespace)
+ if controller.GetService("reviews.example.com") != nil {
+ t.Fatal("service was not deleted")
+ }
+ if got := updater.serviceEvents[len(updater.serviceEvents)-1].event;
got != model.EventDelete {
+ t.Fatalf("last service event = %v, want delete", got)
+ }
+ if got := len(updater.lastEDS(t).endpoints); got != 0 {
+ t.Fatalf("deleted service has %d endpoints, want 0", got)
+ }
+}
+
+type fakeConfigController struct {
+ model.ConfigStore
+ handlers map[config.GroupVersionKind][]model.EventHandler
+}
+
+func newFakeConfigController() *fakeConfigController {
+ return &fakeConfigController{
+ ConfigStore: memory.Make(collections.Dubbo),
+ handlers:
make(map[config.GroupVersionKind][]model.EventHandler),
+ }
+}
+
+func (f *fakeConfigController) RegisterEventHandler(kind
config.GroupVersionKind, handler model.EventHandler) {
+ f.handlers[kind] = append(f.handlers[kind], handler)
+}
+
+func (f *fakeConfigController) Run(stop <-chan struct{}) { <-stop }
+func (f *fakeConfigController) HasSynced() bool { return true }
+
+func (f *fakeConfigController) create(t *testing.T, cfg config.Config) {
+ t.Helper()
+ if _, err := f.ConfigStore.Create(cfg); err != nil {
+ t.Fatal(err)
+ }
+ current := *f.Get(cfg.GroupVersionKind, cfg.Name, cfg.Namespace)
+ for _, handler := range f.handlers[cfg.GroupVersionKind] {
+ handler(config.Config{}, current, model.EventAdd)
+ }
+}
+
+func (f *fakeConfigController) update(t *testing.T, cfg config.Config) {
+ t.Helper()
+ old := *f.Get(cfg.GroupVersionKind, cfg.Name, cfg.Namespace)
+ cfg.ResourceVersion = ""
+ if _, err := f.ConfigStore.Update(cfg); err != nil {
+ t.Fatal(err)
+ }
+ current := *f.Get(cfg.GroupVersionKind, cfg.Name, cfg.Namespace)
+ for _, handler := range f.handlers[cfg.GroupVersionKind] {
+ handler(old, current, model.EventUpdate)
+ }
+}
+
+func (f *fakeConfigController) delete(t *testing.T, kind
config.GroupVersionKind, name, namespace string) {
+ t.Helper()
+ old := *f.Get(kind, name, namespace)
+ if err := f.ConfigStore.Delete(kind, name, namespace, nil); err != nil {
+ t.Fatal(err)
+ }
+ for _, handler := range f.handlers[kind] {
+ handler(old, config.Config{}, model.EventDelete)
+ }
+}
+
+type serviceEvent struct {
+ event model.Event
+}
+
+type edsEvent struct {
+ endpoints []*model.DubboEndpoint
+}
+
+type fakeXDSUpdater struct {
+ serviceEvents []serviceEvent
+ edsEvents []edsEvent
+}
+
+func (*fakeXDSUpdater) ConfigUpdate(*model.PushRequest) {}
+func (f *fakeXDSUpdater) ServiceUpdate(_ model.ShardKey, _, _ string, event
model.Event) {
+ f.serviceEvents = append(f.serviceEvents, serviceEvent{event: event})
+}
+func (f *fakeXDSUpdater) EDSUpdate(_ model.ShardKey, _, _ string, endpoints
[]*model.DubboEndpoint) {
+ f.edsEvents = append(f.edsEvents, edsEvent{endpoints: endpoints})
+}
+func (*fakeXDSUpdater) EDSCacheUpdate(model.ShardKey, string, string,
[]*model.DubboEndpoint) {}
+func (*fakeXDSUpdater) ProxyUpdate(cluster.ID, string)
{}
+
+func (f *fakeXDSUpdater) lastEDS(t *testing.T) edsEvent {
+ t.Helper()
+ if len(f.edsEvents) == 0 {
+ t.Fatal("no EDS update received")
+ }
+ return f.edsEvents[len(f.edsEvents)-1]
+}
diff --git a/go.mod b/go.mod
index a4b5d7c3..6922ab24 100644
--- a/go.mod
+++ b/go.mod
@@ -53,8 +53,8 @@ require (
github.com/hashicorp/go-multierror v1.1.1
github.com/hashicorp/golang-lru/v2 v2.0.5
github.com/heroku/color v0.0.6
- github.com/kdubbo/api v0.0.0-20260711063406-ff5334e781fd
- github.com/kdubbo/client-go v0.0.0-20260711063741-a336aba3d8e8
+ github.com/kdubbo/api v0.0.0-20260713105721-558a18fe4f33
+ github.com/kdubbo/client-go v0.0.0-20260713105923-9a87feab8d96
github.com/kdubbo/xds-api v0.0.0-20260628124158-b93008db9162
github.com/moby/term v0.5.2
github.com/ory/viper v1.7.5
diff --git a/go.sum b/go.sum
index 21fbb2ac..9c8e1324 100644
--- a/go.sum
+++ b/go.sum
@@ -337,10 +337,10 @@ github.com/json-iterator/go v1.1.12/go.mod
h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHm
github.com/jtolds/gls v4.20.0+incompatible/go.mod
h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
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-20260711063406-ff5334e781fd
h1:oMbQY61FOOITVLKRdGX0EirWW3nZN26F+cwttj9DmSM=
-github.com/kdubbo/api v0.0.0-20260711063406-ff5334e781fd/go.mod
h1:8BtJiIovg7QCPsCxXcw3gDf922VcvYq5ihOSvj49Rq8=
-github.com/kdubbo/client-go v0.0.0-20260711063741-a336aba3d8e8
h1:mNvM3yB78CVgozJm4wZxIKK//mBJESofhD2Z0eHLoC0=
-github.com/kdubbo/client-go v0.0.0-20260711063741-a336aba3d8e8/go.mod
h1:HDLiCm6am1zZTLGq+tNt7S/lZfpmU/zRdVJyOp0yymU=
+github.com/kdubbo/api v0.0.0-20260713105721-558a18fe4f33
h1://TxoBFL/igcCqokYHOcA5hA30J6LTpFusDYVQx48Nw=
+github.com/kdubbo/api v0.0.0-20260713105721-558a18fe4f33/go.mod
h1:8BtJiIovg7QCPsCxXcw3gDf922VcvYq5ihOSvj49Rq8=
+github.com/kdubbo/client-go v0.0.0-20260713105923-9a87feab8d96
h1:co2FdpetF4dIozVHMatAd2yYPtKBqdpCgSbpdnXEkKw=
+github.com/kdubbo/client-go v0.0.0-20260713105923-9a87feab8d96/go.mod
h1:NUl5DTlsovJaoYDj0SyHi263H+1Z11EXP7qnk6TPKjo=
github.com/kdubbo/xds-api v0.0.0-20260628124158-b93008db9162
h1:8T6FM/7KYiX84f8wMNmWxZ4YMv+uONCkZqTKiM6m/M8=
github.com/kdubbo/xds-api v0.0.0-20260628124158-b93008db9162/go.mod
h1:o2HDUgL1ntaDbWomZ4cD2tt8jBamuG2qRtjXOa1zZ0Q=
github.com/kevinburke/ssh_config v1.2.0
h1:x584FjTGwHzMwvHx18PXxbBVzfnxogHaAReU4gf13a4=
diff --git a/manifests/charts/base/files/crd-all.gen.yaml
b/manifests/charts/base/files/crd-all.gen.yaml
index fda86cd0..422dd049 100644
--- a/manifests/charts/base/files/crd-all.gen.yaml
+++ b/manifests/charts/base/files/crd-all.gen.yaml
@@ -152,6 +152,341 @@ spec:
---
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
+metadata:
+ annotations:
+ "helm.sh/resource-policy": keep
+ labels:
+ app: dubbo
+ chart: dubbo
+ dubbo: networking
+ heritage: Tiller
+ release: dubbo
+ name: serviceentries.networking.dubbo.apache.org
+spec:
+ group: networking.dubbo.apache.org
+ names:
+ categories:
+ - dubbo
+ - networking
+ kind: ServiceEntry
+ listKind: ServiceEntryList
+ plural: serviceentries
+ shortNames:
+ - se
+ singular: serviceentry
+ scope: Namespaced
+ versions:
+ - additionalPrinterColumns:
+ - description: Service hosts.
+ jsonPath: .spec.hosts
+ name: Hosts
+ type: string
+ - description: Service location.
+ jsonPath: .spec.location
+ name: Location
+ type: string
+ - description: Service resolution.
+ jsonPath: .spec.resolution
+ name: Resolution
+ type: string
+ - description: CreationTimestamp is a timestamp representing the server
time when
+ this object was created.
+ jsonPath: .metadata.creationTimestamp
+ name: Age
+ type: date
+ name: v1alpha3
+ schema:
+ openAPIV3Schema:
+ properties:
+ spec:
+ description: 'Declarative entries for the internal service
registry. See
+ more details at: '
+ properties:
+ addresses:
+ description: Virtual IP addresses associated with the service.
+ items:
+ type: string
+ maxItems: 256
+ type: array
+ endpoints:
+ description: Inline endpoints associated with the service.
+ items:
+ properties:
+ address:
+ description: IP address or fully-qualified DNS name of
the workload.
+ maxLength: 256
+ minLength: 1
+ type: string
+ labels:
+ additionalProperties:
+ type: string
+ description: Labels used by
ServiceEntry.workloadSelector.
+ maxProperties: 256
+ type: object
+ locality:
+ description: Locality in region/zone/subzone form.
+ maxLength: 2048
+ type: string
+ network:
+ description: Network containing the workload.
+ maxLength: 2048
+ type: string
+ ports:
+ additionalProperties:
+ maximum: 4294967295
+ minimum: 0
+ type: integer
+ x-kubernetes-validations:
+ - message: port must be between 1-65535
+ rule: 0 < self && self <= 65535
+ description: Per-service-port endpoint ports.
+ maxProperties: 128
+ type: object
+ serviceAccount:
+ description: Service account used to derive workload
identity.
+ maxLength: 253
+ type: string
+ weight:
+ description: Relative load-balancing weight.
+ maximum: 4294967295
+ minimum: 0
+ type: integer
+ type: object
+ maxItems: 4096
+ type: array
+ exportTo:
+ description: Namespaces to which the service is exported.
+ items:
+ type: string
+ type: array
+ hosts:
+ description: DNS hosts associated with the service.
+ items:
+ type: string
+ maxItems: 256
+ minItems: 1
+ type: array
+ location:
+ description: |2-
+
+
+ Valid Options: MESH_EXTERNAL, MESH_INTERNAL
+ enum:
+ - MESH_EXTERNAL
+ - MESH_INTERNAL
+ type: string
+ ports:
+ description: Ports exposed by the service.
+ items:
+ properties:
+ name:
+ description: Stable name used by WorkloadEntry.ports.
+ maxLength: 256
+ type: string
+ number:
+ maximum: 4294967295
+ minimum: 0
+ type: integer
+ x-kubernetes-validations:
+ - message: port must be between 1-65535
+ rule: 0 < self && self <= 65535
+ protocol:
+ description: Protocol exposed by the port, for example
GRPC,
+ HTTP2, TCP, or TLS.
+ maxLength: 256
+ type: string
+ targetPort:
+ description: Default endpoint port.
+ maximum: 4294967295
+ minimum: 0
+ type: integer
+ required:
+ - number
+ - name
+ type: object
+ maxItems: 256
+ type: array
+ x-kubernetes-list-map-keys:
+ - name
+ x-kubernetes-list-type: map
+ resolution:
+ description: |2-
+
+
+ Valid Options: NONE, STATIC, DNS, DNS_ROUND_ROBIN
+ enum:
+ - NONE
+ - STATIC
+ - DNS
+ - DNS_ROUND_ROBIN
+ type: string
+ subjectAltNames:
+ description: Subject alternative names accepted for the
service.
+ items:
+ type: string
+ type: array
+ workloadSelector:
+ description: Selects WorkloadEntry resources in the same
namespace.
+ properties:
+ matchLabels:
+ additionalProperties:
+ maxLength: 63
+ type: string
+ x-kubernetes-validations:
+ - message: wildcard not allowed in label value match
+ rule: '!self.contains("*")'
+ maxProperties: 4096
+ type: object
+ x-kubernetes-validations:
+ - message: wildcard not allowed in label key match
+ rule: self.all(key, !key.contains("*"))
+ - message: key must not be empty
+ rule: self.all(key, key.size() != 0)
+ type: object
+ required:
+ - hosts
+ type: object
+ x-kubernetes-validations:
+ - message: only one of workloadSelector or endpoints can be set
+ rule: '!(has(self.workloadSelector) && has(self.endpoints))'
+ status:
+ properties:
+ conditions:
+ items:
+ properties:
+ observedGeneration:
+ anyOf:
+ - type: integer
+ - type: string
+ x-kubernetes-int-or-string: true
+ reason:
+ type: string
+ status:
+ type: string
+ type:
+ type: string
+ type: object
+ type: array
+ type: object
+ x-kubernetes-preserve-unknown-fields: true
+ type: object
+ served: true
+ storage: true
+ subresources:
+ status: {}
+---
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+ annotations:
+ "helm.sh/resource-policy": keep
+ labels:
+ app: dubbo
+ chart: dubbo
+ dubbo: networking
+ heritage: Tiller
+ release: dubbo
+ name: workloadentries.networking.dubbo.apache.org
+spec:
+ group: networking.dubbo.apache.org
+ names:
+ categories:
+ - dubbo
+ - networking
+ kind: WorkloadEntry
+ listKind: WorkloadEntryList
+ plural: workloadentries
+ shortNames:
+ - we
+ singular: workloadentry
+ scope: Namespaced
+ versions:
+ - additionalPrinterColumns:
+ - description: Workload address.
+ jsonPath: .spec.address
+ name: Address
+ type: string
+ - description: CreationTimestamp is a timestamp representing the server
time when
+ this object was created.
+ jsonPath: .metadata.creationTimestamp
+ name: Age
+ type: date
+ name: v1alpha3
+ schema:
+ openAPIV3Schema:
+ properties:
+ spec:
+ description: 'Configuration describing a non-Kubernetes workload.
See
+ more details at: '
+ properties:
+ address:
+ description: IP address or fully-qualified DNS name of the
workload.
+ maxLength: 256
+ minLength: 1
+ type: string
+ labels:
+ additionalProperties:
+ type: string
+ description: Labels used by ServiceEntry.workloadSelector.
+ maxProperties: 256
+ type: object
+ locality:
+ description: Locality in region/zone/subzone form.
+ maxLength: 2048
+ type: string
+ network:
+ description: Network containing the workload.
+ maxLength: 2048
+ type: string
+ ports:
+ additionalProperties:
+ maximum: 4294967295
+ minimum: 0
+ type: integer
+ x-kubernetes-validations:
+ - message: port must be between 1-65535
+ rule: 0 < self && self <= 65535
+ description: Per-service-port endpoint ports.
+ maxProperties: 128
+ type: object
+ serviceAccount:
+ description: Service account used to derive workload identity.
+ maxLength: 253
+ type: string
+ weight:
+ description: Relative load-balancing weight.
+ maximum: 4294967295
+ minimum: 0
+ type: integer
+ type: object
+ status:
+ properties:
+ conditions:
+ items:
+ properties:
+ observedGeneration:
+ anyOf:
+ - type: integer
+ - type: string
+ x-kubernetes-int-or-string: true
+ reason:
+ type: string
+ status:
+ type: string
+ type:
+ type: string
+ type: object
+ type: array
+ type: object
+ x-kubernetes-preserve-unknown-fields: true
+ type: object
+ served: true
+ storage: true
+ subresources:
+ status: {}
+---
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
metadata:
annotations:
"helm.sh/resource-policy": keep
diff --git a/pkg/config/protocol/instance.go b/pkg/config/protocol/instance.go
index 74733285..95d3cd51 100644
--- a/pkg/config/protocol/instance.go
+++ b/pkg/config/protocol/instance.go
@@ -26,6 +26,12 @@ func (i Instance) String() string {
}
const (
+ // HTTP declares that the port carries HTTP/1.1 traffic.
+ HTTP Instance = "HTTP"
+ // TCP declares that the port carries opaque TCP traffic.
+ TCP Instance = "TCP"
+ // UDP declares that the port carries UDP traffic.
+ UDP Instance = "UDP"
// GRPC declares that the port carries gRPC traffic.
GRPC Instance = "GRPC"
// GRPCWeb declares that the port carries gRPC traffic.
@@ -44,6 +50,12 @@ const (
// Parse from string ignoring case
func Parse(s string) Instance {
switch strings.ToLower(s) {
+ case "http":
+ return HTTP
+ case "tcp":
+ return TCP
+ case "udp":
+ return UDP
case "grpc":
return GRPC
case "grpc-web":
diff --git a/pkg/config/schema/collections/collections.gen.go
b/pkg/config/schema/collections/collections.gen.go
index 63cbcc63..de2c2a0a 100755
--- a/pkg/config/schema/collections/collections.gen.go
+++ b/pkg/config/schema/collections/collections.gen.go
@@ -434,6 +434,21 @@ var (
ValidateProto: validation.EmptyValidate,
}.MustBuild()
+ ServiceEntry = resource.Builder{
+ Identifier: "ServiceEntry",
+ Group: "networking.dubbo.apache.org",
+ Kind: "ServiceEntry",
+ Plural: "serviceentries",
+ Version: "v1alpha3",
+ Proto: "dubbo.networking.v1alpha3.ServiceEntry",
StatusProto: "dubbo.meta.v1alpha1.DubboStatus",
+ ReflectType:
reflect.TypeOf(&githubcomkdubboapinetworkingv1alpha3.ServiceEntry{}).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.ValidateServiceEntry,
+ }.MustBuild()
+
StatefulSet = resource.Builder{
Identifier: "StatefulSet",
Group: "apps",
@@ -479,6 +494,21 @@ var (
ValidateProto: validation.EmptyValidate,
}.MustBuild()
+ WorkloadEntry = resource.Builder{
+ Identifier: "WorkloadEntry",
+ Group: "networking.dubbo.apache.org",
+ Kind: "WorkloadEntry",
+ Plural: "workloadentries",
+ Version: "v1alpha3",
+ Proto: "dubbo.networking.v1alpha3.WorkloadEntry",
StatusProto: "dubbo.meta.v1alpha1.DubboStatus",
+ ReflectType:
reflect.TypeOf(&githubcomkdubboapinetworkingv1alpha3.WorkloadEntry{}).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.ValidateWorkloadEntry,
+ }.MustBuild()
+
// All contains all collections in the system.
All = collection.NewSchemasBuilder().
MustAdd(AuthorizationPolicy).
@@ -507,9 +537,11 @@ var (
MustAdd(Secret).
MustAdd(Service).
MustAdd(ServiceAccount).
+ MustAdd(ServiceEntry).
MustAdd(StatefulSet).
MustAdd(Telemetry).
MustAdd(ValidatingWebhookConfiguration).
+ MustAdd(WorkloadEntry).
Build()
// Kube contains only kubernetes collections.
@@ -545,7 +577,9 @@ var (
MustAdd(CircuitBreakerPolicy).
MustAdd(PeerAuthentication).
MustAdd(RequestAuthentication).
+ MustAdd(ServiceEntry).
MustAdd(Telemetry).
+ MustAdd(WorkloadEntry).
Build()
// dubboGatewayAPI contains only collections used by Dubbo, including
the full Gateway API.
@@ -559,7 +593,9 @@ var (
MustAdd(PeerAuthentication).
MustAdd(ReferenceGrant).
MustAdd(RequestAuthentication).
+ MustAdd(ServiceEntry).
MustAdd(Telemetry).
+ MustAdd(WorkloadEntry).
Build()
// dubboStableGatewayAPI contains only collections used by Dubbo,
including beta+ Gateway API.
@@ -573,6 +609,8 @@ var (
MustAdd(PeerAuthentication).
MustAdd(ReferenceGrant).
MustAdd(RequestAuthentication).
+ MustAdd(ServiceEntry).
MustAdd(Telemetry).
+ MustAdd(WorkloadEntry).
Build()
)
diff --git a/pkg/config/schema/gvk/resources.gen.go
b/pkg/config/schema/gvk/resources.gen.go
index b8410c53..8cd44e99 100755
--- a/pkg/config/schema/gvk/resources.gen.go
+++ b/pkg/config/schema/gvk/resources.gen.go
@@ -41,9 +41,11 @@ 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"}
+ 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"}
ValidatingWebhookConfiguration = config.GroupVersionKind{Group:
"admissionregistration.k8s.io", Version: "v1", Kind:
"ValidatingWebhookConfiguration"}
+ WorkloadEntry = config.GroupVersionKind{Group:
"networking.dubbo.apache.org", Version: "v1alpha3", Kind: "WorkloadEntry"}
)
// ToGVR converts a GVK to a GVR.
@@ -111,12 +113,16 @@ func ToGVR(g config.GroupVersionKind)
(schema.GroupVersionResource, bool) {
return gvr.Service, true
case ServiceAccount:
return gvr.ServiceAccount, true
+ case ServiceEntry:
+ return gvr.ServiceEntry, true
case StatefulSet:
return gvr.StatefulSet, true
case Telemetry:
return gvr.Telemetry, true
case ValidatingWebhookConfiguration:
return gvr.ValidatingWebhookConfiguration, true
+ case WorkloadEntry:
+ return gvr.WorkloadEntry, true
}
return schema.GroupVersionResource{}, false
@@ -176,12 +182,16 @@ func MustToKind(g config.GroupVersionKind) kind.Kind {
return kind.Service
case ServiceAccount:
return kind.ServiceAccount
+ case ServiceEntry:
+ return kind.ServiceEntry
case StatefulSet:
return kind.StatefulSet
case Telemetry:
return kind.Telemetry
case ValidatingWebhookConfiguration:
return kind.ValidatingWebhookConfiguration
+ case WorkloadEntry:
+ return kind.WorkloadEntry
}
panic("unknown kind: " + g.String())
@@ -252,12 +262,16 @@ func FromGVR(g schema.GroupVersionResource)
(config.GroupVersionKind, bool) {
return Service, true
case gvr.ServiceAccount:
return ServiceAccount, true
+ case gvr.ServiceEntry:
+ return ServiceEntry, true
case gvr.StatefulSet:
return StatefulSet, true
case gvr.Telemetry:
return Telemetry, true
case gvr.ValidatingWebhookConfiguration:
return ValidatingWebhookConfiguration, true
+ case gvr.WorkloadEntry:
+ return WorkloadEntry, true
}
return config.GroupVersionKind{}, false
diff --git a/pkg/config/schema/gvr/resources.gen.go
b/pkg/config/schema/gvr/resources.gen.go
index b75932c4..62f90f2a 100755
--- a/pkg/config/schema/gvr/resources.gen.go
+++ b/pkg/config/schema/gvr/resources.gen.go
@@ -36,9 +36,11 @@ 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"}
+ 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"}
ValidatingWebhookConfiguration = schema.GroupVersionResource{Group:
"admissionregistration.k8s.io", Version: "v1", Resource:
"validatingwebhookconfigurations"}
+ WorkloadEntry = schema.GroupVersionResource{Group:
"networking.dubbo.apache.org", Version: "v1alpha3", Resource: "workloadentries"}
)
func IsClusterScoped(g schema.GroupVersionResource) bool {
@@ -103,12 +105,16 @@ func IsClusterScoped(g schema.GroupVersionResource) bool {
return false
case ServiceAccount:
return false
+ case ServiceEntry:
+ return false
case StatefulSet:
return false
case Telemetry:
return false
case ValidatingWebhookConfiguration:
return true
+ case WorkloadEntry:
+ return false
}
// shouldn't happen
return false
diff --git a/pkg/config/schema/kind/resources.gen.go
b/pkg/config/schema/kind/resources.gen.go
index 77fe9b98..97193ecb 100755
--- a/pkg/config/schema/kind/resources.gen.go
+++ b/pkg/config/schema/kind/resources.gen.go
@@ -32,9 +32,11 @@ const (
Secret
Service
ServiceAccount
+ ServiceEntry
StatefulSet
Telemetry
ValidatingWebhookConfiguration
+ WorkloadEntry
)
func (k Kind) String() string {
@@ -95,12 +97,16 @@ func (k Kind) String() string {
return "Service"
case ServiceAccount:
return "ServiceAccount"
+ case ServiceEntry:
+ return "ServiceEntry"
case StatefulSet:
return "StatefulSet"
case Telemetry:
return "Telemetry"
case ValidatingWebhookConfiguration:
return "ValidatingWebhookConfiguration"
+ case WorkloadEntry:
+ return "WorkloadEntry"
default:
return "Unknown"
}
@@ -164,12 +170,16 @@ func FromString(s string) Kind {
return Service
case "ServiceAccount":
return ServiceAccount
+ case "ServiceEntry":
+ return ServiceEntry
case "StatefulSet":
return StatefulSet
case "Telemetry":
return Telemetry
case "ValidatingWebhookConfiguration":
return ValidatingWebhookConfiguration
+ case "WorkloadEntry":
+ return WorkloadEntry
default:
return Unknown
}
diff --git a/pkg/config/schema/kubeclient/resources.gen.go
b/pkg/config/schema/kubeclient/resources.gen.go
index 2a88d03c..e63b7a7c 100755
--- a/pkg/config/schema/kubeclient/resources.gen.go
+++ b/pkg/config/schema/kubeclient/resources.gen.go
@@ -84,12 +84,16 @@ 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.ServiceEntry:
+ return
c.Dubbo().NetworkingV1alpha3().ServiceEntries(namespace).(ktypes.WriteAPI[T])
case *k8sioapiappsv1.StatefulSet:
return
c.Kube().AppsV1().StatefulSets(namespace).(ktypes.WriteAPI[T])
case *apigithubcomapachedubbokubernetesapitelemetryv1alpha1.Telemetry:
return
c.Dubbo().TelemetryV1alpha1().Telemetries(namespace).(ktypes.WriteAPI[T])
case *k8sioapiadmissionregistrationv1.ValidatingWebhookConfiguration:
return
c.Kube().AdmissionregistrationV1().ValidatingWebhookConfigurations().(ktypes.WriteAPI[T])
+ case
*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.WorkloadEntry:
+ return
c.Dubbo().NetworkingV1alpha3().WorkloadEntries(namespace).(ktypes.WriteAPI[T])
default:
panic(fmt.Sprintf("Unknown type %T", ptr.Empty[T]()))
}
@@ -147,12 +151,16 @@ 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.ServiceEntry:
+ return
c.Dubbo().NetworkingV1alpha3().ServiceEntries(namespace).(ktypes.ReadWriteAPI[T,
TL])
case *k8sioapiappsv1.StatefulSet:
return
c.Kube().AppsV1().StatefulSets(namespace).(ktypes.ReadWriteAPI[T, TL])
case *apigithubcomapachedubbokubernetesapitelemetryv1alpha1.Telemetry:
return
c.Dubbo().TelemetryV1alpha1().Telemetries(namespace).(ktypes.ReadWriteAPI[T,
TL])
case *k8sioapiadmissionregistrationv1.ValidatingWebhookConfiguration:
return
c.Kube().AdmissionregistrationV1().ValidatingWebhookConfigurations().(ktypes.ReadWriteAPI[T,
TL])
+ case
*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.WorkloadEntry:
+ return
c.Dubbo().NetworkingV1alpha3().WorkloadEntries(namespace).(ktypes.ReadWriteAPI[T,
TL])
default:
panic(fmt.Sprintf("Unknown type %T", ptr.Empty[T]()))
}
@@ -210,12 +218,16 @@ func gvrToObject(g schema.GroupVersionResource)
runtime.Object {
return &k8sioapicorev1.Service{}
case gvr.ServiceAccount:
return &k8sioapicorev1.ServiceAccount{}
+ case gvr.ServiceEntry:
+ return
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceEntry{}
case gvr.StatefulSet:
return &k8sioapiappsv1.StatefulSet{}
case gvr.Telemetry:
return
&apigithubcomapachedubbokubernetesapitelemetryv1alpha1.Telemetry{}
case gvr.ValidatingWebhookConfiguration:
return
&k8sioapiadmissionregistrationv1.ValidatingWebhookConfiguration{}
+ case gvr.WorkloadEntry:
+ return
&apigithubcomapachedubbokubernetesapinetworkingv1alpha3.WorkloadEntry{}
default:
panic(fmt.Sprintf("Unknown type %v", g))
}
@@ -401,6 +413,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.ServiceEntry:
+ l = func(options metav1.ListOptions) (runtime.Object, error) {
+ return
c.Dubbo().NetworkingV1alpha3().ServiceEntries(opts.Namespace).List(context.Background(),
options)
+ }
+ w = func(options metav1.ListOptions) (watch.Interface, error) {
+ return
c.Dubbo().NetworkingV1alpha3().ServiceEntries(opts.Namespace).Watch(context.Background(),
options)
+ }
case gvr.StatefulSet:
l = func(options metav1.ListOptions) (runtime.Object, error) {
return
c.Kube().AppsV1().StatefulSets(opts.Namespace).List(context.Background(),
options)
@@ -422,6 +441,13 @@ func getInformerFiltered(c ClientGetter, opts
ktypes.InformerOptions, g schema.G
w = func(options metav1.ListOptions) (watch.Interface, error) {
return
c.Kube().AdmissionregistrationV1().ValidatingWebhookConfigurations().Watch(context.Background(),
options)
}
+ case gvr.WorkloadEntry:
+ l = func(options metav1.ListOptions) (runtime.Object, error) {
+ return
c.Dubbo().NetworkingV1alpha3().WorkloadEntries(opts.Namespace).List(context.Background(),
options)
+ }
+ w = func(options metav1.ListOptions) (watch.Interface, error) {
+ return
c.Dubbo().NetworkingV1alpha3().WorkloadEntries(opts.Namespace).Watch(context.Background(),
options)
+ }
default:
panic(fmt.Sprintf("Unknown type %v", g))
}
diff --git a/pkg/config/schema/kubetypes/resources.gen.go
b/pkg/config/schema/kubetypes/resources.gen.go
index a395b2f8..864ee1ad 100755
--- a/pkg/config/schema/kubetypes/resources.gen.go
+++ b/pkg/config/schema/kubetypes/resources.gen.go
@@ -86,6 +86,10 @@ func getGvk(obj any) (config.GroupVersionKind, bool) {
return gvk.Service, true
case *k8sioapicorev1.ServiceAccount:
return gvk.ServiceAccount, true
+ case *githubcomkdubboapinetworkingv1alpha3.ServiceEntry:
+ return gvk.ServiceEntry, true
+ case
*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.ServiceEntry:
+ return gvk.ServiceEntry, true
case *k8sioapiappsv1.StatefulSet:
return gvk.StatefulSet, true
case *githubcomkdubboapitelemetryv1alpha1.Telemetry:
@@ -94,6 +98,10 @@ func getGvk(obj any) (config.GroupVersionKind, bool) {
return gvk.Telemetry, true
case *k8sioapiadmissionregistrationv1.ValidatingWebhookConfiguration:
return gvk.ValidatingWebhookConfiguration, true
+ case *githubcomkdubboapinetworkingv1alpha3.WorkloadEntry:
+ return gvk.WorkloadEntry, true
+ case
*apigithubcomapachedubbokubernetesapinetworkingv1alpha3.WorkloadEntry:
+ return gvk.WorkloadEntry, true
default:
return config.GroupVersionKind{}, false
}
diff --git a/pkg/config/schema/metadata.yaml b/pkg/config/schema/metadata.yaml
index cb22e54b..68d53812 100644
--- a/pkg/config/schema/metadata.yaml
+++ b/pkg/config/schema/metadata.yaml
@@ -230,6 +230,24 @@ resources:
statusProto: "dubbo.meta.v1alpha1.DubboStatus"
statusProtoPackage: "github.com/kdubbo/api/meta/v1alpha1"
+ - kind: ServiceEntry
+ plural: "serviceentries"
+ group: "networking.dubbo.apache.org"
+ version: "v1alpha3"
+ proto: "dubbo.networking.v1alpha3.ServiceEntry"
+ protoPackage: "github.com/kdubbo/api/networking/v1alpha3"
+ statusProto: "dubbo.meta.v1alpha1.DubboStatus"
+ statusProtoPackage: "github.com/kdubbo/api/meta/v1alpha1"
+
+ - kind: WorkloadEntry
+ plural: "workloadentries"
+ group: "networking.dubbo.apache.org"
+ version: "v1alpha3"
+ proto: "dubbo.networking.v1alpha3.WorkloadEntry"
+ protoPackage: "github.com/kdubbo/api/networking/v1alpha3"
+ statusProto: "dubbo.meta.v1alpha1.DubboStatus"
+ statusProtoPackage: "github.com/kdubbo/api/meta/v1alpha1"
+
- kind: "MeshConfig"
plural: "meshconfigs"
group: ""
diff --git a/pkg/config/validation/validators.go
b/pkg/config/validation/validators.go
index fdb93ebd..0b4fa91f 100644
--- a/pkg/config/validation/validators.go
+++ b/pkg/config/validation/validators.go
@@ -19,13 +19,18 @@ package validation
import (
"fmt"
"math"
+ "net"
"strings"
"github.com/apache/dubbo-kubernetes/pkg/config"
"github.com/apache/dubbo-kubernetes/pkg/config/constants"
+ "github.com/apache/dubbo-kubernetes/pkg/config/labels"
+ "github.com/apache/dubbo-kubernetes/pkg/config/protocol"
+ "github.com/apache/dubbo-kubernetes/pkg/config/visibility"
networking "github.com/kdubbo/api/networking/v1alpha3"
security "github.com/kdubbo/api/security/v1alpha3"
telemetry "github.com/kdubbo/api/telemetry/v1alpha1"
+ kvalidation "k8s.io/apimachinery/pkg/util/validation"
)
// ValidateAuthorizationPolicy checks that an AuthorizationPolicy is
well-formed.
@@ -208,6 +213,125 @@ var ValidateCircuitBreakerPolicy = validateFunc(
return v.Unwrap()
})
+// ValidateServiceEntry checks that a ServiceEntry can be converted into
services and endpoints.
+var ValidateServiceEntry = RegisterValidateFunc("ValidateServiceEntry",
func(cfg config.Config) (Warning, error) {
+ spec, ok := cfg.Spec.(*networking.ServiceEntry)
+ if !ok {
+ return nil, fmt.Errorf("cannot cast to ServiceEntry")
+ }
+ v := Validation{}
+ if len(spec.GetHosts()) == 0 {
+ v = appendValidation(v, fmt.Errorf("hosts must not be empty"))
+ }
+ for i, hostname := range spec.GetHosts() {
+ v = appendValidation(v,
validateServiceEntryHost(fmt.Sprintf("hosts[%d]", i), hostname))
+ }
+ for i, address := range spec.GetAddresses() {
+ if net.ParseIP(address) == nil {
+ if _, _, err := net.ParseCIDR(address); err != nil {
+ v = appendValidation(v,
fmt.Errorf("addresses[%d] %q must be an IP address or CIDR", i, address))
+ }
+ }
+ }
+ if len(spec.GetPorts()) == 0 {
+ v = appendValidation(v, fmt.Errorf("ports must not be empty"))
+ }
+ names := make(map[string]struct{}, len(spec.GetPorts()))
+ numbers := make(map[uint32]struct{}, len(spec.GetPorts()))
+ for i, port := range spec.GetPorts() {
+ if port == nil {
+ v = appendValidation(v, fmt.Errorf("ports[%d] must not
be null", i))
+ continue
+ }
+ if port.GetName() == "" {
+ v = appendValidation(v, fmt.Errorf("ports[%d].name must
not be empty", i))
+ } else if _, found := names[port.GetName()]; found {
+ v = appendValidation(v, fmt.Errorf("ports[%d].name %q
is duplicated", i, port.GetName()))
+ }
+ names[port.GetName()] = struct{}{}
+ if port.GetNumber() == 0 || port.GetNumber() > 65535 {
+ v = appendValidation(v, fmt.Errorf("ports[%d].number
must be between 1 and 65535", i))
+ } else if _, found := numbers[port.GetNumber()]; found {
+ v = appendValidation(v, fmt.Errorf("ports[%d].number %d
is duplicated", i, port.GetNumber()))
+ }
+ numbers[port.GetNumber()] = struct{}{}
+ if port.GetTargetPort() > 65535 {
+ v = appendValidation(v,
fmt.Errorf("ports[%d].targetPort must be between 1 and 65535 when set", i))
+ }
+ if protocol.Parse(port.GetProtocol()) == protocol.Unsupported {
+ v = appendValidation(v, fmt.Errorf("ports[%d].protocol
%q is unsupported", i, port.GetProtocol()))
+ }
+ }
+ if spec.GetWorkloadSelector() != nil && len(spec.GetEndpoints()) > 0 {
+ v = appendValidation(v, fmt.Errorf("only one of
workloadSelector or endpoints can be set"))
+ }
+ v = appendValidation(v,
validateWorkloadSelector(spec.GetWorkloadSelector()))
+ for i, endpoint := range spec.GetEndpoints() {
+ v = appendValidation(v,
validateWorkloadEntry(fmt.Sprintf("endpoints[%d]", i), endpoint))
+ }
+ seenExport := make(map[string]struct{}, len(spec.GetExportTo()))
+ for i, export := range spec.GetExportTo() {
+ if err := visibility.Instance(export).Validate(); err != nil {
+ v = appendValidation(v, fmt.Errorf("exportTo[%d]: %v",
i, err))
+ }
+ if _, found := seenExport[export]; found {
+ v = appendValidation(v, fmt.Errorf("exportTo[%d] %q is
duplicated", i, export))
+ }
+ seenExport[export] = struct{}{}
+ }
+ return v.Unwrap()
+})
+
+// ValidateWorkloadEntry checks that a WorkloadEntry can produce a routable
endpoint.
+var ValidateWorkloadEntry = RegisterValidateFunc("ValidateWorkloadEntry",
func(cfg config.Config) (Warning, error) {
+ spec, ok := cfg.Spec.(*networking.WorkloadEntry)
+ if !ok {
+ return nil, fmt.Errorf("cannot cast to WorkloadEntry")
+ }
+ v := Validation{}
+ v = appendValidation(v, validateWorkloadEntry("workloadEntry", spec))
+ return v.Unwrap()
+})
+
+func validateServiceEntryHost(field, value string) error {
+ if value == "" {
+ return fmt.Errorf("%s must not be empty", field)
+ }
+ if net.ParseIP(value) != nil {
+ return nil
+ }
+ hostname := strings.TrimPrefix(value, "*.")
+ if messages := kvalidation.IsDNS1123Subdomain(hostname); len(messages)
> 0 {
+ return fmt.Errorf("%s %q is not a valid DNS name: %v", field,
value, messages)
+ }
+ if strings.Contains(value, "*") && !strings.HasPrefix(value, "*.") {
+ return fmt.Errorf("%s %q has an invalid wildcard", field, value)
+ }
+ return nil
+}
+
+func validateWorkloadEntry(field string, workload *networking.WorkloadEntry)
error {
+ if workload == nil {
+ return fmt.Errorf("%s must not be null", field)
+ }
+ var errs error
+ if err := validateServiceEntryHost(field+".address",
workload.GetAddress()); err != nil {
+ errs = AppendErrors(errs, err)
+ }
+ for name, port := range workload.GetPorts() {
+ if name == "" {
+ errs = AppendErrors(errs, fmt.Errorf("%s.ports contains
an empty name", field))
+ }
+ if port == 0 || port > 65535 {
+ errs = AppendErrors(errs, fmt.Errorf("%s.ports[%q] must
be between 1 and 65535", field, name))
+ }
+ }
+ if err := labels.Instance(workload.GetLabels()).Validate(); err != nil {
+ errs = AppendErrors(errs, fmt.Errorf("%s.labels: %v", field,
err))
+ }
+ return errs
+}
+
// ValidateTelemetry checks that a Telemetry resource is well-formed.
var ValidateTelemetry = RegisterValidateFunc("ValidateTelemetry",
func(cfg config.Config) (Warning, error) {
diff --git a/pkg/config/validation/validators_test.go
b/pkg/config/validation/validators_test.go
index a8d2c9f2..c6276608 100644
--- a/pkg/config/validation/validators_test.go
+++ b/pkg/config/validation/validators_test.go
@@ -295,6 +295,68 @@ func TestValidateCircuitBreakerPolicy(t *testing.T) {
}
}
+func TestValidateServiceEntry(t *testing.T) {
+ valid := func() *networking.ServiceEntry {
+ return &networking.ServiceEntry{
+ Hosts: []string{"api.example.com"},
+ Addresses: []string{"240.0.0.1"},
+ Resolution: networking.ServiceEntry_STATIC,
+ Ports: []*networking.ServicePort{{Name: "tcp",
Number: 443, Protocol: "TCP"}},
+ Endpoints: []*networking.WorkloadEntry{{Address:
"10.0.0.1", Ports: map[string]uint32{"tcp": 8443}}},
+ }
+ }
+ cases := []struct {
+ name string
+ mutate func(*networking.ServiceEntry)
+ wantErr bool
+ }{
+ {name: "valid", mutate: func(*networking.ServiceEntry) {},
wantErr: false},
+ {name: "missing host", mutate: func(s *networking.ServiceEntry)
{ s.Hosts = nil }, wantErr: true},
+ {name: "bad address", mutate: func(s *networking.ServiceEntry)
{ s.Addresses = []string{"not-an-ip"} }, wantErr: true},
+ {name: "duplicate port", mutate: func(s
*networking.ServiceEntry) {
+ s.Ports = append(s.Ports, &networking.ServicePort{Name:
"tcp", Number: 443, Protocol: "TCP"})
+ }, wantErr: true},
+ {name: "unsupported protocol", mutate: func(s
*networking.ServiceEntry) { s.Ports[0].Protocol = "SMTP" }, wantErr: true},
+ {name: "selector and endpoints", mutate: func(s
*networking.ServiceEntry) {
+ s.WorkloadSelector =
&typev1alpha3.WorkloadSelector{MatchLabels: map[string]string{"app": "api"}}
+ }, wantErr: true},
+ {name: "invalid endpoint", mutate: func(s
*networking.ServiceEntry) { s.Endpoints[0].Address = "" }, wantErr: true},
+ {name: "invalid export", mutate: func(s
*networking.ServiceEntry) { s.ExportTo = []string{"bad_namespace"} }, wantErr:
true},
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ spec := valid()
+ tc.mutate(spec)
+ _, err := ValidateServiceEntry(makeConfig(spec))
+ if got := err != nil; got != tc.wantErr {
+ t.Fatalf("ValidateServiceEntry() error = %v,
wantErr = %v", err, tc.wantErr)
+ }
+ })
+ }
+}
+
+func TestValidateWorkloadEntry(t *testing.T) {
+ cases := []struct {
+ name string
+ spec *networking.WorkloadEntry
+ wantErr bool
+ }{
+ {name: "valid IP", spec: &networking.WorkloadEntry{Address:
"10.0.0.1", Ports: map[string]uint32{"grpc": 50051}}, wantErr: false},
+ {name: "valid DNS", spec: &networking.WorkloadEntry{Address:
"vm.example.com"}, wantErr: false},
+ {name: "missing address", spec: &networking.WorkloadEntry{},
wantErr: true},
+ {name: "invalid port", spec: &networking.WorkloadEntry{Address:
"10.0.0.1", Ports: map[string]uint32{"grpc": 0}}, wantErr: true},
+ {name: "invalid labels", spec:
&networking.WorkloadEntry{Address: "10.0.0.1", Labels: map[string]string{"bad
key": "value"}}, wantErr: true},
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ _, err := ValidateWorkloadEntry(makeConfig(tc.spec))
+ if got := err != nil; got != tc.wantErr {
+ t.Fatalf("ValidateWorkloadEntry() error = %v,
wantErr = %v", err, tc.wantErr)
+ }
+ })
+ }
+}
+
func TestValidateTelemetry(t *testing.T) {
cases := []struct {
name string
