Async nginx config reload

sdk/proto/command.proto

@@ -87,6 +87,25 @@ message DataplaneStatus {
   repeated NginxHealth healths = 5 [(gogoproto.jsontag) = "healths" ]; // moving to DataplaneSoftwareHealth
   repeated DataplaneSoftwareDetails dataplane_software_details = 6 [(gogoproto.jsontag) = "dataplane_software_details"];
   repeated DataplaneSoftwareHealth dataplane_software_healths = 7 [(gogoproto.jsontag) = "dataplane_software_healths"];
+  repeated AgentActivityStatus agent_activity_status = 8 [(gogoproto.jsontag) = "agent_activity_status" ];
+}
+
+message AgentActivityStatus {
+    oneof Status {
+        NginxConfigStatus nginx_config_status = 1 [(gogoproto.jsontag) = "nginx_config_status" ];
+    }
+}
+
+message NginxConfigStatus {
+  string correlation_id = 1 [(gogoproto.jsontag) = "correlation_id" ];
+  Status status = 2 [(gogoproto.jsontag) = "status" ];
+  string message = 3 [(gogoproto.jsontag) = "message" ];
+
+  enum Status {
+    PENDING = 0;
+    OK = 1;
+    ERROR = 2;
+  }
 }
 
 message DataplaneSoftwareHealth {

src/core/topics.go

@@ -19,6 +19,11 @@ const (
 	NginxWorkerProcCreated               = "nginx.worker.created"
 	NginxWorkerProcKilled                = "nginx.worker.killed"
 	NginxDetailProcUpdate                = "nginx.proc.update"
+	NginxConfigValidationPending         = "nginx.config.validation.pending"
+	NginxConfigValidationFailed          = "nginx.config.validation.failed"
+	NginxConfigValidationSucceeded       = "nginx.config.validation.succeeded"
+	NginxConfigApplyFailed               = "nginx.config.apply.failed"
+	NginxConfigApplySucceeded            = "nginx.config.apply.succeeded"
 	CommPrefix                           = "comms."
 	CommStatus                           = CommPrefix + "status"
 	CommMetrics                          = CommPrefix + "metrics"

src/plugins/dataplane_status.go

@@ -15,22 +15,23 @@ import (
 )
 
 type DataPlaneStatus struct {
-	messagePipeline core.MessagePipeInterface
-	ctx             context.Context
-	sendStatus      chan bool
-	healthTicker    *time.Ticker
-	interval        time.Duration
-	meta            *proto.Metadata
-	binary          core.NginxBinary
-	env             core.Environment
-	version         string
-	tags            *[]string
-	configDirs      string
-	lastSendDetails time.Time
-	envHostInfo     *proto.HostInfo
-	statusUrls      map[string]string
-	reportInterval  time.Duration
-	napDetails      *proto.DataplaneSoftwareDetails_AppProtectWafDetails
+	messagePipeline       core.MessagePipeInterface
+	ctx                   context.Context
+	sendStatus            chan bool
+	healthTicker          *time.Ticker
+	interval              time.Duration
+	meta                  *proto.Metadata
+	binary                core.NginxBinary
+	env                   core.Environment
+	version               string
+	tags                  *[]string
+	configDirs            string
+	lastSendDetails       time.Time
+	envHostInfo           *proto.HostInfo
+	statusUrls            map[string]string
+	reportInterval        time.Duration
+	napDetails            *proto.DataplaneSoftwareDetails_AppProtectWafDetails
+	agentActivityStatuses []*proto.AgentActivityStatus
 }
 
 const (
@@ -87,11 +88,54 @@ func (dps *DataPlaneStatus) Process(msg *core.Message) {
 	case msg.Exact(core.NginxAppProtectDetailsGenerated):
 		// If a NAP report was generated sync it
 		dps.syncNAPDetails(msg)
+
+	case msg.Exact(core.NginxConfigValidationPending):
+		log.Tracef("DataplaneStatus: %T message from topic %s received", msg.Data(), msg.Topic())
+		switch data := msg.Data().(type) {
+		case *proto.AgentActivityStatus:
+			dps.updateAgentActivityStatuses(data)
+		default:
+			log.Errorf("Expected the type %T but got %T", &proto.AgentActivityStatus{}, data)
+		}
+	case msg.Exact(core.NginxConfigApplyFailed) || msg.Exact(core.NginxConfigApplySucceeded):
+		log.Tracef("DataplaneStatus: %T message from topic %s received", msg.Data(), msg.Topic())
+		switch data := msg.Data().(type) {
+		case *proto.AgentActivityStatus:
+			dps.updateAgentActivityStatuses(data)
+			dps.sendDataplaneStatus(dps.messagePipeline, false)
+		default:
+			log.Errorf("Expected the type %T but got %T", &proto.AgentActivityStatus{}, data)
+		}
 	}
 }
 
 func (dps *DataPlaneStatus) Subscriptions() []string {
-	return []string{core.AgentConfigChanged, core.NginxAppProtectDetailsGenerated}
+	return []string{
+		core.AgentConfigChanged,
+		core.NginxAppProtectDetailsGenerated,
+		core.NginxConfigValidationPending,
+		core.NginxConfigApplyFailed,
+		core.NginxConfigApplySucceeded,
+	}
+}
+
+func (dps *DataPlaneStatus) updateAgentActivityStatuses(newAgentActivityStatus *proto.AgentActivityStatus) {
+	log.Tracef("DataplaneStatus: Adding %v to agentActivityStatuses", newAgentActivityStatus)
+	if _, ok := newAgentActivityStatus.GetStatus().(*proto.AgentActivityStatus_NginxConfigStatus); ok {
+		foundExistingNginxStatus := false
+		for index, agentActivityStatus := range dps.agentActivityStatuses {
+			if _, ok := agentActivityStatus.GetStatus().(*proto.AgentActivityStatus_NginxConfigStatus); ok {
+				dps.agentActivityStatuses[index] = newAgentActivityStatus
+				log.Tracef("DataplaneStatus: Updated agentActivityStatus with new status %v", newAgentActivityStatus)
+				foundExistingNginxStatus = true
+			}
+		}
+
+		if !foundExistingNginxStatus {
+			dps.agentActivityStatuses = append(dps.agentActivityStatuses, newAgentActivityStatus)
+			log.Tracef("DataplaneStatus: Added new status %v to agentActivityStatus", newAgentActivityStatus)
+		}
+	}
 }
 
 func (dps *DataPlaneStatus) sendDataplaneStatus(pipeline core.MessagePipeInterface, forceDetails bool) {
@@ -133,6 +177,7 @@ func (dps *DataPlaneStatus) dataplaneStatus(forceDetails bool) *proto.DataplaneS
 		Details:                  dps.detailsForProcess(processes, forceDetails),
 		Healths:                  dps.healthForProcess(processes),
 		DataplaneSoftwareDetails: dps.dataplaneSoftwareDetails(),
+		AgentActivityStatus:      dps.agentActivityStatuses,
 	}
 }
 

src/plugins/dataplane_status_test.go

@@ -18,6 +18,83 @@ import (
 )
 
 func TestDataPlaneStatus(t *testing.T) {
+	tests := []struct {
+		testName        string
+		message         *core.Message
+		expectedMessage *core.Message
+	}{
+		{
+			testName: "default status",
+			message:  nil,
+			expectedMessage: core.NewMessage(core.CommStatus, &proto.Command{
+				Meta: nil,
+				Data: &proto.Command_DataplaneStatus{
+					DataplaneStatus: &proto.DataplaneStatus{},
+				},
+			}),
+		},
+		{
+			testName: "successful nginx config apply",
+			message: core.NewMessage(core.NginxConfigApplySucceeded, &proto.AgentActivityStatus{
+				Status: &proto.AgentActivityStatus_NginxConfigStatus{
+					NginxConfigStatus: &proto.NginxConfigStatus{
+						CorrelationId: "123",
+						Status:        proto.NginxConfigStatus_OK,
+						Message:       "config applied",
+					},
+				},
+			}),
+			expectedMessage: core.NewMessage(core.CommStatus, &proto.Command{
+				Meta: nil,
+				Data: &proto.Command_DataplaneStatus{
+					DataplaneStatus: &proto.DataplaneStatus{
+						AgentActivityStatus: []*proto.AgentActivityStatus{
+							{
+								Status: &proto.AgentActivityStatus_NginxConfigStatus{
+									NginxConfigStatus: &proto.NginxConfigStatus{
+										CorrelationId: "123",
+										Status:        proto.NginxConfigStatus_OK,
+										Message:       "config applied",
+									},
+								},
+							},
+						},
+					},
+				},
+			}),
+		},
+		{
+			testName: "nginx config apply failed",
+			message: core.NewMessage(core.NginxConfigApplySucceeded, &proto.AgentActivityStatus{
+				Status: &proto.AgentActivityStatus_NginxConfigStatus{
+					NginxConfigStatus: &proto.NginxConfigStatus{
+						CorrelationId: "123",
+						Status:        proto.NginxConfigStatus_ERROR,
+						Message:       "config applied failed",
+					},
+				},
+			}),
+			expectedMessage: core.NewMessage(core.CommStatus, &proto.Command{
+				Meta: nil,
+				Data: &proto.Command_DataplaneStatus{
+					DataplaneStatus: &proto.DataplaneStatus{
+						AgentActivityStatus: []*proto.AgentActivityStatus{
+							{
+								Status: &proto.AgentActivityStatus_NginxConfigStatus{
+									NginxConfigStatus: &proto.NginxConfigStatus{
+										CorrelationId: "123",
+										Status:        proto.NginxConfigStatus_ERROR,
+										Message:       "config applied failed",
+									},
+								},
+							},
+						},
+					},
+				},
+			}),
+		},
+	}
+
 	processID := "12345"
 	detailsMap := map[string]*proto.NginxDetails{
 		processID: {
@@ -59,30 +136,30 @@ func TestDataPlaneStatus(t *testing.T) {
 	messagePipe.Run()
 	defer dataPlaneStatus.Close()
 
-	// Instance Service
-	t.Run("returns get response", func(t *testing.T) {
-		// sleep for 3 seconds
-		// check messages
-		// need to mock env
-		time.Sleep(3 * time.Second)
-		result := messagePipe.GetProcessedMessages()
-
-		expectedMsg := []string{
-			core.CommStatus,
-		}
-		assert.GreaterOrEqual(t, len(result), len(expectedMsg))
-		for idx, expMsg := range expectedMsg {
-			message := result[idx]
-			assert.Equal(t, expMsg, message.Topic())
-			if expMsg == core.CommStatus {
-				cmd := message.Data().(*proto.Command)
-				dps := cmd.Data.(*proto.Command_DataplaneStatus)
-				assert.NotNil(t, dps)
-				assert.NotNil(t, dps.DataplaneStatus.GetHost().GetHostname())
-				assert.Len(t, dps.DataplaneStatus.GetDataplaneSoftwareDetails(), 1)
+	for _, test := range tests {
+		t.Run(test.testName, func(tt *testing.T) {
+			if test.message != nil {
+				messagePipe.Process(test.message)
+				messagePipe.RunWithoutInit()
 			}
-		}
-	})
+
+			result := messagePipe.GetProcessedMessages()
+
+			message := result[len(result)-1]
+			assert.Equal(t, test.expectedMessage.Topic(), message.Topic())
+
+			cmd := message.Data().(*proto.Command)
+			dps := cmd.Data.(*proto.Command_DataplaneStatus)
+
+			expectedCmd := test.expectedMessage.Data().(*proto.Command)
+			expectedDps := expectedCmd.Data.(*proto.Command_DataplaneStatus)
+
+			assert.NotNil(t, dps)
+			assert.NotNil(t, dps.DataplaneStatus.GetHost().GetHostname())
+			assert.Len(t, dps.DataplaneStatus.GetDataplaneSoftwareDetails(), 1)
+			assert.EqualValues(t, expectedDps.DataplaneStatus.GetAgentActivityStatus(), dps.DataplaneStatus.GetAgentActivityStatus())
+		})
+	}
 }
 
 func TestDPSSyncAgentConfigChange(t *testing.T) {
@@ -245,3 +322,42 @@ func TestDPSSyncNAPDetails(t *testing.T) {
 		})
 	}
 }
+
+func TestDataPlaneSubscriptions(t *testing.T) {
+	expectedSubscriptions := []string{
+		core.AgentConfigChanged,
+		core.NginxAppProtectDetailsGenerated,
+		core.NginxConfigValidationPending,
+		core.NginxConfigApplyFailed,
+		core.NginxConfigApplySucceeded,
+	}
+
+	processID := "12345"
+
+	binary := tutils.NewMockNginxBinary()
+	binary.On("GetNginxDetailsMapFromProcesses", mock.Anything).Return(detailsMap)
+	binary.On("GetNginxIDForProcess", mock.Anything).Return(processID)
+	binary.On("GetNginxDetailsFromProcess", mock.Anything).Return(detailsMap[processID])
+
+	env := tutils.NewMockEnvironment()
+	env.On("Processes", mock.Anything).Return([]core.Process{})
+	env.On("NewHostInfo", mock.Anything, mock.Anything, mock.Anything).Return(&proto.HostInfo{
+		Hostname: "test-host",
+	})
+
+	config := &config.Config{
+		Server:     config.Server{},
+		ConfigDirs: "",
+		Log:        config.LogConfig{},
+		TLS:        config.TLSConfig{},
+		Dataplane: config.Dataplane{
+			Status: config.Status{PollInterval: time.Duration(1)},
+		},
+		AgentMetrics: config.AgentMetrics{},
+		Tags:         []string{},
+	}
+
+	dataPlaneStatus := NewDataPlaneStatus(config, grpc.NewMessageMeta(uuid.New().String()), binary, env, "")
+
+	assert.Equal(t, expectedSubscriptions, dataPlaneStatus.Subscriptions())
+}