[cmd/opampsupervisor]: Implement PackagesAvailable for upgrading agent

.chloggen/feat_supervisor-update-collector.yaml

@@ -0,0 +1,13 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: opampsupervisor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Adds support for agent upgrades"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [34734, 33947]

cmd/opampsupervisor/e2e_test.go

@@ -1251,7 +1251,7 @@ func TestSupervisorWritesAgentFilesToStorageDir(t *testing.T) {
 		"storage_dir": storageDir,
 	})
 
-	require.Nil(t, s.Start())
+	require.NoError(t, s.Start())
 
 	waitForSupervisorConnection(server.supervisorConnected, true)
 
@@ -1647,6 +1647,148 @@ func TestSupervisorOpAmpServerPort(t *testing.T) {
 	}, 10*time.Second, 500*time.Millisecond, "Log never appeared in output")
 }
 
+func TestSupervisorUpgradesAgent(t *testing.T) {
+	tmpDir := t.TempDir()
+	storageDir := filepath.Join(tmpDir, "storage")
+
+	ext := ""
+	if runtime.GOOS == "windows" {
+		ext = ".exe"
+	}
+
+	agentFileName := fmt.Sprintf("otelcontribcol_%s_%s%s", runtime.GOOS, runtime.GOARCH, ext)
+
+	agentFilePath := filepath.Join("..", "..", "bin", agentFileName)
+	agentFileCopyPath := filepath.Join(tmpDir, agentFileName)
+
+	// Upgrading will overwrite the agent binary, so we'll copy to a new path to not affect other tests
+	copyFile(t, agentFilePath, agentFileCopyPath)
+
+	agentIDChan := make(chan []byte, 1)
+	agentDescriptionChan := make(chan *protobufs.AgentDescription, 1)
+	packageStatusesChan := make(chan *protobufs.PackageStatuses, 2)
+
+	server := newOpAMPServer(
+		t,
+		defaultConnectingHandler,
+		types.ConnectionCallbacks{
+			OnMessage: func(_ context.Context, _ types.Connection, message *protobufs.AgentToServer) *protobufs.ServerToAgent {
+				select {
+				case agentIDChan <- message.InstanceUid:
+				default:
+				}
+
+				if message.AgentDescription != nil {
+					select {
+					case agentDescriptionChan <- message.AgentDescription:
+					default:
+					}
+				}
+
+				if message.PackageStatuses != nil {
+					select {
+					case packageStatusesChan <- message.PackageStatuses:
+					default:
+					}
+				}
+
+				return &protobufs.ServerToAgent{}
+			},
+		},
+	)
+
+	s := newSupervisor(t, "upgrade", map[string]string{
+		"url":         server.addr,
+		"storage_dir": storageDir,
+		"agent_path":  agentFileCopyPath,
+	})
+
+	require.Nil(t, s.Start())
+	defer s.Shutdown()
+
+	waitForSupervisorConnection(server.supervisorConnected, true)
+
+	t.Logf("Supervisor connected")
+
+	agentVersion := "0.110.0"
+	agentName := fmt.Sprintf("otelcol-contrib_0.110.0_%s_%s.tar.gz", runtime.GOOS, runtime.GOARCH)
+	agentURL := fmt.Sprintf("https://github.com/open-telemetry/opentelemetry-collector-releases/releases/download/v0.110.0/%s", agentName)
+	agentSigURL := fmt.Sprintf("https://github.com/open-telemetry/opentelemetry-collector-releases/releases/download/v0.110.0/%s.sig", agentName)
+	agentCertURL := fmt.Sprintf("https://github.com/open-telemetry/opentelemetry-collector-releases/releases/download/v0.110.0/%s.pem", agentName)
+
+	agentHash := getHTTPBodyHash(t, agentURL)
+	cert := getHTTPBodyContents(t, agentCertURL)
+	sig := getHTTPBodyContents(t, agentSigURL)
+
+	signatureField := bytes.Join([][]byte{cert, sig}, []byte(" "))
+
+	<-packageStatusesChan
+	<-agentDescriptionChan
+	agentID := <-agentIDChan
+	server.sendToSupervisor(&protobufs.ServerToAgent{
+		InstanceUid: agentID,
+		PackagesAvailable: &protobufs.PackagesAvailable{
+			Packages: map[string]*protobufs.PackageAvailable{
+				"": {
+					Type:    protobufs.PackageType_PackageType_TopLevel,
+					Version: agentVersion,
+					Hash:    []byte{0x01, 0x02},
+					File: &protobufs.DownloadableFile{
+						DownloadUrl: agentURL,
+						ContentHash: agentHash,
+						Signature:   signatureField,
+					},
+				},
+			},
+			AllPackagesHash: []byte{0x03, 0x04},
+		},
+	})
+
+	// Wait for new package statuses
+	ps := <-packageStatusesChan
+	require.Equal(t, &protobufs.PackageStatuses{
+		Packages: map[string]*protobufs.PackageStatus{
+			"": {
+				// TODO: Should initital version be filled in?
+				// What about the hash?
+				Name:                 "",
+				AgentHasVersion:      "",
+				AgentHasHash:         nil,
+				ServerOfferedVersion: agentVersion,
+				ServerOfferedHash:    []byte{0x01, 0x02},
+				Status:               protobufs.PackageStatusEnum_PackageStatusEnum_Installing,
+			},
+		},
+		ServerProvidedAllPackagesHash: []byte{0x03, 0x04},
+	}, ps)
+
+	ps = <-packageStatusesChan
+	require.Equal(t, &protobufs.PackageStatuses{
+		Packages: map[string]*protobufs.PackageStatus{
+			"": {
+				Name:                 "",
+				AgentHasVersion:      agentVersion,
+				AgentHasHash:         []byte{0x01, 0x02},
+				ServerOfferedVersion: agentVersion,
+				ServerOfferedHash:    []byte{0x01, 0x02},
+				Status:               protobufs.PackageStatusEnum_PackageStatusEnum_Installed,
+			},
+		},
+		ServerProvidedAllPackagesHash: []byte{0x03, 0x04},
+	}, ps)
+
+	agentDesc := <-agentDescriptionChan
+	versionFound := false
+	for _, v := range agentDesc.IdentifyingAttributes {
+		if v.Key == semconv.AttributeServiceVersion {
+			versionFound = true
+			require.Equal(t, agentVersion, v.Value.GetStringValue())
+			break
+		}
+	}
+	require.True(t, versionFound, "Agent description after upgrade did not contain the agent version.")
+}
+
 func findRandomPort() (int, error) {
 	l, err := net.Listen("tcp", "localhost:0")
 	if err != nil {
@@ -1662,3 +1804,41 @@ func findRandomPort() (int, error) {
 
 	return port, nil
 }
+
+func getHTTPBodyContents(t *testing.T, url string) []byte {
+	r, err := http.Get(url)
+	require.NoError(t, err)
+	defer r.Body.Close()
+
+	by, err := io.ReadAll(r.Body)
+	require.NoError(t, err)
+
+	return by
+}
+
+func getHTTPBodyHash(t *testing.T, url string) []byte {
+	resp, err := http.Get(url)
+	require.NoError(t, err)
+	defer resp.Body.Close()
+
+	hasher := sha256.New()
+	_, err = io.Copy(hasher, resp.Body)
+	require.NoError(t, err)
+	return hasher.Sum(nil)
+}
+
+func copyFile(t *testing.T, from, to string) {
+	fromFile, err := os.Open(from)
+	require.NoError(t, err)
+	defer fromFile.Close()
+
+	fi, err := fromFile.Stat()
+	require.NoError(t, err)
+
+	toFile, err := os.OpenFile(to, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, fi.Mode())
+	require.NoError(t, err)
+	defer toFile.Close()
+
+	_, err = io.Copy(toFile, fromFile)
+	require.NoError(t, err)
+}