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/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,78 @@
+package internal
+
+import (
+	"context"
+	"sync/atomic"
+	"time"
+
+	"github.com/open-telemetry/opamp-go/protobufs"
+)
+
+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 updates the package status details and calls the passed upateFn to send the new status.
+func (p *downloadReporter) report(ctx context.Context, status *protobufs.PackageStatus, updateFn func(context.Context, bool) error) {
+	go func() {
+		// Make sure we wait at least 1s before reporting the download rate in bps to avoid a panic
+		time.Sleep(time.Second)
+		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
+				}
+				status.DownloadDetails = &protobufs.PackageDownloadDetails{
+					// DownloadPercent may be zero if nothing has been downloaded OR there isn't a Content-Length header in the response.
+					DownloadPercent: downloadPercent,
+					// DownloadBytesPerSecond may be zero if no bytes from the response body have been read yet.
+					DownloadBytesPerSecond: bps,
+				}
+				_ = updateFn(ctx, true)
+			}
+		}
+	}()
+}
+
+// stop the downloadReporter report goroutine
+func (p *downloadReporter) stop() {
+	close(p.done)
+}

client/internal/packagessyncer.go

@@ -5,7 +5,9 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
+	"strconv"
 	"sync"
 
 	"github.com/open-telemetry/opamp-go/client/types"
@@ -273,6 +275,10 @@ func (s *packagesSyncer) shouldDownloadFile(ctx context.Context,
 
 // 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,7 +296,20 @@ func (s *packagesSyncer) downloadFile(ctx context.Context, pkgName string, file
 		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(downloadReporterDefaultInterval, packageLength) // TODO set interval
+	detailsReporter.report(ctx, status, s.reportStatuses)
+	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)
 	}