Detecting File Descriptor Exhaustion in gRPC Servers Using eBPF and OpenTelemetry Metrics — code snippets

snippet-1.go

package main

import (
	"context"
	"log"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	pb "github.com/ashari/grpc-otel-ebpf/proto"
)

// BadPracticeLeakChannel illustrates a common anti-pattern where a gRPC connection
// is established on every request instead of reusing a single client channel.
// This quickly exhausts available file descriptors under high concurrency.
func BadPracticeLeakChannel(ctx context.Context, target string) {
	// INCORRECT: Creating connection on every call. Even if closed, the underlying
	// TCP socket stays in TIME_WAIT, or if close is forgotten, it leaks immediately.
	conn, err := grpc.DialContext(ctx, target, 
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithBlock(),
	)
	if err != nil {
		log.Printf("failed to connect: %v", err)
		return
	}
	
	// If deferred close is omitted or fails due to context cancellation, FD is leaked.
	defer conn.Close() 

	client := pb.NewUserServiceClient(conn)
	_, err = client.GetUserProfile(ctx, &pb.UserRequest{UserId: "usr_100"})
	if err != nil {
		log.Printf("RPC failed: %v", err)
	}
}

// GoodPracticeReusableChannel demonstrates the correct pattern using a singleton client.
type SafeClient struct {
	conn   *grpc.ClientConn
	client pb.UserServiceClient
}

func NewSafeClient(target string) (*SafeClient, error) {
	conn, err := grpc.Dial(target, 
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithKeepaliveParams(grpc.KeepaliveParams{
			Time:                10 * time.Second,
			Timeout:             3 * time.Second,
			PermitWithoutStream: true,
		}),
	)
	if err != nil {
		return nil, err
	}
	return &SafeClient{
		conn:   conn,
		client: pb.NewUserServiceClient(conn),
	}, nil
}

snippet-2.txt

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

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

// Map to track the count of active file descriptors per TGID (Thread Group ID / Process ID)
struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 10240);
    __type(key, u32);   // Key: TGID (PID in userspace)
    __type(value, u64); // Value: Active FD Count
} fd_count_map SEC(".maps");

// Helper function to atomically adjust the FD count for a given process ID
static __always_inline void update_fd_count(u32 tgid, s64 delta) {
    u64 *count = bpf_map_lookup_elem(&fd_count_map, &tgid);
    if (!count) {
        if (delta > 0) {
            u64 val = delta;
            bpf_map_update_elem(&fd_count_map, &tgid, &val, BPF_ANY);
        }
    } else {
        if (delta < 0 && *count < (u64)(-delta)) {
            u64 val = 0;
            bpf_map_update_elem(&fd_count_map, &tgid, &val, BPF_ANY);
        } else {
            u64 val = *count + delta;
            bpf_map_update_elem(&fd_count_map, &tgid, &val, BPF_ANY);
        }
    }
}

// Hook into the exit tracepoint of accept4.
// If the return value is positive, it is a newly allocated file descriptor.
SEC("tracepoint/syscalls/sys_exit_accept4")
int tracepoint_sys_exit_accept4(struct trace_event_raw_sys_exit *ctx) {
    s64 ret = ctx->ret;
    if (ret < 0) {
        return 0; // Failed syscall, no FD allocated
    }
    u64 id = bpf_get_current_pid_tgid();
    u32 tgid = id >> 32; // Extract the TGID (main process ID)
    
    update_fd_count(tgid, 1);
    return 0;
}

// Hook into the exit tracepoint of socket.
SEC("tracepoint/syscalls/sys_exit_socket")
int tracepoint_sys_exit_socket(struct trace_event_raw_sys_exit *ctx) {
    s64 ret = ctx->ret;
    if (ret < 0) {
        return 0;
    }
    u64 id = bpf_get_current_pid_tgid();
    u32 tgid = id >> 32;
    
    update_fd_count(tgid, 1);
    return 0;
}

// Hook into the entry tracepoint of close.
SEC("tracepoint/syscalls/sys_enter_close")
int tracepoint_sys_enter_close(struct trace_event_raw_sys_enter *ctx) {
    u64 id = bpf_get_current_pid_tgid();
    u32 tgid = id >> 32;
    
    update_fd_count(tgid, -1);
    return 0;
}

snippet-3.go

package main

import (
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

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

// Assuming standard go generate bindings or hand-loaded specifications:
//go:generate go run github.com/cilium/ebpf/cmd/bpf2go -target amd64 -type event bpf fd_tracker.c

func loadAndReadEbpfMaps() {
	// 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 eBPF objects (generated via bpf2go)
	var objs bpfObjects
	if err := loadBpfObjects(&objs, nil); err != nil {
		log.Fatalf("loading objects: %v", err)
	}
	defer objs.Close()

	// Attach tracepoints to trace connection cycles
	tpAccept, err := link.Tracepoint("syscalls", "sys_exit_accept4", objs.TracepointSysExitAccept4, nil)
	if err != nil {
		log.Fatalf("opening tracepoint sys_exit_accept4: %v", err)
	}
	defer tpAccept.Close()

	tpSocket, err := link.Tracepoint("syscalls", "sys_exit_socket", objs.TracepointSysExitSocket, nil)
	if err != nil {
		log.Fatalf("opening tracepoint sys_exit_socket: %v", err)
	}
	defer tpSocket.Close()

	tpClose, err := link.Tracepoint("syscalls", "sys_enter_close", objs.TracepointSysEnterClose, nil)
	if err != nil {
		log.Fatalf("opening tracepoint sys_enter_close: %v", err)
	}
	defer tpClose.Close()

	log.Println("eBPF probes successfully attached. Polling maps...")

	// Periodically pull data from maps for illustration
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()

	stopChan := make(chan os.Signal, 1)
	signal.Notify(stopChan, os.Interrupt, syscall.SIGTERM)

	for {
		select {
		case <-ticker.C:
			var targetTgid uint32 = uint32(os.Getpid()) // Monitor self as an example
			var fdCount uint64
			err := objs.FdCountMap.Lookup(targetTgid, &fdCount)
			if err != nil {
				log.Printf("no FD entry in map for TGID %d yet", targetTgid)
			} else {
				log.Printf("TGID %d active sockets/FDs tracked by eBPF: %d", targetTgid, fdCount)
			}
		case <-stopChan:
			log.Println("Terminating loader...")
			return
		}
	}
}

snippet-4.go

package main

import (
	"context"
	"os"

	"github.com/cilium/ebpf"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/metric"
)

type EbpfCollector struct {
	fdMap       *ebpf.Map
	targetTgid  uint32
	fdGauge     metric.Int64ObservableGauge
}

// NewEbpfCollector configures the OTel metric callback tied to the kernel map
func NewEbpfCollector(meter metric.Meter, fdMap *ebpf.Map) (*EbpfCollector, error) {
	collector := &EbpfCollector{
		fdMap:      fdMap,
		targetTgid: uint32(os.Getpid()),
	}

	// Register an asynchronous observable gauge metric
	fdGauge, err := meter.Int64ObservableGauge(
		"process.grpc.active_ebpf_fds",
		metric.WithDescription("Number of active file descriptors (sockets) tracked by eBPF"),
		metric.WithUnit("1"),
	)
	if err != nil {
		return nil, err
	}

	collector.fdGauge = fdGauge

	// Register the callback to fetch data from the eBPF map on collection
	_, err = meter.RegisterCallback(func(ctx context.Context, observer metric.Observer) error {
		var count uint64
		err := collector.fdMap.Lookup(collector.targetTgid, &count)
		if err != nil {
			// If not found in map (no sockets created yet), report 0
			count = 0
		}
		
		observer.ObserveInt64(
			collector.fdGauge,
			int64(count),
			metric.WithAttributes(
				attribute.Int("pid", int(collector.targetTgid)),
				attribute.String("service.name", "grpc-payment-service"),
				attribute.String("environment", "production"),
			),
		)
		return nil
	}, fdGauge)

	return collector, err
}