Reintroduce changes of PR #120

Signed-off-by: Paschalis Tsilias <[email protected]>
open-telemetry · Jun 25, 2024 · 2bf6a4d · 2bf6a4d
1 parent 8f7a652
commit 2bf6a4d
Show file tree

Hide file tree

Showing 8 changed files with 29 additions and 5 deletions.
diff --git a/client/httpclient.go b/client/httpclient.go
@@ -126,6 +126,7 @@ func (c *httpClient) runUntilStopped(ctx context.Context) {
 		&c.common.ClientSyncedState,
 		c.common.PackagesStateProvider,
 		c.common.Capabilities,
+		&c.common.PackageSyncMutex,
 	)
 }
 

diff --git a/client/internal/clientcommon.go b/client/internal/clientcommon.go
@@ -41,6 +41,9 @@ type ClientCommon struct {
 	// PackagesStateProvider provides access to the local state of packages.
 	PackagesStateProvider types.PackagesStateProvider
 
+	// PackageSyncMutex makes sure only one package syncing operation happens at a time.
+	PackageSyncMutex sync.Mutex
+
 	// The transport-specific sender.
 	sender Sender
 

diff --git a/client/internal/httpsender.go b/client/internal/httpsender.go
@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -91,10 +92,11 @@ func (h *HTTPSender) Run(
 	clientSyncedState *ClientSyncedState,
 	packagesStateProvider types.PackagesStateProvider,
 	capabilities protobufs.AgentCapabilities,
+	packageSyncMutex *sync.Mutex,
 ) {
 	h.url = url
 	h.callbacks = callbacks
-	h.receiveProcessor = newReceivedProcessor(h.logger, callbacks, h, clientSyncedState, packagesStateProvider, capabilities)
+	h.receiveProcessor = newReceivedProcessor(h.logger, callbacks, h, clientSyncedState, packagesStateProvider, capabilities, packageSyncMutex)
 
 	for {
 		pollingTimer := time.NewTimer(time.Millisecond * time.Duration(atomic.LoadInt64(&h.pollingIntervalMs)))

diff --git a/client/internal/packagessyncer.go b/client/internal/packagessyncer.go
@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"sync"
 
 	"github.com/open-telemetry/opamp-go/client/types"
 	"github.com/open-telemetry/opamp-go/protobufs"
@@ -20,6 +21,7 @@ type packagesSyncer struct {
 	sender            Sender
 
 	statuses *protobufs.PackageStatuses
+	mut      *sync.Mutex
 	doneCh   chan struct{}
 }
 
@@ -30,6 +32,7 @@ func NewPackagesSyncer(
 	sender Sender,
 	clientSyncedState *ClientSyncedState,
 	packagesStateProvider types.PackagesStateProvider,
+	mut *sync.Mutex,
 ) *packagesSyncer {
 	return &packagesSyncer{
 		logger:            logger,
@@ -38,6 +41,7 @@ func NewPackagesSyncer(
 		clientSyncedState: clientSyncedState,
 		localState:        packagesStateProvider,
 		doneCh:            make(chan struct{}),
+		mut:               mut,
 	}
 }
 
@@ -49,6 +53,7 @@ func (s *packagesSyncer) Sync(ctx context.Context) error {
 	}()
 
 	// Prepare package statuses.
+	s.mut.Lock()
 	if err := s.initStatuses(); err != nil {
 		return err
 	}
@@ -99,6 +104,8 @@ func (s *packagesSyncer) initStatuses() error {
 
 // doSync performs the actual syncing process.
 func (s *packagesSyncer) doSync(ctx context.Context) {
+	defer s.mut.Unlock()
+
 	hash, err := s.localState.AllPackagesHash()
 	if err != nil {
 		s.logger.Errorf(ctx, "Package syncing failed: %V", err)

diff --git a/client/internal/receivedprocessor.go b/client/internal/receivedprocessor.go
@@ -3,6 +3,7 @@ package internal
 import (
 	"context"
 	"fmt"
+	"sync"
 
 	"github.com/open-telemetry/opamp-go/client/types"
 	"github.com/open-telemetry/opamp-go/protobufs"
@@ -24,6 +25,9 @@ type receivedProcessor struct {
 
 	packagesStateProvider types.PackagesStateProvider
 
+	// packageSyncMutex protects against multiple package syncing operations at the same time.
+	packageSyncMutex *sync.Mutex
+
 	// Agent's capabilities defined at Start() time.
 	capabilities protobufs.AgentCapabilities
 }
@@ -35,6 +39,7 @@ func newReceivedProcessor(
 	clientSyncedState *ClientSyncedState,
 	packagesStateProvider types.PackagesStateProvider,
 	capabilities protobufs.AgentCapabilities,
+	packageSyncMutex *sync.Mutex,
 ) receivedProcessor {
 	return receivedProcessor{
 		logger:                logger,
@@ -43,6 +48,7 @@ func newReceivedProcessor(
 		clientSyncedState:     clientSyncedState,
 		packagesStateProvider: packagesStateProvider,
 		capabilities:          capabilities,
+		packageSyncMutex:      packageSyncMutex,
 	}
 }
 
@@ -122,6 +128,7 @@ func (r *receivedProcessor) ProcessReceivedMessage(ctx context.Context, msg *pro
 					r.sender,
 					r.clientSyncedState,
 					r.packagesStateProvider,
+					r.packageSyncMutex,
 				)
 			} else {
 				r.logger.Debugf(ctx, "Ignoring PackagesAvailable, agent does not have AcceptsPackages capability")

diff --git a/client/internal/wsreceiver.go b/client/internal/wsreceiver.go
@@ -3,6 +3,7 @@ package internal
 import (
 	"context"
 	"fmt"
+	"sync"
 
 	"github.com/gorilla/websocket"
 	"github.com/open-telemetry/opamp-go/client/types"
@@ -32,13 +33,14 @@ func NewWSReceiver(
 	clientSyncedState *ClientSyncedState,
 	packagesStateProvider types.PackagesStateProvider,
 	capabilities protobufs.AgentCapabilities,
+	packageSyncMutex *sync.Mutex,
 ) *wsReceiver {
 	w := &wsReceiver{
 		conn:      conn,
 		logger:    logger,
 		sender:    sender,
 		callbacks: callbacks,
-		processor: newReceivedProcessor(logger, callbacks, sender, clientSyncedState, packagesStateProvider, capabilities),
+		processor: newReceivedProcessor(logger, callbacks, sender, clientSyncedState, packagesStateProvider, capabilities, packageSyncMutex),
 		stopped:   make(chan struct{}),
 	}
 

diff --git a/client/internal/wsreceiver_test.go b/client/internal/wsreceiver_test.go
@@ -3,6 +3,7 @@ package internal
 import (
 	"context"
 	"fmt"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -87,7 +88,7 @@ func TestServerToAgentCommand(t *testing.T) {
 			}
 			sender := WSSender{}
 			capabilities := protobufs.AgentCapabilities_AgentCapabilities_AcceptsRestartCommand
-			receiver := NewWSReceiver(TestLogger{t}, callbacks, nil, &sender, &clientSyncedState, nil, capabilities)
+			receiver := NewWSReceiver(TestLogger{t}, callbacks, nil, &sender, &clientSyncedState, nil, capabilities, new(sync.Mutex))
 			receiver.processor.ProcessReceivedMessage(context.Background(), &protobufs.ServerToAgent{
 				Command: test.command,
 			})
@@ -141,7 +142,7 @@ func TestServerToAgentCommandExclusive(t *testing.T) {
 			},
 		}
 		clientSyncedState := ClientSyncedState{}
-		receiver := NewWSReceiver(TestLogger{t}, callbacks, nil, nil, &clientSyncedState, nil, test.capabilities)
+		receiver := NewWSReceiver(TestLogger{t}, callbacks, nil, nil, &clientSyncedState, nil, test.capabilities, new(sync.Mutex))
 		receiver.processor.ProcessReceivedMessage(context.Background(), &protobufs.ServerToAgent{
 			Command: &protobufs.ServerToAgentCommand{
 				Type: protobufs.CommandType_CommandType_Restart,
@@ -204,7 +205,7 @@ func TestReceiverLoopStop(t *testing.T) {
 	}
 	sender := WSSender{}
 	capabilities := protobufs.AgentCapabilities_AgentCapabilities_AcceptsRestartCommand
-	receiver := NewWSReceiver(TestLogger{t}, callbacks, conn, &sender, &clientSyncedState, nil, capabilities)
+	receiver := NewWSReceiver(TestLogger{t}, callbacks, conn, &sender, &clientSyncedState, nil, capabilities, new(sync.Mutex))
 	ctx, cancel := context.WithCancel(context.Background())
 
 	go func() {

diff --git a/client/wsclient.go b/client/wsclient.go
@@ -282,6 +282,7 @@ func (c *wsClient) runOneCycle(ctx context.Context) {
 		&c.common.ClientSyncedState,
 		c.common.PackagesStateProvider,
 		c.common.Capabilities,
+		&c.common.PackageSyncMutex,
 	)
 
 	// When the wsclient is closed, the context passed to runOneCycle will be canceled.