Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Fix] Sinker state checker for otel #1958

Merged
merged 9 commits into from
Nov 2, 2022
Merged

[Fix] Sinker state checker for otel #1958

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
c8946ab
feat(sinker): verify if sink uses opentelemetry before remove from cache
mclcavalcante Nov 2, 2022
47d1cca
feat(sinker): add log
mclcavalcante Nov 2, 2022
e1eb37e
fix(sinker): edit cache instead of removing sink
mclcavalcante Nov 2, 2022
ee95fb1
test
etaques Nov 2, 2022
d4c02f8
Merge branch 'develop' into sinker-test
etaques Nov 2, 2022
beb97f5
remove typo
etaques Nov 2, 2022
c2e92b2
remove
etaques Nov 2, 2022
1efd3c4
remove strconv
etaques Nov 2, 2022
a582cc2
back to right
etaques Nov 2, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 0 additions & 4 deletions maestro/config/config_builder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@ import (
"fmt"
"github.com/ns1labs/orb/pkg/errors"
"gopkg.in/yaml.v2"
"strconv"
"strings"
)

Expand Down Expand Up @@ -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{
Expand Down
23 changes: 15 additions & 8 deletions sinker/config_state_check.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
}
}
}
Expand Down