feat(sinks): enable sinks with noauth.

maestro/config/authentication/authentication_builder.go

@@ -0,0 +1,29 @@
+package authentication
+
+import (
+	maestrobasicauth "github.com/orb-community/orb/maestro/config/authentication/basicauth"
+	maestronoauth "github.com/orb-community/orb/maestro/config/authentication/noauth"
+	"github.com/orb-community/orb/maestro/config/output"
+	"github.com/orb-community/orb/maestro/password"
+	"github.com/orb-community/orb/pkg/types"
+	"github.com/orb-community/orb/sinks/authentication_type/basicauth"
+	"github.com/orb-community/orb/sinks/authentication_type/noauth"
+)
+
+type AuthBuilderService interface {
+	GetExtensionsFromMetadata(config types.Metadata) (output.Extensions, string)
+	DecodeAuth(config types.Metadata) (types.Metadata, error)
+	EncodeAuth(config types.Metadata) (types.Metadata, error)
+}
+
+func GetAuthService(authType string, service password.EncryptionService) AuthBuilderService {
+	switch authType {
+	case basicauth.AuthType:
+		return &maestrobasicauth.BasicAuthBuilder{
+			EncryptionService: service,
+		}
+	case noauth.AuthType:
+		return &maestronoauth.NoAuthBuilder{}
+	}
+	return nil
+}

maestro/config/authentication/basicauth/authentication.go

@@ -0,0 +1,51 @@
+package basicauth
+
+import (
+	"github.com/orb-community/orb/maestro/config/output"
+	"github.com/orb-community/orb/maestro/password"
+	"github.com/orb-community/orb/pkg/types"
+)
+
+const AuthenticationKey = "authentication"
+
+type BasicAuthBuilder struct {
+	EncryptionService password.EncryptionService
+}
+
+func (b *BasicAuthBuilder) GetExtensionsFromMetadata(c types.Metadata) (output.Extensions, string) {
+	authcfg := c.GetSubMetadata(AuthenticationKey)
+	username := authcfg["username"].(string)
+	password := authcfg["password"].(string)
+	return output.Extensions{
+		BasicAuth: &output.BasicAuthenticationExtension{
+			ClientAuth: &output.ClientAuth{
+				Username: username,
+				Password: password,
+			},
+		},
+	}, "basicauth/exporter"
+}
+
+func (b *BasicAuthBuilder) DecodeAuth(c types.Metadata) (types.Metadata, error) {
+	authCfg := c.GetSubMetadata(AuthenticationKey)
+	password := authCfg["password"].(string)
+	decodedPassword, err := b.EncryptionService.DecodePassword(password)
+	if err != nil {
+		return nil, err
+	}
+	authCfg["password"] = decodedPassword
+	c[AuthenticationKey] = authCfg
+	return c, nil
+}
+
+func (b *BasicAuthBuilder) EncodeAuth(c types.Metadata) (types.Metadata, error) {
+	authcfg := c.GetSubMetadata(AuthenticationKey)
+	password := authcfg["password"].(string)
+	encodedPassword, err := b.EncryptionService.EncodePassword(password)
+	if err != nil {
+		return nil, err
+	}
+	authcfg["password"] = encodedPassword
+	c[AuthenticationKey] = authcfg
+	return c, nil
+}

maestro/config/authentication/noauth/authentication.go

@@ -0,0 +1,21 @@
+package noauth
+
+import (
+	"github.com/orb-community/orb/maestro/config/output"
+	"github.com/orb-community/orb/pkg/types"
+)
+
+type NoAuthBuilder struct {
+}
+
+func (b *NoAuthBuilder) GetExtensionsFromMetadata(c types.Metadata) (output.Extensions, string) {
+	return output.Extensions{}, ""
+}
+
+func (b *NoAuthBuilder) DecodeAuth(c types.Metadata) (types.Metadata, error) {
+	return c, nil
+}
+
+func (b *NoAuthBuilder) EncodeAuth(c types.Metadata) (types.Metadata, error) {
+	return c, nil
+}

maestro/config/config_builder.go

@@ -3,6 +3,8 @@ package config
 import (
 	"context"
 	"fmt"
+	"github.com/orb-community/orb/maestro/config/authentication"
+	"github.com/orb-community/orb/maestro/config/output"
 	"strings"
 
 	"github.com/orb-community/orb/pkg/errors"
@@ -367,17 +369,17 @@ func (c *configBuilder) BuildDeploymentConfig(deployment *DeploymentRequest) (st
 
 // ReturnConfigYamlFromSink this is the main method, which will generate the YAML file from the
 func (c *configBuilder) ReturnConfigYamlFromSink(_ context.Context, kafkaUrlConfig string, deployment *DeploymentRequest) (string, error) {
-	authType := deployment.Config.GetSubMetadata(AuthenticationKey)["type"]
+	authType := deployment.Config.GetSubMetadata("authentication")["type"]
 	authTypeStr, ok := authType.(string)
 	if !ok {
 		return "", errors.New("failed to create config invalid authentication type")
 	}
 	// TODO move this into somewhere else
-	authBuilder := GetAuthService(authTypeStr, c.encryptionService)
+	authBuilder := authentication.GetAuthService(authTypeStr, c.encryptionService)
 	if authBuilder == nil {
 		return "", errors.New("invalid authentication type")
 	}
-	exporterBuilder := FromStrategy(deployment.Backend)
+	exporterBuilder := output.FromStrategy(deployment.Backend)
 	if exporterBuilder == nil {
 		return "", errors.New("invalid backend")
 	}
@@ -388,11 +390,15 @@ func (c *configBuilder) ReturnConfigYamlFromSink(_ context.Context, kafkaUrlConf
 	}
 
 	// Add prometheus extension for metrics
-	extensions.PProf = &PProfExtension{
+	extensions.PProf = &output.PProfExtension{
 		Endpoint: "0.0.0.0:1888",
 	}
-	serviceConfig := ServiceConfig{
-		Extensions: []string{"pprof", extensionName},
+	extensionsNames := []string{"pprof"}
+	if extensionName != "" {
+		extensionsNames = append(extensionsNames, extensionName)
+	}
+	serviceConfig := output.ServiceConfig{
+		Extensions: extensionsNames,
 		Pipelines: struct {
 			Metrics struct {
 				Receivers  []string `json:"receivers" yaml:"receivers"`
@@ -410,9 +416,9 @@ func (c *configBuilder) ReturnConfigYamlFromSink(_ context.Context, kafkaUrlConf
 			},
 		},
 	}
-	config := OtelConfigFile{
-		Receivers: Receivers{
-			Kafka: KafkaReceiver{
+	config := output.OtelConfigFile{
+		Receivers: output.Receivers{
+			Kafka: output.KafkaReceiver{
 				Brokers:         []string{kafkaUrlConfig},
 				Topic:           fmt.Sprintf("otlp_metrics-%s", deployment.SinkID),
 				ProtocolVersion: "2.0.0",

maestro/config/config_builder_test.go

@@ -97,6 +97,28 @@ func TestReturnConfigYamlFromSink(t *testing.T) {
 			want:    `---\nreceivers:\n  kafka:\n    brokers:\n    - kafka:9092\n    topic: otlp_metrics-sink-id-22\n    protocol_version: 2.0.0\nextensions:\n  pprof:\n    endpoint: 0.0.0.0:1888\n  basicauth/exporter:\n    client_auth:\n      username: otlp-user\n      password: dbpass\nexporters:\n  otlphttp:\n    endpoint: https://acme.com/otlphttp/push\n    auth:\n      authenticator: basicauth/exporter\nservice:\n  extensions:\n  - pprof\n  - basicauth/exporter\n  pipelines:\n    metrics:\n      receivers:\n      - kafka\n      exporters:\n      - otlphttp\n`,
 			wantErr: false,
 		},
+		{
+			name: "otlp, noauth",
+			args: args{
+				in0:            context.Background(),
+				kafkaUrlConfig: "kafka:9092",
+				sink: &DeploymentRequest{
+					SinkID:  "sink-id-22",
+					OwnerID: "22",
+					Backend: "otlphttp",
+					Config: types.Metadata{
+						"exporter": types.Metadata{
+							"endpoint": "https://acme.com/otlphttp/push",
+						},
+						"authentication": types.Metadata{
+							"type": "noauth",
+						},
+					},
+				},
+			},
+			want:    `---\nreceivers:\n  kafka:\n    brokers:\n    - kafka:9092\n    topic: otlp_metrics-sink-id-22\n    protocol_version: 2.0.0\nextensions:\n  pprof:\n    endpoint: 0.0.0.0:1888\nexporters:\n  otlphttp:\n    endpoint: https://acme.com/otlphttp/push\nservice:\n  extensions:\n  - pprof\n  pipelines:\n    metrics:\n      receivers:\n      - kafka\n      exporters:\n      - otlphttp\n`,
+			wantErr: false,
+		},
 	}
 	for _, tt := range tests {
 		logger := zap.NewNop()

maestro/config/exporter_builder.go → maestro/config/output/exporter_builder.go

@@ -1,6 +1,8 @@
-package config
+package output
 
-import "github.com/orb-community/orb/pkg/types"
+import (
+	"github.com/orb-community/orb/pkg/types"
+)
 
 type ExporterConfigService interface {
 	GetExportersFromMetadata(config types.Metadata, authenticationExtensionName string) (Exporters, string)
@@ -29,19 +31,24 @@ func (p *PrometheusExporterConfig) GetExportersFromMetadata(config types.Metadat
 	if !ok {
 		return Exporters{}, ""
 	}
+	var auth *Auth
+	if authenticationExtensionName != "" {
+		auth = &Auth{Authenticator: authenticationExtensionName}
+	}
+
 	customHeaders, ok := exporterSubMeta["headers"]
 	if !ok || customHeaders == nil {
 		return Exporters{
 			PrometheusRemoteWrite: &PrometheusRemoteWriteExporterConfig{
 				Endpoint: endpointCfg,
-				Auth:     Auth{Authenticator: authenticationExtensionName},
+				Auth:     auth,
 			},
 		}, "prometheusremotewrite"
 	}
 	return Exporters{
 		PrometheusRemoteWrite: &PrometheusRemoteWriteExporterConfig{
 			Endpoint: endpointCfg,
-			Auth:     Auth{Authenticator: authenticationExtensionName},
+			Auth:     auth,
 			Headers:  customHeaders.(map[string]interface{}),
 		},
 	}, "prometheusremotewrite"
@@ -53,19 +60,23 @@ type OTLPHTTPExporterBuilder struct {
 func (O *OTLPHTTPExporterBuilder) GetExportersFromMetadata(config types.Metadata, authenticationExtensionName string) (Exporters, string) {
 	exporterSubMeta := config.GetSubMetadata("exporter")
 	endpointCfg := exporterSubMeta["endpoint"].(string)
+	var auth *Auth
+	if authenticationExtensionName != "" {
+		auth = &Auth{Authenticator: authenticationExtensionName}
+	}
 	customHeaders, ok := exporterSubMeta["headers"]
 	if !ok || customHeaders == nil {
 		return Exporters{
 			OTLPExporter: &OTLPExporterConfig{
 				Endpoint: endpointCfg,
-				Auth:     Auth{Authenticator: authenticationExtensionName},
+				Auth:     auth,
 			},
 		}, "otlphttp"
 	} else {
 		return Exporters{
 			OTLPExporter: &OTLPExporterConfig{
 				Endpoint: endpointCfg,
-				Auth:     Auth{Authenticator: authenticationExtensionName},
+				Auth:     auth,
 				Headers:  customHeaders.(map[string]interface{}),
 			},
 		}, "otlphttp"

maestro/config/types.go → maestro/config/output/types.go

@@ -1,4 +1,4 @@
-package config
+package output
 
 import (
 	"database/sql/driver"
@@ -136,21 +136,17 @@ type LoggingExporterConfig struct {
 type OTLPExporterConfig struct {
 	Endpoint string                 `json:"endpoint" yaml:"endpoint"`
 	Headers  map[string]interface{} `json:"headers,omitempty" yaml:"headers,omitempty"`
-	Auth     struct {
-		Authenticator string `json:"authenticator" yaml:"authenticator"`
-	}
+	Auth     *Auth                  `json:"auth,omitempty" yaml:"auth,omitempty"`
 }
 
 type Auth struct {
-	Authenticator string `json:"authenticator" yaml:"authenticator"`
+	Authenticator string `json:"authenticator,omitempty" yaml:"authenticator"`
 }
 
 type PrometheusRemoteWriteExporterConfig struct {
 	Endpoint string                 `json:"endpoint" yaml:"endpoint"`
 	Headers  map[string]interface{} `json:"headers,omitempty" yaml:"headers,omitempty"`
-	Auth     struct {
-		Authenticator string `json:"authenticator" yaml:"authenticator"`
-	}
+	Auth     *Auth                  `json:"auth,omitempty" yaml:"auth,omitempty"`
 }
 
 type ServiceConfig struct {

maestro/deployment/service.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/orb-community/orb/maestro/config/authentication"
 	"time"
 
 	"github.com/orb-community/orb/maestro/config"
@@ -85,8 +86,8 @@ func (d *deploymentService) CreateDeployment(ctx context.Context, deployment *De
 	return nil
 }
 
-func (d *deploymentService) getAuthBuilder(authType string) config.AuthBuilderService {
-	return config.GetAuthService(authType, d.encryptionService)
+func (d *deploymentService) getAuthBuilder(authType string) authentication.AuthBuilderService {
+	return authentication.GetAuthService(authType, d.encryptionService)
 }
 
 func (d *deploymentService) encodeConfig(deployment *Deployment) (types.Metadata, error) {

maestro/monitor/monitor.go

@@ -5,14 +5,14 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	maestroconfig "github.com/orb-community/orb/maestro/config/output"
 	"io"
 	"strings"
 	"time"
 
 	"github.com/orb-community/orb/maestro/deployment"
 	"github.com/orb-community/orb/maestro/redis/producer"
 
-	maestroconfig "github.com/orb-community/orb/maestro/config"
 	"github.com/orb-community/orb/maestro/kubecontrol"
 	sinkspb "github.com/orb-community/orb/sinks/pb"
 	"go.uber.org/zap"