Automating Real-Time Goroutine Leak Detection in Production Using eBPF and Runtime/pprof — code snippets

snippet-1.txt

// +build ignore

#include <vmlinux.h>
#include <bpf/bpf_helpers.h>
#include <bpf/bpf_tracing.h>

char LICENSE[] SEC("license") = "Dual BSD/GPL";

// Map to temporarily store callerpc between newproc1 and runqput on the same thread
struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 1024);
    __type(key, u64); // key: tgid_pid
    __type(value, u64); // value: callerpc
} pending_newproc SEC(".maps");

// Map to track the parent callerpc of every active goroutine: gp -> callerpc
struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 102400);
    __type(key, u64); // key: gp (goroutine pointer)
    __type(value, u64); // value: callerpc
} goroutine_parent SEC(".maps");

struct caller_metrics {
    u64 created_count;
    u64 active_count;
};

// Map to aggregate stats per callerpc
struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 10240);
    __type(key, u64); // key: callerpc
    __type(value, struct caller_metrics);
} caller_stats SEC(".maps");

SEC("uprobe/newproc1")
int trace_newproc1(struct pt_regs *ctx) {
    u64 tgid_pid = bpf_get_current_pid_tgid();
    // In Go ABI (amd64), 3rd arg is callerpc, passed in RCX
    u64 callerpc = ctx->cx;

    bpf_map_update_elem(&pending_newproc, &tgid_pid, &callerpc, BPF_ANY);
    return 0;
}

SEC("uprobe/runqput")
int trace_runqput(struct pt_regs *ctx) {
    u64 tgid_pid = bpf_get_current_pid_tgid();
    u64 *callerpc_ptr = bpf_map_lookup_elem(&pending_newproc, &tgid_pid);
    if (!callerpc_ptr) {
        return 0;
    }
    u64 callerpc = *callerpc_ptr;
    bpf_map_delete_elem(&pending_newproc, &tgid_pid);

    // In Go ABI (amd64), 2nd arg is gp (*g), passed in RBX
    u64 gp = ctx->bx;

    // Associate gp with its callerpc
    bpf_map_update_elem(&goroutine_parent, &gp, &callerpc, BPF_ANY);

    // Update stats for this callerpc
    struct caller_metrics *metrics = bpf_map_lookup_elem(&caller_stats, &callerpc);
    if (metrics) {
        __sync_fetch_and_add(&metrics->created_count, 1);
        __sync_fetch_and_add(&metrics->active_count, 1);
    } else {
        struct caller_metrics new_metrics = {
            .created_count = 1,
            .active_count = 1,
        };
        bpf_map_update_elem(&caller_stats, &callerpc, &new_metrics, BPF_NOEXIST);
    }

    return 0;
}

SEC("uprobe/goexit1")
int trace_goexit1(struct pt_regs *ctx) {
    // Current g pointer is in R14 on amd64
    u64 gp = ctx->r14;

    u64 *callerpc_ptr = bpf_map_lookup_elem(&goroutine_parent, &gp);
    if (!callerpc_ptr) {
        return 0;
    }
    u64 callerpc = *callerpc_ptr;
    bpf_map_delete_elem(&goroutine_parent, &gp);

    struct caller_metrics *metrics = bpf_map_lookup_elem(&caller_stats, &callerpc);
    if (metrics) {
        if (metrics->active_count > 0) {
            __sync_fetch_and_add(&metrics->active_count, -1);
        }
    }

    return 0;
}

snippet-2.go

package main

import (
	"log"
	"os"

	"github.com/cilium/ebpf/link"
	"github.com/cilium/ebpf/rlimit"
)

//go:generate go run github.com/cilium/ebpf/cmd/bpf2go -target amd64 bpf tracker.bpf.c -- -I/usr/include

func startBPFTracing() (*bpfObjects, func()) {
	// Allow the current process to lock memory for eBPF resources.
	if err := rlimit.RemoveMemlock(); err != nil {
		log.Fatalf("failed to remove memlock: %v", err)
	}

	// Load pre-compiled programs and maps into the kernel.
	objs := bpfObjects{}
	if err := loadBpfObjects(&objs, nil); err != nil {
		log.Fatalf("loading objects: %v", err)
	}

	// Open the current executable for uprobe placement
	executable, err := link.OpenExecutable("/proc/self/exe")
	if err != nil {
		log.Fatalf("failed to open executable: %v", err)
	}

	// Attach uprobes to runtime symbols
	newproc1, err := executable.Uprobe("runtime.newproc1", objs.TraceNewproc1, nil)
	if err != nil {
		log.Fatalf("failed to attach newproc1 uprobe: %v", err)
	}

	runqput, err := executable.Uprobe("runtime.runqput", objs.TraceRunqput, nil)
	if err != nil {
		log.Fatalf("failed to attach runqput uprobe: %v", err)
	}

	goexit1, err := executable.Uprobe("runtime.goexit1", objs.TraceGoexit1, nil)
	if err != nil {
		log.Fatalf("failed to attach goexit1 uprobe: %v", err)
	}

	cleanup := func() {
		newproc1.Close()
		runqput.Close()
		goexit1.Close()
		objs.Close()
	}

	return &objs, cleanup
}

snippet-3.go

package main

import (
	"log"
	"runtime"
	"time"
)

type bpfCallerMetrics struct {
	CreatedCount uint64
	ActiveCount  uint64
}

func monitorLeaks(objs *bpfObjects, leakThreshold uint64, checkInterval time.Duration, onLeakDetected func(string)) {
	ticker := time.NewTicker(checkInterval)
	defer ticker.Stop()

	// Map to track previous active counts to calculate growth rate
	history := make(map[uint64]uint64)

	for range ticker.C {
		var (
			key     uint64
			metrics bpfCallerMetrics
		)

		iter := objs.CallerStats.Iterate()
		for iter.Next(&key, &metrics) {
			prevActive, exists := history[key]
			history[key] = metrics.ActiveCount

			// Heuristic: If active count exceeds threshold and shows continuous growth
			if metrics.ActiveCount > leakThreshold && (!exists || metrics.ActiveCount > prevActive) {
				fn := runtime.FuncForPC(uintptr(key))
				name := "unknown"
				file := "unknown"
				line := 0
				if fn != nil {
					name = fn.Name()
					file, line = fn.FileLine(uintptr(key))
				}

				log.Printf("[LEAK ALERT] Potential leak detected! "+
					"CallerPC: 0x%x (%s at %s:%d) has %d active goroutines (created: %d)",
					key, name, file, line, metrics.ActiveCount, metrics.CreatedCount)

				// Trigger deep diagnostic analysis
				onLeakDetected(name)
			}
		}
		if err := iter.Err(); err != nil {
			log.Printf("error iterating caller_stats map: %v", err)
		}
	}
}

snippet-4.go

package main

import (
	"bytes"
	"fmt"
	"runtime/pprof"
	"sort"
	"strings"

	"github.com/google/pprof/profile"
)

type StackDelta struct {
	Stack      string
	DeltaCount int
}

// captureGoroutineProfile writes the current goroutine profile in binary protobuf format
func captureGoroutineProfile() (*profile.Profile, error) {
	var buf bytes.Buffer
	p := pprof.Lookup("goroutine")
	if p == nil {
		return nil, fmt.Errorf("goroutine profile not found")
	}
	// debug=0 writes binary proto format, which uses less CPU and memory to serialize
	if err := p.WriteTo(&buf, 0); err != nil {
		return nil, err
	}
	return profile.Parse(&buf)
}

// diffProfiles calculates the delta between two profiles, sorting by the highest growth
func diffProfiles(before, after *profile.Profile) []StackDelta {
	beforeStacks := aggregateStacks(before)
	afterStacks := aggregateStacks(after)

	var deltas []StackDelta
	for stack, afterCount := range afterStacks {
		beforeCount := beforeStacks[stack]
		delta := afterCount - beforeCount
		if delta > 0 {
			deltas = append(deltas, StackDelta{
				Stack:      stack,
				DeltaCount: delta,
			})
		}
	}

	sort.Slice(deltas, func(i, j int) bool {
		return deltas[i].DeltaCount > deltas[j].DeltaCount
	})

	return deltas
}

// aggregateStacks processes the locations from leaf-to-root and aggregates matching stack strings
func aggregateStacks(prof *profile.Profile) map[string]int {
	stacks := make(map[string]int)
	for _, sample := range prof.Sample {
		count := int(sample.Value[0]) // First sample value is the count of goroutines
		var callStack []string

		// Location list goes from leaf (execution point) to root (entrypoint)
		for i := len(sample.Location) - 1; i >= 0; i-- {
			loc := sample.Location[i]
			for _, line := range loc.Line {
				fnName := "unknown"
				filename := "unknown"
				lineNum := int64(0)
				if line.Function != nil {
					fnName = line.Function.Name
					filename = line.Function.Filename
					lineNum = line.Line
				}
				callStack = append(callStack, fmt.Sprintf("%s (%s:%d)", fnName, filename, lineNum))
			}
		}

		stackStr := strings.Join(callStack, "\n-> ")
		stacks[stackStr] += count
	}
	return stacks
}

snippet-5.go

package main

import (
	"context"
	"log"
	"time"
)

type LeakDetector struct {
	objs          *bpfObjects
	cleanup       func()
	threshold     uint64
	checkInterval time.Duration
	diffInterval  time.Duration
}

func NewLeakDetector(threshold uint64, checkInterval, diffInterval time.Duration) *LeakDetector {
	objs, cleanup := startBPFTracing()
	return &LeakDetector{
		objs:          objs,
		cleanup:       cleanup,
		threshold:     threshold,
		checkInterval: checkInterval,
		diffInterval:  diffInterval,
	}
}

func (ld *LeakDetector) Run(ctx context.Context) {
	defer ld.cleanup()

	ticker := time.NewTicker(ld.checkInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			ld.checkAndAnalyze()
		}
	}
}

func (ld *LeakDetector) checkAndAnalyze() {
	var (
		key     uint64
		metrics bpfCallerMetrics
	)

	iter := ld.objs.CallerStats.Iterate()
	for iter.Next(&key, &metrics) {
		if metrics.ActiveCount > ld.threshold {
			log.Printf("[ANOMALY] Active goroutines (%d) for PC 0x%x exceeded threshold (%d). Capturing baseline profile...", 
				metrics.ActiveCount, key, ld.threshold)

			before, err := captureGoroutineProfile()
			if err != nil {
				log.Printf("failed to capture baseline profile: %v", err)
				continue
			}

			// Sleep for the configured duration to allow leaking goroutines to accumulate
			time.Sleep(ld.diffInterval)

			after, err := captureGoroutineProfile()
			if err != nil {
				log.Printf("failed to capture target profile: %v", err)
				continue
			}

			deltas := diffProfiles(before, after)
			if len(deltas) == 0 {
				log.Printf("[INFO] Differential analysis complete. No growing stacks found.")
				continue
			}

			log.Printf("[LEAK CONFIRMED] Top growing stacks found:")
			for i, delta := range deltas {
				if i >= 3 { // Print top 3 leaking paths
					break
				}
				log.Printf("Stack #%d (+%d goroutines):\n%s\n", i+1, delta.DeltaCount, delta.Stack)
			}
		}
	}
}