Repository: open-telemetry/opamp-go
Pull request #405 (2025-06-27 17:50:59)

Add DownloadHTTPClient callback
Package Syncer uses the http.DefaultClient and gives users not chance of configuring an HTTP client for the purpose of downloading packages. This means users cannot configure TLS, timeouts, keep-alive, use a custom transport, a proxy, etc to faciliate package file downloads.

This commit add a new DownloadHTTPClient callback that allows users to provide a custom HTTP Client to use with package downloads.

The callback will be called for every DownloadableFile. This is to allow users to tweak the client based on the file URL. Users who do not need to tweak the client for each download can always return the same client. We could call the callback just once per agent and use the same client for all downloads but I felt this gives users more flexibility and makes it very clear that the client is only for downloading files. Happy to change it to a "global" download client though. 

tigrannajaryan: Please fix the build.

Patch:
diff --git a/client/internal/packagessyncer.go b/client/internal/packagessyncer.go
index 2531961..158a186 100644
--- a/client/internal/packagessyncer.go
+++ b/client/internal/packagessyncer.go
@@ -24,9 +24,10 @@ type packagesSyncer struct {
 	sender            Sender
 	reporterInterval  time.Duration
 
-	statuses *protobufs.PackageStatuses
-	mux      *sync.Mutex
-	doneCh   chan struct{}
+	httpClientFactory func(context.Context, *protobufs.DownloadableFile) (*http.Client, error)
+	statuses          *protobufs.PackageStatuses
+	mux               *sync.Mutex
+	doneCh            chan struct{}
 }
 
 // NewPackagesSyncer creates a new packages syncer.
@@ -38,7 +39,11 @@ func NewPackagesSyncer(
 	packagesStateProvider types.PackagesStateProvider,
 	mux *sync.Mutex,
 	reporterInterval time.Duration,
-) *packagesSyncer {
+	httpClientFactory func(context.Context, *protobufs.DownloadableFile) (*http.Client, error),
+) (*packagesSyncer, error) {
+	if httpClientFactory == nil {
+		return nil, fmt.Errorf("httpClientFactory must not be nil")
+	}
 	return &packagesSyncer{
 		logger:            logger,
 		available:         available,
@@ -48,7 +53,8 @@ func NewPackagesSyncer(
 		doneCh:            make(chan struct{}),
 		mux:               mux,
 		reporterInterval:  reporterInterval,
-	}
+		httpClientFactory: httpClientFactory,
+	}, nil
 }
 
 // Sync performs the package syncing process.
@@ -295,7 +301,12 @@ func (s *packagesSyncer) downloadFile(ctx context.Context, pkgName string, file
 		}
 	}
 
-	resp, err := http.DefaultClient.Do(req)
+	client, err := s.httpClientFactory(ctx, file)
+	if err != nil {
+		return fmt.Errorf("failed to create an HTTP Client to download file from %s: %v", file.DownloadUrl, err)
+	}
+
+	resp, err := client.Do(req)
 	if err != nil {
 		return fmt.Errorf("cannot download file from %s: %v", file.DownloadUrl, err)
 	}
diff --git a/client/internal/receivedprocessor.go b/client/internal/receivedprocessor.go
index cbc23c2..5e53b71 100644
--- a/client/internal/receivedprocessor.go
+++ b/client/internal/receivedprocessor.go
@@ -128,7 +128,7 @@ func (r *receivedProcessor) ProcessReceivedMessage(ctx context.Context, msg *pro
 	if msg.PackagesAvailable != nil {
 		if r.hasCapability(protobufs.AgentCapabilities_AgentCapabilities_AcceptsPackages) {
 			msgData.PackagesAvailable = msg.PackagesAvailable
-			msgData.PackageSyncer = NewPackagesSyncer(
+			pkgSyncer, err := NewPackagesSyncer(
 				r.logger,
 				msgData.PackagesAvailable,
 				r.sender,
@@ -136,7 +136,13 @@ func (r *receivedProcessor) ProcessReceivedMessage(ctx context.Context, msg *pro
 				r.packagesStateProvider,
 				r.packageSyncMutex,
 				r.downloadReporterInt,
+				r.callbacks.DownloadHTTPClient,
 			)
+			if err != nil {
+				r.logger.Errorf(ctx, "failed to create package syncer: %v", err)
+			} else {
+				msgData.PackageSyncer = pkgSyncer
+			}
 		} else {
 			r.logger.Debugf(ctx, "Ignoring PackagesAvailable, agent does not have AcceptsPackages capability")
 		}
diff --git a/client/types/callbacks.go b/client/types/callbacks.go
index 48d5f83..25cbe6e 100644
--- a/client/types/callbacks.go
+++ b/client/types/callbacks.go
@@ -130,6 +130,11 @@ type Callbacks struct {
 	//
 	// The responses in the via parameter are passed with their bodies closed.
 	CheckRedirect func(req *http.Request, viaReq []*http.Request, via []*http.Response) error
+
+	// DownloadHTTPClient is called to create an HTTP client that is used to download files by the package syncer.
+	// If the callback is not set, a default HTTP client will be created with the default transport settings.
+	// The callback must return a non-nil HTTP client or an error.
+	DownloadHTTPClient func(ctx context.Context, file *protobufs.DownloadableFile) (*http.Client, error)
 }
 
 func (c *Callbacks) SetDefaults() {
@@ -157,4 +162,20 @@ func (c *Callbacks) SetDefaults() {
 	if c.SaveRemoteConfigStatus == nil {
 		c.SaveRemoteConfigStatus = func(ctx context.Context, status *protobufs.RemoteConfigStatus) {}
 	}
+	if c.DownloadHTTPClient == nil {
+		defaultHttpClient := newHttpClient()
+		c.DownloadHTTPClient = func(ctx context.Context, file *protobufs.DownloadableFile) (*http.Client, error) {
+			return defaultHttpClient, nil
+		}
+	}
+}
+
+func newHttpClient() *http.Client {
+	client := &http.Client{}
+	// Clone the default transport to avoid being affected by global state changes.
+	if tr, ok := http.DefaultTransport.(*http.Transport); ok {
+		tr = tr.Clone()
+		client.Transport = tr
+	}
+	return client
 }


Test patch:
diff --git a/client/internal/httpsender_test.go b/client/internal/httpsender_test.go
index a6c9869..c6a3f86 100644
--- a/client/internal/httpsender_test.go
+++ b/client/internal/httpsender_test.go
@@ -12,12 +12,11 @@ import (
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/assert"
-
 	"github.com/open-telemetry/opamp-go/client/types"
 	sharedinternal "github.com/open-telemetry/opamp-go/internal"
 	"github.com/open-telemetry/opamp-go/internal/testhelpers"
 	"github.com/open-telemetry/opamp-go/protobufs"
+	"github.com/stretchr/testify/assert"
 )
 
 func TestHTTPSenderRetryForStatusTooManyRequests(t *testing.T) {
@@ -247,14 +246,15 @@ func TestPackageUpdatesInParallel(t *testing.T) {
 
 	var messages atomic.Int32
 	var mux sync.Mutex
-	sender.callbacks = types.Callbacks{
-		OnMessage: func(ctx context.Context, msg *types.MessageData) {
-			err := msg.PackageSyncer.Sync(ctx)
-			assert.NoError(t, err)
-			messages.Add(1)
-			doneCh = append(doneCh, msg.PackageSyncer.Done())
-		},
+	callbacks := types.Callbacks{}
+	callbacks.SetDefaults()
+	callbacks.OnMessage = func(ctx context.Context, msg *types.MessageData) {
+		err := msg.PackageSyncer.Sync(ctx)
+		assert.NoError(t, err)
+		messages.Add(1)
+		doneCh = append(doneCh, msg.PackageSyncer.Done())
 	}
+	sender.callbacks = callbacks
 
 	clientSyncedState := &ClientSyncedState{}
 	capabilities := protobufs.AgentCapabilities_AgentCapabilities_AcceptsPackages
@@ -319,14 +319,16 @@ func TestPackageUpdatesWithError(t *testing.T) {
 	localPackageState := types.PackagesStateProvider(nil)
 	var messages atomic.Int32
 	var mux sync.Mutex
-	sender.callbacks = types.Callbacks{
-		OnMessage: func(ctx context.Context, msg *types.MessageData) {
-			// Make sure the call to Sync will return an error due to a nil PackageStateProvider
-			err := msg.PackageSyncer.Sync(ctx)
-			assert.Error(t, err)
-			messages.Add(1)
-		},
+
+	callbacks := types.Callbacks{}
+	callbacks.SetDefaults()
+	callbacks.OnMessage = func(ctx context.Context, msg *types.MessageData) {
+		// Make sure the call to Sync will return an error due to a nil PackageStateProvider
+		err := msg.PackageSyncer.Sync(ctx)
+		assert.Error(t, err)
+		messages.Add(1)
 	}
+	sender.callbacks = callbacks
 
 	clientSyncedState := &ClientSyncedState{}
 
diff --git a/client/internal/packagessyncer_test.go b/client/internal/packagessyncer_test.go
new file mode 100644
index 0000000..7a06b38
--- /dev/null
+++ b/client/internal/packagessyncer_test.go
@@ -0,0 +1,53 @@
+package internal
+
+import (
+	"context"
+	"net/http"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/open-telemetry/opamp-go/protobufs"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewPackagesSyncer(t *testing.T) {
+	tests := []struct {
+		name          string
+		clientFactory func(context.Context, *protobufs.DownloadableFile) (*http.Client, error)
+		err           string
+	}{
+		{
+			name:          "nil client factory",
+			clientFactory: nil,
+			err:           "httpClientFactory must not be nil",
+		},
+		{
+			name: "non-nil client factory",
+			clientFactory: func(context.Context, *protobufs.DownloadableFile) (*http.Client, error) {
+				return &http.Client{}, nil
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s, err := NewPackagesSyncer(
+				nil,
+				&protobufs.PackagesAvailable{},
+				nil,
+				&ClientSyncedState{},
+				&InMemPackagesStore{},
+				&sync.Mutex{},
+				time.Second,
+				tt.clientFactory,
+			)
+			if tt.err != "" {
+				assert.EqualError(t, err, tt.err)
+				assert.Nil(t, s)
+			} else {
+				assert.NoError(t, err)
+				assert.NotNil(t, s)
+			}
+		})
+	}
+}
diff --git a/client/internal/wsreceiver_test.go b/client/internal/wsreceiver_test.go
index 17c1677..0f0fef7 100644
--- a/client/internal/wsreceiver_test.go
+++ b/client/internal/wsreceiver_test.go
@@ -235,13 +235,13 @@ func TestWSPackageUpdatesInParallel(t *testing.T) {
 			<-blockSyncCh
 		}
 	}
-	callbacks := types.Callbacks{
-		OnMessage: func(ctx context.Context, msg *types.MessageData) {
-			err := msg.PackageSyncer.Sync(ctx)
-			assert.NoError(t, err)
-			messages.Add(1)
-			doneCh = append(doneCh, msg.PackageSyncer.Done())
-		},
+	callbacks := types.Callbacks{}
+	callbacks.SetDefaults()
+	callbacks.OnMessage = func(ctx context.Context, msg *types.MessageData) {
+		err := msg.PackageSyncer.Sync(ctx)
+		assert.NoError(t, err)
+		messages.Add(1)
+		doneCh = append(doneCh, msg.PackageSyncer.Done())
 	}
 	clientSyncedState := &ClientSyncedState{}
 	capabilities := protobufs.AgentCapabilities_AgentCapabilities_AcceptsPackages
diff --git a/client/types/callbacks_test.go b/client/types/callbacks_test.go
new file mode 100644
index 0000000..b03b7e8
--- /dev/null
+++ b/client/types/callbacks_test.go
@@ -0,0 +1,42 @@
+package types
+
+import (
+	"context"
+	"net/http"
+	"testing"
+
+	"github.com/open-telemetry/opamp-go/protobufs"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCallbacksDefaults(t *testing.T) {
+	c := Callbacks{}
+
+	c.SetDefaults()
+
+	assert.NotNil(t, c.OnConnect)
+	assert.NotNil(t, c.OnConnectFailed)
+	assert.NotNil(t, c.OnError)
+	assert.NotNil(t, c.OnMessage)
+	assert.NotNil(t, c.OnOpampConnectionSettings)
+	assert.NotNil(t, c.OnCommand)
+	assert.NotNil(t, c.GetEffectiveConfig)
+	assert.NotNil(t, c.SaveRemoteConfigStatus)
+
+	// Test default DownloadHTTPClient
+	require.NotNil(t, c.DownloadHTTPClient)
+	client, err := c.DownloadHTTPClient(context.Background(), &protobufs.DownloadableFile{})
+	require.NoError(t, err)
+	require.NotNil(t, client)
+
+	// ensure transport was set to *http.Transport
+	_, ok := client.Transport.(*http.Transport)
+	require.True(t, ok, "Expected the transport to be of type *http.Transport")
+
+	// ensure it returns the same client on subsequent calls
+	client2, err := c.DownloadHTTPClient(context.Background(), &protobufs.DownloadableFile{})
+	require.NoError(t, err)
+	require.NotNil(t, client)
+	assert.Equal(t, client, client2)
+}