Debugging gRPC Connection Leakage and HTTP/2 Stream Leaks Using eBPF and bpftrace — code snippets

snippet-1.go

package main

import (
	"context"
	"log"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	pb "google.golang.org/grpc/examples/helloworld/helloworld"
)

// LeakConnection demonstrates a connection leak where a new ClientConn is
// created on every iteration and never closed.
func LeakConnection(addr string) {
	for {
		// Bug: Dialing a new connection every iteration and not calling Close()
		conn, err := grpc.Dial(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
		if err != nil {
			log.Printf("failed to dial: %v", err)
			continue
		}
		client := pb.NewGreeterClient(conn)
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
		_, err = client.SayHello(ctx, &pb.HelloRequest{Name: "Leak"})
		cancel() // Cancels the RPC context, but the underlying TCP connection remains open
		if err != nil {
			log.Printf("RPC failed: %v", err)
		}
		time.Sleep(100 * time.Millisecond)
	}
}

snippet-2.go

package main

import (
	"context"
	"log"
	"time"

	"google.golang.org/grpc"
	pb "google.golang.org/grpc/examples/features/proto/echo"
)

// LeakStream demonstrates a stream leak where a server-streaming RPC is invoked,
// but the client abandons it. Without context cancellation, the stream remains active.
func LeakStream(client pb.EchoClient) {
	// Bug: context.Background() cannot be cancelled.
	ctx := context.Background()
	stream, err := client.ServerStreamingEcho(ctx, &pb.EchoRequest{Message: "leak-test"})
	if err != nil {
		log.Fatalf("failed to start stream: %v", err)
	}

	// Read only the first message and return, leaving the stream open
	resp, err := stream.Recv()
	if err != nil {
		log.Printf("recv err: %v", err)
		return
	}
	log.Printf("Received first message: %s", resp.GetMessage())

	// The HTTP/2 stream and Go's internal transport reader goroutines will leak
	// until the parent client connection is completely terminated.
}

snippet-3.txt

#!/usr/bin/env bpftrace

#include <net/sock.h>

BEGIN
{
  printf("Tracing TCP socket state transitions. Ctrl-C to end.\n");
  printf("%-8s %-8s %-16s %-15s %-5s %-15s %-5s %s -> %s\n", 
         "TIME", "PID", "COMM", "SADDR", "SPORT", "DADDR", "DPORT", "OLD", "NEW");
  
  @state_names[1] = "ESTABLISHED";
  @state_names[2] = "SYN_SENT";
  @state_names[3] = "SYN_RECV";
  @state_names[4] = "FIN_WAIT1";
  @state_names[5] = "FIN_WAIT2";
  @state_names[6] = "TIME_WAIT";
  @state_names[7] = "CLOSE";
  @state_names[8] = "CLOSE_WAIT";
  @state_names[9] = "LAST_ACK";
  @state_names[10] = "LISTEN";
  @state_names[11] = "CLOSING";
}

tracepoint:sock:inet_sock_set_state
{
  $sk = (struct sock *)args->skaddr;
  $family = $sk->__sk_common.skc_family;

  if ($family == AF_INET) {
    $daddr = ntop(AF_INET, args->daddr[0]);
    $saddr = ntop(AF_INET, args->saddr[0]);
    $dport = args->dport;
    $sport = args->sport;
    
    $old_str = @state_names[args->oldstate];
    $new_str = @state_names[args->newstate];

    time("%H:%M:%S ");
    printf("%-8d %-16s %-15s %-5d %-15s %-5d %s -> %s\n", 
           pid, comm, $saddr, $sport, $daddr, $dport, $old_str, $new_str);
    
    if (args->newstate == 1) { // TCP_ESTABLISHED
      @conn_start[$sk] = nsecs;
      @conn_info[$sk] = (comm, $saddr, $sport, $daddr, $dport);
    } else if (args->newstate == 7) { // TCP_CLOSE
      $start = @conn_start[$sk];
      if ($start > 0) {
        $duration_ms = (nsecs - $start) / 1000000;
        printf("--> Connection on socket %p closed after %d ms\n", $sk, $duration_ms);
        delete(@conn_start[$sk]);
        delete(@conn_info[$sk]);
      }
    }
  }
}

END
{
  printf("\n--- Leaked Sockets Remaining Open on Exit ---\n");
  for ($sk : @conn_start) {
    $info = @conn_info[$sk];
    printf("Socket %p | Comm: %-16s | %s:%d -> %s:%d | Lifetime: %d ms\n", 
           $sk, $info.0, $info.1, $info.2, $info.3, $info.4, (nsecs - @conn_start[$sk]) / 1000000);
  }
  clear(@state_names);
  clear(@conn_start);
  clear(@conn_info);
}

snippet-4.txt

#!/usr/bin/env bpftrace

tracepoint:syscalls:sys_enter_write
/comm == "grpc_client" || comm == "grpc_server"/
{
  $fd = args.fd;
  $buf = args.buf;
  $count = args.count;

  if ($count >= 9) {
    // Read the 9-byte HTTP/2 frame header from user space
    $b0 = *(uint8*)($buf + 0);
    $b1 = *(uint8*)($buf + 1);
    $b2 = *(uint8*)($buf + 2);
    $b3 = *(uint8*)($buf + 3); // Type
    $b4 = *(uint8*)($buf + 4); // Flags
    $b5 = *(uint8*)($buf + 5);
    $b6 = *(uint8*)($buf + 6);
    $b7 = *(uint8*)($buf + 7);
    $b8 = *(uint8*)($buf + 8);

    $length = ((uint32)$b0 << 16) | ((uint32)$b1 << 8) | (uint32)$b2;
    $type = (uint8)$b3;
    $flags = (uint8)$b4;
    
    // Mask out the reserved 1-bit from the stream ID
    $stream_id = (((uint32)$b5 & 0x7F) << 24) | 
                 ((uint32)$b6 << 16) | 
                 ((uint32)$b7 << 8) | 
                 (uint32)$b8;

    if ($type <= 9) { // Verify type is a standard HTTP/2 frame type (0x0 to 0x9)
      printf("PID: %-6d [%s] FD: %-3d | H2 Frame: Type=%d (", pid, comm, $fd, $type);
      if ($type == 0) { printf("DATA"); }
      else if ($type == 1) { printf("HEADERS"); }
      else if ($type == 2) { printf("PRIORITY"); }
      else if ($type == 3) { printf("RST_STREAM"); }
      else if ($type == 4) { printf("SETTINGS"); }
      else if ($type == 7) { printf("GOAWAY"); }
      else { printf("OTHER"); }
      printf(") | Flags: 0x%02x | StreamID: %d | Len: %d\n", $flags, $stream_id, $length);
    }
  }
}

snippet-5.txt

#!/usr/bin/env bpftrace

// Ensure the path points to your compiled Go binary in production
uprobe:/usr/local/bin/my_grpc_service:"google.golang.org/grpc/internal/transport.(*http2Client).NewStream"
{
  $stream_ptr = reg("dx");
  printf("Time: %s | PID: %d | Stream allocated at pointer: %p\n", usym(reg("ip")), pid, $stream_ptr);
  printf("--- Stack Trace ---\n");
  print(ustack());
  printf("-------------------\n\n");
}

snippet-6.go

package main

import (
	"context"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/keepalive"
)

// NewProductionClient constructs a gRPC client with robust keepalive settings.
func NewProductionClient(addr string) (*grpc.ClientConn, error) {
	kacp := keepalive.ClientParameters{
		Time:                10 * time.Second, // Send a ping to server if connection is idle for 10s
		Timeout:             3 * time.Second,  // Wait 3s for ping ack before closing connection
		PermitWithoutStream: true,             // Permit keepalive pings even with no active streams
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	conn, err := grpc.DialContext(ctx, addr,
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithKeepaliveParams(kacp),
		// Configure round-robin load balancing to ensure traffic spreads across pods
		grpc.WithDefaultServiceConfig(`{"loadBalancingConfig": [{"round_robin":{}}]}`),
	)
	if err != nil {
		return nil, err
	}
	return conn, nil
}

snippet-7.go

package main

import (
	"net"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/keepalive"
)

// StartProductionServer starts a hardened gRPC server.
func StartProductionServer(lis net.Listener) (*grpc.Server, error) {
	kaep := keepalive.EnforcementPolicy{
		MinTime:             5 * time.Second, // Minimum duration between client keepalive pings
		PermitWithoutStream: true,            // Allow pings when no active streams exist
	}

	kasp := keepalive.ServerParameters{
		MaxConnectionIdle:     15 * time.Minute, // Close connection if idle for 15m to free kernel resources
		MaxConnectionAge:      30 * time.Minute, // Max age of any connection before graceful termination
		MaxConnectionAgeGrace: 30 * time.Second, // Allow 30s grace period for active streams to drain
		Time:                  2 * time.Hour,    // Send keepalive ping if no activity for 2h
		Timeout:               20 * time.Second, // Wait 20s for ping response before dropping socket
	}

	server := grpc.NewServer(
		grpc.KeepaliveEnforcementPolicy(kaep),
		grpc.KeepaliveParams(kasp),
		// Hard limit of 100 concurrent streams per TCP connection to prevent OOM
		grpc.MaxConcurrentStreams(100),
	)

	return server, nil
}