Use sumologic extension base URL to determine the appropriate OpAMP endpoint

.changelog/1399.changed.txt

@@ -0,0 +1 @@
+feat: Use sumologic extension base URL to determine the appropriate OpAMP endpoint

pkg/extension/opampextension/go.mod

@@ -3,7 +3,7 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/extension/opamp
 go 1.20
 
 require (
-	github.com/SumoLogic/sumologic-otel-collector/pkg/extension/sumologicextension v0.77.0-sumo-0
+	github.com/SumoLogic/sumologic-otel-collector/pkg/extension/sumologicextension v0.91.0-sumo-0-rc.0
 	github.com/google/uuid v1.3.1
 	github.com/knadh/koanf/parsers/yaml v0.1.0
 	github.com/knadh/koanf/providers/rawbytes v0.1.0

pkg/extension/opampextension/opamp_agent.go

@@ -18,10 +18,12 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"net/url"
 	"os"
 	"path/filepath"
 	"reflect"
 	"runtime"
+	"strings"
 	"syscall"
 
 	"github.com/google/uuid"
@@ -49,6 +51,8 @@ type opampAgent struct {
 	authExtension *sumologicextension.SumologicExtension
 	authHeader    http.Header
 
+	endpoint string
+
 	agentType    string
 	agentVersion string
 
@@ -83,6 +87,15 @@ func (o *opampAgent) Start(ctx context.Context, host component.Host) error {
 		return err
 	}
 
+	var baseURL string
+	if o.authExtension != nil {
+		baseURL = o.authExtension.BaseUrl()
+	}
+
+	if err := o.setEndpoint(baseURL); err != nil {
+		return err
+	}
+
 	if o.authExtension == nil {
 		return o.startClient(ctx)
 	}
@@ -130,7 +143,7 @@ func (o *opampAgent) Reload(ctx context.Context) error {
 func (o *opampAgent) startClient(ctx context.Context) error {
 	settings := types.StartSettings{
 		Header:         o.authHeader,
-		OpAMPServerURL: o.cfg.Endpoint,
+		OpAMPServerURL: o.endpoint,
 		InstanceUid:    o.instanceId.String(),
 		Callbacks: types.CallbacksStruct{
 			OnConnectFunc: func() {
@@ -223,6 +236,33 @@ func (o *opampAgent) watchCredentials(ctx context.Context, callback func(ctx con
 	return nil
 }
 
+// setEndpoint sets the OpAMP endpoint based on the collector endpoint.
+// This is a hack, and it should be removed when the backend is able to
+// correctly redirect our OpAMP client to the correct URL.
+func (o *opampAgent) setEndpoint(baseURL string) error {
+	if baseURL == "" {
+		o.endpoint = o.cfg.Endpoint
+		return nil
+	}
+	u, err := url.Parse(baseURL)
+	if err != nil {
+		return fmt.Errorf("url error, cannot set opamp endpoint: %s", err)
+	}
+
+	u.Scheme = "wss"
+	u.Path = "/v1/opamp"
+
+	// These replacements are specific to Sumo Logic's current domain naming,
+	// and are made provisionally for the OTRM beta. In the future, the backend
+	// will inform the agent of the correct OpAMP URL to use.
+	u.Host = strings.Replace(u.Host, "open-events", "opamp-events", 1)
+	u.Host = strings.Replace(u.Host, "open-collectors", "opamp-collectors", 1)
+
+	o.endpoint = u.String()
+
+	return nil
+}
+
 func newOpampAgent(cfg *Config, logger *zap.Logger, build component.BuildInfo, res pcommon.Resource) (*opampAgent, error) {
 	agentType := build.Command
 

pkg/extension/opampextension/opamp_agent_test.go

@@ -26,6 +26,7 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/config/confighttp"
 	"go.opentelemetry.io/collector/extension/extensiontest"
 	semconv "go.opentelemetry.io/collector/semconv/v1.18.0"
 )
@@ -213,3 +214,68 @@ func TestReload(t *testing.T) {
 	assert.NoError(t, o.Start(ctx, componenttest.NewNopHost()))
 	assert.NoError(t, o.Reload(ctx))
 }
+
+// To be removed when the OpAMP backend correctly advertises its URL.
+// This test is badly written, it reaches into unexported fields to test
+// their contents, but given that it will be removed in a few months, that
+// shouldn't matter too much from a big picture perspective.
+func TestHackSetEndpoint(t *testing.T) {
+	tests := []struct {
+		name         string
+		url          string
+		wantEndpoint string
+	}{
+		{
+			name:         "empty url defaults to config endpoint",
+			wantEndpoint: "wss://example.com",
+		},
+		{
+			name:         "url variant a",
+			url:          "https://sumo-open-events.example.com",
+			wantEndpoint: "wss://sumo-opamp-events.example.com/v1/opamp",
+		},
+		{
+			name:         "url variant b",
+			url:          "https://sumo-open-collectors.example.com",
+			wantEndpoint: "wss://sumo-opamp-collectors.example.com/v1/opamp",
+		},
+		{
+			name:         "url variant c",
+			url:          "https://example.com",
+			wantEndpoint: "wss://example.com/v1/opamp",
+		},
+		{
+			name:         "dev sumologic url",
+			url:          "https://long-open-events.sumologic.net/api/v1",
+			wantEndpoint: "wss://long-opamp-events.sumologic.net/v1/opamp",
+		},
+		{
+			name:         "prod sumologic url",
+			url:          "https://open-collectors.sumologic.com/api/v1",
+			wantEndpoint: "wss://opamp-collectors.sumologic.com/v1/opamp",
+		},
+		{
+			name:         "prod sumologic url with region",
+			url:          "https://open-collectors.au.sumologic.com/api/v1/",
+			wantEndpoint: "wss://opamp-collectors.au.sumologic.com/v1/opamp",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			agent := &opampAgent{cfg: &Config{
+				HTTPClientSettings: confighttp.HTTPClientSettings{
+					Endpoint: "wss://example.com",
+				},
+			}}
+			if err := agent.setEndpoint(test.url); err != nil {
+				// can only happen with an invalid URL, which is quite hard
+				// to even come up with for Go's URL package
+				t.Fatal(err)
+			}
+			if got, want := agent.endpoint, test.wantEndpoint; got != want {
+				t.Errorf("didn't get expected endpoint: got %q, want %q", got, want)
+			}
+		})
+	}
+}