Repository: kumahq/kuma
Pull request #2349 (2021-07-12 01:28:06)

chore(kuma-cp) use resource types for DataplaneInsight tracking (backport #2324)
This is an automatic backport of pull request #2324 done by [Mergify](https://mergify.io).


---


<details>
<summary>Mergify commands and options</summary>

<br />

More conditions and actions can be found in the [documentation](https://docs.mergify.io/).

You can also trigger Mergify actions by commenting on this pull request:

- `@Mergifyio refresh` will re-evaluate the rules
- `@Mergifyio rebase` will rebase this PR on its base branch
- `@Mergifyio update` will merge the base branch into this PR
- `@Mergifyio backport <destination>` will backport this PR on `<destination>` branch

Additionally, on Mergify [dashboard](https://dashboard.mergify.io/) you can:

- look at your merge queues
- generate the Mergify configuration with the config editor.

Finally, you can contact us on https://mergify.io/
</details>



Patch:
diff --git a/pkg/xds/server/callbacks/dataplane_status_sink.go b/pkg/xds/server/callbacks/dataplane_status_sink.go
index 2d23340e1..7d33841e0 100644
--- a/pkg/xds/server/callbacks/dataplane_status_sink.go
+++ b/pkg/xds/server/callbacks/dataplane_status_sink.go
@@ -7,7 +7,7 @@ import (
 
 	mesh_proto "github.com/kumahq/kuma/api/mesh/v1alpha1"
 	"github.com/kumahq/kuma/pkg/core"
-	mesh_core "github.com/kumahq/kuma/pkg/core/resources/apis/mesh"
+	core_mesh "github.com/kumahq/kuma/pkg/core/resources/apis/mesh"
 	"github.com/kumahq/kuma/pkg/core/resources/manager"
 	core_model "github.com/kumahq/kuma/pkg/core/resources/model"
 	"github.com/kumahq/kuma/pkg/core/resources/store"
@@ -20,29 +20,35 @@ type DataplaneInsightSink interface {
 }
 
 type DataplaneInsightStore interface {
-	Upsert(dataplaneId core_model.ResourceKey, subscription *mesh_proto.DiscoverySubscription) error
+	// Upsert creates or updates the subscription, storing it with
+	// the key dataplaneID. dataplaneType gives the resource type of
+	// the dataplane proxy that has subscribed.
+	Upsert(dataplaneType core_model.ResourceType, dataplaneID core_model.ResourceKey, subscription *mesh_proto.DiscoverySubscription) error
 }
 
 func NewDataplaneInsightSink(
+	dataplaneType core_model.ResourceType,
 	accessor SubscriptionStatusAccessor,
 	newTicker func() *time.Ticker,
 	flushBackoff time.Duration,
 	store DataplaneInsightStore) DataplaneInsightSink {
 	return &dataplaneInsightSink{
-		newTicker:    newTicker,
-		accessor:     accessor,
-		flushBackoff: flushBackoff,
-		store:        store,
+		newTicker:     newTicker,
+		dataplaneType: dataplaneType,
+		accessor:      accessor,
+		flushBackoff:  flushBackoff,
+		store:         store,
 	}
 }
 
 var _ DataplaneInsightSink = &dataplaneInsightSink{}
 
 type dataplaneInsightSink struct {
-	newTicker    func() *time.Ticker
-	accessor     SubscriptionStatusAccessor
-	store        DataplaneInsightStore
-	flushBackoff time.Duration
+	newTicker     func() *time.Ticker
+	dataplaneType core_model.ResourceType
+	accessor      SubscriptionStatusAccessor
+	store         DataplaneInsightStore
+	flushBackoff  time.Duration
 }
 
 func (s *dataplaneInsightSink) Start(stop <-chan struct{}) {
@@ -52,12 +58,13 @@ func (s *dataplaneInsightSink) Start(stop <-chan struct{}) {
 	var lastStoredState *mesh_proto.DiscoverySubscription
 
 	flush := func(closing bool) {
-		dataplaneId, currentState := s.accessor.GetStatus()
+		dataplaneID, currentState := s.accessor.GetStatus()
 		if proto.Equal(currentState, lastStoredState) {
 			return
 		}
+
 		copy := proto.Clone(currentState).(*mesh_proto.DiscoverySubscription)
-		if err := s.store.Upsert(dataplaneId, copy); err != nil {
+		if err := s.store.Upsert(s.dataplaneType, dataplaneID, copy); err != nil {
 			switch {
 			case closing:
 				// When XDS stream is closed, Dataplane Status Tracker executes OnStreamClose which closes stop channel
@@ -67,14 +74,22 @@ func (s *dataplaneInsightSink) Start(stop <-chan struct{}) {
 				// 2) upsert fail because it tries to create a new insight, but there is no Dataplane so ownership returns an error
 				// We could build a synchronous mechanism that waits for Sink to be stopped before moving on to next Callbacks, but this is potentially dangerous
 				// that we could block waiting for storage instead of executing next callbacks.
-				sinkLog.V(1).Info("failed to flush Dataplane status on stream close. It can happen when Dataplane is deleted at the same time", "dataplaneid", dataplaneId, "err", err)
+				sinkLog.V(1).Info("failed to flush Dataplane status on stream close. It can happen when Dataplane is deleted at the same time",
+					"dataplaneid", dataplaneID,
+					"err", err)
 			case store.IsResourceConflict(err):
-				sinkLog.V(1).Info("failed to flush DataplaneInsight because it was updated in other place. Will retry in the next tick", "dataplaneid", dataplaneId)
+				sinkLog.V(1).Info("failed to flush DataplaneInsight because it was updated in other place. Will retry in the next tick",
+					"dataplaneid", dataplaneID)
+			case store.IsResourcePreconditionFailed(err):
+				sinkLog.V(1).Info("failed to flush DataplaneInsight for unsupported resource",
+					"dataplaneid", dataplaneID,
+					"err", err,
+				)
 			default:
-				sinkLog.Error(err, "failed to flush DataplaneInsight", "dataplaneid", dataplaneId)
+				sinkLog.Error(err, "failed to flush DataplaneInsight", "dataplaneid", dataplaneID)
 			}
 		} else {
-			sinkLog.V(1).Info("DataplaneInsight saved", "dataplaneid", dataplaneId, "subscription", currentState)
+			sinkLog.V(1).Info("DataplaneInsight saved", "dataplaneid", dataplaneID, "subscription", currentState)
 			lastStoredState = currentState
 		}
 	}
@@ -105,16 +120,24 @@ type dataplaneInsightStore struct {
 	resManager manager.ResourceManager
 }
 
-func (s *dataplaneInsightStore) Upsert(dataplaneId core_model.ResourceKey, subscription *mesh_proto.DiscoverySubscription) error {
-	if dataplaneId.Mesh == core_model.NoMesh {
-		// hack in order to support ZoneIngress which is the only supported Global-scope ProxyType today
-		return manager.Upsert(s.resManager, dataplaneId, mesh_core.NewZoneIngressInsightResource(), func(resource core_model.Resource) {
-			insight := resource.(*mesh_core.ZoneIngressInsightResource)
+func (s *dataplaneInsightStore) Upsert(
+	dataplaneType core_model.ResourceType,
+	dataplaneID core_model.ResourceKey,
+	subscription *mesh_proto.DiscoverySubscription,
+) error {
+	switch dataplaneType {
+	case core_mesh.ZoneIngressType:
+		return manager.Upsert(s.resManager, dataplaneID, core_mesh.NewZoneIngressInsightResource(), func(resource core_model.Resource) {
+			insight := resource.(*core_mesh.ZoneIngressInsightResource)
+			insight.Spec.UpdateSubscription(subscription)
+		})
+	case core_mesh.DataplaneType:
+		return manager.Upsert(s.resManager, dataplaneID, core_mesh.NewDataplaneInsightResource(), func(resource core_model.Resource) {
+			insight := resource.(*core_mesh.DataplaneInsightResource)
 			insight.Spec.UpdateSubscription(subscription)
 		})
+	default:
+		// Return a designated precondition error since we don't expect other dataplane types.
+		return store.ErrorResourcePreconditionFailed(dataplaneType, dataplaneID.Name, dataplaneID.Mesh)
 	}
-	return manager.Upsert(s.resManager, dataplaneId, mesh_core.NewDataplaneInsightResource(), func(resource core_model.Resource) {
-		insight := resource.(*mesh_core.DataplaneInsightResource)
-		insight.Spec.UpdateSubscription(subscription)
-	})
 }
diff --git a/pkg/xds/server/callbacks/dataplane_status_tracker.go b/pkg/xds/server/callbacks/dataplane_status_tracker.go
index acd000511..181b91d93 100644
--- a/pkg/xds/server/callbacks/dataplane_status_tracker.go
+++ b/pkg/xds/server/callbacks/dataplane_status_tracker.go
@@ -4,12 +4,11 @@ import (
 	"context"
 	"sync"
 
-	"google.golang.org/protobuf/types/known/structpb"
-
 	"github.com/golang/protobuf/proto"
 
 	mesh_proto "github.com/kumahq/kuma/api/mesh/v1alpha1"
 	"github.com/kumahq/kuma/pkg/core"
+	core_mesh "github.com/kumahq/kuma/pkg/core/resources/apis/mesh"
 	core_model "github.com/kumahq/kuma/pkg/core/resources/model"
 	core_runtime "github.com/kumahq/kuma/pkg/core/runtime"
 	core_xds "github.com/kumahq/kuma/pkg/core/xds"
@@ -28,10 +27,12 @@ type SubscriptionStatusAccessor interface {
 	GetStatus() (core_model.ResourceKey, *mesh_proto.DiscoverySubscription)
 }
 
-type DataplaneInsightSinkFactoryFunc = func(SubscriptionStatusAccessor) DataplaneInsightSink
+type DataplaneInsightSinkFactoryFunc = func(core_model.ResourceType, SubscriptionStatusAccessor) DataplaneInsightSink
 
-func NewDataplaneStatusTracker(runtimeInfo core_runtime.RuntimeInfo,
-	createStatusSink DataplaneInsightSinkFactoryFunc) DataplaneStatusTracker {
+func NewDataplaneStatusTracker(
+	runtimeInfo core_runtime.RuntimeInfo,
+	createStatusSink DataplaneInsightSinkFactoryFunc,
+) DataplaneStatusTracker {
 	return &dataplaneStatusTracker{
 		runtimeInfo:      runtimeInfo,
 		createStatusSink: createStatusSink,
@@ -114,15 +115,32 @@ func (c *dataplaneStatusTracker) OnStreamRequest(streamID int64, req util_xds.Di
 	state.mu.Lock() // write access to the per Dataplane info
 	defer state.mu.Unlock()
 
-	// infer Dataplane id
 	if state.dataplaneId == (core_model.ResourceKey{}) {
+		var dpType core_model.ResourceType
+		md := core_xds.DataplaneMetadataFromXdsMetadata(req.Metadata())
+
+		// If the dataplane was started with a resource YAML, then it
+		// will be serialized in the node metadata and we would know
+		// the underlying type directly. Since that is optional, we
+		// can't depend on it here, so we map from the proxy type,
+		// which is guaranteed.
+		switch md.GetProxyType() {
+		case mesh_proto.IngressProxyType:
+			dpType = core_mesh.ZoneIngressType
+		case mesh_proto.DataplaneProxyType:
+			dpType = core_mesh.DataplaneType
+		}
+
+		// Infer the Dataplane ID.
 		if proxyId, err := core_xds.ParseProxyIdFromString(req.NodeId()); err == nil {
 			state.dataplaneId = proxyId.ToResourceKey()
-			if err := readVersion(req.Metadata(), state.subscription.Version); err != nil {
+			if md.GetVersion() != nil {
+				state.subscription.Version = md.GetVersion()
+			} else {
 				statusTrackerLog.Error(err, "failed to extract version out of the Envoy metadata", "streamid", streamID, "metadata", req.Metadata())
 			}
-			// kick off async Dataplane status flusher
-			go c.createStatusSink(state).Start(state.stop)
+			// Kick off the async Dataplane status flusher.
+			go c.createStatusSink(dpType, state).Start(state.stop)
 		} else {
 			statusTrackerLog.Error(err, "failed to parse Dataplane Id out of DiscoveryRequest", "streamid", streamID, "req", req)
 		}
@@ -180,17 +198,3 @@ func (s *streamState) GetStatus() (core_model.ResourceKey, *mesh_proto.Discovery
 func (s *streamState) Close() {
 	close(s.stop)
 }
-
-func readVersion(metadata *structpb.Struct, version *mesh_proto.Version) error {
-	if metadata == nil {
-		return nil
-	}
-	rawVersion := metadata.Fields["version"].GetStructValue()
-	if rawVersion != nil {
-		err := util_proto.ToTyped(rawVersion, version)
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
diff --git a/pkg/xds/server/v3/components.go b/pkg/xds/server/v3/components.go
index 41cb7f147..7974153ba 100644
--- a/pkg/xds/server/v3/components.go
+++ b/pkg/xds/server/v3/components.go
@@ -8,6 +8,7 @@ import (
 	envoy_server "github.com/envoyproxy/go-control-plane/pkg/server/v3"
 
 	"github.com/kumahq/kuma/pkg/core"
+	core_model "github.com/kumahq/kuma/pkg/core/resources/model"
 	core_runtime "github.com/kumahq/kuma/pkg/core/runtime"
 	v3 "github.com/kumahq/kuma/pkg/core/xds/v3"
 	util_xds "github.com/kumahq/kuma/pkg/util/xds"
@@ -95,14 +96,16 @@ func DefaultIngressReconciler(rt core_runtime.Runtime, xdsContext v3.XdsContext)
 }
 
 func DefaultDataplaneStatusTracker(rt core_runtime.Runtime) xds_callbacks.DataplaneStatusTracker {
-	return xds_callbacks.NewDataplaneStatusTracker(rt, func(accessor xds_callbacks.SubscriptionStatusAccessor) xds_callbacks.DataplaneInsightSink {
-		return xds_callbacks.NewDataplaneInsightSink(
-			accessor,
-			func() *time.Ticker {
-				return time.NewTicker(rt.Config().XdsServer.DataplaneStatusFlushInterval)
-			},
-			rt.Config().XdsServer.DataplaneStatusFlushInterval/10,
-			xds_callbacks.NewDataplaneInsightStore(rt.ResourceManager()),
-		)
-	})
+	return xds_callbacks.NewDataplaneStatusTracker(rt,
+		func(dataplaneType core_model.ResourceType, accessor xds_callbacks.SubscriptionStatusAccessor) xds_callbacks.DataplaneInsightSink {
+			return xds_callbacks.NewDataplaneInsightSink(
+				dataplaneType,
+				accessor,
+				func() *time.Ticker {
+					return time.NewTicker(rt.Config().XdsServer.DataplaneStatusFlushInterval)
+				},
+				rt.Config().XdsServer.DataplaneStatusFlushInterval/10,
+				xds_callbacks.NewDataplaneInsightStore(rt.ResourceManager()),
+			)
+		})
 }


Test patch:
diff --git a/pkg/test/matchers/proto.go b/pkg/test/matchers/proto.go
index 2e5e1502a..590310b24 100644
--- a/pkg/test/matchers/proto.go
+++ b/pkg/test/matchers/proto.go
@@ -1,11 +1,10 @@
 package matchers
 
 import (
-	"errors"
-
 	"github.com/golang/protobuf/proto"
 	"github.com/onsi/gomega/format"
 	"github.com/onsi/gomega/types"
+	"github.com/pkg/errors"
 
 	util_proto "github.com/kumahq/kuma/pkg/util/proto"
 )
@@ -56,13 +55,13 @@ func (p *ProtoMatcher) yamls(actual interface{}) (string, string, error) {
 	actualProto := actual.(proto.Message)
 	actualYAML, err := util_proto.ToYAML(actualProto)
 	if err != nil {
-		return "", "", errors.New("Proto are not equal (could not parse actual proto to present differences)")
+		return "", "", errors.Errorf("Proto are not equal (could not convert to YAML: %s)", err)
 	}
 
 	expectedProto := p.Expected.(proto.Message)
 	expectedYAML, err := util_proto.ToYAML(expectedProto)
 	if err != nil {
-		return "", "", errors.New("Proto are not equal (could not parse expected proto to present differences)")
+		return "", "", errors.Errorf("Proto are not equal (could not convert to YAML: %s)", err)
 	}
 	return string(actualYAML), string(expectedYAML), nil
 }
diff --git a/pkg/xds/server/callbacks/dataplane_status_sink_test.go b/pkg/xds/server/callbacks/dataplane_status_sink_test.go
index d3de31436..f3b15b356 100644
--- a/pkg/xds/server/callbacks/dataplane_status_sink_test.go
+++ b/pkg/xds/server/callbacks/dataplane_status_sink_test.go
@@ -13,13 +13,10 @@ import (
 	"github.com/golang/protobuf/proto"
 
 	mesh_proto "github.com/kumahq/kuma/api/mesh/v1alpha1"
-	mesh_core "github.com/kumahq/kuma/pkg/core/resources/apis/mesh"
-
+	core_mesh "github.com/kumahq/kuma/pkg/core/resources/apis/mesh"
 	core_model "github.com/kumahq/kuma/pkg/core/resources/model"
 	core_store "github.com/kumahq/kuma/pkg/core/resources/store"
-
 	memory_resources "github.com/kumahq/kuma/pkg/plugins/resources/memory"
-
 	util_proto "github.com/kumahq/kuma/pkg/util/proto"
 )
 
@@ -58,7 +55,7 @@ var _ = Describe("DataplaneInsightSink", func() {
 			var latestUpsert *DataplaneInsightUpsert
 
 			// given
-			sink := callbacks.NewDataplaneInsightSink(accessor, func() *time.Ticker { return ticker }, 1*time.Millisecond, recorder)
+			sink := callbacks.NewDataplaneInsightSink(core_mesh.DataplaneType, accessor, func() *time.Ticker { return ticker }, 1*time.Millisecond, recorder)
 			go sink.Start(stop)
 
 			// when
@@ -136,7 +133,7 @@ var _ = Describe("DataplaneInsightSink", func() {
 
 		BeforeEach(func() {
 			store = core_store.NewPaginationStore(memory_resources.NewStore())
-			err := store.Create(context.Background(), mesh_core.NewMeshResource(), core_store.CreateByKey(core_model.DefaultMesh, core_model.NoMesh))
+			err := store.Create(context.Background(), core_mesh.NewMeshResource(), core_store.CreateByKey(core_model.DefaultMesh, core_model.NoMesh))
 			Expect(err).ToNot(HaveOccurred())
 		})
 
@@ -149,14 +146,15 @@ var _ = Describe("DataplaneInsightSink", func() {
 				ConnectTime:            util_proto.MustTimestampProto(t0),
 				Status:                 mesh_proto.NewSubscriptionStatus(),
 			}
-			dataplaneInsight := mesh_core.NewDataplaneInsightResource()
+			dataplaneType := core_mesh.DataplaneType
+			dataplaneInsight := core_mesh.NewDataplaneInsightResource()
 			lastSeenVersion := ""
 
 			// given
 			statusStore := callbacks.NewDataplaneInsightStore(manager.NewResourceManager(store))
 
 			// when
-			err := statusStore.Upsert(key, proto.Clone(subscription).(*mesh_proto.DiscoverySubscription))
+			err := statusStore.Upsert(dataplaneType, key, proto.Clone(subscription).(*mesh_proto.DiscoverySubscription))
 			// then
 			Expect(err).ToNot(HaveOccurred())
 			// and
@@ -190,7 +188,7 @@ var _ = Describe("DataplaneInsightSink", func() {
 			subscription.Status.Lds.ResponsesSent += 1
 			subscription.Status.Total.ResponsesSent += 1
 			// and
-			err = statusStore.Upsert(key, proto.Clone(subscription).(*mesh_proto.DiscoverySubscription))
+			err = statusStore.Upsert(dataplaneType, key, proto.Clone(subscription).(*mesh_proto.DiscoverySubscription))
 			// then
 			Expect(err).ToNot(HaveOccurred())
 			// and
@@ -247,7 +245,7 @@ type DataplaneInsightStoreRecorder struct {
 	Upserts chan DataplaneInsightUpsert
 }
 
-func (s *DataplaneInsightStoreRecorder) Upsert(dataplaneId core_model.ResourceKey, subscription *mesh_proto.DiscoverySubscription) error {
+func (s *DataplaneInsightStoreRecorder) Upsert(resourceType core_model.ResourceType, dataplaneId core_model.ResourceKey, subscription *mesh_proto.DiscoverySubscription) error {
 	s.Upserts <- DataplaneInsightUpsert{dataplaneId, subscription}
 	return nil
 }
diff --git a/pkg/xds/server/callbacks/dataplane_status_tracker_test.go b/pkg/xds/server/callbacks/dataplane_status_tracker_test.go
index 0f56030f1..f3c30169c 100644
--- a/pkg/xds/server/callbacks/dataplane_status_tracker_test.go
+++ b/pkg/xds/server/callbacks/dataplane_status_tracker_test.go
@@ -64,7 +64,7 @@ var _ = Describe("DataplaneStatusTracker", func() {
 	})
 
 	BeforeEach(func() {
-		tracker = NewDataplaneStatusTracker(runtimeInfo, func(accessor SubscriptionStatusAccessor) DataplaneInsightSink {
+		tracker = NewDataplaneStatusTracker(runtimeInfo, func(dataplaneType core_model.ResourceType, accessor SubscriptionStatusAccessor) DataplaneInsightSink {
 			return DataplaneInsightSinkFunc(func(<-chan struct{}) {})
 		})
 		callbacks = v2.AdaptCallbacks(tracker)
@@ -671,6 +671,7 @@ var _ = Describe("DataplaneStatusTracker", func() {
 			// and
 			accessor, _ := tracker.GetStatusAccessor(streamID)
 			_, sub := accessor.GetStatus()
+			Expect(sub.GetVersion()).ToNot(BeNil())
 			Expect(sub.GetVersion()).To(MatchProto(mesh_proto.NewVersion()))
 		},
 		Entry("when version is a nil struct", versionTestCase{