From 9530140cc519ad4439966215569460a6846ffb1d Mon Sep 17 00:00:00 2001
From: Blake Rouse <blake.rouse@elastic.co>
Date: Mon, 11 Oct 2021 14:07:35 -0400
Subject: [PATCH] Fix self monitor to report starting and wait for fleet-server
 input to be added to policy. (#768)

(cherry picked from commit 334cd2d18c7c1204dcaf68dca491c874604ed158)
---
 internal/pkg/policy/self.go      |   9 ++-
 internal/pkg/policy/self_test.go | 112 +++++++++++++++++++++++++++----
 2 files changed, 108 insertions(+), 13 deletions(-)

diff --git a/internal/pkg/policy/self.go b/internal/pkg/policy/self.go
index 802676133..f1b87734e 100644
--- 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
index 92b07e09f..384cd92c2 100644
--- a/internal/pkg/policy/self_test.go
+++ b/internal/pkg/policy/self_test.go
@@ -71,7 +71,51 @@ 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",
 		},
@@ -79,7 +123,7 @@ func TestSelfMonitor_DefaultPolicy(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	policy := model.Policy{
+	policy = model.Policy{
 		ESDocument: model.ESDocument{
 			Id:      rId,
 			Version: 1,
@@ -88,10 +132,10 @@ 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)
 	}
@@ -99,9 +143,9 @@ func TestSelfMonitor_DefaultPolicy(t *testing.T) {
 		mm.Notify(ctx, []es.HitT{
 			{
 				Id:      rId,
-				SeqNo:   1,
+				SeqNo:   2,
 				Version: 1,
-				Source:  policyData,
+				Source:  pData,
 			},
 		})
 	}()
@@ -337,7 +381,51 @@ 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",
 		},
@@ -345,11 +433,11 @@ func TestSelfMonitor_SpecificPolicy(t *testing.T) {
 	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,
@@ -357,7 +445,7 @@ func TestSelfMonitor_SpecificPolicy(t *testing.T) {
 		RevisionIdx:        1,
 		DefaultFleetServer: true,
 	}
-	policyData, err := json.Marshal(&policy)
+	pData, err = json.Marshal(&policy)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -365,9 +453,9 @@ func TestSelfMonitor_SpecificPolicy(t *testing.T) {
 		mm.Notify(ctx, []es.HitT{
 			{
 				Id:      rId,
-				SeqNo:   1,
+				SeqNo:   2,
 				Version: 1,
-				Source:  policyData,
+				Source:  pData,
 			},
 		})
 	}()