Merge branch 'main' into perf-hostname

.github/workflows/ci.yml

@@ -114,7 +114,7 @@ jobs:
           go install github.com/goreleaser/nfpm/v2/cmd/nfpm@${{ env.NFPM_VERSION }}
           OS_RELEASE="${{ matrix.container.image }}" OS_VERSION="${{ matrix.container.version }}" \
             make integration-test
-      - name: Output Logs 
+      - name: Container Output Logs 
         if: failure()
         run: |
           docker ps -a

go.mod

@@ -19,7 +19,7 @@ require (
 	github.com/nxadm/tail v1.4.8
 	github.com/orcaman/concurrent-map v1.0.0
 	github.com/shirou/gopsutil v3.21.11+incompatible
-	github.com/shirou/gopsutil/v3 v3.23.6
+	github.com/shirou/gopsutil/v3 v3.23.8
 	github.com/sirupsen/logrus v1.9.3
 	github.com/spf13/cobra v1.7.0
 	github.com/spf13/pflag v1.0.5
@@ -295,8 +295,8 @@ require (
 	github.com/tetratelabs/wazero v1.2.1 // indirect
 	github.com/timakin/bodyclose v0.0.0-20221125081123-e39cf3fc478e // indirect
 	github.com/timonwong/loggercheck v0.9.4 // indirect
-	github.com/tklauser/go-sysconf v0.3.11 // indirect
-	github.com/tklauser/numcpus v0.6.0 // indirect
+	github.com/tklauser/go-sysconf v0.3.12 // indirect
+	github.com/tklauser/numcpus v0.6.1 // indirect
 	github.com/tomarrell/wrapcheck/v2 v2.8.1 // indirect
 	github.com/tommy-muehle/go-mnd/v2 v2.5.1 // indirect
 	github.com/toqueteos/webbrowser v1.2.0 // indirect

go.sum

@@ -824,8 +824,8 @@ github.com/shazow/go-diff v0.0.0-20160112020656-b6b7b6733b8c h1:W65qqJCIOVP4jpqP
 github.com/shazow/go-diff v0.0.0-20160112020656-b6b7b6733b8c/go.mod h1:/PevMnwAxekIXwN8qQyfc5gl2NlkB3CQlkizAbOkeBs=
 github.com/shirou/gopsutil v3.21.11+incompatible h1:+1+c1VGhc88SSonWP6foOcLhvnKlUeu/erjjvaPEYiI=
 github.com/shirou/gopsutil v3.21.11+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
-github.com/shirou/gopsutil/v3 v3.23.6 h1:5y46WPI9QBKBbK7EEccUPNXpJpNrvPuTD0O2zHEHT08=
-github.com/shirou/gopsutil/v3 v3.23.6/go.mod h1:j7QX50DrXYggrpN30W0Mo+I4/8U2UUIQrnrhqUeWrAU=
+github.com/shirou/gopsutil/v3 v3.23.8 h1:xnATPiybo6GgdRoC4YoGnxXZFRc3dqQTGi73oLvvBrE=
+github.com/shirou/gopsutil/v3 v3.23.8/go.mod h1:7hmCaBn+2ZwaZOr6jmPBZDfawwMGuo1id3C6aM8EDqQ=
 github.com/shoenig/go-m1cpu v0.1.6 h1:nxdKQNcEB6vzgA2E2bvzKIYRuNj7XNJ4S/aRSwKzFtM=
 github.com/shoenig/go-m1cpu v0.1.6/go.mod h1:1JJMcUBvfNwpq05QDQVAnx3gUHr9IYF7GNg9SUEw2VQ=
 github.com/shoenig/test v0.6.4 h1:kVTaSd7WLz5WZ2IaoM0RSzRsUD+m8wRR+5qvntpn4LU=
@@ -919,10 +919,10 @@ github.com/timakin/bodyclose v0.0.0-20221125081123-e39cf3fc478e h1:MV6KaVu/hzByH
 github.com/timakin/bodyclose v0.0.0-20221125081123-e39cf3fc478e/go.mod h1:27bSVNWSBOHm+qRp1T9qzaIpsWEP6TbUnei/43HK+PQ=
 github.com/timonwong/loggercheck v0.9.4 h1:HKKhqrjcVj8sxL7K77beXh0adEm6DLjV/QOGeMXEVi4=
 github.com/timonwong/loggercheck v0.9.4/go.mod h1:caz4zlPcgvpEkXgVnAJGowHAMW2NwHaNlpS8xDbVhTg=
-github.com/tklauser/go-sysconf v0.3.11 h1:89WgdJhk5SNwJfu+GKyYveZ4IaJ7xAkecBo+KdJV0CM=
-github.com/tklauser/go-sysconf v0.3.11/go.mod h1:GqXfhXY3kiPa0nAXPDIQIWzJbMCB7AmcWpGR8lSZfqI=
-github.com/tklauser/numcpus v0.6.0 h1:kebhY2Qt+3U6RNK7UqpYNA+tJ23IBEGKkB7JQBfDYms=
-github.com/tklauser/numcpus v0.6.0/go.mod h1:FEZLMke0lhOUG6w2JadTzp0a+Nl8PF/GFkQ5UVIcaL4=
+github.com/tklauser/go-sysconf v0.3.12 h1:0QaGUFOdQaIVdPgfITYzaTegZvdCjmYO52cSFAEVmqU=
+github.com/tklauser/go-sysconf v0.3.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI=
+github.com/tklauser/numcpus v0.6.1 h1:ng9scYS7az0Bk4OZLvrNXNSAO2Pxr1XXRAPyjhIx+Fk=
+github.com/tklauser/numcpus v0.6.1/go.mod h1:1XfjsgE2zo8GVw7POkMbHENHzVg3GzmoZ9fESEdAacY=
 github.com/tomarrell/wrapcheck/v2 v2.8.1 h1:HxSqDSN0sAt0yJYsrcYVoEeyM4aI9yAm3KQpIXDJRhQ=
 github.com/tomarrell/wrapcheck/v2 v2.8.1/go.mod h1:/n2Q3NZ4XFT50ho6Hbxg+RV1uyo2Uow/Vdm9NQcl5SE=
 github.com/tommy-muehle/go-mnd/v2 v2.5.1 h1:NowYhSdyE/1zwK9QCLeRb6USWdoif80Ie+v+yU8u1Zw=
@@ -1221,7 +1221,7 @@ golang.org/x/sys v0.3.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.4.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.9.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.11.0 h1:eG7RXZHdqOJ1i+0lgLgCpSXAp6M3LYlAo6osgSi0xOM=
 golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=

main.go

@@ -29,6 +29,7 @@ var (
 	// set at buildtime
 	commit  = ""
 	version = ""
+	env     = &core.EnvironmentType{}
 )
 
 func init() {
@@ -40,13 +41,20 @@ func main() {
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
 
-		env := &core.EnvironmentType{}
-
 		loadedConfig, err := config.GetConfig(env.GetSystemUUID())
 		if err != nil {
 			log.Fatalf("Failed to load configuration: %v", err)
 		}
 
+		eventMeta := events.NewAgentEventMeta(
+			config.MODULE,
+			version,
+			strconv.Itoa(os.Getpid()),
+			env.GetHostname(),
+			env.GetSystemUUID(),
+			loadedConfig.InstanceGroup,
+			loadedConfig.Tags)
+
 		logger.SetLogLevel(loadedConfig.Log.Level)
 		logFile := logger.SetLogFile(loadedConfig.Log.Path)
 		if logFile != nil {
@@ -79,16 +87,7 @@ func main() {
 
 		pipe := core.InitializePipe(ctx, corePlugins, extensionPlugins, agent_config.DefaultPluginSize)
 
-		event := events.NewAgentEventMeta(
-			config.MODULE,
-			version,
-			strconv.Itoa(os.Getpid()),
-			env.GetHostname(),
-			env.GetSystemUUID(),
-			loadedConfig.InstanceGroup,
-			loadedConfig.Tags)
-
-		pipe.Process(core.NewMessage(core.AgentStarted, event))
+		defer pipe.Process(core.NewMessage(core.AgentStarted, eventMeta))
 		core.HandleSignals(ctx, commander, loadedConfig, env, pipe, cancel, controller)
 
 		pipe.Run()

src/core/pipe.go

@@ -194,11 +194,11 @@ func (p *MessagePipe) GetExtensionPlugins() []ExtensionPlugin {
 
 func (p *MessagePipe) initPlugins() {
 	for _, r := range p.plugins {
-		r.Init(p)
+		go r.Init(p)
 	}
 
 	for _, r := range p.extensionPlugins {
-		r.Init(p)
+		go r.Init(p)
 	}
 }
 

src/plugins/nginx.go

@@ -113,17 +113,17 @@ func (n *Nginx) Init(pipeline core.MessagePipeInterface) {
 	log.Info("NginxBinary initializing")
 	n.messagePipeline = pipeline
 	n.nginxBinary.UpdateNginxDetailsFromProcesses(n.getNginxProccessInfo())
-	nginxDetails := n.nginxBinary.GetNginxDetailsMapFromProcesses(n.getNginxProccessInfo())
-
-	pipeline.Process(
-		core.NewMessage(core.NginxPluginConfigured, n),
-		core.NewMessage(core.NginxInstancesFound, nginxDetails),
-	)
 }
 
 // Process processes the messages from the messaging pipe
 func (n *Nginx) Process(message *core.Message) {
 	switch message.Topic() {
+	case core.AgentStarted:
+		nginxDetails := n.nginxBinary.GetNginxDetailsMapFromProcesses(n.getNginxProccessInfo())
+		n.messagePipeline.Process(
+			core.NewMessage(core.NginxPluginConfigured, n),
+			core.NewMessage(core.NginxInstancesFound, nginxDetails),
+		)
 	case core.CommNginxConfig:
 		switch cmd := message.Data().(type) {
 		case *proto.Command:
@@ -208,6 +208,7 @@ func (n *Nginx) Subscriptions() []string {
 		core.NginxConfigValidationPending,
 		core.NginxConfigValidationSucceeded,
 		core.NginxConfigValidationFailed,
+		core.AgentStarted,
 	}
 }
 

src/plugins/nginx_test.go

@@ -162,6 +162,7 @@ func TestNginxConfigApply(t *testing.T) {
 				DirectoryMap: &proto.DirectoryMap{},
 			},
 			msgTopics: []string{
+				core.AgentStarted,
 				core.CommNginxConfig,
 				core.NginxPluginConfigured,
 				core.NginxInstancesFound,
@@ -195,6 +196,7 @@ func TestNginxConfigApply(t *testing.T) {
 				DirectoryMap: &proto.DirectoryMap{},
 			},
 			msgTopics: []string{
+				core.AgentStarted,
 				core.CommNginxConfig,
 				core.NginxPluginConfigured,
 				core.NginxInstancesFound,
@@ -228,6 +230,7 @@ func TestNginxConfigApply(t *testing.T) {
 				DirectoryMap: &proto.DirectoryMap{},
 			},
 			msgTopics: []string{
+				core.AgentStarted,
 				core.CommNginxConfig,
 				core.NginxPluginConfigured,
 				core.NginxInstancesFound,
@@ -281,6 +284,7 @@ func TestNginxConfigApply(t *testing.T) {
 				},
 			},
 			msgTopics: []string{
+				core.AgentStarted,
 				core.CommNginxConfig,
 				core.NginxPluginConfigured,
 				core.NginxInstancesFound,
@@ -318,6 +322,7 @@ func TestNginxConfigApply(t *testing.T) {
 				DirectoryMap: &proto.DirectoryMap{},
 			},
 			msgTopics: []string{
+				core.AgentStarted,
 				core.CommNginxConfig,
 				core.NginxPluginConfigured,
 				core.NginxInstancesFound,
@@ -351,6 +356,7 @@ func TestNginxConfigApply(t *testing.T) {
 				DirectoryMap: &proto.DirectoryMap{},
 			},
 			msgTopics: []string{
+				core.AgentStarted,
 				core.CommNginxConfig,
 				core.NginxPluginConfigured,
 				core.NginxInstancesFound,
@@ -404,6 +410,7 @@ func TestNginxConfigApply(t *testing.T) {
 				},
 			},
 			msgTopics: []string{
+				core.AgentStarted,
 				core.CommNginxConfig,
 				core.NginxPluginConfigured,
 				core.NginxInstancesFound,
@@ -458,6 +465,7 @@ func TestNginxConfigApply(t *testing.T) {
 				},
 			},
 			msgTopics: []string{
+				core.AgentStarted,
 				core.CommNginxConfig,
 				core.NginxPluginConfigured,
 				core.NginxInstancesFound,
@@ -547,7 +555,7 @@ func TestNginxConfigApply(t *testing.T) {
 				Nginx: loadedConfig.Nginx{
 					ConfigReloadMonitoringPeriod: 5 * time.Second,
 				},
-				Features:   []string{agent_config.FeatureNginxConfig},
+				Features:   []string{agent_config.FeatureNginxConfigAsync},
 				Extensions: []string{agent_config.NginxAppProtectExtensionPlugin},
 			}
 
@@ -561,7 +569,7 @@ func TestNginxConfigApply(t *testing.T) {
 			}
 
 			messagePipe := core.SetupMockMessagePipe(t, ctx, []core.Plugin{pluginUnderTest}, []core.ExtensionPlugin{})
-
+			messagePipe.Process(core.NewMessage(core.AgentStarted, nil))
 			messagePipe.Process(core.NewMessage(core.CommNginxConfig, cmd))
 
 			go messagePipe.Run()
@@ -613,11 +621,10 @@ func TestUploadConfigs(t *testing.T) {
 	}
 
 	msgTopics := []string{
+		core.AgentStarted,
 		core.NginxPluginConfigured,
 		core.NginxInstancesFound,
 		core.DataplaneChanged,
-		core.NginxPluginConfigured,
-		core.NginxInstancesFound,
 	}
 
 	env := tutils.GetMockEnvWithProcess()
@@ -631,12 +638,18 @@ func TestUploadConfigs(t *testing.T) {
 	cmdr := tutils.NewMockCommandClient()
 	cmdr.On("Upload", mock.Anything, mock.Anything).Return(nil)
 
-	conf := &loadedConfig.Config{Server: loadedConfig.Server{Host: "127.0.0.1", GrpcPort: 9092}, Features: []string{agent_config.FeatureNginxConfig}}
+	conf := &loadedConfig.Config{Server: loadedConfig.Server{Host: "127.0.0.1", GrpcPort: 9092}, Features: []string{agent_config.FeatureNginxConfigAsync}}
 
 	pluginUnderTest := NewNginx(cmdr, binary, env, conf)
 	messagePipe := core.SetupMockMessagePipe(t, context.TODO(), []core.Plugin{pluginUnderTest}, []core.ExtensionPlugin{})
 
 	pluginUnderTest.Init(messagePipe)
+
+	// calling Run x 2 means AgentStarted finishes before the DataplaneChanged event gets processed.
+	// This is the expected order of the real MessagePipe
+	messagePipe.Process(core.NewMessage(core.AgentStarted, nil))
+	messagePipe.Run()
+
 	messagePipe.Process(core.NewMessage(core.DataplaneChanged, nil))
 	messagePipe.Run()
 
@@ -649,11 +662,10 @@ func TestUploadConfigs(t *testing.T) {
 
 func TestDisableUploadConfigs(t *testing.T) {
 	msgTopics := []string{
+		core.AgentStarted,
 		core.NginxPluginConfigured,
 		core.NginxInstancesFound,
 		core.DataplaneChanged,
-		core.NginxPluginConfigured,
-		core.NginxInstancesFound,
 	}
 
 	env := tutils.GetMockEnvWithProcess()
@@ -668,6 +680,11 @@ func TestDisableUploadConfigs(t *testing.T) {
 	messagePipe := core.SetupMockMessagePipe(t, context.TODO(), []core.Plugin{pluginUnderTest}, []core.ExtensionPlugin{})
 
 	pluginUnderTest.Init(messagePipe)
+	// calling Run x 2 means AgentStarted finishes before the DataplaneChanged event gets processed.
+	// This is the expected order of the real MessagePipe
+	messagePipe.Process(core.NewMessage(core.AgentStarted, nil))
+	messagePipe.Run()
+
 	messagePipe.Process(core.NewMessage(core.DataplaneChanged, nil))
 	messagePipe.Run()
 
@@ -691,6 +708,8 @@ func TestNginxDetailProcUpdate(t *testing.T) {
 	messagePipe := core.SetupMockMessagePipe(t, context.TODO(), []core.Plugin{pluginUnderTest}, []core.ExtensionPlugin{})
 
 	pluginUnderTest.Init(messagePipe)
+	messagePipe.Process(core.NewMessage(core.AgentStarted, nil))
+
 	messagePipe.Process(core.NewMessage(core.NginxDetailProcUpdate, tutils.GetProcesses()))
 	messagePipe.Run()
 
@@ -707,7 +726,7 @@ func TestNginxDetailProcUpdate(t *testing.T) {
 			foundMessage = true
 		}
 	}
-	assert.Len(t, processedMessages, 5)
+	assert.Len(t, processedMessages, 4)
 	assert.True(t, foundMessage)
 }
 
@@ -736,7 +755,7 @@ func TestNginx_Process_NginxConfigUpload(t *testing.T) {
 	binary.On("ReadConfig", "/var/conf", "12345", "12345678").Return(config, nil)
 
 	env := tutils.GetMockEnvWithProcess()
-	conf := &loadedConfig.Config{Server: loadedConfig.Server{Host: "127.0.0.1", GrpcPort: 9092}, Features: []string{agent_config.FeatureNginxConfig}}
+	conf := &loadedConfig.Config{Server: loadedConfig.Server{Host: "127.0.0.1", GrpcPort: 9092}, Features: []string{agent_config.FeatureNginxConfigAsync}}
 
 	pluginUnderTest := NewNginx(cmdr, binary, env, conf)
 	pluginUnderTest.Process(core.NewMessage(core.NginxConfigUpload, configDesc))
@@ -761,6 +780,7 @@ func TestNginx_Subscriptions(t *testing.T) {
 		core.NginxConfigValidationPending,
 		core.NginxConfigValidationSucceeded,
 		core.NginxConfigValidationFailed,
+		core.AgentStarted,
 	}
 	pluginUnderTest := NewNginx(nil, nil, tutils.GetMockEnvWithProcess(), &loadedConfig.Config{})
 
@@ -800,7 +820,7 @@ func TestNginx_validateConfig(t *testing.T) {
 			binary.On("ReadConfig", mock.Anything, mock.Anything, mock.Anything).Return(&proto.NginxConfig{}, nil)
 			binary.On("GetNginxDetailsMapFromProcesses", env.Processes()).Return(tutils.GetDetailsMap())
 			binary.On("UpdateNginxDetailsFromProcesses", env.Processes())
-			conf := &loadedConfig.Config{Server: loadedConfig.Server{Host: "127.0.0.1", GrpcPort: 9092}, Features: []string{agent_config.FeatureNginxConfig}}
+			conf := &loadedConfig.Config{Server: loadedConfig.Server{Host: "127.0.0.1", GrpcPort: 9092}, Features: []string{agent_config.FeatureNginxConfigAsync}}
 
 			pluginUnderTest := NewNginx(&tutils.MockCommandClient{}, binary, env, conf)
 
@@ -830,6 +850,7 @@ func TestNginx_validateConfig(t *testing.T) {
 
 func TestNginx_completeConfigApply(t *testing.T) {
 	expectedTopics := []string{
+		core.AgentStarted,
 		core.NginxConfigValidationSucceeded,
 		core.NginxPluginConfigured,
 		core.NginxInstancesFound,
@@ -883,7 +904,7 @@ func TestNginx_completeConfigApply(t *testing.T) {
 			Host:     "127.0.0.1",
 			GrpcPort: 9092,
 		},
-		Features: []string{agent_config.FeatureNginxConfig},
+		Features: []string{agent_config.FeatureNginxConfigAsync},
 		Nginx: loadedConfig.Nginx{
 			ConfigReloadMonitoringPeriod: 5 * time.Second,
 		},
@@ -919,6 +940,7 @@ func TestNginx_completeConfigApply(t *testing.T) {
 	}
 
 	messagePipe := core.SetupMockMessagePipe(t, context.TODO(), []core.Plugin{pluginUnderTest}, []core.ExtensionPlugin{})
+	messagePipe.Process(core.NewMessage(core.AgentStarted, nil))
 	messagePipe.Process(core.NewMessage(core.NginxConfigValidationSucceeded, response))
 	go messagePipe.Run()
 
@@ -941,6 +963,7 @@ func TestNginx_completeConfigApply(t *testing.T) {
 
 func TestNginx_rollbackConfigApply(t *testing.T) {
 	expectedTopics := []string{
+		core.AgentStarted,
 		core.NginxConfigValidationFailed,
 		core.NginxPluginConfigured,
 		core.NginxInstancesFound,
@@ -980,7 +1003,7 @@ func TestNginx_rollbackConfigApply(t *testing.T) {
 		},
 	)
 
-	conf := &loadedConfig.Config{Server: loadedConfig.Server{Host: "127.0.0.1", GrpcPort: 9092}, Features: []string{agent_config.FeatureNginxConfig}}
+	conf := &loadedConfig.Config{Server: loadedConfig.Server{Host: "127.0.0.1", GrpcPort: 9092}, Features: []string{agent_config.FeatureNginxConfigAsync}}
 
 	pluginUnderTest := NewNginx(commandClient, binary, env, conf)
 
@@ -1012,6 +1035,7 @@ func TestNginx_rollbackConfigApply(t *testing.T) {
 	}
 
 	messagePipe := core.SetupMockMessagePipe(t, context.TODO(), []core.Plugin{pluginUnderTest}, []core.ExtensionPlugin{})
+	messagePipe.Process(core.NewMessage(core.AgentStarted, nil))
 	messagePipe.Process(core.NewMessage(core.NginxConfigValidationFailed, response))
 	messagePipe.Run()