fix(kuma-cp) subscription finalizer, rev 2

api/generic/insights.go

@@ -0,0 +1,19 @@
+package generic
+
+import (
+	"time"
+
+	"github.com/golang/protobuf/proto"
+)
+
+type Insight interface {
+	proto.Message
+	IsOnline() bool
+	GetLastSubscription() Subscription
+	UpdateSubscription(Subscription)
+}
+
+type Subscription interface {
+	proto.Message
+	SetDisconnectTime(time time.Time)
+}

api/mesh/v1alpha1/dataplane_insight_helpers.go

@@ -5,6 +5,8 @@ import (
 	"time"
 
 	"google.golang.org/protobuf/types/known/timestamppb"
+
+	"github.com/kumahq/kuma/api/generic"
 )
 
 func NewSubscriptionStatus() *DiscoverySubscriptionStatus {
@@ -32,74 +34,79 @@ func NewVersion() *Version {
 	}
 }
 
-func (ds *DataplaneInsight) IsOnline() bool {
-	for _, s := range ds.GetSubscriptions() {
-		if s.ConnectTime != nil && s.DisconnectTime == nil {
+func (x *DataplaneInsight) IsOnline() bool {
+	for _, s := range x.GetSubscriptions() {
+		if s.GetConnectTime() != nil && s.GetDisconnectTime() == nil {
 			return true
 		}
 	}
 	return false
 }
 
-func (ds *DataplaneInsight) GetSubscription(id string) (int, *DiscoverySubscription) {
-	for i, s := range ds.GetSubscriptions() {
+func (x *DataplaneInsight) GetSubscription(id string) (int, *DiscoverySubscription) {
+	for i, s := range x.GetSubscriptions() {
 		if s.Id == id {
 			return i, s
 		}
 	}
 	return -1, nil
 }
 
-func (ds *DataplaneInsight) UpdateCert(generation time.Time, expiration time.Time) error {
-	if ds.MTLS == nil {
-		ds.MTLS = &DataplaneInsight_MTLS{}
+func (x *DataplaneInsight) UpdateCert(generation time.Time, expiration time.Time) error {
+	if x.MTLS == nil {
+		x.MTLS = &DataplaneInsight_MTLS{}
 	}
 	ts := timestamppb.New(expiration)
 	if err := ts.CheckValid(); err != nil {
 		return err
 	}
-	ds.MTLS.CertificateExpirationTime = ts
-	ds.MTLS.CertificateRegenerations++
+	x.MTLS.CertificateExpirationTime = ts
+	x.MTLS.CertificateRegenerations++
 	ts = timestamppb.New(generation)
 	if err := ts.CheckValid(); err != nil {
 		return err
 	}
-	ds.MTLS.LastCertificateRegeneration = ts
+	x.MTLS.LastCertificateRegeneration = ts
 	return nil
 }
 
-func (ds *DataplaneInsight) UpdateSubscription(s *DiscoverySubscription) {
-	if ds == nil {
+func (x *DataplaneInsight) UpdateSubscription(s generic.Subscription) {
+	if x == nil {
+		return
+	}
+	discoverySubscription, ok := s.(*DiscoverySubscription)
+	if !ok {
 		return
 	}
-	i, old := ds.GetSubscription(s.Id)
+	i, old := x.GetSubscription(discoverySubscription.Id)
 	if old != nil {
-		ds.Subscriptions[i] = s
+		x.Subscriptions[i] = discoverySubscription
 	} else {
-		ds.finalizeSubscriptions()
-		ds.Subscriptions = append(ds.Subscriptions, s)
+		x.finalizeSubscriptions()
+		x.Subscriptions = append(x.Subscriptions, discoverySubscription)
 	}
 }
 
 // If Kuma CP was killed ungracefully then we can get a subscription without a DisconnectTime.
 // Because of the way we process subscriptions the lack of DisconnectTime on old subscription
 // will cause wrong status.
-func (ds *DataplaneInsight) finalizeSubscriptions() {
+func (x *DataplaneInsight) finalizeSubscriptions() {
 	now := timestamppb.Now()
-	for _, subscription := range ds.GetSubscriptions() {
+	for _, subscription := range x.GetSubscriptions() {
 		if subscription.DisconnectTime == nil {
 			subscription.DisconnectTime = now
 		}
 	}
 }
 
-func (ds *DataplaneInsight) GetLatestSubscription() (*DiscoverySubscription, *time.Time) {
-	if len(ds.GetSubscriptions()) == 0 {
+// todo(lobkovilya): delete GetLatestSubscription, use GetLastSubscription instead
+func (x *DataplaneInsight) GetLatestSubscription() (*DiscoverySubscription, *time.Time) {
+	if len(x.GetSubscriptions()) == 0 {
 		return nil, nil
 	}
 	var idx int = 0
 	var latest *time.Time
-	for i, s := range ds.GetSubscriptions() {
+	for i, s := range x.GetSubscriptions() {
 		if err := s.ConnectTime.CheckValid(); err != nil {
 			continue
 		}
@@ -109,12 +116,23 @@ func (ds *DataplaneInsight) GetLatestSubscription() (*DiscoverySubscription, *ti
 			latest = &t
 		}
 	}
-	return ds.Subscriptions[idx], latest
+	return x.Subscriptions[idx], latest
+}
+
+func (x *DataplaneInsight) GetLastSubscription() generic.Subscription {
+	if len(x.GetSubscriptions()) == 0 {
+		return nil
+	}
+	return x.GetSubscriptions()[len(x.GetSubscriptions())-1]
+}
+
+func (x *DiscoverySubscription) SetDisconnectTime(t time.Time) {
+	x.DisconnectTime = timestamppb.New(t)
 }
 
-func (ds *DataplaneInsight) Sum(v func(*DiscoverySubscription) uint64) uint64 {
+func (x *DataplaneInsight) Sum(v func(*DiscoverySubscription) uint64) uint64 {
 	var result uint64 = 0
-	for _, s := range ds.GetSubscriptions() {
+	for _, s := range x.GetSubscriptions() {
 		result += v(s)
 	}
 	return result

api/mesh/v1alpha1/zone_ingress_insight_helpers.go

@@ -4,6 +4,8 @@ import (
 	"time"
 
 	"google.golang.org/protobuf/types/known/timestamppb"
+
+	"github.com/kumahq/kuma/api/generic"
 )
 
 func (x *ZoneIngressInsight) GetSubscription(id string) (int, *DiscoverySubscription) {
@@ -15,16 +17,20 @@ func (x *ZoneIngressInsight) GetSubscription(id string) (int, *DiscoverySubscrip
 	return -1, nil
 }
 
-func (x *ZoneIngressInsight) UpdateSubscription(s *DiscoverySubscription) {
+func (x *ZoneIngressInsight) UpdateSubscription(s generic.Subscription) {
 	if x == nil {
 		return
 	}
-	i, old := x.GetSubscription(s.Id)
+	discoverySubscription, ok := s.(*DiscoverySubscription)
+	if !ok {
+		return
+	}
+	i, old := x.GetSubscription(discoverySubscription.Id)
 	if old != nil {
-		x.Subscriptions[i] = s
+		x.Subscriptions[i] = discoverySubscription
 	} else {
 		x.finalizeSubscriptions()
-		x.Subscriptions = append(x.Subscriptions, s)
+		x.Subscriptions = append(x.Subscriptions, discoverySubscription)
 	}
 }
 
@@ -49,6 +55,14 @@ func (x *ZoneIngressInsight) IsOnline() bool {
 	return false
 }
 
+func (x *ZoneIngressInsight) GetLastSubscription() generic.Subscription {
+	if len(x.GetSubscriptions()) == 0 {
+		return nil
+	}
+	return x.GetSubscriptions()[len(x.GetSubscriptions())-1]
+}
+
+// todo(lobkovilya): delete GetLatestSubscription, use GetLastSubscription instead
 func (x *ZoneIngressInsight) GetLatestSubscription() (*DiscoverySubscription, *time.Time) {
 	if len(x.GetSubscriptions()) == 0 {
 		return nil, nil

api/system/v1alpha1/zone_insight_helpers.go

@@ -4,6 +4,8 @@ import (
 	"time"
 
 	"google.golang.org/protobuf/types/known/timestamppb"
+
+	"github.com/kumahq/kuma/api/generic"
 )
 
 func NewSubscriptionStatus() *KDSSubscriptionStatus {
@@ -13,22 +15,23 @@ func NewSubscriptionStatus() *KDSSubscriptionStatus {
 	}
 }
 
-func (m *ZoneInsight) GetSubscription(id string) (int, *KDSSubscription) {
-	for i, s := range m.GetSubscriptions() {
+func (x *ZoneInsight) GetSubscription(id string) (int, *KDSSubscription) {
+	for i, s := range x.GetSubscriptions() {
 		if s.Id == id {
 			return i, s
 		}
 	}
 	return -1, nil
 }
 
-func (m *ZoneInsight) GetLatestSubscription() (*KDSSubscription, *time.Time) {
-	if len(m.GetSubscriptions()) == 0 {
+// todo(lobkovilya): delete GetLatestSubscription, use GetLastSubscription instead
+func (x *ZoneInsight) GetLatestSubscription() (*KDSSubscription, *time.Time) {
+	if len(x.GetSubscriptions()) == 0 {
 		return nil, nil
 	}
 	var idx = 0
 	var latest *time.Time
-	for i, s := range m.GetSubscriptions() {
+	for i, s := range x.GetSubscriptions() {
 		if err := s.ConnectTime.CheckValid(); err != nil {
 			continue
 		}
@@ -38,45 +41,60 @@ func (m *ZoneInsight) GetLatestSubscription() (*KDSSubscription, *time.Time) {
 			latest = &t
 		}
 	}
-	return m.Subscriptions[idx], latest
+	return x.Subscriptions[idx], latest
+}
+
+func (x *ZoneInsight) GetLastSubscription() generic.Subscription {
+	if len(x.GetSubscriptions()) == 0 {
+		return nil
+	}
+	return x.GetSubscriptions()[len(x.GetSubscriptions())-1]
 }
 
-func (m *ZoneInsight) IsOnline() bool {
-	for _, s := range m.GetSubscriptions() {
+func (x *ZoneInsight) IsOnline() bool {
+	for _, s := range x.GetSubscriptions() {
 		if s.ConnectTime != nil && s.DisconnectTime == nil {
 			return true
 		}
 	}
 	return false
 }
 
-func (m *ZoneInsight) Sum(v func(*KDSSubscription) uint64) uint64 {
+func (x *KDSSubscription) SetDisconnectTime(time time.Time) {
+	x.DisconnectTime = timestamppb.New(time)
+}
+
+func (x *ZoneInsight) Sum(v func(*KDSSubscription) uint64) uint64 {
 	var result uint64 = 0
-	for _, s := range m.GetSubscriptions() {
+	for _, s := range x.GetSubscriptions() {
 		result += v(s)
 	}
 	return result
 }
 
-func (m *ZoneInsight) UpdateSubscription(s *KDSSubscription) {
-	if m == nil {
+func (x *ZoneInsight) UpdateSubscription(s generic.Subscription) {
+	if x == nil {
+		return
+	}
+	kdsSubscription, ok := s.(*KDSSubscription)
+	if !ok {
 		return
 	}
-	i, old := m.GetSubscription(s.Id)
+	i, old := x.GetSubscription(kdsSubscription.Id)
 	if old != nil {
-		m.Subscriptions[i] = s
+		x.Subscriptions[i] = kdsSubscription
 	} else {
-		m.finalizeSubscriptions()
-		m.Subscriptions = append(m.Subscriptions, s)
+		x.finalizeSubscriptions()
+		x.Subscriptions = append(x.Subscriptions, kdsSubscription)
 	}
 }
 
 // If Global CP was killed ungracefully then we can get a subscription without a DisconnectTime.
 // Because of the way we process subscriptions the lack of DisconnectTime on old subscription
 // will cause wrong status.
-func (m *ZoneInsight) finalizeSubscriptions() {
+func (x *ZoneInsight) finalizeSubscriptions() {
 	now := timestamppb.Now()
-	for _, subscription := range m.GetSubscriptions() {
+	for _, subscription := range x.GetSubscriptions() {
 		if subscription.DisconnectTime == nil {
 			subscription.DisconnectTime = now
 		}

app/kuma-cp/cmd/run.go

@@ -103,10 +103,6 @@ func newRunCmdWithOpts(opts runCmdOpts) *cobra.Command {
 					runLog.Error(err, "unable to set up DNS")
 					return err
 				}
-				if err := gc.Setup(rt); err != nil {
-					runLog.Error(err, "unable to set up GC")
-					return err
-				}
 				if err := xds.Setup(rt); err != nil {
 					runLog.Error(err, "unable to set up XDS")
 					return err
@@ -144,10 +140,6 @@ func newRunCmdWithOpts(opts runCmdOpts) *cobra.Command {
 					runLog.Error(err, "unable to set up DNS")
 					return err
 				}
-				if err := gc.Setup(rt); err != nil {
-					runLog.Error(err, "unable to set up GC")
-					return err
-				}
 				if err := xds.Setup(rt); err != nil {
 					runLog.Error(err, "unable to set up XDS")
 					return err
@@ -195,6 +187,10 @@ func newRunCmdWithOpts(opts runCmdOpts) *cobra.Command {
 				runLog.Error(err, "unable to set up Metrics")
 				return err
 			}
+			if err := gc.Setup(rt); err != nil {
+				runLog.Error(err, "unable to set up GC")
+				return err
+			}
 
 			runLog.Info("starting Control Plane", "version", kuma_version.Build.Version)
 			if err := rt.Start(closeCh); err != nil {

pkg/api-server/config_ws_test.go

@@ -128,15 +128,17 @@ var _ = Describe("Config WS", func() {
 		  "metrics": {
 			"dataplane": {
 			  "enabled": true,
-			  "subscriptionLimit": 2
+			  "subscriptionLimit": 2,
+			  "idleTimeout": "5m0s"
 			},
 			"mesh": {
 			  "maxResyncTimeout": "20s",
 			  "minResyncTimeout": "1s"
 			},
 			"zone": {
 			  "enabled": true,
-			  "subscriptionLimit": 10
+			  "subscriptionLimit": 10,
+			  "idleTimeout": "5m0s"
 			}
 		  },
 		  "mode": "standalone",