fix(kds): detect properly hanging stream for the same zone

api/system/v1alpha1/zone_insight.proto

@@ -25,9 +25,12 @@ message ZoneInsight {
   repeated KDSSubscription subscriptions = 1;
 
   // Statistics about Envoy Admin Streams
+  // Deprecated: use kds_streams instead.
   EnvoyAdminStreams envoy_admin_streams = 2;
 
   HealthCheck health_check = 3;
+  // Information about kds streams that are estabilished between global and zone
+  KDSStreams kds_streams = 4;
 }
 
 message EnvoyAdminStreams {
@@ -39,6 +42,26 @@ message EnvoyAdminStreams {
   string clusters_global_instance_id = 3;
 }
 
+message KDSStreams {
+  // Details of stream that handles Clusters stream.
+  KDSStream clusters = 1;
+  // Details of stream that handles XDS Config Dump stream.
+  KDSStream config_dump = 2;
+  // Details of stream that handles Stats stream.
+  KDSStream stats = 3;
+  // Details of stream that handles global to zone resource sync stream.
+  KDSStream global_to_zone = 4;
+  // Details of stream that handles zone to global resource sync stream.
+  KDSStream zone_to_global = 5;
+}
+
+message KDSStream {
+  // Global instance ID that handles the stream.
+  string global_instance_id = 1;
+  // Time when the stream was open.
+  google.protobuf.Timestamp connect_time = 2;
+}
+
 // KDSSubscription describes a single KDS subscription
 // created by a Zone to the Global.
 // Ideally, there should be only one such subscription per Zone lifecycle.

pkg/intercp/envoyadmin/forwarding_kds_client.go

@@ -168,15 +168,27 @@ func (f *forwardingKdsEnvoyAdminClient) globalInstanceID(ctx context.Context, zo
 	if !zoneInsightRes.Spec.IsOnline() {
 		return "", &ZoneOfflineError{rpcName: rpcName}
 	}
-	streams := zoneInsightRes.Spec.GetEnvoyAdminStreams()
+	streams := zoneInsightRes.Spec.GetKdsStreams()
 	var globalInstanceID string
 	switch rpcName {
 	case service.ConfigDumpRPC:
-		globalInstanceID = streams.GetConfigDumpGlobalInstanceId()
+		if streams.GetConfigDump() != nil {
+			globalInstanceID = streams.GetConfigDump().GetGlobalInstanceId()
+		} else {
+			globalInstanceID = zoneInsightRes.Spec.GetEnvoyAdminStreams().GetConfigDumpGlobalInstanceId()
+		}
 	case service.StatsRPC:
-		globalInstanceID = streams.GetStatsGlobalInstanceId()
+		if streams.GetStats() != nil {
+			globalInstanceID = streams.GetStats().GetGlobalInstanceId()
+		} else {
+			globalInstanceID = zoneInsightRes.Spec.GetEnvoyAdminStreams().GetStatsGlobalInstanceId()
+		}
 	case service.ClustersRPC:
-		globalInstanceID = streams.GetClustersGlobalInstanceId()
+		if streams.GetClusters() != nil {
+			globalInstanceID = streams.GetClusters().GetGlobalInstanceId()
+		} else {
+			globalInstanceID = zoneInsightRes.Spec.GetEnvoyAdminStreams().GetClustersGlobalInstanceId()
+		}
 	default:
 		return "", errors.Errorf("invalid operation %s", rpcName)
 	}

pkg/kds/global/components.go

@@ -153,6 +153,9 @@ func Setup(rt runtime.Runtime) error {
 			rt.KDSContext().GlobalServerFiltersV2,
 			rt.Extensions(),
 			rt.EventBus(),
+			rt.ResourceManager(),
+			rt.Config().Store.Upsert,
+			rt.GetInstanceId(),
 		),
 	),
 		rt.Config().General.ResilientComponentBaseBackoff.Duration,

pkg/kds/mux/zone_sync.go

@@ -2,20 +2,31 @@ package mux
 
 import (
 	"context"
+	"math/rand"
+	"time"
 
 	"github.com/pkg/errors"
+	"github.com/sethvargo/go-retry"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
 	mesh_proto "github.com/kumahq/kuma/api/mesh/v1alpha1"
+	"github.com/kumahq/kuma/api/system/v1alpha1"
+	system_proto "github.com/kumahq/kuma/api/system/v1alpha1"
+	config_store "github.com/kumahq/kuma/pkg/config/core/resources/store"
 	"github.com/kumahq/kuma/pkg/core"
+	"github.com/kumahq/kuma/pkg/core/resources/apis/system"
+	"github.com/kumahq/kuma/pkg/core/resources/manager"
+	"github.com/kumahq/kuma/pkg/core/resources/model"
+	core_store "github.com/kumahq/kuma/pkg/core/resources/store"
 	"github.com/kumahq/kuma/pkg/events"
 	"github.com/kumahq/kuma/pkg/kds"
 	"github.com/kumahq/kuma/pkg/kds/service"
 	"github.com/kumahq/kuma/pkg/kds/util"
 	"github.com/kumahq/kuma/pkg/log"
 	"github.com/kumahq/kuma/pkg/multitenant"
+	"github.com/kumahq/kuma/pkg/util/proto"
 )
 
 type FilterV2 interface {
@@ -44,17 +55,23 @@ type KDSSyncServiceServer struct {
 	extensions     context.Context
 	eventBus       events.EventBus
 	mesh_proto.UnimplementedKDSSyncServiceServer
-	context context.Context
+	context    context.Context
+	resManager manager.ResourceManager
+	upsertCfg  config_store.UpsertConfig
+	instanceID string
 }
 
-func NewKDSSyncServiceServer(ctx context.Context, globalToZoneCb OnGlobalToZoneSyncConnectFunc, zoneToGlobalCb OnZoneToGlobalSyncConnectFunc, filters []FilterV2, extensions context.Context, eventBus events.EventBus) *KDSSyncServiceServer {
+func NewKDSSyncServiceServer(ctx context.Context, globalToZoneCb OnGlobalToZoneSyncConnectFunc, zoneToGlobalCb OnZoneToGlobalSyncConnectFunc, filters []FilterV2, extensions context.Context, eventBus events.EventBus, resManager manager.ResourceManager, upsertCfg config_store.UpsertConfig, instanceID string) *KDSSyncServiceServer {
 	return &KDSSyncServiceServer{
 		context:        ctx,
 		globalToZoneCb: globalToZoneCb,
 		zoneToGlobalCb: zoneToGlobalCb,
 		filters:        filters,
 		extensions:     extensions,
 		eventBus:       eventBus,
+		resManager:     resManager,
+		upsertCfg:      upsertCfg,
+		instanceID:     instanceID,
 	}
 }
 
@@ -66,14 +83,15 @@ func (g *KDSSyncServiceServer) GlobalToZoneSync(stream mesh_proto.KDSSyncService
 	if err != nil {
 		return err
 	}
-	logger = logger.WithValues("clientID", zone)
+	logger = logger.WithValues("clientID", zone, "type", "globalToZone")
 	for _, filter := range g.filters {
 		if err := filter.InterceptServerStream(stream); err != nil {
 			return errors.Wrap(err, "closing KDS stream following a callback error")
 		}
 	}
 
-	shouldDisconnectStream := g.watchZoneHealthCheck(stream.Context(), zone)
+	connectTime := time.Now()
+	shouldDisconnectStream := g.watchZoneHealthCheck(stream.Context(), zone, service.GlobalToZone, connectTime)
 	defer shouldDisconnectStream.Close()
 
 	processingErrorsCh := make(chan error, 1)
@@ -82,6 +100,14 @@ func (g *KDSSyncServiceServer) GlobalToZoneSync(stream mesh_proto.KDSSyncService
 			processingErrorsCh <- err
 		}
 	}()
+	if err := g.storeStreamConnection(stream.Context(), zone, service.GlobalToZone, connectTime); err != nil {
+		if errors.Is(err, context.Canceled) && errors.Is(stream.Context().Err(), context.Canceled) {
+			return status.Error(codes.Canceled, "stream was cancelled")
+		}
+		logger.Error(err, "could not store stream connection")
+		return status.Error(codes.Internal, "could not store stream connection")
+	}
+	logger.Info("stored stream connection")
 
 	select {
 	case <-shouldDisconnectStream.Recv():
@@ -108,14 +134,14 @@ func (g *KDSSyncServiceServer) ZoneToGlobalSync(stream mesh_proto.KDSSyncService
 	if err != nil {
 		return err
 	}
-	logger = logger.WithValues("clientID", zone)
+	logger = logger.WithValues("clientID", zone, "type", "zoneToGlobal")
 	for _, filter := range g.filters {
 		if err := filter.InterceptServerStream(stream); err != nil {
 			return errors.Wrap(err, "closing KDS stream following a callback error")
 		}
 	}
-
-	shouldDisconnectStream := g.watchZoneHealthCheck(stream.Context(), zone)
+	connectTime := time.Now()
+	shouldDisconnectStream := g.watchZoneHealthCheck(stream.Context(), zone, service.ZoneToGlobal, connectTime)
 	defer shouldDisconnectStream.Close()
 
 	processingErrorsCh := make(chan error, 1)
@@ -125,6 +151,15 @@ func (g *KDSSyncServiceServer) ZoneToGlobalSync(stream mesh_proto.KDSSyncService
 		}
 	}()
 
+	if err := g.storeStreamConnection(stream.Context(), zone, service.ZoneToGlobal, connectTime); err != nil {
+		if errors.Is(err, context.Canceled) && errors.Is(stream.Context().Err(), context.Canceled) {
+			return status.Error(codes.Canceled, "stream was cancelled")
+		}
+		logger.Error(err, "could not store stream connection")
+		return status.Error(codes.Internal, "could not store stream connection")
+	}
+	logger.Info("stored stream connection")
+
 	select {
 	case <-shouldDisconnectStream.Recv():
 		logger.Info("ending stream, zone health check failed")
@@ -144,18 +179,74 @@ func (g *KDSSyncServiceServer) ZoneToGlobalSync(stream mesh_proto.KDSSyncService
 	}
 }
 
-func (g *KDSSyncServiceServer) watchZoneHealthCheck(streamContext context.Context, zone string) events.Listener {
+func (g *KDSSyncServiceServer) watchZoneHealthCheck(streamContext context.Context, zone string, typ service.StreamType, connectTime time.Time) events.Listener {
 	tenantID, _ := multitenant.TenantFromCtx(streamContext)
 
 	shouldDisconnectStream := events.NewNeverListener()
-
 	if kds.ContextHasFeature(streamContext, kds.FeatureZonePingHealth) {
 		shouldDisconnectStream = g.eventBus.Subscribe(func(e events.Event) bool {
-			disconnectEvent, ok := e.(service.ZoneWentOffline)
-			return ok && disconnectEvent.TenantID == tenantID && disconnectEvent.Zone == zone
+			switch event := e.(type) {
+			case service.ZoneWentOffline:
+				return event.TenantID == tenantID && event.Zone == zone
+			case service.StreamCancelled:
+				return event.TenantID == tenantID && event.Zone == zone && event.Type == typ && event.ConnTime == connectTime
+			default:
+				return false
+			}
 		})
-		g.eventBus.Send(service.ZoneOpenedStream{Zone: zone, TenantID: tenantID})
+		g.eventBus.Send(service.ZoneOpenedStream{Zone: zone, TenantID: tenantID, Type: typ, ConnTime: connectTime})
 	}
 
 	return shouldDisconnectStream
 }
+
+func (g *KDSSyncServiceServer) storeStreamConnection(ctx context.Context, zone string, typ service.StreamType, connectTime time.Time) error {
+	key := model.ResourceKey{Name: zone}
+
+	// wait for Zone to be created, only then we can create Zone Insight
+	err := retry.Do(
+		ctx,
+		retry.WithMaxRetries(30, retry.NewConstant(1*time.Second)),
+		func(ctx context.Context) error {
+			return retry.RetryableError(g.resManager.Get(ctx, system.NewZoneResource(), core_store.GetBy(key)))
+		},
+	)
+	if err != nil {
+		return err
+	}
+
+	// Add delay for Upsert. If Global CP is behind an HTTP load balancer,
+	// it might be the case that each Envoy Admin stream will land on separate instance.
+	// In this case, all instances will try to update Zone Insight which will result in conflicts.
+	// Since it's unusual to immediately execute envoy admin rpcs after zone is connected, 0-10s delay should be fine.
+	// #nosec G404 - math rand is enough
+	time.Sleep(time.Duration(rand.Int31n(10000)) * time.Millisecond)
+
+	zoneInsight := system.NewZoneInsightResource()
+	return manager.Upsert(ctx, g.resManager, key, zoneInsight, func(resource model.Resource) error {
+		if zoneInsight.Spec.KdsStreams == nil {
+			zoneInsight.Spec.KdsStreams = &v1alpha1.KDSStreams{}
+		}
+		var stream *system_proto.KDSStream
+		switch typ {
+		case service.GlobalToZone:
+			stream = zoneInsight.Spec.GetKdsStreams().GetGlobalToZone()
+		case service.ZoneToGlobal:
+			stream = zoneInsight.Spec.GetKdsStreams().GetZoneToGlobal()
+		}
+		if stream == nil {
+			stream = &v1alpha1.KDSStream{}
+		}
+		if stream.GetConnectTime() == nil || proto.MustTimestampFromProto(stream.ConnectTime).Before(connectTime) {
+			stream.GlobalInstanceId = g.instanceID
+			stream.ConnectTime = proto.MustTimestampProto(connectTime)
+		}
+		switch typ {
+		case service.GlobalToZone:
+			zoneInsight.Spec.KdsStreams.GlobalToZone = stream
+		case service.ZoneToGlobal:
+			zoneInsight.Spec.KdsStreams.ZoneToGlobal = stream
+		}
+		return nil
+	}, manager.WithConflictRetry(g.upsertCfg.ConflictRetryBaseBackoff.Duration, g.upsertCfg.ConflictRetryMaxTimes, g.upsertCfg.ConflictRetryJitterPercent)) // we need retry because zone sink or other RPC may also update the insight.
+}