Revert "add changes to production"

.github/workflows/go-develop.yml

@@ -643,7 +643,6 @@ 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.27.4/bin/linux/amd64/kubectl" && \
+	curl -LO "https://dl.k8s.io/release/v1.22.1/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.24.0 --config=./kind/config.yaml
+	kind create cluster --image kindest/node:v1.22.15 --config=./kind/config.yaml
 
 kind-delete-cluster:
 	kind delete cluster

VERSION

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

cmd/maestro/main.go

@@ -11,8 +11,6 @@ package main
 import (
 	"context"
 	"fmt"
-	"github.com/jmoiron/sqlx"
-	"github.com/orb-community/orb/maestro/postgres"
 	"io"
 	"os"
 	"os/signal"
@@ -38,10 +36,9 @@ import (
 )
 
 const (
-	svcName    = "maestro"
-	envPrefix  = "orb_maestro"
-	sinkPrefix = "orb_sinks"
-	httpPort   = "8500"
+	svcName   = "maestro"
+	envPrefix = "orb_maestro"
+	httpPort  = "8500"
 )
 
 func main() {
@@ -51,9 +48,6 @@ 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
@@ -116,10 +110,8 @@ 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, otelCfg, db, svcCfg)
+	svc := maestro.NewMaestroService(logger, streamEsClient, sinkerEsClient, sinksGRPCClient, streamEsCfg, otelCfg)
 	errs := make(chan error, 2)
 
 	mainContext, mainCancelFunction := context.WithCancel(context.Background())
@@ -140,15 +132,6 @@ 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)
@@ -228,7 +211,7 @@ func loadStreamEsConfig(prefix string) config.EsConfig {
 	cfg.AllowEmptyEnv(true)
 	cfg.AutomaticEnv()
 	var esC config.EsConfig
-	_ = cfg.Unmarshal(&esC)
+	cfg.Unmarshal(&esC)
 	return esC
 }
 
@@ -243,6 +226,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,6 +9,7 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	kitprometheus "github.com/go-kit/kit/metrics/prometheus"
 	"github.com/go-redis/redis/v8"
@@ -19,6 +20,10 @@ 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"
@@ -97,12 +102,6 @@ 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) {
@@ -169,6 +168,8 @@ 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",
@@ -191,7 +192,7 @@ func main() {
 	otelEnabled := otelCfg.Enable == "true"
 	otelKafkaUrl := otelCfg.KafkaUrl
 
-	svc := sinker.New(logger, pubSub, esClient, cacheClient, policiesGRPCClient, fleetGRPCClient, sinksGRPCClient,
+	svc := sinker.New(logger, pubSub, esClient, configRepo, policiesGRPCClient, fleetGRPCClient, sinksGRPCClient,
 		otelKafkaUrl, otelEnabled, gauge, counter, inputCounter, inMemoryCacheConfig.DefaultExpiration)
 	defer func(svc sinker.Service) {
 		err := svc.Stop()
@@ -203,6 +204,7 @@ 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 {
@@ -305,3 +307,11 @@ 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,18 +11,6 @@ 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"
@@ -39,6 +27,17 @@ 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"
@@ -127,7 +126,6 @@ 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)
@@ -195,7 +193,7 @@ func newSinkService(auth mainflux.AuthServiceClient, logger *zap.Logger, esClien
 	mfsdk := mfsdk.NewSDK(config)
 
 	svc := sinks.NewSinkService(logger, auth, repoSink, mfsdk, passwordService)
-	svc = redisprod.NewSinkStreamProducerMiddleware(svc, esClient)
+	svc = redisprod.NewEventStoreMiddleware(svc, esClient)
 	svc = sinkshttp.NewLoggingMiddleware(svc, logger)
 	svc = sinkshttp.MetricsMiddleware(
 		auth,
@@ -288,14 +286,6 @@ 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 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))
+		logger.Error("Bootstrap service failed to subscribe to 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.27.4/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.18.0/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 /