chore(mads/hds): remove custom cache implementation

In mads and hds we were using a custom implementation of a cache. This was unfortunate as go-control-plane has a perfectly working one. The only feature on this cache was the ability to long poll. This was only required in mads and we've changed the handler of mads to make this handling outside of the cache. Signed-off-by: Charly Molter <[email protected]>
kumahq · Sep 27, 2024 · 322eb33 · 322eb33
1 parent 27614ed
commit 322eb33
Show file tree

Hide file tree

Showing 28 changed files with 492 additions and 1,369 deletions.
diff --git a/api/observability/v1/mads_helpers.go b/api/observability/v1/mads_helpers.go
@@ -0,0 +1,15 @@
+package v1
+
+import "fmt"
+
+// GetName returns a name including the mesh which is used for ResourceName in MADS snapshot cache
+func (x *MonitoringAssignment) GetName() string {
+	if x != nil {
+		dpName := ""
+		if len(x.Targets) > 0 {
+			dpName = x.Targets[0].GetName()
+		}
+		return fmt.Sprintf("/meshes/%s/dataplanes/%s", x.GetMesh(), dpName)
+	}
+	return ""
+}
diff --git a/pkg/hds/cache/cache_suite_test.go b/pkg/hds/cache/cache_suite_test.go
diff --git a/pkg/hds/cache/snapshot.go b/pkg/hds/cache/snapshot.go
diff --git a/pkg/hds/cache/snapshot_test.go b/pkg/hds/cache/snapshot_test.go
diff --git a/pkg/hds/components.go b/pkg/hds/components.go
@@ -6,6 +6,7 @@ import (
 
 	envoy_core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	envoy_service_health "github.com/envoyproxy/go-control-plane/envoy/service/health/v3"
+	envoy_cache "github.com/envoyproxy/go-control-plane/pkg/cache/v3"
 
 	config_core "github.com/kumahq/kuma/pkg/config/core"
 	"github.com/kumahq/kuma/pkg/core"
@@ -17,7 +18,6 @@ import (
 	hds_server "github.com/kumahq/kuma/pkg/hds/server"
 	"github.com/kumahq/kuma/pkg/hds/tracker"
 	util_xds "github.com/kumahq/kuma/pkg/util/xds"
-	util_xds_v3 "github.com/kumahq/kuma/pkg/util/xds/v3"
 )
 
 var hdsServerLog = core.Log.WithName("hds-server")
@@ -30,7 +30,7 @@ func Setup(rt core_runtime.Runtime) error {
 		return nil
 	}
 
-	snapshotCache := util_xds_v3.NewSnapshotCache(false, hasher{}, util_xds.NewLogger(hdsServerLog))
+	snapshotCache := envoy_cache.NewSnapshotCache(false, hasher{}, util_xds.NewLogger(hdsServerLog))
 
 	callbacks, err := DefaultCallbacks(rt, snapshotCache)
 	if err != nil {
@@ -44,7 +44,7 @@ func Setup(rt core_runtime.Runtime) error {
 	return nil
 }
 
-func DefaultCallbacks(rt core_runtime.Runtime, cache util_xds_v3.SnapshotCache) (hds_callbacks.Callbacks, error) {
+func DefaultCallbacks(rt core_runtime.Runtime, cache envoy_cache.SnapshotCache) (hds_callbacks.Callbacks, error) {
 	metrics, err := hds_metrics.NewMetrics(rt.Metrics())
 	if err != nil {
 		return nil, err

diff --git a/pkg/hds/server/server.go b/pkg/hds/server/server.go
@@ -14,8 +14,8 @@ import (
 	"google.golang.org/grpc/status"
 
 	mesh_proto "github.com/kumahq/kuma/api/mesh/v1alpha1"
-	hds_cache "github.com/kumahq/kuma/pkg/hds/cache"
 	hds_callbacks "github.com/kumahq/kuma/pkg/hds/callbacks"
+	v3 "github.com/kumahq/kuma/pkg/hds/v3"
 	util_proto "github.com/kumahq/kuma/pkg/util/proto"
 )
 
@@ -133,10 +133,9 @@ func (s *server) process(stream Stream, reqOrRespCh chan *envoy_service_health.H
 				watchCancellation()
 			}
 			watchCancellation = s.cache.CreateWatch(&envoy_cache.Request{
-				Node:          node,
-				TypeUrl:       hds_cache.HealthCheckSpecifierType,
-				ResourceNames: []string{"hcs"},
-				VersionInfo:   lastVersion,
+				Node:        node,
+				TypeUrl:     v3.HealthCheckSpecifierType,
+				VersionInfo: lastVersion,
 			}, envoy_stream.NewStreamState(false, nil), responseChan)
 		case reqOrResp, more := <-reqOrRespCh:
 			if !more {
@@ -168,10 +167,9 @@ func (s *server) process(stream Stream, reqOrRespCh chan *envoy_service_health.H
 				watchCancellation()
 			}
 			watchCancellation = s.cache.CreateWatch(&envoy_cache.Request{
-				Node:          node,
-				TypeUrl:       hds_cache.HealthCheckSpecifierType,
-				ResourceNames: []string{"hcs"},
-				VersionInfo:   lastVersion,
+				Node:        node,
+				TypeUrl:     v3.HealthCheckSpecifierType,
+				VersionInfo: lastVersion,
 			}, envoy_stream.NewStreamState(false, nil), responseChan)
 		}
 	}

diff --git a/pkg/hds/tracker/callbacks.go b/pkg/hds/tracker/callbacks.go
@@ -7,6 +7,7 @@ import (
 
 	envoy_core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	envoy_service_health "github.com/envoyproxy/go-control-plane/envoy/service/health/v3"
+	envoy_cache "github.com/envoyproxy/go-control-plane/pkg/cache/v3"
 	"github.com/go-logr/logr"
 	"github.com/pkg/errors"
 
@@ -47,9 +48,9 @@ func NewCallbacks(
 	log logr.Logger,
 	resourceManager manager.ResourceManager,
 	readOnlyResourceManager manager.ReadOnlyResourceManager,
-	cache util_xds_v3.SnapshotCache,
+	cache envoy_cache.SnapshotCache,
 	config *dp_server.HdsConfig,
-	hasher util_xds_v3.NodeHash,
+	hasher envoy_cache.NodeHash,
 	metrics *hds_metrics.Metrics,
 	defaultAdminPort uint32,
 ) hds_callbacks.Callbacks {
@@ -143,7 +144,7 @@ func (t *tracker) newWatchdog(node *envoy_core.Node) util_xds_v3.Watchdog {
 			t.log.Error(err, "OnTick() failed")
 		},
 		OnStop: func() {
-			if err := t.reconciler.Clear(node); err != nil {
+			if err := t.reconciler.Clear(context.Background(), node); err != nil {
 				t.log.Error(err, "OnTick() failed")
 			}
 		},