fix(kuma-cp) subscription finalizer, rev 2

api/helpers/insights.go

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

api/mesh/v1alpha1/dataplane_insight.proto

@@ -69,6 +69,10 @@ message DiscoverySubscription {
 
   // Version of Envoy and Kuma dataplane
   Version version = 6;
+
+  // Generation is an integer number which is periodically increased by the
+  // status sink
+  uint32 generation = 7;
 }
 
 // DiscoverySubscriptionStatus defines status of an ADS subscription.

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/helpers"
 )
 
 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 helpers.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() helpers.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/helpers"
 )
 
 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 helpers.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() helpers.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.proto

@@ -53,6 +53,10 @@ message KDSSubscription {
 
   // Version of Zone Kuma CP.
   Version version = 6;
+
+  // Generation is an integer number which is periodically increased by the
+  // status sink
+  uint32 generation = 7;
 }
 
 // KDSSubscriptionStatus defines status of an KDS subscription.