add changes to production

.github/workflows/go-develop.yml

@@ -643,6 +643,7 @@ jobs:
       - package-policies
       - package-sinker
       - package-sinks
+      - package-maestro
       - package-ui
 
     runs-on: ubuntu-latest

Makefile

@@ -153,7 +153,7 @@ install-helm:
 
 install-kubectl:
 	cd /tmp && \
-	curl -LO "https://dl.k8s.io/release/v1.22.1/bin/linux/amd64/kubectl" && \
+	curl -LO "https://dl.k8s.io/release/v1.27.4/bin/linux/amd64/kubectl" && \
 	chmod a+x ./kubectl && \
 	sudo mv ./kubectl /usr/local/bin/kubectl
 
@@ -182,7 +182,7 @@ kind-create-all: kind-create-cluster kind-install-orb
 kind-upgrade-all: kind-load-images kind-upgrade-orb
 
 kind-create-cluster:
-	kind create cluster --image kindest/node:v1.22.15 --config=./kind/config.yaml
+	kind create cluster --image kindest/node:v1.24.0 --config=./kind/config.yaml
 
 kind-delete-cluster:
 	kind delete cluster

VERSION

@@ -1 +1 @@
-0.27.0
+0.28.0

cmd/maestro/main.go

@@ -11,6 +11,8 @@ package main
 import (
 	"context"
 	"fmt"
+	"github.com/jmoiron/sqlx"
+	"github.com/orb-community/orb/maestro/postgres"
 	"io"
 	"os"
 	"os/signal"
@@ -36,9 +38,10 @@ import (
 )
 
 const (
-	svcName   = "maestro"
-	envPrefix = "orb_maestro"
-	httpPort  = "8500"
+	svcName    = "maestro"
+	envPrefix  = "orb_maestro"
+	sinkPrefix = "orb_sinks"
+	httpPort   = "8500"
 )
 
 func main() {
@@ -48,6 +51,9 @@ func main() {
 	svcCfg := config.LoadBaseServiceConfig(envPrefix, httpPort)
 	jCfg := config.LoadJaegerConfig(envPrefix)
 	sinksGRPCCfg := config.LoadGRPCConfig("orb", "sinks")
+	dbCfg := config.LoadPostgresConfig(envPrefix, svcName)
+	encryptionKey := config.LoadEncryptionKey(sinkPrefix)
+	svcCfg.EncryptionKey = encryptionKey.Key
 
 	// logger
 	var logger *zap.Logger
@@ -110,8 +116,10 @@ func main() {
 	}
 	sinksGRPCClient := sinksgrpc.NewClient(tracer, sinksGRPCConn, sinksGRPCTimeout, logger)
 	otelCfg := config.LoadOtelConfig(envPrefix)
+	db := connectToDB(dbCfg, logger)
+	defer db.Close()
 
-	svc := maestro.NewMaestroService(logger, streamEsClient, sinkerEsClient, sinksGRPCClient, streamEsCfg, otelCfg)
+	svc := maestro.NewMaestroService(logger, streamEsClient, sinkerEsClient, sinksGRPCClient, otelCfg, db, svcCfg)
 	errs := make(chan error, 2)
 
 	mainContext, mainCancelFunction := context.WithCancel(context.Background())
@@ -132,6 +140,15 @@ func main() {
 	logger.Error(fmt.Sprintf("Maestro service terminated: %s", err))
 }
 
+func connectToDB(cfg config.PostgresConfig, logger *zap.Logger) *sqlx.DB {
+	db, err := postgres.Connect(cfg)
+	if err != nil {
+		logger.Error("Failed to connect to postgres", zap.Error(err))
+		os.Exit(1)
+	}
+	return db
+}
+
 func connectToGRPC(cfg config.GRPCConfig, logger *zap.Logger) *grpc.ClientConn {
 	var opts []grpc.DialOption
 	tls, err := strconv.ParseBool(cfg.ClientTLS)
@@ -211,7 +228,7 @@ func loadStreamEsConfig(prefix string) config.EsConfig {
 	cfg.AllowEmptyEnv(true)
 	cfg.AutomaticEnv()
 	var esC config.EsConfig
-	cfg.Unmarshal(&esC)
+	_ = cfg.Unmarshal(&esC)
 	return esC
 }
 
@@ -226,6 +243,6 @@ func loadSinkerEsConfig(prefix string) config.EsConfig {
 	cfg.AllowEmptyEnv(true)
 	cfg.AutomaticEnv()
 	var esC config.EsConfig
-	cfg.Unmarshal(&esC)
+	_ = cfg.Unmarshal(&esC)
 	return esC
 }

cmd/sinker/main.go

@@ -9,7 +9,6 @@
 package main
 
 import (
-	"context"
 	"fmt"
 	kitprometheus "github.com/go-kit/kit/metrics/prometheus"
 	"github.com/go-redis/redis/v8"
@@ -20,10 +19,6 @@ import (
 	"github.com/orb-community/orb/pkg/config"
 	policiesgrpc "github.com/orb-community/orb/policies/api/grpc"
 	"github.com/orb-community/orb/sinker"
-	sinkconfig "github.com/orb-community/orb/sinker/config"
-	cacheconfig "github.com/orb-community/orb/sinker/redis"
-	"github.com/orb-community/orb/sinker/redis/consumer"
-	"github.com/orb-community/orb/sinker/redis/producer"
 	sinksgrpc "github.com/orb-community/orb/sinks/api/grpc"
 	stdprometheus "github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -102,6 +97,12 @@ func main() {
 	}
 
 	cacheClient := connectToRedis(cacheCfg.URL, cacheCfg.Pass, cacheCfg.DB, logger)
+	defer func(client *redis.Client) {
+		err := client.Close()
+		if err != nil {
+			log.Fatalf(err.Error())
+		}
+	}(cacheClient)
 
 	esClient := connectToRedis(esCfg.URL, esCfg.Pass, esCfg.DB, logger)
 	defer func(esClient *redis.Client) {
@@ -168,8 +169,6 @@ func main() {
 	}
 	sinksGRPCClient := sinksgrpc.NewClient(tracer, sinksGRPCConn, sinksGRPCTimeout, logger)
 
-	configRepo := cacheconfig.NewSinkerCache(cacheClient, logger)
-	configRepo = producer.NewEventStoreMiddleware(configRepo, esClient, logger)
 	gauge := kitprometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
 		Namespace: "sinker",
 		Subsystem: "sink",
@@ -192,7 +191,7 @@ func main() {
 	otelEnabled := otelCfg.Enable == "true"
 	otelKafkaUrl := otelCfg.KafkaUrl
 
-	svc := sinker.New(logger, pubSub, esClient, configRepo, policiesGRPCClient, fleetGRPCClient, sinksGRPCClient,
+	svc := sinker.New(logger, pubSub, esClient, cacheClient, policiesGRPCClient, fleetGRPCClient, sinksGRPCClient,
 		otelKafkaUrl, otelEnabled, gauge, counter, inputCounter, inMemoryCacheConfig.DefaultExpiration)
 	defer func(svc sinker.Service) {
 		err := svc.Stop()
@@ -204,7 +203,6 @@ func main() {
 	errs := make(chan error, 2)
 
 	go startHTTPServer(svcCfg, errs, logger)
-	go subscribeToSinksES(svc, configRepo, esClient, esCfg, logger)
 
 	err = svc.Start()
 	if err != nil {
@@ -307,11 +305,3 @@ func initJaeger(svcName, url string, logger *zap.Logger) (opentracing.Tracer, io
 
 	return tracer, closer
 }
-
-func subscribeToSinksES(svc sinker.Service, configRepo sinkconfig.ConfigRepo, client *redis.Client, cfg config.EsConfig, logger *zap.Logger) {
-	eventStore := consumer.NewEventStore(svc, configRepo, client, cfg.Consumer, logger)
-	logger.Info("Subscribed to Redis Event Store for sinks")
-	if err := eventStore.Subscribe(context.Background()); err != nil {
-		logger.Error("Bootstrap service failed to subscribe to event sourcing", zap.Error(err))
-	}
-}

cmd/sinks/main.go

@@ -11,6 +11,18 @@ package main
 import (
 	"context"
 	"fmt"
+	"io"
+	"io/ioutil"
+	"log"
+	"net"
+	"net/http"
+	"os"
+	"os/signal"
+	"strconv"
+	"strings"
+	"syscall"
+	"time"
+
 	authapi "github.com/mainflux/mainflux/auth/api/grpc"
 	mfsdk "github.com/mainflux/mainflux/pkg/sdk/go"
 	opentracing "github.com/opentracing/opentracing-go"
@@ -27,17 +39,6 @@ import (
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"google.golang.org/grpc/reflection"
-	"io"
-	"io/ioutil"
-	"log"
-	"net"
-	"net/http"
-	"os"
-	"os/signal"
-	"strconv"
-	"strings"
-	"syscall"
-	"time"
 
 	kitprometheus "github.com/go-kit/kit/metrics/prometheus"
 	r "github.com/go-redis/redis/v8"
@@ -126,6 +127,7 @@ func main() {
 	go startHTTPServer(tracer, svc, svcCfg, logger, errs)
 	go startGRPCServer(svc, tracer, sinksGRPCCfg, logger, errs)
 	go subscribeToSinkerES(svc, esClient, esCfg, logger)
+	go subscribeToMaestroStatusES(svc, esClient, esCfg, logger)
 
 	go func() {
 		c := make(chan os.Signal)
@@ -193,7 +195,7 @@ func newSinkService(auth mainflux.AuthServiceClient, logger *zap.Logger, esClien
 	mfsdk := mfsdk.NewSDK(config)
 
 	svc := sinks.NewSinkService(logger, auth, repoSink, mfsdk, passwordService)
-	svc = redisprod.NewEventStoreMiddleware(svc, esClient)
+	svc = redisprod.NewSinkStreamProducerMiddleware(svc, esClient)
 	svc = sinkshttp.NewLoggingMiddleware(svc, logger)
 	svc = sinkshttp.MetricsMiddleware(
 		auth,
@@ -286,6 +288,14 @@ func subscribeToSinkerES(svc sinks.SinkService, client *r.Client, cfg config.EsC
 	eventStore := rediscons.NewEventStore(svc, client, cfg.Consumer, logger)
 	logger.Info("Subscribed to Redis Event Store for sinker")
 	if err := eventStore.Subscribe(context.Background()); err != nil {
-		logger.Error("Bootstrap service failed to subscribe to event sourcing", zap.Error(err))
+		logger.Error("Bootstrap service failed to subscribe to sinker event sourcing", zap.Error(err))
+	}
+}
+
+func subscribeToMaestroStatusES(svc sinks.SinkService, client *r.Client, cfg config.EsConfig, logger *zap.Logger) {
+	eventStore := rediscons.NewSinkStatusListener(logger, client, svc)
+	logger.Info("Subscribed to Redis Event Store for maestro")
+	if err := eventStore.SubscribeToMaestroSinkStatus(context.Background()); err != nil {
+		logger.Error("Bootstrap service failed to subscribe to maestro event sourcing", zap.Error(err))
 	}
 }

docker/Dockerfile

@@ -13,7 +13,7 @@ RUN make $SVC \
 FROM alpine:latest
 ARG SVC
 
-RUN if [[ "maestro" == "$SVC" ]]; then apk update && apk add --no-cache docker-cli bash curl && curl -LO https://storage.googleapis.com/kubernetes-release/release/v1.18.0/bin/linux/amd64/kubectl && chmod +x ./kubectl && mv ./kubectl /usr/local/bin/kubectl; fi
+RUN if [[ "maestro" == "$SVC" ]]; then apk update && apk add --no-cache docker-cli bash curl && curl -LO https://storage.googleapis.com/kubernetes-release/release/v1.27.4/bin/linux/amd64/kubectl && chmod +x ./kubectl && mv ./kubectl /usr/local/bin/kubectl; fi
 # Certificates are needed so that mailing util can work.
 COPY --from=builder /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt
 COPY --from=builder /exe /