From fb06cfd9eedf66316d485af10e020abab181f8f4 Mon Sep 17 00:00:00 2001 From: Frederic Branczyk Date: Fri, 30 Oct 2026 10:44:44 +0000 Subject: [PATCH 1/2] feat: add OTLP profile forwarding mode Convert scraped pprof profiles with the OpenTelemetry translator and export them over OTLP/gRPC. Preserve target labels as resource attributes and pprof labels as sample attributes, with semantic-convention configuration examples. Amp-Thread-ID: https://ampcode.com/threads/T-01a18c4e-d681-71f4-8158-32d8026e071b Co-authored-by: Amp --- README.md | 39 +++++++- deploy/otlp/kubernetes.yaml | 44 ++++++++ deploy/otlp/static.yaml | 30 ++++++ go.mod | 6 ++ go.sum | 24 ++++- pkg/parca/parca.go | 28 +++--- pkg/parca/parca_test.go | 100 +++++++++++++++++++ pkg/profilestore/grpc.go | 85 ++++++++++++++-- pkg/profilestore/grpc_test.go | 182 ++++++++++++++++++++++++++++++++++ pkg/scrape/target_test.go | 52 ++++++++++ 10 files changed, 565 insertions(+), 25 deletions(-) create mode 100644 deploy/otlp/kubernetes.yaml create mode 100644 deploy/otlp/static.yaml create mode 100644 pkg/profilestore/grpc_test.go diff --git a/README.md b/README.md index 2cfe496e5ed..0a523bf7f1c 100644 --- a/README.md +++ b/README.md @@ -74,6 +74,35 @@ The scrape configuration can be changed in the `parca.yaml` in the root of the r ### Configuration +#### Forwarding profiles with OTLP + +Use `--mode=otlp-forwarder --store-address=` to scrape pprof endpoints, +convert the profiles with OpenTelemetry's pprof translator, and export them to an +upstream that supports OTLP profiles over gRPC. No local profile store is started. +The existing `--bearer-token`, `--bearer-token-file`, `--grpc-headers`, and TLS flags +apply to this connection. Use `--insecure` for a plaintext upstream. + +All target labels retained after relabeling and any `--external-label` values +become OTLP resource attributes, without renaming. Labels inside the pprof data +remain attributes on the OTLP samples. Relabeling changes only target metadata, +not labels within profiles. As with other scraping modes, discovery's internal +`__meta_*` labels must be copied by relabeling to retain them. + +Example configs map target metadata to OpenTelemetry semantic-convention names: + +- [Static targets](deploy/otlp/static.yaml) sets `service.name`, + `service.namespace`, `service.instance.id`, and `deployment.environment.name`. +- [Kubernetes pods](deploy/otlp/kubernetes.yaml) sets service attributes and + `k8s.*` attributes from pod discovery metadata. + +Run either example with `--config-path=deploy/otlp/.yaml` and the mode +and store address above. Parca's `job`, `instance`, and `__name__` labels are also +forwarded as resource attributes for compatibility. The examples do not rename +pprof sample types or units. The upstream must support the OTLP profiles +`v1development` protocol used by this build. OTLP forwarding supports pprof +`WriteRaw` requests and OTLP `Export`, but not Parca's Arrow or debuginfo APIs. +The existing `scraper-only` and `forwarder` modes continue to use Parca's protocol. + Flags: @@ -85,8 +114,10 @@ Flags: -h, --help Show context-sensitive help. --config-path="parca.yaml" Path to config file. - --mode="all" Scraper only runs a scraper that sends to a - remote gRPC endpoint. All runs all components. + --mode="all" All runs all components. Scraper-only and + forwarder send profiles to a remote Parca gRPC + endpoint. OTLP-forwarder converts pprof profiles + and sends them to an OTLP profiles gRPC endpoint. --http-address=":7070" Address to bind HTTP server to. --http-read-timeout=5s Timeout duration for HTTP server to read request body. @@ -164,8 +195,8 @@ Flags: TLS. --insecure-skip-verify Skip TLS certificate verification. --external-label=KEY=VALUE;... - Label(s) to attach to all profiles in - scraper-only mode. + Label(s) to attach to all scraped profiles in + forwarding modes. --grpc-headers=KEY=VALUE;... Additional gRPC headers to send with each request to the remote store (key=value pairs). diff --git a/deploy/otlp/kubernetes.yaml b/deploy/otlp/kubernetes.yaml new file mode 100644 index 00000000000..ad8e48e4667 --- /dev/null +++ b/deploy/otlp/kubernetes.yaml @@ -0,0 +1,44 @@ +# Run in-cluster with --mode=otlp-forwarder and --store-address=. +# The service account needs permission to get, list, and watch pods. +# Scrape containers that expose a port named "pprof" and whose pods have an +# app.kubernetes.io/name label. Endpoints must serve Go's /debug/pprof paths. +# Required by config validation, but unused in forwarding modes. +object_storage: + bucket: + type: FILESYSTEM + config: + directory: ./data + +scrape_configs: + - job_name: kubernetes-pods + scrape_interval: 30s + kubernetes_sd_configs: + - role: pod + relabel_configs: + - action: keep + source_labels: [__meta_kubernetes_pod_container_port_name] + regex: pprof + - action: keep + source_labels: [__meta_kubernetes_pod_phase] + regex: Running + - action: keep + source_labels: [__meta_kubernetes_pod_label_app_kubernetes_io_name] + regex: .+ + - source_labels: [__meta_kubernetes_pod_label_app_kubernetes_io_name] + target_label: service.name + - source_labels: [__meta_kubernetes_namespace] + target_label: service.namespace + # A pod UID plus container name distinguishes replicas and containers. + - source_labels: [__meta_kubernetes_pod_uid, __meta_kubernetes_pod_container_name] + separator: / + target_label: service.instance.id + - source_labels: [__meta_kubernetes_namespace] + target_label: k8s.namespace.name + - source_labels: [__meta_kubernetes_pod_name] + target_label: k8s.pod.name + - source_labels: [__meta_kubernetes_pod_uid] + target_label: k8s.pod.uid + - source_labels: [__meta_kubernetes_pod_node_name] + target_label: k8s.node.name + - source_labels: [__meta_kubernetes_pod_container_name] + target_label: k8s.container.name diff --git a/deploy/otlp/static.yaml b/deploy/otlp/static.yaml new file mode 100644 index 00000000000..27d906fdf1a --- /dev/null +++ b/deploy/otlp/static.yaml @@ -0,0 +1,30 @@ +# Run with --mode=otlp-forwarder --store-address=collector.example.com:4317. +# Add --insecure only if the upstream uses plaintext gRPC. +# Required by config validation, but unused in forwarding modes. +object_storage: + bucket: + type: FILESYSTEM + config: + directory: ./data + +scrape_configs: + - job_name: checkout + scrape_interval: 30s + static_configs: + - targets: [checkout:6060] + labels: + namespace: shop + environment: development + relabel_configs: + - source_labels: [job] + target_label: service.name + # The default instance label is assigned AFTER relabeling, so use + # __address__ to identify the service instance here. + - source_labels: [__address__] + target_label: service.instance.id + - source_labels: [namespace] + target_label: service.namespace + - source_labels: [environment] + target_label: deployment.environment.name + - action: labeldrop + regex: namespace|environment diff --git a/go.mod b/go.mod index 8b32f5e5904..661ec51083d 100644 --- a/go.mod +++ b/go.mod @@ -34,6 +34,7 @@ require ( github.com/nanmu42/limitio v1.0.0 github.com/oklog/run v1.2.0 github.com/olekukonko/tablewriter v0.0.5 + github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/pprof v0.157.0 github.com/parquet-go/parquet-go v0.24.0 github.com/pierrec/lz4/v4 v4.1.25 github.com/planetscale/vtprotobuf v0.6.1-0.20250313105119-ba97887b0a25 @@ -45,6 +46,7 @@ require ( github.com/stretchr/testify v1.12.1 github.com/thanos-io/objstore v0.0.0-20241028145108-31c0873d5266 github.com/zeebo/xxh3 v1.1.0 + go.opentelemetry.io/collector/pdata/pprofile v0.157.0 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.70.0 go.opentelemetry.io/contrib/instrumentation/net/http/httptrace/otelhttptrace v0.67.0 go.opentelemetry.io/otel v1.46.0 @@ -196,6 +198,7 @@ require ( github.com/hashicorp/go-multierror v1.1.1 // indirect github.com/hashicorp/go-retryablehttp v0.7.8 // indirect github.com/hashicorp/go-rootcerts v1.0.2 // indirect + github.com/hashicorp/go-version v1.9.0 // indirect github.com/hashicorp/golang-lru v1.0.2 // indirect github.com/hashicorp/nomad/api v0.0.0-20260324203407-b27b0c2e019a // indirect github.com/hashicorp/serf v0.10.1 // indirect @@ -255,10 +258,13 @@ require ( github.com/x448/float16 v0.8.4 // indirect go.etcd.io/bbolt v1.3.7 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/collector/featuregate v1.63.0 // indirect + go.opentelemetry.io/collector/pdata v1.63.0 // indirect go.opentelemetry.io/contrib/detectors/gcp v1.44.0 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.70.0 // indirect go.opentelemetry.io/otel/metric v1.46.0 // indirect go.opentelemetry.io/otel/sdk/metric v1.46.0 // indirect + go.uber.org/multierr v1.11.0 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect go.yaml.in/yaml/v3 v3.0.5 // indirect golang.org/x/crypto v0.57.0 // indirect diff --git a/go.sum b/go.sum index d0260d3de08..29c619c2c94 100644 --- a/go.sum +++ b/go.sum @@ -542,8 +542,8 @@ github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/b github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/Co8= github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= github.com/hashicorp/go-version v1.2.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA= -github.com/hashicorp/go-version v1.8.0 h1:KAkNb1HAiZd1ukkxDFGmokVZe1Xy9HG6NUp+bPle2i4= -github.com/hashicorp/go-version v1.8.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA= +github.com/hashicorp/go-version v1.9.0 h1:CeOIz6k+LoN3qX9Z0tyQrPtiB1DFYRPfCIBtaXPSCnA= +github.com/hashicorp/go-version v1.9.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA= github.com/hashicorp/go.net v0.0.1/go.mod h1:hjKkEWcCURg++eb33jQU7oqQcI9XDCnUzHA0oac0k90= github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= @@ -745,6 +745,10 @@ github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1Cpa github.com/onsi/gomega v1.38.2 h1:eZCjf2xjZAqe+LeWvKb5weQ+NcPwX84kqJ0cZNxok2A= github.com/onsi/gomega v1.38.2/go.mod h1:W2MJcYxRGV63b418Ai34Ud0hEdTVXq9NW9+Sx6uXf3k= github.com/op/go-logging v0.0.0-20160315200505-970db520ece7/go.mod h1:HzydrMdWErDVzsI23lYNej1Htcns9BCg93Dk0bBINWk= +github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/pprof v0.157.0 h1:JrYk+H65J18Gn6JXf5zjNQkCv4HoU7LRHm6nSz6qOVs= +github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/pprof v0.157.0/go.mod h1:Xb2YX/ehDKQHwhpwSkTCe0t6IBDv406KTA3eMpPY8Lw= +github.com/open-telemetry/sig-profiling/profcheck v0.0.0-20260605055552-091960d5da90 h1:Pvu/hHr3XjXFlWQlNk/+dG69b0R1ScGWk8Bgm29lP2g= +github.com/open-telemetry/sig-profiling/profcheck v0.0.0-20260605055552-091960d5da90/go.mod h1:WPdgk1BinVdvYdkbt1KIkgDw6qUiK8CnGq+ftaJ41Ns= github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040= @@ -964,6 +968,14 @@ go.opencensus.io v0.20.2/go.mod h1:6WKK9ahsWS3RSO+PY9ZHZUfv2irvY6gN279GOPZjmmk= go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/collector/featuregate v1.63.0 h1:6EWX1C5AtmIh8hFH97DwK6R7R8Jk3fTLxAUfZPXGutY= +go.opentelemetry.io/collector/featuregate v1.63.0/go.mod h1:4ga1QBMPEejXXmpyJS8lmaRpknJ3Lb9Bvk6e420bUFU= +go.opentelemetry.io/collector/internal/testutil v0.157.0 h1:plojUQwFC5l1ex9KUDaLmCFY/mTxEmf3zrlP7M23IEw= +go.opentelemetry.io/collector/internal/testutil v0.157.0/go.mod h1:Jkjs6rkqs973LqgZ0Fe3zrokQRKULYXPIf4HuqStiEE= +go.opentelemetry.io/collector/pdata v1.63.0 h1:fY2xSG2MnyoBwA4GUhzoogGZMuNS0qHpCoODqaKwiVQ= +go.opentelemetry.io/collector/pdata v1.63.0/go.mod h1:jzozYYhQEkTQ/CCbCBNC+hYUeju9S2J8HIqIDHdxZWk= +go.opentelemetry.io/collector/pdata/pprofile v0.157.0 h1:YRTPhwWzdG0pfJmb8p/qpQm1EdX+JfV20qzwG3ypDqI= +go.opentelemetry.io/collector/pdata/pprofile v0.157.0/go.mod h1:kwy/ufNUBkw8PsFPQnAqCvD12OpGU8h9A1cz5S7xS6g= go.opentelemetry.io/contrib/detectors/gcp v1.44.0 h1:NmLfL734pJhM0JKaYd2Y28+nY9dPRWYAAbxhRCrKXPw= go.opentelemetry.io/contrib/detectors/gcp v1.44.0/go.mod h1:tNAsgd8avTGke1+MndXlU5Cru4PQ9Ai/cCNWQv/ZJ/s= go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.70.0 h1:oECp5f+hN7nkwjU/8BxQ/q23bGPb8FIrD839owX222E= @@ -1000,6 +1012,12 @@ go.opentelemetry.io/proto/otlp/collector/profiles/v1development v0.3.0 h1:AWBLtJ go.opentelemetry.io/proto/otlp/collector/profiles/v1development v0.3.0/go.mod h1:kCdbxyM9kfmv3v4ZQzZ2pACxoQ6lE0IsjP+UcZwQnr4= go.opentelemetry.io/proto/otlp/profiles/v1development v0.3.0 h1:ZQs05qo3Yh4KUHeVH6v89xErwmsvgA/cLX2/w5Ikp+k= go.opentelemetry.io/proto/otlp/profiles/v1development v0.3.0/go.mod h1:3iiRVKaCfVo0UI1ZaSMm5WbCBbINRqVlD9SUmvyBNrY= +go.opentelemetry.io/proto/slim/otlp v1.10.0 h1:iR97Vs/ZDR+y9TfuP9b1XBtdPWeC+OMslIBmhcLU7jM= +go.opentelemetry.io/proto/slim/otlp v1.10.0/go.mod h1:lV9250stpjYLPNA5viFabIgP2QlUGRT1GdTgAf8SIUk= +go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0 h1:RUF5rO0hAlgiJt1fzQVzcVs3vZVNHIcMLgOgG4rWNcQ= +go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0/go.mod h1:I89cynRj8y+383o7tEQVg2SVA6SRgDVIouWPUVXjx0U= +go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0 h1:CQvJSldHRUN6Z8jsUeYv8J0lXRvygALXIzsmAeCcZE0= +go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0/go.mod h1:xSQ+mEfJe/GjK1LXEyVOoSI1N9JV9ZI923X5kup43W4= go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ= @@ -1011,6 +1029,8 @@ go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0= go.uber.org/multierr v1.3.0/go.mod h1:VgVr7evmIr6uPjLBxg28wmKNXyqE9akIJ5XnfpiKl+4= +go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= +go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee/go.mod h1:vJERXedbb3MVM5f9Ejo0C68/HhF8uaILCdgjnY+goOA= go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q= go.uber.org/zap v1.13.0/go.mod h1:zwrFLgMcdUuIBviXEYEH1YKNaOBnKXsx2IPda5bBwHM= diff --git a/pkg/parca/parca.go b/pkg/parca/parca.go index 3b00b90f046..ec99d61ba5c 100644 --- a/pkg/parca/parca.go +++ b/pkg/parca/parca.go @@ -90,12 +90,13 @@ const ( symbolizationInterval = 10 * time.Second flagModeScraperOnly = "scraper-only" flagModeForwarder = "forwarder" + flagModeOTLPForwarder = "otlp-forwarder" metaStoreBadger = "badger" ) type Flags struct { ConfigPath string `default:"parca.yaml" help:"Path to config file."` - Mode string `default:"all" enum:"all,scraper-only,forwarder" help:"Scraper only runs a scraper that sends to a remote gRPC endpoint. All runs all components."` + Mode string `default:"all" enum:"all,scraper-only,forwarder,otlp-forwarder" help:"All runs all components. Scraper-only and forwarder send profiles to a remote Parca gRPC endpoint. OTLP-forwarder converts pprof profiles and sends them to an OTLP profiles gRPC endpoint."` HTTPAddress string `default:":7070" help:"Address to bind HTTP server to."` HTTPReadTimeout time.Duration `default:"5s" help:"Timeout duration for HTTP server to read request body."` HTTPWriteTimeout time.Duration `default:"1m" help:"Timeout duration for HTTP server to write response body."` @@ -127,7 +128,7 @@ type Flags struct { BearerTokenFile string `kong:"help='File to read bearer token from to authenticate with store.'"` 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.'"` + ExternalLabel map[string]string `kong:"help='Label(s) to attach to all scraped profiles in forwarding modes.'"` 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:""` @@ -243,13 +244,12 @@ func Run(ctx context.Context, logger log.Logger, reg *prometheus.Registry, flags return fmt.Errorf("failed to initialize UI filesystem: %w", err) } - if flags.StoreAddress != "" && flags.Mode != flagModeScraperOnly { - return fmt.Errorf("the mode should be set as `--mode=scraper-only`, if `StoreAddress` is set") - } - - if flags.Mode == flagModeScraperOnly || flags.Mode == flagModeForwarder { + if flags.Mode == flagModeScraperOnly || flags.Mode == flagModeForwarder || flags.Mode == flagModeOTLPForwarder { return runForwarder(ctx, logger, reg, tracerProvider, uiFS, flags, version, cfg) } + if flags.StoreAddress != "" { + return fmt.Errorf("--store-address requires --mode=scraper-only, --mode=forwarder, or --mode=otlp-forwarder") + } bucketCfg, err := yaml.Marshal(cfg.ObjectStorage.Bucket) if err != nil { @@ -856,13 +856,14 @@ func runForwarder( if err != nil { return fmt.Errorf("failed to create gRPC connection: %w", err) } + defer conn.Close() dbginfo := debuginfo.NewGRPCForwarder(debuginfopb.NewDebuginfoServiceClient(conn)) client := profilestore.NewClient( profilestorepb.NewProfileStoreServiceClient(conn), otelgrpcprofilingpb.NewProfilesServiceClient(conn), ) - store := profilestore.NewGRPCForwarder(client, logger) + store := profilestore.NewGRPCForwarder(client, logger, flags.Mode == flagModeOTLPForwarder) sdMetrics, err := discovery.CreateAndRegisterSDMetrics(reg) if err != nil { @@ -954,10 +955,13 @@ func runForwarder( server.RegisterableFunc(func(ctx context.Context, srv *grpc.Server, mux *runtime.ServeMux, endpoint string, opts []grpc.DialOption) error { scrapepb.RegisterScrapeServiceServer(srv, m) profilestorepb.RegisterProfileStoreServiceServer(srv, store) - debuginfopb.RegisterDebuginfoServiceServer(srv, dbginfo) - - if err := debuginfopb.RegisterDebuginfoServiceHandlerFromEndpoint(ctx, mux, endpoint, opts); err != nil { - return err + if flags.Mode == flagModeOTLPForwarder { + otelgrpcprofilingpb.RegisterProfilesServiceServer(srv, store) + } else { + debuginfopb.RegisterDebuginfoServiceServer(srv, dbginfo) + if err := debuginfopb.RegisterDebuginfoServiceHandlerFromEndpoint(ctx, mux, endpoint, opts); err != nil { + return err + } } if err := scrapepb.RegisterScrapeServiceHandlerFromEndpoint(ctx, mux, endpoint, opts); err != nil { diff --git a/pkg/parca/parca_test.go b/pkg/parca/parca_test.go index 7baf8953a29..a7614879473 100644 --- a/pkg/parca/parca_test.go +++ b/pkg/parca/parca_test.go @@ -21,8 +21,13 @@ import ( "errors" "io" "math" + "net" + "net/http" + "net/http/httptest" "os" "os/exec" + "path/filepath" + "strings" "testing" "time" @@ -37,9 +42,11 @@ import ( "github.com/prometheus/prometheus/model/timestamp" "github.com/stretchr/testify/require" "go.opentelemetry.io/otel/trace/noop" + otlppb "go.opentelemetry.io/proto/otlp/collector/profiles/v1development" "google.golang.org/grpc" "google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/metadata" "google.golang.org/protobuf/types/known/timestamppb" profilestorepb "github.com/parca-dev/parca/gen/proto/go/parca/profilestore/v1alpha1" @@ -499,3 +506,96 @@ func TestLabels(t *testing.T) { want := map[string]struct{}{} require.Equal(t, want, got, "profile should contain labels from the original profile only") } + +func TestOTLPForwarderScrape(t *testing.T) { + var data bytes.Buffer + loc := &profile.Location{ID: 1, Address: 0x1234} + p := &profile.Profile{ + SampleType: []*profile.ValueType{{Type: "inuse_space", Unit: "bytes"}}, + Sample: []*profile.Sample{{ + Location: []*profile.Location{loc}, Value: []int64{4096}, + Label: map[string][]string{"service.name": {"sample-service"}}, + }}, + Location: []*profile.Location{loc}, TimeNanos: 1700000000000000000, + } + require.NoError(t, p.Write(&data)) + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write(data.Bytes()) + })) + t.Cleanup(target.Close) + + received := make(chan struct { + profiles *otlppb.ExportProfilesServiceRequest + headers metadata.MD + }, 1) + upstream := grpc.NewServer(grpc.UnaryInterceptor(func(ctx context.Context, req interface{}, _ *grpc.UnaryServerInfo, _ grpc.UnaryHandler) (interface{}, error) { + headers, _ := metadata.FromIncomingContext(ctx) + select { + case received <- struct { + profiles *otlppb.ExportProfilesServiceRequest + headers metadata.MD + }{req.(*otlppb.ExportProfilesServiceRequest), headers}: + default: + } + return &otlppb.ExportProfilesServiceResponse{}, nil + })) + otlppb.RegisterProfilesServiceServer(upstream, &otlppb.UnimplementedProfilesServiceServer{}) + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + go func() { _ = upstream.Serve(listener) }() + t.Cleanup(upstream.Stop) + + // Exercise the shipped static example through discovery, relabeling, + // scraping, conversion, and an upstream that implements only OTLP. + contents, err := os.ReadFile("../../deploy/otlp/static.yaml") + require.NoError(t, err) + contents = bytes.ReplaceAll(contents, []byte("checkout:6060"), []byte(strings.TrimPrefix(target.URL, "http://"))) + contents = bytes.ReplaceAll(contents, []byte("scrape_interval: 30s"), []byte("scrape_interval: 1s")) + configPath := filepath.Join(t.TempDir(), "parca.yaml") + require.NoError(t, os.WriteFile(configPath, contents, 0o600)) + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan error, 1) + go func() { + done <- Run(ctx, log.NewNopLogger(), prometheus.NewRegistry(), &Flags{ + ConfigPath: configPath, Mode: flagModeOTLPForwarder, HTTPAddress: "127.0.0.1:0", + StoreAddress: listener.Addr().String(), Insecure: true, BearerToken: "test-token", + GRPCHeaders: map[string]string{"x-tenant": "test-tenant"}, + ExternalLabel: map[string]string{"k8s.cluster.name": "test-cluster"}, + }, "test-version") + }() + t.Cleanup(func() { + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + require.NoError(t, err) + } + case <-time.After(10 * time.Second): + t.Error("forwarder did not stop") + } + }) + select { + case result := <-received: + require.Equal(t, []string{"Bearer test-token"}, result.headers.Get("authorization")) + require.Equal(t, []string{"test-tenant"}, result.headers.Get("x-tenant")) + require.Len(t, result.profiles.ResourceProfiles, 1) + resource := result.profiles.ResourceProfiles[0] + attrs := map[string]string{} + for _, attr := range resource.Resource.Attributes { + attrs[attr.Key] = attr.Value.GetStringValue() + } + require.Equal(t, "checkout", attrs["service.name"]) + require.Equal(t, "shop", attrs["service.namespace"]) + require.Equal(t, strings.TrimPrefix(target.URL, "http://"), attrs["service.instance.id"]) + require.Equal(t, "development", attrs["deployment.environment.name"]) + require.Equal(t, "test-cluster", attrs["k8s.cluster.name"]) + sample := resource.ScopeProfiles[0].Profiles[0].Samples[0] + require.Equal(t, []int64{4096}, sample.Values) + require.Len(t, sample.AttributeIndices, 1) + attr := result.profiles.Dictionary.AttributeTable[sample.AttributeIndices[0]] + require.Equal(t, "service.name", result.profiles.Dictionary.StringTable[attr.KeyStrindex]) + require.Equal(t, "sample-service", attr.Value.GetStringValue()) + case <-time.After(10 * time.Second): + t.Fatal("no OTLP profile received from scraper") + } +} diff --git a/pkg/profilestore/grpc.go b/pkg/profilestore/grpc.go index 0b4dab84544..0dbc1b0895b 100644 --- a/pkg/profilestore/grpc.go +++ b/pkg/profilestore/grpc.go @@ -20,17 +20,24 @@ import ( "github.com/go-kit/log" "github.com/go-kit/log/level" "github.com/gogo/status" + "github.com/google/pprof/profile" + "github.com/google/uuid" + pprof "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/pprof" + "go.opentelemetry.io/collector/pdata/pprofile" + "go.opentelemetry.io/collector/pdata/pprofile/pprofileotlp" otelgrpcprofilingpb "go.opentelemetry.io/proto/otlp/collector/profiles/v1development" "google.golang.org/grpc/codes" + "google.golang.org/protobuf/proto" profilestorepb "github.com/parca-dev/parca/gen/proto/go/parca/profilestore/v1alpha1" ) -// GRPCForwarder forward profiles via gRPC to another Parca instance -// instead of storing the profiles locally in a database. +// GRPCForwarder forwards profiles via gRPC to a remote store, +// optionally converting raw pprof profiles to OTLP. type GRPCForwarder struct { - logger log.Logger - client *Client + logger log.Logger + client *Client + convertToOTLP bool profilestorepb.UnimplementedProfileStoreServiceServer otelgrpcprofilingpb.UnimplementedProfilesServiceServer @@ -51,14 +58,18 @@ func NewClient( } } -func NewGRPCForwarder(client *Client, logger log.Logger) *GRPCForwarder { +func NewGRPCForwarder(client *Client, logger log.Logger, convertToOTLP bool) *GRPCForwarder { return &GRPCForwarder{ - client: client, - logger: logger, + client: client, + logger: logger, + convertToOTLP: convertToOTLP, } } func (s *GRPCForwarder) WriteRaw(ctx context.Context, req *profilestorepb.WriteRawRequest) (*profilestorepb.WriteRawResponse, error) { + if s.convertToOTLP { + return s.writeRawOTLP(ctx, req) + } // TODO: Batch writes to only send a request every now and then. // See https://github.com/parca-dev/parca-agent/blob/main/pkg/agent/write_client.go#L28 resp, err := s.client.WriteRaw(ctx, req) @@ -68,7 +79,67 @@ func (s *GRPCForwarder) WriteRaw(ctx context.Context, req *profilestorepb.WriteR return resp, err } +func (s *GRPCForwarder) writeRawOTLP(ctx context.Context, req *profilestorepb.WriteRawRequest) (*profilestorepb.WriteRawResponse, error) { + profiles := pprofile.NewProfiles() + for _, series := range req.GetSeries() { + for _, sample := range series.GetSamples() { + p, err := profile.ParseData(sample.GetRawProfile()) + if err != nil { + return nil, status.Errorf(codes.InvalidArgument, "parse pprof profile: %v", err) + } + // A period type is optional in pprof, but the translator requires one. + if p.PeriodType == nil { + p.PeriodType = &profile.ValueType{} + } + converted, err := pprof.ConvertPprofToProfiles(p) + if err != nil { + return nil, status.Errorf(codes.InvalidArgument, "convert pprof profile to OTLP: %v", err) + } + for _, resource := range converted.ResourceProfiles().All() { + // Target and external labels describe the resource. Labels inside + // the pprof data remain sample attributes, as set by the translator. + for _, label := range series.GetLabels().GetLabels() { + resource.Resource().Attributes().PutStr(label.GetName(), label.GetValue()) + } + for _, scope := range resource.ScopeProfiles().All() { + for _, p := range scope.Profiles().All() { + p.SetProfileID(pprofile.ProfileID(uuid.New())) + } + } + } + if err := converted.MergeTo(profiles); err != nil { + return nil, status.Errorf(codes.Internal, "merge OTLP profiles: %v", err) + } + } + } + if profiles.ProfileCount() == 0 { + return &profilestorepb.WriteRawResponse{}, nil + } + + // Collector pdata uses its own protobuf types. Bridge through the wire + // format to reuse the existing OTLP client and its connection settings. + data, err := pprofileotlp.NewExportRequestFromProfiles(profiles).MarshalProto() + if err != nil { + return nil, status.Errorf(codes.Internal, "marshal OTLP profiles: %v", err) + } + export := &otelgrpcprofilingpb.ExportProfilesServiceRequest{} + if err := proto.Unmarshal(data, export); err != nil { + return nil, status.Errorf(codes.Internal, "unmarshal OTLP profiles: %v", err) + } + resp, err := s.Export(ctx, export) + if err != nil { + return nil, err + } + if partial := resp.GetPartialSuccess(); partial.GetRejectedProfiles() > 0 { + return nil, status.Errorf(codes.DataLoss, "upstream rejected %d profiles: %s", partial.GetRejectedProfiles(), partial.GetErrorMessage()) + } + return &profilestorepb.WriteRawResponse{}, nil +} + func (s *GRPCForwarder) Write(srv profilestorepb.ProfileStoreService_WriteServer) error { + if s.convertToOTLP { + return status.Error(codes.Unimplemented, "OTLP forwarding supports pprof WriteRaw and OTLP Export, not Arrow Write") + } c, err := s.client.Write(srv.Context()) if err != nil { return status.Errorf(codes.Internal, "failed to create stream: %v", err) diff --git a/pkg/profilestore/grpc_test.go b/pkg/profilestore/grpc_test.go new file mode 100644 index 00000000000..254495bcce1 --- /dev/null +++ b/pkg/profilestore/grpc_test.go @@ -0,0 +1,182 @@ +// Copyright 2026 The Parca Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package profilestore + +import ( + "bytes" + "context" + "net" + "sync" + "testing" + + "github.com/go-kit/log" + "github.com/google/pprof/profile" + "github.com/stretchr/testify/require" + otlppb "go.opentelemetry.io/proto/otlp/collector/profiles/v1development" + commonpb "go.opentelemetry.io/proto/otlp/common/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" + "google.golang.org/grpc/test/bufconn" + + profilestorepb "github.com/parca-dev/parca/gen/proto/go/parca/profilestore/v1alpha1" +) + +type otlpTestServer struct { + otlppb.UnimplementedProfilesServiceServer + mu sync.Mutex + export func(context.Context, *otlppb.ExportProfilesServiceRequest) (*otlppb.ExportProfilesServiceResponse, error) +} + +func (s *otlpTestServer) Export(ctx context.Context, req *otlppb.ExportProfilesServiceRequest) (*otlppb.ExportProfilesServiceResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.export(ctx, req) +} + +func TestOTLPForwarder(t *testing.T) { + ctx := t.Context() + listener := bufconn.Listen(1024 * 1024) + server := grpc.NewServer() + upstream := &otlpTestServer{} + // Register only OTLP: forwarding must not depend on any Parca upstream API. + otlppb.RegisterProfilesServiceServer(server, upstream) + go func() { _ = server.Serve(listener) }() + t.Cleanup(server.Stop) + conn, err := grpc.NewClient("passthrough:///upstream", + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), + ) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + forwarder := NewGRPCForwarder(NewClient(nil, otlppb.NewProfilesServiceClient(conn)), log.NewNopLogger(), true) + + request := &profilestorepb.WriteRawRequest{} + for i, name := range []string{"checkout", "inventory"} { + fn := &profile.Function{ID: 1, Name: name + ".work", Filename: name + ".go"} + loc := &profile.Location{ID: 1, Address: uint64(0x1000 + i), Line: []profile.Line{{Function: fn, Line: 42}}} + p := &profile.Profile{ + SampleType: []*profile.ValueType{{Type: "alloc_objects", Unit: "count"}, {Type: "alloc_space", Unit: "bytes"}}, + Sample: []*profile.Sample{{ + Location: []*profile.Location{loc}, Value: []int64{int64(3 + i), int64(512 + i)}, + Label: map[string][]string{"service.name": {"sample-" + name}}, + NumLabel: map[string][]int64{"request.size": {int64(17 + i)}}, + NumUnit: map[string][]string{"request.size": {"bytes"}}, + }}, + Function: []*profile.Function{fn}, Location: []*profile.Location{loc}, + TimeNanos: 1700000000000000000, DurationNanos: 123456789, + // No PeriodType: valid pprof profiles can omit it. + } + var data bytes.Buffer + if i == 0 { + require.NoError(t, p.Write(&data)) + } else { + require.NoError(t, p.WriteUncompressed(&data)) + } + request.Series = append(request.Series, &profilestorepb.RawProfileSeries{ + Labels: &profilestorepb.LabelSet{Labels: []*profilestorepb.Label{ + {Name: "service.name", Value: name}, {Name: "k8s.namespace.name", Value: "shop"}, + }}, + Samples: []*profilestorepb.RawSample{{RawProfile: data.Bytes()}}, + }) + } + + requests := make(chan *otlppb.ExportProfilesServiceRequest, 1) + upstream.export = func(_ context.Context, req *otlppb.ExportProfilesServiceRequest) (*otlppb.ExportProfilesServiceResponse, error) { + requests <- req + return &otlppb.ExportProfilesServiceResponse{}, nil + } + _, err = forwarder.WriteRaw(ctx, request) + require.NoError(t, err) + received := <-requests + require.NotNil(t, received) + require.Len(t, received.ResourceProfiles, 2) + dictionary := received.Dictionary + ids := map[string]bool{} + for i, resource := range received.ResourceProfiles { + name := []string{"checkout", "inventory"}[i] + attrs := map[string]string{} + for _, attr := range resource.Resource.Attributes { + attrs[attr.Key] = attr.Value.GetStringValue() + } + require.Equal(t, map[string]string{"service.name": name, "k8s.namespace.name": "shop"}, attrs) + require.Len(t, resource.ScopeProfiles, 1) + require.Len(t, resource.ScopeProfiles[0].Profiles, 2) + values := map[string]int64{} + for _, p := range resource.ScopeProfiles[0].Profiles { + require.Len(t, p.ProfileId, 16) + require.NotEqual(t, make([]byte, 16), p.ProfileId) + require.False(t, ids[string(p.ProfileId)]) + ids[string(p.ProfileId)] = true + require.Equal(t, uint64(1700000000000000000), p.TimeUnixNano) + require.Equal(t, uint64(123456789), p.DurationNano) + require.Len(t, p.Samples, 1) + sample := p.Samples[0] + require.Len(t, sample.Values, 1) + values[dictionary.StringTable[p.SampleType.TypeStrindex]] = sample.Values[0] + sampleAttrs := map[string]*commonpb.AnyValue{} + for _, index := range sample.AttributeIndices { + attr := dictionary.AttributeTable[index] + sampleAttrs[dictionary.StringTable[attr.KeyStrindex]] = attr.Value + if dictionary.StringTable[attr.KeyStrindex] == "request.size" { + require.Equal(t, "bytes", dictionary.StringTable[attr.UnitStrindex]) + } + } + require.Len(t, sampleAttrs, 2) + require.Equal(t, "sample-"+name, sampleAttrs["service.name"].GetStringValue()) + require.Equal(t, int64(17+i), sampleAttrs["request.size"].GetIntValue()) + stack := dictionary.StackTable[sample.StackIndex] + require.Len(t, stack.LocationIndices, 1) + loc := dictionary.LocationTable[stack.LocationIndices[0]] + require.Equal(t, uint64(0x1000+i), loc.Address) + require.Equal(t, name+".work", dictionary.StringTable[dictionary.FunctionTable[loc.Lines[0].FunctionIndex].NameStrindex]) + } + require.Equal(t, map[string]int64{"alloc_objects": int64(3 + i), "alloc_space": int64(512 + i)}, values) + } + + t.Run("OTLP conversion stays opt-in", func(t *testing.T) { + legacy := NewGRPCForwarder(NewClient(profilestorepb.NewProfileStoreServiceClient(conn), + otlppb.NewProfilesServiceClient(conn)), log.NewNopLogger(), false) + _, err := legacy.WriteRaw(ctx, request) + require.Equal(t, codes.Unimplemented, status.Code(err), "Parca mode must still use the Parca upstream API") + }) + t.Run("invalid pprof", func(t *testing.T) { + _, err := forwarder.WriteRaw(ctx, &profilestorepb.WriteRawRequest{Series: []*profilestorepb.RawProfileSeries{{ + Samples: []*profilestorepb.RawSample{{RawProfile: []byte("not pprof")}}, + }}}) + require.Equal(t, codes.InvalidArgument, status.Code(err)) + }) + t.Run("upstream error", func(t *testing.T) { + upstream.mu.Lock() + upstream.export = func(context.Context, *otlppb.ExportProfilesServiceRequest) (*otlppb.ExportProfilesServiceResponse, error) { + return nil, status.Error(codes.Unavailable, "upstream offline") + } + upstream.mu.Unlock() + _, err := forwarder.WriteRaw(ctx, request) + require.Equal(t, codes.Unavailable, status.Code(err)) + }) + t.Run("partial rejection", func(t *testing.T) { + upstream.mu.Lock() + upstream.export = func(context.Context, *otlppb.ExportProfilesServiceRequest) (*otlppb.ExportProfilesServiceResponse, error) { + return &otlppb.ExportProfilesServiceResponse{PartialSuccess: &otlppb.ExportProfilesPartialSuccess{ + RejectedProfiles: 1, ErrorMessage: "unsupported sample type", + }}, nil + } + upstream.mu.Unlock() + _, err := forwarder.WriteRaw(ctx, request) + require.Equal(t, codes.DataLoss, status.Code(err)) + require.Contains(t, err.Error(), "unsupported sample type") + }) +} diff --git a/pkg/scrape/target_test.go b/pkg/scrape/target_test.go index 4b87c617355..1fea9de0955 100644 --- a/pkg/scrape/target_test.go +++ b/pkg/scrape/target_test.go @@ -162,3 +162,55 @@ func trueValue() *bool { a := true return &a } + +func TestOTLPExampleConfigs(t *testing.T) { + for _, tc := range []struct { + name string + target model.LabelSet + expected map[string]string + }{ + { + name: "static", + target: model.LabelSet{ + "__address__": "checkout:6060", "namespace": "shop", "environment": "development", + }, + expected: map[string]string{ + "service.name": "checkout", "service.namespace": "shop", + "service.instance.id": "checkout:6060", "deployment.environment.name": "development", + }, + }, + { + name: "kubernetes", + target: model.LabelSet{ + "__address__": "10.0.0.1:6060", "__meta_kubernetes_pod_phase": "Running", + "__meta_kubernetes_pod_container_port_name": "pprof", + "__meta_kubernetes_pod_label_app_kubernetes_io_name": "checkout", + "__meta_kubernetes_namespace": "shop", "__meta_kubernetes_pod_name": "checkout-abc", + "__meta_kubernetes_pod_uid": "pod-uid", "__meta_kubernetes_pod_node_name": "worker-3", + "__meta_kubernetes_pod_container_name": "app", + }, + expected: map[string]string{ + "service.name": "checkout", "service.namespace": "shop", "service.instance.id": "pod-uid/app", + "k8s.namespace.name": "shop", "k8s.pod.name": "checkout-abc", "k8s.pod.uid": "pod-uid", + "k8s.node.name": "worker-3", "k8s.container.name": "app", + }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + cfg, err := config.LoadFile("../../deploy/otlp/" + tc.name + ".yaml") + require.NoError(t, err) + require.NoError(t, cfg.Validate()) + targets, err := targetsFromGroup(&targetgroup.Group{Targets: []model.LabelSet{tc.target}}, + cfg.ScrapeConfigs[0], nil, labels.NewBuilder(labels.EmptyLabels())) + require.NoError(t, err) + require.NotEmpty(t, targets) + for _, target := range targets { + for key, value := range tc.expected { + require.Equal(t, value, target.Labels().Get(key), key) + } + require.False(t, target.Labels().Has("namespace")) + require.False(t, target.Labels().Has("environment")) + } + }) + } +} From 06325f82c5cad578454790eff58cb4cde41f0652 Mon Sep 17 00:00:00 2001 From: Frederic Branczyk Date: Fri, 30 Oct 2026 13:50:09 +0000 Subject: [PATCH 2/2] docs: regenerate CLI help for OTLP forwarding Amp-Thread-ID: https://ampcode.com/threads/T-01a18c4e-d681-71f4-8158-32d8026e071b Co-authored-by: Amp --- README.md | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 0a523bf7f1c..14571f455ff 100644 --- a/README.md +++ b/README.md @@ -115,9 +115,10 @@ Flags: --config-path="parca.yaml" Path to config file. --mode="all" All runs all components. Scraper-only and - forwarder send profiles to a remote Parca gRPC - endpoint. OTLP-forwarder converts pprof profiles - and sends them to an OTLP profiles gRPC endpoint. + forwarder send profiles to a remote Parca + gRPC endpoint. OTLP-forwarder converts pprof + profiles and sends them to an OTLP profiles + gRPC endpoint. --http-address=":7070" Address to bind HTTP server to. --http-read-timeout=5s Timeout duration for HTTP server to read request body.