Add HTTP Client to opampextension

.chloggen/opamp-http.yaml

@@ -0,0 +1,27 @@
+# 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: opampextension
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: enables creating and using an http client
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [31389]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

extension/opampextension/config.go

@@ -5,11 +5,14 @@ package opampextension // import "github.com/open-telemetry/opentelemetry-collec
 
 import (
 	"errors"
+	"net/url"
 
 	"github.com/oklog/ulid/v2"
+	"github.com/open-telemetry/opamp-go/client"
 	"github.com/open-telemetry/opamp-go/protobufs"
 	"go.opentelemetry.io/collector/config/configopaque"
 	"go.opentelemetry.io/collector/config/configtls"
+	"go.uber.org/zap"
 )
 
 // Config contains the configuration for the opamp extension. Trying to mirror
@@ -41,22 +44,82 @@ func (caps Capabilities) toAgentCapabilities() protobufs.AgentCapabilities {
 	return agentCapabilities
 }
 
+type commonFields struct {
+	Endpoint   string                         `mapstructure:"endpoint"`
+	TLSSetting configtls.TLSClientSetting     `mapstructure:"tls,omitempty"`
+	Headers    map[string]configopaque.String `mapstructure:"headers,omitempty"`
+}
+
 // OpAMPServer contains the OpAMP transport configuration.
 type OpAMPServer struct {
-	WS *OpAMPWebsocket `mapstructure:"ws"`
+	WS   *commonFields `mapstructure:"ws,omitempty"`
+	HTTP *commonFields `mapstructure:"http,omitempty"`
 }
 
-// OpAMPWebsocket contains the OpAMP websocket transport configuration.
-type OpAMPWebsocket struct {
-	Endpoint   string                         `mapstructure:"endpoint"`
-	TLSSetting configtls.TLSClientSetting     `mapstructure:"tls,omitempty"`
-	Headers    map[string]configopaque.String `mapstructure:"headers,omitempty"`
+func (c *commonFields) Scheme() string {
+	uri, err := url.ParseRequestURI(c.Endpoint)
+	if err != nil {
+		return ""
+	}
+	return uri.Scheme
+}
+
+func (c *commonFields) Validate() error {
+	if c.Endpoint == "" {
+		return errors.New("opamp server endpoint must be provided")
+	}
+	return nil
+}
+
+func (s OpAMPServer) GetClient(logger *zap.Logger) client.OpAMPClient {
+	if s.WS != nil {
+		return client.NewWebSocket(newLoggerFromZap(logger.With(zap.String("client", "ws"))))
+	}
+	return client.NewHTTP(newLoggerFromZap(logger.With(zap.String("client", "http"))))
+}
+
+func (s OpAMPServer) GetHeaders() map[string]configopaque.String {
+	if s.WS != nil {
+		return s.WS.Headers
+	} else if s.HTTP != nil {
+		return s.HTTP.Headers
+	}
+	return map[string]configopaque.String{}
+}
+
+func (s OpAMPServer) GetTLSSetting() configtls.TLSClientSetting {
+	if s.WS != nil {
+		return s.WS.TLSSetting
+	} else if s.HTTP != nil {
+		return s.HTTP.TLSSetting
+	}
+	return configtls.TLSClientSetting{}
+}
+
+func (s OpAMPServer) GetEndpoint() string {
+	if s.WS != nil {
+		return s.WS.Endpoint
+	} else if s.HTTP != nil {
+		return s.HTTP.Endpoint
+	}
+	return ""
 }
 
 // Validate checks if the extension configuration is valid
 func (cfg *Config) Validate() error {
-	if cfg.Server.WS.Endpoint == "" {
-		return errors.New("opamp server websocket endpoint must be provided")
+	switch {
+	case cfg.Server.WS == nil && cfg.Server.HTTP == nil:
+		return errors.New("opamp server must have at least ws or http set")
+	case cfg.Server.WS != nil && cfg.Server.HTTP != nil:
+		return errors.New("opamp server must have only ws or http set")
+	case cfg.Server.WS != nil:
+		if err := cfg.Server.WS.Validate(); err != nil {
+			return err
+		}
+	case cfg.Server.HTTP != nil:
+		if err := cfg.Server.HTTP.Validate(); err != nil {
+			return err
+		}
 	}
 
 	if cfg.InstanceUID != "" {

extension/opampextension/config_test.go

@@ -10,6 +10,8 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configopaque"
+	"go.opentelemetry.io/collector/config/configtls"
 	"go.opentelemetry.io/collector/confmap"
 	"go.opentelemetry.io/collector/confmap/confmaptest"
 )
@@ -30,7 +32,7 @@ func TestUnmarshalConfig(t *testing.T) {
 	assert.Equal(t,
 		&Config{
 			Server: &OpAMPServer{
-				WS: &OpAMPWebsocket{
+				WS: &commonFields{
 					Endpoint: "wss://127.0.0.1:4320/v1/opamp",
 				},
 			},
@@ -41,20 +43,218 @@ func TestUnmarshalConfig(t *testing.T) {
 		}, cfg)
 }
 
-func TestConfigValidate(t *testing.T) {
-	cfg := &Config{
-		Server: &OpAMPServer{
-			WS: &OpAMPWebsocket{},
+func TestUnmarshalHttpConfig(t *testing.T) {
+	cm, err := confmaptest.LoadConf(filepath.Join("testdata", "config_http.yaml"))
+	require.NoError(t, err)
+	factory := NewFactory()
+	cfg := factory.CreateDefaultConfig()
+	assert.NoError(t, component.UnmarshalConfig(cm, cfg))
+	assert.Equal(t,
+		&Config{
+			Server: &OpAMPServer{
+				HTTP: &commonFields{
+					Endpoint: "https://127.0.0.1:4320/v1/opamp",
+				},
+			},
+			InstanceUID: "01BX5ZZKBKACTAV9WEVGEMMVRZ",
+			Capabilities: Capabilities{
+				ReportsEffectiveConfig: true,
+			},
+		}, cfg)
+}
+
+func TestConfig_Getters(t *testing.T) {
+	type fields struct {
+		Server *OpAMPServer
+	}
+	type expected struct {
+		headers  assert.ValueAssertionFunc
+		tls      assert.ValueAssertionFunc
+		endpoint assert.ValueAssertionFunc
+	}
+	tests := []struct {
+		name     string
+		fields   fields
+		expected expected
+	}{
+		{
+			name: "nothing set",
+			fields: fields{
+				Server: &OpAMPServer{},
+			},
+			expected: expected{
+				headers:  assert.Empty,
+				tls:      assert.Empty,
+				endpoint: assert.Empty,
+			},
+		},
+		{
+			name: "WS valid endpoint, headers, tls",
+			fields: fields{
+				Server: &OpAMPServer{
+					WS: &commonFields{
+						Endpoint: "wss://127.0.0.1:4320/v1/opamp",
+						Headers: map[string]configopaque.String{
+							"test": configopaque.String("test"),
+						},
+						TLSSetting: configtls.TLSClientSetting{
+							Insecure: true,
+						},
+					},
+				},
+			},
+			expected: expected{
+				headers:  assert.NotEmpty,
+				tls:      assert.NotEmpty,
+				endpoint: assert.NotEmpty,
+			},
 		},
+		{
+			name: "HTTP valid endpoint and valid instance id",
+			fields: fields{
+				Server: &OpAMPServer{
+					HTTP: &commonFields{
+						Endpoint: "https://127.0.0.1:4320/v1/opamp",
+						Headers: map[string]configopaque.String{
+							"test": configopaque.String("test"),
+						},
+						TLSSetting: configtls.TLSClientSetting{
+							Insecure: true,
+						},
+					},
+				},
+			},
+			expected: expected{
+				headers:  assert.NotEmpty,
+				tls:      assert.NotEmpty,
+				endpoint: assert.NotEmpty,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tt.expected.headers(t, tt.fields.Server.GetHeaders())
+			tt.expected.tls(t, tt.fields.Server.GetTLSSetting())
+			tt.expected.endpoint(t, tt.fields.Server.GetEndpoint())
+		})
+	}
+}
+
+func TestConfig_Validate(t *testing.T) {
+	type fields struct {
+		Server       *OpAMPServer
+		InstanceUID  string
+		Capabilities Capabilities
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		wantErr assert.ErrorAssertionFunc
+	}{
+		{
+			name: "WS must have endpoint",
+			fields: fields{
+				Server: &OpAMPServer{
+					WS: &commonFields{},
+				},
+			},
+			wantErr: func(t assert.TestingT, err error, _ ...any) bool {
+				return assert.Equal(t, "opamp server endpoint must be provided", err.Error())
+			},
+		},
+		{
+			name: "WS valid endpoint and invalid instance id",
+			fields: fields{
+				Server: &OpAMPServer{
+					WS: &commonFields{
+						Endpoint: "wss://127.0.0.1:4320/v1/opamp",
+					},
+				},
+				InstanceUID: "01BX5ZZKBKACTAV9WEVGEMMVRZFAIL",
+			},
+			wantErr: func(t assert.TestingT, err error, _ ...any) bool {
+				return assert.Equal(t, "opamp instance_uid is invalid", err.Error())
+			},
+		},
+		{
+			name: "WS valid endpoint and valid instance id",
+			fields: fields{
+				Server: &OpAMPServer{
+					WS: &commonFields{
+						Endpoint: "wss://127.0.0.1:4320/v1/opamp",
+					},
+				},
+				InstanceUID: "01BX5ZZKBKACTAV9WEVGEMMVRZ",
+			},
+			wantErr: assert.NoError,
+		},
+		{
+			name: "HTTP must have endpoint",
+			fields: fields{
+				Server: &OpAMPServer{
+					HTTP: &commonFields{},
+				},
+			},
+			wantErr: func(t assert.TestingT, err error, _ ...any) bool {
+				return assert.Equal(t, "opamp server endpoint must be provided", err.Error())
+			},
+		},
+		{
+			name: "HTTP valid endpoint and invalid instance id",
+			fields: fields{
+				Server: &OpAMPServer{
+					HTTP: &commonFields{
+						Endpoint: "https://127.0.0.1:4320/v1/opamp",
+					},
+				},
+				InstanceUID: "01BX5ZZKBKACTAV9WEVGEMMVRZFAIL",
+			},
+			wantErr: func(t assert.TestingT, err error, _ ...any) bool {
+				return assert.Equal(t, "opamp instance_uid is invalid", err.Error())
+			},
+		},
+		{
+			name: "HTTP valid endpoint and valid instance id",
+			fields: fields{
+				Server: &OpAMPServer{
+					HTTP: &commonFields{
+						Endpoint: "https://127.0.0.1:4320/v1/opamp",
+					},
+				},
+				InstanceUID: "01BX5ZZKBKACTAV9WEVGEMMVRZ",
+			},
+			wantErr: assert.NoError,
+		},
+		{
+			name: "neither config set",
+			fields: fields{
+				Server: &OpAMPServer{},
+			},
+			wantErr: func(t assert.TestingT, err error, _ ...any) bool {
+				return assert.Equal(t, "opamp server must have at least ws or http set", err.Error())
+			},
+		},
+		{
+			name: "both config set",
+			fields: fields{
+				Server: &OpAMPServer{
+					WS:   &commonFields{},
+					HTTP: &commonFields{},
+				},
+			},
+			wantErr: func(t assert.TestingT, err error, _ ...any) bool {
+				return assert.Equal(t, "opamp server must have only ws or http set", err.Error())
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg := &Config{
+				Server:       tt.fields.Server,
+				InstanceUID:  tt.fields.InstanceUID,
+				Capabilities: tt.fields.Capabilities,
+			}
+			tt.wantErr(t, cfg.Validate())
+		})
 	}
-	err := cfg.Validate()
-	assert.Equal(t, "opamp server websocket endpoint must be provided", err.Error())
-	cfg.Server.WS.Endpoint = "wss://127.0.0.1:4320/v1/opamp"
-	assert.NoError(t, cfg.Validate())
-	cfg.InstanceUID = "01BX5ZZKBKACTAV9WEVGEMMVRZFAIL"
-	err = cfg.Validate()
-	require.Error(t, err)
-	assert.Equal(t, "opamp instance_uid is invalid", err.Error())
-	cfg.InstanceUID = "01BX5ZZKBKACTAV9WEVGEMMVRZ"
-	require.NoError(t, cfg.Validate())
 }

extension/opampextension/factory.go

@@ -23,9 +23,7 @@ func NewFactory() extension.Factory {
 
 func createDefaultConfig() component.Config {
 	return &Config{
-		Server: &OpAMPServer{
-			WS: &OpAMPWebsocket{},
-		},
+		Server: &OpAMPServer{},
 		Capabilities: Capabilities{
 			ReportsEffectiveConfig: true,
 		},