From e2ed140de2517136a164eb76db5202d51cdc0f25 Mon Sep 17 00:00:00 2001 From: Alexander Emelin Date: Tue, 9 Nov 2021 19:12:10 +0300 Subject: [PATCH] Live: rename remote write backends to write configs (#41467) --- pkg/api/api.go | 8 +- pkg/services/live/live.go | 74 ++++----- pkg/services/live/pipeline/config.go | 140 ++++++++++-------- .../live/pipeline/data_output_loki.go | 4 +- pkg/services/live/pipeline/devdata.go | 24 ++- .../live/pipeline/frame_output_loki.go | 22 ++- .../pipeline/frame_output_remote_write.go | 17 +-- .../frame_output_remote_write_test.go | 4 +- pkg/services/live/pipeline/storage_file.go | 118 +++++++-------- .../features/live/pages/CloudAdminPage.tsx | 4 +- 10 files changed, 210 insertions(+), 205 deletions(-) diff --git a/pkg/api/api.go b/pkg/api/api.go index 5e332290bb1..3366ddb177e 100644 --- a/pkg/api/api.go +++ b/pkg/api/api.go @@ -441,10 +441,10 @@ func (hs *HTTPServer) registerRoutes() { liveRoute.Post("/channel-rules", routing.Wrap(hs.Live.HandleChannelRulesPostHTTP), reqOrgAdmin) liveRoute.Put("/channel-rules", routing.Wrap(hs.Live.HandleChannelRulesPutHTTP), reqOrgAdmin) liveRoute.Delete("/channel-rules", routing.Wrap(hs.Live.HandleChannelRulesDeleteHTTP), reqOrgAdmin) - liveRoute.Get("/remote-write-backends", routing.Wrap(hs.Live.HandleRemoteWriteBackendsListHTTP), reqOrgAdmin) - liveRoute.Post("/remote-write-backends", routing.Wrap(hs.Live.HandleRemoteWriteBackendsPostHTTP), reqOrgAdmin) - liveRoute.Put("/remote-write-backends", routing.Wrap(hs.Live.HandleRemoteWriteBackendsPutHTTP), reqOrgAdmin) - liveRoute.Delete("/remote-write-backends", routing.Wrap(hs.Live.HandleRemoteWriteBackendsDeleteHTTP), reqOrgAdmin) + liveRoute.Get("/write-configs", routing.Wrap(hs.Live.HandleWriteConfigsListHTTP), reqOrgAdmin) + liveRoute.Post("/write-configs", routing.Wrap(hs.Live.HandleWriteConfigsPostHTTP), reqOrgAdmin) + liveRoute.Put("/write-configs", routing.Wrap(hs.Live.HandleWriteConfigsPutHTTP), reqOrgAdmin) + liveRoute.Delete("/write-configs", routing.Wrap(hs.Live.HandleWriteConfigsDeleteHTTP), reqOrgAdmin) } }) diff --git a/pkg/services/live/live.go b/pkg/services/live/live.go index 09ed4139310..295cc5641af 100644 --- a/pkg/services/live/live.go +++ b/pkg/services/live/live.go @@ -1005,19 +1005,19 @@ type DryRunRuleStorage struct { ChannelRules []pipeline.ChannelRule } -func (s *DryRunRuleStorage) GetRemoteWriteBackend(_ context.Context, _ int64, _ pipeline.RemoteWriteBackendGetCmd) (pipeline.RemoteWriteBackend, bool, error) { - return pipeline.RemoteWriteBackend{}, false, errors.New("not implemented by dry run rule storage") +func (s *DryRunRuleStorage) GetWriteConfig(_ context.Context, _ int64, _ pipeline.WriteConfigGetCmd) (pipeline.WriteConfig, bool, error) { + return pipeline.WriteConfig{}, false, errors.New("not implemented by dry run rule storage") } -func (s *DryRunRuleStorage) CreateRemoteWriteBackend(_ context.Context, _ int64, _ pipeline.RemoteWriteBackendCreateCmd) (pipeline.RemoteWriteBackend, error) { - return pipeline.RemoteWriteBackend{}, errors.New("not implemented by dry run rule storage") +func (s *DryRunRuleStorage) CreateWriteConfig(_ context.Context, _ int64, _ pipeline.WriteConfigCreateCmd) (pipeline.WriteConfig, error) { + return pipeline.WriteConfig{}, errors.New("not implemented by dry run rule storage") } -func (s *DryRunRuleStorage) UpdateRemoteWriteBackend(_ context.Context, _ int64, _ pipeline.RemoteWriteBackendUpdateCmd) (pipeline.RemoteWriteBackend, error) { - return pipeline.RemoteWriteBackend{}, errors.New("not implemented by dry run rule storage") +func (s *DryRunRuleStorage) UpdateWriteConfig(_ context.Context, _ int64, _ pipeline.WriteConfigUpdateCmd) (pipeline.WriteConfig, error) { + return pipeline.WriteConfig{}, errors.New("not implemented by dry run rule storage") } -func (s *DryRunRuleStorage) DeleteRemoteWriteBackend(_ context.Context, _ int64, _ pipeline.RemoteWriteBackendDeleteCmd) error { +func (s *DryRunRuleStorage) DeleteWriteConfig(_ context.Context, _ int64, _ pipeline.WriteConfigDeleteCmd) error { return errors.New("not implemented by dry run rule storage") } @@ -1033,7 +1033,7 @@ func (s *DryRunRuleStorage) DeleteChannelRule(_ context.Context, _ int64, _ pipe return errors.New("not implemented by dry run rule storage") } -func (s *DryRunRuleStorage) ListRemoteWriteBackends(_ context.Context, _ int64) ([]pipeline.RemoteWriteBackend, error) { +func (s *DryRunRuleStorage) ListWriteConfigs(_ context.Context, _ int64) ([]pipeline.WriteConfig, error) { return nil, nil } @@ -1161,60 +1161,60 @@ func (g *GrafanaLive) HandlePipelineEntitiesListHTTP(_ *models.ReqContext) respo }) } -// HandleRemoteWriteBackendsListHTTP ... -func (g *GrafanaLive) HandleRemoteWriteBackendsListHTTP(c *models.ReqContext) response.Response { - backends, err := g.pipelineStorage.ListRemoteWriteBackends(c.Req.Context(), c.OrgId) +// HandleWriteConfigsListHTTP ... +func (g *GrafanaLive) HandleWriteConfigsListHTTP(c *models.ReqContext) response.Response { + backends, err := g.pipelineStorage.ListWriteConfigs(c.Req.Context(), c.OrgId) if err != nil { - return response.Error(http.StatusInternalServerError, "Failed to get remote write backends", err) + return response.Error(http.StatusInternalServerError, "Failed to get write configs", err) } - result := make([]pipeline.RemoteWriteBackendDto, 0, len(backends)) + result := make([]pipeline.WriteConfigDto, 0, len(backends)) for _, b := range backends { - result = append(result, pipeline.RemoteWriteBackendToDto(b)) + result = append(result, pipeline.WriteConfigToDto(b)) } return response.JSON(http.StatusOK, util.DynMap{ - "remoteWriteBackends": result, + "writeConfigs": result, }) } -// HandleChannelRulesPostHTTP ... -func (g *GrafanaLive) HandleRemoteWriteBackendsPostHTTP(c *models.ReqContext) response.Response { +// HandleWriteConfigsPostHTTP ... +func (g *GrafanaLive) HandleWriteConfigsPostHTTP(c *models.ReqContext) response.Response { body, err := ioutil.ReadAll(c.Req.Body) if err != nil { return response.Error(http.StatusInternalServerError, "Error reading body", err) } - var cmd pipeline.RemoteWriteBackendCreateCmd + var cmd pipeline.WriteConfigCreateCmd err = json.Unmarshal(body, &cmd) if err != nil { - return response.Error(http.StatusBadRequest, "Error decoding remote write backend", err) + return response.Error(http.StatusBadRequest, "Error decoding write config create command", err) } - result, err := g.pipelineStorage.CreateRemoteWriteBackend(c.Req.Context(), c.OrgId, cmd) + result, err := g.pipelineStorage.CreateWriteConfig(c.Req.Context(), c.OrgId, cmd) if err != nil { - return response.Error(http.StatusInternalServerError, "Failed to create remote write backend", err) + return response.Error(http.StatusInternalServerError, "Failed to create write config", err) } return response.JSON(http.StatusOK, util.DynMap{ - "remoteWriteBackend": pipeline.RemoteWriteBackendToDto(result), + "writeConfig": pipeline.WriteConfigToDto(result), }) } -// HandleChannelRulesPutHTTP ... -func (g *GrafanaLive) HandleRemoteWriteBackendsPutHTTP(c *models.ReqContext) response.Response { +// HandleWriteConfigsPutHTTP ... +func (g *GrafanaLive) HandleWriteConfigsPutHTTP(c *models.ReqContext) response.Response { body, err := ioutil.ReadAll(c.Req.Body) if err != nil { return response.Error(http.StatusInternalServerError, "Error reading body", err) } - var cmd pipeline.RemoteWriteBackendUpdateCmd + var cmd pipeline.WriteConfigUpdateCmd err = json.Unmarshal(body, &cmd) if err != nil { - return response.Error(http.StatusBadRequest, "Error decoding remote write backend", err) + return response.Error(http.StatusBadRequest, "Error decoding write config update command", err) } if cmd.UID == "" { return response.Error(http.StatusBadRequest, "UID required", nil) } - existingBackend, ok, err := g.pipelineStorage.GetRemoteWriteBackend(c.Req.Context(), c.OrgId, pipeline.RemoteWriteBackendGetCmd{ + existingBackend, ok, err := g.pipelineStorage.GetWriteConfig(c.Req.Context(), c.OrgId, pipeline.WriteConfigGetCmd{ UID: cmd.UID, }) if err != nil { - return response.Error(http.StatusInternalServerError, "Failed to get remote write backend", err) + return response.Error(http.StatusInternalServerError, "Failed to get write config", err) } if ok { if cmd.SecureSettings == nil { @@ -1231,32 +1231,32 @@ func (g *GrafanaLive) HandleRemoteWriteBackendsPutHTTP(c *models.ReqContext) res } } } - result, err := g.pipelineStorage.UpdateRemoteWriteBackend(c.Req.Context(), c.OrgId, cmd) + result, err := g.pipelineStorage.UpdateWriteConfig(c.Req.Context(), c.OrgId, cmd) if err != nil { - return response.Error(http.StatusInternalServerError, "Failed to update remote write backend", err) + return response.Error(http.StatusInternalServerError, "Failed to update write config", err) } return response.JSON(http.StatusOK, util.DynMap{ - "remoteWriteBackend": pipeline.RemoteWriteBackendToDto(result), + "writeConfig": pipeline.WriteConfigToDto(result), }) } -// HandleChannelRulesDeleteHTTP ... -func (g *GrafanaLive) HandleRemoteWriteBackendsDeleteHTTP(c *models.ReqContext) response.Response { +// HandleWriteConfigsDeleteHTTP ... +func (g *GrafanaLive) HandleWriteConfigsDeleteHTTP(c *models.ReqContext) response.Response { body, err := ioutil.ReadAll(c.Req.Body) if err != nil { return response.Error(http.StatusInternalServerError, "Error reading body", err) } - var cmd pipeline.RemoteWriteBackendDeleteCmd + var cmd pipeline.WriteConfigDeleteCmd err = json.Unmarshal(body, &cmd) if err != nil { - return response.Error(http.StatusBadRequest, "Error decoding remote write backend", err) + return response.Error(http.StatusBadRequest, "Error decoding write config delete command", err) } if cmd.UID == "" { return response.Error(http.StatusBadRequest, "UID required", nil) } - err = g.pipelineStorage.DeleteRemoteWriteBackend(c.Req.Context(), c.OrgId, cmd) + err = g.pipelineStorage.DeleteWriteConfig(c.Req.Context(), c.OrgId, cmd) if err != nil { - return response.Error(http.StatusInternalServerError, "Failed to delete remote write backend", err) + return response.Error(http.StatusInternalServerError, "Failed to delete write config", err) } return response.JSON(http.StatusOK, util.DynMap{}) } diff --git a/pkg/services/live/pipeline/config.go b/pkg/services/live/pipeline/config.go index aa9fac6a47e..fb30f5f3311 100644 --- a/pkg/services/live/pipeline/config.go +++ b/pkg/services/live/pipeline/config.go @@ -153,53 +153,53 @@ func typeRegistered(entityType string, registry []EntityInfo) bool { return false } -func RemoteWriteBackendToDto(b RemoteWriteBackend) RemoteWriteBackendDto { +func WriteConfigToDto(b WriteConfig) WriteConfigDto { secureFields := make(map[string]bool, len(b.SecureSettings)) for k := range b.SecureSettings { secureFields[k] = true } - return RemoteWriteBackendDto{ + return WriteConfigDto{ UID: b.UID, Settings: b.Settings, SecureFields: secureFields, } } -type RemoteWriteBackendDto struct { - UID string `json:"uid"` - Settings RemoteWriteSettings `json:"settings"` - SecureFields map[string]bool `json:"secureFields"` +type WriteConfigDto struct { + UID string `json:"uid"` + Settings WriteSettings `json:"settings"` + SecureFields map[string]bool `json:"secureFields"` } -type RemoteWriteBackendGetCmd struct { +type WriteConfigGetCmd struct { UID string `json:"uid"` } -type RemoteWriteBackendCreateCmd struct { - UID string `json:"uid"` - Settings RemoteWriteSettings `json:"settings"` - SecureSettings map[string]string `json:"secureSettings"` +type WriteConfigCreateCmd struct { + UID string `json:"uid"` + Settings WriteSettings `json:"settings"` + SecureSettings map[string]string `json:"secureSettings"` } // TODO: add version field later. -type RemoteWriteBackendUpdateCmd struct { - UID string `json:"uid"` - Settings RemoteWriteSettings `json:"settings"` - SecureSettings map[string]string `json:"secureSettings"` +type WriteConfigUpdateCmd struct { + UID string `json:"uid"` + Settings WriteSettings `json:"settings"` + SecureSettings map[string]string `json:"secureSettings"` } -type RemoteWriteBackendDeleteCmd struct { +type WriteConfigDeleteCmd struct { UID string `json:"uid"` } -type RemoteWriteBackend struct { - OrgId int64 `json:"-"` - UID string `json:"uid"` - Settings RemoteWriteSettings `json:"settings"` - SecureSettings map[string][]byte `json:"secureSettings,omitempty"` +type WriteConfig struct { + OrgId int64 `json:"-"` + UID string `json:"uid"` + Settings WriteSettings `json:"settings"` + SecureSettings map[string][]byte `json:"secureSettings,omitempty"` } -func (r RemoteWriteBackend) Valid() (bool, string) { +func (r WriteConfig) Valid() (bool, string) { if r.UID == "" { return false, "uid required" } @@ -209,9 +209,7 @@ func (r RemoteWriteBackend) Valid() (bool, string) { return true, "" } -type RemoteWriteSettings struct { - // Endpoint to send streaming frames to. - Endpoint string `json:"endpoint"` +type BasicAuth struct { // User is a user for remote write request. User string `json:"user,omitempty"` // Password is a plain text non-encrypted password. @@ -219,8 +217,15 @@ type RemoteWriteSettings struct { Password string `json:"password,omitempty"` } -type RemoteWriteBackends struct { - Backends []RemoteWriteBackend `json:"remoteWriteBackends"` +type WriteSettings struct { + // Endpoint to send streaming frames to. + Endpoint string `json:"endpoint"` + // BasicAuth is an optional basic auth settings. + BasicAuth *BasicAuth `json:"basicAuth,omitempty"` +} + +type WriteConfigs struct { + Configs []WriteConfig `json:"writeConfigs"` } type ChannelRules struct { @@ -276,11 +281,11 @@ type ChannelRuleDeleteCmd struct { } type Storage interface { - ListRemoteWriteBackends(_ context.Context, orgID int64) ([]RemoteWriteBackend, error) - GetRemoteWriteBackend(_ context.Context, orgID int64, cmd RemoteWriteBackendGetCmd) (RemoteWriteBackend, bool, error) - CreateRemoteWriteBackend(_ context.Context, orgID int64, cmd RemoteWriteBackendCreateCmd) (RemoteWriteBackend, error) - UpdateRemoteWriteBackend(_ context.Context, orgID int64, cmd RemoteWriteBackendUpdateCmd) (RemoteWriteBackend, error) - DeleteRemoteWriteBackend(_ context.Context, orgID int64, cmd RemoteWriteBackendDeleteCmd) error + ListWriteConfigs(_ context.Context, orgID int64) ([]WriteConfig, error) + GetWriteConfig(_ context.Context, orgID int64, cmd WriteConfigGetCmd) (WriteConfig, bool, error) + CreateWriteConfig(_ context.Context, orgID int64, cmd WriteConfigCreateCmd) (WriteConfig, error) + UpdateWriteConfig(_ context.Context, orgID int64, cmd WriteConfigUpdateCmd) (WriteConfig, error) + DeleteWriteConfig(_ context.Context, orgID int64, cmd WriteConfigDeleteCmd) error ListChannelRules(_ context.Context, orgID int64) ([]ChannelRule, error) CreateChannelRule(_ context.Context, orgID int64, cmd ChannelRuleCreateCmd) (ChannelRule, error) UpdateChannelRule(_ context.Context, orgID int64, cmd ChannelRuleUpdateCmd) (ChannelRule, error) @@ -422,23 +427,31 @@ func (f *StorageRuleBuilder) extractFrameConditionChecker(config *FrameCondition } } -func (f *StorageRuleBuilder) getWritePassword(remoteWriteBackend RemoteWriteBackend) (string, error) { +func (f *StorageRuleBuilder) constructBasicAuth(writeConfig WriteConfig) (*BasicAuth, error) { + if writeConfig.Settings.BasicAuth == nil { + return nil, nil + } var password string - hasSecurePassword := len(remoteWriteBackend.SecureSettings["password"]) > 0 + hasSecurePassword := len(writeConfig.SecureSettings["basicAuthPassword"]) > 0 if hasSecurePassword { - passwordBytes, err := f.EncryptionService.Decrypt(context.Background(), remoteWriteBackend.SecureSettings["password"], setting.SecretKey) + passwordBytes, err := f.EncryptionService.Decrypt(context.Background(), writeConfig.SecureSettings["basicAuthPassword"], setting.SecretKey) if err != nil { - return "", fmt.Errorf("password can't be decrypted: %w", err) + return nil, fmt.Errorf("basicAuthPassword can't be decrypted: %w", err) } password = string(passwordBytes) } else { // Use plain text password (should be removed upon database integration). - password = remoteWriteBackend.Settings.Password + if writeConfig.Settings.BasicAuth != nil { + password = writeConfig.Settings.BasicAuth.Password + } } - return password, nil + return &BasicAuth{ + User: writeConfig.Settings.BasicAuth.User, + Password: password, + }, nil } -func (f *StorageRuleBuilder) extractFrameOutputter(config *FrameOutputterConfig, remoteWriteBackends []RemoteWriteBackend) (FrameOutputter, error) { +func (f *StorageRuleBuilder) extractFrameOutputter(config *FrameOutputterConfig, writeConfigs []WriteConfig) (FrameOutputter, error) { if config == nil { return nil, nil } @@ -456,7 +469,7 @@ func (f *StorageRuleBuilder) extractFrameOutputter(config *FrameOutputterConfig, var outputters []FrameOutputter for _, outConf := range config.MultipleOutputterConfig.Outputters { out := outConf - outputter, err := f.extractFrameOutputter(&out, remoteWriteBackends) + outputter, err := f.extractFrameOutputter(&out, writeConfigs) if err != nil { return nil, err } @@ -475,7 +488,7 @@ func (f *StorageRuleBuilder) extractFrameOutputter(config *FrameOutputterConfig, if err != nil { return nil, err } - outputter, err := f.extractFrameOutputter(config.ConditionalOutputConfig.Outputter, remoteWriteBackends) + outputter, err := f.extractFrameOutputter(config.ConditionalOutputConfig.Outputter, writeConfigs) if err != nil { return nil, err } @@ -489,36 +502,34 @@ func (f *StorageRuleBuilder) extractFrameOutputter(config *FrameOutputterConfig, if config.RemoteWriteOutputConfig == nil { return nil, missingConfiguration } - remoteWriteBackend, ok := f.getRemoteWriteBackend(config.RemoteWriteOutputConfig.UID, remoteWriteBackends) + writeConfig, ok := f.getWriteConfig(config.RemoteWriteOutputConfig.UID, writeConfigs) if !ok { - return nil, fmt.Errorf("unknown remote write backend uid: %s", config.RemoteWriteOutputConfig.UID) + return nil, fmt.Errorf("unknown write config uid: %s", config.RemoteWriteOutputConfig.UID) } - password, err := f.getWritePassword(remoteWriteBackend) + basicAuth, err := f.constructBasicAuth(writeConfig) if err != nil { return nil, fmt.Errorf("error getting password: %w", err) } return NewRemoteWriteFrameOutput( - remoteWriteBackend.Settings.Endpoint, - remoteWriteBackend.Settings.User, - password, + writeConfig.Settings.Endpoint, + basicAuth, config.RemoteWriteOutputConfig.SampleMilliseconds, ), nil case FrameOutputTypeLoki: if config.LokiOutputConfig == nil { return nil, missingConfiguration } - remoteWriteBackend, ok := f.getRemoteWriteBackend(config.LokiOutputConfig.UID, remoteWriteBackends) + writeConfig, ok := f.getWriteConfig(config.LokiOutputConfig.UID, writeConfigs) if !ok { return nil, fmt.Errorf("unknown loki backend uid: %s", config.LokiOutputConfig.UID) } - password, err := f.getWritePassword(remoteWriteBackend) + basicAuth, err := f.constructBasicAuth(writeConfig) if err != nil { return nil, fmt.Errorf("error getting password: %w", err) } return NewLokiFrameOutput( - remoteWriteBackend.Settings.Endpoint, - remoteWriteBackend.Settings.User, - password, + writeConfig.Settings.Endpoint, + basicAuth, ), nil case FrameOutputTypeChangeLog: if config.ChangeLogOutputConfig == nil { @@ -530,7 +541,7 @@ func (f *StorageRuleBuilder) extractFrameOutputter(config *FrameOutputterConfig, } } -func (f *StorageRuleBuilder) extractDataOutputter(config *DataOutputterConfig, remoteWriteBackends []RemoteWriteBackend) (DataOutputter, error) { +func (f *StorageRuleBuilder) extractDataOutputter(config *DataOutputterConfig, writeConfigs []WriteConfig) (DataOutputter, error) { if config == nil { return nil, nil } @@ -545,18 +556,17 @@ func (f *StorageRuleBuilder) extractDataOutputter(config *DataOutputterConfig, r if config.LokiOutputConfig == nil { return nil, missingConfiguration } - remoteWriteBackend, ok := f.getRemoteWriteBackend(config.LokiOutputConfig.UID, remoteWriteBackends) + writeConfig, ok := f.getWriteConfig(config.LokiOutputConfig.UID, writeConfigs) if !ok { return nil, fmt.Errorf("unknown loki backend uid: %s", config.LokiOutputConfig.UID) } - password, err := f.getWritePassword(remoteWriteBackend) + basicAuth, err := f.constructBasicAuth(writeConfig) if err != nil { - return nil, fmt.Errorf("error getting password: %w", err) + return nil, fmt.Errorf("error constructing basicAuth: %w", err) } return NewLokiDataOutput( - remoteWriteBackend.Settings.Endpoint, - remoteWriteBackend.Settings.User, - password, + writeConfig.Settings.Endpoint, + basicAuth, ), nil case DataOutputTypeBuiltin: return NewBuiltinDataOutput(f.ChannelHandlerGetter), nil @@ -567,13 +577,13 @@ func (f *StorageRuleBuilder) extractDataOutputter(config *DataOutputterConfig, r } } -func (f *StorageRuleBuilder) getRemoteWriteBackend(uid string, remoteWriteBackends []RemoteWriteBackend) (RemoteWriteBackend, bool) { - for _, rwb := range remoteWriteBackends { +func (f *StorageRuleBuilder) getWriteConfig(uid string, writeConfigs []WriteConfig) (WriteConfig, bool) { + for _, rwb := range writeConfigs { if rwb.UID == uid { return rwb, true } } - return RemoteWriteBackend{}, false + return WriteConfig{}, false } func (f *StorageRuleBuilder) BuildRules(ctx context.Context, orgID int64) ([]*LiveChannelRule, error) { @@ -582,7 +592,7 @@ func (f *StorageRuleBuilder) BuildRules(ctx context.Context, orgID int64) ([]*Li return nil, err } - remoteWriteBackends, err := f.Storage.ListRemoteWriteBackends(ctx, orgID) + writeConfigs, err := f.Storage.ListWriteConfigs(ctx, orgID) if err != nil { return nil, err } @@ -622,7 +632,7 @@ func (f *StorageRuleBuilder) BuildRules(ctx context.Context, orgID int64) ([]*Li var dataOutputters []DataOutputter for _, outConfig := range ruleConfig.Settings.DataOutputters { - out, err := f.extractDataOutputter(outConfig, remoteWriteBackends) + out, err := f.extractDataOutputter(outConfig, writeConfigs) if err != nil { return nil, fmt.Errorf("error building data outputter for %s: %w", rule.Pattern, err) } @@ -632,7 +642,7 @@ func (f *StorageRuleBuilder) BuildRules(ctx context.Context, orgID int64) ([]*Li var outputters []FrameOutputter for _, outConfig := range ruleConfig.Settings.FrameOutputters { - out, err := f.extractFrameOutputter(outConfig, remoteWriteBackends) + out, err := f.extractFrameOutputter(outConfig, writeConfigs) if err != nil { return nil, fmt.Errorf("error building frame outputter for %s: %w", rule.Pattern, err) } diff --git a/pkg/services/live/pipeline/data_output_loki.go b/pkg/services/live/pipeline/data_output_loki.go index 22b2cd0b4ec..79e732ae2d1 100644 --- a/pkg/services/live/pipeline/data_output_loki.go +++ b/pkg/services/live/pipeline/data_output_loki.go @@ -10,9 +10,9 @@ type LokiDataOutput struct { lokiWriter *lokiWriter } -func NewLokiDataOutput(endpoint, user, password string) *LokiDataOutput { +func NewLokiDataOutput(endpoint string, basicAuth *BasicAuth) *LokiDataOutput { return &LokiDataOutput{ - lokiWriter: newLokiWriter(endpoint, user, password), + lokiWriter: newLokiWriter(endpoint, basicAuth), } } diff --git a/pkg/services/live/pipeline/devdata.go b/pkg/services/live/pipeline/devdata.go index af893178d1a..b5a9d3ce925 100644 --- a/pkg/services/live/pipeline/devdata.go +++ b/pkg/services/live/pipeline/devdata.go @@ -103,8 +103,10 @@ func (f *DevRuleBuilder) BuildRules(_ context.Context, _ int64) ([]*LiveChannelR DataOutputters: []DataOutputter{ NewLokiDataOutput( os.Getenv("GF_LIVE_LOKI_ENDPOINT"), - os.Getenv("GF_LIVE_LOKI_USER"), - os.Getenv("GF_LIVE_LOKI_PASSWORD"), + &BasicAuth{ + User: os.Getenv("GF_LIVE_LOKI_USER"), + Password: os.Getenv("GF_LIVE_LOKI_PASSWORD"), + }, ), }, Converter: NewJsonFrameConverter(JsonFrameConverterConfig{}), @@ -112,8 +114,10 @@ func (f *DevRuleBuilder) BuildRules(_ context.Context, _ int64) ([]*LiveChannelR NewManagedStreamFrameOutput(f.ManagedStream), NewRemoteWriteFrameOutput( os.Getenv("GF_LIVE_REMOTE_WRITE_ENDPOINT"), - os.Getenv("GF_LIVE_REMOTE_WRITE_USER"), - os.Getenv("GF_LIVE_REMOTE_WRITE_PASSWORD"), + &BasicAuth{ + User: os.Getenv("GF_LIVE_REMOTE_WRITE_USER"), + Password: os.Getenv("GF_LIVE_REMOTE_WRITE_PASSWORD"), + }, 1000, ), }, @@ -298,8 +302,10 @@ func (f *DevRuleBuilder) BuildRules(_ context.Context, _ int64) ([]*LiveChannelR NewManagedStreamFrameOutput(f.ManagedStream), NewRemoteWriteFrameOutput( os.Getenv("GF_LIVE_REMOTE_WRITE_ENDPOINT"), - os.Getenv("GF_LIVE_REMOTE_WRITE_USER"), - os.Getenv("GF_LIVE_REMOTE_WRITE_PASSWORD"), + &BasicAuth{ + User: os.Getenv("GF_LIVE_REMOTE_WRITE_USER"), + Password: os.Getenv("GF_LIVE_REMOTE_WRITE_PASSWORD"), + }, 0, ), NewChangeLogFrameOutput(f.FrameStorage, ChangeLogOutputConfig{ @@ -333,8 +339,10 @@ func (f *DevRuleBuilder) BuildRules(_ context.Context, _ int64) ([]*LiveChannelR NewManagedStreamFrameOutput(f.ManagedStream), NewRemoteWriteFrameOutput( os.Getenv("GF_LIVE_REMOTE_WRITE_ENDPOINT"), - os.Getenv("GF_LIVE_REMOTE_WRITE_USER"), - os.Getenv("GF_LIVE_REMOTE_WRITE_PASSWORD"), + &BasicAuth{ + User: os.Getenv("GF_LIVE_REMOTE_WRITE_USER"), + Password: os.Getenv("GF_LIVE_REMOTE_WRITE_PASSWORD"), + }, 0, ), }, diff --git a/pkg/services/live/pipeline/frame_output_loki.go b/pkg/services/live/pipeline/frame_output_loki.go index f16e20cebf0..f4f01b73db1 100644 --- a/pkg/services/live/pipeline/frame_output_loki.go +++ b/pkg/services/live/pipeline/frame_output_loki.go @@ -20,9 +20,9 @@ type LokiFrameOutput struct { lokiWriter *lokiWriter } -func NewLokiFrameOutput(endpoint, user, password string) *LokiFrameOutput { +func NewLokiFrameOutput(endpoint string, basicAuth *BasicAuth) *LokiFrameOutput { return &LokiFrameOutput{ - lokiWriter: newLokiWriter(endpoint, user, password), + lokiWriter: newLokiWriter(endpoint, basicAuth), } } @@ -65,18 +65,14 @@ type lokiWriter struct { buffer []LokiStream // Endpoint to send streaming frames to. - endpoint string - // User is a user for remote write request. - user string - // Password for remote write endpoint. - password string + endpoint string + basicAuth *BasicAuth } -func newLokiWriter(endpoint, user, password string) *lokiWriter { +func newLokiWriter(endpoint string, basicAuth *BasicAuth) *lokiWriter { w := &lokiWriter{ - endpoint: endpoint, - user: user, - password: password, + endpoint: endpoint, + basicAuth: basicAuth, httpClient: &http.Client{ Timeout: 2 * time.Second, }, @@ -129,8 +125,8 @@ func (w *lokiWriter) flush(streams []LokiStream) error { return fmt.Errorf("error constructing loki push request: %w", err) } req.Header.Set("Content-Type", "application/json") - if w.user != "" { - req.SetBasicAuth(w.user, w.password) + if w.basicAuth != nil { + req.SetBasicAuth(w.basicAuth.User, w.basicAuth.Password) } started := time.Now() diff --git a/pkg/services/live/pipeline/frame_output_remote_write.go b/pkg/services/live/pipeline/frame_output_remote_write.go index fbd9f26334d..cb4ff658016 100644 --- a/pkg/services/live/pipeline/frame_output_remote_write.go +++ b/pkg/services/live/pipeline/frame_output_remote_write.go @@ -21,10 +21,10 @@ type RemoteWriteFrameOutput struct { // Endpoint to send streaming frames to. Endpoint string - // User is a user for remote write request. - User string - // Password for remote write endpoint. - Password string + + // BasicAuth is an optional basic auth params. + BasicAuth *BasicAuth + // SampleMilliseconds allow defining an interval to sample points inside a channel // when outputting to remote write endpoint (on __name__ label basis). For example // when having a 20Hz stream and SampleMilliseconds 1000 then only one point in a @@ -38,11 +38,10 @@ type RemoteWriteFrameOutput struct { buffer []prompb.TimeSeries } -func NewRemoteWriteFrameOutput(endpoint, user, password string, sampleMilliseconds int64) *RemoteWriteFrameOutput { +func NewRemoteWriteFrameOutput(endpoint string, basicAuth *BasicAuth, sampleMilliseconds int64) *RemoteWriteFrameOutput { out := &RemoteWriteFrameOutput{ Endpoint: endpoint, - User: user, - Password: password, + BasicAuth: basicAuth, SampleMilliseconds: sampleMilliseconds, httpClient: &http.Client{Timeout: 2 * time.Second}, } @@ -153,8 +152,8 @@ func (out *RemoteWriteFrameOutput) flush(timeSeries []prompb.TimeSeries) error { req.Header.Set("Content-Type", "application/x-protobuf") req.Header.Set("Content-Encoding", "snappy") req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0") - if out.User != "" { - req.SetBasicAuth(out.User, out.Password) + if out.BasicAuth != nil { + req.SetBasicAuth(out.BasicAuth.User, out.BasicAuth.Password) } started := time.Now() diff --git a/pkg/services/live/pipeline/frame_output_remote_write_test.go b/pkg/services/live/pipeline/frame_output_remote_write_test.go index fa282b62dd9..3ef8acc4172 100644 --- a/pkg/services/live/pipeline/frame_output_remote_write_test.go +++ b/pkg/services/live/pipeline/frame_output_remote_write_test.go @@ -51,7 +51,7 @@ func TestRemoteWriteFrameOutput_sample(t *testing.T) { }, }, } - out := NewRemoteWriteFrameOutput("", "", "", 500) + out := NewRemoteWriteFrameOutput("", nil, 500) sampledTimeSeries := out.sample(timeSeries) require.Len(t, sampledTimeSeries, 2) @@ -123,7 +123,7 @@ func TestRemoteWriteFrameOutput_sample_merge(t *testing.T) { }, }, } - out := NewRemoteWriteFrameOutput("", "", "", 50) + out := NewRemoteWriteFrameOutput("", nil, 50) sampledTimeSeries := out.sample(timeSeries) require.Len(t, sampledTimeSeries, 2) diff --git a/pkg/services/live/pipeline/storage_file.go b/pkg/services/live/pipeline/storage_file.go index f461a086b5f..d20c546b584 100644 --- a/pkg/services/live/pipeline/storage_file.go +++ b/pkg/services/live/pipeline/storage_file.go @@ -20,45 +20,37 @@ type FileStorage struct { EncryptionService encryption.Service } -func (f *FileStorage) ListRemoteWriteBackends(_ context.Context, orgID int64) ([]RemoteWriteBackend, error) { - cfgfile := filepath.Join(f.DataPath, "pipeline", "remote-write-backends.json") - var backends []RemoteWriteBackend - // Safe to ignore gosec warning G304. - // nolint:gosec - backendBytes, err := ioutil.ReadFile(cfgfile) +func (f *FileStorage) ListWriteConfigs(_ context.Context, orgID int64) ([]WriteConfig, error) { + writeConfigs, err := f.readWriteConfigs() if err != nil { - return backends, fmt.Errorf("can't read %s file: %w", cfgfile, err) + return nil, fmt.Errorf("can't read write configs: %w", err) } - var remoteWriteBackends RemoteWriteBackends - err = json.Unmarshal(backendBytes, &remoteWriteBackends) - if err != nil { - return nil, fmt.Errorf("can't unmarshal remote-write-backends.json data: %w", err) - } - for _, b := range remoteWriteBackends.Backends { + var orgConfigs []WriteConfig + for _, b := range writeConfigs.Configs { if b.OrgId == orgID || (orgID == 1 && b.OrgId == 0) { - backends = append(backends, b) + orgConfigs = append(orgConfigs, b) } } - return backends, nil + return orgConfigs, nil } -func (f *FileStorage) GetRemoteWriteBackend(_ context.Context, orgID int64, cmd RemoteWriteBackendGetCmd) (RemoteWriteBackend, bool, error) { - remoteWriteBackends, err := f.readRemoteWriteBackends() +func (f *FileStorage) GetWriteConfig(_ context.Context, orgID int64, cmd WriteConfigGetCmd) (WriteConfig, bool, error) { + writeConfigs, err := f.readWriteConfigs() if err != nil { - return RemoteWriteBackend{}, false, fmt.Errorf("can't remote write backends: %w", err) + return WriteConfig{}, false, fmt.Errorf("can't read write configs: %w", err) } - for _, existingBackend := range remoteWriteBackends.Backends { + for _, existingBackend := range writeConfigs.Configs { if uidMatch(orgID, cmd.UID, existingBackend) { return existingBackend, true, nil } } - return RemoteWriteBackend{}, false, nil + return WriteConfig{}, false, nil } -func (f *FileStorage) CreateRemoteWriteBackend(ctx context.Context, orgID int64, cmd RemoteWriteBackendCreateCmd) (RemoteWriteBackend, error) { - remoteWriteBackends, err := f.readRemoteWriteBackends() +func (f *FileStorage) CreateWriteConfig(ctx context.Context, orgID int64, cmd WriteConfigCreateCmd) (WriteConfig, error) { + writeConfigs, err := f.readWriteConfigs() if err != nil { - return RemoteWriteBackend{}, fmt.Errorf("can't read remote write backends: %w", err) + return WriteConfig{}, fmt.Errorf("can't read write configs: %w", err) } if cmd.UID == "" { cmd.UID = util.GenerateShortUID() @@ -66,10 +58,10 @@ func (f *FileStorage) CreateRemoteWriteBackend(ctx context.Context, orgID int64, secureSettings, err := f.EncryptionService.EncryptJsonData(ctx, cmd.SecureSettings, setting.SecretKey) if err != nil { - return RemoteWriteBackend{}, fmt.Errorf("error encrypting data: %w", err) + return WriteConfig{}, fmt.Errorf("error encrypting data: %w", err) } - backend := RemoteWriteBackend{ + backend := WriteConfig{ OrgId: orgID, UID: cmd.UID, Settings: cmd.Settings, @@ -78,30 +70,30 @@ func (f *FileStorage) CreateRemoteWriteBackend(ctx context.Context, orgID int64, ok, reason := backend.Valid() if !ok { - return RemoteWriteBackend{}, fmt.Errorf("invalid remote write backend: %s", reason) + return WriteConfig{}, fmt.Errorf("invalid write config: %s", reason) } - for _, existingBackend := range remoteWriteBackends.Backends { + for _, existingBackend := range writeConfigs.Configs { if uidMatch(orgID, backend.UID, existingBackend) { - return RemoteWriteBackend{}, fmt.Errorf("backend already exists in org: %s", backend.UID) + return WriteConfig{}, fmt.Errorf("backend already exists in org: %s", backend.UID) } } - remoteWriteBackends.Backends = append(remoteWriteBackends.Backends, backend) - err = f.saveRemoteWriteBackends(orgID, remoteWriteBackends) + writeConfigs.Configs = append(writeConfigs.Configs, backend) + err = f.saveWriteConfigs(orgID, writeConfigs) return backend, err } -func (f *FileStorage) UpdateRemoteWriteBackend(ctx context.Context, orgID int64, cmd RemoteWriteBackendUpdateCmd) (RemoteWriteBackend, error) { - remoteWriteBackends, err := f.readRemoteWriteBackends() +func (f *FileStorage) UpdateWriteConfig(ctx context.Context, orgID int64, cmd WriteConfigUpdateCmd) (WriteConfig, error) { + writeConfigs, err := f.readWriteConfigs() if err != nil { - return RemoteWriteBackend{}, fmt.Errorf("can't read remote write backends: %w", err) + return WriteConfig{}, fmt.Errorf("can't read write configs: %w", err) } secureSettings, err := f.EncryptionService.EncryptJsonData(ctx, cmd.SecureSettings, setting.SecretKey) if err != nil { - return RemoteWriteBackend{}, fmt.Errorf("error encrypting data: %w", err) + return WriteConfig{}, fmt.Errorf("error encrypting data: %w", err) } - backend := RemoteWriteBackend{ + backend := WriteConfig{ OrgId: orgID, UID: cmd.UID, Settings: cmd.Settings, @@ -110,35 +102,35 @@ func (f *FileStorage) UpdateRemoteWriteBackend(ctx context.Context, orgID int64, ok, reason := backend.Valid() if !ok { - return RemoteWriteBackend{}, fmt.Errorf("invalid channel rule: %s", reason) + return WriteConfig{}, fmt.Errorf("invalid channel rule: %s", reason) } index := -1 - for i, existingBackend := range remoteWriteBackends.Backends { + for i, existingBackend := range writeConfigs.Configs { if uidMatch(orgID, backend.UID, existingBackend) { index = i break } } if index > -1 { - remoteWriteBackends.Backends[index] = backend + writeConfigs.Configs[index] = backend } else { - return f.CreateRemoteWriteBackend(ctx, orgID, RemoteWriteBackendCreateCmd(cmd)) + return f.CreateWriteConfig(ctx, orgID, WriteConfigCreateCmd(cmd)) } - err = f.saveRemoteWriteBackends(orgID, remoteWriteBackends) + err = f.saveWriteConfigs(orgID, writeConfigs) return backend, err } -func (f *FileStorage) DeleteRemoteWriteBackend(_ context.Context, orgID int64, cmd RemoteWriteBackendDeleteCmd) error { - remoteWriteBackends, err := f.readRemoteWriteBackends() +func (f *FileStorage) DeleteWriteConfig(_ context.Context, orgID int64, cmd WriteConfigDeleteCmd) error { + writeConfigs, err := f.readWriteConfigs() if err != nil { - return fmt.Errorf("can't read remote write backends: %w", err) + return fmt.Errorf("can't read write configs: %w", err) } index := -1 - for i, existingBackend := range remoteWriteBackends.Backends { + for i, existingBackend := range writeConfigs.Configs { if uidMatch(orgID, cmd.UID, existingBackend) { index = i break @@ -146,12 +138,12 @@ func (f *FileStorage) DeleteRemoteWriteBackend(_ context.Context, orgID int64, c } if index > -1 { - remoteWriteBackends.Backends = removeRemoteWriteBackendByIndex(remoteWriteBackends.Backends, index) + writeConfigs.Configs = removeWriteConfigByIndex(writeConfigs.Configs, index) } else { - return fmt.Errorf("remote write backend not found") + return fmt.Errorf("write config not found") } - return f.saveRemoteWriteBackends(orgID, remoteWriteBackends) + return f.saveWriteConfigs(orgID, writeConfigs) } func (f *FileStorage) ListChannelRules(_ context.Context, orgID int64) ([]ChannelRule, error) { @@ -198,7 +190,7 @@ func patternMatch(orgID int64, pattern string, existingRule ChannelRule) bool { return pattern == existingRule.Pattern && (existingRule.OrgId == orgID || (existingRule.OrgId == 0 && orgID == 1)) } -func uidMatch(orgID int64, uid string, existingBackend RemoteWriteBackend) bool { +func uidMatch(orgID int64, uid string, existingBackend WriteConfig) bool { return uid == existingBackend.UID && (existingBackend.OrgId == orgID || (existingBackend.OrgId == 0 && orgID == 1)) } @@ -306,44 +298,44 @@ func (f *FileStorage) DeleteChannelRule(_ context.Context, orgID int64, cmd Chan return f.saveChannelRules(orgID, channelRules) } -func removeRemoteWriteBackendByIndex(s []RemoteWriteBackend, index int) []RemoteWriteBackend { +func removeWriteConfigByIndex(s []WriteConfig, index int) []WriteConfig { return append(s[:index], s[index+1:]...) } -func (f *FileStorage) remoteWriteFilePath() string { - return filepath.Join(f.DataPath, "pipeline", "remote-write-backends.json") +func (f *FileStorage) writeConfigsFilePath() string { + return filepath.Join(f.DataPath, "pipeline", "write-configs.json") } -func (f *FileStorage) readRemoteWriteBackends() (RemoteWriteBackends, error) { - filePath := f.remoteWriteFilePath() +func (f *FileStorage) readWriteConfigs() (WriteConfigs, error) { + filePath := f.writeConfigsFilePath() // Safe to ignore gosec warning G304. // nolint:gosec bytes, err := ioutil.ReadFile(filePath) if err != nil { - return RemoteWriteBackends{}, fmt.Errorf("can't read %s file: %w", filePath, err) + return WriteConfigs{}, fmt.Errorf("can't read %s file: %w", filePath, err) } - var remoteWriteBackends RemoteWriteBackends - err = json.Unmarshal(bytes, &remoteWriteBackends) + var writeConfigs WriteConfigs + err = json.Unmarshal(bytes, &writeConfigs) if err != nil { - return RemoteWriteBackends{}, fmt.Errorf("can't unmarshal %s data: %w", filePath, err) + return WriteConfigs{}, fmt.Errorf("can't unmarshal %s data: %w", filePath, err) } - return remoteWriteBackends, nil + return writeConfigs, nil } -func (f *FileStorage) saveRemoteWriteBackends(_ int64, remoteWriteBackends RemoteWriteBackends) error { - filePath := f.remoteWriteFilePath() +func (f *FileStorage) saveWriteConfigs(_ int64, writeConfigs WriteConfigs) error { + filePath := f.writeConfigsFilePath() // Safe to ignore gosec warning G304. // nolint:gosec file, err := os.OpenFile(filePath, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0600) if err != nil { - return fmt.Errorf("can't open channel remote write backends file: %w", err) + return fmt.Errorf("can't open channel write configs file: %w", err) } defer func() { _ = file.Close() }() enc := json.NewEncoder(file) enc.SetIndent("", " ") - err = enc.Encode(remoteWriteBackends) + err = enc.Encode(writeConfigs) if err != nil { - return fmt.Errorf("can't save remote write backends to file: %w", err) + return fmt.Errorf("can't save write configs to file: %w", err) } return nil } diff --git a/public/app/features/live/pages/CloudAdminPage.tsx b/public/app/features/live/pages/CloudAdminPage.tsx index 70699cd3229..b24dba5e6c4 100644 --- a/public/app/features/live/pages/CloudAdminPage.tsx +++ b/public/app/features/live/pages/CloudAdminPage.tsx @@ -15,9 +15,9 @@ export default function CloudAdminPage() { useEffect(() => { getBackendSrv() - .get(`api/live/remote-write-backends`) + .get(`api/live/write-configs`) .then((data) => { - setCloud(data.remoteWriteBackends); + setCloud(data.writeConfigs); }) .catch((e) => { if (e.data) {