diff --git a/hub/Gemfile b/hub/Gemfile index 62b1bee5..d57f5b3d 100644 --- a/hub/Gemfile +++ b/hub/Gemfile @@ -7,6 +7,11 @@ gem "bootsnap", require: false gem "json", "~> 3.0" gem "jwt", "~> 3.3" gem "oaken", "1.0.0" +gem "opentelemetry-exporter-otlp", require: false +gem "opentelemetry-instrumentation-action_pack", require: false +gem "opentelemetry-instrumentation-active_job", require: false +gem "opentelemetry-instrumentation-net_http", require: false +gem "opentelemetry-sdk", require: false gem "propshaft" gem "puma", ">= 5.0" gem "rails", "~> 8.1.4" diff --git a/hub/Gemfile.lock b/hub/Gemfile.lock index 36d13bad..d17da70c 100644 --- a/hub/Gemfile.lock +++ b/hub/Gemfile.lock @@ -123,6 +123,26 @@ GEM raabro (~> 1.4) globalid (1.4.0) activesupport (>= 6.1) + google-protobuf (4.36.2) + bigdecimal + rake (~> 13.3) + google-protobuf (4.36.2-aarch64-linux-gnu) + bigdecimal + rake (~> 13.3) + google-protobuf (4.36.2-aarch64-linux-musl) + bigdecimal + rake (~> 13.3) + google-protobuf (4.36.2-arm64-darwin) + bigdecimal + rake (~> 13.3) + google-protobuf (4.36.2-x86_64-linux-gnu) + bigdecimal + rake (~> 13.3) + google-protobuf (4.36.2-x86_64-linux-musl) + bigdecimal + rake (~> 13.3) + googleapis-common-protos-types (1.23.0) + google-protobuf (~> 4.26) hashdiff (1.2.1) i18n (1.15.2) concurrent-ruby (~> 1.0) @@ -182,6 +202,51 @@ GEM oj (3.17.6) bigdecimal (>= 3.0) ostruct (>= 0.2) + opentelemetry-api (1.11.1) + logger + opentelemetry-common (0.25.2) + opentelemetry-api (~> 1.0) + opentelemetry-exporter-otlp (0.37.0) + google-protobuf (>= 3.18) + googleapis-common-protos-types (~> 1.3) + opentelemetry-api (~> 1.1) + opentelemetry-common (~> 0.20) + opentelemetry-exporter-otlp-common (~> 0.2) + opentelemetry-sdk (~> 1.10) + opentelemetry-semantic_conventions + opentelemetry-exporter-otlp-common (0.3.0) + google-protobuf (>= 3.18) + googleapis-common-protos-types (~> 1.3) + opentelemetry-api (~> 1.1) + opentelemetry-common (~> 0.20) + opentelemetry-instrumentation-action_pack (0.18.1) + opentelemetry-instrumentation-rack (~> 0.29) + opentelemetry-instrumentation-active_job (0.13.0) + opentelemetry-instrumentation-base (~> 0.25) + opentelemetry-instrumentation-base (0.27.0) + opentelemetry-api (~> 1.7) + opentelemetry-common (~> 0.21) + opentelemetry-logs-api (~> 0.6) + opentelemetry-metrics-api (~> 0.9) + opentelemetry-registry (~> 0.1) + opentelemetry-instrumentation-net_http (0.29.2) + opentelemetry-instrumentation-base (~> 0.25) + opentelemetry-instrumentation-rack (0.31.2) + opentelemetry-instrumentation-base (~> 0.25) + opentelemetry-logs-api (0.6.0) + opentelemetry-api (~> 1.0) + opentelemetry-metrics-api (0.9.0) + opentelemetry-api (~> 1.0) + opentelemetry-registry (0.6.3) + opentelemetry-api (~> 1.1) + opentelemetry-sdk (1.13.1) + logger + opentelemetry-api (~> 1.1) + opentelemetry-common (~> 0.20) + opentelemetry-registry (~> 0.2) + opentelemetry-semantic_conventions + opentelemetry-semantic_conventions (1.44.0) + opentelemetry-api (~> 1.0) ostruct (0.6.3) parallel (2.2.0) parser (3.3.12.0) @@ -377,6 +442,11 @@ DEPENDENCIES json (~> 3.0) jwt (~> 3.3) oaken (= 1.0.0) + opentelemetry-exporter-otlp + opentelemetry-instrumentation-action_pack + opentelemetry-instrumentation-active_job + opentelemetry-instrumentation-net_http + opentelemetry-sdk propshaft puma (>= 5.0) rails (~> 8.1.4) @@ -435,6 +505,13 @@ CHECKSUMS form_props (0.2.2) sha256=a46f8f7fd216b26ed6d6416d22334344a531188aee0e854aebc81761ee280cad fugit (1.13.0) sha256=a4f093fce740da52f216740a5041e2a594ea763cdb89e8b2754ca4399634ab18 globalid (1.4.0) sha256=037f12fbf1d9d7a014d501c2d5c77356fd4ddd96d7a7991d6700bba96706f427 + google-protobuf (4.36.2) sha256=2f8daff50ae14aad36ab0ccdcf284065fa71d726c444efe114e6db82ffe62494 + google-protobuf (4.36.2-aarch64-linux-gnu) sha256=c3279ec6eaa59f6ed6276eda2a33cffb4306471183ab79a332e698b8564f5169 + google-protobuf (4.36.2-aarch64-linux-musl) sha256=0772c9dde1e6dce9a6b44d1d35c28c29adee99d9bd6aff5e21840daeb22161ce + google-protobuf (4.36.2-arm64-darwin) sha256=f70d810e6bbab9d8ad9c1d5fd685a21de4c10ad3ad32efb9455e17e5b1286917 + google-protobuf (4.36.2-x86_64-linux-gnu) sha256=d25a820873e423f1d209ae8c672a4a28b019e2429debd0901d26670a0b4e14db + google-protobuf (4.36.2-x86_64-linux-musl) sha256=0fa17686c896da96a60a705e56ddc7d4ff56413f52787c655bd4c9fb33ffe14f + googleapis-common-protos-types (1.23.0) sha256=992e740a523794d9fc5f29a504465d8fc737aaa16c930fe7228e3346860faf0a hashdiff (1.2.1) sha256=9c079dbc513dfc8833ab59c0c2d8f230fa28499cc5efb4b8dd276cf931457cd1 i18n (1.15.2) sha256=00f9eb62412fe593b2a65a97daa75300d37abb8f7202ec748e94b6d46a9dd1b5 io-console (0.9.2) sha256=efa74f891dd03c0939a931dfc6e74c2813d904763d456ea9762b0525e748db08 @@ -465,6 +542,20 @@ CHECKSUMS nokogiri (1.19.4-x86_64-linux-musl) sha256=17dfb7c1fa194ae02fbf7c51a7afc8d278045ab3fdacfd86f91d02d7b274470b oaken (1.0.0) sha256=e89249bc4f6cd3ab9b5e3abdc06c377fa772e6bcb3005cc09ff044bb8d3f1dc2 oj (3.17.6) sha256=f511257e01a12030c3ceb8023cd2e96fca924ea9c741f57ea366816f1fd8e68c + opentelemetry-api (1.11.1) sha256=d07f9987ea6240b3fd5ed5dd7450f93c58e403e0553e804069c23ae18ba61ff7 + opentelemetry-common (0.25.2) sha256=c035d44987789356fe8feeeb870708b44fe7564ffc032dec7200c39da92a911d + opentelemetry-exporter-otlp (0.37.0) sha256=edc8bda657dce5f4d527ac735f75d44043bb5924de6df0e3f01260108aa76136 + opentelemetry-exporter-otlp-common (0.3.0) sha256=855d10d788a7272dfe5b14efa229507619abbef29085ef4a8acfd2c0d3af6171 + opentelemetry-instrumentation-action_pack (0.18.1) sha256=0b827d703baa2951458e36f2bdfa4a2cdb17a0130cfb0e1b6fe5d67fadc85c2d + opentelemetry-instrumentation-active_job (0.13.0) sha256=42da156924bbabdaca6268be1e46963a04829bd0b2b82fc6e368597a3525cde2 + opentelemetry-instrumentation-base (0.27.0) sha256=e55edb82cd0b018e82e746be61fa36cb05b7b68b2fa37f0e4a96b923ad3d7735 + opentelemetry-instrumentation-net_http (0.29.2) sha256=4256105f058404eb0bd43e4a987e48f9e48fcade06b40a388092c380b2e946eb + opentelemetry-instrumentation-rack (0.31.2) sha256=4a183066110455711a0c77754e3f46652f646221d7704d968a48d155b47f8309 + opentelemetry-logs-api (0.6.0) sha256=b889ac61d90326bffadb83bf7278d246d1086e2349c9052aeb072ed375ce73d0 + opentelemetry-metrics-api (0.9.0) sha256=e11a12055113e11421313f71c758e5e10e4662e9e5c3b91786218a692e81206d + opentelemetry-registry (0.6.3) sha256=568fa8eaea94394f29f4fce3f0339ab10c3bf6d6c8beed6b3cfee6fd37fcc608 + opentelemetry-sdk (1.13.1) sha256=91c9cc315ac7ef30243549eeff8b27cd0d39bd02bdf9c9ae481ab96a1cec6754 + opentelemetry-semantic_conventions (1.44.0) sha256=b09b8793422f13b01c08f7a6f29b5f511f47f37bcdb57c0513465a0464168513 ostruct (0.6.3) sha256=95a2ed4a4bd1d190784e666b47b2d3f078e4a9efda2fccf18f84ddc6538ed912 parallel (2.2.0) sha256=e1059c5fd7b649558a0aec38a769f06a42942bdb40503d005a59c352fe011cd8 parser (3.3.12.0) sha256=21a6d7f755d5a24dfbdc6e6b772e4e879a52e7631a88bc5a3a134606052c9828 diff --git a/hub/app/models/github_client/batch.rb b/hub/app/models/github_client/batch.rb index 0b62a62c..f5668d00 100644 --- a/hub/app/models/github_client/batch.rb +++ b/hub/app/models/github_client/batch.rb @@ -8,9 +8,16 @@ class GithubClient::Batch def self.map(items, &operation) return [] if items.empty? + tracing_context = OpenTelemetry::Context.current if defined?(OpenTelemetry::Context) executor = Concurrent::FixedThreadPool.new([ items.size, MAX_CONCURRENCY ].min) futures = items.map do |item| - Concurrent::Promises.future_on(executor) { operation.call(item) } + Concurrent::Promises.future_on(executor) do + if tracing_context + OpenTelemetry::Context.with_current(tracing_context) { operation.call(item) } + else + operation.call(item) + end + end end futures.map(&:value!) ensure diff --git a/hub/config/application.rb b/hub/config/application.rb index ac8fcab9..7b3974f6 100644 --- a/hub/config/application.rb +++ b/hub/config/application.rb @@ -30,7 +30,7 @@ class Application < Rails::Application # Please, add to the `ignore` list any other `lib` subdirectories that do # not contain `.rb` files, or that should not be reloaded or eager loaded. # Common ones are `templates`, `generators`, or `middleware`, for example. - config.autoload_lib(ignore: %w[assets tasks]) + config.autoload_lib(ignore: %w[assets hub_tracing tasks]) # Configuration for the application, engines, and railties goes here. # @@ -85,4 +85,7 @@ class Application < Rails::Application # only to replica-learning and never stores it. See docs/copilot.md. config.x.copilot_token = ENV["REPLICA_COPILOT_TOKEN"].presence end + + require_relative "../lib/hub_tracing" + HubTracing.start! end diff --git a/hub/docs/README.md b/hub/docs/README.md index 91add69c..96370916 100644 --- a/hub/docs/README.md +++ b/hub/docs/README.md @@ -4,7 +4,7 @@ Start with [Deployment](deployment.md) to run Replica Hub or [Development](devel The [shared contributor guides](../../docs/README.md) apply across Replica. This index covers Hub-specific behavior and operations. -- **Operate Hub:** [Deployment](deployment.md), [GitHub App setup](github-app-setup.md), [security](security.md), and [backups](backups-and-restore.md). +- **Operate Hub:** [Deployment](deployment.md), [server-side tracing](deployment.md#optional-server-side-tracing), [GitHub App setup](github-app-setup.md), [security](security.md), and [backups](backups-and-restore.md). - **GitHub and Builds:** [App setup](github-app-setup.md), [access](authentication-and-repository-authorization.md), [discovery](build-discovery.md), [Actions](github-actions-builds.md), and [storage](artifact-storage.md). - **Review:** [Gallery](gallery.md), [previews](preview-serving.md), [collaboration](review-collaboration.md), and [share links](share-links.md). - **Analytics:** [Repository activity events](analytics.md). diff --git a/hub/docs/deployment.md b/hub/docs/deployment.md index f6851d96..58034aa0 100644 --- a/hub/docs/deployment.md +++ b/hub/docs/deployment.md @@ -13,6 +13,117 @@ The recommended deployment needs: The default deployment runs one Replica Hub container and stores its SQLite databases and Build files in one persistent Docker volume. +## Optional server-side tracing + +Hub can send server-side traces to an OpenTelemetry Protocol (OTLP) endpoint. Traces describe Rails requests, Active Job enqueueing and execution, instrumented Active Record operations, and outbound Net::HTTP calls. The published image includes OpenTelemetry instrumentation and its HTTP/protobuf exporter, not vendor-specific APM SDKs. + +Tracing is off by default. To enable it, add these settings to the private deployment `.env` or your container platform's environment: + +```dotenv +OTEL_TRACES_EXPORTER=otlp +OTEL_SERVICE_NAME=replica-hub +OTEL_RESOURCE_ATTRIBUTES=deployment.environment.name=production +OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=http://otel-collector:4318/v1/traces +OTEL_EXPORTER_OTLP_TRACES_PROTOCOL=http/protobuf +``` + +This example assumes a Collector named `otel-collector` is reachable from Hub on a trusted private container network. Hub does not install or start that service. Use HTTPS for endpoints outside a trusted private network. + +Apply every tracing configuration change to all Hub web and job processes. With the Compose bundle, run `./replica up`; it recreates Hub with the private `.env`. Other container platforms must recreate or restart every process with the changed variables. No image rebuild is needed. + +### Check traces with a local viewer + +Use a local OpenTelemetry viewer before connecting a production recipient. After `./replica up` creates the default Replica Compose network, the standalone Aspire dashboard can join it and receive traces directly from Hub: + +```sh +docker run --rm -d \ + --name replica-otel-dashboard \ + --network replica-hub_default \ + -p 127.0.0.1:18888:18888 \ + -e ASPIRE_DASHBOARD_UNSECURED_ALLOW_ANONYMOUS=true \ + mcr.microsoft.com/aspire/dashboard:latest +``` + +Add this endpoint to Hub's private `.env`: + +```dotenv +OTEL_TRACES_EXPORTER=otlp +OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=http://replica-otel-dashboard:18890/v1/traces +OTEL_EXPORTER_OTLP_TRACES_PROTOCOL=http/protobuf +``` + +Run `./replica up` to recreate Hub with the changed environment. + +Open , use Hub, and confirm Rails requests and jobs appear. The command exposes the unsecured dashboard only on the local machine and leaves its OTLP receiver on the private Compose network. It is for temporary local inspection, not production retention. + +When inspection is complete, first set `OTEL_TRACES_EXPORTER=none` or configure a replacement recipient, then run `./replica up`. Stop the local viewer afterward with `docker stop replica-otel-dashboard`. Stopping it while Hub still targets it causes later batches to fail and log delivery errors. On another container platform, recreate or restart every Hub web and job process with the replacement configuration before stopping the viewer. + +| Setting | Purpose | +| --- | --- | +| `OTEL_TRACES_EXPORTER` | `otlp` enables export; `none` disables tracing and is Hub's default. Other exporters are not included. | +| `OTEL_SERVICE_NAME` | Service identity. It overrides `service.name` in `OTEL_RESOURCE_ATTRIBUTES`; Hub defaults to `replica-hub` only when neither setting supplies a name. | +| `OTEL_RESOURCE_ATTRIBUTES` | Comma-separated deployment metadata, such as `deployment.environment.name=production,service.version=1.0.0`. Do not put secrets or personal data here. | +| `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` | Complete HTTP(S) trace endpoint, including its required path. Credentials, query strings, and fragments are not accepted in the URL. | +| `OTEL_EXPORTER_OTLP_TRACES_HEADERS` | Authentication headers required by the recipient, in the exporter's comma-separated `key=value` format. Keep this value private. | +| `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL` | Only `http/protobuf` is supported. A gRPC-only endpoint needs a Collector or gateway that accepts HTTP/protobuf. | +| `OTEL_TRACES_SAMPLER` | Standard SDK sampling policy, for example `parentbased_traceidratio`. The SDK default is parent-based, with all new root traces sampled. | +| `OTEL_TRACES_SAMPLER_ARG` | Sampling-policy argument; for `parentbased_traceidratio`, a fraction from `0` to `1`, such as `0.1`. | +| `OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE` | Optional path inside the container to a trusted CA certificate for a private endpoint. Mount the certificate when needed; keep TLS verification enabled. | +| `OTEL_SDK_DISABLED` | Set to `true` to disable tracing regardless of the exporter setting. | +| `REPLICA_OTEL_INCLUDE_SENSITIVE_DATA` | Set to `true` to include SQL text, filtered job arguments, and exception diagnostics. Defaults to `false`; credential-keyed job fields remain filtered. | +| `REPLICA_OTEL_TRUST_INCOMING_CONTEXT` | Set to `true` only behind a trusted ingress that removes client-supplied trace headers and supplies its own. Defaults to `false`. | +| `REPLICA_OTEL_UNTRACED_HOSTS` | Optional comma-separated exact DNS hostnames, local names, or IPv4/IPv6 literals to exclude from outbound Net::HTTP tracing and trace-header injection. Do not include a scheme, port, path, IPv6 brackets, or prefix length. Subdomains are not included automatically. Empty or malformed entries are rejected. | + +The exporter also accepts the common `OTEL_EXPORTER_OTLP_ENDPOINT`, `OTEL_EXPORTER_OTLP_HEADERS`, and `OTEL_EXPORTER_OTLP_PROTOCOL` settings. Trace-specific values take precedence. The common endpoint is a base URL to which the exporter appends `/v1/traces`; the trace-specific endpoint is used as supplied. Enabling export requires an explicit endpoint. Hub rejects unsupported exporters, protocols, and malformed endpoint URLs at startup. + +### Choose a recipient + +Send directly to an APM provider that accepts OTLP over HTTP/protobuf, using its documented trace endpoint and authentication headers. Alternatively, send to a company-managed OpenTelemetry Collector. The Collector can filter, sample, translate, and forward traces to one or more APMs without changing Hub's image. Choose a Collector distribution that includes the exporters your providers need. + +Sentry documents a direct OTLP trace endpoint. Datadog provides a Collector exporter. These paths do not guarantee the same features as each provider's native SDK; check the provider's supported signals, release status, and deployment version. + +The deployment operator controls the destination and credentials. Restrict recipients and network access through your deployment's outbound-network policy. Hub's repository analytics destination settings do not configure or restrict tracing. + +For mutual TLS, the exporter accepts `OTEL_EXPORTER_OTLP_TRACES_CLIENT_CERTIFICATE` and `OTEL_EXPORTER_OTLP_TRACES_CLIENT_KEY`, pointing to a mounted client certificate and RSA private key. Keep certificate verification enabled. + +### Trace-context boundaries + +Public HTTP requests start new traces by default. Hub ignores their incoming `traceparent`, so a client cannot override the deployment's root sampling decision by supplying a sampled parent. Enable `REPLICA_OTEL_TRUST_INCOMING_CONTEXT` only when all traffic passes through an ingress that sanitizes client-supplied trace headers. This setting trusts every incoming request; it does not authenticate a proxy or enforce a network restriction. + +Hub discards incoming `tracestate` and baggage even with trusted ingress. Jobs continue the context serialized by Hub through its trusted job queue. Protect queue storage and access independently. + +Outbound Net::HTTP calls propagate trace context by default. Use `REPLICA_OTEL_UNTRACED_HOSTS` to exclude specific destinations; those calls also lose their HTTP spans. This setting does not block the requests or restrict telemetry export destinations. Hub fixes its propagation policy in code; `OTEL_PROPAGATORS` does not override these boundaries. + +### Data and delivery limits + +`REPLICA_OTEL_INCLUDE_SENSITIVE_DATA` controls diagnostic content as one startup policy: + +| Trace data | `false` (default) | `true` | +| --- | --- | --- | +| HTTP method, route template, status and service address | Exported | Exported | +| Database system | Exported | Exported | +| SQL query text | Excluded | Exported | +| Queue, job ID, adapter, provider job ID and priority | Exported | Exported | +| Serialized job arguments | Excluded | Exported after key-based filtering | +| Error status | Exported without a description | Exported without a description | +| Exception type, message and stack trace | Excluded | Exported | + +The exporter enforces this policy before a span leaves Hub. Hub avoids attaching SQL text and job arguments when the setting is `false`. Active Job instrumentation can record exception events in process memory before the exporter removes them. Changing the environment variable in a running process does not change its policy. + +Set `REPLICA_OTEL_INCLUDE_SENSITIVE_DATA=true` to include SQL query text, serialized job arguments, and exception type, message, and stack trace. Hub filters structured job arguments with Rails' configured parameter filters. The opt-in permits identity fields such as `email`, `name`, `username`, `user_name`, `first_name`, `last_name`, `display_name`, `given_name`, `family_name`, and `login`. Other configured filters still apply, including password, token, credential, and authorization rules. Cookie-keyed values remain filtered. The flag does not attach signed-in user records or enable request bodies or headers. + +Filtering is key-based, not a guarantee that exported data contains no personal information or secrets. SQL text can contain literal values, exception messages and stack traces can contain application data, and positional or free-text job arguments lack field names to filter. SQL bind values are not attached separately. Serialized model arguments may include Global IDs. Configure your APM's access and retention policies accordingly, and apply additional Collector filtering if needed. Operator-supplied resource attributes are exported as configured. + +The deployment operator controls the sensitive-data flag through environment configuration. Compose deployments apply a change with `./replica up`; other platforms must recreate or restart every web and job process. The new setting applies to diagnostics collected by those processes; data already sent remains subject to the recipient's retention policy. + +Hub always excludes raw HTTP URL paths and queries, request and response headers and bodies, arbitrary span events, span links, SQL bind values, and error status descriptions from export. When sensitive diagnostics are enabled, exception events retain only `exception.type`, `exception.message`, and `exception.stacktrace`. HTTP span names use methods and route templates instead of concrete URLs. + +Active Job execution and Hub's parallel GitHub calls continue their submitting trace when context is available. The SDK's bounded background queue exports batches outside the request path and resets its worker after process forks. Hub allows up to five seconds to flush at normal process exit. Collector or provider outages are logged and may drop traces; a timed-out HTTP connection is discarded before the next batch. Delivery is not durable and does not gate successful Hub requests. Forced termination can lose pending spans. + +Health checks at `/up` are excluded. Asset compilation with `SECRET_KEY_BASE_DUMMY` does not initialize tracing. This integration does not enable browser monitoring, metric or log export, or vendor-native error reporting. Providers that discard OTLP span events may not display the exported exceptions. + +For exporter settings and their formats, see the [Ruby OTLP exporter](https://github.com/open-telemetry/opentelemetry-ruby/tree/main/exporter/otlp) and the [Ruby SDK documentation](https://opentelemetry.io/docs/languages/ruby/). Provider setup examples are available for [Sentry OTLP](https://docs.sentry.io/concepts/otlp/direct/traces/) and the [Datadog Collector exporter](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/exporter/datadogexporter). + ## Choose an isolated domain Hub displays prototypes inside its review pages. It requires each preview to use a separate HTTPS subdomain, such as `b123.preview.hub.example.com` under `hub.example.com`. This lets preview cookies work while the prototype is embedded. Hub keeps its own sign-in cookie separate. diff --git a/hub/lib/hub_tracing.rb b/hub/lib/hub_tracing.rb new file mode 100644 index 00000000..e0655463 --- /dev/null +++ b/hub/lib/hub_tracing.rb @@ -0,0 +1,119 @@ +require "uri" +require "ipaddr" + +module HubTracing + autoload :Enabled, File.expand_path("hub_tracing/enabled", __dir__) + + HOSTNAME = / + \A + [a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])? + (?:\.[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)* + \z + /ix + + def self.start! + return if ENV["SECRET_KEY_BASE_DUMMY"] || ENV["OTEL_SDK_DISABLED"]&.casecmp?("true") + + exporter = ENV.fetch("OTEL_TRACES_EXPORTER", "none") + unless %w[none otlp].include?(exporter) + raise ArgumentError, "OTEL_TRACES_EXPORTER must be none or otlp" + end + return if exporter == "none" + + include_sensitive_data = boolean_setting("REPLICA_OTEL_INCLUDE_SENSITIVE_DATA") + trust_incoming_context = boolean_setting("REPLICA_OTEL_TRUST_INCOMING_CONTEXT") + untraced_hosts = if ENV.key?("REPLICA_OTEL_UNTRACED_HOSTS") + ENV.fetch("REPLICA_OTEL_UNTRACED_HOSTS").split(",", -1).map(&:strip) + else + [] + end + if ENV.key?("REPLICA_OTEL_UNTRACED_HOSTS") && + (untraced_hosts.empty? || untraced_hosts.any? { !valid_untraced_host?(_1) }) + raise ArgumentError, + "REPLICA_OTEL_UNTRACED_HOSTS must contain comma-separated hostnames or IP addresses" + end + untraced_hosts = untraced_hosts.map do |host| + Regexp.new("\\A#{Regexp.escape(host)}\\z", Regexp::IGNORECASE) + end + + protocol = ENV["OTEL_EXPORTER_OTLP_TRACES_PROTOCOL"] || + ENV.fetch("OTEL_EXPORTER_OTLP_PROTOCOL", "http/protobuf") + unless protocol == "http/protobuf" + raise ArgumentError, "Hub tracing supports only the http/protobuf OTLP protocol" + end + endpoint_name = if ENV.key?("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT") + "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT" + elsif ENV.key?("OTEL_EXPORTER_OTLP_ENDPOINT") + "OTEL_EXPORTER_OTLP_ENDPOINT" + end + unless endpoint_name + raise ArgumentError, + "Hub tracing requires OTEL_EXPORTER_OTLP_TRACES_ENDPOINT or OTEL_EXPORTER_OTLP_ENDPOINT" + end + endpoint = ENV.fetch(endpoint_name) + begin + uri = URI.parse(endpoint) + rescue URI::InvalidURIError + raise ArgumentError, "#{endpoint_name} must be an absolute HTTP(S) URL", cause: nil + end + unless %w[http https].include?(uri.scheme) && uri.host && !uri.host.empty? && + !uri.userinfo && !uri.query && !uri.fragment + raise ArgumentError, + "#{endpoint_name} must be an absolute HTTP(S) URL without credentials, query, or fragment" + end + if endpoint_name == "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT" && uri.path.empty? + raise ArgumentError, "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT must include a request path" + end + validate_otlp_headers! + + Enabled.start!(include_sensitive_data:, trust_incoming_context:, untraced_hosts:) + end + + def self.boolean_setting(name) + value = ENV.fetch(name, "false") + unless %w[true false].include?(value.downcase) + raise ArgumentError, "#{name} must be true or false" + end + + value.casecmp?("true") + end + + def self.validate_otlp_headers! + name = if ENV.key?("OTEL_EXPORTER_OTLP_TRACES_HEADERS") + "OTEL_EXPORTER_OTLP_TRACES_HEADERS" + elsif ENV.key?("OTEL_EXPORTER_OTLP_HEADERS") + "OTEL_EXPORTER_OTLP_HEADERS" + end + return unless name + + entries = ENV.fetch(name).split(",", -1) + valid = entries.any? && entries.all? do |entry| + key, value = entry.split("=", 2) + key && value && + !URI.decode_uri_component(key).strip.empty? && + !URI.decode_uri_component(value).strip.empty? + end + return if valid + + raise ArgumentError, "#{name} must contain comma-separated URL-encoded key=value headers" + rescue ArgumentError, Encoding::CompatibilityError + raise ArgumentError, + "#{name} must contain comma-separated URL-encoded key=value headers", + cause: nil + end + + def self.valid_untraced_host?(host) + return false if host.empty? || host.bytesize > 253 + + if host.include?(":") || host.match?(/\A[0-9.]+\z/) + return false if host.start_with?("[") || host.end_with?("]") || host.include?("/") + + IPAddr.new(host) + return true + end + + host.match?(HOSTNAME) + rescue IPAddr::InvalidAddressError + false + end +end diff --git a/hub/lib/hub_tracing/batch_span_processor.rb b/hub/lib/hub_tracing/batch_span_processor.rb new file mode 100644 index 00000000..dfd49bc2 --- /dev/null +++ b/hub/lib/hub_tracing/batch_span_processor.rb @@ -0,0 +1,64 @@ +module HubTracing + class BatchSpanProcessor < OpenTelemetry::SDK::Trace::Export::BatchSpanProcessor + SUCCESS = OpenTelemetry::SDK::Trace::Export::SUCCESS + FAILURE = OpenTelemetry::SDK::Trace::Export::FAILURE + TIMEOUT = OpenTelemetry::SDK::Trace::Export::TIMEOUT + + def shutdown(timeout: nil) + lock { reset_on_fork(restart_thread: false) } + super + end + + private + + # The pinned SDK does not include its export mutex wait in force-flush deadlines. + def export_batch(span_array, timeout: @exporter_timeout_seconds) + deadline = monotonic_time + timeout if timeout + unless lock_export_before(deadline) + report_result(TIMEOUT, span_array) + return TIMEOUT + end + + begin + remaining_timeout = deadline && deadline - monotonic_time + if remaining_timeout && remaining_timeout <= 0 + report_result(TIMEOUT, span_array) + return TIMEOUT + end + + batch = span_array.map(&:to_span_data) + result_code = @exporter.export(batch, timeout: remaining_timeout) + report_result(result_code, span_array) + result_code + ensure + @export_mutex.unlock + end + rescue StandardError => error + report_result(FAILURE, span_array) + @metrics_reporter.add_to_counter("otel.bsp.error", labels: { "reason" => error.class.to_s }) + FAILURE + end + + def lock_export_before(deadline) + return @export_mutex.lock unless deadline + + loop do + return true if @export_mutex.try_lock + + remaining = deadline - monotonic_time + return false if remaining <= 0 + + sleep([ remaining, 0.001 ].min) + end + end + + def monotonic_time + Process.clock_gettime(Process::CLOCK_MONOTONIC) + end + + def reset_on_fork(restart_thread: true) + @exporter.reset_after_fork if @pid && @pid != Process.pid + super + end + end +end diff --git a/hub/lib/hub_tracing/data_filter.rb b/hub/lib/hub_tracing/data_filter.rb new file mode 100644 index 00000000..4cf2fa7c --- /dev/null +++ b/hub/lib/hub_tracing/data_filter.rb @@ -0,0 +1,15 @@ +module HubTracing + class DataFilter + def initialize + filters = Rails.application.config.filter_parameters + # The opt-in permits identity fields, including Rails' baseline email filter. + filters = filters.reject { |filter| filter == :email || filter == "email" } + @filter = ActiveSupport::ParameterFilter.new(filters + [ :cookie ]) + end + + def job_arguments(arguments) + serialized = ActiveJob::Arguments.serialize(arguments) + JSON.generate(@filter.filter("arguments" => serialized).fetch("arguments")) + end + end +end diff --git a/hub/lib/hub_tracing/enabled.rb b/hub/lib/hub_tracing/enabled.rb new file mode 100644 index 00000000..008719e8 --- /dev/null +++ b/hub/lib/hub_tracing/enabled.rb @@ -0,0 +1,89 @@ +require "opentelemetry/exporter/otlp" +require "opentelemetry/instrumentation/action_pack" +require "opentelemetry/instrumentation/active_job" +require "opentelemetry/instrumentation/net/http" +require "opentelemetry/sdk" +require_relative "batch_span_processor" +require_relative "data_filter" +require_relative "exporter" +require_relative "propagator" + +module HubTracing + module Enabled + def self.start!(include_sensitive_data:, trust_incoming_context:, untraced_hosts:) + span_exporter = Exporter.new(include_sensitive_data:) do + OpenTelemetry::Exporter::OTLP::Exporter.new + end + processor = BatchSpanProcessor.new(span_exporter) + OpenTelemetry::SDK.configure do |config| + resource_service_name = \ + OpenTelemetry::SDK::Resources::Resource.default.attribute_enumerator.to_h["service.name"] + config.service_name = "replica-hub" if resource_service_name == "unknown_service" + config.propagators = [ Propagator.new(trust_incoming_context:) ] + config.add_span_processor(processor) + config.use "OpenTelemetry::Instrumentation::Rack", untraced_endpoints: [ "/up" ] + config.use "OpenTelemetry::Instrumentation::ActionPack", span_naming: :class + config.use "OpenTelemetry::Instrumentation::ActiveJob", + propagation_style: :child, span_naming: :job_class + config.use "OpenTelemetry::Instrumentation::Net::HTTP", untraced_hosts: + end + + collect_job_arguments(include_sensitive_data:) + trace_sql(include_sensitive_data:) + flush_at_exit + end + + def self.collect_job_arguments(include_sensitive_data:) + return unless include_sensitive_data + + # Snapshot the original rules before Rails compiles them into combined regular expressions. + Rails.application.config.after_initialize do + data_filter = DataFilter.new + ActiveSupport.on_load(:active_job) do + record_arguments = proc do |job| + span = OpenTelemetry::Trace.current_span + next unless span.recording? + + begin + span.set_attribute("replica.job.arguments", data_filter.job_arguments(job.arguments)) + rescue ActiveJob::SerializationError, JSON::GeneratorError + Rails.logger.warn("OpenTelemetry could not serialize job arguments; diagnostic arguments omitted") + end + end + before_enqueue(&record_arguments) + before_perform(&record_arguments) + end + end + end + + def self.trace_sql(include_sensitive_data:) + tracer = OpenTelemetry.tracer_provider.tracer("replica-hub") + ActiveSupport::Notifications.subscribe("sql.active_record") do |_, started, finished, _, payload| + next if payload[:name] == "SCHEMA" || payload[:cached] + next unless OpenTelemetry::Trace.current_span.recording? + + span = tracer.start_span \ + "database query", + kind: :client, + start_timestamp: started, + attributes: { + "db.system.name" => "sqlite", + "db.query.text" => include_sensitive_data ? payload[:sql] : nil + }.compact + span.status = OpenTelemetry::Trace::Status.error if payload[:exception] + if include_sensitive_data && payload[:exception_object] + span.record_exception(payload[:exception_object]) + end + span.finish(end_timestamp: finished) + end + end + + def self.flush_at_exit + at_exit do + Timeout.timeout(5) { OpenTelemetry.tracer_provider.shutdown(timeout: 5) } + rescue Timeout::Error + OpenTelemetry.handle_error(message: "OpenTelemetry shutdown exceeded its deadline; pending traces dropped") + end + end + end +end diff --git a/hub/lib/hub_tracing/exporter.rb b/hub/lib/hub_tracing/exporter.rb new file mode 100644 index 00000000..49f4916f --- /dev/null +++ b/hub/lib/hub_tracing/exporter.rb @@ -0,0 +1,144 @@ +require "timeout" + +module HubTracing + class Exporter + OPERATIONAL_ATTRIBUTES = %w[ + http.request.method http.response.status_code http.method http.status_code http.route + url.scheme http.scheme server.address server.port net.peer.name net.peer.port + db.system.name + messaging.system messaging.destination messaging.message.id messaging.active_job.adapter.name + messaging.active_job.message.provider_job_id messaging.active_job.message.priority + code.namespace code.function + ].freeze + SENSITIVE_ATTRIBUTES = %w[db.query.text replica.job.arguments].freeze + EXCEPTION_ATTRIBUTES = %w[exception.type exception.message exception.stacktrace].freeze + TIMEOUT_RESULT = OpenTelemetry::SDK::Trace::Export::TIMEOUT + + def initialize(include_sensitive_data: false, &exporter_factory) + raise ArgumentError, "exporter factory is required" unless exporter_factory + + @exporter_factory = exporter_factory + @exporter = @exporter_factory.call + @include_sensitive_data = include_sensitive_data + @mutex = Mutex.new + @pid = Process.pid + end + + def reset_after_fork + pid = Process.pid + return if @pid == pid + + @exporter = @exporter_factory.call + @mutex = Mutex.new + @pid = pid + end + + def export(spans, timeout: nil) + synchronize_exporter(timeout:) do |remaining_timeout| + export_with_deadline(spans, timeout: remaining_timeout) + end + end + + def shutdown(timeout: nil) + synchronize_exporter(timeout:) do |remaining_timeout| + @exporter.shutdown(timeout: remaining_timeout) + end + end + + def force_flush(timeout: nil) + synchronize_exporter(timeout:) do |remaining_timeout| + @exporter.force_flush(timeout: remaining_timeout) + end + end + + private + + def synchronize_exporter(timeout:) + deadline = monotonic_time + timeout if timeout + return TIMEOUT_RESULT unless lock_before(deadline) + + begin + remaining_timeout = deadline && deadline - monotonic_time + return TIMEOUT_RESULT if remaining_timeout && remaining_timeout <= 0 + + yield remaining_timeout + ensure + @mutex.unlock + end + end + + def lock_before(deadline) + return @mutex.lock unless deadline + + loop do + return true if @mutex.try_lock + + remaining = deadline - monotonic_time + return false if remaining <= 0 + + sleep([ remaining, 0.001 ].min) + end + end + + def monotonic_time + Process.clock_gettime(Process::CLOCK_MONOTONIC) + end + + def export_with_deadline(spans, timeout:) + raise Timeout::Error if timeout && timeout <= 0 + + # The bundled OTLP exporter's retry sleeps can outlast its request timeout. + Timeout.timeout(timeout) do + @exporter.export(spans.map { |span| sanitized(span) }, timeout:) + end + rescue Timeout::Error + replace_timed_out_exporter + OpenTelemetry.handle_error(message: "OTLP export exceeded its deadline; trace batch dropped") + TIMEOUT_RESULT + end + + def replace_timed_out_exporter + timed_out_exporter = @exporter + begin + timed_out_exporter.shutdown(timeout: 0) + rescue StandardError => error + OpenTelemetry.handle_error \ + exception: error, + message: "OpenTelemetry could not close a timed-out OTLP exporter" + end + @exporter = @exporter_factory.call + end + + def sanitized(span) + result = span.dup + attributes = OPERATIONAL_ATTRIBUTES + attributes += SENSITIVE_ATTRIBUTES if @include_sensitive_data + result.attributes = (span.attributes || {}).slice(*attributes) + result.events = sanitized_events(span.events) + result.links = [] + result.status = case span.status.code + when OpenTelemetry::Trace::Status::ERROR then OpenTelemetry::Trace::Status.error + when OpenTelemetry::Trace::Status::OK then OpenTelemetry::Trace::Status.ok + else OpenTelemetry::Trace::Status.unset + end + method = result.attributes["http.request.method"] || result.attributes["http.method"] + if span.kind == :server || method + result.name = [ method || "HTTP", result.attributes["http.route"] ].compact.join(" ") + end + result + end + + def sanitized_events(events) + return [] unless @include_sensitive_data + + (events || []).filter_map do |event| + next unless event.name == "exception" + + OpenTelemetry::SDK::Trace::Event.new \ + event.name, + (event.attributes || {}).slice(*EXCEPTION_ATTRIBUTES), + event.timestamp + end + end + end +end diff --git a/hub/lib/hub_tracing/propagator.rb b/hub/lib/hub_tracing/propagator.rb new file mode 100644 index 00000000..8bdb82b7 --- /dev/null +++ b/hub/lib/hub_tracing/propagator.rb @@ -0,0 +1,29 @@ +module HubTracing + class Propagator + def initialize(trust_incoming_context: false) + @trust_incoming_context = trust_incoming_context + @trace_context = OpenTelemetry::Trace::Propagation::TraceContext.text_map_propagator + end + + def extract(carrier, context: OpenTelemetry::Context.current, + getter: OpenTelemetry::Context::Propagation.text_map_getter) + if getter.equal?(OpenTelemetry::Common::Propagation.rack_env_getter) && !@trust_incoming_context + return OpenTelemetry::Trace.context_with_span \ + OpenTelemetry::Trace::Span::INVALID, + parent_context: context + end + + # Vendor state and baggage never cross a Hub boundary, even with trusted ingress. + @trace_context.extract({ "traceparent" => getter.get(carrier, "traceparent") }, context:) + end + + def inject(carrier, context: OpenTelemetry::Context.current, + setter: OpenTelemetry::Context::Propagation.text_map_setter) + @trace_context.inject(carrier, context:, setter:) + end + + def fields + [ "traceparent" ] + end + end +end diff --git a/hub/test/integration/hub_tracing_test.rb b/hub/test/integration/hub_tracing_test.rb new file mode 100644 index 00000000..63435f8f --- /dev/null +++ b/hub/test/integration/hub_tracing_test.rb @@ -0,0 +1,458 @@ +require "test_helper" +require "open3" +require "base64" +require "opentelemetry/exporter/otlp" + +class HubTracingIntegrationTest < ActiveSupport::TestCase + self.use_transactional_tests = false + + test "normal boot isolates inbound context and excludes sensitive diagnostics" do + result = run_traced_request(failing_job: true) + + assert_equal 200, result.fetch("status") + assert_equal "completed", result.fetch("body") + assert_equal 200, result.fetch("health_status") + refute result.fetch("baggage_propagated"), "incoming baggage reached the outbound service" + refute result.fetch("tracestate_propagated"), "incoming tracestate reached the outbound service" + assert result.fetch("traceparent_propagated"), "ordinary outbound HTTP lost context injection" + payloads = result.fetch("payloads").map { |payload| Base64.strict_decode64(payload) } + assert_not_empty payloads, result.fetch("diagnostics") + payloads.each do |payload| + %w[ + request-private-canary identity-email-canary identity-name-canary password-canary token-canary + sql-diagnostic-canary argument-diagnostic-canary job-failure-diagnostic-canary + ].each do |private_value| + refute_includes payload, private_value + end + end + spans = decoded_spans(result) + server = spans.find { |span| span.kind == :SPAN_KIND_SERVER } + assert server, "Rails boot did not produce a server span" + assert_equal "POST /tracing-probe", server.name + refute_equal "1234567890abcdef1234567890abcdef", server.trace_id.unpack1("H*") + assert_empty server.parent_span_id + assert_equal 1, spans.count { |span| span.kind == :SPAN_KIND_SERVER }, "health check was traced" + jobs = spans.select { |span| span.name.include?("TracingProbeJob") } + assert_operator jobs.size, :>=, 2, "enqueue and serialized execution must both be traced" + assert jobs.all? { |span| span.trace_id == server.trace_id }, "job execution lost its request trace" + jobs.each do |job| + assert spans.any? { |parent| parent.span_id == job.parent_span_id }, "job has no exported parent" + end + metadata = spans.flat_map(&:attributes) + assert metadata.any? { |attribute| attribute.key == "db.system.name" } + jobs.each do |job| + attributes = span_attributes(job) + assert_equal "default", attributes.fetch("messaging.destination").string_value + assert attributes.fetch("messaging.message.id").string_value.present? + assert_equal "test", attributes.fetch("messaging.active_job.adapter.name").string_value + assert_equal "3", attributes.fetch("messaging.active_job.message.priority").string_value + refute attributes.key?("replica.job.arguments") + end + refute metadata.any? { |attribute| attribute.key == "db.query.text" } + refute metadata.any? { |attribute| attribute.key.match?(/bind|parameter/) } + assert spans.any? { |span| span.kind == :SPAN_KIND_CLIENT }, "outbound HTTP was not traced" + spans.each do |span| + assert_empty span.events + assert_empty span.status.message if span.has_status? + end + end + + test "trusted inbound context retains sampled parent but discards tracestate and baggage" do + result = run_traced_request(configuration: { + "REPLICA_OTEL_TRUST_INCOMING_CONTEXT" => "TrUe", + "OTEL_TRACES_SAMPLER" => "parentbased_traceidratio", + "OTEL_TRACES_SAMPLER_ARG" => "0" + }) + + server = decoded_spans(result).find { |span| span.kind == :SPAN_KIND_SERVER } + assert server, result.fetch("diagnostics") + assert_equal "1234567890abcdef1234567890abcdef", server.trace_id.unpack1("H*") + assert_equal "1234567890abcdef", server.parent_span_id.unpack1("H*") + refute result.fetch("baggage_propagated") + refute result.fetch("tracestate_propagated") + assert result.fetch("traceparent_propagated") + assert decoded_spans(result).all? { |span| span.trace_state.empty? } + end + + test "public inbound sampled context cannot override zero sampling" do + result = run_traced_request(configuration: { + "OTEL_TRACES_SAMPLER" => "parentbased_traceidratio", + "OTEL_TRACES_SAMPLER_ARG" => "0" + }) + + assert_equal 200, result.fetch("status") + assert result.fetch("job_performed") + assert_empty decoded_spans(result), "an untrusted sampled parent bypassed the configured sampler" + end + + test "explicit false keeps identity filtering and incoming context isolation" do + result = run_traced_request(configuration: { + "REPLICA_OTEL_TRUST_INCOMING_CONTEXT" => "FaLsE", + "REPLICA_OTEL_INCLUDE_SENSITIVE_DATA" => "FaLsE" + }) + + server = decoded_spans(result).find { |span| span.kind == :SPAN_KIND_SERVER } + assert server, result.fetch("diagnostics") + refute_equal "1234567890abcdef1234567890abcdef", server.trace_id.unpack1("H*") + refute_includes result.fetch("payloads").map { |payload| Base64.strict_decode64(payload) }.join, "identity-email-canary" + end + + test "service name preserves SDK resource precedence" do + [ + [ {}, "replica-hub" ], + [ { "OTEL_RESOURCE_ATTRIBUTES" => "service.name=resource-service" }, "resource-service" ], + [ + { + "OTEL_RESOURCE_ATTRIBUTES" => "service.name=resource-service", + "OTEL_SERVICE_NAME" => "explicit-service" + }, + "explicit-service" + ] + ].each do |configuration, expected| + result = run_traced_request(configuration:) + + assert_equal [ expected ], resource_service_names(result), result.fetch("diagnostics") + end + end + + test "pathless common endpoint appends the OTLP traces path" do + result = run_traced_request(configuration: { + "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT" => nil, + "OTEL_EXPORTER_OTLP_ENDPOINT" => "http://collector.example" + }) + + assert_equal 200, result.fetch("status") + assert_operator result.fetch("export_attempts"), :>, 0 + assert_not_empty decoded_spans(result) + end + + test "sensitive opt in exports SQL filtered job arguments and exception diagnostics" do + result = run_traced_request( + configuration: { "REPLICA_OTEL_INCLUDE_SENSITIVE_DATA" => "TrUe" }, + failing_job: true + ) + + payload = result.fetch("payloads").map { |encoded| Base64.strict_decode64(encoded) }.join + %w[ + password-canary token-canary app-secret-canary authorization-canary cookies-canary + request-private-canary + ].each { |private_value| refute_includes payload, private_value } + spans = decoded_spans(result) + jobs = spans.select { |span| span.name.include?("TracingProbeJob") } + assert_operator jobs.size, :>=, 2 + jobs.each do |job| + arguments = JSON.parse(span_attributes(job).fetch("replica.job.arguments").string_value).last + assert_equal "identity-email-canary@example.test", arguments.fetch("email"), "#{job.name} did not honor identity opt in" + assert_equal "identity-name-canary", arguments.fetch("name") + assert_equal "[FILTERED]", arguments.fetch("password") + assert_equal "[FILTERED]", arguments.fetch("token") + assert_equal "[FILTERED]", arguments.fetch("customer_secret") + assert_equal "[FILTERED]", arguments.fetch("authorization") + assert_equal "[FILTERED]", arguments.fetch("cookies") + end + successful_query = spans.find do |span| + span_attributes(span)["db.query.text"]&.string_value == "SELECT 'sql-diagnostic-canary'" + end + assert successful_query, "successful SQL query was not exported" + + unparented_query = spans.find do |span| + span_attributes(span)["db.query.text"]&.string_value == "SELECT 'unparented-sql-diagnostic-canary'" + end + refute unparented_query, "unparented SQL query was exported" + + failed_query = spans.find do |span| + span_attributes(span)["db.query.text"]&.string_value&.include?("missing_diagnostic_table") + end + assert failed_query, "failed SQL query was not exported" + assert_equal :STATUS_CODE_ERROR, failed_query.status.code + exception = jobs.flat_map(&:events).find { |event| event.name == "exception" } + assert exception, "job failure has no exported exception event" + attributes = exception.attributes.to_h { |attribute| [ attribute.key, attribute.value.string_value ] } + assert_equal "RuntimeError", attributes.fetch("exception.type") + assert_includes attributes.fetch("exception.message"), "job-failure-diagnostic-canary" + assert attributes.fetch("exception.stacktrace").present? + assert_equal %w[exception.message exception.stacktrace exception.type], attributes.keys.sort + assert spans.flat_map(&:attributes).none? { |attribute| attribute.key.match?(/bind|parameter/) } + end + + test "sensitive collection setting remains fixed until restart" do + [ [ "true", "false" ], [ "false", "true" ] ].each do |startup, changed| + result = run_traced_request( + configuration: { "REPLICA_OTEL_INCLUDE_SENSITIVE_DATA" => startup }, + change_sensitive_setting: changed, + failing_job: true + ) + spans = decoded_spans(result) + jobs = spans.select { |span| span.name.include?("TracingProbeJob") } + assert_operator jobs.size, :>=, 2 + if startup == "true" + assert jobs.all? { |job| span_attributes(job).key?("replica.job.arguments") } + assert spans.any? { |span| span_attributes(span).key?("db.query.text") } + assert jobs.any? { |job| job.events.any? { |event| event.name == "exception" } } + else + assert jobs.none? { |job| span_attributes(job).key?("replica.job.arguments") } + assert spans.none? { |span| span_attributes(span).key?("db.query.text") } + assert jobs.all? { |job| job.events.empty? } + end + end + end + + test "untraced exact hosts suppress HTTP spans and propagation without suppressing other hosts or jobs" do + result = run_traced_request(configuration: { + "REPLICA_OTEL_UNTRACED_HOSTS" => " upstream.example, LocalHost,127.0.0.1,::1 " + }) + + refute result.fetch("traceparent_propagated"), "mixed-case untraced host received trace context" + refute result.fetch("tracestate_propagated") + refute result.fetch("baggage_propagated") + assert result.fetch("other_traceparent_propagated"), "hostname subdomain was incorrectly suppressed" + spans = decoded_spans(result) + http = spans.select { |span| span_attributes(span).key?("http.method") || span_attributes(span).key?("http.request.method") } + refute http.any? { |span| span.kind == :SPAN_KIND_CLIENT && span_attributes(span).values.any? { |value| value.string_value&.casecmp?("upstream.example") } } + assert http.any? { |span| span.kind == :SPAN_KIND_CLIENT && span_attributes(span).values.any? { |value| value.string_value == "sub.upstream.example" } } + server = spans.find { |span| span.kind == :SPAN_KIND_SERVER } + jobs = spans.select { |span| span.name.include?("TracingProbeJob") } + assert_operator jobs.size, :>=, 2 + assert jobs.all? { |span| span.trace_id == server.trace_id } + end + + test "an unavailable collector does not interrupt a request or serialized job and flush is bounded" do + result = run_traced_request(collector_failure: true) + + assert_equal 200, result.fetch("status") + assert_equal "completed", result.fetch("body") + assert result.fetch("job_performed") + assert_operator result.fetch("export_attempts"), :>, 0 + assert_operator result.fetch("request_seconds"), :<, 5 + assert_operator result.fetch("flush_seconds"), :<, 5 + end + + test "a forked worker exports new job spans without replaying its parents pending request" do + result = run_traced_request(fork_worker: true) + + assert_equal 0, result.fetch("worker_exitstatus") + assert result.fetch("worker_performed") + spans = result.fetch("worker_payloads").flat_map do |payload| + request = Opentelemetry::Proto::Collector::Trace::V1::ExportTraceServiceRequest.decode \ + Base64.strict_decode64(payload) + request.resource_spans.flat_map(&:scope_spans).flat_map(&:spans) + end + assert spans.any? { |span| span.name.include?("TracingProbeJob") }, "forked worker did not export" + refute spans.any? { |span| span.kind == :SPAN_KIND_SERVER }, "child re-exported its parent's request" + spans.each { |span| assert_empty span.events } + end + + private + + def decoded_spans(result) + result.fetch("payloads").flat_map do |payload| + request = Opentelemetry::Proto::Collector::Trace::V1::ExportTraceServiceRequest.decode \ + Base64.strict_decode64(payload) + request.resource_spans.flat_map(&:scope_spans).flat_map(&:spans) + end + end + + def span_attributes(span) + span.attributes.to_h { |attribute| [ attribute.key, attribute.value ] } + end + + def resource_service_names(result) + result.fetch("payloads").flat_map do |payload| + request = Opentelemetry::Proto::Collector::Trace::V1::ExportTraceServiceRequest.decode \ + Base64.strict_decode64(payload) + request.resource_spans.filter_map do |resource_spans| + attribute = resource_spans.resource.attributes.find { |candidate| candidate.key == "service.name" } + attribute&.value&.string_value + end + end.uniq + end + + def run_traced_request(collector_failure: false, fork_worker: false, configuration: {}, failing_job: false, change_sensitive_setting: nil) + environment = ENV.keys.grep(/\A(?:OTEL_|REPLICA_OTEL_)/).to_h { |key| [ key, nil ] } + environment.merge! \ + "SECRET_KEY_BASE_DUMMY" => nil, + "RAILS_ENV" => "test", + "OTEL_TRACES_EXPORTER" => "otlp", + "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT" => "http://collector.example/v1/traces", + "OTEL_EXPORTER_OTLP_TIMEOUT" => "1", + "OTEL_BSP_SCHEDULE_DELAY" => "60000", + "OTEL_BSP_EXPORT_TIMEOUT" => "1000" + environment.merge!(configuration) + script = <<~RUBY + require "webmock" + require "json" + require "base64" + WebMock.enable! + WebMock.disable_net_connect! + require_relative "config/environment" + Rails.application.config.filter_parameters += [ :customer_secret ] + ENV["REPLICA_OTEL_INCLUDE_SENSITIVE_DATA"] = #{change_sensitive_setting.inspect} unless #{change_sensitive_setting.nil?} + payloads = [] + export_attempts = 0 + baggage_propagated = false + tracestate_propagated = false + traceparent_propagated = false + other_traceparent_propagated = false + intake = WebMock.stub_request(:post, "http://collector.example/v1/traces") + if #{collector_failure} + intake.to_return do + export_attempts += 1 + raise Errno::ECONNREFUSED + end + else + intake.to_return do |request| + export_attempts += 1 + payloads << Base64.strict_encode64(request.body) + { status: 200, body: "" } + end + end + WebMock.stub_request(:get, "http://UpStReAm.ExAmPlE/request-private-canary?token=request-private-canary") + .to_return do |request| + baggage_propagated ||= request.headers.key?("Baggage") + tracestate_propagated ||= request.headers.key?("Tracestate") + traceparent_propagated ||= request.headers.key?("Traceparent") + { status: 200, body: "request-private-canary" } + end + WebMock.stub_request(:get, "http://sub.upstream.example/") + .to_return do |request| + other_traceparent_propagated ||= request.headers.key?("Traceparent") + { status: 200, body: "" } + end + class TracingProbeJob < ActiveJob::Base + def perform(diagnostic, values) + raise "job argument changed" unless diagnostic == "argument-diagnostic-canary" && + values.fetch("email") == "identity-email-canary@example.test" && + values.fetch("name") == "identity-name-canary" && + values.fetch("password") == "password-canary" && + values.fetch("token") == "token-canary" && + values.fetch("customer_secret") == "app-secret-canary" && + values.fetch("authorization") == "authorization-canary" && + values.fetch("cookies") == "cookies-canary" && + values.fetch("on") == Date.new(2026, 10, 19) + raise "job-failure-diagnostic-canary" if #{failing_job} + self.class.performed = true + end + class_attribute :performed, default: false + end + TracingProbeJob.queue_adapter = :test + class TracingProbeController < ActionController::Base + def create + ActiveRecord::Base.connection.select_value("SELECT 'sql-diagnostic-canary'") + if #{failing_job} + begin + ActiveRecord::Base.connection.select_value("SELECT * FROM missing_diagnostic_table") + rescue ActiveRecord::StatementInvalid + end + end + Net::HTTP.get(URI("http://UpStReAm.ExAmPlE/request-private-canary?token=request-private-canary")) + Net::HTTP.get(URI("http://sub.upstream.example/")) + TracingProbeJob.set(priority: 3).perform_later("argument-diagnostic-canary", { + "email" => "identity-email-canary@example.test", + "name" => "identity-name-canary", + "password" => "password-canary", + "token" => "token-canary", + "customer_secret" => "app-secret-canary", + "authorization" => "authorization-canary", + "cookies" => "cookies-canary", + "on" => Date.new(2026, 10, 19) + }) + queued = TracingProbeJob.queue_adapter.enqueued_jobs.shift.select { |key, _| key.is_a?(String) } + serialized = JSON.parse(JSON.generate(queued)) + OpenTelemetry::Context.with_current(OpenTelemetry::Context.empty) do + begin + ActiveJob::Base.execute(serialized) + rescue RuntimeError => error + raise unless #{failing_job} && error.message == "job-failure-diagnostic-canary" + end + end + render plain: "completed" + end + end + Rails.application.routes.draw do + post "/tracing-probe", to: "tracing_probe#create" + get "/up", to: "rails/health#show" + end + OpenTelemetry::Context.with_current(OpenTelemetry::Context.empty) do + ActiveRecord::Base.connection.select_value("SELECT 'unparented-sql-diagnostic-canary'") + end + started = Process.clock_gettime(Process::CLOCK_MONOTONIC) + request = Rack::MockRequest.env_for \ + "/tracing-probe?token=request-private-canary", + method: "POST", + "HTTP_AUTHORIZATION" => "Bearer private-value-canary", + "HTTP_BAGGAGE" => "secret=request-private-canary", + "HTTP_TRACEPARENT" => "00-1234567890abcdef1234567890abcdef-1234567890abcdef-01", + "HTTP_TRACESTATE" => "vendor=request-private-canary", + input: "request-private-canary", + "CONTENT_TYPE" => "text/plain" + status, headers, body = Rails.application.call(request) + response_body = body.each.to_a.join + body.close if body.respond_to?(:close) + request_seconds = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + health_status, _, health_body = Rails.application.call(Rack::MockRequest.env_for("/up")) + health_body.each { |_part| } + health_body.close if health_body.respond_to?(:close) + worker = {} + if #{fork_worker} + reader, writer = IO.pipe + pid = fork do + reader.close + payloads.clear + TracingProbeJob.performed = false + TracingProbeJob.perform_now("argument-diagnostic-canary", { + "email" => "identity-email-canary@example.test", + "name" => "identity-name-canary", + "password" => "password-canary", + "token" => "token-canary", + "customer_secret" => "app-secret-canary", + "authorization" => "authorization-canary", + "cookies" => "cookies-canary", + "on" => Date.new(2026, 10, 19) + }) + OpenTelemetry.tracer_provider.force_flush(timeout: 1) + OpenTelemetry.tracer_provider.shutdown(timeout: 1) + writer.write(JSON.generate(payloads: payloads, performed: TracingProbeJob.performed)) + writer.close + exit! 0 + end + writer.close + worker = JSON.parse(reader.read) + reader.close + _, worker_status = Process.wait2(pid) + worker["exitstatus"] = worker_status.exitstatus + end + started = Process.clock_gettime(Process::CLOCK_MONOTONIC) + OpenTelemetry.tracer_provider.force_flush(timeout: 1) + flush_seconds = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + result = JSON.generate \ + status: status, + body: response_body, + job_performed: TracingProbeJob.performed, + health_status: health_status, + baggage_propagated: baggage_propagated, + tracestate_propagated: tracestate_propagated, + traceparent_propagated: traceparent_propagated, + other_traceparent_propagated: other_traceparent_propagated, + payloads: payloads, + export_attempts: export_attempts, + request_seconds: request_seconds, + flush_seconds: flush_seconds, + worker_payloads: worker["payloads"], + worker_performed: worker["performed"], + worker_exitstatus: worker["exitstatus"] + puts "HUB_TRACING_RESULT=" + result + RUBY + output, error, status = Open3.capture3 \ + environment, + RbConfig.ruby, + "-rbundler/setup", + "-e", + script, + chdir: Rails.root + assert status.success?, "Tracing subprocess failed:\n#{output}\n#{error}" + result = output.lines.find { |line| line.start_with?("HUB_TRACING_RESULT=") } + assert result, "Tracing subprocess returned no result:\n#{output}\n#{error}" + JSON.parse(result.delete_prefix("HUB_TRACING_RESULT=")) + .merge("diagnostics" => "#{output}\n#{error}") + end +end diff --git a/hub/test/lib/hub_tracing_batch_span_processor_test.rb b/hub/test/lib/hub_tracing_batch_span_processor_test.rb new file mode 100644 index 00000000..f209a4fa --- /dev/null +++ b/hub/test/lib/hub_tracing_batch_span_processor_test.rb @@ -0,0 +1,161 @@ +require "test_helper" +require "opentelemetry/sdk" +require_relative "../../lib/hub_tracing/exporter" +require_relative "../../lib/hub_tracing/batch_span_processor" + +class HubTracingBatchSpanProcessorTest < ActiveSupport::TestCase + self.use_transactional_tests = false + + test "force flush times out while a background batch holds the export lock" do + entered = Queue.new + release = Queue.new + exports = 0 + exporter = Object.new + exporter.define_singleton_method(:export) do |_, timeout:| + exports += 1 + if exports == 1 + entered << true + release.pop + end + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + exporter.define_singleton_method(:force_flush) do |timeout:| + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + exporter.define_singleton_method(:shutdown) do |timeout:| + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + processor = HubTracing::BatchSpanProcessor.new \ + exporter, + exporter_timeout: 1_000, + schedule_delay: 1, + max_queue_size: 10, + max_export_batch_size: 1 + provider = OpenTelemetry::SDK::Trace::TracerProvider.new + provider.add_span_processor(processor) + tracer = provider.tracer("hub-tracing-batch-processor-test") + tracer.start_span("in flight").finish + Timeout.timeout(1) { entered.pop } + tracer.start_span("queued").finish + + flush = Thread.new { provider.force_flush(timeout: 0.01) } + + assert flush.join(0.2), "force flush waited for the background export" + assert_equal OpenTelemetry::SDK::Trace::Export::TIMEOUT, flush.value + assert_equal 1, exports + ensure + release&.push(true) + flush&.join(1) + provider&.shutdown(timeout: 1) + end + + test "fork reset runs once before concurrent child export flush and shutdown" do + entered = Queue.new + release = Queue.new + factory_entered = Queue.new + release_factory = Queue.new + calls = [] + factories = 0 + blocked = false + processor_errors = [] + metrics = Object.new + metrics.define_singleton_method(:add_to_counter) do |name, increment: nil, labels: {}| + processor_errors << labels["reason"] if name == "otel.bsp.error" + end + metrics.define_singleton_method(:observe_value) do |name, value:| + end + exporter = HubTracing::Exporter.new do + identifier = factories += 1 + if identifier > 1 + factory_entered << identifier + release_factory.pop + end + Object.new.tap do |delegate| + delegate.define_singleton_method(:export) do |_, timeout:| + calls << [ identifier, :export ] + unless blocked + blocked = true + entered << true + release.pop + end + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + delegate.define_singleton_method(:force_flush) do |timeout:| + calls << [ identifier, :force_flush ] + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + delegate.define_singleton_method(:shutdown) do |timeout:| + calls << [ identifier, :shutdown ] + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + end + end + processor = HubTracing::BatchSpanProcessor.new \ + exporter, + exporter_timeout: 1_000, + schedule_delay: 1, + max_queue_size: 10, + max_export_batch_size: 1, + start_thread_on_boot: false, + metrics_reporter: metrics + provider = OpenTelemetry::SDK::Trace::TracerProvider.new + provider.add_span_processor(processor) + reader, writer = IO.pipe + pid = fork do + reader.close + result = begin + tracer = provider.tracer("hub-tracing-batch-processor-fork-test") + span = tracer.start_span("child") + finish = Thread.new { span.finish } + first_flush = Thread.new { provider.force_flush(timeout: 1) } + Timeout.timeout(1) { factory_entered.pop } + sleep 0.05 + 2.times { release_factory << true } + Timeout.timeout(1) { entered.pop } + concurrent_flush = Thread.new { provider.force_flush(timeout: 1) } + Timeout.timeout(1) { Thread.pass until concurrent_flush.status == "sleep" } + release << true + Timeout.timeout(1) { finish.value } + first_flush_result = Timeout.timeout(1) { first_flush.value } + concurrent_flush_result = Timeout.timeout(1) { concurrent_flush.value } + shutdown_result = provider.shutdown(timeout: 1) + { + factories:, + calls:, + processor_errors:, + first_flush_result:, + concurrent_flush_result:, + shutdown_result: + } + rescue StandardError => error + { error: [ error.class.name, error.message ] } + ensure + 2.times { release_factory << true } + release << true + end + writer.write(Marshal.dump(result)) + writer.close + exit! 0 + end + writer.close + child = Marshal.load(reader.read) + reader.close + _, status = Process.wait2(pid) + + assert status.success? + refute child.key?(:error), child[:error]&.join(": ") + assert_equal 2, child.fetch(:factories) + assert_empty child.fetch(:processor_errors) + assert child.fetch(:calls).all? { |identifier, _| identifier == 2 } + assert_includes child.fetch(:calls), [ 2, :export ] + assert_includes child.fetch(:calls), [ 2, :force_flush ] + assert_includes child.fetch(:calls), [ 2, :shutdown ] + assert_equal OpenTelemetry::SDK::Trace::Export::SUCCESS, child.fetch(:first_flush_result) + assert_equal OpenTelemetry::SDK::Trace::Export::SUCCESS, child.fetch(:concurrent_flush_result) + assert_equal OpenTelemetry::SDK::Trace::Export::SUCCESS, child.fetch(:shutdown_result) + ensure + release_factory&.push(true) + release&.push(true) + provider&.shutdown(timeout: 1) + end +end diff --git a/hub/test/lib/hub_tracing_data_filter_test.rb b/hub/test/lib/hub_tracing_data_filter_test.rb new file mode 100644 index 00000000..54829094 --- /dev/null +++ b/hub/test/lib/hub_tracing_data_filter_test.rb @@ -0,0 +1,81 @@ +require "test_helper" +require "json" +require_relative "../../lib/hub_tracing/data_filter" + +class HubTracingDataFilterTest < ActiveSupport::TestCase + test "opt in retains identities while preserving credential cookie and application filters" do + original_filters = Rails.application.config.filter_parameters + Rails.application.config.filter_parameters = [ + :email, "email", :password, :token, :customer_secret, :authorization, /private_email/, + ->(key, value) { value.replace("[APP FILTERED]") if key == "internal_note" } + ] + filter = HubTracing::DataFilter.new + arguments = [ + { + "email" => "person@example.test", + "name" => "Example Person", + "login" => "example-person", + "password" => "password-canary", + "token" => "token-canary", + "customer_secret" => "app-secret-canary", + "authorization" => "authorization-canary", + "cookies" => "cookies-canary", + "private_email" => "private@example.test", + "internal_note" => "note-canary" + } + ] + original = Marshal.load(Marshal.dump(arguments)) + + result = JSON.parse(filter.job_arguments(arguments)).first + + assert_equal "person@example.test", result.fetch("email") + assert_equal "Example Person", result.fetch("name") + assert_equal "example-person", result.fetch("login") + %w[password token customer_secret authorization cookies private_email].each do |key| + assert_equal "[FILTERED]", result.fetch(key) + end + assert_equal "[APP FILTERED]", result.fetch("internal_note") + assert_equal original, arguments + ensure + Rails.application.config.filter_parameters = original_filters + end + + test "constructed filters retain identity opt in and credentials after Rails replaces its source with precompiled rules" do + original_filters = Rails.application.config.filter_parameters + Rails.application.config.filter_parameters = [ :email, :password, :token, :customer_secret ] + filter = HubTracing::DataFilter.new + source = Rails.application.config.filter_parameters + source.replace(ActiveSupport::ParameterFilter.precompile_filters(source)) + arguments = [ + { + "email" => "person@example.test", + "name" => "Example Person", + "password" => "password-canary", + "token" => "token-canary", + "customer_secret" => "app-secret-canary" + } + ] + + result = JSON.parse(filter.job_arguments(arguments)).first + + assert_equal "person@example.test", result.fetch("email") + assert_equal "Example Person", result.fetch("name") + %w[password token customer_secret].each { |key| assert_equal "[FILTERED]", result.fetch(key) } + ensure + Rails.application.config.filter_parameters = original_filters + end + + test "sensitive opt in does not remove an application rule for another identity field" do + original_filters = Rails.application.config.filter_parameters + Rails.application.config.filter_parameters = [ :email, :name, /email/ ] + arguments = [ { "email" => "person@example.test", "name" => "Example Person", "login" => "example-person" } ] + + result = JSON.parse(HubTracing::DataFilter.new.job_arguments(arguments)).first + + assert_equal "[FILTERED]", result.fetch("email") + assert_equal "[FILTERED]", result.fetch("name") + assert_equal "example-person", result.fetch("login") + ensure + Rails.application.config.filter_parameters = original_filters + end +end diff --git a/hub/test/lib/hub_tracing_test.rb b/hub/test/lib/hub_tracing_test.rb new file mode 100644 index 00000000..cbb3fe0f --- /dev/null +++ b/hub/test/lib/hub_tracing_test.rb @@ -0,0 +1,595 @@ +require "test_helper" +require "open3" +require "socket" +require "opentelemetry/sdk" +require "opentelemetry/exporter/otlp" +require "opentelemetry/sdk/trace/export/in_memory_span_exporter" +require_relative "../../lib/hub_tracing/exporter" + +class HubTracingTest < ActiveSupport::TestCase + self.use_transactional_tests = false + + test "default disabled and dummy builds do not load the tracing SDK" do + [ + {}, + { "OTEL_TRACES_EXPORTER" => "none" }, + { "OTEL_TRACES_EXPORTER" => "otlp", "OTEL_SDK_DISABLED" => "true" }, + { "OTEL_TRACES_EXPORTER" => "otlp", "SECRET_KEY_BASE_DUMMY" => "1" }, + { "REPLICA_OTEL_INCLUDE_SENSITIVE_DATA" => "invalid", "REPLICA_OTEL_TRUST_INCOMING_CONTEXT" => "invalid" } + ].each do |configuration| + output, error, status = run_startup(configuration, <<~RUBY) + HubTracing.start! + abort "SDK initialized while disabled" if defined?(OpenTelemetry::SDK) + puts "disabled" + RUBY + + assert status.success?, error + assert_equal "disabled", output.strip + end + end + + test "disabled Rails eager loading does not load tracing SDK or support files" do + environment = ENV.keys.grep(/\A(?:OTEL_|REPLICA_OTEL_)/).to_h { |key| [ key, nil ] } + .merge("RAILS_ENV" => "test", "SECRET_KEY_BASE_DUMMY" => nil) + script = <<~RUBY + require_relative "config/environment" + Rails.application.eager_load! + abort "tracing SDK loaded while disabled" if defined?(OpenTelemetry::SDK) + support_files = $LOADED_FEATURES.grep(%r{/lib/hub_tracing/}) + abort "tracing support files loaded while disabled: \#{support_files.join(", ")}" if support_files.any? + %i[Exporter DataFilter Propagator].each do |name| + abort "\#{name} loaded while tracing disabled" if HubTracing.const_defined?(name, false) + end + puts "eager loaded without tracing" + RUBY + output, error, status = Open3.capture3 \ + environment, + RbConfig.ruby, + "-rbundler/setup", + "-e", + script, + chdir: Rails.root + + assert status.success?, "#{output}\n#{error}" + assert_includes output, "eager loaded without tracing" + end + + test "invalid enabled configuration fails without printing endpoint credentials" do + [ + { "OTEL_EXPORTER_OTLP_ENDPOINT" => nil }, + { "OTEL_EXPORTER_OTLP_ENDPOINT" => "https://user:credential-canary@collector.example" }, + { "OTEL_EXPORTER_OTLP_ENDPOINT" => "https://collector.example?token=credential-canary" }, + { "OTEL_EXPORTER_OTLP_ENDPOINT" => "https://collector.example/#credential-canary" }, + { "OTEL_EXPORTER_OTLP_ENDPOINT" => "https://[credential-canary" }, + { "OTEL_EXPORTER_OTLP_ENDPOINT" => "file:///credential-canary" }, + { "OTEL_EXPORTER_OTLP_ENDPOINT" => "http:///v1/traces" }, + { "OTEL_EXPORTER_OTLP_ENDPOINT" => "https://:4318/v1/traces" }, + { "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT" => "http://collector.example" }, + { "OTEL_EXPORTER_OTLP_PROTOCOL" => "grpc" }, + { "OTEL_EXPORTER_OTLP_TRACES_HEADERS" => "Authorization=credential-canary%ZZ" }, + { "OTEL_EXPORTER_OTLP_HEADERS" => "Authorization=credential-canary%ZZ" }, + { "OTEL_TRACES_EXPORTER" => "console,otlp" }, + { "REPLICA_OTEL_INCLUDE_SENSITIVE_DATA" => "credential-canary" }, + { "REPLICA_OTEL_INCLUDE_SENSITIVE_DATA" => "" }, + { "REPLICA_OTEL_INCLUDE_SENSITIVE_DATA" => "1" }, + { "REPLICA_OTEL_TRUST_INCOMING_CONTEXT" => "credential-canary" }, + { "REPLICA_OTEL_TRUST_INCOMING_CONTEXT" => "" }, + { "REPLICA_OTEL_TRUST_INCOMING_CONTEXT" => "yes" }, + { "REPLICA_OTEL_UNTRACED_HOSTS" => "" }, + { "REPLICA_OTEL_UNTRACED_HOSTS" => "upstream.example," }, + { "REPLICA_OTEL_UNTRACED_HOSTS" => ",upstream.example" }, + { "REPLICA_OTEL_UNTRACED_HOSTS" => "upstream.example, ,other.example" }, + { "REPLICA_OTEL_UNTRACED_HOSTS" => "https://upstream.example" }, + { "REPLICA_OTEL_UNTRACED_HOSTS" => "upstream.example:443" }, + { "REPLICA_OTEL_UNTRACED_HOSTS" => "upstream.example/path" }, + { "REPLICA_OTEL_UNTRACED_HOSTS" => "-upstream.example" }, + { "REPLICA_OTEL_UNTRACED_HOSTS" => "upstream..example" }, + { "REPLICA_OTEL_UNTRACED_HOSTS" => "999.0.0.1" }, + { "REPLICA_OTEL_UNTRACED_HOSTS" => "2001:db8::gggg" }, + { "REPLICA_OTEL_UNTRACED_HOSTS" => "[::1]" }, + { "REPLICA_OTEL_UNTRACED_HOSTS" => "::1/128" } + ].each do |invalid| + header_name = invalid.keys.find { _1.match?(/\AOTEL_EXPORTER_OTLP(?:_TRACES)?_HEADERS\z/) } + endpoint_name = invalid.keys.find { _1.match?(/\AOTEL_EXPORTER_OTLP(?:_TRACES)?_ENDPOINT\z/) } + expected_message = if header_name + "#{header_name} must contain comma-separated URL-encoded key=value headers" + elsif endpoint_name && invalid.fetch(endpoint_name).nil? + "Hub tracing requires OTEL_EXPORTER_OTLP_TRACES_ENDPOINT or OTEL_EXPORTER_OTLP_ENDPOINT" + elsif endpoint_name == "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT" && + invalid.fetch(endpoint_name) == "http://collector.example" + "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT must include a request path" + end + expected_prefix = "#{endpoint_name} " if endpoint_name && !expected_message + configuration = { + "OTEL_TRACES_EXPORTER" => "otlp", + "OTEL_EXPORTER_OTLP_ENDPOINT" => "http://collector.example" + }.merge(invalid) + output, error, status = run_startup(configuration, <<~RUBY) + begin + HubTracing.start! + rescue ArgumentError => error + expected_message = #{expected_message.inspect} + expected_prefix = #{expected_prefix.inspect} + abort "unexpected configuration error: \#{error.message}" \ + if expected_message && error.message != expected_message + abort "configuration error did not name #{endpoint_name}: \#{error.message}" \ + if expected_prefix && !error.message.start_with?(expected_prefix) + abort "configuration error retained its cause" if error.cause + puts error.full_message(highlight: false) + exit 42 + end + RUBY + + assert_equal 42, status.exitstatus, "#{configuration.keys}: #{output} #{error}" + refute_includes output + error, "credential-canary" + end + end + + test "export deadline interrupts a sleeping exporter even when it rescues ordinary errors" do + delegate = Object.new + entered = false + completed = false + discarded = false + delegate.define_singleton_method(:export) do |_, timeout:| + entered = true + sleep 0.25 + completed = true + OpenTelemetry::SDK::Trace::Export::SUCCESS + rescue StandardError + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + delegate.define_singleton_method(:shutdown) do |timeout:| + discarded = true + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + exporter = HubTracing::Exporter.new { delegate } + + assert_equal OpenTelemetry::SDK::Trace::Export::TIMEOUT, exporter.export([], timeout: 0.01) + assert entered, "delegate was never called" + refute completed, "delegate continued beyond the export deadline" + assert discarded, "timed-out delegate was not discarded" + end + + test "flush times out while another export holds the lifecycle lock" do + entered = Queue.new + release = Queue.new + flush_calls = 0 + delegate = Object.new + delegate.define_singleton_method(:export) do |_, timeout:| + entered << true + release.pop + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + delegate.define_singleton_method(:force_flush) do |timeout:| + flush_calls += 1 + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + delegate.define_singleton_method(:shutdown) do |timeout:| + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + exporter = HubTracing::Exporter.new { delegate } + export = Thread.new { exporter.export([], timeout: nil) } + Timeout.timeout(1) { entered.pop } + + flush = Thread.new { exporter.force_flush(timeout: 0.01) } + + assert flush.join(0.2), "flush waited beyond its caller timeout" + assert_equal OpenTelemetry::SDK::Trace::Export::TIMEOUT, flush.value + assert_equal 0, flush_calls + ensure + release&.push(true) + export&.join(1) + flush&.join(1) + exporter&.shutdown(timeout: 1) + end + + test "flush passes only its remaining timeout after waiting for the lifecycle lock" do + entered = Queue.new + release = Queue.new + flush_started = Queue.new + flush_timeouts = Queue.new + delegate = Object.new + delegate.define_singleton_method(:export) do |_, timeout:| + entered << true + release.pop + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + delegate.define_singleton_method(:force_flush) do |timeout:| + flush_timeouts << timeout + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + delegate.define_singleton_method(:shutdown) do |timeout:| + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + exporter = HubTracing::Exporter.new { delegate } + export = Thread.new { exporter.export([], timeout: nil) } + Timeout.timeout(1) { entered.pop } + flush = Thread.new do + flush_started << true + exporter.force_flush(timeout: 1) + end + Timeout.timeout(1) { flush_started.pop } + Timeout.timeout(1) { Thread.pass until flush.status == "sleep" } + sleep 0.05 + release << true + + assert_equal OpenTelemetry::SDK::Trace::Export::SUCCESS, Timeout.timeout(1) { flush.value } + remaining = Timeout.timeout(1) { flush_timeouts.pop } + assert_operator remaining, :>, 0 + assert_operator remaining, :<, 1 + ensure + release&.push(true) + export&.join(1) + flush&.join(1) + exporter&.shutdown(timeout: 1) + end + + test "exporter replaces a timed-out OTLP connection before the next batch" do + collector = TCPServer.new("127.0.0.1", 0) + accepted_connections = Queue.new + completed_connections = Queue.new + first_request = Queue.new + release_first_response = Queue.new + server = Thread.new do + 2.times do |index| + client = nil + client = collector.accept + accepted_connections << index + read_http_request(client) + if index.zero? + first_request << true + release_first_response.pop + end + client.write("HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + rescue Errno::EPIPE, IOError + ensure + client&.close + completed_connections << index if client + end + end + factories = 0 + endpoint = "http://127.0.0.1:#{collector.local_address.ip_port}/v1/traces" + exporter = HubTracing::Exporter.new do + factories += 1 + OpenTelemetry::Exporter::OTLP::Exporter.new(endpoint:) + end + originals, provider, _sensitive = traced_spans + + first_export = Thread.new { exporter.export(originals, timeout: 0.05) } + Timeout.timeout(1) { first_request.pop } + assert_equal 0, Timeout.timeout(1) { accepted_connections.pop } + assert_equal OpenTelemetry::SDK::Trace::Export::TIMEOUT, Timeout.timeout(1) { first_export.value } + release_first_response << true + assert_equal OpenTelemetry::SDK::Trace::Export::SUCCESS, exporter.export(originals, timeout: 1) + assert_equal 1, Timeout.timeout(1) { accepted_connections.pop } + assert_equal [ 0, 1 ], 2.times.map { Timeout.timeout(1) { completed_connections.pop } } + Timeout.timeout(1) { server.value } + assert_equal 2, factories + ensure + release_first_response&.push(true) + first_export&.join(1) + exporter&.shutdown(timeout: 1) + provider&.shutdown(timeout: 1) + collector&.close + server&.join(1) + end + + test "forked exporters use independent keep-alive collector connections" do + collector = TCPServer.new("127.0.0.1", 0) + requests = Queue.new + server = Thread.new do + handlers = 2.times.map do |connection| + client = collector.accept + Thread.new do + (connection.zero? ? 2 : 1).times do + body = read_http_request(client) + requests << [ connection, decoded_spans(body).map(&:name) ] + client.write("HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: keep-alive\r\n\r\n") + end + ensure + client.close + end + end + handlers.each(&:value) + end + endpoint = "http://127.0.0.1:#{collector.local_address.ip_port}/v1/traces" + exporter = HubTracing::Exporter.new do + OpenTelemetry::Exporter::OTLP::Exporter.new(endpoint:) + end + originals, provider, _sensitive = traced_spans + template = originals.last + + assert_equal \ + OpenTelemetry::SDK::Trace::Export::SUCCESS, + exporter.export(named_batch(template, "before fork"), timeout: 1) + + reader, writer = IO.pipe + pid = fork do + reader.close + exporter.reset_after_fork + result = exporter.export(named_batch(template, "child after fork"), timeout: 1) + shutdown = exporter.shutdown(timeout: 1) + writer.write(Marshal.dump(export: result, shutdown:)) + writer.close + exit! 0 + end + writer.close + child = Marshal.load(reader.read) + reader.close + _, child_status = Process.wait2(pid) + + assert child_status.success? + assert_equal OpenTelemetry::SDK::Trace::Export::SUCCESS, child.fetch(:export) + assert_equal OpenTelemetry::SDK::Trace::Export::SUCCESS, child.fetch(:shutdown) + assert_equal \ + OpenTelemetry::SDK::Trace::Export::SUCCESS, + exporter.export(named_batch(template, "parent after fork"), timeout: 1) + assert_equal OpenTelemetry::SDK::Trace::Export::SUCCESS, exporter.shutdown(timeout: 1) + Timeout.timeout(1) { server.value } + + names_by_connection = Hash.new { |names, connection| names[connection] = [] } + 3.times do + connection, names = requests.pop + names_by_connection[connection].concat(names) + end + assert_equal [ "before fork", "parent after fork" ], names_by_connection.fetch(0) + assert_equal [ "child after fork" ], names_by_connection.fetch(1) + ensure + exporter&.shutdown(timeout: 1) + provider&.shutdown(timeout: 1) + collector&.close + server&.join(1) + end + + test "explicit fork reset replaces inherited delegates before flush and shutdown" do + %i[force_flush shutdown].each do |operation| + calls = [] + factories = 0 + exporter = HubTracing::Exporter.new do + identifier = factories += 1 + Object.new.tap do |delegate| + delegate.define_singleton_method(:force_flush) do |timeout:| + calls << [ identifier, :force_flush ] + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + delegate.define_singleton_method(:shutdown) do |timeout:| + calls << [ identifier, :shutdown ] + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + end + end + reader, writer = IO.pipe + pid = fork do + reader.close + exporter.reset_after_fork + result = exporter.public_send(operation, timeout: 1) + writer.write(Marshal.dump(factories:, calls:, result:)) + writer.close + exit! 0 + end + writer.close + child = Marshal.load(reader.read) + reader.close + _, status = Process.wait2(pid) + + assert status.success? + assert_equal 2, child.fetch(:factories) + assert_equal [ [ 2, operation ] ], child.fetch(:calls) + assert_equal OpenTelemetry::SDK::Trace::Export::SUCCESS, child.fetch(:result) + ensure + exporter&.shutdown(timeout: 1) + end + end + + test "expired export deadlines do not send spans" do + delegate = Object.new + calls = 0 + delegate.define_singleton_method(:export) do |_, timeout:| + calls += 1 + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + delegate.define_singleton_method(:shutdown) do |timeout:| + OpenTelemetry::SDK::Trace::Export::SUCCESS + end + exporter = HubTracing::Exporter.new { delegate } + + [ 0, -1 ].each do |timeout| + assert_equal OpenTelemetry::SDK::Trace::Export::TIMEOUT, exporter.export([], timeout:) + end + assert_equal 0, calls + end + + test "exporter removes sensitive diagnostics and unrelated private content by default" do + originals, provider, sensitive = traced_spans + exporter, requests, intake = exporter_requests + + assert_equal OpenTelemetry::SDK::Trace::Export::SUCCESS, exporter.export(originals, timeout: 1) + + assert_requested intake, times: 1 + payload = requests.first.body + %w[sql-diagnostic-canary argument-diagnostic-canary exception-diagnostic-canary].each do |diagnostic| + refute_includes payload, diagnostic + end + refute_includes payload, sensitive + decoded_spans(payload).zip(originals).each do |span, original| + attributes = span.attributes.to_h { |attribute| [ attribute.key, attribute.value ] } + assert_equal original.attributes.fetch("http.request.method"), attributes.fetch("http.request.method").string_value + assert_equal "sqlite", attributes.fetch("db.system.name").string_value + assert_equal "default", attributes.fetch("messaging.destination").string_value + assert_equal "TracingProbeJob", attributes.fetch("code.namespace").string_value + refute attributes.key?("db.query.text") + refute attributes.key?("replica.job.arguments") + refute attributes.key?("db.query.parameter.0") + assert_empty span.events + assert_empty span.links + assert_empty span.status.message + end + assert_equal "POST /#{sensitive}", originals.first.name + assert_equal sensitive, originals.first.events.first.name + assert_equal sensitive, originals.first.status.description + ensure + exporter&.shutdown(timeout: 1) + provider&.shutdown(timeout: 1) + end + + test "sensitive exporter retains intended diagnostics but removes unrelated private content" do + originals, provider, sensitive = traced_spans + exporter, requests, intake = exporter_requests(include_sensitive_data: true) + + assert_equal OpenTelemetry::SDK::Trace::Export::SUCCESS, exporter.export(originals, timeout: 1) + + assert_requested intake, times: 1 + assert_equal "application/x-protobuf", requests.first.headers["Content-Type"] + payload = requests.first.body + refute_includes payload, sensitive + exported = decoded_spans(payload) + assert_equal [ "POST /builds/:id", "GET /builds/:id" ], exported.map(&:name) + exported.zip(originals).each do |actual, original| + assert_equal original.trace_id, actual.trace_id + assert_equal original.span_id, actual.span_id + assert_equal original.start_timestamp, actual.start_time_unix_nano + assert_equal original.end_timestamp, actual.end_time_unix_nano + assert_empty actual.links + metadata = actual.attributes.to_h { |attribute| [ attribute.key, attribute.value ] } + assert_equal "SELECT 'sql-diagnostic-canary'", metadata.fetch("db.query.text").string_value + assert_equal \ + '["argument-diagnostic-canary",{"password":"[FILTERED]"}]', + metadata.fetch("replica.job.arguments").string_value + refute metadata.key?("db.query.parameter.0") + refute metadata.key?("http.request.header.authorization") + refute metadata.key?("http.request.body") + refute metadata.key?("url.full") + end + assert_equal [ "exception" ], exported.first.events.map(&:name) + event = exported.first.events.first + assert_equal originals.first.events.last.timestamp, event.time_unix_nano + expected_exception = { + "exception.type" => "RuntimeError", + "exception.message" => "exception-diagnostic-canary", + "exception.stacktrace" => "probe.rb:1:in `perform'" + } + assert_equal expected_exception, event.attributes.to_h { |attribute| [ attribute.key, attribute.value.string_value ] } + assert_empty exported.last.events + assert_equal :STATUS_CODE_ERROR, exported.first.status.code + assert_empty exported.first.status.message + ensure + exporter&.shutdown(timeout: 1) + provider&.shutdown(timeout: 1) + end + + private + + def traced_spans + memory = OpenTelemetry::SDK::Trace::Export::InMemorySpanExporter.new + provider = OpenTelemetry::SDK::Trace::TracerProvider.new + provider.add_span_processor(OpenTelemetry::SDK::Trace::Export::SimpleSpanProcessor.new(memory)) + tracer = provider.tracer("hub-tracing-test") + sensitive = "private-value-canary" + attributes = { + "http.request.method" => "POST", + "http.response.status_code" => 500, + "http.route" => "/builds/:id", + "http.method" => "POST", + "http.status_code" => 500, + "url.scheme" => "https", + "http.scheme" => "https", + "server.address" => "example.com", + "server.port" => 443, + "net.peer.name" => "example.com", + "net.peer.port" => 443, + "db.system.name" => "sqlite", + "db.query.text" => "SELECT 'sql-diagnostic-canary'", + "messaging.system" => "active_job", + "messaging.destination" => "default", + "messaging.message.id" => "job-id", + "messaging.active_job.adapter.name" => "ActiveJob::QueueAdapters::TestAdapter", + "messaging.active_job.message.provider_job_id" => "provider-id", + "messaging.active_job.message.priority" => "3", + "code.namespace" => "TracingProbeJob", + "code.function" => "perform", + "replica.job.arguments" => '["argument-diagnostic-canary",{"password":"[FILTERED]"}]', + "http.request.header.authorization" => "Bearer #{sensitive}", + "http.request.body" => sensitive, + "url.full" => "https://user:#{sensitive}@example.com/#{sensitive}?token=#{sensitive}", + "db.statement" => "SELECT '#{sensitive}'", + "db.query.parameter.0" => sensitive, + "job.arguments" => sensitive, + "db.system" => sensitive, + "db.operation" => sensitive, + "db.operation.name" => sensitive, + "messaging.operation" => sensitive, + "messaging.operation.name" => sensitive, + "messaging.operation.type" => sensitive, + "messaging.destination.name" => sensitive, + "messaging.destination_kind" => sensitive + } + link = OpenTelemetry::Trace::Link.new \ + OpenTelemetry::Trace::SpanContext.new, + { "secret" => sensitive } + span = tracer.start_span("POST /#{sensitive}", kind: :server, attributes:, links: [ link ]) + span.add_event(sensitive, attributes: { "exception.message" => sensitive }) + span.add_event \ + "exception", + attributes: { + "exception.type" => "RuntimeError", + "exception.message" => "exception-diagnostic-canary", + "exception.stacktrace" => "probe.rb:1:in `perform'", + "exception.escaped" => true, + "request.body" => sensitive + } + span.status = OpenTelemetry::Trace::Status.error(sensitive) + span.finish + client = tracer.start_span \ + "GET https://example.com/#{sensitive}", + kind: :client, + attributes: attributes.merge("http.request.method" => "GET") + client.finish + [ memory.finished_spans, provider, sensitive ] + end + + def exporter_requests(include_sensitive_data: false) + requests = [] + intake = stub_request(:post, "http://collector.example/v1/traces").with do |request| + requests << request + true + end.to_return(status: 200, body: "") + delegate = OpenTelemetry::Exporter::OTLP::Exporter.new(endpoint: "http://collector.example/v1/traces") + [ HubTracing::Exporter.new(include_sensitive_data:) { delegate }, requests, intake ] + end + + def decoded_spans(payload) + decoded = Opentelemetry::Proto::Collector::Trace::V1::ExportTraceServiceRequest.decode(payload) + decoded.resource_spans.flat_map(&:scope_spans).flat_map(&:spans) + end + + def read_http_request(socket) + request = +"" + request << socket.readpartial(1024) until request.include?("\r\n\r\n") + headers, body = request.split("\r\n\r\n", 2) + content_length = headers[/\r\nContent-Length:\s*(\d+)/i, 1].to_i + body << socket.read(content_length - body.bytesize) if body.bytesize < content_length + body + end + + def named_batch(template, name) + span = template.dup + span.name = name + span.attributes = {} + [ span ] + end + + def run_startup(configuration, script) + environment = ENV.keys.grep(/\A(?:OTEL_|REPLICA_OTEL_)/).to_h { |key| [ key, nil ] } + .merge("SECRET_KEY_BASE_DUMMY" => nil) + .merge(configuration) + Open3.capture3 \ + environment, + RbConfig.ruby, + "-rbundler/setup", + "-Ilib", + "-rhub_tracing", + "-e", + script, + chdir: Rails.root + end +end diff --git a/hub/test/models/github_client_batch_test.rb b/hub/test/models/github_client_batch_test.rb index ada61181..59307e87 100644 --- a/hub/test/models/github_client_batch_test.rb +++ b/hub/test/models/github_client_batch_test.rb @@ -40,4 +40,23 @@ class GithubClientBatchTest < ActiveSupport::TestCase end assert_equal 3, completed.size end + + test "restores the submitting tracing context in every worker" do + require "opentelemetry/sdk" + + provider = OpenTelemetry::SDK::Trace::TracerProvider.new + span = provider.tracer("github-client-batch-test").start_span("request") + + results = OpenTelemetry::Trace.with_span(span) do + GithubClient::Batch.map([ 3, 1, 2 ]) do |item| + [ item, OpenTelemetry::Trace.current_span.context.span_id ] + end + end + + assert_equal [ 3, 1, 2 ], results.map(&:first) + assert_equal [ span.context.span_id ] * 3, results.map(&:last) + ensure + span&.finish + provider&.shutdown + end end