OTA-1418: USC: Drop unknown messages from the API (#1143)

* USC: Drop unknown messages from the API

Assume the informers, in each message, include a list of all active insight IDs that infomer knows about. The Status API can then drop the unknown insights formerly reported by that informer (=insights present in the API but not known to be active by the responsible informer)

This is only a part of OTA-1418. The remaining changes will be as follows:

- Do not drop the unknown insights right away but make them expire after some some time of not being renewed (handle the case where informer starts up and starts sending messages about insights as it discovers them)
- Actually make the informers remember their insights and populate the `knownInsights` field of the message
- Add additional `deletedInsights` field to the message, to allow informers explicitly tell the API some insight went away
- Make the USC respect `deletedInsights`
- Make informers fill `deletedInsight`
- Potentially add a out-of-API memory cache for insights to be expired / deleted but allow them to reappear with formerly known content

Co-authored-by: David Hurta <[email protected]>

* Apply suggestions from code review

Co-authored-by: David Hurta <[email protected]>

* Fix comment formatting in updatestatuscontroller.go

---------

Co-authored-by: David Hurta <[email protected]>
Co-authored-by: David Hurta <[email protected]>

Author: petr-muller

pkg/updatestatus/updatestatuscontroller.go

@@ -10,6 +10,7 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"gopkg.in/yaml.v3"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/klog/v2"
 
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -29,6 +30,11 @@ import (
 // we have the Status API we need to serialize ourselves anyway.
 type informerMsg struct {
 	informer string
+	// knownInsights contains the UIDs of insights known by the informer, so the controller can remove insights formerly
+	// reported by the informer but no longer known to it (e.g. because the informer was restarted and the culprit
+	// condition ceased to exist in the meantime). The `uid` of the insight in the message payload is always assumed
+	// to be known, and is not required to be included in `knownInsights` by the informers (but informers can do so).
+	knownInsights []string
 
 	uid     string
 	insight []byte
@@ -153,6 +159,7 @@ func (c *updateStatusController) processInsightMsg(message informerMsg) bool {
 	}
 	klog.Infof("USC :: Collector :: Received insight from informer %q (uid=%s)", message.informer, message.uid)
 	c.updateInsightInStatusApi(message)
+	c.removeUnknownInsights(message)
 
 	return true
 }
@@ -213,7 +220,21 @@ func (c *updateStatusController) updateInsightInStatusApi(msg informerMsg) {
 			klog.Infof("USC :: Collector :: Insight (uid=%s) content did not change (len=%d)", msg.uid, len(updatedContent))
 		}
 	}
+}
 
+// removeUnknownInsights removes insights from the status API that are no longer reported as known to the informer
+// that originally reported them.
+// Assumes the statusApi field is locked.
+func (c *updateStatusController) removeUnknownInsights(message informerMsg) {
+	known := sets.New(message.knownInsights...)
+	known.Insert(message.uid)
+	informerPrefix := fmt.Sprintf("usc.%s.", message.informer)
+	for key := range c.statusApi.cm.Data {
+		if strings.HasPrefix(key, informerPrefix) && !known.Has(strings.TrimPrefix(key, informerPrefix)) {
+			delete(c.statusApi.cm.Data, key)
+			klog.V(2).Infof("USC :: Collector :: Dropped insight %q because it is no longer reported as known by informer %q", key, message.informer)
+		}
+	}
 }
 
 func (c *updateStatusController) commitStatusApiAsConfigMap(ctx context.Context) error {

pkg/updatestatus/updatestatuscontroller_test.go

@@ -77,24 +77,28 @@ func Test_updateStatusController(t *testing.T) {
 			},
 			informerMsg: []informerMsg{
 				{
-					informer: "cpi",
-					uid:      "new-item",
-					insight:  []byte("msg1"),
+					informer:      "cpi",
+					uid:           "new-item",
+					insight:       []byte("msg1"),
+					knownInsights: []string{"kept", "overwritten"},
 				},
 				{
-					informer: "cpi",
-					uid:      "overwritten",
-					insight:  []byte("msg2 (overwritten intermediate)"),
+					informer:      "cpi",
+					uid:           "overwritten",
+					insight:       []byte("msg2 (overwritten intermediate)"),
+					knownInsights: []string{"kept", "new-item"},
 				},
 				{
-					informer: "cpi",
-					uid:      "another",
-					insight:  []byte("msg3"),
+					informer:      "cpi",
+					uid:           "another",
+					insight:       []byte("msg3"),
+					knownInsights: []string{"kept", "overwritten", "new-item"},
 				},
 				{
-					informer: "cpi",
-					uid:      "overwritten",
-					insight:  []byte("msg4 (overwritten final)"),
+					informer:      "cpi",
+					uid:           "overwritten",
+					insight:       []byte("msg4 (overwritten final)"),
+					knownInsights: []string{"kept", "new-item", "another"},
 				},
 			},
 			expected: &corev1.ConfigMap{
@@ -175,6 +179,25 @@ func Test_updateStatusController(t *testing.T) {
 			},
 			expected: nil,
 		},
+		{
+			name: "unknown message gets removed from state",
+			controllerConfigMap: &corev1.ConfigMap{
+				Data: map[string]string{
+					"usc.one.old": "payload",
+				},
+			},
+			informerMsg: []informerMsg{{
+				informer:      "one",
+				uid:           "new",
+				insight:       []byte("new payload"),
+				knownInsights: nil,
+			}},
+			expected: &corev1.ConfigMap{
+				Data: map[string]string{
+					"usc.one.new": "new payload",
+				},
+			},
+		},
 	}
 
 	for _, tc := range testCases {