Merge branch 'v3' into chore/report-nap-instances

Author: dhurley

api/grpc/mpi/v1/files.pb.go

@@ -272,10 +272,10 @@ message X509Name {
     repeated string postal_code = 7 [(buf.validate.field).repeated.items.string.min_len = 1];
 
     // Serial Number (SN): Unique identifier or serial number.
-    string serial_number = 8 [(buf.validate.field).string.min_len = 0];
+    string serial_number = 8;
 
     // Common Name (CN): Typically the person’s or entity's full name.
-    string common_name = 9 [(buf.validate.field).string.min_len = 1];
+    string common_name = 9;
 
     // Parsed attributes including any non-standard attributes, as specified in RFC 2253.
     // These attributes are parsed but not marshaled by this package.

api/grpc/mpi/v1/files.proto

@@ -15,6 +15,8 @@ const (
 	ConfigUploadRequestTopic     = "config-upload-request"
 	DataPlaneResponseTopic       = "data-plane-response"
 	ConnectionCreatedTopic       = "connection-created"
+	CredentialUpdatedTopic       = "credential-updated"
+	ConnectionResetTopic         = "connection-reset"
 	ConfigApplyRequestTopic      = "config-apply-request"
 	WriteConfigSuccessfulTopic   = "write-config-successful"
 	ConfigApplySuccessfulTopic   = "config-apply-successful"

internal/bus/topics.go

@@ -59,6 +59,8 @@ func (s *NginxStubStatusScraper) ID() component.ID {
 
 func (s *NginxStubStatusScraper) Start(_ context.Context, _ component.Host) error {
 	httpClient := http.DefaultClient
+	httpClient.Timeout = s.cfg.ClientConfig.Timeout
+
 	if strings.HasPrefix(s.cfg.APIDetails.Listen, "unix:") {
 		httpClient.Transport = &http.Transport{
 			DialContext: func(_ context.Context, _, _ string) (net.Conn, error) {

internal/collector/nginxossreceiver/internal/scraper/stubstatus/stub_status_scraper.go

@@ -49,7 +49,9 @@ func newNginxPlusScraper(
 	endpoint := strings.TrimPrefix(cfg.APIDetails.URL, "unix:")
 	logger := settings.Logger
 	logger.Info("Creating NGINX Plus scraper")
+
 	httpClient := http.DefaultClient
+	httpClient.Timeout = cfg.ClientConfig.Timeout
 
 	mb := metadata.NewMetricsBuilder(cfg.MetricsBuilderConfig, settings)
 	rb := mb.NewResourceBuilder()

internal/collector/nginxplusreceiver/scraper.go

@@ -120,6 +120,10 @@ func (oc *Collector) Init(ctx context.Context, mp bus.MessagePipeInterface) erro
 		oc.processReceivers(ctx, oc.config.Collector.Receivers.OtlpReceivers)
 	}
 
+	if !oc.stopped {
+		return errors.New("OTel collector already running")
+	}
+
 	bootErr := oc.bootup(runCtx)
 	if bootErr != nil {
 		slog.ErrorContext(runCtx, "Unable to start OTel Collector", "error", bootErr)
@@ -219,6 +223,7 @@ func (oc *Collector) Close(ctx context.Context) error {
 			slog.ErrorContext(ctx, "Failed to shutdown OTel Collector", "error", err, "state", oc.service.GetState())
 		} else {
 			slog.InfoContext(ctx, "OTel Collector shutdown", "state", oc.service.GetState())
+			oc.stopped = true
 		}
 	}
 
@@ -374,6 +379,11 @@ func (oc *Collector) restartCollector(ctx context.Context) {
 	var runCtx context.Context
 	runCtx, oc.cancel = context.WithCancel(ctx)
 
+	if !oc.stopped {
+		slog.ErrorContext(ctx, "Unable to restart OTel collector, failed to stop collector")
+		return
+	}
+
 	bootErr := oc.bootup(runCtx)
 	if bootErr != nil {
 		slog.ErrorContext(runCtx, "Unable to start OTel Collector", "error", bootErr)

internal/collector/otel_collector_plugin.go

@@ -31,6 +31,7 @@ type (
 		UpdateDataPlaneStatus(ctx context.Context, resource *mpi.Resource) error
 		UpdateDataPlaneHealth(ctx context.Context, instanceHealths []*mpi.InstanceHealth) error
 		SendDataPlaneResponse(ctx context.Context, response *mpi.DataPlaneResponse) error
+		UpdateClient(client mpi.CommandServiceClient)
 		Subscribe(ctx context.Context)
 		IsConnected() bool
 		CreateConnection(ctx context.Context, resource *mpi.Resource) (*mpi.CreateConnectionResponse, error)
@@ -86,6 +87,8 @@ func (cp *CommandPlugin) Info() *bus.Info {
 
 func (cp *CommandPlugin) Process(ctx context.Context, msg *bus.Message) {
 	switch msg.Topic {
+	case bus.ConnectionResetTopic:
+		cp.processConnectionReset(ctx, msg)
 	case bus.ResourceUpdateTopic:
 		cp.processResourceUpdate(ctx, msg)
 	case bus.InstanceHealthTopic:
@@ -172,8 +175,22 @@ func (cp *CommandPlugin) processDataPlaneResponse(ctx context.Context, msg *bus.
 	}
 }
 
+func (cp *CommandPlugin) processConnectionReset(ctx context.Context, msg *bus.Message) {
+	slog.DebugContext(ctx, "Command plugin received connection reset")
+	if newConnection, ok := msg.Data.(grpc.GrpcConnectionInterface); ok {
+		err := cp.conn.Close(ctx)
+		if err != nil {
+			slog.ErrorContext(ctx, "Command plugin: unable to close connection", "error", err)
+		}
+		cp.conn = newConnection
+		cp.commandService.UpdateClient(cp.conn.CommandServiceClient())
+		slog.DebugContext(ctx, "Command service client reset successfully")
+	}
+}
+
 func (cp *CommandPlugin) Subscriptions() []string {
 	return []string{
+		bus.ConnectionResetTopic,
 		bus.ResourceUpdateTopic,
 		bus.InstanceHealthTopic,
 		bus.DataPlaneHealthResponseTopic,

internal/command/command_plugin.go

@@ -45,6 +45,7 @@ func TestCommandPlugin_Subscriptions(t *testing.T) {
 	assert.Equal(
 		t,
 		[]string{
+			bus.ConnectionResetTopic,
 			bus.ResourceUpdateTopic,
 			bus.InstanceHealthTopic,
 			bus.DataPlaneHealthResponseTopic,
@@ -142,6 +143,12 @@ func TestCommandPlugin_Process(t *testing.T) {
 	})
 	require.Equal(t, 1, fakeCommandService.UpdateDataPlaneHealthCallCount())
 	require.Equal(t, 1, fakeCommandService.SendDataPlaneResponseCallCount())
+
+	commandPlugin.Process(ctx, &bus.Message{
+		Topic: bus.ConnectionResetTopic,
+		Data:  commandPlugin.conn,
+	})
+	require.Equal(t, 1, fakeCommandService.UpdateClientCallCount())
 }
 
 func TestCommandPlugin_monitorSubscribeChannel(t *testing.T) {