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/08/06 03:15:53 UTC
(dubbo-kubernetes) branch master updated: [navi] add discovery init mesh config v4 (#771)
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 3835a353 [navi] add discovery init mesh config v4 (#771)
3835a353 is described below
commit 3835a353479da31f79cdddf199c679e2f760927e
Author: Jian Zhong <11...@qq.com>
AuthorDate: Wed Aug 6 11:15:48 2025 +0800
[navi] add discovery init mesh config v4 (#771)
---
go.mod | 2 +-
go.sum | 4 +-
navigator/cmd/navi-discovery/app/cmd.go | 2 +
navigator/pkg/bootstrap/mesh.go | 4 +-
navigator/pkg/bootstrap/options.go | 2 +
navigator/pkg/bootstrap/server.go | 33 +-
pkg/config/constants/constants.go | 2 +
pkg/config/labels/instance.go | 164 ++++++
pkg/config/mesh/kubemesh/watcher.go | 2 +-
pkg/config/mesh/meshwatcher/mesh.go | 29 +
pkg/config/mesh/watchers.go | 43 ++
pkg/config/schema/kubeclient/common.go | 26 -
pkg/config/schema/kubeclient/resource.gen.go | 178 ++++++
pkg/config/schema/kubetypes/resources.gen.go | 3 +
pkg/filewatcher/fakefilewatcher.go | 146 -----
pkg/kube/client.go | 60 +-
pkg/kube/client_config.go | 49 +-
pkg/kube/client_factory.go | 88 ++-
pkg/kube/informerfactory/factory.go | 175 +++---
pkg/kube/kclient/client.go | 10 +-
pkg/kube/krt/collection.go | 847 +++++++++++++++------------
pkg/kube/krt/core.go | 25 +-
pkg/kube/krt/debug.go | 14 -
pkg/kube/krt/fetch.go | 53 +-
pkg/kube/krt/filter.go | 23 +-
pkg/kube/krt/helpers.go | 7 +-
pkg/kube/krt/internal.go | 33 +-
pkg/kube/krt/options.go | 14 +-
pkg/kube/krt/singleton.go | 4 +
pkg/mesh/mesh.go | 43 +-
pkg/queue/instance.go | 170 ++++++
pkg/queue/util.go | 39 ++
32 files changed, 1524 insertions(+), 770 deletions(-)
diff --git a/go.mod b/go.mod
index f31a05ad..facc807e 100644
--- a/go.mod
+++ b/go.mod
@@ -56,7 +56,7 @@ require (
helm.sh/helm/v3 v3.18.4
istio.io/api v1.26.3
istio.io/client-go v1.26.0-alpha.0.0.20250801111706-8b4229d430b7
- istio.io/istio v0.0.0-20250803052519-90ec2958505f
+ istio.io/istio v0.0.0-20250804181122-c5fdc78b0741
k8s.io/api v0.33.2
k8s.io/apiextensions-apiserver v0.33.2
k8s.io/apimachinery v0.33.2
diff --git a/go.sum b/go.sum
index aa58533e..5ac9549e 100644
--- a/go.sum
+++ b/go.sum
@@ -906,8 +906,8 @@ istio.io/api v1.26.3 h1:/TiA7bJi24yBQSgpLy5vHhFkobf4DWS1L+CuUxNk4os=
istio.io/api v1.26.3/go.mod h1:DTVGH6CLXj5W8FF9JUD3Tis78iRgT1WeuAnxfTz21Wg=
istio.io/client-go v1.26.0-alpha.0.0.20250801111706-8b4229d430b7 h1:UfQlmnTALAQJkGHzaQVBLqeWYLOFU2vhRQhEFBAXjdg=
istio.io/client-go v1.26.0-alpha.0.0.20250801111706-8b4229d430b7/go.mod h1:M7elDrU4/g2Rgr/K/u5E4VQP6l4lSMwBdNj5NeT21mg=
-istio.io/istio v0.0.0-20250803052519-90ec2958505f h1:Yx4yf5/ikWf/X/tt6rBKZ6NcWpEROYW5JyDAr1MuAQQ=
-istio.io/istio v0.0.0-20250803052519-90ec2958505f/go.mod h1:TY8IgDEUQKj+8eRtT8zxs8Mfsqxv1d/ecErZhfzF73U=
+istio.io/istio v0.0.0-20250804181122-c5fdc78b0741 h1:dxfcWWEWYto5gpKo8rVyLh42iJHGaqRfmv4n91IxK6U=
+istio.io/istio v0.0.0-20250804181122-c5fdc78b0741/go.mod h1:TY8IgDEUQKj+8eRtT8zxs8Mfsqxv1d/ecErZhfzF73U=
k8s.io/api v0.33.2 h1:YgwIS5jKfA+BZg//OQhkJNIfie/kmRsO0BmNaVSimvY=
k8s.io/api v0.33.2/go.mod h1:fhrbphQJSM2cXzCWgqU29xLDuks4mu7ti9vveEnpSXs=
k8s.io/apiextensions-apiserver v0.33.2 h1:6gnkIbngnaUflR3XwE1mCefN3YS8yTD631JXQhsU6M8=
diff --git a/navigator/cmd/navi-discovery/app/cmd.go b/navigator/cmd/navi-discovery/app/cmd.go
index e7adfb02..cb5f17e8 100644
--- a/navigator/cmd/navi-discovery/app/cmd.go
+++ b/navigator/cmd/navi-discovery/app/cmd.go
@@ -113,6 +113,8 @@ func addFlags(c *cobra.Command) {
"Directory to watch for updates to config yaml files. If specified, the files will be used as the source of config, rather than a CRD client.")
c.PersistentFlags().StringVar(&serverArgs.RegistryOptions.KubeOptions.DomainSuffix, "domain", constants.DefaultClusterLocalDomain,
"DNS domain suffix")
+ c.PersistentFlags().StringVarP(&serverArgs.Namespace, "namespace", "n", bootstrap.PodNamespace,
+ "Select a namespace where the controller resides. If not set, uses ${POD_NAMESPACE} environment variable")
c.PersistentFlags().StringVar((*string)(&serverArgs.RegistryOptions.KubeOptions.ClusterID), "clusterID", features.ClusterName,
"The ID of the cluster that this Dubbod instance resides")
c.PersistentFlags().StringToStringVar(&serverArgs.RegistryOptions.KubeOptions.ClusterAliases, "clusterAliases", map[string]string{},
diff --git a/navigator/pkg/bootstrap/mesh.go b/navigator/pkg/bootstrap/mesh.go
index 2a517d28..f3655124 100644
--- a/navigator/pkg/bootstrap/mesh.go
+++ b/navigator/pkg/bootstrap/mesh.go
@@ -39,7 +39,7 @@ func (s *Server) initMeshConfiguration(args *NaviArgs, fileWatcher filewatcher.F
}
func (s *Server) getMeshConfiguration(args *NaviArgs, fileWatcher filewatcher.FileWatcher) krt.Singleton[meshwatcher.MeshConfigResource] {
- opts := krt.NewOptionsBuilder(s.internalStop, "")
+ opts := krt.NewOptionsBuilder(s.internalStop, "", args.KrtDebugger)
sources := s.getConfigurationSources(args, fileWatcher, args.MeshConfigFile, kubemesh.MeshConfigKey)
if len(sources) == 0 {
fmt.Printf("\nUsing default mesh - missing file %s and no k8s client\n", args.MeshConfigFile)
@@ -48,7 +48,7 @@ func (s *Server) getMeshConfiguration(args *NaviArgs, fileWatcher filewatcher.Fi
}
func (s *Server) getConfigurationSources(args *NaviArgs, fileWatcher filewatcher.FileWatcher, file string, cmKey string) []meshwatcher.MeshConfigSource {
- opts := krt.NewOptionsBuilder(s.internalStop, "")
+ opts := krt.NewOptionsBuilder(s.internalStop, "", args.KrtDebugger)
var userMeshConfig *meshwatcher.MeshConfigSource
if features.SharedMeshConfig != "" && s.kubeClient != nil {
userMeshConfig = ptr.Of(kubemesh.NewConfigMapSource(s.kubeClient, args.Namespace, features.SharedMeshConfig, cmKey, opts))
diff --git a/navigator/pkg/bootstrap/options.go b/navigator/pkg/bootstrap/options.go
index f5b0a30a..fd3d21af 100644
--- a/navigator/pkg/bootstrap/options.go
+++ b/navigator/pkg/bootstrap/options.go
@@ -23,6 +23,7 @@ import (
"github.com/apache/dubbo-kubernetes/pkg/ctrlz"
"github.com/apache/dubbo-kubernetes/pkg/env"
"github.com/apache/dubbo-kubernetes/pkg/keepalive"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/krt"
)
var (
@@ -46,6 +47,7 @@ type NaviArgs struct {
Namespace string
CtrlZOptions *ctrlz.Options
KeepaliveOptions *keepalive.Options
+ KrtDebugger *krt.DebugHandler `json:"-"`
}
type DiscoveryServerOptions struct {
diff --git a/navigator/pkg/bootstrap/server.go b/navigator/pkg/bootstrap/server.go
index c99efe9f..ed91ad34 100644
--- a/navigator/pkg/bootstrap/server.go
+++ b/navigator/pkg/bootstrap/server.go
@@ -28,12 +28,11 @@ import (
"github.com/apache/dubbo-kubernetes/pkg/filewatcher"
"github.com/apache/dubbo-kubernetes/pkg/h2c"
dubbokeepalive "github.com/apache/dubbo-kubernetes/pkg/keepalive"
+ kubelib "github.com/apache/dubbo-kubernetes/pkg/kube"
"github.com/apache/dubbo-kubernetes/pkg/mesh"
"github.com/apache/dubbo-kubernetes/pkg/network"
"golang.org/x/net/http2"
"google.golang.org/grpc"
- cluster2 "istio.io/istio/pkg/cluster"
- kubelib "istio.io/istio/pkg/kube"
"k8s.io/client-go/rest"
"net"
"net/http"
@@ -93,10 +92,13 @@ func NewServer(args *NaviArgs, initFuncs ...func(*Server)) (*Server, error) {
}
func (s *Server) Start(stop <-chan struct{}) error {
+ fmt.Printf("\nStarting Dubbod Server with primary cluster %s\n", s.clusterID)
if err := s.server.Start(stop); err != nil {
return err
}
- // TODO waitForCacheSync
+ if !s.waitForCacheSync(stop) {
+ return fmt.Errorf("failed to sync cache")
+ }
// TODO XDSserver CacheSynced
if s.secureGrpcAddress != "" {
@@ -143,6 +145,7 @@ func (s *Server) Start(stop <-chan struct{}) error {
func (s *Server) initKubeClient(args *NaviArgs) error {
if s.kubeClient != nil {
+ // Already initialized by startup arguments
return nil
}
hasK8SConfigStore := false
@@ -168,16 +171,16 @@ func (s *Server) initKubeClient(args *NaviArgs) error {
}
if hasK8SConfigStore || hasKubeRegistry(args.RegistryOptions.Registries) {
+ // Used by validation
kubeRestConfig, err := kubelib.DefaultRestConfig(args.RegistryOptions.KubeConfig, "", func(config *rest.Config) {
config.QPS = args.RegistryOptions.KubeOptions.KubernetesAPIQPS
config.Burst = args.RegistryOptions.KubeOptions.KubernetesAPIBurst
})
-
if err != nil {
return fmt.Errorf("failed creating kube config: %v", err)
}
- s.kubeClient, err = kubelib.NewClient(kubelib.NewClientConfigForRestConfig(kubeRestConfig), cluster2.ID(s.clusterID))
+ s.kubeClient, err = kubelib.NewClient(kubelib.NewClientConfigForRestConfig(kubeRestConfig), s.clusterID)
if err != nil {
return fmt.Errorf("failed creating kube client: %v", err)
}
@@ -248,3 +251,23 @@ func getClusterID(args *NaviArgs) cluster.ID {
}
return clusterID
}
+
+func (s *Server) cachesSynced() bool {
+ // TODO multiclusterController HasSynced
+ // TODO ServiceController().HasSynced
+ // TODO configController.HasSynced
+ return true
+}
+
+func (s *Server) waitForCacheSync(stop <-chan struct{}) bool {
+ start := time.Now()
+ fmt.Println("\nWaiting for caches to be synced")
+ if !kubelib.WaitForCacheSync("server", stop, s.cachesSynced) {
+ fmt.Println("\nFailed waiting for cache sync")
+ return false
+ }
+ fmt.Printf("\nAll controller caches have been synced up in %v\n", time.Since(start))
+ // TODO XDSServer.InboundUpdates.Load
+ // TODO return kubelib.WaitForCacheSync("push context", stop, func() bool { return s.pushContextReady(expected) })
+ return false
+}
diff --git a/pkg/config/constants/constants.go b/pkg/config/constants/constants.go
index 9375dd66..8a787e5a 100644
--- a/pkg/config/constants/constants.go
+++ b/pkg/config/constants/constants.go
@@ -22,4 +22,6 @@ const (
DefaultClusterLocalDomain = "cluster.local"
DefaultClusterName = "Kubernetes"
ServiceClusterName = "dubbo-proxy"
+ ConfigPathDir = "./etc/dubbo/proxy"
+ BinaryPathFilename = "/usr/local/bin/envoy"
)
diff --git a/pkg/config/labels/instance.go b/pkg/config/labels/instance.go
new file mode 100644
index 00000000..094ab3d6
--- /dev/null
+++ b/pkg/config/labels/instance.go
@@ -0,0 +1,164 @@
+// Copyright Istio Authors
+//
+// Licensed 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 labels
+
+import (
+ "fmt"
+ "regexp"
+ "strings"
+
+ "github.com/hashicorp/go-multierror"
+
+ "istio.io/istio/pkg/maps"
+ "istio.io/istio/pkg/slices"
+)
+
+const (
+ DNS1123LabelMaxLength = 63 // Public for testing only.
+ dns1123LabelFmt = "[a-zA-Z0-9](?:[-a-zA-Z0-9]*[a-zA-Z0-9])?"
+ // a wild-card prefix is an '*', a normal DNS1123 label with a leading '*' or '*-', or a normal DNS1123 label
+ wildcardPrefix = `(\*|(\*|\*-)?` + dns1123LabelFmt + `)`
+
+ // Using kubernetes requirement, a valid key must be a non-empty string consist
+ // of alphanumeric characters, '-', '_' or '.', and must start and end with an
+ // alphanumeric character (e.g. 'MyValue', or 'my_value', or '12345'
+ qualifiedNameFmt = "(?:[A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9]"
+
+ // In Kubernetes, label names can start with a DNS name followed by a '/':
+ dnsNamePrefixFmt = dns1123LabelFmt + `(?:\.` + dns1123LabelFmt + `)*/`
+ dnsNamePrefixMaxLength = 253
+)
+
+var (
+ tagRegexp = regexp.MustCompile("^(" + dnsNamePrefixFmt + ")?(" + qualifiedNameFmt + ")$") // label value can be an empty string
+ labelValueRegexp = regexp.MustCompile("^" + "(" + qualifiedNameFmt + ")?" + "$")
+ dns1123LabelRegexp = regexp.MustCompile("^" + dns1123LabelFmt + "$")
+ wildcardPrefixRegexp = regexp.MustCompile("^" + wildcardPrefix + "$")
+)
+
+// Instance is a non empty map of arbitrary strings. Each version of a service can
+// be differentiated by a unique set of labels associated with the version. These
+// labels are assigned to all instances of a particular service version. For
+// example, lets say catalog.mystore.com has 2 versions v1 and v2. v1 instances
+// could have labels gitCommit=aeiou234, region=us-east, while v2 instances could
+// have labels name=kittyCat,region=us-east.
+type Instance map[string]string
+
+// SubsetOf is true if the label has same values for the keys
+func (i Instance) SubsetOf(that Instance) bool {
+ if len(i) == 0 {
+ return true
+ }
+
+ if len(that) == 0 || len(that) < len(i) {
+ return false
+ }
+
+ for k, v1 := range i {
+ if v2, ok := that[k]; !ok || v1 != v2 {
+ return false
+ }
+ }
+ return true
+}
+
+// Match is true if the label has same values for the keys.
+// if len(i) == 0, will return false. It is mainly used for service -> workload
+func (i Instance) Match(that Instance) bool {
+ if len(i) == 0 {
+ return false
+ }
+
+ return i.SubsetOf(that)
+}
+
+// Equals returns true if the labels are equal.
+func (i Instance) Equals(that Instance) bool {
+ return maps.Equal(i, that)
+}
+
+// Validate ensures tag is well-formed
+func (i Instance) Validate() error {
+ if i == nil {
+ return nil
+ }
+ var errs error
+ for k, v := range i {
+ if err := validateTagKey(k); err != nil {
+ errs = multierror.Append(errs, err)
+ }
+ if !labelValueRegexp.MatchString(v) {
+ errs = multierror.Append(errs, fmt.Errorf("invalid tag value: %q", v))
+ }
+ }
+ return errs
+}
+
+// IsDNS1123Label tests for a string that conforms to the definition of a label in
+// DNS (RFC 1123).
+func IsDNS1123Label(value string) bool {
+ return len(value) <= DNS1123LabelMaxLength && dns1123LabelRegexp.MatchString(value)
+}
+
+// IsWildcardDNS1123Label tests for a string that conforms to the definition of a label in DNS (RFC 1123), but allows
+// the wildcard label (`*`), and typical labels with a leading astrisk instead of alphabetic character (e.g. "*-foo")
+func IsWildcardDNS1123Label(value string) bool {
+ return len(value) <= DNS1123LabelMaxLength && wildcardPrefixRegexp.MatchString(value)
+}
+
+// validateTagKey checks that a string is valid as a Kubernetes label name.
+func validateTagKey(k string) error {
+ match := tagRegexp.FindStringSubmatch(k)
+ if match == nil {
+ return fmt.Errorf("invalid tag key: %q", k)
+ }
+
+ if len(match[1]) > 0 {
+ dnsPrefixLength := len(match[1]) - 1 // exclude the trailing / from the length
+ if dnsPrefixLength > dnsNamePrefixMaxLength {
+ return fmt.Errorf("invalid tag key: %q (DNS prefix is too long)", k)
+ }
+ }
+
+ if len(match[2]) > DNS1123LabelMaxLength {
+ return fmt.Errorf("invalid tag key: %q (name is too long)", k)
+ }
+
+ return nil
+}
+
+func (i Instance) String() string {
+ // Ensure stable ordering
+ keys := slices.Sort(maps.Keys(i))
+
+ var buffer strings.Builder
+ // Assume each kv pair is roughly 25 characters. We could be under or over, this is just a guess to optimize
+ buffer.Grow(len(keys) * 25)
+ first := true
+ for _, k := range keys {
+ v := i[k]
+ if !first {
+ buffer.WriteString(",")
+ } else {
+ first = false
+ }
+ if len(v) > 0 {
+ buffer.WriteString(k + "=" + v)
+ } else {
+ buffer.WriteString(k)
+ }
+ }
+ return buffer.String()
+}
diff --git a/pkg/config/mesh/kubemesh/watcher.go b/pkg/config/mesh/kubemesh/watcher.go
index b1b2f796..06797646 100644
--- a/pkg/config/mesh/kubemesh/watcher.go
+++ b/pkg/config/mesh/kubemesh/watcher.go
@@ -19,11 +19,11 @@ package kubemesh
import (
"fmt"
+ "github.com/apache/dubbo-kubernetes/pkg/kube"
"github.com/apache/dubbo-kubernetes/pkg/kube/kclient"
"github.com/apache/dubbo-kubernetes/pkg/kube/krt"
"github.com/apache/dubbo-kubernetes/pkg/mesh/meshwatcher"
"github.com/apache/dubbo-kubernetes/pkg/ptr"
- "istio.io/istio/pkg/kube"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/fields"
diff --git a/pkg/config/mesh/meshwatcher/mesh.go b/pkg/config/mesh/meshwatcher/mesh.go
new file mode 100644
index 00000000..e510bc3d
--- /dev/null
+++ b/pkg/config/mesh/meshwatcher/mesh.go
@@ -0,0 +1,29 @@
+package meshwatcher
+
+import (
+ "google.golang.org/protobuf/proto"
+
+ meshconfig "istio.io/api/mesh/v1alpha1"
+)
+
+// MeshConfigResource holds the current MeshConfig state
+type MeshConfigResource struct {
+ *meshconfig.MeshConfig
+}
+
+func (m MeshConfigResource) ResourceName() string { return "MeshConfigResource" }
+
+func (m MeshConfigResource) Equals(other MeshConfigResource) bool {
+ return proto.Equal(m.MeshConfig, other.MeshConfig)
+}
+
+// MeshNetworksResource holds the current MeshNetworks state
+type MeshNetworksResource struct {
+ *meshconfig.MeshNetworks
+}
+
+func (m MeshNetworksResource) ResourceName() string { return "MeshNetworksResource" }
+
+func (m MeshNetworksResource) Equals(other MeshNetworksResource) bool {
+ return proto.Equal(m.MeshNetworks, other.MeshNetworks)
+}
diff --git a/pkg/config/mesh/watchers.go b/pkg/config/mesh/watchers.go
new file mode 100644
index 00000000..03a3a0e6
--- /dev/null
+++ b/pkg/config/mesh/watchers.go
@@ -0,0 +1,43 @@
+package mesh
+
+import (
+ "istio.io/api/mesh/v1alpha1"
+)
+
+// NetworksWatcher watches changes to the mesh networks config.
+type NetworksWatcher interface {
+ // AddNetworksHandler registers a callback handler for changes to the networks config.
+ AddNetworksHandler(func()) *WatcherHandlerRegistration
+
+ // DeleteNetworksHandler unregisters a callback handler when remote cluster is removed.
+ DeleteNetworksHandler(registration *WatcherHandlerRegistration)
+}
+
+// Holder of a mesh configuration.
+type Holder interface {
+ Mesh() *v1alpha1.MeshConfig
+}
+
+// Watcher is a Holder whose mesh config can be updated asynchronously.
+type Watcher interface {
+ Holder
+
+ // AddMeshHandler registers a callback handler for changes to the mesh config.
+ AddMeshHandler(h func()) *WatcherHandlerRegistration
+
+ // DeleteMeshHandler unregisters a callback handler when remote cluster is removed.
+ DeleteMeshHandler(registration *WatcherHandlerRegistration)
+}
+
+// WatcherHandlerRegistration will be returned to caller to remove the handler later.
+type WatcherHandlerRegistration struct {
+ remove func()
+}
+
+func NewWatcherHandlerRegistration(f func()) *WatcherHandlerRegistration {
+ return &WatcherHandlerRegistration{remove: f}
+}
+
+func (r *WatcherHandlerRegistration) Remove() {
+ r.remove()
+}
diff --git a/pkg/config/schema/kubeclient/common.go b/pkg/config/schema/kubeclient/common.go
index 42310ee4..02141def 100644
--- a/pkg/config/schema/kubeclient/common.go
+++ b/pkg/config/schema/kubeclient/common.go
@@ -151,29 +151,3 @@ func stripUnusedFields(obj any) (any, error) {
t.GetObjectMeta().SetManagedFields(nil)
return obj, nil
}
-
-func getInformerFiltered(c ClientGetter, opts ktypes.InformerOptions, g schema.GroupVersionResource) informerfactory.StartableInformer {
- var l func(options metav1.ListOptions) (runtime.Object, error)
- var w func(options metav1.ListOptions) (watch.Interface, error)
- return c.Informers().InformerFor(g, opts, func() cache.SharedIndexInformer {
- inf := cache.NewSharedIndexInformer(
- &cache.ListWatch{
- ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
- options.FieldSelector = opts.FieldSelector
- options.LabelSelector = opts.LabelSelector
- return l(options)
- },
- WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
- options.FieldSelector = opts.FieldSelector
- options.LabelSelector = opts.LabelSelector
- return w(options)
- },
- },
- gvrToObject(g),
- 0,
- cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
- )
- setupInformer(opts, inf)
- return inf
- })
-}
diff --git a/pkg/config/schema/kubeclient/resource.gen.go b/pkg/config/schema/kubeclient/resource.gen.go
index 56761fe3..f350fc65 100644
--- a/pkg/config/schema/kubeclient/resource.gen.go
+++ b/pkg/config/schema/kubeclient/resource.gen.go
@@ -1,19 +1,69 @@
package kubeclient
import (
+ "context"
"fmt"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/gvr"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/informerfactory"
ktypes "github.com/apache/dubbo-kubernetes/pkg/kube/kubetypes"
"github.com/apache/dubbo-kubernetes/pkg/ptr"
+ k8sioapiadmissionregistrationv1 "k8s.io/api/admissionregistration/v1"
+ k8sioapiappsv1 "k8s.io/api/apps/v1"
+ k8sioapicertificatesv1 "k8s.io/api/certificates/v1"
+ k8sioapicertificatesv1beta1 "k8s.io/api/certificates/v1beta1"
+ k8sioapicoordinationv1 "k8s.io/api/coordination/v1"
k8sioapicorev1 "k8s.io/api/core/v1"
+ k8sioapidiscoveryv1 "k8s.io/api/discovery/v1"
+ k8sioapinetworkingv1 "k8s.io/api/networking/v1"
+ k8sioapipolicyv1 "k8s.io/api/policy/v1"
+ k8sioapiextensionsapiserverpkgapisapiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
+ "k8s.io/apimachinery/pkg/watch"
+ "k8s.io/client-go/tools/cache"
)
func GetWriteClient[T runtime.Object](c ClientGetter, namespace string) ktypes.WriteAPI[T] {
switch any(ptr.Empty[T]()).(type) {
+ case *k8sioapicertificatesv1.CertificateSigningRequest:
+ return c.Kube().CertificatesV1().CertificateSigningRequests().(ktypes.WriteAPI[T])
+ case *k8sioapicertificatesv1beta1.ClusterTrustBundle:
+ return c.Kube().CertificatesV1beta1().ClusterTrustBundles().(ktypes.WriteAPI[T])
case *k8sioapicorev1.ConfigMap:
return c.Kube().CoreV1().ConfigMaps(namespace).(ktypes.WriteAPI[T])
+ case *k8sioapiextensionsapiserverpkgapisapiextensionsv1.CustomResourceDefinition:
+ return c.Ext().ApiextensionsV1().CustomResourceDefinitions().(ktypes.WriteAPI[T])
+ case *k8sioapiappsv1.DaemonSet:
+ return c.Kube().AppsV1().DaemonSets(namespace).(ktypes.WriteAPI[T])
+ case *k8sioapiappsv1.Deployment:
+ return c.Kube().AppsV1().Deployments(namespace).(ktypes.WriteAPI[T])
+ case *k8sioapidiscoveryv1.EndpointSlice:
+ return c.Kube().DiscoveryV1().EndpointSlices(namespace).(ktypes.WriteAPI[T])
+ case *k8sioapicorev1.Endpoints:
+ return c.Kube().CoreV1().Endpoints(namespace).(ktypes.WriteAPI[T])
+ case *k8sioapinetworkingv1.Ingress:
+ return c.Kube().NetworkingV1().Ingresses(namespace).(ktypes.WriteAPI[T])
+ case *k8sioapinetworkingv1.IngressClass:
+ return c.Kube().NetworkingV1().IngressClasses().(ktypes.WriteAPI[T])
+ case *k8sioapicoordinationv1.Lease:
+ return c.Kube().CoordinationV1().Leases(namespace).(ktypes.WriteAPI[T])
+ case *k8sioapiadmissionregistrationv1.MutatingWebhookConfiguration:
+ return c.Kube().AdmissionregistrationV1().MutatingWebhookConfigurations().(ktypes.WriteAPI[T])
+ case *k8sioapicorev1.Namespace:
+ return c.Kube().CoreV1().Namespaces().(ktypes.WriteAPI[T])
+ case *k8sioapicorev1.Node:
+ return c.Kube().CoreV1().Nodes().(ktypes.WriteAPI[T])
+ case *k8sioapicorev1.Pod:
+ return c.Kube().CoreV1().Pods(namespace).(ktypes.WriteAPI[T])
+ case *k8sioapipolicyv1.PodDisruptionBudget:
+ return c.Kube().PolicyV1().PodDisruptionBudgets(namespace).(ktypes.WriteAPI[T])
+ case *k8sioapicorev1.Secret:
+ return c.Kube().CoreV1().Secrets(namespace).(ktypes.WriteAPI[T])
+ case *k8sioapicorev1.Service:
+ return c.Kube().CoreV1().Services(namespace).(ktypes.WriteAPI[T])
+ case *k8sioapicorev1.ServiceAccount:
+ return c.Kube().CoreV1().ServiceAccounts(namespace).(ktypes.WriteAPI[T])
default:
panic(fmt.Sprintf("Unknown type %T", ptr.Empty[T]()))
}
@@ -23,7 +73,135 @@ func gvrToObject(g schema.GroupVersionResource) runtime.Object {
switch g {
case gvr.ConfigMap:
return &k8sioapicorev1.ConfigMap{}
+ case gvr.CustomResourceDefinition:
+ return &k8sioapiextensionsapiserverpkgapisapiextensionsv1.CustomResourceDefinition{}
+ case gvr.DaemonSet:
+ return &k8sioapiappsv1.DaemonSet{}
+ case gvr.Deployment:
+ return &k8sioapiappsv1.Deployment{}
+ case gvr.MutatingWebhookConfiguration:
+ return &k8sioapiadmissionregistrationv1.MutatingWebhookConfiguration{}
+ case gvr.Namespace:
+ return &k8sioapicorev1.Namespace{}
+ case gvr.Secret:
+ return &k8sioapicorev1.Secret{}
+ case gvr.Service:
+ return &k8sioapicorev1.Service{}
+ case gvr.ServiceAccount:
+ return &k8sioapicorev1.ServiceAccount{}
+ case gvr.StatefulSet:
+ return &k8sioapiappsv1.StatefulSet{}
+ case gvr.ValidatingWebhookConfiguration:
+ return &k8sioapiadmissionregistrationv1.ValidatingWebhookConfiguration{}
default:
panic(fmt.Sprintf("Unknown type %v", g))
}
}
+
+func getInformerFiltered(c ClientGetter, opts ktypes.InformerOptions, g schema.GroupVersionResource) informerfactory.StartableInformer {
+ var l func(options metav1.ListOptions) (runtime.Object, error)
+ var w func(options metav1.ListOptions) (watch.Interface, error)
+
+ switch g {
+ case gvr.ConfigMap:
+ l = func(options metav1.ListOptions) (runtime.Object, error) {
+ return c.Kube().CoreV1().ConfigMaps(opts.Namespace).List(context.Background(), options)
+ }
+ w = func(options metav1.ListOptions) (watch.Interface, error) {
+ return c.Kube().CoreV1().ConfigMaps(opts.Namespace).Watch(context.Background(), options)
+ }
+ case gvr.CustomResourceDefinition:
+ l = func(options metav1.ListOptions) (runtime.Object, error) {
+ return c.Ext().ApiextensionsV1().CustomResourceDefinitions().List(context.Background(), options)
+ }
+ w = func(options metav1.ListOptions) (watch.Interface, error) {
+ return c.Ext().ApiextensionsV1().CustomResourceDefinitions().Watch(context.Background(), options)
+ }
+ case gvr.DaemonSet:
+ l = func(options metav1.ListOptions) (runtime.Object, error) {
+ return c.Kube().AppsV1().DaemonSets(opts.Namespace).List(context.Background(), options)
+ }
+ w = func(options metav1.ListOptions) (watch.Interface, error) {
+ return c.Kube().AppsV1().DaemonSets(opts.Namespace).Watch(context.Background(), options)
+ }
+ case gvr.Deployment:
+ l = func(options metav1.ListOptions) (runtime.Object, error) {
+ return c.Kube().AppsV1().Deployments(opts.Namespace).List(context.Background(), options)
+ }
+ w = func(options metav1.ListOptions) (watch.Interface, error) {
+ return c.Kube().AppsV1().Deployments(opts.Namespace).Watch(context.Background(), options)
+ }
+ case gvr.MutatingWebhookConfiguration:
+ l = func(options metav1.ListOptions) (runtime.Object, error) {
+ return c.Kube().AdmissionregistrationV1().MutatingWebhookConfigurations().List(context.Background(), options)
+ }
+ w = func(options metav1.ListOptions) (watch.Interface, error) {
+ return c.Kube().AdmissionregistrationV1().MutatingWebhookConfigurations().Watch(context.Background(), options)
+ }
+ case gvr.Namespace:
+ l = func(options metav1.ListOptions) (runtime.Object, error) {
+ return c.Kube().CoreV1().Namespaces().List(context.Background(), options)
+ }
+ w = func(options metav1.ListOptions) (watch.Interface, error) {
+ return c.Kube().CoreV1().Namespaces().Watch(context.Background(), options)
+ }
+ case gvr.Secret:
+ l = func(options metav1.ListOptions) (runtime.Object, error) {
+ return c.Kube().CoreV1().Secrets(opts.Namespace).List(context.Background(), options)
+ }
+ w = func(options metav1.ListOptions) (watch.Interface, error) {
+ return c.Kube().CoreV1().Secrets(opts.Namespace).Watch(context.Background(), options)
+ }
+ case gvr.Service:
+ l = func(options metav1.ListOptions) (runtime.Object, error) {
+ return c.Kube().CoreV1().Services(opts.Namespace).List(context.Background(), options)
+ }
+ w = func(options metav1.ListOptions) (watch.Interface, error) {
+ return c.Kube().CoreV1().Services(opts.Namespace).Watch(context.Background(), options)
+ }
+ case gvr.ServiceAccount:
+ l = func(options metav1.ListOptions) (runtime.Object, error) {
+ return c.Kube().CoreV1().ServiceAccounts(opts.Namespace).List(context.Background(), options)
+ }
+ w = func(options metav1.ListOptions) (watch.Interface, error) {
+ return c.Kube().CoreV1().ServiceAccounts(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)
+ }
+ w = func(options metav1.ListOptions) (watch.Interface, error) {
+ return c.Kube().AppsV1().StatefulSets(opts.Namespace).Watch(context.Background(), options)
+ }
+ case gvr.ValidatingWebhookConfiguration:
+ l = func(options metav1.ListOptions) (runtime.Object, error) {
+ return c.Kube().AdmissionregistrationV1().ValidatingWebhookConfigurations().List(context.Background(), options)
+ }
+ w = func(options metav1.ListOptions) (watch.Interface, error) {
+ return c.Kube().AdmissionregistrationV1().ValidatingWebhookConfigurations().Watch(context.Background(), options)
+ }
+ default:
+ panic(fmt.Sprintf("Unknown type %v", g))
+ }
+ return c.Informers().InformerFor(g, opts, func() cache.SharedIndexInformer {
+ inf := cache.NewSharedIndexInformer(
+ &cache.ListWatch{
+ ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
+ options.FieldSelector = opts.FieldSelector
+ options.LabelSelector = opts.LabelSelector
+ return l(options)
+ },
+ WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
+ options.FieldSelector = opts.FieldSelector
+ options.LabelSelector = opts.LabelSelector
+ return w(options)
+ },
+ },
+ gvrToObject(g),
+ 0,
+ cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
+ )
+ setupInformer(opts, inf)
+ return inf
+ })
+}
diff --git a/pkg/config/schema/kubetypes/resources.gen.go b/pkg/config/schema/kubetypes/resources.gen.go
index b8627ba2..09a73e0e 100644
--- a/pkg/config/schema/kubetypes/resources.gen.go
+++ b/pkg/config/schema/kubetypes/resources.gen.go
@@ -3,6 +3,7 @@ package kubetypes
import (
"github.com/apache/dubbo-kubernetes/operator/pkg/config"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/gvk"
+ istioioapimeshv1alpha1 "istio.io/api/mesh/v1alpha1"
k8sioapicorev1 "k8s.io/api/core/v1"
)
@@ -10,6 +11,8 @@ func getGvk(obj any) (config.GroupVersionKind, bool) {
switch obj.(type) {
case *k8sioapicorev1.ConfigMap:
return gvk.ConfigMap, true
+ case *istioioapimeshv1alpha1.MeshConfig:
+ return gvk.MeshConfig, true
default:
return config.GroupVersionKind{}, false
}
diff --git a/pkg/filewatcher/fakefilewatcher.go b/pkg/filewatcher/fakefilewatcher.go
deleted file mode 100644
index 731c6c67..00000000
--- a/pkg/filewatcher/fakefilewatcher.go
+++ /dev/null
@@ -1,146 +0,0 @@
-/*
- * 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 filewatcher
-
-import (
- "errors"
- "fmt"
- "sync"
-
- "github.com/fsnotify/fsnotify"
-)
-
-// NewFileWatcherFunc returns a function which creates a new file
-// watcher. This may be used to provide test hooks for using the
-// FakeWatcher implementation below.
-type NewFileWatcherFunc func() FileWatcher
-
-// FakeWatcher provides a fake file watcher implementation for unit
-// tests. Production code should use the `NewWatcher()`.
-type FakeWatcher struct {
- sync.Mutex
-
- events map[string]chan fsnotify.Event
- errors map[string]chan error
- changedFunc func(path string, added bool)
-}
-
-// InjectEvent injects an event into the fake file watcher.
-func (w *FakeWatcher) InjectEvent(path string, event fsnotify.Event) {
- w.Lock()
- ch, ok := w.events[path]
- w.Unlock()
-
- if ok {
- ch <- event
- }
-}
-
-// InjectError injects an error into the fake file watcher.
-func (w *FakeWatcher) InjectError(path string, err error) {
- w.Lock()
- ch, ok := w.errors[path]
- w.Unlock()
-
- if ok {
- ch <- err
- }
-}
-
-// NewFakeWatcher returns a function which creates a new fake watcher for unit
-// testing. This allows observe callers to inject events and errors per-watched
-// path. changedFunc() provides a callback notification when a new watch is added
-// or removed. Production code should use `NewWatcher()`.
-func NewFakeWatcher(changedFunc func(path string, added bool)) (NewFileWatcherFunc, *FakeWatcher) {
- w := &FakeWatcher{
- events: make(map[string]chan fsnotify.Event),
- errors: make(map[string]chan error),
- changedFunc: changedFunc,
- }
- return func() FileWatcher {
- return w
- }, w
-}
-
-// Add is a fake implementation of the FileWatcher interface.
-func (w *FakeWatcher) Add(path string) error {
- w.Lock()
-
- // w.events and w.errors are always updated togeather. We only check
- // the first to determine existence.
- if _, ok := w.events[path]; ok {
- w.Unlock()
- return fmt.Errorf("path %v already exists", path)
- }
-
- w.events[path] = make(chan fsnotify.Event, 1000)
- w.errors[path] = make(chan error, 1000)
-
- w.Unlock()
-
- if w.changedFunc != nil {
- w.changedFunc(path, true)
- }
- return nil
-}
-
-// Remove is a fake implementation of the FileWatcher interface.
-func (w *FakeWatcher) Remove(path string) error {
- w.Lock()
- defer w.Unlock()
-
- if _, ok := w.events[path]; !ok {
- return errors.New("path doesn't exist")
- }
-
- delete(w.events, path)
- delete(w.errors, path)
- if w.changedFunc != nil {
- w.changedFunc(path, false)
- }
- return nil
-}
-
-// Close is a fake implementation of the FileWatcher interface.
-func (w *FakeWatcher) Close() error {
- w.Lock()
- for path, ch := range w.events {
- close(ch)
- delete(w.events, path)
- }
- for path, ch := range w.errors {
- close(ch)
- delete(w.errors, path)
- }
- defer w.Unlock()
- return nil
-}
-
-// Events is a fake implementation of the FileWatcher interface.
-func (w *FakeWatcher) Events(path string) chan fsnotify.Event {
- w.Lock()
- defer w.Unlock()
- return w.events[path]
-}
-
-// Errors is a fake implementation of the FileWatcher interface.
-func (w *FakeWatcher) Errors(path string) chan error {
- w.Lock()
- defer w.Unlock()
- return w.errors[path]
-}
diff --git a/pkg/kube/client.go b/pkg/kube/client.go
index 269da7c1..a4bf722e 100644
--- a/pkg/kube/client.go
+++ b/pkg/kube/client.go
@@ -25,18 +25,17 @@ import (
"github.com/apache/dubbo-kubernetes/pkg/kube/informerfactory"
"github.com/apache/dubbo-kubernetes/pkg/lazy"
"github.com/apache/dubbo-kubernetes/pkg/sleep"
- cclient "istio.io/client-go/pkg/clientset/versioned"
+ istioclient "istio.io/client-go/pkg/clientset/versioned"
kubeExtClient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
"k8s.io/apimachinery/pkg/api/meta"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime/schema"
kubeVersion "k8s.io/apimachinery/pkg/version"
- "k8s.io/client-go/metadata"
- "k8s.io/client-go/tools/cache"
-
"k8s.io/client-go/dynamic"
"k8s.io/client-go/kubernetes"
+ "k8s.io/client-go/metadata"
"k8s.io/client-go/rest"
+ "k8s.io/client-go/tools/cache"
"k8s.io/client-go/tools/clientcmd"
"net/http"
"time"
@@ -46,7 +45,7 @@ type client struct {
extSet kubeExtClient.Interface
config *rest.Config
revision string
- factory *clientFactory
+ clientFactory *clientFactory
version lazy.Lazy[*kubeVersion.Info]
informerFactory informerfactory.InformerFactory
dynamic dynamic.Interface
@@ -54,7 +53,7 @@ type client struct {
mapper meta.ResettableRESTMapper
metadata metadata.Interface
http *http.Client
- dubbo cclient.Interface
+ dubbo istioclient.Interface
clusterID cluster.ID
}
@@ -76,64 +75,68 @@ type Client interface {
type CLIClient interface {
Client
- UtilFactory() PartialFactory
DynamicClientFor(gvk schema.GroupVersionKind, obj *unstructured.Unstructured, namespace string) (dynamic.ResourceInterface, error)
}
type ClientOption func(cliClient CLIClient) CLIClient
func NewCLIClient(clientCfg clientcmd.ClientConfig, opts ...ClientOption) (CLIClient, error) {
- return newInternalClient(newClientFactory(clientCfg, false), opts...)
+ return newClientInternal(newClientFactory(clientCfg, false), opts...)
}
-func NewClient(clientConfig clientcmd.ClientConfig, cluster cluster.ID) (Client, error) {
- return newInternalClient(newClientFactory(clientConfig, false), WithCluster(cluster))
+func NewClient(clientCfg clientcmd.ClientConfig, cluster cluster.ID) (Client, error) {
+ return newClientInternal(newClientFactory(clientCfg, false), WithCluster(cluster))
}
-func newInternalClient(factory *clientFactory, opts ...ClientOption) (CLIClient, error) {
+func newClientInternal(clientFactory *clientFactory, opts ...ClientOption) (*client, error) {
var c client
var err error
- c.factory = factory
- c.config, err = factory.ToRestConfig()
+
+ c.clientFactory = clientFactory
+
+ c.config, err = clientFactory.ToRESTConfig()
if err != nil {
return nil, err
}
+
for _, opt := range opts {
opt(&c)
}
- c.mapper, err = factory.mapper.Get()
+
+ c.mapper, err = clientFactory.mapper.Get()
if err != nil {
return nil, err
}
+
+ c.informerFactory = informerfactory.NewSharedInformerFactory()
+
c.kube, err = kubernetes.NewForConfig(c.config)
if err != nil {
return nil, err
}
- c.dynamic, err = dynamic.NewForConfig(c.config)
- if err != nil {
- return nil, err
- }
+
c.metadata, err = metadata.NewForConfig(c.config)
if err != nil {
return nil, err
}
- c.extSet, err = kubeExtClient.NewForConfig(c.config)
+
+ c.dynamic, err = dynamic.NewForConfig(c.config)
if err != nil {
return nil, err
}
- c.informerFactory = informerfactory.NewSharedInformerFactory()
- c.kube, err = kubernetes.NewForConfig(c.config)
+
+ c.extSet, err = kubeExtClient.NewForConfig(c.config)
if err != nil {
return nil, err
}
- c.http = &http.Client{
- Timeout: time.Second * 15,
- }
+
+ c.http = &http.Client{}
if c.config != nil && c.config.Timeout != 0 {
c.http.Timeout = c.config.Timeout
} else {
c.http.Timeout = time.Second * 15
}
+
var clientWithTimeout kubernetes.Interface
clientWithTimeout = c.kube
restConfig := c.RESTConfig()
@@ -141,6 +144,7 @@ func newInternalClient(factory *clientFactory, opts ...ClientOption) (CLIClient,
if restConfig.Timeout == 0 {
restConfig.Timeout = time.Second * 5
}
+
kubeClient, err := kubernetes.NewForConfig(restConfig)
if err == nil {
clientWithTimeout = kubeClient
@@ -224,10 +228,6 @@ func (c *client) bestEffortToGVR(gvk schema.GroupVersionKind, obj *unstructured.
return gvr, namespaced
}
-func (c *client) UtilFactory() PartialFactory {
- return c.factory
-}
-
func WaitForCacheSync(name string, stop <-chan struct{}, cacheSyncs ...cache.InformerSynced) (r bool) {
t0 := time.Now()
maximum := time.Millisecond * 100
@@ -243,9 +243,9 @@ func WaitForCacheSync(name string, stop <-chan struct{}, cacheSyncs ...cache.Inf
attempt := 0
defer func() {
if r {
- fmt.Printf("sync complete: name=%s, time=%v\n", name, time.Since(t0))
+ fmt.Printf("\nsync complete: name=%s, time=%v", name, time.Since(t0))
} else {
- fmt.Printf("sync failed: name=%s, time=%v\n", name, time.Since(t0))
+ fmt.Printf("\nsync failed: name=%s, time=%v", name, time.Since(t0))
}
}()
for {
diff --git a/pkg/kube/client_config.go b/pkg/kube/client_config.go
index cd5eb575..31e7ae74 100644
--- a/pkg/kube/client_config.go
+++ b/pkg/kube/client_config.go
@@ -24,23 +24,27 @@ import (
)
const (
- contextName = "context0"
- clusterName = "cluster0"
+ contextName = "context0"
+ clusterName = "cluster0"
+ authInfoName = "authInfo0"
)
var _ clientcmd.ClientConfig = &clientConfig{}
+// clientConfig is a utility that allows construction of a k8s ClientConfig from
+// a k8s rest.Config
type clientConfig struct {
restConfig rest.Config
}
+// NewClientConfigForRestConfig creates a new k8s clientcmd.ClientConfig from the given rest.Config.
func NewClientConfigForRestConfig(restConfig *rest.Config) clientcmd.ClientConfig {
return &clientConfig{
restConfig: *restConfig,
}
}
-func (c clientConfig) RawConfig() (api.Config, error) {
+func (c *clientConfig) RawConfig() (api.Config, error) {
cfg := api.Config{
Kind: "Config",
APIVersion: "v1",
@@ -48,31 +52,54 @@ func (c clientConfig) RawConfig() (api.Config, error) {
Clusters: map[string]*api.Cluster{
clusterName: newCluster(&c.restConfig),
},
+ AuthInfos: map[string]*api.AuthInfo{
+ authInfoName: newAuthInfo(&c.restConfig),
+ },
Contexts: map[string]*api.Context{
contextName: {
- Cluster: clusterName,
+ Cluster: clusterName,
+ AuthInfo: authInfoName,
},
},
CurrentContext: contextName,
}
+
return cfg, nil
}
-func (c clientConfig) ClientConfig() (*rest.Config, error) {
+func (c *clientConfig) ClientConfig() (*rest.Config, error) {
return c.copyRestConfig(), nil
}
+func (c *clientConfig) Namespace() (string, bool, error) {
+ return "default", false, nil
+}
+
+func (c *clientConfig) ConfigAccess() clientcmd.ConfigAccess {
+ return nil
+}
+
func (c *clientConfig) copyRestConfig() *rest.Config {
out := c.restConfig
return &out
}
-func (c clientConfig) Namespace() (string, bool, error) {
- return "default", false, nil
-}
-
-func (c clientConfig) ConfigAccess() clientcmd.ConfigAccess {
- return nil
+func newAuthInfo(restConfig *rest.Config) *api.AuthInfo {
+ return &api.AuthInfo{
+ ClientCertificate: restConfig.CertFile,
+ ClientCertificateData: restConfig.CertData,
+ ClientKey: restConfig.KeyFile,
+ ClientKeyData: restConfig.KeyData,
+ Token: restConfig.BearerToken,
+ TokenFile: restConfig.BearerTokenFile,
+ Impersonate: restConfig.Impersonate.UserName,
+ ImpersonateGroups: restConfig.Impersonate.Groups,
+ ImpersonateUserExtra: restConfig.Impersonate.Extra,
+ Username: restConfig.Username,
+ Password: restConfig.Password,
+ AuthProvider: restConfig.AuthProvider,
+ Exec: restConfig.ExecProvider,
+ }
}
func newCluster(restConfig *rest.Config) *api.Cluster {
diff --git a/pkg/kube/client_factory.go b/pkg/kube/client_factory.go
index e2deb18d..d9f8ed30 100644
--- a/pkg/kube/client_factory.go
+++ b/pkg/kube/client_factory.go
@@ -37,31 +37,40 @@ import (
"time"
)
+var _ PartialFactory = &clientFactory{}
+
// clientFactory partially implements the kubectl util.Factory, which is provides access to various k8s clients.
// The full Factory can be built with MakeKubeFactory.
// This split is to avoid huge dependencies.
type clientFactory struct {
- clientConfig clientcmd.ClientConfig
- expander lazy.Lazy[meta.RESTMapper]
- mapper lazy.Lazy[meta.ResettableRESTMapper]
+ clientConfig clientcmd.ClientConfig
+
+ expander lazy.Lazy[meta.RESTMapper]
+ mapper lazy.Lazy[meta.ResettableRESTMapper]
+
discoveryClient lazy.Lazy[discovery.CachedDiscoveryInterface]
}
// newClientFactory creates a new util.Factory from the given clientcmd.ClientConfig.
func newClientFactory(clientConfig clientcmd.ClientConfig, diskCache bool) *clientFactory {
- cf := &clientFactory{
+ out := &clientFactory{
clientConfig: clientConfig,
}
- cf.discoveryClient = lazy.NewWithRetry(func() (discovery.CachedDiscoveryInterface, error) {
- restConfig, err := cf.ToRestConfig()
+
+ out.discoveryClient = lazy.NewWithRetry(func() (discovery.CachedDiscoveryInterface, error) {
+ restConfig, err := out.ToRESTConfig()
if err != nil {
return nil, err
}
// Setup cached discovery. CLIs uses disk cache, controllers use memory cache.
if diskCache {
+ // From https://github.com/kubernetes/cli-runtime/blob/4fdf49ae46a0caa7fafdfe97825c6129d5153f06/pkg/genericclioptions/config_flags.go#L288
+
cacheDir := filepath.Join(homedir.HomeDir(), ".kube", "cache")
+
httpCacheDir := filepath.Join(cacheDir, "http")
discoveryCacheDir := computeDiscoverCacheDir(filepath.Join(cacheDir, "discovery"), restConfig.Host)
+
return diskcached.NewCachedDiscoveryClientForConfig(restConfig, discoveryCacheDir, httpCacheDir, 6*time.Hour)
}
d, err := discovery.NewDiscoveryClientForConfig(restConfig)
@@ -70,73 +79,49 @@ func newClientFactory(clientConfig clientcmd.ClientConfig, diskCache bool) *clie
}
return memory.NewMemCacheClient(d), nil
})
- cf.mapper = lazy.NewWithRetry(func() (meta.ResettableRESTMapper, error) {
- discoveryClient, err := cf.ToDiscoveryClient()
+ out.mapper = lazy.NewWithRetry(func() (meta.ResettableRESTMapper, error) {
+ discoveryClient, err := out.ToDiscoveryClient()
if err != nil {
return nil, err
}
return restmapper.NewDeferredDiscoveryRESTMapper(discoveryClient), nil
})
- cf.expander = lazy.NewWithRetry(func() (meta.RESTMapper, error) {
- discoveryClient, err := cf.discoveryClient.Get()
+ out.expander = lazy.NewWithRetry(func() (meta.RESTMapper, error) {
+ discoveryClient, err := out.discoveryClient.Get()
if err != nil {
return nil, err
}
- mapper, err := cf.mapper.Get()
+ mapper, err := out.mapper.Get()
if err != nil {
return nil, err
}
return restmapper.NewShortcutExpander(mapper, discoveryClient, func(string) {}), nil
})
- return cf
+ return out
}
-func (c *clientFactory) RestClient() (*rest.RESTClient, error) {
- clientConfig, err := c.ToRestConfig()
+func (c *clientFactory) ToRESTConfig() (*rest.Config, error) {
+ restConfig, err := c.clientConfig.ClientConfig()
if err != nil {
return nil, err
}
- return rest.RESTClientFor(clientConfig)
+ return SetRestDefaults(restConfig), nil
}
func (c *clientFactory) ToDiscoveryClient() (discovery.CachedDiscoveryInterface, error) {
return c.discoveryClient.Get()
}
-func (c *clientFactory) ToRestConfig() (*rest.Config, error) {
- restConfig, err := c.clientConfig.ClientConfig()
- if err != nil {
- return nil, err
- }
- return SetRestDefaults(restConfig), nil
-}
-
// overlyCautiousIllegalFileCharacters matches characters that *might* not be supported. Windows is really restrictive, so this is really restrictive
var overlyCautiousIllegalFileCharacters = regexp.MustCompile(`[^(\w/.)]`)
-func computeDiscoverCacheDir(dir, host string) string {
- schemelesshost := strings.Replace(strings.Replace(host, "https://", "", 1), "http://", "", 1)
- safehost := overlyCautiousIllegalFileCharacters.ReplaceAllString(schemelesshost, "_")
- return filepath.Join(dir, safehost)
-}
-
-type rESTClientGetter interface {
- // ToRESTConfig returns restconfig
- ToRESTConfig() (*rest.Config, error)
- // ToDiscoveryClient returns discovery client
- ToDiscoveryClient() (discovery.CachedDiscoveryInterface, error)
- // ToRESTMapper returns a restmapper
- ToRESTMapper() (meta.RESTMapper, error)
- // ToRawKubeConfigLoader return kubeconfig loader as-is
- ToRawKubeConfigLoader() clientcmd.ClientConfig
-}
-
-func (c *clientFactory) ToRESTConfig() (*rest.Config, error) {
- restConfig, err := c.clientConfig.ClientConfig()
- if err != nil {
- return nil, err
- }
- return SetRestDefaults(restConfig), nil
+// computeDiscoverCacheDir takes the parentDir and the host and comes up with a "usually non-colliding" name.
+func computeDiscoverCacheDir(parentDir, host string) string {
+ // strip the optional scheme from host if its there:
+ schemelessHost := strings.Replace(strings.Replace(host, "https://", "", 1), "http://", "", 1)
+ // now do a simple collapse of non-AZ09 characters. Collisions are possible but unlikely. Even if we do collide the problem is short lived
+ safeHost := overlyCautiousIllegalFileCharacters.ReplaceAllString(schemelessHost, "_")
+ return filepath.Join(parentDir, safeHost)
}
func (c *clientFactory) ToRESTMapper() (meta.RESTMapper, error) {
@@ -172,6 +157,17 @@ func (c *clientFactory) RESTClient() (*rest.RESTClient, error) {
return rest.RESTClientFor(clientConfig)
}
+type rESTClientGetter interface {
+ // ToRESTConfig returns restconfig
+ ToRESTConfig() (*rest.Config, error)
+ // ToDiscoveryClient returns discovery client
+ ToDiscoveryClient() (discovery.CachedDiscoveryInterface, error)
+ // ToRESTMapper returns a restmapper
+ ToRESTMapper() (meta.RESTMapper, error)
+ // ToRawKubeConfigLoader return kubeconfig loader as-is
+ ToRawKubeConfigLoader() clientcmd.ClientConfig
+}
+
type PartialFactory interface {
rESTClientGetter
diff --git a/pkg/kube/informerfactory/factory.go b/pkg/kube/informerfactory/factory.go
index f9dd9ab5..34d3244b 100644
--- a/pkg/kube/informerfactory/factory.go
+++ b/pkg/kube/informerfactory/factory.go
@@ -18,96 +18,73 @@ type StartableInformer struct {
start func(stopCh <-chan struct{})
}
+func (s StartableInformer) Start(stopCh <-chan struct{}) {
+ s.start(stopCh)
+}
+
+// InformerFactory provides access to a shared informer factory
type InformerFactory interface {
+ // Start initializes all requested informers. They are handled in goroutines
+ // which run until the stop channel gets closed.
Start(stopCh <-chan struct{})
+
+ // InformerFor returns the SharedIndexInformer the provided type.
InformerFor(resource schema.GroupVersionResource, opts kubetypes.InformerOptions, newFunc NewInformerFunc) StartableInformer
+
+ // WaitForCacheSync blocks until all started informers' caches were synced
+ // or the stop channel gets closed.
WaitForCacheSync(stopCh <-chan struct{}) bool
+
+ // Shutdown marks a factory as shutting down. At that point no new
+ // informers can be started anymore and Start will return without
+ // doing anything.
+ //
+ // In addition, Shutdown blocks until all goroutines have terminated. For that
+ // to happen, the close channel(s) that they were started with must be closed,
+ // either before Shutdown gets called or while it is waiting.
+ //
+ // Shutdown may be called multiple times, even concurrently. All such calls will
+ // block until all goroutines have terminated.
Shutdown()
}
+// NewSharedInformerFactory constructs a new instance of informerFactory for all namespaces.
+func NewSharedInformerFactory() InformerFactory {
+ return &informerFactory{
+ informers: map[informerKey]builtInformer{},
+ startedInformers: sets.New[informerKey](),
+ }
+}
+
// InformerKey represents a unique informer
type informerKey struct {
gvr schema.GroupVersionResource
labelSelector string
fieldSelector string
+ informerType kubetypes.InformerType
namespace string
}
-type informerFactory struct {
- lock sync.Mutex
- informers map[informerKey]builtInformer
- startedInformers sets.Set[informerKey]
- wg sync.WaitGroup
- shuttingDown bool
-}
-
type builtInformer struct {
informer cache.SharedIndexInformer
objectTransform func(obj any) (any, error)
}
-func NewSharedInformerFactory() InformerFactory {
- return &informerFactory{
- informers: map[informerKey]builtInformer{},
- startedInformers: sets.New[informerKey](),
- }
-}
-
-func (s StartableInformer) Start(stopCh <-chan struct{}) {
- s.start(stopCh)
-}
-
-// Start initializes all requested informers.
-func (f *informerFactory) Start(stopCh <-chan struct{}) {
- f.lock.Lock()
- defer f.lock.Unlock()
-
- if f.shuttingDown {
- return
- }
-
- for informerType, informer := range f.informers {
- if !f.startedInformers.Contains(informerType) {
- informer := informer
- f.wg.Add(1)
- go func() {
- defer f.wg.Done()
- informer.informer.Run(stopCh)
- }()
- f.startedInformers.Insert(informerType)
- }
- }
-}
-
-func (f *informerFactory) WaitForCacheSync(stopCh <-chan struct{}) bool {
- informers := func() []cache.SharedIndexInformer {
- f.lock.Lock()
- defer f.lock.Unlock()
- informers := make([]cache.SharedIndexInformer, 0, len(f.informers))
- for informerKey, informer := range f.informers {
- if f.startedInformers.Contains(informerKey) {
- informers = append(informers, informer.informer)
- }
- }
- return informers
- }()
+type informerFactory struct {
+ lock sync.Mutex
+ informers map[informerKey]builtInformer
+ // startedInformers is used for tracking which informers have been started.
+ // This allows Start() to be called multiple times safely.
+ startedInformers sets.Set[informerKey]
- for _, informer := range informers {
- if !cache.WaitForCacheSync(stopCh, informer.HasSynced) {
- return false
- }
- }
- return true
+ // wg tracks how many goroutines were started.
+ wg sync.WaitGroup
+ // shuttingDown is true when Shutdown has been called. It may still be running
+ // because it needs to wait for goroutines.
+ shuttingDown bool
}
-func (f *informerFactory) Shutdown() {
- defer f.wg.Wait()
-
- f.lock.Lock()
- defer f.lock.Unlock()
-
- f.shuttingDown = true
-}
+var _ InformerFactory = &informerFactory{}
func (f *informerFactory) InformerFor(resource schema.GroupVersionResource, opts kubetypes.InformerOptions, newFunc NewInformerFunc) StartableInformer {
f.lock.Lock()
@@ -117,11 +94,11 @@ func (f *informerFactory) InformerFor(resource schema.GroupVersionResource, opts
gvr: resource,
labelSelector: opts.LabelSelector,
fieldSelector: opts.FieldSelector,
+ informerType: opts.InformerType,
namespace: opts.Namespace,
}
inf, exists := f.informers[key]
if exists {
- checkInformerOverlap(inf, resource, opts)
return f.makeStartableInformer(inf.informer, key)
}
@@ -134,12 +111,6 @@ func (f *informerFactory) InformerFor(resource schema.GroupVersionResource, opts
return f.makeStartableInformer(informer, key)
}
-func checkInformerOverlap(inf builtInformer, resource schema.GroupVersionResource, opts kubetypes.InformerOptions) {
- if fmt.Sprintf("%p", inf.objectTransform) == fmt.Sprintf("%p", opts.ObjectTransform) {
- return
- }
-}
-
func (f *informerFactory) makeStartableInformer(informer cache.SharedIndexInformer, key informerKey) StartableInformer {
return StartableInformer{
Informer: informer,
@@ -170,3 +141,59 @@ func (f *informerFactory) startOne(stopCh <-chan struct{}, informerType informer
f.startedInformers.Insert(informerType)
}
}
+
+// Start initializes all requested informers.
+func (f *informerFactory) Start(stopCh <-chan struct{}) {
+ f.lock.Lock()
+ defer f.lock.Unlock()
+
+ if f.shuttingDown {
+ return
+ }
+
+ for informerType, informer := range f.informers {
+ if !f.startedInformers.Contains(informerType) {
+ f.wg.Add(1)
+ // We need a new variable in each loop iteration,
+ // otherwise the goroutine would use the loop variable
+ // and that keeps changing.
+ informer := informer
+ go func() {
+ defer f.wg.Done()
+ informer.informer.Run(stopCh)
+ }()
+ f.startedInformers.Insert(informerType)
+ }
+ }
+}
+
+// WaitForCacheSync waits for all started informers' cache were synced.
+func (f *informerFactory) WaitForCacheSync(stopCh <-chan struct{}) bool {
+ informers := func() []cache.SharedIndexInformer {
+ f.lock.Lock()
+ defer f.lock.Unlock()
+ informers := make([]cache.SharedIndexInformer, 0, len(f.informers))
+ for informerKey, informer := range f.informers {
+ if f.startedInformers.Contains(informerKey) {
+ informers = append(informers, informer.informer)
+ }
+ }
+ return informers
+ }()
+
+ for _, informer := range informers {
+ if !cache.WaitForCacheSync(stopCh, informer.HasSynced) {
+ return false
+ }
+ }
+ return true
+}
+
+func (f *informerFactory) Shutdown() {
+ // Will return immediately if there is nothing to wait for.
+ defer f.wg.Wait()
+
+ f.lock.Lock()
+ defer f.lock.Unlock()
+ f.shuttingDown = true
+}
diff --git a/pkg/kube/kclient/client.go b/pkg/kube/kclient/client.go
index 60ea3240..8593f4cc 100644
--- a/pkg/kube/kclient/client.go
+++ b/pkg/kube/kclient/client.go
@@ -5,15 +5,15 @@ import (
"fmt"
"github.com/apache/dubbo-kubernetes/navigator/pkg/features"
dubbogvr "github.com/apache/dubbo-kubernetes/pkg/config/schema/gvr"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/kubeclient"
types "github.com/apache/dubbo-kubernetes/pkg/config/schema/kubetypes"
+ "github.com/apache/dubbo-kubernetes/pkg/kube"
"github.com/apache/dubbo-kubernetes/pkg/kube/controllers"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/informerfactory"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/kubetypes"
"github.com/apache/dubbo-kubernetes/pkg/ptr"
+ "github.com/apache/dubbo-kubernetes/pkg/util/sets"
"github.com/apache/dubbo-kubernetes/pkg/util/slices"
- "istio.io/istio/pkg/config/schema/kubeclient"
- "istio.io/istio/pkg/kube"
- "istio.io/istio/pkg/kube/informerfactory"
- "istio.io/istio/pkg/kube/kubetypes"
- "istio.io/istio/pkg/util/sets"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
klabels "k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/runtime/schema"
diff --git a/pkg/kube/krt/collection.go b/pkg/kube/krt/collection.go
index 3ceac44a..86267466 100644
--- a/pkg/kube/krt/collection.go
+++ b/pkg/kube/krt/collection.go
@@ -5,26 +5,33 @@ import (
"github.com/apache/dubbo-kubernetes/pkg/kube/controllers"
"github.com/apache/dubbo-kubernetes/pkg/maps"
"github.com/apache/dubbo-kubernetes/pkg/ptr"
+ "github.com/apache/dubbo-kubernetes/pkg/queue"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
"github.com/apache/dubbo-kubernetes/pkg/util/slices"
- "istio.io/istio/pkg/queue"
"sync"
)
type indexedDependencyType uint8
-type Key[O any] string
-
const (
unknownIndexType indexedDependencyType = iota
indexType indexedDependencyType = iota
getKeyType indexedDependencyType = iota
)
-type indexedDependency struct {
- id collectionUID
- key string
- typ indexedDependencyType
+// EnableAssertions, if true, will enable assertions. These typically are violations of the krt collection requirements.
+const EnableAssertions = false
+
+type dependencyState[I any] struct {
+ // collectionDependencies specifies the set of collections we depend on from within the transformation functions (via Fetch).
+ // These are keyed by the internal uid() function on collections.
+ // Note this does not include `parent`, which is the *primary* dependency declared outside of transformation functions.
+ collectionDependencies sets.Set[collectionUID]
+ // Stores a map of I -> secondary dependencies (added via Fetch)
+ objectDependencies map[Key[I]][]*dependency
+ indexedDependencies map[indexedDependency]sets.Set[Key[I]]
+ // indexedDependenciesExtractor stores a map of [collection,fetch type,(optional)index id] => an extractor to get change keys.
+ indexedDependenciesExtractor map[extractorKey]objectKeyExtractor
}
type extractorKey struct {
@@ -33,130 +40,206 @@ type extractorKey struct {
typ indexedDependencyType
}
-type objectKeyExtractor = func(o any) []string
-
-type multiIndex[I, O any] struct {
- outputs map[Key[O]]O
- inputs map[Key[I]]I
- mappings map[Key[I]]sets.Set[Key[O]]
-}
-
-type collectionIndex[I, O any] struct {
- extract func(o O) []string
- index map[string]sets.Set[Key[O]]
- parent *manyCollection[I, O]
-}
+func (i dependencyState[I]) update(key Key[I], deps []*dependency) {
+ // Update the I -> Dependency mapping
+ i.objectDependencies[key] = deps
+ for _, d := range deps {
+ if depKeys, typ, extractor, filterID, ok := d.filter.reverseIndexKey(); ok {
+ for _, depKey := range depKeys {
+ k := indexedDependency{
+ id: d.id,
+ key: depKey,
+ typ: typ,
+ }
+ if typ == unknownIndexType && extractor == nil {
+ // no need to make keys in the reverse index
+ // if no extractor specified
+ continue
+ }
-type dependencyState[I any] struct {
- collectionDependencies sets.Set[collectionUID]
- objectDependencies map[Key[I]][]*dependency
- indexedDependencies map[indexedDependency]sets.Set[Key[I]]
- indexedDependenciesExtractor map[extractorKey]objectKeyExtractor
+ sets.InsertOrNew(i.indexedDependencies, k, key)
+ kk := extractorKey{
+ filterUID: filterID,
+ uid: d.id,
+ typ: typ,
+ }
+ i.indexedDependenciesExtractor[kk] = extractor
+ }
+ }
+ }
}
-func NewCollection[I, O any](c Collection[I], hf TransformationSingle[I, O], opts ...CollectionOption) Collection[O] {
- hm := func(ctx HandlerContext, i I) []O {
- res := hf(ctx, i)
- if res == nil {
- return nil
- }
- return []O{*res}
+func (i dependencyState[I]) delete(key Key[I]) {
+ old, f := i.objectDependencies[key]
+ if !f {
+ return
}
- o := buildCollectionOptions(opts...)
- if o.name == "" {
- o.name = fmt.Sprintf("Collection[%v,%v]", ptr.TypeName[I](), ptr.TypeName[O]())
+ delete(i.objectDependencies, key)
+ for _, d := range old {
+ if depKeys, typ, _, _, ok := d.filter.reverseIndexKey(); ok {
+ for _, depKey := range depKeys {
+ k := indexedDependency{
+ id: d.id,
+ key: depKey,
+ typ: typ,
+ }
+ sets.DeleteCleanupLast(i.indexedDependencies, k, key)
+ }
+ }
}
- return newManyCollection(c, hm, o, nil)
}
-func newManyCollection[I, O any](
- cc Collection[I],
- hf TransformationMulti[I, O],
- opts collectionOptions,
- onPrimaryInputEventHandler func([]Event[I]),
-) Collection[O] {
- c := cc.(internalCollection[I])
+func (i dependencyState[I]) changedInputKeys(sourceCollection collectionUID, events []Event[any]) sets.Set[Key[I]] {
+ changedInputKeys := sets.Set[Key[I]]{}
+ // Check old and new
+ for _, ev := range events {
+ // We have a possibly dependant object changed. For each input object, see if it depends on the object.
+ // Naively, we can look through every item in this collection and check if it matches the filter. However, this is
+ // inefficient, especially when the dependency changes frequently and the collection is large.
+ // Where possible, we utilize the reverse-indexing to get the precise list of potentially changed objects.
+ foundAny := false
- h := &manyCollection[I, O]{
- transformation: hf,
- collectionName: opts.name,
- id: nextUID(),
- parent: c,
- dependencyState: dependencyState[I]{
- collectionDependencies: sets.New[collectionUID](),
- objectDependencies: map[Key[I]][]*dependency{},
- indexedDependencies: map[indexedDependency]sets.Set[Key[I]]{},
- indexedDependenciesExtractor: map[extractorKey]func(o any) []string{},
- },
- collectionState: multiIndex[I, O]{
- inputs: map[Key[I]]I{},
- outputs: map[Key[O]]O{},
- mappings: map[Key[I]]sets.Set[Key[O]]{},
- },
- indexes: make(map[string]collectionIndex[I, O]),
- eventHandlers: newHandlerSet[O](),
- augmentation: opts.augmentation,
- synced: make(chan struct{}),
- stop: opts.stop,
- onPrimaryInputEventHandler: onPrimaryInputEventHandler,
- }
+ // find all the reverse indexes related to the sourceCollection
+ // N here is usually going to be small (the number of FilterKey/FilterIndex)
+ extractorKeys := []extractorKey{}
+ for k := range i.indexedDependenciesExtractor {
+ if k.typ != unknownIndexType && k.uid == sourceCollection {
+ extractorKeys = append(extractorKeys, k)
+ }
+ }
- if opts.metadata != nil {
- h.metadata = opts.metadata
+ for _, ekey := range extractorKeys {
+ if extractor, f := i.indexedDependenciesExtractor[ekey]; f {
+ foundAny = true
+ // We have a reverse index
+ for _, item := range ev.Items() {
+ // Find all the reverse index keys for this object. For each key we will find impacted input objects.
+ keys := extractor(item)
+ for _, key := range keys {
+ for iKey := range i.indexedDependencies[indexedDependency{id: sourceCollection, key: key, typ: ekey.typ}] {
+ if changedInputKeys.Contains(iKey) {
+ // We may have already found this item, skip it
+ continue
+ }
+ dependencies := i.objectDependencies[iKey]
+ if changed := objectChanged(dependencies, sourceCollection, ev, true); changed {
+ changedInputKeys.Insert(iKey)
+ }
+ }
+ }
+ }
+ }
+ }
+ if !foundAny {
+ for iKey, dependencies := range i.objectDependencies {
+ if changed := objectChanged(dependencies, sourceCollection, ev, false); changed {
+ changedInputKeys.Insert(iKey)
+ }
+ }
+ }
}
+ return changedInputKeys
+}
- h.syncer = channelSyncer{
- name: h.collectionName,
- synced: h.synced,
+func objectChanged(dependencies []*dependency, sourceCollection collectionUID, ev Event[any], preFiltered bool) bool {
+ for _, dep := range dependencies {
+ id := dep.id
+ if id != sourceCollection {
+ continue
+ }
+ // For each input, we will check if it depends on this event.
+ // We use Items() to check both the old and new object; we will recompute if either matched
+ for _, item := range ev.Items() {
+ match := dep.filter.Matches(item, preFiltered)
+ if match {
+ // Its a match! Return now. We don't need to check all dependencies, since we just need to find if any of them changed
+ return true
+ }
+ }
}
+ return false
+}
- h.queue = queue.NewWithSync(func() {
- close(h.synced)
- fmt.Printf("\n%v synced (uid %v)\n", h.name(), h.uid())
- }, h.collectionName)
-
- go h.runQueue()
+// manyCollection builds a mapping from I->O.
+// This can be built from transformation functions of I->*O or I->[]O; both are implemented by this same struct.
+type manyCollection[I, O any] struct {
+ // collectionName provides the collectionName for this collection.
+ collectionName string
+ id collectionUID
+ // parent is the input collection we are building off of.
+ parent Collection[I]
+
+ // log is a logger for the collection, with additional labels already added to identify it.
+ // This can be acquired with blockNewEvents held, but only with strict ordering (mu inside blockNewEvents)
+ // mu protects all items grouped below(collectionState, dependencyState, indexes).
+ // This is acquired for reads and writes of data.
+ mu sync.RWMutex
+ collectionState multiIndex[I, O]
+ dependencyState dependencyState[I]
+ // internal indexes
+ indexes map[string]collectionIndex[I, O]
+
+ // eventHandlers is a list of event handlers registered for the collection. On any changes, each will be notified.
+ eventHandlers *handlerSet[O]
+
+ transformation TransformationMulti[I, O]
+
+ // augmentation allows transforming an object into another for usage throughout the library. See WithObjectAugmentation.
+ augmentation func(a any) any
+ synced chan struct{}
+ stop <-chan struct{}
+ queue queue.Instance
+ metadata Metadata
+
+ // onPrimaryInputEventHandler is a specialized internal handler that runs synchronously when a primary input changes
+ onPrimaryInputEventHandler func(o []Event[I])
- return h
+ syncer Syncer
}
-func (h *manyCollection[I, O]) name() string {
- return h.collectionName
+type collectionIndex[I, O any] struct {
+ extract func(o O) []string
+ index map[string]sets.Set[Key[O]]
+ parent *manyCollection[I, O]
}
-// nolint: unused // (not true, its to implement an interface)
-func (h *manyCollection[I, O]) uid() collectionUID {
- return h.id
+func (c collectionIndex[I, O]) Lookup(key string) []O {
+ c.parent.mu.RLock()
+ defer c.parent.mu.RUnlock()
+ keys := c.index[key]
+
+ res := make([]O, 0, len(keys))
+ for k := range keys {
+ v, f := c.parent.collectionState.outputs[k]
+ if !f {
+ continue
+ }
+ res = append(res, v)
+ }
+ return res
}
-type internalCollection[T any] interface {
- Collection[T]
- name() string
- uid() collectionUID
- dump() CollectionDump
- augment(any) any
- index(name string, extract func(o T) []string) indexer[T]
+func (c collectionIndex[I, O]) delete(o O, oKey Key[O]) {
+ oldIndexKeys := c.extract(o)
+ for _, oldIndexKey := range oldIndexKeys {
+ sets.DeleteCleanupLast(c.index, oldIndexKey, oKey)
+ }
}
-type manyCollection[I, O any] struct {
- collectionName string
- id collectionUID
- parent Collection[I]
- mu sync.RWMutex
- collectionState multiIndex[I, O]
- dependencyState dependencyState[I]
- indexes map[string]collectionIndex[I, O]
- eventHandlers *handlerSet[O]
- transformation TransformationMulti[I, O]
- augmentation func(a any) any
- synced chan struct{}
- stop <-chan struct{}
- metadata Metadata
- onPrimaryInputEventHandler func(o []Event[I])
- syncer Syncer
- queue queue.Instance
+func (c collectionIndex[I, O]) update(ev Event[O], oKey Key[O]) {
+ if ev.Old != nil {
+ c.delete(*ev.Old, oKey)
+ }
+ if ev.New != nil {
+ newIndexKeys := c.extract(*ev.New)
+ for _, newIndexKey := range newIndexKeys {
+ sets.InsertOrNew(c.index, newIndexKey, oKey)
+ }
+ }
}
+var _ internalCollection[any] = &manyCollection[any, any]{}
+
type handlers[O any] struct {
mu sync.RWMutex
h []*singletonHandlerRegistration[O]
@@ -167,6 +250,15 @@ type singletonHandlerRegistration[O any] struct {
fn func(o []Event[O])
}
+func (o *handlers[O]) MarkInitialized() []func(o []Event[O]) {
+ o.mu.Lock()
+ defer o.mu.Unlock()
+ o.init = true
+ return slices.Map(o.h, func(e *singletonHandlerRegistration[O]) func(o []Event[O]) {
+ return e.fn
+ })
+}
+
func (o *handlers[O]) Insert(f func(o []Event[O])) *singletonHandlerRegistration[O] {
o.mu.Lock()
defer o.mu.Unlock()
@@ -191,85 +283,97 @@ func (o *handlers[O]) Get() []func(o []Event[O]) {
})
}
-func (h *manyCollection[I, O]) WaitUntilSynced(stop <-chan struct{}) bool {
- return h.syncer.WaitUntilSynced(stop)
+// multiIndex stores input and output objects.
+// Each input and output can be looked up by its key.
+// Additionally, a mapping of input key -> output keys stores the transformation.
+type multiIndex[I, O any] struct {
+ outputs map[Key[O]]O
+ inputs map[Key[I]]I
+ mappings map[Key[I]]sets.Set[Key[O]]
}
-func (h *manyCollection[I, O]) GetKey(k string) (res *O) {
- h.mu.RLock()
- defer h.mu.RUnlock()
- rf, f := h.collectionState.outputs[Key[O](k)]
- if f {
- return &rf
- }
- return nil
+func (h *manyCollection[I, O]) HasSynced() bool {
+ return h.syncer.HasSynced()
}
-func (h *manyCollection[I, O]) List() (res []O) {
+func (h *manyCollection[I, O]) WaitUntilSynced(stop <-chan struct{}) bool {
+ return h.syncer.WaitUntilSynced(stop)
+}
+
+// nolint: unused // (not true, its to implement an interface)
+func (h *manyCollection[I, O]) dump() CollectionDump {
h.mu.RLock()
defer h.mu.RUnlock()
- return maps.Values(h.collectionState.outputs)
-}
-func (h *manyCollection[I, O]) Register(f func(o Event[O])) HandlerRegistration {
- return registerHandlerAsBatched(h, f)
-}
-
-func (h *manyCollection[I, O]) RegisterBatch(f func(o []Event[O]), runExistingState bool) HandlerRegistration {
- if !runExistingState {
- // If we don't to run the initial state this is simple, we just register the handler.
- h.mu.Lock()
- defer h.mu.Unlock()
- return h.eventHandlers.Insert(f, h, nil, h.stop)
+ inputs := make(map[string]InputDump, len(h.collectionState.inputs))
+ for k, v := range h.collectionState.mappings {
+ output := make([]string, 0, len(v))
+ for vv := range v {
+ output = append(output, string(vv))
+ }
+ slices.Sort(output)
+ inputs[string(k)] = InputDump{
+ Outputs: output,
+ Dependencies: nil, // filled later
+ }
}
- // We need to run the initial state, but we don't want to get duplicate events.
- // We should get "ADD initialObject1, ADD initialObjectN, UPDATE someLaterUpdate" without mixing the initial ADDs
- // To do this we block any new event processing
- // Get initial state
- h.mu.RLock()
- defer h.mu.RUnlock()
-
- events := make([]Event[O], 0, len(h.collectionState.outputs))
- for _, o := range h.collectionState.outputs {
- events = append(events, Event[O]{
- New: &o,
- Event: controllers.EventAdd,
- })
+ for k, deps := range h.dependencyState.objectDependencies {
+ depss := make([]string, 0, len(deps))
+ for _, dep := range deps {
+ depss = append(depss, dep.collectionName)
+ }
+ slices.Sort(depss)
+ cur := inputs[string(k)]
+ cur.Dependencies = depss
+ inputs[string(k)] = cur
}
- // Send out all the initial objects to the handler. We will then unlock the new events so it gets the future updates.
- return h.eventHandlers.Insert(f, h, events, h.stop)
+ return CollectionDump{
+ Outputs: eraseMap(h.collectionState.outputs),
+ Inputs: inputs,
+ InputCollection: h.parent.(internalCollection[I]).name(),
+ Synced: h.HasSynced(),
+ }
}
-func (h *manyCollection[I, O]) HasSynced() bool {
- return h.syncer.HasSynced()
+// nolint: unused // (not true, its to implement an interface)
+func (h *manyCollection[I, O]) augment(a any) any {
+ if h.augmentation != nil {
+ return h.augmentation(a)
+ }
+ return a
}
-func (h *manyCollection[I, O]) Metadata() Metadata {
- return h.metadata
-}
+// nolint: unused // (not true)
+func (h *manyCollection[I, O]) index(name string, extract func(o O) []string) indexer[O] {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ if idx, ok := h.indexes[name]; ok {
+ return idx
+ }
-func (h *manyCollection[I, O]) runQueue() {
- c := h.parent
- if !c.WaitUntilSynced(h.stop) {
- return
+ idx := collectionIndex[I, O]{
+ extract: extract,
+ index: make(map[string]sets.Set[Key[O]]),
+ parent: h,
}
- syncer := c.RegisterBatch(func(o []Event[I]) {
- if h.onPrimaryInputEventHandler != nil {
- h.onPrimaryInputEventHandler(o)
- }
- h.queue.Push(func() error {
- h.onPrimaryInputEvent(o)
- return nil
- })
- }, true)
- if !syncer.WaitUntilSynced(h.stop) {
- return
+ for k, v := range h.collectionState.outputs {
+ idx.update(Event[O]{
+ Old: nil,
+ New: &v,
+ Event: controllers.EventAdd,
+ }, k)
}
- h.queue.Run(h.stop)
+ h.indexes[name] = idx
+ return idx
}
+// onPrimaryInputEvent takes a list of I's that changed and reruns the handler over them.
+// This is called either when I directly changes, or if a secondary dependency changed. In this case, we compute which I's depended
+// on the secondary dependency, and call onPrimaryInputEvent with them
func (h *manyCollection[I, O]) onPrimaryInputEvent(items []Event[I]) {
+ // Between the events being enqueued and now, the input may have changed. Update with latest info.
+ // Note we now have the `blockNewEvents` lock so this is safe; any futures calls will do the same so always have up-to-date information.
for idx, ev := range items {
iKey := GetKey(ev.Latest())
iObj := h.parent.GetKey(iKey)
@@ -288,19 +392,7 @@ func (h *manyCollection[I, O]) onPrimaryInputEvent(items []Event[I]) {
h.handleChangedPrimaryInputEvents(items)
}
-type collectionDependencyTracker[I, O any] struct {
- *manyCollection[I, O]
- d []*dependency
- key Key[I]
- discardUpdate bool
-}
-
-func (i *collectionDependencyTracker[I, O]) DiscardResult() {
- i.discardUpdate = true
-}
-
-const EnableAssertions = false
-
+// handleChangedPrimaryInputEvents takes a list of I's that changed and reruns the handler over them.
func (h *manyCollection[I, O]) handleChangedPrimaryInputEvents(items []Event[I]) {
var events []Event[O]
recomputedResults := make([]map[Key[O]]O, len(items))
@@ -314,7 +406,10 @@ func (h *manyCollection[I, O]) handleChangedPrimaryInputEvents(items []Event[I])
i := a.Latest()
iKey := getTypedKey(i)
- ctx := &collectionDependencyTracker[I, O]{manyCollection: h, key: iKey}
+ ctx := &collectionDependencyTracker[I, O]{
+ manyCollection: h,
+ key: iKey,
+ }
results := slices.GroupUnique(h.transformation(ctx, i), getTypedKey[O])
recomputedResults[idx] = results
// Store new dependency state, to insert in the next loop under the lock
@@ -342,6 +437,7 @@ func (h *manyCollection[I, O]) handleChangedPrimaryInputEvents(items []Event[I])
for _, index := range h.indexes {
index.delete(oldRes, oKey)
}
+
}
delete(h.collectionState.mappings, iKey)
delete(h.collectionState.inputs, iKey)
@@ -409,244 +505,253 @@ func (h *manyCollection[I, O]) handleChangedPrimaryInputEvents(items []Event[I])
if len(events) == 0 {
return
}
-
h.eventHandlers.Distribute(events, !h.HasSynced())
}
-func (h *manyCollection[I, O]) assertIndexConsistency() {
- oToI := map[Key[O]]Key[I]{}
- for i, os := range h.collectionState.mappings {
- if _, f := h.collectionState.inputs[i]; !f {
- panic(fmt.Sprintf("for mapping key %v, no input found", i))
- }
- for o := range os {
- if ci, f := oToI[o]; f {
- panic(fmt.Sprintf("duplicate mapping %v: input %v and %v both map to it", o, ci, i))
- }
- oToI[o] = i
- if _, f := h.collectionState.outputs[o]; !f {
- panic(fmt.Sprintf("for mapping key %v->%v, no output found", i, o))
- }
+func (h *manyCollection[I, O]) Metadata() Metadata {
+ return h.metadata
+}
+
+// NewCollection transforms a Collection[I] to a Collection[O] by applying the provided transformation function.
+// This applies for one-to-one relationships between I and O.
+// For zero-to-one, use NewSingleton. For one-to-many, use NewManyCollection.
+func NewCollection[I, O any](c Collection[I], hf TransformationSingle[I, O], opts ...CollectionOption) Collection[O] {
+ // For implementation simplicity, represent TransformationSingle as a TransformationMulti so we can share an implementation.
+ hm := func(ctx HandlerContext, i I) []O {
+ res := hf(ctx, i)
+ if res == nil {
+ return nil
}
+ return []O{*res}
+ }
+ o := buildCollectionOptions(opts...)
+ if o.name == "" {
+ // NOTE: this will print Collection[nil, nil] if I or O are interfaces
+ o.name = fmt.Sprintf("Collection[%v,%v]", ptr.TypeName[I](), ptr.TypeName[O]())
}
+ return newManyCollection(c, hm, o, nil)
}
-func (i dependencyState[I]) update(key Key[I], deps []*dependency) {
- // Update the I -> Dependency mapping
- i.objectDependencies[key] = deps
- for _, d := range deps {
- if depKeys, typ, extractor, filterID, ok := d.filter.reverseIndexKey(); ok {
- for _, depKey := range depKeys {
- k := indexedDependency{
- id: d.id,
- key: depKey,
- typ: typ,
- }
- if typ == unknownIndexType && extractor == nil {
- // no need to make keys in the reverse index
- // if no extractor specified
- continue
- }
+func newManyCollection[I, O any](
+ cc Collection[I],
+ hf TransformationMulti[I, O],
+ opts collectionOptions,
+ onPrimaryInputEventHandler func([]Event[I]),
+) Collection[O] {
+ c := cc.(internalCollection[I])
- sets.InsertOrNew(i.indexedDependencies, k, key)
- kk := extractorKey{
- filterUID: filterID,
- uid: d.id,
- typ: typ,
- }
- i.indexedDependenciesExtractor[kk] = extractor
- }
- }
+ h := &manyCollection[I, O]{
+ transformation: hf,
+ collectionName: opts.name,
+ id: nextUID(),
+ parent: c,
+ dependencyState: dependencyState[I]{
+ collectionDependencies: sets.New[collectionUID](),
+ objectDependencies: map[Key[I]][]*dependency{},
+ indexedDependencies: map[indexedDependency]sets.Set[Key[I]]{},
+ indexedDependenciesExtractor: map[extractorKey]func(o any) []string{},
+ },
+ collectionState: multiIndex[I, O]{
+ inputs: map[Key[I]]I{},
+ outputs: map[Key[O]]O{},
+ mappings: map[Key[I]]sets.Set[Key[O]]{},
+ },
+ indexes: make(map[string]collectionIndex[I, O]),
+ eventHandlers: newHandlerSet[O](),
+ augmentation: opts.augmentation,
+ synced: make(chan struct{}),
+ stop: opts.stop,
+ onPrimaryInputEventHandler: onPrimaryInputEventHandler,
+ }
+
+ if opts.metadata != nil {
+ h.metadata = opts.metadata
+ }
+
+ h.syncer = channelSyncer{
+ name: h.collectionName,
+ synced: h.synced,
}
+
+ h.queue = queue.NewWithSync(func() {
+ close(h.synced)
+ fmt.Printf("%v synced (uid %v)\n", h.name(), h.uid())
+ }, h.collectionName)
+
+ go h.runQueue()
+
+ return h
}
-func (i dependencyState[I]) delete(key Key[I]) {
- old, f := i.objectDependencies[key]
- if !f {
+func (h *manyCollection[I, O]) runQueue() {
+ c := h.parent
+ // Wait for primary dependency to be ready
+ if !c.WaitUntilSynced(h.stop) {
return
}
- delete(i.objectDependencies, key)
- for _, d := range old {
- if depKeys, typ, _, _, ok := d.filter.reverseIndexKey(); ok {
- for _, depKey := range depKeys {
- k := indexedDependency{
- id: d.id,
- key: depKey,
- typ: typ,
- }
- sets.DeleteCleanupLast(i.indexedDependencies, k, key)
- }
+ // Now register to our primary collection. On any event, we will enqueue the update.
+ syncer := c.RegisterBatch(func(o []Event[I]) {
+ if h.onPrimaryInputEventHandler != nil {
+ h.onPrimaryInputEventHandler(o)
}
+ h.queue.Push(func() error {
+ h.onPrimaryInputEvent(o)
+ return nil
+ })
+ }, true)
+ // Wait for everything initial state to be enqueued
+ if !syncer.WaitUntilSynced(h.stop) {
+ return
}
+ h.queue.Run(h.stop)
}
-func (i dependencyState[I]) changedInputKeys(sourceCollection collectionUID, events []Event[any]) sets.Set[Key[I]] {
- changedInputKeys := sets.Set[Key[I]]{}
- // Check old and new
- for _, ev := range events {
- // We have a possibly dependant object changed. For each input object, see if it depends on the object.
- // Naively, we can look through every item in this collection and check if it matches the filter. However, this is
- // inefficient, especially when the dependency changes frequently and the collection is large.
- // Where possible, we utilize the reverse-indexing to get the precise list of potentially changed objects.
- foundAny := false
+func (h *manyCollection[I, O]) onSecondaryDependencyEvent(sourceCollection collectionUID, events []Event[any]) {
+ // A secondary dependency changed...
+ // Got an event. Now we need to find out who depends on it..
+ changedInputKeys := h.dependencyState.changedInputKeys(sourceCollection, events)
- // find all the reverse indexes related to the sourceCollection
- // N here is usually going to be small (the number of FilterKey/FilterIndex)
- extractorKeys := []extractorKey{}
- for k := range i.indexedDependenciesExtractor {
- if k.typ != unknownIndexType && k.uid == sourceCollection {
- extractorKeys = append(extractorKeys, k)
- }
- }
-
- for _, ekey := range extractorKeys {
- if extractor, f := i.indexedDependenciesExtractor[ekey]; f {
- foundAny = true
- // We have a reverse index
- for _, item := range ev.Items() {
- // Find all the reverse index keys for this object. For each key we will find impacted input objects.
- keys := extractor(item)
- for _, key := range keys {
- for iKey := range i.indexedDependencies[indexedDependency{id: sourceCollection, key: key, typ: ekey.typ}] {
- if changedInputKeys.Contains(iKey) {
- // We may have already found this item, skip it
- continue
- }
- dependencies := i.objectDependencies[iKey]
- if changed := objectChanged(dependencies, sourceCollection, ev, true); changed {
- changedInputKeys.Insert(iKey)
- }
- }
- }
+ toRun := make([]Event[I], 0, len(changedInputKeys))
+ // Now we have the set of input keys that changed. We need to recompute all of these.
+ // While we could just do that manually, to re-use code, we will convert these into Event[I] and use the same logic as
+ // we would if the input itself changed.
+ for i := range changedInputKeys {
+ iObj := h.parent.GetKey(string(i))
+ if iObj == nil {
+ // Object no longer found means it has been deleted.
+ for oKey := range h.collectionState.mappings[i] {
+ _, f := h.collectionState.outputs[oKey]
+ if !f {
+ continue
}
- }
- }
- if !foundAny {
- for iKey, dependencies := range i.objectDependencies {
- if changed := objectChanged(dependencies, sourceCollection, ev, false); changed {
- changedInputKeys.Insert(iKey)
+ e := Event[I]{
+ Event: controllers.EventDelete,
+ Old: ptr.Of(h.collectionState.inputs[i]),
}
+ toRun = append(toRun, e)
}
+ } else {
+ // Typically an EventUpdate should have Old and New. We only have New here.
+ // In practice, this is an internal surface only so we just make sure onPrimaryInputEvent handles this.
+ toRun = append(toRun, Event[I]{
+ Event: controllers.EventUpdate,
+ New: iObj,
+ })
}
}
- return changedInputKeys
+ h.handleChangedPrimaryInputEvents(toRun)
}
-func objectChanged(dependencies []*dependency, sourceCollection collectionUID, ev Event[any], preFiltered bool) bool {
- for _, dep := range dependencies {
- id := dep.id
- if id != sourceCollection {
- continue
- }
- // For each input, we will check if it depends on this event.
- // We use Items() to check both the old and new object; we will recompute if either matched
- for _, item := range ev.Items() {
- match := dep.filter.Matches(item, preFiltered)
- if match {
- // Its a match! Return now. We don't need to check all dependencies, since we just need to find if any of them changed
- return true
- }
- }
- }
- return false
+// nolint: unused // it is used to implement interface
+func (h *manyCollection[I, O]) _internalHandler() {
}
-func (h *manyCollection[I, O]) dump() CollectionDump {
+func (h *manyCollection[I, O]) GetKey(k string) (res *O) {
h.mu.RLock()
defer h.mu.RUnlock()
-
- inputs := make(map[string]InputDump, len(h.collectionState.inputs))
- for k, v := range h.collectionState.mappings {
- output := make([]string, 0, len(v))
- for vv := range v {
- output = append(output, string(vv))
- }
- slices.Sort(output)
- inputs[string(k)] = InputDump{
- Outputs: output,
- Dependencies: nil, // filled later
- }
- }
- for k, deps := range h.dependencyState.objectDependencies {
- depss := make([]string, 0, len(deps))
- for _, dep := range deps {
- depss = append(depss, dep.collectionName)
- }
- slices.Sort(depss)
- cur := inputs[string(k)]
- cur.Dependencies = depss
- inputs[string(k)] = cur
+ rf, f := h.collectionState.outputs[Key[O](k)]
+ if f {
+ return &rf
}
+ return nil
+}
- return CollectionDump{
- Outputs: eraseMap(h.collectionState.outputs),
- Inputs: inputs,
- InputCollection: h.parent.(internalCollection[I]).name(),
- Synced: h.HasSynced(),
- }
+func (h *manyCollection[I, O]) List() (res []O) {
+ h.mu.RLock()
+ defer h.mu.RUnlock()
+ return maps.Values(h.collectionState.outputs)
}
-func (h *manyCollection[I, O]) index(name string, extract func(o O) []string) indexer[O] {
- h.mu.Lock()
- defer h.mu.Unlock()
- if idx, ok := h.indexes[name]; ok {
- return idx
- }
+func (h *manyCollection[I, O]) Register(f func(o Event[O])) HandlerRegistration {
+ return registerHandlerAsBatched(h, f)
+}
- idx := collectionIndex[I, O]{
- extract: extract,
- index: make(map[string]sets.Set[Key[O]]),
- parent: h,
+func (h *manyCollection[I, O]) RegisterBatch(f func(o []Event[O]), runExistingState bool) HandlerRegistration {
+ if !runExistingState {
+ // If we don't to run the initial state this is simple, we just register the handler.
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ return h.eventHandlers.Insert(f, h, nil, h.stop)
}
- for k, v := range h.collectionState.outputs {
- idx.update(Event[O]{
- Old: nil,
- New: &v,
+ h.mu.RLock()
+ defer h.mu.RUnlock()
+
+ events := make([]Event[O], 0, len(h.collectionState.outputs))
+ for _, o := range h.collectionState.outputs {
+ events = append(events, Event[O]{
+ New: &o,
Event: controllers.EventAdd,
- }, k)
+ })
}
- h.indexes[name] = idx
- return idx
+
+ return h.eventHandlers.Insert(f, h, events, h.stop)
}
-func (h *manyCollection[I, O]) augment(a any) any {
- if h.augmentation != nil {
- return h.augmentation(a)
- }
- return a
+func (h *manyCollection[I, O]) name() string {
+ return h.collectionName
}
-func (c collectionIndex[I, O]) Lookup(key string) []O {
- c.parent.mu.RLock()
- defer c.parent.mu.RUnlock()
- keys := c.index[key]
+// nolint: unused // (not true, its to implement an interface)
+func (h *manyCollection[I, O]) uid() collectionUID {
+ return h.id
+}
- res := make([]O, 0, len(keys))
- for k := range keys {
- v, f := c.parent.collectionState.outputs[k]
- if !f {
- continue
+func (h *manyCollection[I, O]) assertIndexConsistency() {
+ oToI := map[Key[O]]Key[I]{}
+ for i, os := range h.collectionState.mappings {
+ if _, f := h.collectionState.inputs[i]; !f {
+ panic(fmt.Sprintf("for mapping key %v, no input found", i))
+ }
+ for o := range os {
+ if ci, f := oToI[o]; f {
+ panic(fmt.Sprintf("duplicate mapping %v: input %v and %v both map to it", o, ci, i))
+ }
+ oToI[o] = i
+ if _, f := h.collectionState.outputs[o]; !f {
+ panic(fmt.Sprintf("for mapping key %v->%v, no output found", i, o))
+ }
}
- res = append(res, v)
}
- return res
}
-func (c collectionIndex[I, O]) update(ev Event[O], oKey Key[O]) {
- if ev.Old != nil {
- c.delete(*ev.Old, oKey)
- }
- if ev.New != nil {
- newIndexKeys := c.extract(*ev.New)
- for _, newIndexKey := range newIndexKeys {
- sets.InsertOrNew(c.index, newIndexKey, oKey)
- }
- }
+type collectionDependencyTracker[I, O any] struct {
+ *manyCollection[I, O]
+ d []*dependency
+ key Key[I]
+ discardUpdate bool
}
-func (c collectionIndex[I, O]) delete(o O, oKey Key[O]) {
- oldIndexKeys := c.extract(o)
- for _, oldIndexKey := range oldIndexKeys {
- sets.DeleteCleanupLast(c.index, oldIndexKey, oKey)
+func (i *collectionDependencyTracker[I, O]) name() string {
+ return fmt.Sprintf("%s{%s}", i.collectionName, i.key)
+}
+
+// registerDependency track a dependency. This is in the context of a specific input I type, as we create a collectionDependencyTracker
+// per I.
+func (i *collectionDependencyTracker[I, O]) registerDependency(
+ d *dependency,
+ syncer Syncer,
+ register func(f erasedEventHandler) Syncer,
+) {
+ i.d = append(i.d, d)
+
+ i.mu.Lock()
+ existed := i.dependencyState.collectionDependencies.InsertContains(d.id)
+ i.mu.Unlock()
+ // For any new collections we depend on, start watching them if its the first time we have watched them.
+ if !existed {
+ syncer.WaitUntilSynced(i.stop)
+ register(func(o []Event[any]) {
+ i.queue.Push(func() error {
+ i.onSecondaryDependencyEvent(d.id, o)
+ return nil
+ })
+ }).WaitUntilSynced(i.stop)
}
}
+
+func (i *collectionDependencyTracker[I, O]) _internalHandler() {
+}
+
+func (i *collectionDependencyTracker[I, O]) DiscardResult() {
+ i.discardUpdate = true
+}
diff --git a/pkg/kube/krt/core.go b/pkg/kube/krt/core.go
index 0fd0a1c2..befee765 100644
--- a/pkg/kube/krt/core.go
+++ b/pkg/kube/krt/core.go
@@ -2,20 +2,23 @@ package krt
import "github.com/apache/dubbo-kubernetes/pkg/kube/controllers"
+type Metadata map[string]any
+
type FetchOption func(*dependency)
type HandlerContext interface {
DiscardResult()
+ _internalHandler()
}
+type Key[O any] string
+
type (
TransformationEmpty[T any] func(ctx HandlerContext) *T
TransformationMulti[I, O any] func(ctx HandlerContext, i I) []O
TransformationSingle[I, O any] func(ctx HandlerContext, i I) *O
)
-type Metadata map[string]any
-
type Event[T any] struct {
Old *T
New *T
@@ -90,3 +93,21 @@ type ResourceNamer interface {
type uidable interface {
uid() collectionUID
}
+
+type internalCollection[T any] interface {
+ Collection[T]
+
+ // Name is a human facing name for this collection.
+ // Note this may not be universally unique
+ name() string
+ // Uid is an internal unique ID for this collection. MUST be globally unique
+ uid() collectionUID
+
+ dump() CollectionDump
+
+ // Augment mutates an object for use in various function calls. See WithObjectAugmentation
+ augment(any) any
+
+ // Create a new index into the collection
+ index(name string, extract func(o T) []string) indexer[T]
+}
diff --git a/pkg/kube/krt/debug.go b/pkg/kube/krt/debug.go
index 860579be..6b213b25 100644
--- a/pkg/kube/krt/debug.go
+++ b/pkg/kube/krt/debug.go
@@ -47,20 +47,6 @@ func (p DebugCollection) MarshalJSON() ([]byte, error) {
})
}
-func maybeRegisterCollectionForDebugging[T any](c Collection[T], handler *DebugHandler) {
- if handler == nil {
- return
- }
- cc := c.(internalCollection[T])
- handler.mu.Lock()
- defer handler.mu.Unlock()
- handler.debugCollections = append(handler.debugCollections, DebugCollection{
- name: cc.name(),
- dump: cc.dump,
- uid: cc.uid(),
- })
-}
-
// nolint: unused // (not true, not sure why it thinks it is!)
func eraseMap[T any](l map[Key[T]]T) map[string]any {
nm := make(map[string]any, len(l))
diff --git a/pkg/kube/krt/fetch.go b/pkg/kube/krt/fetch.go
index 4d8408c2..6b1d52fa 100644
--- a/pkg/kube/krt/fetch.go
+++ b/pkg/kube/krt/fetch.go
@@ -1,5 +1,9 @@
package krt
+import (
+ "github.com/apache/dubbo-kubernetes/pkg/util/slices"
+)
+
func FetchOne[T any](ctx HandlerContext, c Collection[T], opts ...FetchOption) *T {
res := Fetch[T](ctx, c, opts...)
switch len(res) {
@@ -17,5 +21,52 @@ func Fetch[T any](ctx HandlerContext, cc Collection[T], opts ...FetchOption) []T
}
func fetch[T any](ctx HandlerContext, cc Collection[T], allowMissingContext bool, opts ...FetchOption) []T {
- return nil
+ c := cc.(internalCollection[T])
+ d := &dependency{
+ id: c.uid(),
+ collectionName: c.name(),
+ filter: &filter{},
+ }
+ for _, o := range opts {
+ o(d)
+ }
+ if ctx != nil {
+ h := ctx.(registerDependency)
+ // Important: register before we List(), so we cannot miss any events
+ h.registerDependency(d, c, func(f erasedEventHandler) Syncer {
+ ff := func(o []Event[T]) {
+ f(slices.Map(o, castEvent[T, any]))
+ }
+ // Skip calling all the existing state for secondary dependencies, otherwise we end up with a deadlock due to
+ // rerunning the same collection's recomputation at the same time (once for the initial event, then for the initial registration).
+ return c.RegisterBatch(ff, false)
+ })
+ } else if !allowMissingContext {
+ panic("Fetch() requires a valid context")
+ }
+
+ // Now we can do the real fetching
+ // Compute our list of all possible objects that can match. Then we will filter them later.
+ // This pre-filtering upfront avoids extra work
+ var list []T
+ if !d.filter.keys.IsNil() {
+ // If they fetch a set of keys, directly Get these. Usually this is a single resource.
+ list = make([]T, 0, d.filter.keys.Len())
+ for _, k := range d.filter.keys.List() {
+ if i := c.GetKey(k); i != nil {
+ list = append(list, *i)
+ }
+ }
+ } else if d.filter.index != nil {
+ // Otherwise from an index; fetch from there. Often this is a list of a namespace
+ list = d.filter.index.list().([]T)
+ } else {
+ // Otherwise get everything
+ list = c.List()
+ }
+ list = slices.FilterInPlace(list, func(i T) bool {
+ o := c.augment(i)
+ return d.filter.Matches(o, true)
+ })
+ return list
}
diff --git a/pkg/kube/krt/filter.go b/pkg/kube/krt/filter.go
index 451c1723..782c4efa 100644
--- a/pkg/kube/krt/filter.go
+++ b/pkg/kube/krt/filter.go
@@ -2,9 +2,8 @@ package krt
import (
"fmt"
+ "github.com/apache/dubbo-kubernetes/pkg/config/labels"
"github.com/apache/dubbo-kubernetes/pkg/util/smallset"
- "istio.io/istio/pkg/config/labels"
- "istio.io/istio/pkg/log"
"reflect"
)
@@ -35,6 +34,8 @@ func getKeyExtractor(o any) []string {
return []string{GetKey(o)}
}
+type objectKeyExtractor = func(o any) []string
+
func (f *filter) reverseIndexKey() ([]string, indexedDependencyType, objectKeyExtractor, collectionUID, bool) {
if f.keys.Len() > 0 {
if f.index != nil {
@@ -87,17 +88,11 @@ func (f *filter) Matches(object any, forList bool) bool {
// First, lookup directly by key. This is cheap
// an empty set will match none
if !f.keys.IsNil() && !f.keys.Contains(GetKey[any](object)) {
- if log.DebugEnabled() {
- log.Debugf("no match key: %q vs %q", f.keys, GetKey[any](object))
- }
return false
}
// Index is also cheap, and often used to filter namespaces out. Make sure we do this early
if f.index != nil {
if !f.index.indexMatches(object) {
- if log.DebugEnabled() {
- log.Debugf("no match index")
- }
return false
}
}
@@ -105,27 +100,15 @@ func (f *filter) Matches(object any, forList bool) bool {
// Rest is expensive
if f.selects != nil && !labels.Instance(getLabelSelector(object)).SubsetOf(f.selects) {
- if log.DebugEnabled() {
- log.Debugf("no match selects: %q vs %q", f.selects, getLabelSelector(object))
- }
return false
}
if f.selectsNonEmpty != nil && !labels.Instance(getLabelSelector(object)).Match(f.selectsNonEmpty) {
- if log.DebugEnabled() {
- log.Debugf("no match selectsNonEmpty: %q vs %q", f.selectsNonEmpty, getLabelSelector(object))
- }
return false
}
if f.labels != nil && !labels.Instance(f.labels).SubsetOf(getLabels(object)) {
- if log.DebugEnabled() {
- log.Debugf("no match labels: %q vs %q", f.labels, getLabels(object))
- }
return false
}
if f.generic != nil && !f.generic(object) {
- if log.DebugEnabled() {
- log.Debugf("no match generic")
- }
return false
}
return true
diff --git a/pkg/kube/krt/helpers.go b/pkg/kube/krt/helpers.go
index 6da0cc10..73d9ad1b 100644
--- a/pkg/kube/krt/helpers.go
+++ b/pkg/kube/krt/helpers.go
@@ -81,7 +81,8 @@ func GetKey[O any](a O) string {
if ack != nil {
return *ack
}
- panic(fmt.Sprintf("Cannot get Key, got %T", a))
+ // panic(fmt.Sprintf("Cannot get Key, got %T", a))
+ return fmt.Sprintf("Cannot get Key, got %T", a)
}
// Named is a convenience struct. It is ideal to be embedded into a type that has a name and namespace,
@@ -144,9 +145,9 @@ func waitForCacheSync(name string, stop <-chan struct{}, collections ...<-chan s
t0 := time.Now()
defer func() {
if r {
- fmt.Printf("sync complete: name=%s, time=%v\n", name, time.Since(t0))
+ fmt.Printf("\nsync complete: name=%s, time=%v\n", name, time.Since(t0))
} else {
- fmt.Printf("sync failed: name=%s, time=%v\n", name, time.Since(t0))
+ fmt.Printf("\nsync failed: name=%s, time=%v\n", name, time.Since(t0))
}
}()
for _, col := range collections {
diff --git a/pkg/kube/krt/internal.go b/pkg/kube/krt/internal.go
index 1f942f34..200ab18b 100644
--- a/pkg/kube/krt/internal.go
+++ b/pkg/kube/krt/internal.go
@@ -3,6 +3,7 @@ package krt
import (
"fmt"
"github.com/apache/dubbo-kubernetes/pkg/config"
+ "github.com/apache/dubbo-kubernetes/pkg/ptr"
"go.uber.org/atomic"
"google.golang.org/protobuf/proto"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -47,6 +48,12 @@ func nextUID() collectionUID {
return collectionUID(globalUIDCounter.Inc())
}
+type indexedDependency struct {
+ id collectionUID
+ key string
+ typ indexedDependencyType
+}
+
func registerHandlerAsBatched[T any](c internalCollection[T], f func(o Event[T])) HandlerRegistration {
return c.RegisterBatch(func(events []Event[T]) {
for _, o := range events {
@@ -68,23 +75,24 @@ func Equal[O any](a, b O) bool {
if pk, ok := any(&a).(Equaler[*O]); ok {
return pk.Equals(&b)
}
- // Future improvement: add a default Kubernetes object implementation
- // ResourceVersion is tempting but probably not safe. If we are comparing objects from the API server its fine,
- // but often we will be operating on types generated by the controller itself.
- // We should have a way to opt-in to RV comparison, but not default to it.
ap, ok := any(a).(proto.Message)
if ok {
if reflect.TypeOf(ap.ProtoReflect().Interface()) == reflect.TypeOf(ap) {
return proto.Equal(ap, any(b).(proto.Message))
}
- // If not, this is an embedded proto most likely... Sneaky.
- // DeepEqual on proto is broken, so fail fast to avoid subtle errors.
panic(fmt.Sprintf("unable to compare object %T; perhaps it is embedding a protobuf? Provide an Equaler implementation", a))
}
return reflect.DeepEqual(a, b)
}
+type erasedEventHandler = func(o []Event[any])
+
+type registerDependency interface {
+ registerDependency(*dependency, Syncer, func(f erasedEventHandler) Syncer)
+ name() string
+}
+
func getLabels(a any) map[string]string {
al, ok := a.(Labeler)
if ok {
@@ -104,3 +112,16 @@ func getLabels(a any) map[string]string {
}
panic(fmt.Sprintf("No Labels, got %T", a))
}
+
+func castEvent[I, O any](o Event[I]) Event[O] {
+ e := Event[O]{
+ Event: o.Event,
+ }
+ if o.Old != nil {
+ e.Old = ptr.Of(any(*o.Old).(O))
+ }
+ if o.New != nil {
+ e.New = ptr.Of(any(*o.New).(O))
+ }
+ return e
+}
diff --git a/pkg/kube/krt/options.go b/pkg/kube/krt/options.go
index 58ddb030..609c56e5 100644
--- a/pkg/kube/krt/options.go
+++ b/pkg/kube/krt/options.go
@@ -3,14 +3,16 @@ package krt
type OptionsBuilder struct {
namePrefix string
stop <-chan struct{}
+ debugger *DebugHandler
}
type BuilderOption func(opt CollectionOption) OptionsBuilder
-func NewOptionsBuilder(stop <-chan struct{}, namePrefix string) OptionsBuilder {
+func NewOptionsBuilder(stop <-chan struct{}, namePrefix string, debugger *DebugHandler) OptionsBuilder {
return OptionsBuilder{
namePrefix: namePrefix,
stop: stop,
+ debugger: debugger,
}
}
@@ -18,12 +20,20 @@ func (k OptionsBuilder) Stop() <-chan struct{} {
return k.stop
}
+func (k OptionsBuilder) With(opts ...CollectionOption) []CollectionOption {
+ return append([]CollectionOption{WithDebugging(k.debugger), WithStop(k.stop)}, opts...)
+}
+
+func (k OptionsBuilder) Debugger() *DebugHandler {
+ return k.debugger
+}
+
func (k OptionsBuilder) WithName(n string) []CollectionOption {
name := n
if k.namePrefix != "" {
name = k.namePrefix + "/" + name
}
- return []CollectionOption{WithStop(k.stop), WithName(name)}
+ return []CollectionOption{WithDebugging(k.debugger), WithStop(k.stop), WithName(name)}
}
func WithStop(stop <-chan struct{}) CollectionOption {
diff --git a/pkg/kube/krt/singleton.go b/pkg/kube/krt/singleton.go
index 62708675..bcea2cc0 100644
--- a/pkg/kube/krt/singleton.go
+++ b/pkg/kube/krt/singleton.go
@@ -95,6 +95,10 @@ func (d *static[T]) RegisterBatch(f func(o []Event[T]), runExistingState bool) H
}}
}
+func (c collectionAdapter[T]) uid() collectionUID {
+ return c.c.(uidable).uid()
+}
+
type staticHandler struct {
Syncer
remove func()
diff --git a/pkg/mesh/mesh.go b/pkg/mesh/mesh.go
index 65471cee..5dc9d67d 100644
--- a/pkg/mesh/mesh.go
+++ b/pkg/mesh/mesh.go
@@ -7,13 +7,28 @@ import (
"github.com/apache/dubbo-kubernetes/pkg/util/protomarshal"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
"github.com/hashicorp/go-multierror"
+ "google.golang.org/protobuf/types/known/durationpb"
+ wrappers "google.golang.org/protobuf/types/known/wrapperspb"
meshconfig "istio.io/api/mesh/v1alpha1"
+ "istio.io/api/networking/v1alpha3"
"os"
"sigs.k8s.io/yaml"
+ "time"
)
func DefaultProxyConfig() *meshconfig.ProxyConfig {
- return &meshconfig.ProxyConfig{}
+ return &meshconfig.ProxyConfig{
+ ConfigPath: constants.ConfigPathDir,
+ ClusterName: &meshconfig.ProxyConfig_ServiceCluster{ServiceCluster: constants.ServiceClusterName},
+ DrainDuration: durationpb.New(45 * time.Second),
+ TerminationDrainDuration: durationpb.New(5 * time.Second),
+ ProxyAdminPort: 15000,
+ // TODO authpolicy
+ DiscoveryAddress: "dubbod.dubbo-system.svc:15012",
+ BinaryPath: constants.BinaryPathFilename,
+ StatNameLength: 189,
+ StatusPort: 15020,
+ }
}
func ReadMeshConfig(filename string) (*meshconfig.MeshConfig, error) {
@@ -90,8 +105,32 @@ func ApplyMeshConfig(yaml string, defaultConfig *meshconfig.MeshConfig) (*meshco
func DefaultMeshConfig() *meshconfig.MeshConfig {
proxyConfig := DefaultProxyConfig()
return &meshconfig.MeshConfig{
+ EnableTracing: true,
+ AccessLogFile: "",
+ AccessLogEncoding: meshconfig.MeshConfig_TEXT,
+ AccessLogFormat: "",
+ EnableEnvoyAccessLogService: false,
+ ProtocolDetectionTimeout: durationpb.New(0),
+ TrustDomain: constants.DefaultClusterLocalDomain,
+ TrustDomainAliases: []string{},
+ EnableAutoMtls: wrappers.Bool(true),
+ OutboundTrafficPolicy: &meshconfig.MeshConfig_OutboundTrafficPolicy{Mode: meshconfig.MeshConfig_OutboundTrafficPolicy_ALLOW_ANY},
+ InboundTrafficPolicy: &meshconfig.MeshConfig_InboundTrafficPolicy{Mode: meshconfig.MeshConfig_InboundTrafficPolicy_PASSTHROUGH},
+ LocalityLbSetting: &v1alpha3.LocalityLoadBalancerSetting{
+ Enabled: wrappers.Bool(true),
+ },
+ Certificates: []*meshconfig.Certificate{},
DefaultConfig: proxyConfig,
- RootNamespace: constants.DubboSystemNamespace,
+
+ RootNamespace: constants.DubboSystemNamespace,
+ ProxyListenPort: 15001,
+ ProxyInboundListenPort: 15006,
+ ConnectTimeout: durationpb.New(10 * time.Second),
+ DefaultServiceExportTo: []string{"*"},
+ DefaultVirtualServiceExportTo: []string{"*"},
+ DefaultDestinationRuleExportTo: []string{"*"},
+ DnsRefreshRate: durationpb.New(60 * time.Second),
+ DefaultProviders: &meshconfig.MeshConfig_DefaultProviders{},
}
}
diff --git a/pkg/queue/instance.go b/pkg/queue/instance.go
new file mode 100644
index 00000000..e6f4e98e
--- /dev/null
+++ b/pkg/queue/instance.go
@@ -0,0 +1,170 @@
+// Copyright Istio Authors
+//
+// Licensed 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 queue
+
+import (
+ "sync"
+ "time"
+
+ "go.uber.org/atomic"
+ "k8s.io/apimachinery/pkg/util/rand"
+)
+
+// Task to be performed.
+type Task func() error
+
+type queueTask struct {
+ task Task
+ enqueueTime time.Time
+ startTime time.Time
+}
+
+// Instance of work tickets processed using a rate-limiting loop
+type baseInstance interface {
+ // Push a task.
+ Push(task Task)
+ // Run the loop until a signal on the channel
+ Run(<-chan struct{})
+ // Closed returns a chan that will be signaled when the Instance has stopped processing tasks.
+ Closed() <-chan struct{}
+}
+
+type Instance interface {
+ baseInstance
+ // HasSynced returns true once the queue has synced.
+ // Syncing indicates that all items in the queue *before* Run was called have been processed.
+ HasSynced() bool
+}
+
+type queueImpl struct {
+ delay time.Duration
+ tasks []*queueTask
+ cond *sync.Cond
+ closing bool
+ closed chan struct{}
+ closeOnce *sync.Once
+ // initialSync indicates the queue has initially "synced".
+ initialSync *atomic.Bool
+ id string
+ syncCallback func()
+}
+
+// NewQueue instantiates a queue with a processing function
+func NewQueue(errorDelay time.Duration) Instance {
+ return NewQueueWithID(errorDelay, rand.String(10))
+}
+
+// NewQueue instantiates a queue with a processing function
+func NewWithSync(f func(), name string) Instance {
+ q := NewQueueWithID(time.Second, name)
+ q.(*queueImpl).syncCallback = f
+ return q
+}
+
+func NewQueueWithID(errorDelay time.Duration, name string) Instance {
+ return &queueImpl{
+ delay: errorDelay,
+ tasks: make([]*queueTask, 0),
+ closing: false,
+ closed: make(chan struct{}),
+ closeOnce: &sync.Once{},
+ initialSync: atomic.NewBool(false),
+ cond: sync.NewCond(&sync.Mutex{}),
+ id: name,
+ }
+}
+
+func (q *queueImpl) Push(item Task) {
+ q.cond.L.Lock()
+ defer q.cond.L.Unlock()
+ if !q.closing {
+ q.tasks = append(q.tasks, &queueTask{task: item, enqueueTime: time.Now()})
+ }
+ q.cond.Signal()
+}
+
+func (q *queueImpl) Closed() <-chan struct{} {
+ return q.closed
+}
+
+// get blocks until it can return a task to be processed. If shutdown = true,
+// the processing go routine should stop.
+func (q *queueImpl) get() (task *queueTask, shutdown bool) {
+ q.cond.L.Lock()
+ defer q.cond.L.Unlock()
+ // wait for closing to be set, or a task to be pushed
+ for !q.closing && len(q.tasks) == 0 {
+ q.cond.Wait()
+ }
+
+ if q.closing && len(q.tasks) == 0 {
+ // We must be shutting down.
+ return nil, true
+ }
+ task = q.tasks[0]
+ // Slicing will not free the underlying elements of the array, so explicitly clear them out here
+ q.tasks[0] = nil
+ q.tasks = q.tasks[1:]
+
+ task.startTime = time.Now()
+
+ return task, false
+}
+
+func (q *queueImpl) processNextItem() bool {
+ // Wait until there is a new item in the queue
+ task, shuttingdown := q.get()
+ if shuttingdown {
+ return false
+ }
+
+ // Run the task.
+ if err := task.task(); err != nil {
+ delay := q.delay
+ time.AfterFunc(delay, func() {
+ q.Push(task.task)
+ })
+ }
+ return true
+}
+
+func (q *queueImpl) HasSynced() bool {
+ return q.initialSync.Load()
+}
+
+func (q *queueImpl) Run(stop <-chan struct{}) {
+ defer func() {
+ q.closeOnce.Do(func() {
+ close(q.closed)
+ })
+ }()
+ go func() {
+ <-stop
+ q.cond.L.Lock()
+ q.cond.Signal()
+ q.closing = true
+ q.cond.L.Unlock()
+ }()
+
+ q.Push(func() error {
+ q.initialSync.Store(true)
+ if q.syncCallback != nil {
+ q.syncCallback()
+ }
+ return nil
+ })
+ for q.processNextItem() {
+ }
+}
diff --git a/pkg/queue/util.go b/pkg/queue/util.go
new file mode 100644
index 00000000..f16618fa
--- /dev/null
+++ b/pkg/queue/util.go
@@ -0,0 +1,39 @@
+// Copyright Istio Authors
+//
+// Licensed 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 queue
+
+import (
+ "fmt"
+ "time"
+)
+
+// WaitForClose blocks until the Instance has stopped processing tasks or the timeout expires.
+// If the timeout is zero, it will wait until the queue is done processing.
+// WaitForClose an error if the timeout expires.
+func WaitForClose(q Instance, timeout time.Duration) error {
+ closed := q.Closed()
+ if timeout == 0 {
+ <-closed
+ return nil
+ }
+ timer := time.NewTimer(timeout)
+ defer timer.Stop()
+ select {
+ case <-closed:
+ return nil
+ case <-timer.C:
+ return fmt.Errorf("timeout waiting for queue to close after %v", timeout)
+ }
+}