Client reports download status and download details

client/clientimpl_test.go

@@ -1527,6 +1527,11 @@ func TestUpdatePackages(t *testing.T) {
 	errorOnCallback.errorOnCallback = true
 	tests = append(tests, errorOnCallback)
 
+	// Check that the downloading status is sent
+	downloading := createPackageTestCase("download status set", downloadSrv)
+	downloading.expectedStatus.Packages["package1"].Status = protobufs.PackageStatusEnum_PackageStatusEnum_Downloading
+	tests = append(tests, downloading)
+
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			verifyUpdatePackages(t, test)

client/internal/clientcommon.go

@@ -5,6 +5,7 @@
 	"errors"
 	"fmt"
 	"sync"
+	"time"
 
 	"google.golang.org/protobuf/proto"
 
@@ -59,6 +60,8 @@
 
 	// Indicates that the Client is fully stopped.
 	stoppedSignal chan struct{}
+
+	downloadReporterInterval time.Duration
 }
 
 // NewClientCommon creates a new ClientCommon.
@@ -144,9 +147,15 @@
 		return err
 	}
 
+	if settings.DownloadReporterInterval != nil && settings.DownloadReporterInterval < time.Second {
+		c.downloadReporterInterval = time.Second
+	} else if settings.DownloadReporterInterval != nil {
+		c.downloadReporterInterval = settings.DownloadReporterInterval
+	}
+
 	return nil
 }
 
 // Stop stops the client. It returns an error if the client is not started.
 func (c *ClientCommon) Stop(ctx context.Context) error {
 	if !c.isStarted {

client/internal/httpsender.go

@@ -96,7 +96,7 @@
 ) {
 	h.url = url
 	h.callbacks = callbacks
-	h.receiveProcessor = newReceivedProcessor(h.logger, callbacks, h, clientSyncedState, packagesStateProvider, capabilities, packageSyncMutex)
+	h.receiveProcessor = newReceivedProcessor(h.logger, callbacks, h, clientSyncedState, packagesStateProvider, capabilities, packageSyncMutex, c.common.downloadReporterInterval)
 
 	// we need to detect if the redirect was ever set, if not, we want default behaviour
 	if callbacks.CheckRedirect != nil {

client/internal/inmempackagestore.go

@@ -3,6 +3,7 @@ package internal
 import (
 	"context"
 	"io"
+	"maps"
 
 	"github.com/open-telemetry/opamp-go/client/types"
 	"github.com/open-telemetry/opamp-go/protobufs"
@@ -92,7 +93,7 @@ func (l *InMemPackagesStore) SetAllPackagesHash(hash []byte) error {
 }
 
 func (l *InMemPackagesStore) GetContent() map[string][]byte {
-	return l.fileContents
+	return maps.Clone(l.fileContents)
 }
 
 func (l *InMemPackagesStore) GetSignature() map[string][]byte {

client/internal/package_download_details_reporter.go

@@ -0,0 +1,68 @@
+package internal
+
+import (
+	"context"
+	"sync/atomic"
+	"time"
+)
+
+const downloadReporterDefaultInterval = time.Second * 10
+
+type downloadReporter struct {
+	start         time.Time
+	interval      time.Duration
+	packageLength float64
+
+	downloaded atomic.Uint64
+
+	done chan struct{}
+}
+
+func newDownloadReporter(interval time.Duration, length int) *downloadReporter {
+	if interval <= 0 {
+		interval = downloadReporterDefaultInterval
+	}
+	return &downloadReporter{
+		start:         time.Now(),
+		interval:      interval,
+		packageLength: float64(length),
+		done:          make(chan struct{}),
+	}
+}
+
+// Write tracks the number of bytes downloaded. It will never return an error.
+func (p *downloadReporter) Write(b []byte) (int, error) {
+	n := len(b)
+	p.downloaded.Add(uint64(n))
+	return n, nil
+}
+
+// report periodically calls the passed function to with the download percent and rate to update the status of a package.
+func (p *downloadReporter) report(ctx context.Context, updateFn func(context.Context, float, float) error) {
+	go func() {
+		ticker := time.NewTicker(p.interval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-p.done:
+				return
+			case <-ticker.C:
+				downloadTime := time.Now().Sub(p.start)
+				downloaded := float64(p.downloaded.Load())
+				bps := downloaded / float64(downloadTime/time.Second)
+				var downloadPercent float64
+				if p.packageLength > 0 {
+					downloadPercent = downloaded / p.packageLength * 100
+				}
+				_ = updateFn(ctx, downloadPercent, bps)
+			}
+		}
+	}()
+}
+
+// stop the downloadReporter report goroutine
+func (p *downloadReporter) stop() {
+	close(p.done)
+}

client/internal/packagessyncer.go

@@ -5,8 +5,11 @@
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
+	"strconv"
 	"sync"
+	"time"
 
 	"github.com/open-telemetry/opamp-go/client/types"
 	"github.com/open-telemetry/opamp-go/protobufs"
@@ -19,6 +22,7 @@
 	clientSyncedState *ClientSyncedState
 	localState        types.PackagesStateProvider
 	sender            Sender
+	reporterInterval  time.Duration
 
 	statuses *protobufs.PackageStatuses
 	mux      *sync.Mutex
@@ -33,6 +37,7 @@
 	clientSyncedState *ClientSyncedState,
 	packagesStateProvider types.PackagesStateProvider,
 	mux *sync.Mutex,
+	reporterInterval time.Duration,
 ) *packagesSyncer {
 	return &packagesSyncer{
 		logger:            logger,
@@ -42,6 +47,7 @@
 		localState:        packagesStateProvider,
 		doneCh:            make(chan struct{}),
 		mux:               mux,
+		reporterInterval:  reporterInterval,
 	}
 }
 
@@ -273,6 +279,10 @@
 
 // downloadFile downloads the file from the server.
 func (s *packagesSyncer) downloadFile(ctx context.Context, pkgName string, file *protobufs.DownloadableFile) error {
+	status := s.statuses.Packages[pkgName]
+	status.Status = protobufs.PackageStatusEnum_PackageStatusEnum_Downloading
+	_ = s.reportStatuses(ctx, true)
+
 	s.logger.Debugf(ctx, "Downloading package %s file from %s", pkgName, file.DownloadUrl)
 
 	req, err := http.NewRequestWithContext(ctx, "GET", file.DownloadUrl, nil)
@@ -290,13 +300,37 @@
 		return fmt.Errorf("cannot download file from %s, HTTP response=%v", file.DownloadUrl, resp.StatusCode)
 	}
 
-	err = s.localState.UpdateContent(ctx, pkgName, resp.Body, file.ContentHash, file.Signature)
+	// Package length is required to be able to report download percent.
+	packageLength := -1
+	if contentLength := resp.Header.Get("Content-Length"); contentLength != "" {
+		if length, err := strconv.Atoi(contentLength); err == nil {
+			packageLength = length
+		}
+	}
+	// start the download reporter
+	detailsReporter := newDownloadReporter(s.reporterInterval, packageLength)
+	detailsReporter.report(ctx, s.getDownloadDetailsFn(pkgName))
+	defer detailsReporter.stop()
+
+	tr := io.TeeReader(resp.Body, detailsReporter)
+	err = s.localState.UpdateContent(ctx, pkgName, tr, file.ContentHash, file.Signature)
 	if err != nil {
 		return fmt.Errorf("failed to install/update the package %s downloaded from %s: %v", pkgName, file.DownloadUrl, err)
 	}
 	return nil
 }
 
+func (s *packagesSyncer) getDownloadDetailsFn(pkgName string) func(context.Context, float64, float64) error {
+	return func(ctx context.Context, percent, rate float64) error {
+		status := s.statuses.Packages[pkgName]
+		status.DownloadDetails = &protobufs.PackageDownloadDetails{
+			DownloadPercent:        percent,
+			DownloadBytesPerSecond: rate,
+		}
+		return s.reportStatuses(ctx, true)
+	}
+}
+
 // deleteUnneededLocalPackages deletes local packages that are not
 // needed anymore. This is done by comparing the local package state
 // with the server's package state.

client/internal/receivedprocessor.go

@@ -31,6 +31,10 @@ type receivedProcessor struct {
 
 	// Agent's capabilities defined at Start() time.
 	capabilities protobufs.AgentCapabilities
+
+	// Download reporter interval value
+	// a negative number indicates that the default should be used instead.
+	reporterInterval time.Duration
 }
 
 func newReceivedProcessor(
@@ -41,6 +45,7 @@ func newReceivedProcessor(
 	packagesStateProvider types.PackagesStateProvider,
 	capabilities protobufs.AgentCapabilities,
 	packageSyncMutex *sync.Mutex,
+	reporterInt time.Duration,
 ) receivedProcessor {
 	return receivedProcessor{
 		logger:                logger,
@@ -50,6 +55,7 @@ func newReceivedProcessor(
 		packagesStateProvider: packagesStateProvider,
 		capabilities:          capabilities,
 		packageSyncMutex:      packageSyncMutex,
+		reporterInterval:      reporterInt,
 	}
 }
 
@@ -129,6 +135,7 @@ func (r *receivedProcessor) ProcessReceivedMessage(ctx context.Context, msg *pro
 				r.clientSyncedState,
 				r.packagesStateProvider,
 				r.packageSyncMutex,
+				r.reporterInterval,
 			)
 		} else {
 			r.logger.Debugf(ctx, "Ignoring PackagesAvailable, agent does not have AcceptsPackages capability")

client/internal/wsreceiver.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/gorilla/websocket"
 	"github.com/open-telemetry/opamp-go/client/types"
@@ -34,13 +35,14 @@ func NewWSReceiver(
 	packagesStateProvider types.PackagesStateProvider,
 	capabilities protobufs.AgentCapabilities,
 	packageSyncMutex *sync.Mutex,
+	reporterInterval time.Duration,
 ) *wsReceiver {
 	w := &wsReceiver{
 		conn:      conn,
 		logger:    logger,
 		sender:    sender,
 		callbacks: callbacks,
-		processor: newReceivedProcessor(logger, callbacks, sender, clientSyncedState, packagesStateProvider, capabilities, packageSyncMutex),
+		processor: newReceivedProcessor(logger, callbacks, sender, clientSyncedState, packagesStateProvider, capabilities, packageSyncMutex, reporterInterval),
 		stopped:   make(chan struct{}),
 	}
 

client/types/startsettings.go

@@ -64,4 +64,9 @@ type StartSettings struct {
 	//
 	// If the ReportsHeartbeat capability is disabled, this option has no effect.
 	HeartbeatInterval *time.Duration
+
+	// Optional DownloadReporterInterval to configure how often a client reports the status of a package that is being downloaded.
+	// If nil, the default reporter interval (10s) will be used.
+	// If specified a minimum value of 1s will be enforced.
+	DownloadReporterInterval *time.Duration
 }

client/wsclient.go

@@ -359,6 +359,7 @@ func (c *wsClient) runOneCycle(ctx context.Context) {
 		c.common.PackagesStateProvider,
 		c.common.Capabilities,
 		&c.common.PackageSyncMutex,
+		c.common.downloadReporterInterval,
 	)
 
 	// When the wsclient is closed, the context passed to runOneCycle will be canceled.