You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by zh...@apache.org on 2025/10/05 19:39:11 UTC
(dubbo-kubernetes) branch master updated: [discovery] Update xds ads and webhook (#801)
This is an automated email from the ASF dual-hosted git repository.
zhongxjian 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 1aede46a [discovery] Update xds ads and webhook (#801)
1aede46a is described below
commit 1aede46a2f123a80627bfd5e90d60cd35edf76d4
Author: Jian Zhong <11...@qq.com>
AuthorDate: Mon Oct 6 03:39:06 2025 +0800
[discovery] Update xds ads and webhook (#801)
---
.../schema/{resource => collection}/schema.go | 2 +-
pkg/config/schema/collection/schemas.go | 29 ++++-----
pkg/config/schema/collections/collections.agent.go | 13 ++--
pkg/config/schema/collections/collections.go | 21 +++---
pkg/config/schema/gvk/resources.go | 18 ++---
pkg/config/schema/gvr/resources.go | 10 +--
pkg/config/schema/kind/resources.go | 3 +
pkg/kube/inject/webhook.go | 10 ++-
pkg/webhooks/server/server.go | 3 +-
pkg/webhooks/validation/controller/controller.go | 3 +-
pkg/webhooks/webhookpatch.go | 2 +
sail/pkg/bootstrap/configcontroller.go | 2 +-
sail/pkg/bootstrap/server.go | 14 ++--
sail/pkg/config/aggregate/config.go | 2 +-
sail/pkg/config/kube/crd/conversion.go | 12 ++--
sail/pkg/config/kube/crdclient/client.go | 5 +-
sail/pkg/config/kube/file/controller.go | 2 +-
sail/pkg/credentials/kube/secrets.go | 18 +----
sail/pkg/model/push_context.go | 76 +++++++++-------------
sail/pkg/xds/discovery.go | 8 +--
sail/pkg/xds/eds.go | 5 +-
21 files changed, 118 insertions(+), 140 deletions(-)
diff --git a/pkg/config/schema/resource/schema.go b/pkg/config/schema/collection/schema.go
similarity index 99%
rename from pkg/config/schema/resource/schema.go
rename to pkg/config/schema/collection/schema.go
index 5012b662..015ab2a6 100644
--- a/pkg/config/schema/resource/schema.go
+++ b/pkg/config/schema/collection/schema.go
@@ -1,4 +1,4 @@
-package resource
+package collection
import (
"errors"
diff --git a/pkg/config/schema/collection/schemas.go b/pkg/config/schema/collection/schemas.go
index d5cadbf2..f2eb0a44 100644
--- a/pkg/config/schema/collection/schemas.go
+++ b/pkg/config/schema/collection/schemas.go
@@ -9,18 +9,17 @@ import (
"k8s.io/apimachinery/pkg/runtime/schema"
"github.com/apache/dubbo-kubernetes/pkg/config"
- "github.com/apache/dubbo-kubernetes/pkg/config/schema/resource"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
)
// Schemas contains metadata about configuration resources.
type Schemas struct {
- byCollection map[config.GroupVersionKind]resource.Schema
- byAddOrder []resource.Schema
+ byCollection map[config.GroupVersionKind]Schema
+ byAddOrder []Schema
}
// SchemasFor is a shortcut for creating Schemas. It uses MustAdd for each element.
-func SchemasFor(schemas ...resource.Schema) Schemas {
+func SchemasFor(schemas ...Schema) Schemas {
b := NewSchemasBuilder()
for _, s := range schemas {
b.MustAdd(s)
@@ -36,7 +35,7 @@ type SchemasBuilder struct {
// NewSchemasBuilder returns a new instance of SchemasBuilder.
func NewSchemasBuilder() *SchemasBuilder {
s := Schemas{
- byCollection: make(map[config.GroupVersionKind]resource.Schema),
+ byCollection: make(map[config.GroupVersionKind]Schema),
}
return &SchemasBuilder{
@@ -45,7 +44,7 @@ func NewSchemasBuilder() *SchemasBuilder {
}
// Add a new collection to the schemas.
-func (b *SchemasBuilder) Add(s resource.Schema) error {
+func (b *SchemasBuilder) Add(s Schema) error {
if _, found := b.schemas.byCollection[s.GroupVersionKind()]; found {
return fmt.Errorf("collection already exists: %v", s.GroupVersionKind())
}
@@ -56,7 +55,7 @@ func (b *SchemasBuilder) Add(s resource.Schema) error {
}
// MustAdd calls Add and panics if it fails.
-func (b *SchemasBuilder) MustAdd(s resource.Schema) *SchemasBuilder {
+func (b *SchemasBuilder) MustAdd(s Schema) *SchemasBuilder {
if err := b.Add(s); err != nil {
panic(fmt.Sprintf("SchemasBuilder.MustAdd: %v", err))
}
@@ -74,7 +73,7 @@ func (b *SchemasBuilder) Build() Schemas {
}
// ForEach executes the given function on each contained schema, until the function returns true.
-func (s Schemas) ForEach(handleSchema func(resource.Schema) (done bool)) {
+func (s Schemas) ForEach(handleSchema func(Schema) (done bool)) {
for _, schema := range s.byAddOrder {
if handleSchema(schema) {
return
@@ -113,7 +112,7 @@ func (s Schemas) Intersect(otherSchemas Schemas) Schemas {
}
// FindByGroupVersionKind searches and returns the first schema with the given GVK
-func (s Schemas) FindByGroupVersionKind(gvk config.GroupVersionKind) (resource.Schema, bool) {
+func (s Schemas) FindByGroupVersionKind(gvk config.GroupVersionKind) (Schema, bool) {
for _, rs := range s.byAddOrder {
if rs.GroupVersionKind() == gvk {
return rs, true
@@ -125,7 +124,7 @@ func (s Schemas) FindByGroupVersionKind(gvk config.GroupVersionKind) (resource.S
// FindByGroupVersionAliasesKind searches and returns the first schema with the given GVK,
// if not found, it will search for version aliases for the schema to see if there is a match.
-func (s Schemas) FindByGroupVersionAliasesKind(gvk config.GroupVersionKind) (resource.Schema, bool) {
+func (s Schemas) FindByGroupVersionAliasesKind(gvk config.GroupVersionKind) (Schema, bool) {
for _, rs := range s.byAddOrder {
for _, va := range rs.GroupVersionAliasKinds() {
if va == gvk {
@@ -139,7 +138,7 @@ func (s Schemas) FindByGroupVersionAliasesKind(gvk config.GroupVersionKind) (res
// FindByGroupKind searches and returns the first schema with the given GVK, ignoring versions.
// Generally it's a good idea to use FindByGroupVersionAliasesKind, which validates the version as well.
// FindByGroupKind provides future proofing against versions we don't yet know about; given we don't know them, its risky.
-func (s Schemas) FindByGroupKind(gvk config.GroupVersionKind) (resource.Schema, bool) {
+func (s Schemas) FindByGroupKind(gvk config.GroupVersionKind) (Schema, bool) {
for _, rs := range s.byAddOrder {
if rs.Group() == gvk.Group && rs.Kind() == gvk.Kind {
return rs, true
@@ -149,7 +148,7 @@ func (s Schemas) FindByGroupKind(gvk config.GroupVersionKind) (resource.Schema,
}
// FindByGroupVersionResource searches and returns the first schema with the given GVR
-func (s Schemas) FindByGroupVersionResource(gvr schema.GroupVersionResource) (resource.Schema, bool) {
+func (s Schemas) FindByGroupVersionResource(gvr schema.GroupVersionResource) (Schema, bool) {
for _, rs := range s.byAddOrder {
if rs.GroupVersionResource() == gvr {
return rs, true
@@ -160,7 +159,7 @@ func (s Schemas) FindByGroupVersionResource(gvr schema.GroupVersionResource) (re
}
// All returns all known Schemas
-func (s Schemas) All() []resource.Schema {
+func (s Schemas) All() []Schema {
return slices.Clone(s.byAddOrder)
}
@@ -174,7 +173,7 @@ func (s Schemas) GroupVersionKinds() []config.GroupVersionKind {
}
// Add creates a copy of this Schemas with the given schemas added.
-func (s Schemas) Add(toAdd ...resource.Schema) Schemas {
+func (s Schemas) Add(toAdd ...Schema) Schemas {
b := NewSchemasBuilder()
for _, s := range s.byAddOrder {
@@ -189,7 +188,7 @@ func (s Schemas) Add(toAdd ...resource.Schema) Schemas {
}
// Remove creates a copy of this Schemas with the given schemas removed.
-func (s Schemas) Remove(toRemove ...resource.Schema) Schemas {
+func (s Schemas) Remove(toRemove ...Schema) Schemas {
b := NewSchemasBuilder()
for _, s := range s.byAddOrder {
diff --git a/pkg/config/schema/collections/collections.agent.go b/pkg/config/schema/collections/collections.agent.go
index 21bd9682..a840452a 100644
--- a/pkg/config/schema/collections/collections.agent.go
+++ b/pkg/config/schema/collections/collections.agent.go
@@ -5,7 +5,6 @@ package collections
import (
"github.com/apache/dubbo-kubernetes/pkg/config/schema/collection"
- "github.com/apache/dubbo-kubernetes/pkg/config/schema/resource"
istioioapimetav1alpha1 "istio.io/api/meta/v1alpha1"
istioioapinetworkingv1alpha3 "istio.io/api/networking/v1alpha3"
istioioapisecurityv1beta1 "istio.io/api/security/v1beta1"
@@ -14,7 +13,7 @@ import (
)
var (
- PeerAuthentication = resource.Builder{
+ PeerAuthentication = collection.Builder{
Identifier: "PeerAuthentication",
Group: "security.istio.io",
Kind: "PeerAuthentication",
@@ -30,7 +29,7 @@ var (
Synthetic: false,
Builtin: false,
}.MustBuild()
- RequestAuthentication = resource.Builder{
+ RequestAuthentication = collection.Builder{
Identifier: "RequestAuthentication",
Group: "security.istio.io",
Kind: "RequestAuthentication",
@@ -46,7 +45,7 @@ var (
Synthetic: false,
Builtin: false,
}.MustBuild()
- DestinationRule = resource.Builder{
+ DestinationRule = collection.Builder{
Identifier: "DestinationRule",
Group: "networking.istio.io",
Kind: "DestinationRule",
@@ -63,7 +62,7 @@ var (
Synthetic: false,
Builtin: false,
}.MustBuild()
- VirtualService = resource.Builder{
+ VirtualService = collection.Builder{
Identifier: "VirtualService",
Group: "networking.istio.io",
Kind: "VirtualService",
@@ -80,7 +79,7 @@ var (
Synthetic: false,
Builtin: false,
}.MustBuild()
- ValidatingWebhookConfiguration = resource.Builder{
+ ValidatingWebhookConfiguration = collection.Builder{
Identifier: "ValidatingWebhookConfiguration",
Group: "admissionregistration.k8s.io",
Kind: "ValidatingWebhookConfiguration",
@@ -93,7 +92,7 @@ var (
Synthetic: false,
Builtin: true,
}.MustBuild()
- MutatingWebhookConfiguration = resource.Builder{
+ MutatingWebhookConfiguration = collection.Builder{
Identifier: "MutatingWebhookConfiguration",
Group: "admissionregistration.k8s.io",
Kind: "MutatingWebhookConfiguration",
diff --git a/pkg/config/schema/collections/collections.go b/pkg/config/schema/collections/collections.go
index ace155e2..457283ad 100644
--- a/pkg/config/schema/collections/collections.go
+++ b/pkg/config/schema/collections/collections.go
@@ -5,7 +5,6 @@ package collections
import (
"github.com/apache/dubbo-kubernetes/pkg/config/schema/collection"
- "github.com/apache/dubbo-kubernetes/pkg/config/schema/resource"
istioioapimetav1alpha1 "istio.io/api/meta/v1alpha1"
istioioapinetworkingv1alpha3 "istio.io/api/networking/v1alpha3"
istioioapisecurityv1beta1 "istio.io/api/security/v1beta1"
@@ -14,9 +13,9 @@ import (
)
var (
- PeerAuthentication = resource.Builder{
+ PeerAuthentication = collection.Builder{
Identifier: "PeerAuthentication",
- Group: "security.istio.io",
+ Group: "security.dubbo.io",
Kind: "PeerAuthentication",
Plural: "peerauthentications",
Version: "v1",
@@ -30,9 +29,9 @@ var (
Synthetic: false,
Builtin: false,
}.MustBuild()
- RequestAuthentication = resource.Builder{
+ RequestAuthentication = collection.Builder{
Identifier: "RequestAuthentication",
- Group: "security.istio.io",
+ Group: "security.dubbo.io",
Kind: "RequestAuthentication",
Plural: "requestauthentications",
Version: "v1",
@@ -46,9 +45,9 @@ var (
Synthetic: false,
Builtin: false,
}.MustBuild()
- DestinationRule = resource.Builder{
+ DestinationRule = collection.Builder{
Identifier: "DestinationRule",
- Group: "networking.istio.io",
+ Group: "networking.dubbo.io",
Kind: "DestinationRule",
Plural: "destinationrules",
Version: "v1",
@@ -63,9 +62,9 @@ var (
Synthetic: false,
Builtin: false,
}.MustBuild()
- VirtualService = resource.Builder{
+ VirtualService = collection.Builder{
Identifier: "VirtualService",
- Group: "networking.istio.io",
+ Group: "networking.dubbo.io",
Kind: "VirtualService",
Plural: "virtualservices",
Version: "v1",
@@ -80,7 +79,7 @@ var (
Synthetic: false,
Builtin: false,
}.MustBuild()
- ValidatingWebhookConfiguration = resource.Builder{
+ ValidatingWebhookConfiguration = collection.Builder{
Identifier: "ValidatingWebhookConfiguration",
Group: "admissionregistration.k8s.io",
Kind: "ValidatingWebhookConfiguration",
@@ -93,7 +92,7 @@ var (
Synthetic: false,
Builtin: true,
}.MustBuild()
- MutatingWebhookConfiguration = resource.Builder{
+ MutatingWebhookConfiguration = collection.Builder{
Identifier: "MutatingWebhookConfiguration",
Group: "admissionregistration.k8s.io",
Kind: "MutatingWebhookConfiguration",
diff --git a/pkg/config/schema/gvk/resources.go b/pkg/config/schema/gvk/resources.go
index 2a4af1fb..8fb566d9 100644
--- a/pkg/config/schema/gvk/resources.go
+++ b/pkg/config/schema/gvk/resources.go
@@ -37,11 +37,11 @@ var (
Service = config.GroupVersionKind{Group: "", Version: "v1", Kind: "Service"}
ServiceAccount = config.GroupVersionKind{Group: "", Version: "v1", Kind: "ServiceAccount"}
MeshConfig = config.GroupVersionKind{Group: "", Version: "v1alpha1", Kind: "MeshConfig"}
- RequestAuthentication = config.GroupVersionKind{Group: "security.istio.io", Version: "v1", Kind: "RequestAuthentication"}
- PeerAuthentication = config.GroupVersionKind{Group: "security.istio.io", Version: "v1", Kind: "PeerAuthentication"}
- AuthorizationPolicy = config.GroupVersionKind{Group: "security.istio.io", Version: "v1", Kind: "AuthorizationPolicy"}
- DestinationRule = config.GroupVersionKind{Group: "networking.istio.io", Version: "v1", Kind: "DestinationRule"}
- VirtualService = config.GroupVersionKind{Group: "networking.istio.io", Version: "v1", Kind: "VirtualService"}
+ RequestAuthentication = config.GroupVersionKind{Group: "security.dubbo.io", Version: "v1", Kind: "RequestAuthentication"}
+ PeerAuthentication = config.GroupVersionKind{Group: "security.dubbo.io", Version: "v1", Kind: "PeerAuthentication"}
+ AuthorizationPolicy = config.GroupVersionKind{Group: "security.dubbo.io", Version: "v1", Kind: "AuthorizationPolicy"}
+ DestinationRule = config.GroupVersionKind{Group: "networking.dubbo.io", Version: "v1", Kind: "DestinationRule"}
+ VirtualService = config.GroupVersionKind{Group: "networking.dubbo.io", Version: "v1", Kind: "VirtualService"}
)
func ToGVR(g config.GroupVersionKind) (schema.GroupVersionResource, bool) {
@@ -97,10 +97,10 @@ func FromGVR(g schema.GroupVersionResource) (config.GroupVersionKind, bool) {
switch g {
case gvr.CustomResourceDefinition:
return CustomResourceDefinition, true
- // case gvr.MutatingWebhookConfiguration:
- // return MutatingWebhookConfiguration, true
- // case gvr.ValidatingWebhookConfiguration:
- // return ValidatingWebhookConfiguration, true
+ case gvr.MutatingWebhookConfiguration:
+ return MutatingWebhookConfiguration, true
+ case gvr.ValidatingWebhookConfiguration:
+ return ValidatingWebhookConfiguration, true
case gvr.Namespace:
return Namespace, true
case gvr.Deployment:
diff --git a/pkg/config/schema/gvr/resources.go b/pkg/config/schema/gvr/resources.go
index dd55ca6d..35f8c75e 100644
--- a/pkg/config/schema/gvr/resources.go
+++ b/pkg/config/schema/gvr/resources.go
@@ -35,11 +35,11 @@ var (
Service = schema.GroupVersionResource{Group: "", Version: "v1", Resource: "services"}
ServiceAccount = schema.GroupVersionResource{Group: "", Version: "v1", Resource: "serviceaccounts"}
MeshConfig = schema.GroupVersionResource{Group: "", Version: "v1alpha1", Resource: "meshconfigs"}
- RequestAuthentication = schema.GroupVersionResource{Group: "security.istio.io", Version: "v1", Resource: "requestauthentications"}
- PeerAuthentication = schema.GroupVersionResource{Group: "security.istio.io", Version: "v1", Resource: "peerauthentications"}
- AuthorizationPolicy = schema.GroupVersionResource{Group: "security.istio.io", Version: "v1", Resource: "authorizationpolicies"}
- DestinationRule = schema.GroupVersionResource{Group: "networking.istio.io", Version: "v1", Resource: "destinationrules"}
- VirtualService = schema.GroupVersionResource{Group: "networking.istio.io", Version: "v1", Resource: "virtualservices"}
+ RequestAuthentication = schema.GroupVersionResource{Group: "security.dubbo.io", Version: "v1", Resource: "requestauthentications"}
+ PeerAuthentication = schema.GroupVersionResource{Group: "security.dubbo.io", Version: "v1", Resource: "peerauthentications"}
+ AuthorizationPolicy = schema.GroupVersionResource{Group: "security.dubbo.io", Version: "v1", Resource: "authorizationpolicies"}
+ DestinationRule = schema.GroupVersionResource{Group: "networking.dubbo.io", Version: "v1", Resource: "destinationrules"}
+ VirtualService = schema.GroupVersionResource{Group: "networking.dubbo.io", Version: "v1", Resource: "virtualservices"}
)
func IsClusterScoped(g schema.GroupVersionResource) bool {
diff --git a/pkg/config/schema/kind/resources.go b/pkg/config/schema/kind/resources.go
index 09d6606f..80ce5182 100644
--- a/pkg/config/schema/kind/resources.go
+++ b/pkg/config/schema/kind/resources.go
@@ -11,6 +11,7 @@ const (
Secret
Service
ServiceAccount
+ ServiceEntry
StatefulSet
ValidatingWebhookConfiguration
MutatingWebhookConfiguration
@@ -40,6 +41,8 @@ func (k Kind) String() string {
return "Service"
case ServiceAccount:
return "ServiceAccount"
+ case ServiceEntry:
+ return "ServiceEntry"
case StatefulSet:
return "StatefulSet"
case ValidatingWebhookConfiguration:
diff --git a/pkg/kube/inject/webhook.go b/pkg/kube/inject/webhook.go
index 134074b9..d8717c17 100644
--- a/pkg/kube/inject/webhook.go
+++ b/pkg/kube/inject/webhook.go
@@ -290,6 +290,7 @@ func postProcessPod(pod *corev1.Pod, injectedPod corev1.Pod, req InjectionParame
if pod.Labels == nil {
pod.Labels = map[string]string{}
}
+
return nil
}
@@ -348,13 +349,14 @@ func (wh *Webhook) inject(ar *kube.AdmissionReview, path string) *kube.Admission
pod.ManagedFields = nil
- potentialPodName(pod.ObjectMeta)
+ podName := potentialPodName(pod.ObjectMeta)
if pod.ObjectMeta.Namespace == "" {
pod.ObjectMeta.Namespace = req.Namespace
}
-
+ klog.Infof("Namespace: %v podName: %s", pod.Namespace+"/"+podName)
klog.Infof("Process proxyless injection request")
+ wh.mu.RLock()
proxyConfig := wh.env.GetProxyConfigOrDefault(pod.Namespace, pod.Labels, pod.Annotations, wh.meshConfig)
deploy, typeMeta := kube.GetDeployMetaFromPod(&pod)
params := InjectionParameters{
@@ -372,10 +374,14 @@ func (wh *Webhook) inject(ar *kube.AdmissionReview, path string) *kube.Admission
revision: wh.revision,
}
+ wh.mu.RUnlock()
+
patchBytes, err := injectPod(params)
if err != nil {
+ klog.Errorf("Pod injection failed: %v", err)
return toAdmissionResponse(err)
}
+
reviewResponse := kube.AdmissionResponse{
Allowed: true,
Patch: patchBytes,
diff --git a/pkg/webhooks/server/server.go b/pkg/webhooks/server/server.go
index 9612ec72..d75287f0 100644
--- a/pkg/webhooks/server/server.go
+++ b/pkg/webhooks/server/server.go
@@ -7,7 +7,6 @@ import (
"fmt"
"github.com/apache/dubbo-kubernetes/pkg/config/constants"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/collection"
- "github.com/apache/dubbo-kubernetes/pkg/config/schema/resource"
"github.com/apache/dubbo-kubernetes/pkg/config/validation"
"github.com/apache/dubbo-kubernetes/pkg/kube"
"github.com/apache/dubbo-kubernetes/sail/pkg/config/kube/crd"
@@ -113,7 +112,7 @@ func (wh *Webhook) validate(request *kube.AdmissionRequest) *kube.AdmissionRespo
// "Version" is not relevant for Istio types; each version has the same schema. So do a lookup that does not consider
// version. This ensures if a new version comes out and Istiod is not updated, we won't reject it.
- s, exists := wh.schemas.FindByGroupKind(resource.FromKubernetesGVK(&gvk))
+ s, exists := wh.schemas.FindByGroupKind(collection.FromKubernetesGVK(&gvk))
if !exists {
klog.Infof("unrecognized type %v", addDryRunMessageIfNeeded(obj.GroupVersionKind().String()))
return toAdmissionResponse(fmt.Errorf("unrecognized type %v", obj.GroupVersionKind()))
diff --git a/pkg/webhooks/validation/controller/controller.go b/pkg/webhooks/validation/controller/controller.go
index 4745ffcf..169f2582 100644
--- a/pkg/webhooks/validation/controller/controller.go
+++ b/pkg/webhooks/validation/controller/controller.go
@@ -73,8 +73,7 @@ func failurePolicyIsIgnore(current *kubeApiAdmission.ValidatingWebhookConfigurat
func (c *Controller) readyForFailClose() bool {
if !c.dryRunOfInvalidConfigRejected {
- klog.Info("Endpoint successfully rejected invalid config. Switching to fail-close.")
- c.dryRunOfInvalidConfigRejected = true
+
// Sync all webhooks; this ensures if we have multiple webhooks all of them are updated
c.syncAll()
}
diff --git a/pkg/webhooks/webhookpatch.go b/pkg/webhooks/webhookpatch.go
index 9a7d5522..f8b507ac 100644
--- a/pkg/webhooks/webhookpatch.go
+++ b/pkg/webhooks/webhookpatch.go
@@ -87,6 +87,8 @@ func (w *WebhookCertPatcher) patchMutatingWebhookConfig(webhookConfigName string
if !ok {
return nil
}
+ klog.Infof("This is webhook label: %v", v)
+
if v != w.revision {
return errWrongRevision
}
diff --git a/sail/pkg/bootstrap/configcontroller.go b/sail/pkg/bootstrap/configcontroller.go
index a6d8e35d..85c1a049 100644
--- a/sail/pkg/bootstrap/configcontroller.go
+++ b/sail/pkg/bootstrap/configcontroller.go
@@ -134,7 +134,7 @@ func (s *Server) initConfigSources(args *SailArgs) (err error) {
Config: adsc.Config{
Namespace: args.Namespace,
Workload: args.PodName,
- Revision: "", // TODO
+ Revision: args.Revision,
Meta: nil,
GrpcOpts: []grpc.DialOption{
args.KeepaliveOptions.ConvertToClientOption(),
diff --git a/sail/pkg/bootstrap/server.go b/sail/pkg/bootstrap/server.go
index 3812cb02..d601fcd0 100644
--- a/sail/pkg/bootstrap/server.go
+++ b/sail/pkg/bootstrap/server.go
@@ -672,11 +672,13 @@ func (s *Server) initSDSServer() {
// Make sure we have security
klog.Warningf("skipping Kubernetes credential reader; SAIL_ENABLE_XDS_IDENTITY_CHECK must be set to true for this feature.")
} else {
- // s.XDSServer.ConfigUpdate(&model.PushRequest{
- // Full: false,
- // ConfigsUpdated: sets.New(model.ConfigKey{Kind: k, Name: name, Namespace: namespace}),
- // Reason: model.NewReasonStats(model.SecretTrigger),
- // })
+ // TODO ConfigUpdated Multicluster get secret and configmap
+ s.XDSServer.ConfigUpdate(&model.PushRequest{
+ Full: false,
+ ConfigsUpdated: nil,
+ Reason: model.NewReasonStats(model.SecretTrigger),
+ })
+
}
}
@@ -759,7 +761,7 @@ func (s *Server) cachesSynced() bool {
func (s *Server) pushContextReady(expected int64) bool {
committed := s.XDSServer.CommittedUpdates.Load()
if committed < expected {
- klog.Infof("Waiting for pushcontext to process inbound updates, inbound: %v, committed : %v", expected, committed)
+ klog.V(2).Infof("Waiting for pushcontext to process inbound updates, inbound: %v, committed : %v", expected, committed)
return false
}
return true
diff --git a/sail/pkg/config/aggregate/config.go b/sail/pkg/config/aggregate/config.go
index 8a8478ac..c5f56081 100644
--- a/sail/pkg/config/aggregate/config.go
+++ b/sail/pkg/config/aggregate/config.go
@@ -3,12 +3,12 @@ package aggregate
import (
"errors"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/collection"
"github.com/apache/dubbo-kubernetes/pkg/slices"
"k8s.io/apimachinery/pkg/types"
"github.com/apache/dubbo-kubernetes/pkg/config"
- "github.com/apache/dubbo-kubernetes/pkg/config/schema/collection"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
"github.com/apache/dubbo-kubernetes/sail/pkg/model"
)
diff --git a/sail/pkg/config/kube/crd/conversion.go b/sail/pkg/config/kube/crd/conversion.go
index 8cf499e1..90c53664 100644
--- a/sail/pkg/config/kube/crd/conversion.go
+++ b/sail/pkg/config/kube/crd/conversion.go
@@ -5,14 +5,14 @@ import (
"encoding/json"
"fmt"
"github.com/apache/dubbo-kubernetes/pkg/config"
- "github.com/apache/dubbo-kubernetes/pkg/config/schema/resource"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/collection"
"io"
kubeyaml "k8s.io/apimachinery/pkg/util/yaml"
"k8s.io/klog/v2"
"reflect"
)
-type ConversionFunc = func(s resource.Schema, js string) (config.Spec, error)
+type ConversionFunc = func(s collection.Schema, js string) (config.Spec, error)
func parseInputsImpl(inputs string, withValidate bool) ([]config.Config, []DubboKind, error) {
var varr []config.Config
@@ -45,7 +45,7 @@ func ParseInputs(inputs string) ([]config.Config, []DubboKind, error) {
return parseInputsImpl(inputs, true)
}
-func FromJSON(s resource.Schema, js string) (config.Spec, error) {
+func FromJSON(s collection.Schema, js string) (config.Spec, error) {
c, err := s.NewInstance()
if err != nil {
return nil, err
@@ -56,11 +56,11 @@ func FromJSON(s resource.Schema, js string) (config.Spec, error) {
return c, nil
}
-func ConvertObject(schema resource.Schema, object DubboObject, domain string) (*config.Config, error) {
+func ConvertObject(schema collection.Schema, object DubboObject, domain string) (*config.Config, error) {
return ConvertObjectInternal(schema, object, domain, FromJSON)
}
-func StatusJSONFromMap(schema resource.Schema, jsonMap *json.RawMessage) (config.Status, error) {
+func StatusJSONFromMap(schema collection.Schema, jsonMap *json.RawMessage) (config.Status, error) {
if jsonMap == nil {
return nil, nil
}
@@ -79,7 +79,7 @@ func StatusJSONFromMap(schema resource.Schema, jsonMap *json.RawMessage) (config
return status, nil
}
-func ConvertObjectInternal(schema resource.Schema, object DubboObject, domain string, convert ConversionFunc) (*config.Config, error) {
+func ConvertObjectInternal(schema collection.Schema, object DubboObject, domain string, convert ConversionFunc) (*config.Config, error) {
js, err := json.Marshal(object.GetSpec())
if err != nil {
return nil, err
diff --git a/sail/pkg/config/kube/crdclient/client.go b/sail/pkg/config/kube/crdclient/client.go
index da1ebc19..37537237 100644
--- a/sail/pkg/config/kube/crdclient/client.go
+++ b/sail/pkg/config/kube/crdclient/client.go
@@ -5,7 +5,6 @@ import (
"fmt"
"github.com/apache/dubbo-kubernetes/pkg/config"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/collection"
- "github.com/apache/dubbo-kubernetes/pkg/config/schema/resource"
"github.com/apache/dubbo-kubernetes/pkg/kube"
"github.com/apache/dubbo-kubernetes/pkg/kube/controllers"
"github.com/apache/dubbo-kubernetes/pkg/kube/kclient"
@@ -30,7 +29,7 @@ type Client struct {
kinds map[config.GroupVersionKind]nsStore
kindsMu sync.RWMutex
domainSuffix string
- schemasByCRDName map[string]resource.Schema
+ schemasByCRDName map[string]collection.Schema
schemas collection.Schemas
client kube.Client
filtersByGVK map[config.GroupVersionKind]kubetypes.Filter
@@ -52,7 +51,7 @@ type Option struct {
}
func NewForSchemas(client kube.Client, opts Option, schemas collection.Schemas) *Client {
- schemasByCRDName := map[string]resource.Schema{}
+ schemasByCRDName := map[string]collection.Schema{}
for _, s := range schemas.All() {
// From the spec: "Its name MUST be in the format <.spec.name>.<.spec.group>."
name := fmt.Sprintf("%s.%s", s.Plural(), s.Group())
diff --git a/sail/pkg/config/kube/file/controller.go b/sail/pkg/config/kube/file/controller.go
index 0a1ec6c9..37836882 100644
--- a/sail/pkg/config/kube/file/controller.go
+++ b/sail/pkg/config/kube/file/controller.go
@@ -2,11 +2,11 @@ package file
import (
"fmt"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/collection"
"github.com/apache/dubbo-kubernetes/sail/pkg/config/kube/crd"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/apache/dubbo-kubernetes/pkg/config"
- "github.com/apache/dubbo-kubernetes/pkg/config/schema/collection"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/collections"
"github.com/apache/dubbo-kubernetes/pkg/kube/controllers"
"github.com/apache/dubbo-kubernetes/pkg/kube/krt"
diff --git a/sail/pkg/credentials/kube/secrets.go b/sail/pkg/credentials/kube/secrets.go
index 191d2484..fd2ee9dd 100644
--- a/sail/pkg/credentials/kube/secrets.go
+++ b/sail/pkg/credentials/kube/secrets.go
@@ -8,25 +8,11 @@ import (
)
const (
- // The ID/name for the certificate chain in kubernetes generic secret.
- GenericScrtCert = "cert"
- // The ID/name for the private key in kubernetes generic secret.
- GenericScrtKey = "key"
- // The ID/name for the CA certificate in kubernetes generic secret.
GenericScrtCaCert = "cacert"
- // The ID/name for the CRL in kubernetes generic secret.
- GenericScrtCRL = "crl"
+ GenericScrtCRL = "crl"
- // The ID/name for the certificate chain in kubernetes tls secret.
- TLSSecretCert = "tls.crt"
- // The ID/name for the k8sKey in kubernetes tls secret.
- TLSSecretKey = "tls.key"
- // The ID/name for the certificate OCSP staple in kubernetes tls secret
- TLSSecretOcspStaple = "tls.ocsp-staple"
- // The ID/name for the CA certificate in kubernetes tls secret
TLSSecretCaCert = "ca.crt"
- // The ID/name for the CRL in kubernetes tls secret.
- TLSSecretCrl = "ca.crl"
+ TLSSecretCrl = "ca.crl"
)
func hasKeys(d map[string][]byte, keys ...string) bool {
diff --git a/sail/pkg/model/push_context.go b/sail/pkg/model/push_context.go
index 8efbe55e..a6fde7c4 100644
--- a/sail/pkg/model/push_context.go
+++ b/sail/pkg/model/push_context.go
@@ -102,21 +102,7 @@ type PushRequest struct {
type ResourceDelta = xds.ResourceDelta
func NewPushContext() *PushContext {
- return &PushContext{
- // ServiceIndex: serviceIndex{},
- // ProxyStatus: map[string]map[string]ProxyPushStatus{},
- // serviceAccounts: map[serviceAccountKey][]string{},
- }
-}
-
-func newServiceIndex() serviceIndex {
- return serviceIndex{
- public: []*Service{},
- privateByNamespace: map[string][]*Service{},
- exportedToNamespace: map[string][]*Service{},
- HostnameAndNamespace: map[host.Name]map[string]*Service{},
- instancesByPort: map[string]map[int][]*DubboEndpoint{},
- }
+ return &PushContext{}
}
type ConfigKey struct {
@@ -252,10 +238,6 @@ func (pr *PushRequest) Merge(other *PushRequest) *PushRequest {
return pr
}
-func (r ReasonStats) Has(reason TriggerReason) bool {
- return r[reason] > 0
-}
-
func (pr *PushRequest) IsRequest() bool {
return len(pr.Reason) == 1 && pr.Reason.Has(ProxyRequest)
}
@@ -279,32 +261,6 @@ func (ps *PushContext) UpdateMetrics() {
defer ps.proxyStatusMutex.RUnlock()
}
-func NewReasonStats(reasons ...TriggerReason) ReasonStats {
- ret := make(ReasonStats)
- for _, reason := range reasons {
- ret.Add(reason)
- }
- return ret
-}
-
-func (r ReasonStats) Add(reason TriggerReason) {
- r[reason]++
-}
-
-func (r ReasonStats) Merge(other ReasonStats) {
- for reason, count := range other {
- r[reason] += count
- }
-}
-
-func (r ReasonStats) Count() int {
- var ret int
- for _, count := range r {
- ret += count
- }
- return ret
-}
-
func (ps *PushContext) GetAllServices() []*Service {
return ps.servicesExportedToNamespace(NamespaceAll)
}
@@ -330,3 +286,33 @@ func (ps *PushContext) servicesExportedToNamespace(ns string) []*Service {
return out
}
+
+func NewReasonStats(reasons ...TriggerReason) ReasonStats {
+ ret := make(ReasonStats)
+ for _, reason := range reasons {
+ ret.Add(reason)
+ }
+ return ret
+}
+
+func (r ReasonStats) Has(reason TriggerReason) bool {
+ return r[reason] > 0
+}
+
+func (r ReasonStats) Add(reason TriggerReason) {
+ r[reason]++
+}
+
+func (r ReasonStats) Merge(other ReasonStats) {
+ for reason, count := range other {
+ r[reason] += count
+ }
+}
+
+func (r ReasonStats) Count() int {
+ var ret int
+ for _, count := range r {
+ ret += count
+ }
+ return ret
+}
diff --git a/sail/pkg/xds/discovery.go b/sail/pkg/xds/discovery.go
index 822bf218..6ca0ea4a 100644
--- a/sail/pkg/xds/discovery.go
+++ b/sail/pkg/xds/discovery.go
@@ -111,7 +111,7 @@ func (s *DiscoveryServer) Shutdown() {
func (s *DiscoveryServer) Push(req *model.PushRequest) {
if !req.Full {
req.Push = s.globalPushContext()
- // s.AdsPushAll(req)
+ s.AdsPushAll(req)
return
}
@@ -125,7 +125,7 @@ func (s *DiscoveryServer) Push(req *model.PushRequest) {
versionLocal := s.NextVersion()
push := s.initPushContext(req, oldPushContext, versionLocal)
req.Push = push
- // s.AdsPushAll(req)
+ s.AdsPushAll(req)
}
func (s *DiscoveryServer) initPushContext(req *model.PushRequest, oldPushContext *model.PushContext, version string) *model.PushContext {
@@ -157,8 +157,6 @@ func (s *DiscoveryServer) ConfigUpdate(req *model.PushRequest) {
}
s.InboundUpdates.Inc()
- klog.Infof("this is req: %v", req)
-
s.pushChannel <- req
}
@@ -222,14 +220,12 @@ func debounce(ch chan *model.PushRequest, stopCh <-chan struct{}, opts DebounceO
freeCh := make(chan struct{}, 1)
push := func(req *model.PushRequest, debouncedEvents int, startDebounce time.Time) {
- klog.Info("This is push func")
pushFn(req)
updateSent.Add(int64(debouncedEvents))
freeCh <- struct{}{}
}
pushWorker := func() {
- klog.Info("This is pushWorker func")
eventDelay := time.Since(startDebounce)
quietTime := time.Since(lastConfigUpdateTime)
// it has been too long or quiet enough
diff --git a/sail/pkg/xds/eds.go b/sail/pkg/xds/eds.go
index 21fe09b8..811dacd7 100644
--- a/sail/pkg/xds/eds.go
+++ b/sail/pkg/xds/eds.go
@@ -1,9 +1,12 @@
package xds
import (
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/kind"
+ "github.com/apache/dubbo-kubernetes/pkg/util/sets"
"github.com/apache/dubbo-kubernetes/sail/pkg/model"
)
+// TODO EDS
func (s *DiscoveryServer) EDSUpdate(shard model.ShardKey, serviceName string, namespace string,
dubboEndpoints []*model.DubboEndpoint,
) {
@@ -13,7 +16,7 @@ func (s *DiscoveryServer) EDSUpdate(shard model.ShardKey, serviceName string, na
// Trigger a push
s.ConfigUpdate(&model.PushRequest{
Full: pushType == model.FullPush,
- ConfigsUpdated: nil,
+ ConfigsUpdated: sets.New(model.ConfigKey{Kind: kind.ServiceEntry, Name: serviceName, Namespace: namespace}),
Reason: model.NewReasonStats(model.EndpointUpdate),
})
}