Live: api to show available pipeline entities (#39469)

Co-authored-by: Ryan McKinley <ryantxu@gmail.com>

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)
 			}
 		})

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)

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)
 }

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}
 }

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)
 			}
 		}
 	}

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
 		}

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
 	}

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

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 {

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)

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)
 }

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)
 }

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 := &centrifuge.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)
 	}

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

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)

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
 }

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
 }

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)
 }

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)
 }
 

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

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 {

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)
 		}
 	}

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 {

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
 	}

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

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

pkg/services/live/pipeline/subscribe_permission_denied.go

@@ -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
-}