Fix self monitor to report starting and wait for fleet-server input t…

…o be added to policy.
elastic · Oct 11, 2021 · b8f6877 · b8f6877
1 parent cb956b8
commit b8f6877
Show file tree

Hide file tree

Showing 2 changed files with 108 additions and 13 deletions.
diff --git a/internal/pkg/policy/self.go b/internal/pkg/policy/self.go
@@ -218,7 +218,14 @@ func (m *selfMonitorT) updateStatus(ctx context.Context) (proto.StateObserved_St
 		return proto.StateObserved_FAILED, err
 	}
 	if !data.HasType("fleet-server") {
-		return proto.StateObserved_FAILED, errors.New("assigned policy does not have fleet-server input")
+		// no fleet-server input
+		m.status = proto.StateObserved_STARTING
+		if m.policyId == "" {
+			m.reporter.Status(proto.StateObserved_STARTING, "Waiting on fleet-server input to be added to default policy", nil)
+		} else {
+			m.reporter.Status(proto.StateObserved_STARTING, fmt.Sprintf("Waiting on fleet-server input to be added to policy: %s", m.policyId), nil)
+		}
+		return proto.StateObserved_STARTING, nil
 	}
 
 	status := proto.StateObserved_HEALTHY

diff --git a/internal/pkg/policy/self_test.go b/internal/pkg/policy/self_test.go
@@ -71,15 +71,59 @@ func TestSelfMonitor_DefaultPolicy(t *testing.T) {
 
 	policyId := uuid.Must(uuid.NewV4()).String()
 	rId := xid.New().String()
-	policyContents, err := json.Marshal(&policyData{Inputs: []policyInput{
+	policyContents, err := json.Marshal(&policyData{Inputs: []policyInput{}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	policy := model.Policy{
+		ESDocument: model.ESDocument{
+			Id:      rId,
+			Version: 1,
+			SeqNo:   1,
+		},
+		PolicyId:           policyId,
+		CoordinatorIdx:     1,
+		Data:               policyContents,
+		RevisionIdx:        1,
+		DefaultFleetServer: true,
+	}
+	pData, err := json.Marshal(&policy)
+	if err != nil {
+		t.Fatal(err)
+	}
+	go func() {
+		mm.Notify(ctx, []es.HitT{
+			{
+				Id:      rId,
+				SeqNo:   1,
+				Version: 1,
+				Source:  pData,
+			},
+		})
+	}()
+
+	// should still be set to starting
+	ftesting.Retry(t, ctx, func(ctx context.Context) error {
+		status, msg, _ := reporter.Current()
+		if status != proto.StateObserved_STARTING {
+			return fmt.Errorf("should be reported as starting; instead its %s", status)
+		}
+		if msg != "Waiting on fleet-server input to be added to default policy" {
+			return fmt.Errorf("should be matching with default policy")
+		}
+		return nil
+	})
+
+	rId = xid.New().String()
+	policyContents, err = json.Marshal(&policyData{Inputs: []policyInput{
 		{
 			Type: "fleet-server",
 		},
 	}})
 	if err != nil {
 		t.Fatal(err)
 	}
-	policy := model.Policy{
+	policy = model.Policy{
 		ESDocument: model.ESDocument{
 			Id:      rId,
 			Version: 1,
@@ -88,20 +132,20 @@ func TestSelfMonitor_DefaultPolicy(t *testing.T) {
 		PolicyId:           policyId,
 		CoordinatorIdx:     1,
 		Data:               policyContents,
-		RevisionIdx:        1,
+		RevisionIdx:        2,
 		DefaultFleetServer: true,
 	}
-	policyData, err := json.Marshal(&policy)
+	pData, err = json.Marshal(&policy)
 	if err != nil {
 		t.Fatal(err)
 	}
 	go func() {
 		mm.Notify(ctx, []es.HitT{
 			{
 				Id:      rId,
-				SeqNo:   1,
+				SeqNo:   2,
 				Version: 1,
-				Source:  policyData,
+				Source:  pData,
 			},
 		})
 	}()
@@ -337,37 +381,81 @@ func TestSelfMonitor_SpecificPolicy(t *testing.T) {
 	}, ftesting.RetrySleep(1*time.Second))
 
 	rId := xid.New().String()
-	policyContents, err := json.Marshal(&policyData{Inputs: []policyInput{
+	policyContents, err := json.Marshal(&policyData{Inputs: []policyInput{}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	policy := model.Policy{
+		ESDocument: model.ESDocument{
+			Id:      rId,
+			Version: 1,
+			SeqNo:   1,
+		},
+		PolicyId:           policyId,
+		CoordinatorIdx:     1,
+		Data:               policyContents,
+		RevisionIdx:        2,
+		DefaultFleetServer: true,
+	}
+	pData, err := json.Marshal(&policy)
+	if err != nil {
+		t.Fatal(err)
+	}
+	go func() {
+		mm.Notify(ctx, []es.HitT{
+			{
+				Id:      rId,
+				SeqNo:   1,
+				Version: 1,
+				Source:  pData,
+			},
+		})
+	}()
+
+	// should still be set to starting
+	ftesting.Retry(t, ctx, func(ctx context.Context) error {
+		status, msg, _ := reporter.Current()
+		if status != proto.StateObserved_STARTING {
+			return fmt.Errorf("should be reported as starting; instead its %s", status)
+		}
+		if msg != fmt.Sprintf("Waiting on fleet-server input to be added to policy: %s", policyId) {
+			return fmt.Errorf("should be matching with specific policy")
+		}
+		return nil
+	}, ftesting.RetrySleep(1*time.Second))
+
+	rId = xid.New().String()
+	policyContents, err = json.Marshal(&policyData{Inputs: []policyInput{
 		{
 			Type: "fleet-server",
 		},
 	}})
 	if err != nil {
 		t.Fatal(err)
 	}
-	policy := model.Policy{
+	policy = model.Policy{
 		ESDocument: model.ESDocument{
 			Id:      rId,
 			Version: 1,
-			SeqNo:   1,
+			SeqNo:   2,
 		},
 		PolicyId:           policyId,
 		CoordinatorIdx:     1,
 		Data:               policyContents,
 		RevisionIdx:        1,
 		DefaultFleetServer: true,
 	}
-	policyData, err := json.Marshal(&policy)
+	pData, err = json.Marshal(&policy)
 	if err != nil {
 		t.Fatal(err)
 	}
 	go func() {
 		mm.Notify(ctx, []es.HitT{
 			{
 				Id:      rId,
-				SeqNo:   1,
+				SeqNo:   2,
 				Version: 1,
-				Source:  policyData,
+				Source:  pData,
 			},
 		})
 	}()