[Fix] Sinker state checker for otel (#1958)

* feat(sinker): verify if sink uses opentelemetry before remove from cache

* feat(sinker): add log

* fix(sinker): edit cache instead of removing sink

* test

* remove typo

* remove

* remove strconv

* back to right

Co-authored-by: mclcavalcante <mariana.cavalcante@encora.com>

maestro/config/config_builder.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"github.com/ns1labs/orb/pkg/errors"
 	"gopkg.in/yaml.v2"
-	"strconv"
 	"strings"
 )
 
@@ -180,9 +179,6 @@ func GetDeploymentJson(sinkId, sinkUrl, sinkUsername, sinkPassword string) (stri
 
 // ReturnConfigYamlFromSink this is the main method, which will generate the YAML file from the
 func ReturnConfigYamlFromSink(_ context.Context, kafkaUrlConfig, sinkId, sinkUrl, sinkUsername, sinkPassword string) (string, error) {
-	if _, err := strconv.Atoi(sinkUsername); err != nil {
-		sinkUsername = "#$#" + sinkUsername
-	}
 	config := OtelConfigFile{
 		Receivers: Receivers{
 			Kafka: KafkaReceiver{

sinker/config_state_check.go

@@ -35,14 +35,21 @@ func (svc *SinkerService) checkState(_ time.Time) {
 			if cfg.LastRemoteWrite.Add(DefaultTimeout).Before(time.Now()) {
 				svc.logger.Info("opentelemetry:", zap.String("otel", cfg.Opentelemetry))
 				if cfg.State == config.Active {
-					err := cfg.State.SetFromString("idle")
-					if err != nil {
-						svc.logger.Error("error updating sink config cache", zap.Error(err))
-						return
-					}
-					if err := svc.sinkerCache.Edit(cfg); err != nil {
-						svc.logger.Error("error updating sink config cache", zap.Error(err))
-						return
+					if cfg.Opentelemetry == "enabled" {
+						err := cfg.State.SetFromString("idle")
+						if err != nil {
+							svc.logger.Error("error updating sink state otel", zap.Error(err))
+							return
+						}
+						if err := svc.sinkerCache.Edit(cfg); err != nil {
+							svc.logger.Error("error updating sink config cache otel", zap.Error(err))
+							return
+						}
+					} else {
+						if err := svc.sinkerCache.Remove(cfg.OwnerID, cfg.SinkID); err != nil {
+							svc.logger.Error("error updating sink config cache", zap.Error(err))
+							return
+						}
 					}
 				}
 			}