Pyroscope fixes (#3697)

* pyroscope.ebpf: Retry after eBPF errors and buffer config updates. (#3381)

* fix: Keep ebpf component loop running

If the eBPF component encounter an error, it stops the main processing
loops, which then caused issues because any config reload for alloy
would block the whole agent.

Fixes #3332

Also for long running scrapes we were blocking the component evaluation
as we are waiting for the config to be updated. I switched this to be a
buffered channel, so the config updated are unblocked.

In case of multiple updated buffered it will skip the in between updates
and go straight to the latest version.

Fixes #1371

* Fix data racec in test

* Limit tries to start an ebpf session

* Add info logs at session start/end

* Pass context around to shutdown properly

* fix(pyroscope.scrape): godeltaprof scraping (#3542)

* fix(pyroscope.scrape): godeltaprof scraping

* cl

* update changelog

---------

Co-authored-by: Christian Simon <[email protected]>
Co-authored-by: Tolya Korniltsev <[email protected]>

Author: 3 people

CHANGELOG.md

@@ -7,7 +7,14 @@ This document contains a historical list of changes between releases. Only
 changes that impact end-user behavior are listed; changes to documentation or
 internal API changes are not present.
 
-v1.9.0-rc.0
+v1.9.0-rc.2
+-----------------
+
+### Bugfixes
+
+- Fix `pyroscope.scrape` scraping godeltaprof profiles. (@korniltsev)
+
+v1.9.0-rc.1
 -----------------
 
 ### Breaking changes

internal/component/pyroscope/ebpf/ebpf_linux.go

@@ -15,7 +15,6 @@ import (
 	"github.com/grafana/pyroscope/ebpf/pprof"
 	"github.com/grafana/pyroscope/ebpf/sd"
 	"github.com/grafana/pyroscope/ebpf/symtab"
-	"github.com/oklog/run"
 
 	"github.com/grafana/alloy/internal/component"
 	"github.com/grafana/alloy/internal/component/pyroscope"
@@ -61,7 +60,7 @@ func New(opts component.Options, args Arguments) (component.Component, error) {
 		args:         args,
 		targetFinder: targetFinder,
 		session:      session,
-		argsUpdate:   make(chan Arguments),
+		argsUpdate:   make(chan Arguments, 4),
 	}
 	res.metrics.targetsActive.Set(float64(len(res.targetFinder.DebugInfo())))
 	return res, nil
@@ -104,48 +103,89 @@ type Component struct {
 	debugInfo     DebugInfo
 	debugInfoLock sync.Mutex
 	metrics       *metrics
+
+	healthMut sync.RWMutex
+	health    component.Health
 }
 
 func (c *Component) Run(ctx context.Context) error {
-	err := c.session.Start()
-	if err != nil {
-		return fmt.Errorf("ebpf profiling session start: %w", err)
-	}
-	defer c.session.Stop()
-
-	var g run.Group
-	g.Add(func() error {
-		collectInterval := c.args.CollectInterval
-		t := time.NewTicker(collectInterval)
-		defer t.Stop()
-		for {
-			select {
-			case <-ctx.Done():
-				return nil
-			case newArgs := <-c.argsUpdate:
-				c.args = newArgs
-				c.session.UpdateTargets(targetsOptionFromArgs(c.args))
-				c.metrics.targetsActive.Set(float64(len(c.targetFinder.DebugInfo())))
-				err := c.session.Update(convertSessionOptions(c.args, c.metrics))
-				if err != nil {
-					return nil
-				}
-				c.appendable.UpdateChildren(newArgs.ForwardTo)
-				if c.args.CollectInterval != collectInterval {
-					t.Reset(c.args.CollectInterval)
-					collectInterval = c.args.CollectInterval
-				}
-			case <-t.C:
-				err := c.collectProfiles()
+	var (
+		sessionStarted   = false
+		sessionErrors    = 0
+		sessionMaxErrors = 3
+	)
+
+	collectInterval := c.args.CollectInterval
+	t := time.NewTicker(collectInterval)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case newArgs := <-c.argsUpdate:
+			// ensure there are no other updates queued. this might happen if the collection takes a very long time
+			newArgs = getLatestArgsFromChannel(c.argsUpdate, newArgs)
+
+			// update targets
+			c.args = newArgs
+			c.session.UpdateTargets(targetsOptionFromArgs(c.args))
+			c.metrics.targetsActive.Set(float64(len(c.targetFinder.DebugInfo())))
+			err := c.session.Update(convertSessionOptions(c.args, c.metrics))
+			if err != nil {
+				level.Error(c.options.Logger).Log("msg", "failed to update profiling session", "err", err)
+				c.reportUnhealthy(err)
+				continue
+			}
+			c.appendable.UpdateChildren(newArgs.ForwardTo)
+			if c.args.CollectInterval != collectInterval {
+				t.Reset(c.args.CollectInterval)
+				collectInterval = c.args.CollectInterval
+			}
+		case <-t.C:
+			if !sessionStarted {
+				err := c.session.Start()
 				if err != nil {
-					c.metrics.profilingSessionsFailingTotal.Inc()
-					return err
+					sessionErrors++
+					if sessionErrors > sessionMaxErrors {
+						level.Error(c.options.Logger).Log("msg", "too many errors starting profiling session, giving up", "tries", sessionErrors, "last_error", err)
+						t.Stop()
+						continue
+					}
+					level.Error(c.options.Logger).Log("msg", "failed to start profiling session", "err", err)
+					c.reportUnhealthy(err)
+					continue
 				}
-				c.updateDebugInfo()
+				sessionErrors = 0
+				defer func() {
+					c.session.Stop()
+					level.Info(c.options.Logger).Log("msg", "ebpf profiling session stopped")
+				}()
+				sessionStarted = true
+				level.Info(c.options.Logger).Log("msg", "ebpf profiling session started")
 			}
+
+			err := c.collectProfiles(ctx)
+			if err != nil {
+				level.Error(c.options.Logger).Log("msg", "failed to collect profiles", "err", err)
+				c.reportUnhealthy(err)
+				c.metrics.profilingSessionsFailingTotal.Inc()
+				continue
+			}
+			c.reportHealthy()
+			c.updateDebugInfo()
+		}
+	}
+}
+
+func getLatestArgsFromChannel[A any](ch chan A, current A) A {
+	for {
+		select {
+		case x := <-ch:
+			current = x
+		default:
+			return current
 		}
-	}, func(error) {})
-	return g.Run()
+	}
 }
 
 func (c *Component) Update(args component.Arguments) error {
@@ -154,13 +194,38 @@ func (c *Component) Update(args component.Arguments) error {
 	return nil
 }
 
+func (c *Component) reportUnhealthy(err error) {
+	c.healthMut.Lock()
+	defer c.healthMut.Unlock()
+	c.health = component.Health{
+		Health:     component.HealthTypeUnhealthy,
+		Message:    err.Error(),
+		UpdateTime: time.Now(),
+	}
+}
+
+func (c *Component) reportHealthy() {
+	c.healthMut.Lock()
+	defer c.healthMut.Unlock()
+	c.health = component.Health{
+		Health:     component.HealthTypeHealthy,
+		UpdateTime: time.Now(),
+	}
+}
+
+func (c *Component) CurrentHealth() component.Health {
+	c.healthMut.RLock()
+	defer c.healthMut.RUnlock()
+	return c.health
+}
+
 func (c *Component) DebugInfo() interface{} {
 	c.debugInfoLock.Lock()
 	defer c.debugInfoLock.Unlock()
 	return c.debugInfo
 }
 
-func (c *Component) collectProfiles() error {
+func (c *Component) collectProfiles(ctx context.Context) error {
 	c.metrics.profilingSessionsTotal.Inc()
 	level.Debug(c.options.Logger).Log("msg", "ebpf  collectProfiles")
 	args := c.args
@@ -176,6 +241,11 @@ func (c *Component) collectProfiles() error {
 	level.Debug(c.options.Logger).Log("msg", "ebpf collectProfiles done", "profiles", len(builders.Builders))
 	bytesSent := 0
 	for _, builder := range builders.Builders {
+		// check if the context is done
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+
 		serviceName := builder.Labels.Get("service_name")
 		c.metrics.pprofsTotal.WithLabelValues(serviceName).Inc()
 		c.metrics.pprofSamplesTotal.WithLabelValues(serviceName).Add(float64(len(builder.Profile.Sample)))
@@ -192,7 +262,7 @@ func (c *Component) collectProfiles() error {
 		c.metrics.pprofBytesTotal.WithLabelValues(serviceName).Add(float64(len(rawProfile)))
 
 		samples := []*pyroscope.RawSample{{RawProfile: rawProfile}}
-		err = appender.Append(context.Background(), builder.Labels, samples)
+		err = appender.Append(ctx, builder.Labels, samples)
 		if err != nil {
 			level.Error(c.options.Logger).Log("msg", "ebpf pprof write", "err", err)
 			continue

internal/component/pyroscope/ebpf/ebpf_linux_test.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"sync"
 	"testing"
 	"time"
 
@@ -26,11 +27,12 @@ import (
 )
 
 type mockSession struct {
-	options      ebpfspy.SessionOptions
-	collectError error
-	collected    int
-	data         [][]string
-	dataTarget   *sd.Target
+	options         ebpfspy.SessionOptions
+	collectCallback func() error
+	collected       int
+	data            [][]string
+	dataTarget      *sd.Target
+	mtx             sync.Mutex
 }
 
 func (m *mockSession) Start() error {
@@ -42,6 +44,8 @@ func (m *mockSession) Stop() {
 }
 
 func (m *mockSession) Update(options ebpfspy.SessionOptions) error {
+	m.mtx.Lock()
+	defer m.mtx.Unlock()
 	m.options = options
 	return nil
 }
@@ -52,8 +56,8 @@ func (m *mockSession) UpdateTargets(_ sd.TargetsOptions) {
 
 func (m *mockSession) CollectProfiles(f pprof.CollectProfilesCallback) error {
 	m.collected++
-	if m.collectError != nil {
-		return m.collectError
+	if m.collectCallback != nil {
+		return m.collectCallback()
 	}
 	for _, stack := range m.data {
 		f(
@@ -104,7 +108,7 @@ func (m *mockSession) DebugInfo() interface{} {
 	}
 }
 
-func TestShutdownOnError(t *testing.T) {
+func TestTargetUpdatesWithLongCollection(t *testing.T) {
 	logger := util.TestAlloyLogger(t)
 	ms := newMetrics(nil)
 	targetFinder, err := sd.NewTargetFinder(os.DirFS("/foo"), logger, sd.TargetsOptions{
@@ -126,9 +130,92 @@ func TestShutdownOnError(t *testing.T) {
 		ms,
 	)
 
-	session.collectError = fmt.Errorf("mocked error collecting profiles")
-	err = c.Run(t.Context())
-	require.Error(t, err)
+	collection := make(chan struct{})
+
+	// simulate a long collection
+	session.collectCallback = func() error {
+		<-collection
+		return nil
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	ctx, cancel := context.WithCancel(t.Context())
+	go func() {
+		err = c.Run(ctx)
+		require.NoError(t, err)
+		wg.Done()
+	}()
+
+	// now schedule 3 updates
+	c.Update(arguments)
+	c.Update(arguments)
+	c.Update(arguments)
+	argX := NewDefaultArguments()
+	argX.SampleRate = 1234
+	c.Update(argX)
+
+	// unblock the collection
+	close(collection)
+
+	// wait for the session to be updated
+	require.Eventually(t, func() bool {
+		session.mtx.Lock()
+		defer session.mtx.Unlock()
+		return session.options.SampleRate == 1234
+	}, time.Second*1, time.Millisecond*10)
+
+	cancel()
+	wg.Wait()
+}
+
+func TestReportingCollectError(t *testing.T) {
+	logger := util.TestAlloyLogger(t)
+	ms := newMetrics(nil)
+	targetFinder, err := sd.NewTargetFinder(os.DirFS("/foo"), logger, sd.TargetsOptions{
+		ContainerCacheSize: 1024,
+	})
+	require.NoError(t, err)
+	session := &mockSession{}
+	arguments := NewDefaultArguments()
+	arguments.CollectInterval = time.Millisecond * 100
+	c := newTestComponent(
+		component.Options{
+			Logger:        logger,
+			Registerer:    prometheus.NewRegistry(),
+			OnStateChange: func(e component.Exports) {},
+		},
+		arguments,
+		session,
+		targetFinder,
+		ms,
+	)
+
+	session.collectCallback = func() error { return fmt.Errorf("mocked error collecting profiles") }
+	var wg sync.WaitGroup
+	wg.Add(1)
+	ctx, cancel := context.WithCancel(t.Context())
+	go func() {
+		err = c.Run(ctx)
+		require.NoError(t, err)
+		wg.Done()
+	}()
+
+	// expect the component to be unhealthy
+	require.Eventually(t, func() bool {
+		if c.CurrentHealth().Health == component.HealthTypeUnhealthy {
+			require.Equal(t, c.CurrentHealth().Message, "ebpf session collectProfiles mocked error collecting profiles")
+			return true
+		}
+		return false
+	}, time.Second*1, time.Millisecond*10)
+
+	// the component should still be handling update requests
+	err = c.Update(arguments)
+	require.NoError(t, err)
+
+	cancel()
+	wg.Wait()
 }
 
 func TestContextShutdown(t *testing.T) {

internal/component/pyroscope/scrape/scrape_loop.go

@@ -168,11 +168,15 @@ func newScrapeLoop(t *Target, scrapeClient *http.Client, appendable pyroscope.Ap
 		timeout += interval - time.Second
 	}
 
+	appender := appendable.Appender()
+	if !t.godeltaprof {
+		appender = NewDeltaAppender(appender, t.allLabels)
+	}
 	return &scrapeLoop{
 		Target:       t,
 		logger:       logger,
 		scrapeClient: scrapeClient,
-		appender:     NewDeltaAppender(appendable.Appender(), t.allLabels),
+		appender:     appender,
 		interval:     interval,
 		timeout:      timeout,
 	}