diff --git a/pkg/api/api.go b/pkg/api/api.go index 6a2ceeb76e6..42dd5f85428 100644 --- a/pkg/api/api.go +++ b/pkg/api/api.go @@ -438,6 +438,7 @@ func (hs *HTTPServer) registerRoutes() { // POST Live data to be processed according to channel rules. liveRoute.Post("/push/:streamId/:path", hs.LivePushGateway.HandlePath) liveRoute.Get("/channel-rules", routing.Wrap(hs.Live.HandleChannelRulesListHTTP), reqOrgAdmin) + liveRoute.Get("/pipeline-entities", routing.Wrap(hs.Live.HandlePipelineEntitiesListHTTP), reqOrgAdmin) liveRoute.Get("/remote-write-backends", routing.Wrap(hs.Live.HandleRemoteWriteBackendsListHTTP), reqOrgAdmin) } }) diff --git a/pkg/services/live/live.go b/pkg/services/live/live.go index 5dc17b394b1..5f61a0c8f2e 100644 --- a/pkg/services/live/live.go +++ b/pkg/services/live/live.go @@ -868,6 +868,94 @@ func (g *GrafanaLive) HandleChannelRulesListHTTP(c *models.ReqContext) response. }) } +type configInfo struct { + Type string `json:"type"` + Description string `json:"description"` + Example interface{} `json:"example,omitempty"` +} + +// HandlePipelineEntitiesListHTTP ... +func (g *GrafanaLive) HandlePipelineEntitiesListHTTP(_ *models.ReqContext) response.Response { + return response.JSON(http.StatusOK, util.DynMap{ + "subscribers": []configInfo{ + { + Type: pipeline.SubscriberTypeBuiltin, + Description: "list the fields that should be removed", + // Example: pipeline.Bu{}, + }, + { + Type: pipeline.SubscriberTypeManagedStream, + Description: "list the fields that should be removed", + }, + { + Type: pipeline.SubscriberTypeMultiple, + }, + }, + "outputs": []configInfo{ + { + Type: pipeline.OutputTypeManagedStream, + Description: "Only send schema when structure changes. Note this also requires a matching subscriber", + Example: pipeline.ManagedStreamOutputConfig{}, + }, + { + Type: pipeline.OutputTypeMultiple, + Description: "Send the output to multiple destinations", + Example: pipeline.MultipleOutputterConfig{}, + }, + { + Type: pipeline.OutputTypeConditional, + Description: "send to an output depending on frame values", + Example: pipeline.ConditionalOutputConfig{}, + }, + { + Type: pipeline.OutputTypeRedirect, + }, + { + Type: pipeline.OutputTypeThreshold, + }, + { + Type: pipeline.OutputTypeChangeLog, + }, + { + Type: pipeline.OutputTypeRemoteWrite, + }, + }, + "converters": []configInfo{ + { + Type: pipeline.ConverterTypeJsonAuto, + }, + { + Type: pipeline.ConverterTypeJsonExact, + }, + { + Type: pipeline.ConverterTypeInfluxAuto, + Description: "accept influx line protocol", + Example: pipeline.AutoInfluxConverterConfig{}, + }, + { + Type: pipeline.ConverterTypeJsonFrame, + }, + }, + "processors": []configInfo{ + { + Type: pipeline.ProcessorTypeKeepFields, + Description: "list the fields that should stay", + Example: pipeline.KeepFieldsProcessorConfig{}, + }, + { + Type: pipeline.ProcessorTypeDropFields, + Description: "list the fields that should be removed", + Example: pipeline.DropFieldsProcessorConfig{}, + }, + { + Type: pipeline.ProcessorTypeMultiple, + Description: "apply multiplie processors", + Example: pipeline.MultipleProcessorConfig{}, + }, + }, + }) +} + // HandleRemoteWriteBackendsListHTTP ... func (g *GrafanaLive) HandleRemoteWriteBackendsListHTTP(c *models.ReqContext) response.Response { result, err := g.channelRuleStorage.ListRemoteWriteBackends(c.Req.Context(), c.OrgId) diff --git a/pkg/services/live/pipeline/condition_checker.go b/pkg/services/live/pipeline/condition_checker.go index 46d4607612d..a50fd67ec0d 100644 --- a/pkg/services/live/pipeline/condition_checker.go +++ b/pkg/services/live/pipeline/condition_checker.go @@ -8,5 +8,6 @@ import ( // ConditionChecker checks conditions in context of data.Frame being processed. type ConditionChecker interface { + Type() string CheckCondition(ctx context.Context, frame *data.Frame) (bool, error) } diff --git a/pkg/services/live/pipeline/condition_checker_multiple.go b/pkg/services/live/pipeline/condition_checker_multiple.go index dee496331df..83e7c9ba069 100644 --- a/pkg/services/live/pipeline/condition_checker_multiple.go +++ b/pkg/services/live/pipeline/condition_checker_multiple.go @@ -16,24 +16,30 @@ const ( // MultipleConditionChecker can check multiple conditions according to ConditionType. type MultipleConditionChecker struct { - Type ConditionType - Conditions []ConditionChecker + ConditionType ConditionType + Conditions []ConditionChecker } -func (m MultipleConditionChecker) CheckCondition(ctx context.Context, frame *data.Frame) (bool, error) { - for _, c := range m.Conditions { - ok, err := c.CheckCondition(ctx, frame) +const ConditionCheckerTypeMultiple = "multiple" + +func (c *MultipleConditionChecker) Type() string { + return ConditionCheckerTypeMultiple +} + +func (c *MultipleConditionChecker) CheckCondition(ctx context.Context, frame *data.Frame) (bool, error) { + for _, cond := range c.Conditions { + ok, err := cond.CheckCondition(ctx, frame) if err != nil { return false, err } - if ok && m.Type == ConditionAny { + if ok && c.ConditionType == ConditionAny { return true, nil } - if !ok && m.Type == ConditionAll { + if !ok && c.ConditionType == ConditionAll { return false, nil } } - if m.Type == ConditionAny { + if c.ConditionType == ConditionAny { return false, nil } return true, nil @@ -41,5 +47,5 @@ func (m MultipleConditionChecker) CheckCondition(ctx context.Context, frame *dat // NewMultipleConditionChecker creates new MultipleConditionChecker. func NewMultipleConditionChecker(conditionType ConditionType, conditions ...ConditionChecker) *MultipleConditionChecker { - return &MultipleConditionChecker{Type: conditionType, Conditions: conditions} + return &MultipleConditionChecker{ConditionType: conditionType, Conditions: conditions} } diff --git a/pkg/services/live/pipeline/condition_number_compare.go b/pkg/services/live/pipeline/condition_number_compare.go index 429415c5b30..afcefe03c2a 100644 --- a/pkg/services/live/pipeline/condition_number_compare.go +++ b/pkg/services/live/pipeline/condition_number_compare.go @@ -27,10 +27,16 @@ const ( NumberCompareOpNe NumberCompareOp = "ne" ) -func (f NumberCompareCondition) CheckCondition(_ context.Context, frame *data.Frame) (bool, error) { +const ConditionCheckerTypeNumberCompare = "numberCompare" + +func (c *NumberCompareCondition) Type() string { + return ConditionCheckerTypeNumberCompare +} + +func (c *NumberCompareCondition) CheckCondition(_ context.Context, frame *data.Frame) (bool, error) { for _, field := range frame.Fields { // TODO: support other numeric types. - if field.Name == f.FieldName && (field.Type() == data.FieldTypeNullableFloat64) { + if field.Name == c.FieldName && (field.Type() == data.FieldTypeNullableFloat64) { value, ok := field.At(0).(*float64) if !ok { return false, fmt.Errorf("unexpected value type: %T", field.At(0)) @@ -38,21 +44,21 @@ func (f NumberCompareCondition) CheckCondition(_ context.Context, frame *data.Fr if value == nil { return false, nil } - switch f.Op { + switch c.Op { case NumberCompareOpGt: - return *value > f.Value, nil + return *value > c.Value, nil case NumberCompareOpGte: - return *value >= f.Value, nil + return *value >= c.Value, nil case NumberCompareOpLte: - return *value <= f.Value, nil + return *value <= c.Value, nil case NumberCompareOpLt: - return *value < f.Value, nil + return *value < c.Value, nil case NumberCompareOpEq: - return *value == f.Value, nil + return *value == c.Value, nil case NumberCompareOpNe: - return *value != f.Value, nil + return *value != c.Value, nil default: - return false, fmt.Errorf("unknown comparison operator: %s", f.Op) + return false, fmt.Errorf("unknown comparison operator: %s", c.Op) } } } diff --git a/pkg/services/live/pipeline/config.go b/pkg/services/live/pipeline/config.go index 61ec3373f9d..01021a674e1 100644 --- a/pkg/services/live/pipeline/config.go +++ b/pkg/services/live/pipeline/config.go @@ -128,11 +128,11 @@ func (f *StorageRuleBuilder) extractSubscriber(config *SubscriberConfig) (Subscr } missingConfiguration := fmt.Errorf("missing configuration for %s", config.Type) switch config.Type { - case "builtin": + case SubscriberTypeBuiltin: return NewBuiltinSubscriber(f.ChannelHandlerGetter), nil - case "managedStream": + case SubscriberTypeManagedStream: return NewManagedStreamSubscriber(f.ManagedStream), nil - case "multiple": + case SubscriberTypeMultiple: if config.MultipleSubscriberConfig == nil { return nil, missingConfiguration } @@ -157,22 +157,22 @@ func (f *StorageRuleBuilder) extractConverter(config *ConverterConfig) (Converte } missingConfiguration := fmt.Errorf("missing configuration for %s", config.Type) switch config.Type { - case "jsonAuto": + case ConverterTypeJsonAuto: if config.AutoJsonConverterConfig == nil { return nil, missingConfiguration } return NewAutoJsonConverter(*config.AutoJsonConverterConfig), nil - case "jsonExact": + case ConverterTypeJsonExact: if config.ExactJsonConverterConfig == nil { return nil, missingConfiguration } return NewExactJsonConverter(*config.ExactJsonConverterConfig), nil - case "jsonFrame": + case ConverterTypeJsonFrame: if config.JsonFrameConverterConfig == nil { return nil, missingConfiguration } return NewJsonFrameConverter(*config.JsonFrameConverterConfig), nil - case "influxAuto": + case ConverterTypeInfluxAuto: if config.AutoInfluxConverterConfig == nil { return nil, missingConfiguration } @@ -188,17 +188,17 @@ func (f *StorageRuleBuilder) extractProcessor(config *ProcessorConfig) (Processo } missingConfiguration := fmt.Errorf("missing configuration for %s", config.Type) switch config.Type { - case "dropFields": + case ProcessorTypeDropFields: if config.DropFieldsProcessorConfig == nil { return nil, missingConfiguration } return NewDropFieldsProcessor(*config.DropFieldsProcessorConfig), nil - case "keepFields": + case ProcessorTypeKeepFields: if config.KeepFieldsProcessorConfig == nil { return nil, missingConfiguration } return NewKeepFieldsProcessor(*config.KeepFieldsProcessorConfig), nil - case "multiple": + case ProcessorTypeMultiple: if config.MultipleProcessorConfig == nil { return nil, missingConfiguration } @@ -223,13 +223,13 @@ func (f *StorageRuleBuilder) extractConditionChecker(config *ConditionCheckerCon } missingConfiguration := fmt.Errorf("missing configuration for %s", config.Type) switch config.Type { - case "numberCompare": + case ConditionCheckerTypeNumberCompare: if config.NumberCompareConditionConfig == nil { return nil, missingConfiguration } c := *config.NumberCompareConditionConfig return NewNumberCompareCondition(c.FieldName, c.Op, c.Value), nil - case "multiple": + case ConditionCheckerTypeMultiple: var conditions []ConditionChecker if config.MultipleConditionCheckerConfig == nil { return nil, missingConfiguration @@ -254,12 +254,12 @@ func (f *StorageRuleBuilder) extractOutputter(config *OutputterConfig, remoteWri } missingConfiguration := fmt.Errorf("missing configuration for %s", config.Type) switch config.Type { - case "redirect": + case OutputTypeRedirect: if config.RedirectOutputConfig == nil { return nil, missingConfiguration } return NewRedirectOutput(*config.RedirectOutputConfig), nil - case "multiple": + case OutputTypeMultiple: if config.MultipleOutputterConfig == nil { return nil, missingConfiguration } @@ -273,11 +273,11 @@ func (f *StorageRuleBuilder) extractOutputter(config *OutputterConfig, remoteWri outputters = append(outputters, outputter) } return NewMultipleOutput(outputters...), nil - case "managedStream": + case OutputTypeManagedStream: return NewManagedStreamOutput(f.ManagedStream), nil - case "localSubscribers": + case OutputTypeLocalSubscribers: return NewLocalSubscribersOutput(f.Node), nil - case "conditional": + case OutputTypeConditional: if config.ConditionalOutputConfig == nil { return nil, missingConfiguration } @@ -290,12 +290,12 @@ func (f *StorageRuleBuilder) extractOutputter(config *OutputterConfig, remoteWri return nil, err } return NewConditionalOutput(condition, outputter), nil - case "threshold": + case OutputTypeThreshold: if config.ThresholdOutputConfig == nil { return nil, missingConfiguration } return NewThresholdOutput(f.FrameStorage, *config.ThresholdOutputConfig), nil - case "remoteWrite": + case OutputTypeRemoteWrite: if config.RemoteWriteOutputConfig == nil { return nil, missingConfiguration } @@ -304,7 +304,7 @@ func (f *StorageRuleBuilder) extractOutputter(config *OutputterConfig, remoteWri return nil, fmt.Errorf("unknown remote write backend uid: %s", config.RemoteWriteOutputConfig.UID) } return NewRemoteWriteOutput(*remoteWriteConfig), nil - case "changeLog": + case OutputTypeChangeLog: if config.ChangeLogOutputConfig == nil { return nil, missingConfiguration } diff --git a/pkg/services/live/pipeline/converter_influx_auto.go b/pkg/services/live/pipeline/converter_influx_auto.go index 810751fee6d..18ac1c8e9e2 100644 --- a/pkg/services/live/pipeline/converter_influx_auto.go +++ b/pkg/services/live/pipeline/converter_influx_auto.go @@ -22,8 +22,14 @@ func NewAutoInfluxConverter(config AutoInfluxConverterConfig) *AutoInfluxConvert return &AutoInfluxConverter{config: config, converter: convert.NewConverter()} } -func (i AutoInfluxConverter) Convert(_ context.Context, vars Vars, body []byte) ([]*ChannelFrame, error) { - frameWrappers, err := i.converter.Convert(body, i.config.FrameFormat) +const ConverterTypeInfluxAuto = "influxAuto" + +func (c *AutoInfluxConverter) Type() string { + return ConverterTypeInfluxAuto +} + +func (c *AutoInfluxConverter) Convert(_ context.Context, vars Vars, body []byte) ([]*ChannelFrame, error) { + frameWrappers, err := c.converter.Convert(body, c.config.FrameFormat) if err != nil { return nil, err } diff --git a/pkg/services/live/pipeline/converter_json_auto.go b/pkg/services/live/pipeline/converter_json_auto.go index 0e16fd9820a..110631505ef 100644 --- a/pkg/services/live/pipeline/converter_json_auto.go +++ b/pkg/services/live/pipeline/converter_json_auto.go @@ -18,6 +18,12 @@ func NewAutoJsonConverter(c AutoJsonConverterConfig) *AutoJsonConverter { return &AutoJsonConverter{config: c} } +const ConverterTypeJsonAuto = "jsonAuto" + +func (c *AutoJsonConverter) Type() string { + return ConverterTypeJsonAuto +} + // Automatic conversion works this way: // * Time added automatically // * Nulls dropped diff --git a/pkg/services/live/pipeline/converter_json_exact.go b/pkg/services/live/pipeline/converter_json_exact.go index 020ca490fa2..38e0e7981d7 100644 --- a/pkg/services/live/pipeline/converter_json_exact.go +++ b/pkg/services/live/pipeline/converter_json_exact.go @@ -27,6 +27,12 @@ func NewExactJsonConverter(c ExactJsonConverterConfig) *ExactJsonConverter { return &ExactJsonConverter{config: c} } +const ConverterTypeJsonExact = "jsonExact" + +func (c *ExactJsonConverter) Type() string { + return ConverterTypeJsonExact +} + func (c *ExactJsonConverter) Convert(_ context.Context, vars Vars, body []byte) ([]*ChannelFrame, error) { //obj, err := oj.Parse(body) //if err != nil { diff --git a/pkg/services/live/pipeline/converter_json_frame.go b/pkg/services/live/pipeline/converter_json_frame.go index d8a7426b720..ea8264caada 100644 --- a/pkg/services/live/pipeline/converter_json_frame.go +++ b/pkg/services/live/pipeline/converter_json_frame.go @@ -20,6 +20,12 @@ func NewJsonFrameConverter(c JsonFrameConverterConfig) *JsonFrameConverter { } } +const ConverterTypeJsonFrame = "jsonFrame" + +func (c *JsonFrameConverter) Type() string { + return ConverterTypeJsonFrame +} + func (c *JsonFrameConverter) Convert(_ context.Context, _ Vars, body []byte) ([]*ChannelFrame, error) { var frame data.Frame err := json.Unmarshal(body, &frame) diff --git a/pkg/services/live/pipeline/output_changelog.go b/pkg/services/live/pipeline/output_changelog.go index 91072a7f586..f2769806677 100644 --- a/pkg/services/live/pipeline/output_changelog.go +++ b/pkg/services/live/pipeline/output_changelog.go @@ -24,13 +24,19 @@ func NewChangeLogOutput(frameStorage FrameGetSetter, config ChangeLogOutputConfi return &ChangeLogOutput{frameStorage: frameStorage, config: config} } -func (l ChangeLogOutput) Output(_ context.Context, vars OutputVars, frame *data.Frame) ([]*ChannelFrame, error) { - previousFrame, previousFrameOK, err := l.frameStorage.Get(vars.OrgID, l.config.Channel) +const OutputTypeChangeLog = "changeLog" + +func (out *ChangeLogOutput) Type() string { + return OutputTypeChangeLog +} + +func (out *ChangeLogOutput) Output(_ context.Context, vars OutputVars, frame *data.Frame) ([]*ChannelFrame, error) { + previousFrame, previousFrameOK, err := out.frameStorage.Get(vars.OrgID, out.config.Channel) if err != nil { return nil, err } - fieldName := l.config.FieldName + fieldName := out.config.FieldName previousFrameFieldIndex := -1 if previousFrameOK { @@ -78,15 +84,15 @@ func (l ChangeLogOutput) Output(_ context.Context, vars OutputVars, frame *data. if fTime.Len() > 0 { changeFrame := data.NewFrame("change", fTime, f1, f2) - err := l.frameStorage.Set(vars.OrgID, l.config.Channel, frame) + err := out.frameStorage.Set(vars.OrgID, out.config.Channel, frame) if err != nil { return nil, err } return []*ChannelFrame{{ - Channel: l.config.Channel, + Channel: out.config.Channel, Frame: changeFrame, }}, nil } - return nil, l.frameStorage.Set(vars.OrgID, l.config.Channel, frame) + return nil, out.frameStorage.Set(vars.OrgID, out.config.Channel, frame) } diff --git a/pkg/services/live/pipeline/output_conditional.go b/pkg/services/live/pipeline/output_conditional.go index e47960c74e1..fef42cadebc 100644 --- a/pkg/services/live/pipeline/output_conditional.go +++ b/pkg/services/live/pipeline/output_conditional.go @@ -15,13 +15,19 @@ func NewConditionalOutput(condition ConditionChecker, outputter Outputter) *Cond return &ConditionalOutput{Condition: condition, Outputter: outputter} } -func (l ConditionalOutput) Output(ctx context.Context, vars OutputVars, frame *data.Frame) ([]*ChannelFrame, error) { - ok, err := l.Condition.CheckCondition(ctx, frame) +const OutputTypeConditional = "conditional" + +func (out *ConditionalOutput) Type() string { + return OutputTypeConditional +} + +func (out ConditionalOutput) Output(ctx context.Context, vars OutputVars, frame *data.Frame) ([]*ChannelFrame, error) { + ok, err := out.Condition.CheckCondition(ctx, frame) if err != nil { return nil, err } if !ok { return nil, nil } - return l.Outputter.Output(ctx, vars, frame) + return out.Outputter.Output(ctx, vars, frame) } diff --git a/pkg/services/live/pipeline/output_local_subscribers.go b/pkg/services/live/pipeline/output_local_subscribers.go index 5a8a2931536..ed1adf4788f 100644 --- a/pkg/services/live/pipeline/output_local_subscribers.go +++ b/pkg/services/live/pipeline/output_local_subscribers.go @@ -20,7 +20,13 @@ func NewLocalSubscribersOutput(node *centrifuge.Node) *LocalSubscribersOutput { return &LocalSubscribersOutput{node: node} } -func (l *LocalSubscribersOutput) Output(_ context.Context, vars OutputVars, frame *data.Frame) ([]*ChannelFrame, error) { +const OutputTypeLocalSubscribers = "localSubscribers" + +func (out *LocalSubscribersOutput) Type() string { + return OutputTypeLocalSubscribers +} + +func (out *LocalSubscribersOutput) Output(_ context.Context, vars OutputVars, frame *data.Frame) ([]*ChannelFrame, error) { channelID := vars.Channel channel := orgchannel.PrependOrgID(vars.OrgID, channelID) frameJSON, err := json.Marshal(frame) @@ -30,7 +36,7 @@ func (l *LocalSubscribersOutput) Output(_ context.Context, vars OutputVars, fram pub := ¢rifuge.Publication{ Data: frameJSON, } - err = l.node.Hub().BroadcastPublication(channel, pub, centrifuge.StreamPosition{}) + err = out.node.Hub().BroadcastPublication(channel, pub, centrifuge.StreamPosition{}) if err != nil { return nil, fmt.Errorf("error publishing %s: %w", string(frameJSON), err) } diff --git a/pkg/services/live/pipeline/output_managed_stream.go b/pkg/services/live/pipeline/output_managed_stream.go index bd225d704a4..5ec929ec487 100644 --- a/pkg/services/live/pipeline/output_managed_stream.go +++ b/pkg/services/live/pipeline/output_managed_stream.go @@ -16,8 +16,14 @@ func NewManagedStreamOutput(managedStream *managedstream.Runner) *ManagedStreamO return &ManagedStreamOutput{managedStream: managedStream} } -func (l *ManagedStreamOutput) Output(_ context.Context, vars OutputVars, frame *data.Frame) ([]*ChannelFrame, error) { - stream, err := l.managedStream.GetOrCreateStream(vars.OrgID, vars.Scope, vars.Namespace) +const OutputTypeManagedStream = "managedStream" + +func (out *ManagedStreamOutput) Type() string { + return OutputTypeManagedStream +} + +func (out *ManagedStreamOutput) Output(_ context.Context, vars OutputVars, frame *data.Frame) ([]*ChannelFrame, error) { + stream, err := out.managedStream.GetOrCreateStream(vars.OrgID, vars.Scope, vars.Namespace) if err != nil { logger.Error("Error getting stream", "error", err) return nil, err diff --git a/pkg/services/live/pipeline/output_multiple.go b/pkg/services/live/pipeline/output_multiple.go index d763bf259ec..bd21934b438 100644 --- a/pkg/services/live/pipeline/output_multiple.go +++ b/pkg/services/live/pipeline/output_multiple.go @@ -12,9 +12,15 @@ type MultipleOutput struct { Outputters []Outputter } -func (m MultipleOutput) Output(ctx context.Context, vars OutputVars, frame *data.Frame) ([]*ChannelFrame, error) { +const OutputTypeMultiple = "multiple" + +func (out *MultipleOutput) Type() string { + return OutputTypeMultiple +} + +func (out MultipleOutput) Output(ctx context.Context, vars OutputVars, frame *data.Frame) ([]*ChannelFrame, error) { var frames []*ChannelFrame - for _, out := range m.Outputters { + for _, out := range out.Outputters { f, err := out.Output(ctx, vars, frame) if err != nil { logger.Error("Error outputting frame", "error", err) diff --git a/pkg/services/live/pipeline/output_redirect.go b/pkg/services/live/pipeline/output_redirect.go index d562af77cf3..803a14f27f2 100644 --- a/pkg/services/live/pipeline/output_redirect.go +++ b/pkg/services/live/pipeline/output_redirect.go @@ -22,12 +22,18 @@ func NewRedirectOutput(config RedirectOutputConfig) *RedirectOutput { return &RedirectOutput{config: config} } -func (l *RedirectOutput) Output(_ context.Context, vars OutputVars, frame *data.Frame) ([]*ChannelFrame, error) { - if vars.Channel == l.config.Channel { - return nil, fmt.Errorf("redirect to the same channel: %s", l.config.Channel) +const OutputTypeRedirect = "redirect" + +func (out *RedirectOutput) Type() string { + return OutputTypeRedirect +} + +func (out *RedirectOutput) Output(_ context.Context, vars OutputVars, frame *data.Frame) ([]*ChannelFrame, error) { + if vars.Channel == out.config.Channel { + return nil, fmt.Errorf("redirect to the same channel: %s", out.config.Channel) } return []*ChannelFrame{{ - Channel: l.config.Channel, + Channel: out.config.Channel, Frame: frame, }}, nil } diff --git a/pkg/services/live/pipeline/output_remote_write.go b/pkg/services/live/pipeline/output_remote_write.go index 8d6eb10457f..30e158f78e8 100644 --- a/pkg/services/live/pipeline/output_remote_write.go +++ b/pkg/services/live/pipeline/output_remote_write.go @@ -32,8 +32,14 @@ func NewRemoteWriteOutput(config RemoteWriteConfig) *RemoteWriteOutput { } } -func (r RemoteWriteOutput) Output(_ context.Context, _ OutputVars, frame *data.Frame) ([]*ChannelFrame, error) { - if r.config.Endpoint == "" { +const OutputTypeRemoteWrite = "remoteWrite" + +func (out *RemoteWriteOutput) Type() string { + return OutputTypeRemoteWrite +} + +func (out *RemoteWriteOutput) Output(_ context.Context, _ OutputVars, frame *data.Frame) ([]*ChannelFrame, error) { + if out.config.Endpoint == "" { logger.Debug("Skip sending to remote write: no url") return nil, nil } @@ -45,8 +51,8 @@ func (r RemoteWriteOutput) Output(_ context.Context, _ OutputVars, frame *data.F return nil, err } - logger.Debug("Sending to remote write endpoint", "url", r.config.Endpoint, "bodyLength", len(remoteWriteData)) - req, err := http.NewRequest(http.MethodPost, r.config.Endpoint, bytes.NewReader(remoteWriteData)) + logger.Debug("Sending to remote write endpoint", "url", out.config.Endpoint, "bodyLength", len(remoteWriteData)) + req, err := http.NewRequest(http.MethodPost, out.config.Endpoint, bytes.NewReader(remoteWriteData)) if err != nil { logger.Error("Error constructing remote write request", "error", err) return nil, err @@ -54,10 +60,10 @@ func (r RemoteWriteOutput) Output(_ context.Context, _ OutputVars, frame *data.F 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") - req.SetBasicAuth(r.config.User, r.config.Password) + req.SetBasicAuth(out.config.User, out.config.Password) started := time.Now() - resp, err := r.httpClient.Do(req) + resp, err := out.httpClient.Do(req) if err != nil { logger.Error("Error sending remote write request", "error", err) return nil, err @@ -67,6 +73,6 @@ func (r RemoteWriteOutput) Output(_ context.Context, _ OutputVars, frame *data.F logger.Error("Unexpected response code from remote write endpoint", "code", resp.StatusCode) return nil, errors.New("unexpected response code from remote write endpoint") } - logger.Debug("Successfully sent to remote write endpoint", "url", r.config.Endpoint, "elapsed", time.Since(started)) + logger.Debug("Successfully sent to remote write endpoint", "url", out.config.Endpoint, "elapsed", time.Since(started)) return nil, nil } diff --git a/pkg/services/live/pipeline/output_threshold.go b/pkg/services/live/pipeline/output_threshold.go index bda29b2b5fa..48a26b2e678 100644 --- a/pkg/services/live/pipeline/output_threshold.go +++ b/pkg/services/live/pipeline/output_threshold.go @@ -31,15 +31,21 @@ func NewThresholdOutput(frameStorage FrameGetSetter, config ThresholdOutputConfi return &ThresholdOutput{frameStorage: frameStorage, config: config} } -func (l *ThresholdOutput) Output(_ context.Context, vars OutputVars, frame *data.Frame) ([]*ChannelFrame, error) { +const OutputTypeThreshold = "threshold" + +func (out *ThresholdOutput) Type() string { + return OutputTypeThreshold +} + +func (out *ThresholdOutput) Output(_ context.Context, vars OutputVars, frame *data.Frame) ([]*ChannelFrame, error) { if frame == nil { return nil, nil } - previousFrame, previousFrameOk, err := l.frameStorage.Get(vars.OrgID, l.config.Channel) + previousFrame, previousFrameOk, err := out.frameStorage.Get(vars.OrgID, out.config.Channel) if err != nil { return nil, err } - fieldName := l.config.FieldName + fieldName := out.config.FieldName currentFrameFieldIndex := -1 for i, f := range frame.Fields { @@ -136,15 +142,15 @@ func (l *ThresholdOutput) Output(_ context.Context, vars OutputVars, frame *data if fTime.Len() > 0 { stateFrame := data.NewFrame("state", fTime, f1, f2, f3) - err := l.frameStorage.Set(vars.OrgID, l.config.Channel, frame) + err := out.frameStorage.Set(vars.OrgID, out.config.Channel, frame) if err != nil { return nil, err } return []*ChannelFrame{{ - Channel: l.config.Channel, + Channel: out.config.Channel, Frame: stateFrame, }}, nil } - return nil, l.frameStorage.Set(vars.OrgID, l.config.Channel, frame) + return nil, out.frameStorage.Set(vars.OrgID, out.config.Channel, frame) } diff --git a/pkg/services/live/pipeline/pipeline.go b/pkg/services/live/pipeline/pipeline.go index f6a23eb02f0..28d75ce98f7 100644 --- a/pkg/services/live/pipeline/pipeline.go +++ b/pkg/services/live/pipeline/pipeline.go @@ -47,22 +47,26 @@ type OutputVars struct { // of resulting slice will be then individually processed and outputted // according configured channel rules. type Converter interface { + Type() string Convert(ctx context.Context, vars Vars, body []byte) ([]*ChannelFrame, error) } // Processor can modify data.Frame in a custom way before it will be outputted. type Processor interface { + Type() string Process(ctx context.Context, vars ProcessorVars, frame *data.Frame) (*data.Frame, error) } // Outputter outputs data.Frame to a custom destination. Or simply // do nothing if some conditions not met. type Outputter interface { + Type() string Output(ctx context.Context, vars OutputVars, frame *data.Frame) ([]*ChannelFrame, error) } // Subscriber can handle channel subscribe events. type Subscriber interface { + Type() string Subscribe(ctx context.Context, vars Vars) (models.SubscribeReply, backend.SubscribeStreamStatus, error) } diff --git a/pkg/services/live/pipeline/pipeline_test.go b/pkg/services/live/pipeline/pipeline_test.go index cf1b0c2e4ed..95b49571b7d 100644 --- a/pkg/services/live/pipeline/pipeline_test.go +++ b/pkg/services/live/pipeline/pipeline_test.go @@ -16,7 +16,7 @@ type testRuleGetter struct { rules map[string]*LiveChannelRule } -func (t *testRuleGetter) Get(orgID int64, channel string) (*LiveChannelRule, bool, error) { +func (t *testRuleGetter) Get(_ int64, channel string) (*LiveChannelRule, bool, error) { t.mu.Lock() defer t.mu.Unlock() rule, ok := t.rules[channel] @@ -48,12 +48,20 @@ type testConverter struct { frame *data.Frame } +func (t *testConverter) Type() string { + return "test" +} + func (t *testConverter) Convert(_ context.Context, _ Vars, _ []byte) ([]*ChannelFrame, error) { return []*ChannelFrame{{Channel: t.channel, Frame: t.frame}}, nil } type testProcessor struct{} +func (t *testProcessor) Type() string { + return "test" +} + func (t *testProcessor) Process(_ context.Context, _ ProcessorVars, frame *data.Frame) (*data.Frame, error) { return frame, nil } @@ -63,6 +71,10 @@ type testOutputter struct { frame *data.Frame } +func (t *testOutputter) Type() string { + return "test" +} + func (t *testOutputter) Output(_ context.Context, _ OutputVars, frame *data.Frame) ([]*ChannelFrame, error) { if t.err != nil { return nil, t.err diff --git a/pkg/services/live/pipeline/processor_drop_field.go b/pkg/services/live/pipeline/processor_drop_field.go index 536579ad269..e0917269cea 100644 --- a/pkg/services/live/pipeline/processor_drop_field.go +++ b/pkg/services/live/pipeline/processor_drop_field.go @@ -23,8 +23,14 @@ func NewDropFieldsProcessor(config DropFieldsProcessorConfig) *DropFieldsProcess return &DropFieldsProcessor{config: config} } -func (d DropFieldsProcessor) Process(_ context.Context, _ ProcessorVars, frame *data.Frame) (*data.Frame, error) { - for _, f := range d.config.FieldNames { +const ProcessorTypeDropFields = "dropFields" + +func (p *DropFieldsProcessor) Type() string { + return ProcessorTypeDropFields +} + +func (p *DropFieldsProcessor) Process(_ context.Context, _ ProcessorVars, frame *data.Frame) (*data.Frame, error) { + for _, f := range p.config.FieldNames { inner: for i, field := range frame.Fields { if f == field.Name { diff --git a/pkg/services/live/pipeline/processor_keep_field.go b/pkg/services/live/pipeline/processor_keep_field.go index e71c411c729..a4943142808 100644 --- a/pkg/services/live/pipeline/processor_keep_field.go +++ b/pkg/services/live/pipeline/processor_keep_field.go @@ -28,10 +28,16 @@ func stringInSlice(str string, slice []string) bool { return false } -func (d KeepFieldsProcessor) Process(_ context.Context, _ ProcessorVars, frame *data.Frame) (*data.Frame, error) { +const ProcessorTypeKeepFields = "keepFields" + +func (p *KeepFieldsProcessor) Type() string { + return ProcessorTypeKeepFields +} + +func (p *KeepFieldsProcessor) Process(_ context.Context, _ ProcessorVars, frame *data.Frame) (*data.Frame, error) { var fieldsToKeep []*data.Field for _, field := range frame.Fields { - if stringInSlice(field.Name, d.config.FieldNames) { + if stringInSlice(field.Name, p.config.FieldNames) { fieldsToKeep = append(fieldsToKeep, field) } } diff --git a/pkg/services/live/pipeline/processor_multiple.go b/pkg/services/live/pipeline/processor_multiple.go index 40d40798971..91f228db632 100644 --- a/pkg/services/live/pipeline/processor_multiple.go +++ b/pkg/services/live/pipeline/processor_multiple.go @@ -12,8 +12,14 @@ type MultipleProcessor struct { Processors []Processor } -func (m MultipleProcessor) Process(ctx context.Context, vars ProcessorVars, frame *data.Frame) (*data.Frame, error) { - for _, p := range m.Processors { +const ProcessorTypeMultiple = "multiple" + +func (p *MultipleProcessor) Type() string { + return ProcessorTypeMultiple +} + +func (p *MultipleProcessor) Process(ctx context.Context, vars ProcessorVars, frame *data.Frame) (*data.Frame, error) { + for _, p := range p.Processors { var err error frame, err = p.Process(ctx, vars, frame) if err != nil { diff --git a/pkg/services/live/pipeline/subscribe_builtin.go b/pkg/services/live/pipeline/subscribe_builtin.go index 7788ab1c28a..b4e7de92f56 100644 --- a/pkg/services/live/pipeline/subscribe_builtin.go +++ b/pkg/services/live/pipeline/subscribe_builtin.go @@ -18,16 +18,22 @@ type ChannelHandlerGetter interface { GetChannelHandler(user *models.SignedInUser, channel string) (models.ChannelHandler, live.Channel, error) } +const SubscriberTypeBuiltin = "builtin" + func NewBuiltinSubscriber(channelHandlerGetter ChannelHandlerGetter) *BuiltinSubscriber { return &BuiltinSubscriber{channelHandlerGetter: channelHandlerGetter} } -func (m *BuiltinSubscriber) Subscribe(ctx context.Context, vars Vars) (models.SubscribeReply, backend.SubscribeStreamStatus, error) { +func (s *BuiltinSubscriber) Type() string { + return SubscriberTypeBuiltin +} + +func (s *BuiltinSubscriber) Subscribe(ctx context.Context, vars Vars) (models.SubscribeReply, backend.SubscribeStreamStatus, error) { u, ok := livecontext.GetContextSignedUser(ctx) if !ok { return models.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, nil } - handler, _, err := m.channelHandlerGetter.GetChannelHandler(u, vars.Channel) + handler, _, err := s.channelHandlerGetter.GetChannelHandler(u, vars.Channel) if err != nil { return models.SubscribeReply{}, 0, err } diff --git a/pkg/services/live/pipeline/subscribe_managed_stream.go b/pkg/services/live/pipeline/subscribe_managed_stream.go index 9ddd21d16ce..cd34bef0cc6 100644 --- a/pkg/services/live/pipeline/subscribe_managed_stream.go +++ b/pkg/services/live/pipeline/subscribe_managed_stream.go @@ -14,12 +14,18 @@ type ManagedStreamSubscriber struct { managedStream *managedstream.Runner } +const SubscriberTypeManagedStream = "managedStream" + func NewManagedStreamSubscriber(managedStream *managedstream.Runner) *ManagedStreamSubscriber { return &ManagedStreamSubscriber{managedStream: managedStream} } -func (m *ManagedStreamSubscriber) Subscribe(ctx context.Context, vars Vars) (models.SubscribeReply, backend.SubscribeStreamStatus, error) { - stream, err := m.managedStream.GetOrCreateStream(vars.OrgID, vars.Scope, vars.Namespace) +func (s *ManagedStreamSubscriber) Type() string { + return SubscriberTypeManagedStream +} + +func (s *ManagedStreamSubscriber) Subscribe(ctx context.Context, vars Vars) (models.SubscribeReply, backend.SubscribeStreamStatus, error) { + stream, err := s.managedStream.GetOrCreateStream(vars.OrgID, vars.Scope, vars.Namespace) if err != nil { logger.Error("Error getting managed stream", "error", err) return models.SubscribeReply{}, 0, err diff --git a/pkg/services/live/pipeline/subscribe_multiple.go b/pkg/services/live/pipeline/subscribe_multiple.go index b508cddb627..24c271b9456 100644 --- a/pkg/services/live/pipeline/subscribe_multiple.go +++ b/pkg/services/live/pipeline/subscribe_multiple.go @@ -15,10 +15,16 @@ func NewMultipleSubscriber(subscribers ...Subscriber) *MultipleSubscriber { return &MultipleSubscriber{Subscribers: subscribers} } -func (m *MultipleSubscriber) Subscribe(ctx context.Context, vars Vars) (models.SubscribeReply, backend.SubscribeStreamStatus, error) { +const SubscriberTypeMultiple = "multiple" + +func (s *MultipleSubscriber) Type() string { + return SubscriberTypeMultiple +} + +func (s *MultipleSubscriber) Subscribe(ctx context.Context, vars Vars) (models.SubscribeReply, backend.SubscribeStreamStatus, error) { finalReply := models.SubscribeReply{} - for _, s := range m.Subscribers { + for _, s := range s.Subscribers { reply, status, err := s.Subscribe(ctx, vars) if err != nil { return models.SubscribeReply{}, 0, err diff --git a/pkg/services/live/pipeline/subscribe_permission_denied.go b/pkg/services/live/pipeline/subscribe_permission_denied.go deleted file mode 100644 index bfd3fe8a68d..00000000000 --- a/pkg/services/live/pipeline/subscribe_permission_denied.go +++ /dev/null @@ -1,18 +0,0 @@ -package pipeline - -import ( - "context" - - "github.com/grafana/grafana-plugin-sdk-go/backend" - "github.com/grafana/grafana/pkg/models" -) - -type PermissionDeniedSubscriber struct{} - -func NewPermissionDeniedSubscriber() *PermissionDeniedSubscriber { - return &PermissionDeniedSubscriber{} -} - -func (m *PermissionDeniedSubscriber) Subscribe(ctx context.Context, vars Vars) (models.SubscribeReply, backend.SubscribeStreamStatus, error) { - return models.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, nil -}