feat. X-Ray - Real time stream debugging for Flow components

cmd/internal/flowmode/cmd_run.go

@@ -34,6 +34,7 @@ import (
 	"github.com/grafana/agent/service/labelstore"
 	otel_service "github.com/grafana/agent/service/otel"
 	uiservice "github.com/grafana/agent/service/ui"
+	"github.com/grafana/agent/service/xray"
 	"github.com/grafana/ckit/advertise"
 	"github.com/grafana/ckit/peer"
 	"github.com/grafana/river/diag"
@@ -243,9 +244,12 @@ func (fr *flowRun) Run(configPath string) error {
 		EnablePProf:      fr.enablePprof,
 	})
 
+	xrayService := xray.New(l)
+
 	uiService := uiservice.New(uiservice.Options{
 		UIPrefix: fr.uiPrefix,
 		Cluster:  clusterService.Data().(cluster.Cluster),
+		Xray:     xrayService,
 	})
 
 	otelService := otel_service.New(l)
@@ -263,6 +267,7 @@ func (fr *flowRun) Run(configPath string) error {
 		Reg:      reg,
 		Services: []service.Service{
 			httpService,
+			xrayService,
 			uiService,
 			clusterService,
 			otelService,

component/discovery/discovery.go

@@ -2,13 +2,15 @@ package discovery
 
 import (
 	"context"
+	"fmt"
 	"sort"
 	"strings"
 	"sync"
 	"time"
 
 	"github.com/grafana/agent/component"
 	"github.com/grafana/agent/service/cluster"
+	"github.com/grafana/agent/service/xray"
 	"github.com/grafana/ckit/shard"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/discovery"
@@ -108,17 +110,25 @@ type Component struct {
 	discMut       sync.Mutex
 	latestDisc    discovery.Discoverer
 	newDiscoverer chan struct{}
+	xray          *xray.Service
 
 	creator Creator
 }
 
 // New creates a discovery component given arguments and a concrete Discovery implementation function.
 func New(o component.Options, args component.Arguments, creator Creator) (*Component, error) {
+	data, err := o.GetServiceData(xray.ServiceName)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get information about X-Ray service: %w", err)
+	}
+	xray := data.(*xray.Service)
+
 	c := &Component{
 		opts:    o,
 		creator: creator,
 		// buffered to avoid deadlock from the first immediate update
 		newDiscoverer: make(chan struct{}, 1),
+		xray:          xray,
 	}
 	return c, c.Update(args)
 }
@@ -198,6 +208,10 @@ func (c *Component) runDiscovery(ctx context.Context, d Discoverer) {
 				allTargets = append(allTargets, labels)
 			}
 		}
+
+		if ds := c.xray.GetDebugStream(c.opts.ID); ds != nil {
+			ds(fmt.Sprintf("%s", allTargets))
+		}
 		c.opts.OnStateChange(Exports{Targets: allTargets})
 	}
 

component/otelcol/exporter/exporter.go

@@ -5,16 +5,18 @@ package exporter
 import (
 	"context"
 	"errors"
+	"fmt"
 	"os"
 
 	"github.com/grafana/agent/component"
 	"github.com/grafana/agent/component/otelcol"
 	"github.com/grafana/agent/component/otelcol/internal/lazycollector"
-	"github.com/grafana/agent/component/otelcol/internal/lazyconsumer"
+	"github.com/grafana/agent/component/otelcol/internal/lazyexporterconsumer"
 	"github.com/grafana/agent/component/otelcol/internal/scheduler"
 	"github.com/grafana/agent/component/otelcol/internal/views"
 	"github.com/grafana/agent/pkg/build"
 	"github.com/grafana/agent/pkg/util/zapadapter"
+	"github.com/grafana/agent/service/xray"
 	"github.com/prometheus/client_golang/prometheus"
 	otelcomponent "go.opentelemetry.io/collector/component"
 	otelexporter "go.opentelemetry.io/collector/exporter"
@@ -79,7 +81,7 @@ type Exporter struct {
 
 	opts     component.Options
 	factory  otelexporter.Factory
-	consumer *lazyconsumer.Consumer
+	consumer *lazyexporterconsumer.Consumer
 
 	sched     *scheduler.Scheduler
 	collector *lazycollector.Collector
@@ -101,9 +103,18 @@ var (
 // The registered component must be registered to export the
 // otelcol.ConsumerExports type, otherwise New will panic.
 func New(opts component.Options, f otelexporter.Factory, args Arguments, supportedSignals TypeSignal) (*Exporter, error) {
+	data, err := opts.GetServiceData(xray.ServiceName)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get information about X-Ray service: %w", err)
+	}
+	xray := data.(*xray.Service)
+	debugStreamCallback := func() func(string) {
+		return xray.GetDebugStream(opts.ID)
+	}
+
 	ctx, cancel := context.WithCancel(context.Background())
 
-	consumer := lazyconsumer.New(ctx)
+	consumer := lazyexporterconsumer.New(ctx, debugStreamCallback)
 
 	// Create a lazy collector where metrics from the upstream component will be
 	// forwarded.

component/otelcol/exporter/loki/loki.go

@@ -3,13 +3,16 @@ package loki
 
 import (
 	"context"
+	"fmt"
+	"time"
 
 	"github.com/go-kit/log"
 	"github.com/grafana/agent/component"
 	"github.com/grafana/agent/component/common/loki"
 	"github.com/grafana/agent/component/otelcol"
 	"github.com/grafana/agent/component/otelcol/exporter/loki/internal/convert"
 	"github.com/grafana/agent/component/otelcol/internal/lazyconsumer"
+	"github.com/grafana/agent/service/xray"
 )
 
 func init() {
@@ -34,20 +37,34 @@ type Component struct {
 	log  log.Logger
 	opts component.Options
 
-	converter *convert.Converter
+	converter               *convert.Converter
+	logsReceiverStreamDebug *logsReceiverStreamDebug
+	xray                    *xray.Service
 }
 
-var _ component.Component = (*Component)(nil)
+var (
+	_ component.Component = (*Component)(nil)
+)
 
 // New creates a new otelcol.exporter.loki component.
 func New(o component.Options, c Arguments) (*Component, error) {
 	converter := convert.New(o.Logger, o.Registerer, c.ForwardTo)
 
+	data, err := o.GetServiceData(xray.ServiceName)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get information about X-Ray service: %w", err)
+	}
+	xray := data.(*xray.Service)
+
 	res := &Component{
 		log:  o.Logger,
 		opts: o,
 
 		converter: converter,
+		logsReceiverStreamDebug: &logsReceiverStreamDebug{
+			entries: make(chan loki.Entry),
+		},
+		xray: xray,
 	}
 	if err := res.Update(c); err != nil {
 		return nil, err
@@ -65,13 +82,29 @@ func New(o component.Options, c Arguments) (*Component, error) {
 
 // Run implements Component.
 func (c *Component) Run(ctx context.Context) error {
-	<-ctx.Done()
-	return nil
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case entry := <-c.logsReceiverStreamDebug.Chan():
+			if ds := c.xray.GetDebugStream(c.opts.ID); ds != nil {
+				ds(fmt.Sprintf("ts=%s, labels=%s, entry=%s", entry.Timestamp.Format(time.RFC3339Nano), entry.Labels.String(), entry.Line))
+			}
+		}
+	}
 }
 
 // Update implements Component.
 func (c *Component) Update(newConfig component.Arguments) error {
 	cfg := newConfig.(Arguments)
-	c.converter.UpdateFanout(cfg.ForwardTo)
+	c.converter.UpdateFanout(append(cfg.ForwardTo, c.logsReceiverStreamDebug))
 	return nil
 }
+
+type logsReceiverStreamDebug struct {
+	entries chan loki.Entry
+}
+
+func (l *logsReceiverStreamDebug) Chan() chan loki.Entry {
+	return l.entries
+}

component/otelcol/exporter/prometheus/prometheus.go

@@ -14,6 +14,10 @@ import (
 	"github.com/grafana/agent/component/otelcol/internal/lazyconsumer"
 	"github.com/grafana/agent/component/prometheus"
 	"github.com/grafana/agent/service/labelstore"
+	"github.com/prometheus/prometheus/model/exemplar"
+	"github.com/prometheus/prometheus/model/histogram"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/model/metadata"
 	"github.com/prometheus/prometheus/storage"
 )
 
@@ -72,8 +76,9 @@ type Component struct {
 	fanout    *prometheus.Fanout
 	converter *convert.Converter
 
-	mut sync.RWMutex
-	cfg Arguments
+	mut                 sync.RWMutex
+	cfg                 Arguments
+	debugStreamCallback func(func() string)
 }
 
 var _ component.Component = (*Component)(nil)
@@ -85,17 +90,55 @@ func New(o component.Options, c Arguments) (*Component, error) {
 		return nil, err
 	}
 	ls := service.(labelstore.LabelStore)
-	fanout := prometheus.NewFanout(nil, o.ID, o.Registerer, ls)
-
-	converter := convert.New(o.Logger, fanout, convertArgumentsToConvertOptions(c))
 
 	res := &Component{
-		log:  o.Logger,
-		opts: o,
-
-		fanout:    fanout,
-		converter: converter,
+		log:                 o.Logger,
+		opts:                o,
+		debugStreamCallback: func(func() string) {},
 	}
+
+	res.fanout = prometheus.NewFanout(nil, o.ID, o.Registerer, ls)
+
+	interceptor := prometheus.NewInterceptor(res.fanout, ls,
+		prometheus.WithAppendHook(func(globalRef storage.SeriesRef, l labels.Labels, t int64, v float64, next storage.Appender) (storage.SeriesRef, error) {
+			localID := ls.GetLocalRefID(res.opts.ID, uint64(globalRef))
+			_, nextErr := next.Append(storage.SeriesRef(localID), l, t, v)
+			res.debugStreamCallback(func() string { return fmt.Sprintf("ts=%d, labels=%s, value=%f", t, l, v) })
+			return globalRef, nextErr
+		}),
+		prometheus.WithHistogramHook(func(globalRef storage.SeriesRef, l labels.Labels, t int64, h *histogram.Histogram, fh *histogram.FloatHistogram, next storage.Appender) (storage.SeriesRef, error) {
+			localID := ls.GetLocalRefID(res.opts.ID, uint64(globalRef))
+			_, nextErr := next.AppendHistogram(storage.SeriesRef(localID), l, t, h, fh)
+			res.debugStreamCallback(func() string {
+				if h != nil {
+					return fmt.Sprintf("ts=%d, labels=%s, histogram=%s", t, l, h.String())
+				} else if fh != nil {
+					return fmt.Sprintf("ts=%d, labels=%s, float_histogram=%s", t, l, fh.String())
+				}
+				return fmt.Sprintf("ts=%d, labels=%s, no_value", t, l)
+			})
+			return globalRef, nextErr
+		}),
+		prometheus.WithMetadataHook(func(globalRef storage.SeriesRef, l labels.Labels, m metadata.Metadata, next storage.Appender) (storage.SeriesRef, error) {
+			localID := ls.GetLocalRefID(res.opts.ID, uint64(globalRef))
+			_, nextErr := next.UpdateMetadata(storage.SeriesRef(localID), l, m)
+			res.debugStreamCallback(func() string {
+				return fmt.Sprintf("labels=%s, type=%s, unit=%s, help=%s", l, m.Type, m.Unit, m.Help)
+			})
+			return globalRef, nextErr
+		}),
+		prometheus.WithExemplarHook(func(globalRef storage.SeriesRef, l labels.Labels, e exemplar.Exemplar, next storage.Appender) (storage.SeriesRef, error) {
+			localID := ls.GetLocalRefID(res.opts.ID, uint64(globalRef))
+			_, nextErr := next.AppendExemplar(storage.SeriesRef(localID), l, e)
+			res.debugStreamCallback(func() string {
+				return fmt.Sprintf("ts=%d, labels=%s, exemplar_labels=%s, value=%f", e.Ts, l, e.Labels, e.Value)
+			})
+			return globalRef, nextErr
+		}),
+	)
+
+	res.converter = convert.New(o.Logger, interceptor, convertArgumentsToConvertOptions(c))
+
 	if err := res.Update(c); err != nil {
 		return nil, err
 	}
@@ -104,7 +147,7 @@ func New(o component.Options, c Arguments) (*Component, error) {
 	// remain the same throughout the component's lifetime, so we do this during
 	// component construction.
 	export := lazyconsumer.New(context.Background())
-	export.SetConsumers(nil, converter, nil)
+	export.SetConsumers(nil, res.converter, nil)
 	o.OnStateChange(otelcol.ConsumerExports{Input: export})
 
 	return res, nil
@@ -159,3 +202,7 @@ func convertArgumentsToConvertOptions(args Arguments) convert.Options {
 		ResourceToTelemetryConversion: args.ResourceToTelemetryConversion,
 	}
 }
+
+func (c *Component) HookDebugStream(active bool, debugStreamCallback func(computeDataFunc func() string)) {
+	c.debugStreamCallback = debugStreamCallback
+}