Merge pull request #279 from meshery/feat/discovery/cr

Add ability to discover new custom resources.

Makefile

@@ -43,6 +43,7 @@ docker-run:
 PHONY: nats
 ## Runs a local instance of NATS server in detached mode
 nats:
+	(docker rm -f nats) || true
 	docker run --name nats --rm -p 4222:4222 -p 8222:8222 -d nats --http_port 8222 
 
 #-----------------------------------------------------------------------------

go.mod

@@ -12,7 +12,7 @@ toolchain go1.21.1
 require (
 	github.com/buger/jsonparser v1.1.1
 	github.com/google/uuid v1.4.0
-	github.com/layer5io/meshkit v0.6.88
+	github.com/layer5io/meshkit v0.7.2
 	github.com/myntra/pipeline v0.0.0-20180618182531-2babf4864ce8
 	github.com/spf13/viper v1.17.0
 	golang.org/x/exp v0.0.0-20230905200255-921286631fa9
@@ -45,7 +45,7 @@ require (
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/docker/cli v24.0.6+incompatible // indirect
 	github.com/docker/distribution v2.8.2+incompatible // indirect
-	github.com/docker/docker v24.0.6+incompatible // indirect
+	github.com/docker/docker v24.0.7+incompatible // indirect
 	github.com/docker/docker-credential-helpers v0.7.0 // indirect
 	github.com/docker/go-connections v0.4.0 // indirect
 	github.com/docker/go-metrics v0.0.1 // indirect
@@ -162,7 +162,7 @@ require (
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	gorm.io/driver/postgres v1.5.3 // indirect
 	gorm.io/driver/sqlite v1.5.4 // indirect
-	helm.sh/helm/v3 v3.13.0 // indirect
+	helm.sh/helm/v3 v3.13.2 // indirect
 	k8s.io/apiextensions-apiserver v0.28.3 // indirect
 	k8s.io/apiserver v0.28.3 // indirect
 	k8s.io/cli-runtime v0.28.3 // indirect

go.sum

@@ -121,8 +121,8 @@ github.com/docker/cli v24.0.6+incompatible h1:fF+XCQCgJjjQNIMjzaSmiKJSCcfcXb3TWT
 github.com/docker/cli v24.0.6+incompatible/go.mod h1:JLrzqnKDaYBop7H2jaqPtU4hHvMKP+vjCwu2uszcLI8=
 github.com/docker/distribution v2.8.2+incompatible h1:T3de5rq0dB1j30rp0sA2rER+m322EBzniBPB6ZIzuh8=
 github.com/docker/distribution v2.8.2+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
-github.com/docker/docker v24.0.6+incompatible h1:hceabKCtUgDqPu+qm0NgsaXf28Ljf4/pWFL7xjWWDgE=
-github.com/docker/docker v24.0.6+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
+github.com/docker/docker v24.0.7+incompatible h1:Wo6l37AuwP3JaMnZa226lzVXGA3F9Ig1seQen0cKYlM=
+github.com/docker/docker v24.0.7+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/docker-credential-helpers v0.7.0 h1:xtCHsjxogADNZcdv1pKUHXryefjlVRqWqIhk/uXJp0A=
 github.com/docker/docker-credential-helpers v0.7.0/go.mod h1:rETQfLdHNT3foU5kuNkFR1R1V12OJRRO5lzt2D1b5X0=
 github.com/docker/go-connections v0.4.0 h1:El9xVISelRB7BuFusrZozjnkIM5YnzCViNKohAFqRJQ=
@@ -349,8 +349,8 @@ github.com/lann/builder v0.0.0-20180802200727-47ae307949d0 h1:SOEGU9fKiNWd/HOJuq
 github.com/lann/builder v0.0.0-20180802200727-47ae307949d0/go.mod h1:dXGbAdH5GtBTC4WfIxhKZfyBF/HBFgRZSWwZ9g/He9o=
 github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0 h1:P6pPBnrTSX3DEVR4fDembhRWSsG5rVo6hYhAB/ADZrk=
 github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0/go.mod h1:vmVJ0l/dxyfGW6FmdpVm2joNMFikkuWg0EoCKLGUMNw=
-github.com/layer5io/meshkit v0.6.88 h1:JvL/lzIorXxGR5a3BKo1nSzHmtssh/HstAY4yYc9aqM=
-github.com/layer5io/meshkit v0.6.88/go.mod h1:8dujy2ZSjnmtI7bFhMhnBUwFSlkPDyOowp6iLvxMslk=
+github.com/layer5io/meshkit v0.7.2 h1:MseAfNKbGJPR/cD0cdySyV8qOSsyYgLYKoNmGEoXEjU=
+github.com/layer5io/meshkit v0.7.2/go.mod h1:E5Zmn+CeBiGsnM2vsAsXKQtS39JDB9E91V9KRJMuShU=
 github.com/lib/pq v1.2.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
 github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
 github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
@@ -915,8 +915,8 @@ gorm.io/gorm v1.25.5 h1:zR9lOiiYf09VNh5Q1gphfyia1JpiClIWG9hQaxB/mls=
 gorm.io/gorm v1.25.5/go.mod h1:hbnx/Oo0ChWMn1BIhpy1oYozzpM15i4YPuHDmfYtwg8=
 gotest.tools/v3 v3.4.0 h1:ZazjZUfuVeZGLAmlKKuyv3IKP5orXcwtOwDQH6YVr6o=
 gotest.tools/v3 v3.4.0/go.mod h1:CtbdzLSsqVhDgMtKsx03ird5YTGB3ar27v0u/yKBW5g=
-helm.sh/helm/v3 v3.13.0 h1:XPJKIU30K4JTQ6VX/6e0hFAmEIonYa8E7wx5aqv4xOc=
-helm.sh/helm/v3 v3.13.0/go.mod h1:2PBEKsMWKLVZTojUOqMS3Eadv5mP43FBWrRgLNkNm9Y=
+helm.sh/helm/v3 v3.13.2 h1:IcO9NgmmpetJODLZhR3f3q+6zzyXVKlRizKFwbi7K8w=
+helm.sh/helm/v3 v3.13.2/go.mod h1:GIHDwZggaTGbedevTlrQ6DB++LBN6yuQdeGj0HNaDx0=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190106161140-3f1c8253044a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190418001031-e561f6794a2a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

internal/channels/system.go

@@ -37,3 +37,7 @@ type ReSyncChannel chan struct{}
 func (ch ReSyncChannel) Stop() {
 	<-ch
 }
+
+func (ch ReSyncChannel) ReSyncInformer() {
+	ch <- struct{}{}
+}

internal/config/default_config.go

@@ -12,12 +12,14 @@ var (
 		"startedat": time.Now().String(),
 	}
 
+	DefaultPublishingSubject = "meshery.meshsync.core"
+
 	Pipelines = map[string]PipelineConfigs{
 		GlobalResourceKey: []PipelineConfig{
 			// Core Resources
 			{
 				Name:      "namespaces.v1.",
-				PublishTo: "meshery.meshsync.core",
+				PublishTo: DefaultPublishingSubject,
 			},
 			{
 				Name:      "configmaps.v1.",
@@ -39,99 +41,78 @@ var (
 				Name:      "persistentvolumeclaims.v1.",
 				PublishTo: "meshery.meshsync.core",
 			},
-			{
-				Name: "prometheuses.v1.monitoring.coreos.com",
-				PublishTo: "meshery.meshsync.core",
-			},
-			{
-				Name: "grafanas.v1beta1.grafana.integreatly.org",
-				PublishTo: "meshery.meshsync.core",
-			},
 		},
 		LocalResourceKey: []PipelineConfig{
 			// Core Resources
 			{
 				Name:      "replicasets.v1.apps",
-				PublishTo: "meshery.meshsync.core",
+				PublishTo: DefaultPublishingSubject,
 			},
 			{
 				Name:      "pods.v1.",
-				PublishTo: "meshery.meshsync.core",
+				PublishTo: DefaultPublishingSubject,
 			},
 			{
 				Name:      "services.v1.",
-				PublishTo: "meshery.meshsync.core",
+				PublishTo: DefaultPublishingSubject,
 			},
 			{
 				Name:      "deployments.v1.apps",
-				PublishTo: "meshery.meshsync.core",
+				PublishTo: DefaultPublishingSubject,
 			},
 			{
 				Name:      "statefulsets.v1.apps",
-				PublishTo: "meshery.meshsync.core",
+				PublishTo: DefaultPublishingSubject,
 			},
 			{
 				Name:      "daemonsets.v1.apps",
-				PublishTo: "meshery.meshsync.core",
+				PublishTo: DefaultPublishingSubject,
 			},
 			//Added Ingress support
 			{
 				Name:      "ingresses.v1.networking.k8s.io",
-				PublishTo: "meshery.meshsync.core",
+				PublishTo: DefaultPublishingSubject,
 			},
-			//Added endpoint support
+			// Added endpoint support
 			{
 				Name:      "endpoints.v1.",
-				PublishTo: "meshery.meshsync.core",
+				PublishTo: DefaultPublishingSubject,
 			},
 			//Added endpointslice support
 			{
 				Name:      "endpointslices.v1.discovery.k8s.io",
-				PublishTo: "meshery.meshsync.core",
+				PublishTo: DefaultPublishingSubject,
 			},
-			//Added cronJob support
+			// Added cronJob support
 			{
 				Name:      "cronjobs.v1.batch",
-				PublishTo: "meshery.meshsync.core",
+				PublishTo: DefaultPublishingSubject,
 			},
 			//Added ReplicationController support
 			{
 				Name:      "replicationcontrollers.v1.",
-				PublishTo: "meshery.meshsync.core",
+				PublishTo: DefaultPublishingSubject,
 			},
 			//Added storageClass support
 			{
 				Name:      "storageclasses.v1.storage.k8s.io",
-				PublishTo: "meshery.meshsync.core",
+				PublishTo: DefaultPublishingSubject,
 			},
 			//Added ClusterRole support
 			{
 				Name:      "clusterroles.v1.rbac.authorization.k8s.io",
-				PublishTo: "meshery.meshsync.core",
+				PublishTo: DefaultPublishingSubject,
 			},
 			//Added VolumeAttachment support
 			{
 				Name:      "volumeattachments.v1.storage.k8s.io",
-				PublishTo: "meshery.meshsync.core",
+				PublishTo: DefaultPublishingSubject,
 			},
 			//Added apiservice support
 			{
 				Name:      "apiservices.v1.apiregistration.k8s.io",
-				PublishTo: "meshery.meshsync.core",
+				PublishTo: DefaultPublishingSubject,
 			},
-			// Istio Resources
-			// {
-			// 	Name:      "virtualservices.v1beta1.networking.istio.io",
-			// 	PublishTo: "meshery.meshsync.istio",
-			// },
-			// {
-			// 	Name:      "gateways.v1beta1.networking.istio.io",
-			// 	PublishTo: "meshery.meshsync.istio",
-			// },
-			// {
-			// 	Name:      "destinationrules.v1beta1.networking.istio.io",
-			// 	PublishTo: "meshery.meshsync.istio",
-			// },
 		},
 	}
 

internal/config/types.go

@@ -1,5 +1,7 @@
 package config
 
+import "golang.org/x/exp/slices"
+
 const (
 	ServerKey                 = "server-config"
 	PipelineNameKey           = "meshsync-pipeline"
@@ -19,6 +21,20 @@ const (
 
 type PipelineConfigs []PipelineConfig
 
+func(p PipelineConfigs) Add(pc PipelineConfig) PipelineConfigs {
+	p = append(p, pc)
+	return p
+}
+
+func(p PipelineConfigs) Delete(pc PipelineConfig) PipelineConfigs {
+	for index, pipelineConfig := range p {
+		if pipelineConfig.Name == pc.Name {
+			p = slices.Delete[PipelineConfigs](p, index, index + 1)
+			break
+		}
+	}
+	return p
+}
 type PipelineConfig struct {
 	Name      string   `json:"name" yaml:"name"`
 	PublishTo string   `json:"publish-to" yaml:"publish-to"`

internal/pipeline/handlers.go

@@ -12,8 +12,8 @@ import (
 	"k8s.io/client-go/tools/cache"
 )
 
-func (ri *RegisterInformer) registerHandlers(s cache.SharedIndexInformer) {
-	handlers := cache.ResourceEventHandlerFuncs{
+func(ri *RegisterInformer) GetEventHandlers() cache.ResourceEventHandlerFuncs {
+	return cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			err := ri.publishItem(obj.(*unstructured.Unstructured), broker.Add, ri.config)
 			if err != nil {
@@ -66,7 +66,10 @@ func (ri *RegisterInformer) registerHandlers(s cache.SharedIndexInformer) {
 			ri.log.Info("Received DELETE event for: ", obj.(*unstructured.Unstructured).GetName(), "/", obj.(*unstructured.Unstructured).GetNamespace(), " of kind: ", obj.(*unstructured.Unstructured).GroupVersionKind().Kind)
 		},
 	}
-	s.AddEventHandler(handlers) // nolint
+}
+
+func (ri *RegisterInformer) registerHandlers(s cache.SharedIndexInformer) {
+	s.AddEventHandler(ri.GetEventHandlers()) // nolint
 }
 
 func (ri *RegisterInformer) publishItem(obj *unstructured.Unstructured, evtype broker.EventType, config internalconfig.PipelineConfig) error {

main.go

@@ -116,6 +116,8 @@ func main() {
 		os.Exit(1)
 	}
 
+	go meshsyncHandler.WatchCRDs()
+
 	go meshsyncHandler.Run()
 	go meshsyncHandler.ListenToRequests()