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 · d528d46 · d528d46
1 parent 27614ed
commit d528d46
Show file tree

Hide file tree

Showing 28 changed files with 491 additions and 1,370 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
@@ -3,7 +3,6 @@ package server
 import (
 	"context"
 	"sync/atomic"
-
 	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"
@@ -14,8 +13,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 +132,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 +166,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)
 		}
 	}