OTA-1418: USC: Make messages more robust and decoupled (#1146)

* OTA-1418: Identify the informer sending insight message

Make USC aware of the identities of the informers that send insights to it, by including the identifier in the message sent to USC. In the future API, we will have separated containers per informer:

```
informers:
- name: name-1
  insights: [...]
- name: name-2
  insights: [...]
```

We do not have this in our ConfigMap API mock so for now we encode the informer into the key of the item stored in the ConfigMap. USC also takes ownership of the `usc.` prefix (formerly `usc-`) which identifies insight keys, and the informers no longer need to maintain this contract.

* USC: Drop & log invalid messages

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

* USC: Validate insight message creation in producers

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

* USC: Producers no longer need to prefix insight UIDs

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

* USC: Address review

---------

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

Author: petr-muller

pkg/updatestatus/controlplaneinformer.go

@@ -9,7 +9,6 @@ import (
 	"strings"
 	"time"
 
-	"gopkg.in/yaml.v3"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -87,8 +86,9 @@ func clusterOperatorEventFilterFunc(obj interface{}) bool {
 }
 
 const (
-	clusterVersionKindName  = "ClusterVersion"
-	clusterOperatorKindName = "ClusterOperator"
+	clusterVersionKindName   = "ClusterVersion"
+	clusterOperatorKindName  = "ClusterOperator"
+	controlPlaneInformerName = "cpi"
 )
 
 // sync is called for any controller event. It will assess the state and health of the control plane, indicated by
@@ -118,9 +118,19 @@ func (c *controlPlaneInformerController) sync(ctx context.Context, syncCtx facto
 
 		now := c.now()
 		cvInsight, healthInsights := assessClusterVersion(clusterVersion, now)
-		msgs = append(msgs, makeInsightMsgForClusterVersion(cvInsight, now))
+		msg, err := makeInsightMsgForClusterVersion(cvInsight, now)
+		if err != nil {
+			klog.Errorf("BUG: Could not create insight message: %v", err)
+			return nil
+		}
+		msgs = append(msgs, msg)
 		for item := range healthInsights {
-			msgs = append(msgs, makeInsightMsgForHealthInsight(healthInsights[item], now))
+			msg, err := makeInsightMsgForHealthInsight(healthInsights[item], now)
+			if err != nil {
+				klog.Errorf("BUG: Could not create insight message: %v", err)
+				return nil
+			}
+			msgs = append(msgs, msg)
 		}
 
 	case clusterOperatorKindName:
@@ -144,7 +154,12 @@ func (c *controlPlaneInformerController) sync(ctx context.Context, syncCtx facto
 		if err != nil {
 			return fmt.Errorf("failed to assess cluster operator %s: %w", name, err)
 		}
-		msgs = append(msgs, makeInsightMsgForClusterOperator(insight, now))
+		msg, err := makeInsightMsgForClusterOperator(insight, now)
+		if err != nil {
+			klog.Errorf("BUG: Could not create insight message: %v", err)
+			return nil
+		}
+		msgs = append(msgs, msg)
 	default:
 		return fmt.Errorf("invalid queue key %s with unexpected type %s", queueKey, t)
 	}
@@ -161,22 +176,16 @@ func (c *controlPlaneInformerController) sync(ctx context.Context, syncCtx facto
 	return nil
 }
 
-func makeInsightMsgForClusterOperator(coInsight *ClusterOperatorStatusInsight, acquiredAt metav1.Time) informerMsg {
-	uid := fmt.Sprintf("usc-co-%s", coInsight.Name)
+func makeInsightMsgForClusterOperator(coInsight *ClusterOperatorStatusInsight, acquiredAt metav1.Time) (informerMsg, error) {
 	insight := ControlPlaneInsight{
-		UID:        uid,
+		UID:        fmt.Sprintf("co-%s", coInsight.Name),
 		AcquiredAt: acquiredAt,
 		ControlPlaneInsightUnion: ControlPlaneInsightUnion{
 			Type:                         ClusterOperatorStatusInsightType,
 			ClusterOperatorStatusInsight: coInsight,
 		},
 	}
-	// Should handle errors, but ultimately we will have a proper API and won’t need to serialize ourselves
-	rawInsight, _ := yaml.Marshal(insight)
-	return informerMsg{
-		uid:     uid,
-		insight: rawInsight,
-	}
+	return makeControlPlaneInsightMsg(insight, controlPlaneInformerName)
 }
 
 func assessClusterOperator(ctx context.Context, operator *configv1.ClusterOperator, targetVersion string, appsClient appsv1client.AppsV1Interface, now metav1.Time) (*ClusterOperatorStatusInsight, error) {
@@ -300,22 +309,16 @@ func getImagePullSpec(ctx context.Context, name string, appsClient appsv1client.
 // makeInsightMsgForClusterVersion creates an informerMsg for the given ClusterVersionStatusInsight. It defines an uid
 // name and serializes the insight as YAML. Serialization is convenient because it prevents any data sharing issues
 // between controllers.
-func makeInsightMsgForClusterVersion(cvInsight *ClusterVersionStatusInsight, acquiredAt metav1.Time) informerMsg {
-	uid := fmt.Sprintf("usc-cv-%s", cvInsight.Resource.Name)
+func makeInsightMsgForClusterVersion(cvInsight *ClusterVersionStatusInsight, acquiredAt metav1.Time) (informerMsg, error) {
 	insight := ControlPlaneInsight{
-		UID:        uid,
+		UID:        fmt.Sprintf("cv-%s", cvInsight.Resource.Name),
 		AcquiredAt: acquiredAt,
 		ControlPlaneInsightUnion: ControlPlaneInsightUnion{
 			Type:                        ClusterVersionStatusInsightType,
 			ClusterVersionStatusInsight: cvInsight,
 		},
 	}
-	// Should handle errors, but ultimately we will have a proper API and won’t need to serialize ourselves
-	rawInsight, _ := yaml.Marshal(insight)
-	return informerMsg{
-		uid:     uid,
-		insight: rawInsight,
-	}
+	return makeControlPlaneInsightMsg(insight, controlPlaneInformerName)
 }
 
 func uidForHealthInsight(healthInsight *HealthInsight) string {
@@ -332,26 +335,19 @@ func uidForHealthInsight(healthInsight *HealthInsight) string {
 	encoded := base64.StdEncoding.EncodeToString(sum)
 	encoded = strings.TrimRight(encoded, "=")
 
-	return fmt.Sprintf("usc-%s", encoded)
+	return encoded
 }
 
-func makeInsightMsgForHealthInsight(healthInsight *HealthInsight, acquiredAt metav1.Time) informerMsg {
-	uid := uidForHealthInsight(healthInsight)
+func makeInsightMsgForHealthInsight(healthInsight *HealthInsight, acquiredAt metav1.Time) (informerMsg, error) {
 	insight := ControlPlaneInsight{
-		UID:        uid,
+		UID:        uidForHealthInsight(healthInsight),
 		AcquiredAt: acquiredAt,
 		ControlPlaneInsightUnion: ControlPlaneInsightUnion{
 			Type:          HealthInsightType,
 			HealthInsight: healthInsight,
 		},
 	}
-
-	// Should handle errors, but ultimately we will have a proper API and won’t need to serialize ourselves
-	rawInsight, _ := yaml.Marshal(insight)
-	return informerMsg{
-		uid:     uid,
-		insight: rawInsight,
-	}
+	return makeControlPlaneInsightMsg(insight, controlPlaneInformerName)
 }
 
 // assessClusterVersion produces a ClusterVersion status insight from the current state of the ClusterVersion resource.

pkg/updatestatus/controlplaneinformer_test.go

@@ -96,8 +96,8 @@ func Test_sync_with_cv(t *testing.T) {
 			cvProgressing: &progressingTrue,
 			cvHistory:     []configv1.UpdateHistory{inProgress418},
 			expectedMsgs: map[string]ControlPlaneInsight{
-				"usc-cv-version": {
-					UID:        "usc-cv-version",
+				"cv-version": {
+					UID:        "cv-version",
 					AcquiredAt: now,
 					ControlPlaneInsightUnion: ControlPlaneInsightUnion{
 						Type: ClusterVersionStatusInsightType,
@@ -131,8 +131,8 @@ func Test_sync_with_cv(t *testing.T) {
 			cvProgressing: &progressingFalse,
 			cvHistory:     []configv1.UpdateHistory{completed418},
 			expectedMsgs: map[string]ControlPlaneInsight{
-				"usc-cv-version": {
-					UID:        "usc-cv-version",
+				"cv-version": {
+					UID:        "cv-version",
 					AcquiredAt: now,
 					ControlPlaneInsightUnion: ControlPlaneInsightUnion{
 						Type: ClusterVersionStatusInsightType,
@@ -167,8 +167,8 @@ func Test_sync_with_cv(t *testing.T) {
 			cvProgressing: &progressingTrue,
 			cvHistory:     []configv1.UpdateHistory{inProgress419, completed418},
 			expectedMsgs: map[string]ControlPlaneInsight{
-				"usc-cv-version": {
-					UID:        "usc-cv-version",
+				"cv-version": {
+					UID:        "cv-version",
 					AcquiredAt: now,
 					ControlPlaneInsightUnion: ControlPlaneInsightUnion{
 						Type: ClusterVersionStatusInsightType,
@@ -203,8 +203,8 @@ func Test_sync_with_cv(t *testing.T) {
 				uscForceHealthInsightAnnotation: "value-does-not-matter",
 			},
 			expectedMsgs: map[string]ControlPlaneInsight{
-				"usc-0kmuaUQRUJDOAIAF1KWTmg": {
-					UID:        "usc-0kmuaUQRUJDOAIAF1KWTmg",
+				"0kmuaUQRUJDOAIAF1KWTmg": {
+					UID:        "0kmuaUQRUJDOAIAF1KWTmg",
 					AcquiredAt: now,
 					ControlPlaneInsightUnion: ControlPlaneInsightUnion{
 						Type: HealthInsightType,
@@ -228,8 +228,8 @@ func Test_sync_with_cv(t *testing.T) {
 						},
 					},
 				},
-				"usc-cv-version": {
-					UID:        "usc-cv-version",
+				"cv-version": {
+					UID:        "cv-version",
 					AcquiredAt: now,
 					ControlPlaneInsightUnion: ControlPlaneInsightUnion{
 						Type: ClusterVersionStatusInsightType,
@@ -293,8 +293,9 @@ func Test_sync_with_cv(t *testing.T) {
 					t.Fatalf("Failed to marshal expected insight: %v", err)
 				}
 				expectedMsgs = append(expectedMsgs, informerMsg{
-					uid:     uid,
-					insight: raw,
+					informer: controlPlaneInformerName,
+					uid:      uid,
+					insight:  raw,
 				})
 			}
 
@@ -304,6 +305,11 @@ func Test_sync_with_cv(t *testing.T) {
 			if diff := cmp.Diff(expectedMsgs, actualMsgs, ignoreOrder, cmp.AllowUnexported(informerMsg{})); diff != "" {
 				t.Errorf("Sync messages differ from expected:\n%s", diff)
 			}
+			for _, msg := range actualMsgs {
+				if err := msg.validate(); err != nil {
+					t.Errorf("Received message is invalid: %v\nMessage content: %v", err, msg)
+				}
+			}
 		})
 	}
 }
@@ -525,8 +531,8 @@ func Test_sync_with_co(t *testing.T) {
 		{
 			name: "Cluster during installation",
 			expectedMsgs: map[string]ControlPlaneInsight{
-				"usc-co-some-co": {
-					UID:        "usc-co-some-co",
+				"co-some-co": {
+					UID:        "co-some-co",
 					AcquiredAt: now,
 					ControlPlaneInsightUnion: ControlPlaneInsightUnion{
 						Type: ClusterOperatorStatusInsightType,
@@ -585,14 +591,24 @@ func Test_sync_with_co(t *testing.T) {
 					t.Fatalf("Failed to marshal expected insight: %v", err)
 				}
 				expectedMsgs = append(expectedMsgs, informerMsg{
-					uid:     uid,
-					insight: raw,
+					informer: controlPlaneInformerName,
+					uid:      uid,
+					insight:  raw,
 				})
 			}
 
-			if diff := cmp.Diff(expectedMsgs, actualMsgs, cmp.AllowUnexported(informerMsg{})); diff != "" {
+			ignoreOrder := cmpopts.SortSlices(func(a, b informerMsg) bool {
+				return a.uid < b.uid
+			})
+			if diff := cmp.Diff(expectedMsgs, actualMsgs, ignoreOrder, cmp.AllowUnexported(informerMsg{})); diff != "" {
 				t.Errorf("Sync messages differ from expected:\n%s", diff)
 			}
+
+			for _, msg := range actualMsgs {
+				if err := msg.validate(); err != nil {
+					t.Errorf("Received message is invalid: %v\nMessage content: %v", err, msg)
+				}
+			}
 		})
 	}
 }

pkg/updatestatus/nodeinformer.go

@@ -6,8 +6,6 @@ import (
 	"strings"
 	"time"
 
-	"gopkg.in/yaml.v3"
-
 	corev1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -126,7 +124,11 @@ func (c *nodeInformerController) sync(ctx context.Context, syncCtx factory.SyncC
 
 		now := c.now()
 		if insight := assessNode(node, mcp, machineConfigVersions, mostRecentVersionInCVHistory, now); insight != nil {
-			msg = makeInsightMsgForNode(insight, now)
+			msg, err = makeInsightMsgForNode(insight, now)
+			if err != nil {
+				klog.Errorf("BUG: Could not create insight message: %v", err)
+				return nil
+			}
 		}
 	default:
 		return fmt.Errorf("invalid queue key %s with unexpected type %s", queueKey, t)
@@ -140,22 +142,17 @@ func (c *nodeInformerController) sync(ctx context.Context, syncCtx factory.SyncC
 	return nil
 }
 
-func makeInsightMsgForNode(nodeInsight *NodeStatusInsight, acquiredAt metav1.Time) informerMsg {
-	uid := fmt.Sprintf("usc-node-%s", nodeInsight.Resource.Name)
+func makeInsightMsgForNode(nodeInsight *NodeStatusInsight, acquiredAt metav1.Time) (informerMsg, error) {
 	insight := WorkerPoolInsight{
-		UID:        uid,
+		UID:        fmt.Sprintf("node-%s", nodeInsight.Resource.Name),
 		AcquiredAt: acquiredAt,
 		WorkerPoolInsightUnion: WorkerPoolInsightUnion{
 			Type:              NodeStatusInsightType,
 			NodeStatusInsight: nodeInsight,
 		},
 	}
-	// Should handle errors, but ultimately we will have a proper API and won’t need to serialize ourselves
-	rawInsight, _ := yaml.Marshal(insight)
-	return informerMsg{
-		uid:     uid,
-		insight: rawInsight,
-	}
+
+	return makeWorkerPoolsInsightMsg(insight, nodesInformerName)
 }
 
 func whichMCP(node *corev1.Node, pools []*machineconfigv1.MachineConfigPool) (*machineconfigv1.MachineConfigPool, error) {
@@ -376,7 +373,8 @@ func assessNode(node *corev1.Node, mcp *machineconfigv1.MachineConfigPool, machi
 }
 
 const (
-	nodeKindName = "Node"
+	nodeKindName      = "Node"
+	nodesInformerName = "ni"
 )
 
 func parseNodeInformerControllerQueueKey(queueKey string) (string, string, error) {

pkg/updatestatus/nodeinformer_test.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	"gopkg.in/yaml.v3"
 
 	corev1 "k8s.io/api/core/v1"
@@ -915,8 +916,8 @@ func Test_sync_with_node(t *testing.T) {
 				},
 			},
 			expectedMsgs: map[string]WorkerPoolInsight{
-				"usc-node-worker-1": {
-					UID:        "usc-node-worker-1",
+				"node-worker-1": {
+					UID:        "node-worker-1",
 					AcquiredAt: now,
 					WorkerPoolInsightUnion: WorkerPoolInsightUnion{
 						Type: NodeStatusInsightType,
@@ -957,8 +958,8 @@ func Test_sync_with_node(t *testing.T) {
 				},
 			},
 			expectedMsgs: map[string]WorkerPoolInsight{
-				"usc-node-worker-1": {
-					UID:        "usc-node-worker-1",
+				"node-worker-1": {
+					UID:        "node-worker-1",
 					AcquiredAt: now,
 					WorkerPoolInsightUnion: WorkerPoolInsightUnion{
 						Type: NodeStatusInsightType,
@@ -1040,14 +1041,25 @@ func Test_sync_with_node(t *testing.T) {
 					t.Fatalf("Failed to marshal expected insight: %v", err)
 				}
 				expectedMsgs = append(expectedMsgs, informerMsg{
-					uid:     uid,
-					insight: raw,
+					informer: nodesInformerName,
+					uid:      uid,
+					insight:  raw,
 				})
 			}
 
-			if diff := cmp.Diff(expectedMsgs, actualMsgs, cmp.AllowUnexported(informerMsg{})); diff != "" {
+			ignoreOrder := cmpopts.SortSlices(func(a, b informerMsg) bool {
+				return a.uid < b.uid
+			})
+
+			if diff := cmp.Diff(expectedMsgs, actualMsgs, ignoreOrder, cmp.AllowUnexported(informerMsg{})); diff != "" {
 				t.Errorf("Sync messages differ from expected:\n%s", diff)
 			}
+
+			for _, msg := range actualMsgs {
+				if err := msg.validate(); err != nil {
+					t.Errorf("Received message is invalid: %v\nMessage content: %v", err, msg)
+				}
+			}
 		})
 	}
 }