Add token path to Agent config Command block

Makefile.packaging

@@ -10,7 +10,7 @@ BINARY_PATH         := $(BUILD_DIR)/$(BINARY_NAME)
 GPG_PUBLIC_KEY      := .key
 PACKAGE_BUILD       ?= 1
 PACKAGE_VERSION     := $(shell echo ${VERSION} | tr -d 'v')-$(PACKAGE_BUILD)
-APK_PACKAGE_VERSION := $(shell echo ${VERSION} | tr -d 'v')_$(PACKAGE_BUILD)
+APK_PACKAGE_VERSION := $(shell echo ${VERSION} | tr -d 'v').$(PACKAGE_BUILD)
 TARBALL_NAME        := $(PACKAGE_PREFIX)v3.tar.gz
 
 DEB_DISTROS         ?= ubuntu-noble-24.04 ubuntu-jammy-22.04 ubuntu-focal-20.04 debian-bookworm-12 debian-bullseye-11

internal/backoff/backoff.go

@@ -48,7 +48,7 @@ import (
 // Information from https://pkg.go.dev/github.com/cenkalti/backoff/v4#section-readme
 func WaitUntil(
 	ctx context.Context,
-	backoffSettings *config.CommonSettings,
+	backoffSettings *config.BackOff,
 	operation backoff.Operation,
 ) error {
 	eb := backoff.NewExponentialBackOff()
@@ -68,7 +68,7 @@ func WaitUntil(
 // nolint: ireturn
 func WaitUntilWithData[T any](
 	ctx context.Context,
-	backoffSettings *config.CommonSettings,
+	backoffSettings *config.BackOff,
 	operation backoff.OperationWithData[T],
 ) (T, error) {
 	backoffWithContext := Context(ctx, backoffSettings)
@@ -77,7 +77,7 @@ func WaitUntilWithData[T any](
 }
 
 // nolint: ireturn
-func Context(ctx context.Context, backoffSettings *config.CommonSettings) backoff.BackOffContext {
+func Context(ctx context.Context, backoffSettings *config.BackOff) backoff.BackOffContext {
 	eb := backoff.NewExponentialBackOff()
 	eb.InitialInterval = backoffSettings.InitialInterval
 	eb.MaxInterval = backoffSettings.MaxInterval

internal/backoff/backoff_test.go

@@ -82,14 +82,16 @@ func TestWaitUntil(t *testing.T) {
 
 	for _, test := range tests {
 		invocations = 0
-		settings := &config.CommonSettings{
-			InitialInterval:     test.initialInterval,
-			MaxInterval:         test.maxInterval,
-			MaxElapsedTime:      test.maxElapsedTime,
-			RandomizationFactor: config.DefBackoffRandomizationFactor,
-			Multiplier:          config.DefBackoffMultiplier,
+		settings := &config.Client{
+			Backoff: &config.BackOff{
+				InitialInterval:     test.initialInterval,
+				MaxInterval:         test.maxInterval,
+				MaxElapsedTime:      test.maxElapsedTime,
+				RandomizationFactor: config.DefBackoffRandomizationFactor,
+				Multiplier:          config.DefBackoffMultiplier,
+			},
 		}
-		result := WaitUntil(test.context, settings, test.operation)
+		result := WaitUntil(test.context, settings.Backoff, test.operation)
 
 		if test.expectedError {
 			assert.Errorf(t, result, test.name)
@@ -164,7 +166,7 @@ func TestWaitUntilWithData(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		settings := &config.CommonSettings{
+		settings := &config.BackOff{
 			InitialInterval:     test.initialInterval,
 			MaxInterval:         test.maxInterval,
 			MaxElapsedTime:      test.maxElapsedTime,
@@ -185,7 +187,7 @@ func TestWaitUntilWithData(t *testing.T) {
 }
 
 func TestContext(t *testing.T) {
-	settings := &config.CommonSettings{
+	settings := &config.BackOff{
 		InitialInterval:     10 * time.Millisecond,
 		MaxInterval:         10 * time.Millisecond,
 		MaxElapsedTime:      10 * time.Millisecond,

internal/collector/otel_collector_plugin.go

@@ -203,9 +203,9 @@ func (oc *Collector) Close(ctx context.Context) error {
 		oc.service.Shutdown()
 		oc.cancel()
 
-		settings := oc.config.Common
+		settings := oc.config.Client.Backoff
 		settings.MaxElapsedTime = maxTimeToWaitForShutdown
-		err := backoff.WaitUntil(ctx, oc.config.Common, func() error {
+		err := backoff.WaitUntil(ctx, settings, func() error {
 			if oc.service.GetState() == otelcol.StateClosed {
 				return nil
 			}

internal/command/command_service.go

@@ -11,7 +11,6 @@ import (
 	"log/slog"
 	"sync"
 	"sync/atomic"
-	"time"
 
 	"github.com/cenkalti/backoff/v4"
 
@@ -29,20 +28,21 @@ import (
 var _ commandService = (*CommandService)(nil)
 
 const (
-	retryInterval                  = 5 * time.Second
 	createConnectionMaxElapsedTime = 0
 )
 
 type (
 	CommandService struct {
-		commandServiceClient mpi.CommandServiceClient
-		subscribeClient      mpi.CommandService_SubscribeClient
-		agentConfig          *config.Config
-		isConnected          *atomic.Bool
-		subscribeCancel      context.CancelFunc
-		subscribeChannel     chan *mpi.ManagementPlaneRequest
-		subscribeMutex       sync.Mutex
-		subscribeClientMutex sync.Mutex
+		commandServiceClient         mpi.CommandServiceClient
+		subscribeClient              mpi.CommandService_SubscribeClient
+		agentConfig                  *config.Config
+		isConnected                  *atomic.Bool
+		subscribeCancel              context.CancelFunc
+		subscribeChannel             chan *mpi.ManagementPlaneRequest
+		configApplyRequestQueue      map[string][]*mpi.ManagementPlaneRequest // key is the instance ID
+		subscribeMutex               sync.Mutex
+		subscribeClientMutex         sync.Mutex
+		configApplyRequestQueueMutex sync.Mutex
 	}
 )
 
@@ -56,10 +56,11 @@ func NewCommandService(
 	isConnected.Store(false)
 
 	commandService := &CommandService{
-		commandServiceClient: commandServiceClient,
-		agentConfig:          agentConfig,
-		isConnected:          isConnected,
-		subscribeChannel:     subscribeChannel,
+		commandServiceClient:    commandServiceClient,
+		agentConfig:             agentConfig,
+		isConnected:             isConnected,
+		subscribeChannel:        subscribeChannel,
+		configApplyRequestQueue: make(map[string][]*mpi.ManagementPlaneRequest),
 	}
 
 	var subscribeCtx context.Context
@@ -95,7 +96,7 @@ func (cs *CommandService) UpdateDataPlaneStatus(
 		Resource: resource,
 	}
 
-	backOffCtx, backoffCancel := context.WithTimeout(ctx, cs.agentConfig.Common.MaxElapsedTime)
+	backOffCtx, backoffCancel := context.WithTimeout(ctx, cs.agentConfig.Client.Backoff.MaxElapsedTime)
 	defer backoffCancel()
 
 	sendDataPlaneStatus := func() (*mpi.UpdateDataPlaneStatusResponse, error) {
@@ -119,7 +120,7 @@ func (cs *CommandService) UpdateDataPlaneStatus(
 
 	response, err := backoff.RetryWithData(
 		sendDataPlaneStatus,
-		backoffHelpers.Context(backOffCtx, cs.agentConfig.Common),
+		backoffHelpers.Context(backOffCtx, cs.agentConfig.Client.Backoff),
 	)
 	if err != nil {
 		return err
@@ -145,12 +146,12 @@ func (cs *CommandService) UpdateDataPlaneHealth(ctx context.Context, instanceHea
 		InstanceHealths: instanceHealths,
 	}
 
-	backOffCtx, backoffCancel := context.WithTimeout(ctx, cs.agentConfig.Common.MaxElapsedTime)
+	backOffCtx, backoffCancel := context.WithTimeout(ctx, cs.agentConfig.Client.Backoff.MaxElapsedTime)
 	defer backoffCancel()
 
 	response, err := backoff.RetryWithData(
 		cs.dataPlaneHealthCallback(ctx, request),
-		backoffHelpers.Context(backOffCtx, cs.agentConfig.Common),
+		backoffHelpers.Context(backOffCtx, cs.agentConfig.Client.Backoff),
 	)
 	if err != nil {
 		return err
@@ -164,12 +165,17 @@ func (cs *CommandService) UpdateDataPlaneHealth(ctx context.Context, instanceHea
 func (cs *CommandService) SendDataPlaneResponse(ctx context.Context, response *mpi.DataPlaneResponse) error {
 	slog.DebugContext(ctx, "Sending data plane response", "response", response)
 
-	backOffCtx, backoffCancel := context.WithTimeout(ctx, cs.agentConfig.Common.MaxElapsedTime)
+	backOffCtx, backoffCancel := context.WithTimeout(ctx, cs.agentConfig.Client.Backoff.MaxElapsedTime)
 	defer backoffCancel()
 
+	err := cs.handleConfigApplyResponse(ctx, response)
+	if err != nil {
+		return err
+	}
+
 	return backoff.Retry(
 		cs.sendDataPlaneResponseCallback(ctx, response),
-		backoffHelpers.Context(backOffCtx, cs.agentConfig.Common),
+		backoffHelpers.Context(backOffCtx, cs.agentConfig.Client.Backoff),
 	)
 }
 
@@ -184,12 +190,12 @@ func (cs *CommandService) CancelSubscription(ctx context.Context) {
 }
 
 func (cs *CommandService) subscribe(ctx context.Context) {
-	commonSettings := &config.CommonSettings{
-		InitialInterval:     cs.agentConfig.Common.InitialInterval,
-		MaxInterval:         cs.agentConfig.Common.MaxInterval,
+	commonSettings := &config.BackOff{
+		InitialInterval:     cs.agentConfig.Client.Backoff.InitialInterval,
+		MaxInterval:         cs.agentConfig.Client.Backoff.MaxInterval,
 		MaxElapsedTime:      createConnectionMaxElapsedTime,
-		RandomizationFactor: cs.agentConfig.Common.RandomizationFactor,
-		Multiplier:          cs.agentConfig.Common.Multiplier,
+		RandomizationFactor: cs.agentConfig.Client.Backoff.RandomizationFactor,
+		Multiplier:          cs.agentConfig.Client.Backoff.Multiplier,
 	}
 
 	for {
@@ -223,12 +229,12 @@ func (cs *CommandService) CreateConnection(
 		Resource: resource,
 	}
 
-	commonSettings := &config.CommonSettings{
-		InitialInterval:     cs.agentConfig.Common.InitialInterval,
-		MaxInterval:         cs.agentConfig.Common.MaxInterval,
+	commonSettings := &config.BackOff{
+		InitialInterval:     cs.agentConfig.Client.Backoff.InitialInterval,
+		MaxInterval:         cs.agentConfig.Client.Backoff.MaxInterval,
 		MaxElapsedTime:      createConnectionMaxElapsedTime,
-		RandomizationFactor: cs.agentConfig.Common.RandomizationFactor,
-		Multiplier:          cs.agentConfig.Common.Multiplier,
+		RandomizationFactor: cs.agentConfig.Client.Backoff.RandomizationFactor,
+		Multiplier:          cs.agentConfig.Client.Backoff.Multiplier,
 	}
 
 	slog.DebugContext(ctx, "Sending create connection request", "request", request)
@@ -273,6 +279,81 @@ func (cs *CommandService) sendDataPlaneResponseCallback(
 	}
 }
 
+func (cs *CommandService) handleConfigApplyResponse(
+	ctx context.Context,
+	response *mpi.DataPlaneResponse,
+) error {
+	cs.configApplyRequestQueueMutex.Lock()
+	defer cs.configApplyRequestQueueMutex.Unlock()
+
+	isConfigApplyResponse := false
+	var indexOfConfigApplyRequest int
+
+	for index, configApplyRequest := range cs.configApplyRequestQueue[response.GetInstanceId()] {
+		if configApplyRequest.GetMessageMeta().GetCorrelationId() == response.GetMessageMeta().GetCorrelationId() {
+			indexOfConfigApplyRequest = index
+			isConfigApplyResponse = true
+
+			break
+		}
+	}
+
+	if isConfigApplyResponse {
+		err := cs.sendResponseForQueuedConfigApplyRequests(ctx, response, indexOfConfigApplyRequest)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (cs *CommandService) sendResponseForQueuedConfigApplyRequests(
+	ctx context.Context,
+	response *mpi.DataPlaneResponse,
+	indexOfConfigApplyRequest int,
+) error {
+	instanceID := response.GetInstanceId()
+	for i := 0; i < indexOfConfigApplyRequest; i++ {
+		newResponse := response
+
+		newResponse.GetMessageMeta().MessageId = proto.GenerateMessageID()
+
+		request := cs.configApplyRequestQueue[instanceID][i]
+		newResponse.GetMessageMeta().CorrelationId = request.GetMessageMeta().GetCorrelationId()
+
+		slog.DebugContext(
+			ctx,
+			"Sending data plane response for queued config apply request",
+			"response", newResponse,
+		)
+
+		backOffCtx, backoffCancel := context.WithTimeout(ctx, cs.agentConfig.Client.Backoff.MaxElapsedTime)
+
+		err := backoff.Retry(
+			cs.sendDataPlaneResponseCallback(ctx, newResponse),
+			backoffHelpers.Context(backOffCtx, cs.agentConfig.Client.Backoff),
+		)
+		if err != nil {
+			slog.ErrorContext(ctx, "Failed to send data plane response", "error", err)
+			backoffCancel()
+
+			return err
+		}
+
+		backoffCancel()
+	}
+
+	cs.configApplyRequestQueue[instanceID] = cs.configApplyRequestQueue[instanceID][indexOfConfigApplyRequest+1:]
+	slog.DebugContext(ctx, "Removed config apply requests from queue", "queue", cs.configApplyRequestQueue[instanceID])
+
+	if len(cs.configApplyRequestQueue[instanceID]) > 0 {
+		cs.subscribeChannel <- cs.configApplyRequestQueue[instanceID][len(cs.configApplyRequestQueue[instanceID])-1]
+	}
+
+	return nil
+}
+
 // Retry callback for sending a data plane health status to the Management Plane.
 func (cs *CommandService) dataPlaneHealthCallback(
 	ctx context.Context,
@@ -335,7 +416,25 @@ func (cs *CommandService) receiveCallback(ctx context.Context) func() error {
 			return recvError
 		}
 
-		cs.subscribeChannel <- request
+		switch request.GetRequest().(type) {
+		case *mpi.ManagementPlaneRequest_ConfigApplyRequest:
+			cs.configApplyRequestQueueMutex.Lock()
+			defer cs.configApplyRequestQueueMutex.Unlock()
+
+			instanceID := request.GetConfigApplyRequest().GetOverview().GetConfigVersion().GetInstanceId()
+			cs.configApplyRequestQueue[instanceID] = append(cs.configApplyRequestQueue[instanceID], request)
+			if len(cs.configApplyRequestQueue[instanceID]) == 1 {
+				cs.subscribeChannel <- request
+			} else {
+				slog.DebugContext(
+					ctx,
+					"Config apply request is already in progress, queuing new config apply request",
+					"request", request,
+				)
+			}
+		default:
+			cs.subscribeChannel <- request
+		}
 
 		return nil
 	}