refactor: use hashicorp go-plugin

.github/workflows/release.yaml

@@ -41,10 +41,10 @@ jobs:
           username: ${{ secrets.DOCKERHUB_USERNAME }}
           password: ${{ secrets.DOCKERHUB_TOKEN }}
       - name: Run GoReleaser
-        uses: goreleaser/goreleaser-action@v2.6.1
+        uses: goreleaser/goreleaser-action@v4
         with:
           distribution: goreleaser
           version: latest
-          args: --rm-dist
+          args: release --clean
         env:
           GITHUB_TOKEN: ${{ secrets.GO_RELEASER_TOKEN }}

.github/workflows/release_cortex_provider_plugin.yaml

@@ -4,7 +4,7 @@ on:
   workflow_dispatch:
 
 jobs:
-  release:
+  release_plugin:
     runs-on: ubuntu-latest
     steps:
       - name: Set up Go 1.x

.gitignore

@@ -29,5 +29,6 @@ tmp
 dist/
 siren
 !siren/
+plugin/
 
 third_party/

.goreleaser.yml

@@ -67,7 +67,7 @@ brews:
   - name: siren
     homepage: "https://github.com/goto/siren"
     description: "Universal data observability tool."
-    tap:
+    repository:
       owner: goto
       name: homebrew-taps
     license: "Apache 2.0"

Makefile

@@ -2,21 +2,30 @@ NAME="github.com/goto/siren"
 LAST_COMMIT := $(shell git rev-parse --short HEAD)
 LAST_TAG := "$(shell git rev-list --tags --max-count=1)"
 APP_VERSION := "$(shell git describe --tags ${LAST_TAG})-next"
-PROTON_COMMIT := "96e1f08d485a1ff38d0939f6533726fb68f36126"
+PROTON_COMMIT := "4fc96a5406f640044ebe0572385336af7bd221ae"
 
 .PHONY: all build test clean dist vet proto install
 
 all: build
 
-build: ## Build the siren binary
+
+build: build-main build-plugins
+
+build-main: ## Build the siren binary
 	@echo " > building siren version ${APP_VERSION}"
 	go build -ldflags "-X main.Version=${APP_VERSION}" ${NAME}
+	@echo " > building plugins version ${APP_VERSION}"
+	@echo " - build complete"
+
+build-plugins:
+	@echo " > building plugins"
+	@go list -m | grep providers | while read path; do go build -o ./plugin/${shell basename "$$path"} "$$path"; done
 	@echo " - build complete"
 
 test: ## Run the tests
 	go test -race $(shell go list ./... | grep -v /test/) -covermode=atomic -coverprofile=coverage.out
 
-e2e-test: ## Run all e2e tests
+e2e-test: build-plugins ## Run all e2e tests
 	go test -v -race ./test/e2e_test/... -coverprofile=coverage.out --timeout 300s
 
 coverage: ## Print code coverage

buf.gen.yaml

@@ -8,7 +8,7 @@ plugins:
     opt: paths=source_relative,require_unimplemented_servers=true
   - plugin: buf.build/bufbuild/validate-go:v0.9.0
     out: proto
-    opt: "paths=source_relative"
+    opt: paths=source_relative
   - plugin: buf.build/grpc-ecosystem/gateway:v2.15.1
     out: proto
     opt:

cli/deps.go

@@ -22,7 +22,7 @@ import (
 	"github.com/goto/siren/pkg/pgc"
 	"github.com/goto/siren/pkg/secret"
 	"github.com/goto/siren/pkg/telemetry"
-	"github.com/goto/siren/plugins/providers/cortex"
+	"github.com/goto/siren/plugins/providers"
 	"github.com/goto/siren/plugins/receivers/file"
 	"github.com/goto/siren/plugins/receivers/httpreceiver"
 	"github.com/goto/siren/plugins/receivers/pagerduty"
@@ -36,7 +36,7 @@ func InitDeps(
 	logger saltlog.Logger,
 	cfg config.Config,
 	queue notification.Queuer,
-) (*api.Deps, *newrelic.Application, *pgc.Client, map[string]notification.Notifier, error) {
+) (*api.Deps, *newrelic.Application, *pgc.Client, map[string]notification.Notifier, *providers.PluginManager, error) {
 
 	telemetry.Init(ctx, cfg.Telemetry, logger)
 
@@ -57,17 +57,17 @@ func InitDeps(
 
 	dbClient, err := db.New(cfg.DB)
 	if err != nil {
-		return nil, nil, nil, nil, err
+		return nil, nil, nil, nil, nil, err
 	}
 
 	pgClient, err := pgc.NewClient(logger, dbClient)
 	if err != nil {
-		return nil, nil, nil, nil, err
+		return nil, nil, nil, nil, nil, err
 	}
 
 	encryptor, err := secret.New(cfg.Service.EncryptionKey)
 	if err != nil {
-		return nil, nil, nil, nil, fmt.Errorf("cannot initialize encryptor: %w", err)
+		return nil, nil, nil, nil, nil, fmt.Errorf("cannot initialize encryptor: %w", err)
 	}
 
 	templateRepository := postgres.NewTemplateRepository(pgClient)
@@ -79,29 +79,42 @@ func InitDeps(
 	logRepository := postgres.NewLogRepository(pgClient)
 	logService := log.NewService(logRepository)
 
-	cortexPluginService := cortex.NewPluginService(logger, cfg.Providers.Cortex)
+	providersPluginManager := providers.NewPluginManager(logger, cfg.Providers)
+	providerPluginClients := providersPluginManager.InitClients()
+	providerPlugins, err := providersPluginManager.DispenseClients(providerPluginClients)
+	if err != nil {
+		return nil, nil, nil, nil, nil, err
+	}
+	if err := providersPluginManager.InitConfigs(ctx, providerPlugins, cfg.Log.Level); err != nil {
+		return nil, nil, nil, nil, nil, err
+	}
+
+	var configSyncers = make(map[string]namespace.ConfigSyncer, 0)
+	var alertTransformers = make(map[string]alert.AlertTransformer, 0)
+	var ruleUploaders = make(map[string]rule.RuleUploader, 0)
+
+	for k, pc := range providerPlugins {
+		alertTransformers[k] = pc.(alert.AlertTransformer)
+		configSyncers[k] = pc.(namespace.ConfigSyncer)
+		ruleUploaders[k] = pc.(rule.RuleUploader)
+	}
+
 	alertRepository := postgres.NewAlertRepository(pgClient)
 	alertService := alert.NewService(
 		alertRepository,
 		logService,
-		map[string]alert.AlertTransformer{
-			provider.TypeCortex: cortexPluginService,
-		},
+		alertTransformers,
 	)
 
 	namespaceRepository := postgres.NewNamespaceRepository(pgClient)
-	namespaceService := namespace.NewService(encryptor, namespaceRepository, providerService, map[string]namespace.ConfigSyncer{
-		provider.TypeCortex: cortexPluginService,
-	})
+	namespaceService := namespace.NewService(encryptor, namespaceRepository, providerService, configSyncers)
 
 	ruleRepository := postgres.NewRuleRepository(pgClient)
 	ruleService := rule.NewService(
 		ruleRepository,
 		templateService,
 		namespaceService,
-		map[string]rule.RuleUploader{
-			provider.TypeCortex: cortexPluginService,
-		},
+		ruleUploaders,
 	)
 
 	silenceRepository := postgres.NewSilenceRepository(pgClient)
@@ -171,6 +184,6 @@ func InitDeps(
 			SubscriptionService: subscriptionService,
 			NotificationService: notificationService,
 			SilenceService:      silenceService,
-		}, nrApp, pgClient, notifierRegistry,
+		}, nrApp, pgClient, notifierRegistry, providersPluginManager,
 		nil
 }

cli/job.go

@@ -12,6 +12,7 @@ import (
 	"github.com/goto/siren/core/notification"
 	"github.com/goto/siren/internal/jobs"
 	"github.com/goto/siren/pkg/errors"
+	"github.com/goto/siren/pkg/zaputil"
 	"github.com/goto/siren/plugins/queues"
 	"github.com/goto/siren/plugins/queues/postgresq"
 	"github.com/spf13/cobra"
@@ -153,9 +154,9 @@ func jobRunCleanupIdempotencyCommand() *cobra.Command {
 				return err
 			}
 
-			logger := initLogger(cfg.Log)
+			logger := zaputil.InitLogger(serviceName, cfg.Log.Level, cfg.Log.GCPCompatible)
 
-			apiDeps, _, pgClient, _, err := InitDeps(cmd.Context(), logger, cfg, nil)
+			apiDeps, _, pgClient, _, providersPluginManager, err := InitDeps(cmd.Context(), logger, cfg, nil)
 			if err != nil {
 				return err
 			}
@@ -188,6 +189,8 @@ func jobRunCleanupIdempotencyCommand() *cobra.Command {
 				logger.Error(err.Error())
 			}
 
+			providersPluginManager.Stop()
+
 			return nil
 		},
 	}

cli/server.go

@@ -6,7 +6,6 @@ import (
 	"time"
 
 	"github.com/MakeNowJust/heredoc"
-	"github.com/goto/salt/log"
 	"github.com/goto/salt/printer"
 	"github.com/goto/siren/config"
 	"github.com/goto/siren/core/notification"
@@ -18,8 +17,6 @@ import (
 	"github.com/goto/siren/plugins/queues/inmemory"
 	"github.com/goto/siren/plugins/queues/postgresq"
 	"github.com/spf13/cobra"
-	"go.uber.org/zap"
-	"go.uber.org/zap/zapcore"
 )
 
 func serverCmd() *cobra.Command {
@@ -123,7 +120,7 @@ func serverMigrateCommand() *cobra.Command {
 }
 
 func StartServer(ctx context.Context, cfg config.Config) error {
-	logger := initLogger(cfg.Log)
+	logger := zaputil.InitLogger(serviceName, cfg.Log.Level, cfg.Log.GCPCompatible)
 
 	var queue, dlq notification.Queuer
 	var err error
@@ -142,7 +139,7 @@ func StartServer(ctx context.Context, cfg config.Config) error {
 		dlq = inmemory.New(logger, 10)
 	}
 
-	apiDeps, nrApp, pgClient, notifierRegistry, err := InitDeps(ctx, logger, cfg, queue)
+	apiDeps, nrApp, pgClient, notifierRegistry, providersPluginManager, err := InitDeps(ctx, logger, cfg, queue)
 	if err != nil {
 		return err
 	}
@@ -196,6 +193,10 @@ func StartServer(ctx context.Context, cfg config.Config) error {
 	timeoutCtx, cancel := context.WithTimeout(context.Background(), gracefulStopQueueWaitPeriod)
 	defer cancel()
 
+	logger.Warn("stopping plugins")
+	providersPluginManager.Stop()
+	logger.Warn("all plugins stopped")
+
 	logger.Warn("stopping queue...")
 	if err := queue.Stop(timeoutCtx); err != nil {
 		logger.Error("error stopping queue", "error", err)
@@ -214,55 +215,11 @@ func StartServer(ctx context.Context, cfg config.Config) error {
 	}
 	logger.Warn("db closed...")
 
-	return err
-}
-
-func initLogger(cfg config.Log) log.Logger {
-	defaultConfig := zap.NewProductionConfig()
-	defaultConfig.Level = zap.NewAtomicLevelAt(getZapLogLevelFromString(cfg.Level))
-
-	if cfg.GCPCompatible {
-		defaultConfig = zap.Config{
-			Level:       zap.NewAtomicLevelAt(getZapLogLevelFromString(cfg.Level)),
-			Encoding:    "json",
-			Development: false,
-			Sampling: &zap.SamplingConfig{
-				Initial:    100,
-				Thereafter: 100,
-			},
-			EncoderConfig:    zaputil.EncoderConfig,
-			OutputPaths:      []string{"stdout"},
-			ErrorOutputPaths: []string{"stderr"},
-		}
+	logger.Warn("exiting plugins...")
+	if err := pgClient.Close(); err != nil {
+		logger.Error("error when closing db", "err", err)
 	}
+	logger.Warn("plugins exited...")
 
-	return log.NewZap(log.ZapWithConfig(
-		defaultConfig,
-		zap.Fields(zaputil.ServiceContext(serviceName)),
-		zap.AddCaller(),
-		zap.AddStacktrace(zap.DPanicLevel),
-	))
-
-}
-
-// getZapLogLevelFromString helps to set logLevel from string
-func getZapLogLevelFromString(level string) zapcore.Level {
-	switch level {
-	case "debug":
-		return zap.DebugLevel
-	case "info":
-		return zap.InfoLevel
-	case "warn":
-		return zap.WarnLevel
-	case "error":
-		return zap.ErrorLevel
-	case "dpanic":
-		return zap.DPanicLevel
-	case "panic":
-		return zap.PanicLevel
-	case "fatal":
-		return zap.FatalLevel
-	default:
-		return zap.InfoLevel
-	}
+	return err
 }

cli/worker.go

@@ -9,6 +9,7 @@ import (
 	"github.com/goto/siren/config"
 	"github.com/goto/siren/core/notification"
 	"github.com/goto/siren/pkg/worker"
+	"github.com/goto/siren/pkg/zaputil"
 	"github.com/goto/siren/plugins/queues"
 	"github.com/goto/siren/plugins/queues/postgresq"
 	"github.com/spf13/cobra"
@@ -128,9 +129,9 @@ func workerStartNotificationDLQHandlerCommand() *cobra.Command {
 }
 
 func StartNotificationHandlerWorker(ctx context.Context, cfg config.Config, cancelWorkerChan chan struct{}) error {
-	logger := initLogger(cfg.Log)
+	logger := zaputil.InitLogger(serviceName, cfg.Log.Level, cfg.Log.GCPCompatible)
 
-	_, nrApp, pgClient, notifierRegistry, err := InitDeps(ctx, logger, cfg, nil)
+	_, nrApp, pgClient, notifierRegistry, providersPluginManager, err := InitDeps(ctx, logger, cfg, nil)
 	if err != nil {
 		return err
 	}
@@ -158,6 +159,10 @@ func StartNotificationHandlerWorker(ctx context.Context, cfg config.Config, canc
 			return notificationHandler.Process(ctx, runningAt)
 		})
 
+		logger.Warn("stopping plugins")
+		providersPluginManager.Stop()
+		logger.Warn("all plugins stopped")
+
 		logger.Info("closing all clients")
 		if err := pgClient.Close(); err != nil {
 			logger.Error(err.Error())
@@ -168,9 +173,9 @@ func StartNotificationHandlerWorker(ctx context.Context, cfg config.Config, canc
 }
 
 func StartNotificationDLQHandlerWorker(ctx context.Context, cfg config.Config, cancelWorkerChan chan struct{}) error {
-	logger := initLogger(cfg.Log)
+	logger := zaputil.InitLogger(serviceName, cfg.Log.Level, cfg.Log.GCPCompatible)
 
-	_, nrApp, pgClient, notifierRegistry, err := InitDeps(ctx, logger, cfg, nil)
+	_, nrApp, pgClient, notifierRegistry, providersPluginManager, err := InitDeps(ctx, logger, cfg, nil)
 	if err != nil {
 		return err
 	}
@@ -198,6 +203,10 @@ func StartNotificationDLQHandlerWorker(ctx context.Context, cfg config.Config, c
 			return notificationHandler.Process(ctx, runningAt)
 		})
 
+		logger.Warn("stopping plugins")
+		providersPluginManager.Stop()
+		logger.Warn("all plugins stopped")
+
 		logger.Info("closing all clients")
 		if err := pgClient.Close(); err != nil {
 			logger.Error(err.Error())

config/config.go

@@ -10,7 +10,7 @@ import (
 	"github.com/goto/siren/internal/server"
 	"github.com/goto/siren/pkg/errors"
 	"github.com/goto/siren/pkg/telemetry"
-	"github.com/goto/siren/plugins/providers"
+	"github.com/goto/siren/plugins"
 	"github.com/goto/siren/plugins/receivers"
 )
 
@@ -41,7 +41,7 @@ type Config struct {
 	Telemetry    telemetry.Config    `mapstructure:"telemetry" yaml:"telemetry"`
 	Service      server.Config       `mapstructure:"service" yaml:"service"`
 	Log          Log                 `mapstructure:"log" yaml:"log"`
-	Providers    providers.Config    `mapstructure:"providers" yaml:"providers"`
+	Providers    plugins.Config      `mapstructure:"providers" yaml:"providers"`
 	Receivers    receivers.Config    `mapstructure:"receivers" yaml:"receivers"`
 	Notification notification.Config `mapstructure:"notification" yaml:"notification"`
 }