Add support for arbitrary headers (#6080)

* Add support for arbitrary headers

* proto: Update

* .github: Update buf

* Makefile: Update buf dependency update command

* proto: Regenerate descriptor.ts after merge with main

The main branch had an outdated version of descriptor.ts that was
missing several fields from the latest protobuf descriptors. After
merging main, regenerated the proto files to bring them back in sync
with the proto definitions.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

---------

Co-authored-by: Claude <[email protected]>

Author: brancz

.github/workflows/proto-gen.yaml

@@ -45,10 +45,9 @@ jobs:
     steps:
       - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
 
-      - uses: bufbuild/buf-action@8f4a1456a0ab6a1eb80ba68e53832e6fcfacc16c # v1.3.0
+      - uses: bufbuild/buf-setup-action@a47c93e0b1648d5651a065437926377d060baa99 # v1.50.0
         with:
-          version: 1.55.1
-          setup_only: true
+          version: 1.61.0
 
       - name: Generate
         run:

.github/workflows/proto-pr.yaml

@@ -46,6 +46,8 @@ jobs:
       - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
 
       - uses: bufbuild/buf-setup-action@a47c93e0b1648d5651a065437926377d060baa99 # v1.50.0
+        with:
+          version: 1.61.0
 
       - name: version
         run: buf --version

.github/workflows/proto-push.yaml

@@ -49,6 +49,8 @@ jobs:
       - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
 
       - uses: bufbuild/buf-setup-action@a47c93e0b1648d5651a065437926377d060baa99 # v1.50.0
+        with:
+          version: 1.61.0
 
       - name: version
         run: buf --version

Makefile

@@ -140,7 +140,7 @@ proto/generate: proto/vendor
 
 .PHONY: proto/vendor
 proto/vendor: proto/google/pprof/profile.proto
-	cd proto && buf mod update
+	cd proto && buf dep update
 
 proto/google/pprof/profile.proto:
 	mkdir -p proto/google/pprof

README.md

@@ -166,6 +166,9 @@ Flags:
       --external-label=KEY=VALUE;...
                                  Label(s) to attach to all profiles in
                                  scraper-only mode.
+      --grpc-headers=KEY=VALUE;...
+                                 Additional gRPC headers to send with each
+                                 request to the remote store (key=value pairs).
 ```
 <!-- prettier-ignore-end -->
 

pkg/parca/parca.go

@@ -57,6 +57,7 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/metadata"
 	"gopkg.in/yaml.v3"
 
 	debuginfopb "github.com/parca-dev/parca/gen/proto/go/parca/debuginfo/v1alpha1"
@@ -125,6 +126,7 @@ type Flags struct {
 	Insecure           bool              `kong:"help='Send gRPC requests via plaintext instead of TLS.'"`
 	InsecureSkipVerify bool              `kong:"help='Skip TLS certificate verification.'"`
 	ExternalLabel      map[string]string `kong:"help='Label(s) to attach to all profiles in scraper-only mode.'"`
+	GRPCHeaders        map[string]string `kong:"help='Additional gRPC headers to send with each request to the remote store (key=value pairs).'"`
 
 	Hidden FlagsHidden `embed:"" prefix:""`
 }
@@ -661,6 +663,28 @@ func Run(ctx context.Context, logger log.Logger, reg *prometheus.Registry, flags
 	return nil
 }
 
+// customHeadersUnaryInterceptor adds custom headers to all unary RPC calls.
+func customHeadersUnaryInterceptor(headers map[string]string) grpc.UnaryClientInterceptor {
+	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+		// Add headers to outgoing context
+		for key, value := range headers {
+			ctx = metadata.AppendToOutgoingContext(ctx, key, value)
+		}
+		return invoker(ctx, method, req, reply, cc, opts...)
+	}
+}
+
+// customHeadersStreamInterceptor adds custom headers to all streaming RPC calls.
+func customHeadersStreamInterceptor(headers map[string]string) grpc.StreamClientInterceptor {
+	return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
+		// Add headers to outgoing context
+		for key, value := range headers {
+			ctx = metadata.AppendToOutgoingContext(ctx, key, value)
+		}
+		return streamer(ctx, desc, cc, method, opts...)
+	}
+}
+
 func runForwarder(
 	ctx context.Context,
 	logger log.Logger,
@@ -686,17 +710,27 @@ func runForwarder(
 
 	propagators := propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{})
 
+	// Build interceptor chains
+	unaryInterceptors := []grpc.UnaryClientInterceptor{
+		metrics.UnaryClientInterceptor(),
+	}
+	streamInterceptors := []grpc.StreamClientInterceptor{
+		metrics.StreamClientInterceptor(),
+	}
+
+	// Add custom headers interceptor if headers are configured
+	if len(flags.GRPCHeaders) > 0 {
+		unaryInterceptors = append([]grpc.UnaryClientInterceptor{customHeadersUnaryInterceptor(flags.GRPCHeaders)}, unaryInterceptors...)
+		streamInterceptors = append([]grpc.StreamClientInterceptor{customHeadersStreamInterceptor(flags.GRPCHeaders)}, streamInterceptors...)
+	}
+
 	opts := []grpc.DialOption{
 		grpc.WithStatsHandler(otelgrpc.NewServerHandler(
 			otelgrpc.WithTracerProvider(tracer),
 			otelgrpc.WithPropagators(propagators),
 		)),
-		grpc.WithChainUnaryInterceptor(
-			metrics.UnaryClientInterceptor(),
-		),
-		grpc.WithChainStreamInterceptor(
-			metrics.StreamClientInterceptor(),
-		),
+		grpc.WithChainUnaryInterceptor(unaryInterceptors...),
+		grpc.WithChainStreamInterceptor(streamInterceptors...),
 	}
 	if flags.Insecure {
 		opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))