Merge branch 'main' of github.com:control-theory/opamp-go

.github/workflows/build-and-test.yml

@@ -5,6 +5,7 @@ on:
     tags:
       - "v[0-9]+.[0-9]+.[0-9]+*"
   pull_request:
+    types: [opened, synchronize, reopened, labeled, unlabeled]
 
 jobs:
   setup-environment:
@@ -13,9 +14,9 @@ jobs:
       - name: Checkout Repo
         uses: actions/checkout@v4
       - name: Setup Go
-        uses: actions/setup-go@v5.0.1
+        uses: actions/setup-go@v5.1.0
         with:
-          go-version: "1.20"
+          go-version: "1.22"
       - name: Setup Go Environment
         run: |
           echo "GOPATH=$(go env GOPATH)" >> $GITHUB_ENV
@@ -46,9 +47,9 @@ jobs:
       - name: Checkout Repo
         uses: actions/checkout@v4
       - name: Setup Go
-        uses: actions/setup-go@v5.0.1
+        uses: actions/setup-go@v5.1.0
         with:
-          go-version: "1.20"
+          go-version: "1.22"
       - name: Setup Go Environment
         run: |
           echo "GOPATH=$(go env GOPATH)" >> $GITHUB_ENV
@@ -76,7 +77,7 @@ jobs:
       - name: Setup Go
         uses: actions/setup-go@v5
         with:
-          go-version: "1.20"
+          go-version: "1.22"
       - name: Setup Go Environment
         run: |
           echo "GOPATH=$(go env GOPATH)" >> $GITHUB_ENV
@@ -101,7 +102,7 @@ jobs:
       - name: Run Unit Tests With Coverage
         run: make test-with-cover
       - name: Upload coverage report
-        uses: codecov/codecov-action@v4
+        uses: codecov/codecov-action@v5
         with:
           token: ${{ secrets.CODECOV_TOKEN }}
           file: ./coverage.out

.github/workflows/static-check.yml

@@ -2,6 +2,7 @@ name: static-check
 
 on:
   pull_request:
+    types: [opened, synchronize, reopened, labeled, unlabeled]
   push:
     tags:
       - 'v*'
@@ -19,7 +20,7 @@ jobs:
       - id: install_go
         uses: WillAbides/setup-go-faster@v1.14.0
         with:
-          go-version: "1.20.x"
+          go-version: "1.22.x"
       - uses: dominikh/staticcheck-action@v1.3.1
         with:
           install-go: false

.github/workflows/tidy.yml

@@ -0,0 +1,47 @@
+name: go-mod-tidy
+on:
+  pull_request_target:
+    types: [opened, ready_for_review, synchronize, reopened, labeled, unlabeled]
+    branches:
+      - main
+jobs:
+  setup-environment:
+    runs-on: ubuntu-latest
+    if: ${{ (github.actor == 'renovate[bot]' || contains(github.event.pull_request.labels.*.name, 'renovatebot')) }}
+    steps:
+      - name: Checkout Repo
+        uses: actions/checkout@v4
+        with:
+          ref: ${{ github.head_ref }}
+      - name: Setup Go
+        uses: actions/setup-go@v5.1.0
+        with:
+          go-version: "1.22"
+      - name: Setup Go Environment
+        run: |
+          echo "GOPATH=$(go env GOPATH)" >> $GITHUB_ENV
+          echo "$(go env GOPATH)/bin" >> $GITHUB_PATH
+      - name: Cache Go
+        id: module-cache
+        uses: actions/cache@v4
+        with:
+          path: /home/runner/go/pkg/mod
+          key: go-pkg-mod-${{ runner.os }}-${{ hashFiles('**/go.mod', '**/go.sum') }}
+      - name: Install dependencies
+        if: steps.module-cache.outputs.cache-hit != 'true'
+        run: make gomoddownload
+      - name: Install Tools
+        if: steps.module-cache.outputs.cache-hit != 'true'
+        run: make install-tools
+      - name: go mod tidy, make all
+        run: |
+          make tidy && make all
+          git config user.name opentelemetrybot
+          git config user.email 107717825+opentelemetrybot@users.noreply.github.com
+          echo "git diff --exit-code || (git add . && git commit -m \"go mod tidy, make all\" && git push)"
+          git diff --exit-code || (git add . && git commit -m "go mod tidy, make all" && git push)
+        env:
+          GITHUB_TOKEN: ${{ secrets.OPENTELEMETRYBOT_GITHUB_TOKEN }}
+      - uses: actions-ecosystem/action-remove-labels@v1
+        with:
+          labels: renovatebot

client/clientimpl_test.go

@@ -340,6 +340,79 @@ func TestConnectWithHeader(t *testing.T) {
 	})
 }
 
+func TestConnectWithHeaderFunc(t *testing.T) {
+	testClients(t, func(t *testing.T, client OpAMPClient) {
+		// Start a server.
+		srv := internal.StartMockServer(t)
+		var conn atomic.Value
+		srv.OnConnect = func(r *http.Request) {
+			authHdr := r.Header.Get("Authorization")
+			assert.EqualValues(t, "Bearer 12345678", authHdr)
+			userAgentHdr := r.Header.Get("User-Agent")
+			assert.EqualValues(t, "custom-agent/1.0", userAgentHdr)
+			conn.Store(true)
+		}
+
+		hf := func(header http.Header) http.Header {
+			header.Set("Authorization", "Bearer 12345678")
+			header.Set("User-Agent", "custom-agent/1.0")
+			return header
+		}
+
+		// Start a client.
+		settings := types.StartSettings{
+			OpAMPServerURL: "ws://" + srv.Endpoint,
+			HeaderFunc:     hf,
+		}
+		startClient(t, settings, client)
+
+		// Wait for connection to be established.
+		eventually(t, func() bool { return conn.Load() != nil })
+
+		// Shutdown the Server and the client.
+		srv.Close()
+		_ = client.Stop(context.Background())
+	})
+}
+
+func TestConnectWithHeaderAndHeaderFunc(t *testing.T) {
+	testClients(t, func(t *testing.T, client OpAMPClient) {
+		// Start a server.
+		srv := internal.StartMockServer(t)
+		var conn atomic.Value
+		srv.OnConnect = func(r *http.Request) {
+			authHdr := r.Header.Get("Authorization")
+			assert.EqualValues(t, "Bearer 12345678", authHdr)
+			userAgentHdr := r.Header.Get("User-Agent")
+			assert.EqualValues(t, "custom-agent/1.0", userAgentHdr)
+			conn.Store(true)
+		}
+
+		baseHeader := http.Header{}
+		baseHeader.Set("User-Agent", "custom-agent/1.0")
+
+		hf := func(header http.Header) http.Header {
+			header.Set("Authorization", "Bearer 12345678")
+			return header
+		}
+
+		// Start a client.
+		settings := types.StartSettings{
+			OpAMPServerURL: "ws://" + srv.Endpoint,
+			Header:         baseHeader,
+			HeaderFunc:     hf,
+		}
+		startClient(t, settings, client)
+
+		// Wait for connection to be established.
+		eventually(t, func() bool { return conn.Load() != nil })
+
+		// Shutdown the Server and the client.
+		srv.Close()
+		_ = client.Stop(context.Background())
+	})
+}
+
 func TestConnectWithTLS(t *testing.T) {
 	testClients(t, func(t *testing.T, client OpAMPClient) {
 		// Start a server.
@@ -1199,6 +1272,7 @@ type packageTestCase struct {
 	available           *protobufs.PackagesAvailable
 	expectedStatus      *protobufs.PackageStatuses
 	expectedFileContent map[string][]byte
+	expectedSignature   map[string][]byte
 	expectedError       string
 }
 
@@ -1320,6 +1394,10 @@ func verifyUpdatePackages(t *testing.T, testCase packageTestCase) {
 			for pkgName, receivedContent := range localPackageState.GetContent() {
 				expectedContent := testCase.expectedFileContent[pkgName]
 				assert.EqualValues(t, expectedContent, receivedContent)
+
+				actualSignature := localPackageState.GetSignature()[pkgName]
+				expectedSignature := testCase.expectedSignature[pkgName]
+				assert.EqualValues(t, expectedSignature, actualSignature)
 			}
 		}
 
@@ -1394,6 +1472,7 @@ func createPackageTestCase(name string, downloadSrv *httptest.Server) packageTes
 					File: &protobufs.DownloadableFile{
 						DownloadUrl: downloadSrv.URL + packageFileURL,
 						ContentHash: []byte{4, 5},
+						Signature:   []byte{6, 7},
 					},
 					Hash: []byte{1, 2, 3},
 				},
@@ -1419,6 +1498,10 @@ func createPackageTestCase(name string, downloadSrv *httptest.Server) packageTes
 		expectedFileContent: map[string][]byte{
 			"package1": packageFileContent,
 		},
+
+		expectedSignature: map[string][]byte{
+			"package1": {6, 7},
+		},
 	}
 }
 
@@ -1905,8 +1988,6 @@ func TestSendCustomMessagePendingError(t *testing.T) {
 		}
 		client.SetCustomCapabilities(clientCustomCapabilities)
 
-		assert.NoError(t, client.Start(context.Background(), settings))
-
 		customMessage1 := &protobufs.CustomMessage{
 			Capability: "local.test.example",
 			Type:       "hello",
@@ -1918,6 +1999,7 @@ func TestSendCustomMessagePendingError(t *testing.T) {
 			Data:       []byte("test message 2"),
 		}
 
+		// Send a message to the unstarted client.
 		_, err := client.SendCustomMessage(customMessage1)
 		assert.NoError(t, err)
 
@@ -1926,6 +2008,9 @@ func TestSendCustomMessagePendingError(t *testing.T) {
 		assert.ErrorIs(t, err, types.ErrCustomMessagePending)
 		assert.NotNil(t, sendingChan)
 
+		// Start the client so we can start processing messages properly.
+		assert.NoError(t, client.Start(context.Background(), settings))
+
 		// Receive the first custom message
 		eventually(
 			t,

client/httpclient.go

@@ -44,7 +44,7 @@ func (c *httpClient) Start(ctx context.Context, settings types.StartSettings) er
 	c.opAMPServerURL = settings.OpAMPServerURL
 
 	// Prepare Server connection settings.
-	c.sender.SetRequestHeader(settings.Header)
+	c.sender.SetRequestHeader(settings.Header, settings.HeaderFunc)
 
 	// Add TLS configuration into httpClient
 	c.sender.AddTLSConfig(settings.TLSConfig)
@@ -131,6 +131,7 @@ func (c *httpClient) runUntilStopped(ctx context.Context) {
 		&c.common.ClientSyncedState,
 		c.common.PackagesStateProvider,
 		c.common.Capabilities,
+		&c.common.PackageSyncMutex,
 	)
 }
 

client/httpclient_test.go

@@ -145,3 +145,81 @@ func TestHTTPClientSetPollingInterval(t *testing.T) {
 	err := client.Stop(context.Background())
 	assert.NoError(t, err)
 }
+
+func TestHTTPClientStartWithHeartbeatInterval(t *testing.T) {
+	tests := []struct {
+		name             string
+		enableHeartbeat  bool
+		expectHeartbeats bool
+	}{
+		{"client enable heartbeat", true, true},
+		{"client disable heartbeat", false, false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Start a Server.
+			srv := internal.StartMockServer(t)
+			var rcvCounter int64
+			srv.OnMessage = func(msg *protobufs.AgentToServer) *protobufs.ServerToAgent {
+				if msg == nil {
+					t.Error("unexpected nil msg")
+					return nil
+				}
+				assert.EqualValues(t, rcvCounter, msg.SequenceNum)
+				atomic.AddInt64(&rcvCounter, 1)
+				return nil
+			}
+
+			// Start a client.
+			heartbeat := 10 * time.Millisecond
+			settings := types.StartSettings{
+				OpAMPServerURL:    "http://" + srv.Endpoint,
+				HeartbeatInterval: &heartbeat,
+			}
+			if tt.enableHeartbeat {
+				settings.Capabilities = protobufs.AgentCapabilities_AgentCapabilities_ReportsHeartbeat
+			}
+			client := NewHTTP(nil)
+			prepareClient(t, &settings, client)
+
+			assert.NoError(t, client.Start(context.Background(), settings))
+
+			// Verify that status report is delivered.
+			eventually(t, func() bool { return atomic.LoadInt64(&rcvCounter) == 1 })
+
+			if tt.expectHeartbeats {
+				assert.Eventually(t, func() bool { return atomic.LoadInt64(&rcvCounter) >= 2 }, 5*time.Second, 10*time.Millisecond)
+			} else {
+				assert.Never(t, func() bool { return atomic.LoadInt64(&rcvCounter) >= 2 }, 50*time.Millisecond, 10*time.Millisecond)
+			}
+
+			// Shutdown the Server.
+			srv.Close()
+
+			// Shutdown the client.
+			err := client.Stop(context.Background())
+			assert.NoError(t, err)
+		})
+	}
+}
+
+func TestHTTPClientStartWithZeroHeartbeatInterval(t *testing.T) {
+	srv := internal.StartMockServer(t)
+
+	// Start a client.
+	heartbeat := 0 * time.Millisecond
+	settings := types.StartSettings{
+		OpAMPServerURL:    "http://" + srv.Endpoint,
+		HeartbeatInterval: &heartbeat,
+		Capabilities:      protobufs.AgentCapabilities_AgentCapabilities_ReportsHeartbeat,
+	}
+	client := NewHTTP(nil)
+	prepareClient(t, &settings, client)
+
+	// Zero heartbeat interval is invalid for http client.
+	assert.Error(t, client.Start(context.Background(), settings))
+
+	// Shutdown the Server.
+	srv.Close()
+}

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
 
@@ -134,6 +137,12 @@ func (c *ClientCommon) PrepareStart(
 		c.Callbacks = types.CallbacksStruct{}
 	}
 
+	if c.Capabilities&protobufs.AgentCapabilities_AgentCapabilities_ReportsHeartbeat != 0 && settings.HeartbeatInterval != nil {
+		if err := c.sender.SetHeartbeatInterval(*settings.HeartbeatInterval); err != nil {
+			return err
+		}
+	}
+
 	if err := c.sender.SetInstanceUid(settings.InstanceUid); err != nil {
 		return err
 	}