diff --git a/Cargo.lock b/Cargo.lock index 5e204e6ad4d..4f98b50079f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3875,6 +3875,10 @@ dependencies = [ "num-integer", "num-traits", "object_store", + "opentelemetry", + "opentelemetry-http", + "opentelemetry-otlp", + "opentelemetry_sdk", "parking_lot", "petgraph 0.8.3", "portable-atomic", @@ -5847,6 +5851,76 @@ dependencies = [ "vcpkg", ] +[[package]] +name = "opentelemetry" +version = "0.33.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8c1f1a6ac9c875a4941e4c7c085b61e8f4daa326de03b4701768c3491a6ad9ee" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "pin-project-lite", + "thiserror 2.0.18", +] + +[[package]] +name = "opentelemetry-http" +version = "0.33.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "302249bd8ee804789faaa7b5b2cbf79bc61cb864714079c644e3828af5b7e811" +dependencies = [ + "async-trait", + "bytes", + "http 1.4.2", + "opentelemetry", + "reqwest 0.13.2", +] + +[[package]] +name = "opentelemetry-otlp" +version = "0.33.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0756af24841e367128ef31c831999937b8b5c5197d47914590d8f4e2e88c2e1a" +dependencies = [ + "http 1.4.2", + "httpdate", + "opentelemetry", + "opentelemetry-http", + "opentelemetry-proto", + "opentelemetry_sdk", + "prost", + "reqwest 0.13.2", + "thiserror 2.0.18", +] + +[[package]] +name = "opentelemetry-proto" +version = "0.33.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d576f5dddc3299341bcd85a04be17b2564844c0c9563214f02d85f2524d1afae" +dependencies = [ + "opentelemetry", + "opentelemetry_sdk", + "prost", +] + +[[package]] +name = "opentelemetry_sdk" +version = "0.33.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "372a912db2974d1990207006fd0e3aa0ef3fd4ad3be43c65cfaadf2206866061" +dependencies = [ + "futures-channel", + "futures-executor", + "futures-util", + "opentelemetry", + "percent-encoding", + "portable-atomic", + "rand 0.9.3", + "thiserror 2.0.18", +] + [[package]] name = "option-ext" version = "0.2.0" @@ -6957,6 +7031,7 @@ checksum = "ab3f43e3283ab1488b624b44b0e988d0acea0b3214e694730a055cb6b2efa801" dependencies = [ "base64", "bytes", + "futures-channel", "futures-core", "futures-util", "h2", diff --git a/Cargo.toml b/Cargo.toml index a956a514a95..321a0b0a566 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -80,6 +80,11 @@ indicatif = "0.18" Inflector = "0.11.3" itertools = "0.15.0" lazy_static = "1.5.0" +# Default features off: no metrics, no logs, no `tracing`-based internal logging. +opentelemetry = { version = "0.33.1", default-features = false, features = ["trace"] } +opentelemetry_sdk = { version = "0.33.1", default-features = false, features = ["trace"] } +opentelemetry-http = { version = "0.33.1", default-features = false } +opentelemetry-otlp = { version = "0.33.1", default-features = false, features = ["http-proto", "reqwest-blocking-client", "trace"] } prost = "0.14" prost-types = "0.14" redis = { version = "1.2.2", features = [ diff --git a/NEWS.md b/NEWS.md index 8068ecbc9fa..c99bfacdadf 100644 --- a/NEWS.md +++ b/NEWS.md @@ -12,6 +12,7 @@ $ docker pull graphprotocol/graph-node:v0.45.0 ### What's New +- **OpenTelemetry tracing.** Setting `GRAPH_OTEL_TRACING=true` exports traces over OTLP (`http/protobuf`) to any collector: one trace per processed block with the handler and, optionally, host function spans under it, one per batch written by the write queue, linked to the traces of the blocks it contains, and one per GraphQL request that joins the trace of an incoming `traceparent` header. `GRAPH_OTEL_SPANS` picks the span groups; the exporter and sampler are configured with the standard `OTEL_*` variables. Off by default, with negligible overhead when off. See [docs/tracing.md](docs/tracing.md). - **Postponed attribute index creation re-enabled, now built in the background.** `GRAPH_POSTPONE_ATTRIBUTE_INDEX_CREATION` (introduced in v0.44.0 but inert — hardcoded off) now actually defers non-critical attribute indexes until a new deployment nears chain head (`GRAPH_POSTPONE_INDEXES_CREATION_THRESHOLD`, default 10000 blocks from head). Index creation now runs in a background task instead of blocking block processing, and interrupted `CREATE INDEX CONCURRENTLY` runs are detected and rebuilt on retry instead of being silently skipped. ([#6608](https://github.com/graphprotocol/graph-node/pull/6608)) - **`ethereum.decodeParams` host function.** Decodes calldata/event data whose top-level type is a dynamic tuple (e.g. Gnosis Safe's `execTransaction`), which `ethereum.decode` cannot handle since it expects a single ABI value with a leading offset word. Gated behind apiVersion 0.0.10; `ethereum.decode` itself is unchanged. ([#6649](https://github.com/graphprotocol/graph-node/pull/6649)) diff --git a/chain/ethereum/src/runtime/runtime_adapter.rs b/chain/ethereum/src/runtime/runtime_adapter.rs index a5d68a761a7..322cb285be7 100644 --- a/chain/ethereum/src/runtime/runtime_adapter.rs +++ b/chain/ethereum/src/runtime/runtime_adapter.rs @@ -208,6 +208,14 @@ async fn ethereum_call( asc_get::<_, AscUnresolvedContractCall, _>(ctx.heap, wasm_ptr.into(), &ctx.gas, 0)? }; + ctx.metrics.stopwatch.set_attrs_with(|| { + [ + ("contract", call.contract_name.clone().into()), + ("function", call.function_name.clone().into()), + ("address", format!("{:#x}", call.contract_address).into()), + ] + }); + let result = eth_call( eth_adapter, call_cache, diff --git a/chain/ethereum/src/trigger.rs b/chain/ethereum/src/trigger.rs index b5d51d9a379..7d4c185af6f 100644 --- a/chain/ethereum/src/trigger.rs +++ b/chain/ethereum/src/trigger.rs @@ -22,6 +22,7 @@ use graph::runtime::HostExportError; use graph::runtime::asc_new; use graph::runtime::gas::GasCounter; use graph::semver::Version; +use graph::telemetry::AttrValue; use graph_runtime_wasm::module::ToAscPtr; use std::{cmp::Ordering, sync::Arc}; @@ -71,6 +72,36 @@ impl MappingTriggerTrait for MappingTrigger { None => String::new(), } } + + fn trace_attributes(&self) -> Vec<(&'static str, AttrValue)> { + match self { + MappingTrigger::Log { log, .. } => { + transaction_attributes(log.transaction_hash, log.log_index) + } + MappingTrigger::Call { call, .. } => { + transaction_attributes(call.transaction_hash, None) + } + MappingTrigger::Block { .. } => Vec::new(), + } + } +} + +/// The span attributes that locate a trigger inside its block. +fn transaction_attributes( + tx_hash: Option, + log_index: Option, +) -> Vec<(&'static str, AttrValue)> { + let mut attrs = Vec::new(); + if let Some(tx_hash) = tx_hash { + attrs.push(("tx_hash", format!("{tx_hash:#x}").into())); + } + if let Some(log_index) = log_index { + attrs.push(( + "log_index", + i64::try_from(log_index).unwrap_or(i64::MAX).into(), + )); + } + attrs } // Logging the block is too verbose, so this strips the block from the trigger for Debug. @@ -649,3 +680,26 @@ impl<'a> EthereumCallData<'a> { &self.call.to } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn transaction_attributes_name_the_transaction() { + let tx_hash = B256::from([0xab; 32]); + let attrs = transaction_attributes(Some(tx_hash), Some(7)); + assert_eq!( + attrs, + vec![ + ("tx_hash", AttrValue::from(format!("0x{}", "ab".repeat(32)))), + ("log_index", AttrValue::from(7i64)), + ] + ); + assert_eq!( + transaction_attributes(Some(tx_hash), None), + vec![("tx_hash", AttrValue::from(format!("0x{}", "ab".repeat(32))))] + ); + assert!(transaction_attributes(None, None).is_empty()); + } +} diff --git a/core/src/subgraph/runner/mod.rs b/core/src/subgraph/runner/mod.rs index 026be3dfc27..e876858d68e 100644 --- a/core/src/subgraph/runner/mod.rs +++ b/core/src/subgraph/runner/mod.rs @@ -42,6 +42,7 @@ use graph::prelude::{ }; use graph::schema::EntityKey; use graph::slog::{Logger, debug, error, info, o, trace, warn}; +use graph::telemetry::SpanGroup; use graph::util::lfu_cache::EvictStats; use graph::util::{backoff::ExponentialBackoff, lfu_cache::LfuCache}; use std::sync::Arc; @@ -578,9 +579,29 @@ where let block_start = Instant::now(); let action = { - let stopwatch = &self.metrics.stream.stopwatch; - let _section = stopwatch.start_section(PROCESS_BLOCK_SECTION_NAME); - self.process_block(block, cursor).await + let stopwatch = self.metrics.stream.stopwatch.cheap_clone(); + // Each block is its own trace. + let _section = stopwatch.start_root_section( + PROCESS_BLOCK_SECTION_NAME, + SpanGroup::Block, + || { + [ + ( + "subgraph_id", + self.inputs.deployment.hash.to_string().into(), + ), + ("block_number", block_ptr.number.into()), + ("block_hash", block_ptr.hash_hex().into()), + ("network", self.inputs.network.clone().into()), + ] + }, + &[], + ); + let action = self.process_block(block, cursor).await; + if let Err(err) = &action { + stopwatch.set_error_with(|| err.to_string()); + } + action }; let action = self.handle_action(block_start, block_ptr, action).await?; @@ -720,7 +741,11 @@ where // Transact entity operations into the store and update the // subgraph's block stream pointer - let _section = self.metrics.host.stopwatch.start_section("transact_block"); + let _section = self + .metrics + .host + .stopwatch + .start_section_with("transact_block", || [("entity_count", mods.len().into())]); let start = Instant::now(); // If a deterministic error has happened, make the PoI to be the only entity that'll be stored. @@ -1098,7 +1123,7 @@ where vid_gen, ); - let _section = self + let section = self .metrics .stream .stopwatch @@ -1134,6 +1159,10 @@ where Err(MappingError::PossibleReorg(e)) => return Err(ProcessingError::PossibleReorg(e)), } + // The later stages have sections of their own and are not part + // of trigger processing. + section.end(); + // Check if there are any datasources that have expired in this block. ie: the end_block // of that data source is equal to the block number of the current block. let has_expired_data_sources = self.inputs.end_blocks.contains(&block_ptr.number); @@ -1485,7 +1514,7 @@ where close_to_chain_head(block_ptr, &self.state.cached_head_ptr, CAUGHT_UP_DISTANCE); if is_caught_up { // Stop recording time-to-sync metrics. - self.metrics.stream.stopwatch.disable(); + self.metrics.stream.stopwatch.stop_recording(); } Ok(is_caught_up) } diff --git a/docs/environment-variables.md b/docs/environment-variables.md index 7a601e32e64..92b8741d546 100644 --- a/docs/environment-variables.md +++ b/docs/environment-variables.md @@ -160,6 +160,43 @@ those. value will not be cached. The default is 3. A value of 0 means that there is no limit on the size of a cache entry. +## OpenTelemetry tracing + +Tracing is off by default and adds one atomic load per instrumented +section when off. [tracing.md](./tracing.md) describes what is traced and +how to set up a collector. + +- `GRAPH_OTEL_TRACING`: set to `true` to export spans over OTLP. Defaults to + `false`. +- `GRAPH_OTEL_SPANS`: comma-separated list of the span groups to emit: + `block` (one trace per processed block and the runner sections under it), + `handler` (one span per trigger handler), `host_fn` (one span per WASM host + function call, which can be thousands per handler), `store` (write batches + and the SQL queries that serve GraphQL) and `graphql` (one trace per + GraphQL request). `all` and `none` are accepted. Defaults to + `block,handler,store,graphql`. + +The exporter speaks OTLP over `http/protobuf` and is configured with the +standard OpenTelemetry variables, which the SDK reads directly: + +- `OTEL_EXPORTER_OTLP_ENDPOINT`: base URL of the collector; traces go to + `/v1/traces`. Defaults to `http://localhost:4318`. + `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` overrides the full traces URL. +- `OTEL_EXPORTER_OTLP_PROTOCOL`: only `http/protobuf`, the default, is + supported. +- `OTEL_EXPORTER_OTLP_HEADERS`, `OTEL_EXPORTER_OTLP_TIMEOUT`: passed to the + collector as described in the OpenTelemetry specification. +- `OTEL_SERVICE_NAME`: the `service.name` resource attribute. Defaults to + `graph-node`. +- `OTEL_TRACES_SAMPLER` and `OTEL_TRACES_SAMPLER_ARG`: sampling is decided + once per trace at its root span. Use `parentbased_traceidratio` with an + argument between 0 and 1 to sample a fraction of blocks and requests. + Defaults to `parentbased_always_on`. + +Spans are exported in batches. On SIGINT or SIGTERM, graph-node flushes the +buffered spans before exiting; on SIGKILL the last few seconds of spans are +lost. + ## Miscellaneous - `GRAPH_NODE_ID`: sets the node ID, allowing to run multiple Graph Nodes diff --git a/docs/tracing.md b/docs/tracing.md new file mode 100644 index 00000000000..a60d2cdd0a2 --- /dev/null +++ b/docs/tracing.md @@ -0,0 +1,254 @@ +# OpenTelemetry tracing + +graph-node can export traces of block processing, database writes and +GraphQL requests to any OpenTelemetry collector or vendor agent that accepts +OTLP. Tracing is off by default. When it is off, no exporter is installed +and the instrumented code costs one atomic load at the start and one at the +end of each section. + +Traces complement the [Prometheus metrics](./metrics.md), which say how +long things take on average, by showing where the time went in one specific +block or request. + +## Enabling tracing + +Set `GRAPH_OTEL_TRACING=true` and point the exporter at a collector: + +```sh +export GRAPH_OTEL_TRACING=true +export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318 +export OTEL_SERVICE_NAME=graph-node +``` + +graph-node logs one line at startup when tracing is on, `OpenTelemetry +tracing enabled`, naming the endpoint, the service name and the enabled span +groups. + +The exporter speaks OTLP over `http/protobuf`, which the OpenTelemetry +Collector and most vendor agents accept, some after enabling an OTLP HTTP +receiver. gRPC is not supported. Everything about +the exporter is configured with the standard `OTEL_*` variables, listed in +[environment-variables.md](./environment-variables.md#opentelemetry-tracing). +graph-node adds only two variables of its own: `GRAPH_OTEL_TRACING` and +`GRAPH_OTEL_SPANS`. + +### A local collector + +The quickest way to see traces is the OpenTelemetry Collector with its +debug exporter, which prints every span it receives: + +```yaml +# otel-collector.yaml +receivers: + otlp: + protocols: + http: + endpoint: 0.0.0.0:4318 +exporters: + debug: + verbosity: detailed +service: + pipelines: + traces: + receivers: [otlp] + exporters: [debug] +``` + +```sh +docker run --rm -p 4318:4318 \ + -v "$PWD/otel-collector.yaml:/etc/otelcol/config.yaml" \ + otel/opentelemetry-collector:latest --config /etc/otelcol/config.yaml +``` + +To keep the traces, add an `otlp` or `otlphttp` exporter that forwards them +to a backend such as Jaeger or Tempo, both of which accept OTLP. Vendor +exporters usually need the `otel/opentelemetry-collector-contrib` image and +their own configuration. + +### Datadog + +The Datadog Agent accepts OTLP directly. Enable its HTTP receiver with +`DD_OTLP_CONFIG_RECEIVER_PROTOCOLS_HTTP_ENDPOINT=0.0.0.0:4318` and point +`OTEL_EXPORTER_OTLP_ENDPOINT` at the agent. The `service.name` resource +attribute becomes the Datadog service, so set `OTEL_SERVICE_NAME` to +whatever the service should be called there. graph-node does not read +`DD_SERVICE`, `DD_ENV` or `DD_VERSION`; the equivalent resource attributes +go in `OTEL_RESOURCE_ATTRIBUTES`, for example +`deployment.environment.name=prod` for the `env` tag. The agent's own +configuration can still add tags on its side. + +## What is traced + +### Span groups + +Spans belong to groups that are turned on and off together with +`GRAPH_OTEL_SPANS`. The default is `block,handler,store,graphql`. + +| Group | Spans | Volume | +| --- | --- | --- | +| `block` | One trace per processed block, with the runner's sections under it. | One root span and a handful of children per block per subgraph. | +| `handler` | One span per trigger matched and per handler run. | One or two spans per trigger. | +| `host_fn` | One span per host function call made by a handler, such as `store.set` or `ethereum.call`. | Can be thousands per handler. Off by default. | +| `store` | One trace per batch written by the write queue, plus one span per SQL query that serves a GraphQL request. | One root and a few children per batch; one span per entity query. | +| `graphql` | One trace per GraphQL request. | One root span per request. | + +`block`, `store` and `graphql` own the roots of traces. Turning one of them +off removes those traces entirely, including spans of other groups that +would have nested in them: `handler` and `host_fn` spans need `block`, and +`db_query` spans need `graphql`. Turning `store` off removes the +`write_batch` traces and the `db_query` spans, but not the store sections +under `transact_block`, which belong to the block's trace. + +`handler` and `host_fn` only add spans inside a trace. Turning one off skips +its spans, and anything nested in them attaches to the nearest span that +is left. With `handler` off and `host_fn` on, for example, host function +spans sit directly under `process_block`. + +### The span tree + +``` +process_block {subgraph_id, block_number, block_hash, network, shard, stage=process} +├── process_triggers +│ ├── match_and_decode (one per trigger, handler) +│ └── run_handler {handler, trigger, tx_hash?, log_index?} (handler) +│ ├── host_export_ethereum_call {host_function, contract, function, address} (host_fn) +│ ├── host_export_store_set {host_function, entity_type} (host_fn) +│ ├── host_export_store_get, host_export_ipfs_cat, ... {host_function} (host_fn) +│ └── host_export_other {host_function} (host_fn) +├── as_modifications +├── transact_block {entity_count} +└── update_proof_of_indexing + +write_batch {subgraph_id, first_block, last_block, entity_count, dropped_links, shard, stage=writer} +│ links -> the process_block spans whose changes it contains +├── transact_blocks_get_conn +├── apply_entity_modifications +│ ├── apply_entity_modifications_delete +│ ├── check_interface_entity_uniqueness +│ └── apply_entity_modifications_insert +└── transact_blocks_prune + +graphql_query {subgraph, api_version, query_id?, deployment?, error_count?} +│ parent <- the traceparent header of the request, if any +└── db_query {subgraph_id, entity_count, sql?} (one per SQL query) +``` + +The span names are the section names of the existing stopwatch metrics, so +a `deployment_sync_secs{section="run_handler"}` time series and a +`run_handler` span measure the same thing. Store, IPFS and chain host +functions have a section of their own, named `host_export_` followed by the +import name with dots replaced by underscores. Every other host function, +from `bigInt.plus` to `dataSource.create`, shares the `host_export_other` +section. All of them carry the import name in `host_function`, so the +spans of one function can be found or dropped in the collector. + +Attributes worth knowing: + +- `tx_hash` is set on `run_handler` for event and call handlers, and + `log_index` for event handlers, which is where a trace of a block connects + to a transaction. Block handlers have neither. `trigger` is `onchain`, + `offchain` or `subgraph`. +- `shard` is the database shard the deployment lives in and `stage` is + `process` for the runner and `writer` for the write queue. +- `sql` on `db_query` is the query with its bind values, cut at 20 KiB. It is + only set when `GRAPH_LOG_QUERY_TIMING` includes `sql`, the same gate as + the query timing logs, because it can be large and can contain query + variables. +- `query_id` on `graphql_query` is the id that the query timing logs carry, + and the `route` in the comment that prefixes every SQL query, so a span, a + log line and a statement in `pg_stat_activity` can be matched up. It is + missing when the request did not parse. `deployment` is set once the + result names one, and `error_count` only when there were errors. + +A span whose work failed carries an error status with the error message, +cut at 4 KiB. A failed block has the error on `process_block`; a GraphQL +request whose result has errors has them joined on `graphql_query`, with +`error_count` saying how many there were. + +### Writes are their own traces + +The write queue batches the changes of several blocks and writes them after +those blocks were processed, so a write does not belong to any one block. +Each batch is a trace of its own rooted at `write_batch`, and that span +carries a [span link](https://opentelemetry.io/docs/concepts/signals/traces/#span-links) +to the `process_block` span of every block in the batch that has one, so +blocks whose trace was not sampled, or `block` being off, leave gaps. Tracing backends +show links next to the span; follow one to get from a slow write back to +the blocks that caused it. Going the other way, from a block to the batch +that wrote it, depends on whether the backend indexes links. + +A batch keeps at most 32 links. When more blocks than that end up in one +batch, `dropped_links` says how many were left out. + +With the write queue disabled (`GRAPH_STORE_WRITE_QUEUE=0`), writes happen +inside `transact_block` and the store sections nest under the block's +trace instead. + +### GraphQL requests + +A request's trace starts at the HTTP server. If the request carries a W3C +[`traceparent`](https://www.w3.org/TR/trace-context/) header, the +`graphql_query` span joins that trace, so a gateway or client that traces +its own requests sees graph-node's work inside them. Without the header, +the request starts a new trace. + +Every entity query run for the request becomes a `db_query` child. Other +SQL the request needs, such as looking up the deployment, is not traced. +Results served from the query cache run no queries and have no children. + +This is separate from the `GRAPH_GRAPHQL_TRACE_TOKEN` feature, which +returns timing information in the response body of a single request. + +## Sampling + +Sampling is decided once per trace, at its root span, by the sampler +configured with `OTEL_TRACES_SAMPLER` and `OTEL_TRACES_SAMPLER_ARG`. The +default, `parentbased_always_on`, exports everything. On a node that +indexes many subgraphs or serves many queries that is a lot of spans, so +either sample a fraction: + +```sh +export OTEL_TRACES_SAMPLER=parentbased_traceidratio +export OTEL_TRACES_SAMPLER_ARG=0.01 +``` + +or sample in the collector with the +[tail sampling processor](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/processor/tailsamplingprocessor), +which can keep only slow or failed traces. Tail sampling needs all the +spans first, so budget for the export volume even if little of it is kept. + +A `parentbased_*` sampler follows the decision in the `traceparent` header +of a GraphQL request, so a client that samples its own requests decides +whether graph-node's part is kept too. + +An unsampled trace records and exports nothing: once the sampler has said +no at the root, no spans are created under it. What remains is the cost of +building that root and asking the sampler, once per trace. + +## Shutdown + +Spans are buffered and exported in batches by a background thread. When +graph-node receives SIGINT or SIGTERM, it flushes the buffer before +exiting. The flush is best effort: with the collector unreachable it gives +up after the exporter timeout, once per batch left in the buffer. On +SIGKILL, or if the process panics, the spans of the last few seconds are +lost. + +## Overhead + +With tracing off, each instrumented section does one atomic load at its +start and one at its end more than before, and nothing else. Once a +deployment is synced, graph-node stops timing its sections, as it always +has, since the `deployment_sync_secs` counter measures the time to sync. + +With tracing on, the sections of a synced deployment keep running so that +the blocks processed after sync are traced; only the counter stops. Each +section then costs a mutex and a small allocation on top of its span. + +With tracing on, every sampled span allocates its attributes and is handed +to the batch processor. The `block`, `store` and `graphql` groups add a few +spans per block or request and are cheap. `handler` adds one or two spans +per trigger. `host_fn` is the expensive one: a handler that reads and writes +thousands of entities makes thousands of spans, which is why it is off by +default. Turn it on for a specific investigation, together with a low +sampling ratio. diff --git a/graph/Cargo.toml b/graph/Cargo.toml index 13364f1a9fd..c2cf257f496 100644 --- a/graph/Cargo.toml +++ b/graph/Cargo.toml @@ -38,6 +38,10 @@ futures01 = { package = "futures", version = "0.1.31" } lru_time_cache = "0.11" graphql-tools = { workspace = true } lazy_static = { workspace = true } +opentelemetry = { workspace = true } +opentelemetry_sdk = { workspace = true } +opentelemetry-http = { workspace = true } +opentelemetry-otlp = { workspace = true } num-bigint = { version = "=0.2.6", features = ["serde"] } num-integer = { version = "=0.1.46" } num-traits = "=0.2.19" diff --git a/graph/src/blockchain/mod.rs b/graph/src/blockchain/mod.rs index df942d19910..46d5fb06734 100644 --- a/graph/src/blockchain/mod.rs +++ b/graph/src/blockchain/mod.rs @@ -57,6 +57,7 @@ use self::{ client::ChainClient, }; use crate::components::network_provider::ChainName; +use crate::telemetry::AttrValue; #[async_trait] pub trait BlockIngestor: 'static + Send + Sync { @@ -492,6 +493,13 @@ pub trait MappingTriggerTrait { /// If there is an error when processing this trigger, this will called to add relevant context. /// For example an useful return is: `"block # (), transaction ". fn error_context(&self) -> String; + + /// Attributes that identify this trigger on the span of the handler + /// that processes it, like the transaction hash. Only called when + /// tracing is on. + fn trace_attributes(&self) -> Vec<(&'static str, AttrValue)> { + Vec::new() + } } /// A callback that is called after the triggers have been decoded. diff --git a/graph/src/components/metrics/stopwatch.rs b/graph/src/components/metrics/stopwatch.rs index ce6ddfd0ff8..cc1cde7519a 100644 --- a/graph/src/components/metrics/stopwatch.rs +++ b/graph/src/components/metrics/stopwatch.rs @@ -3,6 +3,7 @@ use std::time::Instant; use crate::derive::CheapClone; use crate::prelude::*; +use crate::telemetry::{self, AttrValue, Context, Span, SpanGroup}; /// This is a "section guard", that closes the section on drop. pub struct Section { @@ -34,6 +35,16 @@ impl Drop for Section { /// // do stuff... /// // At the end of the scope `_main_section` is dropped, which is equivalent to calling /// // `_main_section.end()`. +/// ``` +/// +/// When tracing is enabled (see [`crate::telemetry`]), sections also +/// produce spans. A section started with [`StopwatchMetrics::start_root_section`] +/// begins a new trace; a section started while one of its ancestors on the +/// section stack has a span becomes a child of the nearest such ancestor. +/// Sections outside any root, like the base `unknown` section, have no +/// span. Parenting comes from the section stack alone, never from the +/// thread-local context, so it works unchanged across the threads that +/// share a `StopwatchMetrics`. #[derive(Clone, CheapClone)] pub struct StopwatchMetrics { disabled: Arc, @@ -66,12 +77,14 @@ impl StopwatchMetrics { logger, stage, shard, + recording: true, section_stack: Vec::new(), + spans: Vec::new(), timer: Instant::now(), }; // Start a base section so that all time is accounted for. - inner.start_section("unknown".to_owned()); + inner.start_section("unknown".to_owned(), SpanStart::Child, Vec::new); StopwatchMetrics { disabled: Arc::new(AtomicBool::new(false)), @@ -80,9 +93,53 @@ impl StopwatchMetrics { } pub fn start_section(&self, id: &str) -> Section { + self.start_section_with(id, Vec::new) + } + + /// Like [`Self::start_section`], with the attributes `attrs` returns + /// set on the section's span. `attrs` is only called if the section + /// gets a span that is recorded, so building them costs nothing + /// otherwise. It runs while the stopwatch is locked and must not call + /// back into it. + pub fn start_section_with(&self, id: &str, attrs: impl FnOnce() -> I) -> Section + where + I: IntoIterator, + { + self.start(id, SpanStart::Child, attrs) + } + + /// Start a section whose span, if `group` is enabled, is the root of a + /// new trace rather than a child of the enclosing section's span. + /// `links` point the new trace at other traces, for work that was + /// caused by several of them. `attrs` is only called if the root is + /// recorded, and runs while the stopwatch is locked, so it must not + /// call back into it. The root also carries the stopwatch's `shard` + /// and `stage`. + pub fn start_root_section( + &self, + id: &str, + group: SpanGroup, + attrs: impl FnOnce() -> I, + links: &[Context], + ) -> Section + where + I: IntoIterator, + { + self.start(id, SpanStart::Root { group, links }, attrs) + } + + fn start(&self, id: &str, span: SpanStart<'_>, attrs: impl FnOnce() -> I) -> Section + where + I: IntoIterator, + { let id = id.to_owned(); if !self.disabled.load(Ordering::SeqCst) { - self.inner.lock().unwrap().start_section(id.clone()) + let mut inner = self.inner.lock().unwrap(); + // `stop_recording` sets `disabled` under the lock; a section + // that lost the race to it must not start. + if !self.disabled.load(Ordering::SeqCst) { + inner.start_section(id.clone(), span, attrs) + } } // If disabled, this will do nothing on drop. @@ -92,9 +149,67 @@ impl StopwatchMetrics { } } - /// Turns `start_section` and `end_section` into no-ops, no more metrics will be updated. - pub fn disable(&self) { - self.disabled.store(true, Ordering::SeqCst) + /// Set an attribute on the span of the current section, for details + /// that are only known after the section started. Does nothing if the + /// current section has no span. + pub fn set_attr(&self, key: &'static str, value: impl Into) { + self.set_attrs_with(|| [(key, value.into())]) + } + + /// Set the attributes `attrs` returns on the span of the current + /// section. `attrs` is only called if the current section has a span, + /// so building them costs nothing otherwise. It runs while the + /// stopwatch is locked and must not call back into it. + pub fn set_attrs_with(&self, attrs: impl FnOnce() -> I) + where + I: IntoIterator, + { + if telemetry::enabled() && !self.disabled.load(Ordering::SeqCst) { + self.inner.lock().unwrap().set_attrs_with(attrs) + } + } + + /// Mark the span of the current section as failed. Does nothing if the + /// current section has no span. + pub fn set_error(&self, message: &str) { + self.set_error_with(|| message.to_owned()) + } + + /// Like [`Self::set_error`], but the message is only built if the + /// current section has a span. Use it when formatting the message is + /// not free. + pub fn set_error_with(&self, message: impl FnOnce() -> String) { + if telemetry::enabled() && !self.disabled.load(Ordering::SeqCst) { + self.inner.lock().unwrap().set_error_with(message) + } + } + + /// The context of the span that a section started now would be + /// parented to, if any. Used to link work that happens elsewhere, like + /// a queued write, back to the current trace. + pub fn current_context(&self) -> Option { + if telemetry::enabled() && !self.disabled.load(Ordering::SeqCst) { + self.inner.lock().unwrap().current_context() + } else { + None + } + } + + /// Stop recording time into the `deployment_sync_secs` counter, which + /// measures the time to sync a deployment; called once the deployment + /// is synced. With tracing off, the stopwatch does nothing from here + /// on. With tracing on, sections keep producing spans for the blocks + /// processed after sync, and only the counter stops. + pub fn stop_recording(&self) { + let mut inner = self.inner.lock().unwrap(); + if telemetry::enabled() { + inner.recording = false; + } else { + // Under the lock, so that a `start_section` that already + // passed its check of `disabled` cannot push a section that + // `end_section` would then ignore. + self.disabled.store(true, Ordering::SeqCst); + } } fn end_section(&self, id: String) { @@ -108,6 +223,42 @@ impl StopwatchMetrics { } } +/// How the span of a new section, if any, relates to the trace. +enum SpanStart<'a> { + /// A child of the nearest enclosing section that has a span. + Child, + /// The root of a new trace, if `group` is enabled. + Root { + group: SpanGroup, + links: &'a [Context], + }, +} + +/// What a section on the stack contributes to the trace. +enum SectionSpan { + /// No span; sections nested in it attach to the nearest span further + /// down the stack. + None, + /// A root that got no span because its group is off or its trace was + /// not sampled. Sections nested in it get no span either, rather than + /// escaping into an enclosing trace. + Boundary, + Span(Span), +} + +/// The group that decides whether a non-root section gets a span, for +/// the sections that have one of their own. Any other section follows its +/// enclosing root. +fn section_group(id: &str) -> Option { + if id.starts_with("host_export_") { + Some(SpanGroup::HostFn) + } else if id == "run_handler" || id == "match_and_decode" { + Some(SpanGroup::Handler) + } else { + None + } +} + /// We want to account for all subgraph indexing time, based on "wall clock" time. To do this we /// break down indexing into _sequential_ sections, and register the total time spent in each. So /// that there is no double counting, time spent in child sections doesn't count for the parent. @@ -117,9 +268,16 @@ struct StopwatchInner { // Counter for the total time the subgraph spent syncing in various sections. counter: CounterVec, + // Whether time is still recorded into `counter`. False once the + // deployment is synced. + recording: bool, + // The top section (last item) is the one that's currently executing. section_stack: Vec, + // The span of each section in `section_stack`, in lockstep with it. + spans: Vec, + // The timer is reset whenever a section starts or ends. timer: Instant, @@ -132,6 +290,9 @@ struct StopwatchInner { impl StopwatchInner { fn record_and_reset(&mut self) { + if !self.recording { + return; + } if let Some(section) = self.section_stack.last() { // Register the current timer. let elapsed = self.timer.elapsed().as_secs_f64(); @@ -149,11 +310,89 @@ impl StopwatchInner { self.timer = Instant::now(); } - fn start_section(&mut self, id: String) { + fn start_section(&mut self, id: String, span: SpanStart<'_>, attrs: impl FnOnce() -> I) + where + I: IntoIterator, + { #[cfg(debug_assertions)] self.record_section_relation(&id); self.record_and_reset(); + let span = self.start_span(&id, span, attrs); self.section_stack.push(id); + self.spans.push(span); + } + + fn start_span(&self, id: &str, span: SpanStart<'_>, attrs: impl FnOnce() -> I) -> SectionSpan + where + I: IntoIterator, + { + if !telemetry::enabled() { + return SectionSpan::None; + } + + let mut span = match span { + SpanStart::Root { group, links } => { + if !telemetry::group_enabled(group) { + return SectionSpan::Boundary; + } + let mut span = Span::root_with_links(id.to_owned(), links); + // An unsampled root is a no-op span; its subtree is skipped. + if !span.is_recording() { + return SectionSpan::Boundary; + } + span.set_attrs([ + ("shard", self.shard.clone().into()), + ("stage", self.stage.clone().into()), + ]); + span + } + SpanStart::Child => { + let Some(parent) = self.current_context() else { + return SectionSpan::None; + }; + if let Some(group) = section_group(id) + && !telemetry::group_enabled(group) + { + return SectionSpan::None; + } + let span = Span::child(id.to_owned(), &parent); + if !span.is_recording() { + return SectionSpan::None; + } + span + } + }; + + span.set_attrs(attrs()); + SectionSpan::Span(span) + } + + /// The context of the nearest section on the stack that has a span, + /// stopping at a root that has none. + fn current_context(&self) -> Option { + for span in self.spans.iter().rev() { + match span { + SectionSpan::Span(span) => return span.context(), + SectionSpan::Boundary => return None, + SectionSpan::None => continue, + } + } + None + } + + fn set_attrs_with(&mut self, attrs: impl FnOnce() -> I) + where + I: IntoIterator, + { + if let Some(SectionSpan::Span(span)) = self.spans.last_mut() { + span.set_attrs(attrs()); + } + } + + fn set_error_with(&mut self, message: impl FnOnce() -> String) { + if let Some(SectionSpan::Span(span)) = self.spans.last_mut() { + span.set_error(&message()); + } } fn end_section(&mut self, id: String) { @@ -162,6 +401,9 @@ impl StopwatchInner { Some(current_section) if current_section == &id => { self.record_and_reset(); self.section_stack.pop(); + if let Some(SectionSpan::Span(span)) = self.spans.pop() { + span.end(); + } } Some(current_section) => error!(self.logger, "`end_section` with mismatched section"; "current" => current_section, @@ -236,3 +478,349 @@ impl StopwatchInner { } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::telemetry::SpanGroups; + use crate::telemetry::testing::{ExportedSpan, in_subprocess, install, install_with}; + + fn stopwatch() -> StopwatchMetrics { + StopwatchMetrics::new( + crate::log::discard(), + DeploymentHash::new("QmStopwatchSpans").unwrap(), + "test", + Arc::new(MetricsRegistry::mock()), + "shard".to_string(), + ) + } + + fn test_attr(name: &'static str) -> (&'static str, AttrValue) { + ("test", AttrValue::from(name)) + } + + fn names(spans: &[ExportedSpan]) -> Vec<&str> { + spans.iter().map(|span| span.name.as_str()).collect() + } + + #[test] + fn sections_become_nested_spans() { + let handle = install(); + let stopwatch = stopwatch(); + + // Sections outside a root have no span, and no error message is + // built for them. + let outside = stopwatch.start_section("scan_blocks"); + assert!(stopwatch.current_context().is_none()); + stopwatch.set_error_with(|| panic!("message built for a section without a span")); + stopwatch.set_attrs_with(|| -> [(&'static str, AttrValue); 0] { + panic!("attributes built for a section without a span") + }); + let unspanned = + stopwatch.start_section_with("no_span", || -> [(&'static str, AttrValue); 0] { + panic!("attributes built for a section without a span") + }); + unspanned.end(); + outside.end(); + + let root = stopwatch.start_root_section( + "process_block", + SpanGroup::Block, + || { + [ + test_attr("sections_become_nested_spans"), + ("block_number", 7.into()), + ] + }, + &[], + ); + let root_ctx = stopwatch.current_context().expect("root has a context"); + + let triggers = stopwatch.start_section("process_triggers"); + let handler = + stopwatch.start_section_with("run_handler", || [("handler", "handleX".into())]); + stopwatch.set_attr("tx_hash", "0xabc"); + stopwatch.set_error("handler failed"); + let host_fn = stopwatch.start_section("host_export_store_set"); + host_fn.end(); + handler.end(); + triggers.end(); + let transact = stopwatch.start_section("transact_block"); + transact.end(); + root.end(); + assert!(stopwatch.current_context().is_none()); + + let roots = handle.spans_with_attr("test", "sections_become_nested_spans"); + assert_eq!(roots.len(), 1); + let root = &roots[0]; + assert_eq!(root.name, "process_block"); + assert_eq!(root.parent_span_id, None); + assert_eq!(root.trace_id, root_ctx.trace_id()); + assert_eq!(root.attr("block_number"), Some("7")); + assert_eq!(root.attr("shard"), Some("shard")); + assert_eq!(root.attr("stage"), Some("test")); + + // Five spans in the trace, none for `scan_blocks` or `unknown`. + let trace = handle.trace(&root.trace_id); + assert_eq!(trace.len(), 5); + + let children = handle.children_of(root); + assert_eq!(names(&children), ["process_triggers", "transact_block"]); + let handlers = handle.children_of(&children[0]); + assert_eq!(names(&handlers), ["run_handler"]); + assert_eq!(handlers[0].attr("handler"), Some("handleX")); + assert_eq!(handlers[0].attr("tx_hash"), Some("0xabc")); + assert!(handlers[0].is_error); + assert!(!root.is_error); + let host_fns = handle.children_of(&handlers[0]); + assert_eq!(names(&host_fns), ["host_export_store_set"]); + } + + #[test] + fn root_sections_can_link_to_other_traces() { + let handle = install(); + let stopwatch = stopwatch(); + + let mut linked_to = Vec::new(); + for _ in 0..2 { + let block = stopwatch.start_root_section( + "process_block", + SpanGroup::Block, + || [test_attr("root_sections_can_link_to_other_traces")], + &[], + ); + linked_to.push(stopwatch.current_context().unwrap()); + block.end(); + } + + let write = stopwatch.start_root_section( + "write_batch", + SpanGroup::Store, + || [test_attr("root_sections_can_link_to_other_traces")], + &linked_to, + ); + let write_ctx = stopwatch.current_context().unwrap(); + write.end(); + + let spans = handle.spans_with_attr("test", "root_sections_can_link_to_other_traces"); + let write = spans.iter().find(|s| s.name == "write_batch").unwrap(); + assert_eq!(write.trace_id, write_ctx.trace_id()); + assert!(linked_to.iter().all(|ctx| ctx.trace_id() != write.trace_id)); + let mut links: Vec<_> = write + .links + .iter() + .map(|(trace_id, _)| trace_id.as_str()) + .collect(); + links.sort(); + let mut expected: Vec<_> = linked_to.iter().map(|ctx| ctx.trace_id()).collect(); + expected.sort(); + assert_eq!(links, expected); + } + + #[test] + fn mismatched_end_leaves_spans_consistent() { + install(); + let stopwatch = stopwatch(); + + let root = stopwatch.start_root_section("process_block", SpanGroup::Block, || [], &[]); + let child = stopwatch.start_section("a"); + stopwatch + .inner + .lock() + .unwrap() + .end_section("wrong".to_owned()); + { + let inner = stopwatch.inner.lock().unwrap(); + assert_eq!(inner.section_stack, ["unknown", "process_block", "a"]); + assert_eq!(inner.spans.len(), 3); + assert!(matches!(inner.spans[0], SectionSpan::None)); + assert!(matches!(inner.spans[1], SectionSpan::Span(_))); + assert!(matches!(inner.spans[2], SectionSpan::Span(_))); + } + child.end(); + root.end(); + { + let inner = stopwatch.inner.lock().unwrap(); + assert_eq!(inner.section_stack, ["unknown"]); + assert_eq!(inner.spans.len(), 1); + } + } + + #[test] + fn stop_recording_keeps_spans() { + let handle = install(); + let stopwatch = stopwatch(); + let counter = |section: &str| { + stopwatch + .inner + .lock() + .unwrap() + .counter + .get_metric_with_label_values(&[section, "test", "shard"]) + .unwrap() + .get() + }; + + let root = stopwatch.start_root_section( + "process_block", + SpanGroup::Block, + || [test_attr("stop_recording_keeps_spans")], + &[], + ); + let before = stopwatch.start_section("before"); + std::thread::sleep(Duration::from_millis(1)); + before.end(); + assert!(counter("before") > 0.0); + + stopwatch.stop_recording(); + let block_secs = counter("process_block"); + + // The counter no longer moves; sections and their spans go on. + let after = + stopwatch.start_section_with("after", || [test_attr("stop_recording_keeps_spans")]); + std::thread::sleep(Duration::from_millis(1)); + assert!(stopwatch.current_context().is_some()); + after.end(); + root.end(); + assert_eq!(counter("after"), 0.0); + assert_eq!(counter("process_block"), block_secs); + + let spans = handle.spans_with_attr("test", "stop_recording_keeps_spans"); + assert_eq!(names(&spans), ["after", "process_block"]); + assert_eq!( + spans[0].parent_span_id.as_deref(), + Some(spans[1].span_id.as_str()) + ); + assert_eq!(stopwatch.inner.lock().unwrap().section_stack, ["unknown"]); + } + + #[test] + fn stop_recording_without_tracing_disables_sections() { + in_subprocess( + "components::metrics::stopwatch::tests::stop_recording_without_tracing_disables_sections", + &[], + || { + assert!(!telemetry::enabled()); + let stopwatch = stopwatch(); + stopwatch.stop_recording(); + let section = stopwatch.start_section("a"); + assert_eq!(stopwatch.inner.lock().unwrap().section_stack, ["unknown"]); + section.end(); + assert_eq!(stopwatch.inner.lock().unwrap().section_stack, ["unknown"]); + }, + ); + } + + #[test] + fn groups_gate_their_sections() { + in_subprocess( + "components::metrics::stopwatch::tests::groups_gate_their_sections", + &[], + || { + // Everything but host functions. + let handle = install_with(SpanGroups::default()); + let stopwatch = stopwatch(); + + let root = stopwatch.start_root_section( + "process_block", + SpanGroup::Block, + || [test_attr("groups_gate_their_sections")], + &[], + ); + let handler = stopwatch.start_section("run_handler"); + let host_fn = stopwatch.start_section("host_export_store_set"); + // A section under a gated-off section attaches to the + // nearest ancestor with a span. + let nested = stopwatch.start_section("under_host_fn"); + nested.end(); + host_fn.end(); + handler.end(); + root.end(); + + let roots = handle.spans_with_attr("test", "groups_gate_their_sections"); + assert_eq!(roots.len(), 1); + let trace = handle.trace(&roots[0].trace_id); + assert_eq!( + names(&trace), + ["under_host_fn", "run_handler", "process_block"] + ); + let handler = trace.iter().find(|s| s.name == "run_handler").unwrap(); + assert_eq!(names(&handle.children_of(handler)), ["under_host_fn"]); + + // A root in a disabled group produces no trace at all, and + // nested inside a traced root it is a boundary: its + // sections do not escape into the enclosing trace. + let outer = stopwatch.start_root_section( + "process_block", + SpanGroup::Block, + || [test_attr("groups_gate_their_sections_boundary")], + &[], + ); + let outer_ctx = stopwatch.current_context().unwrap(); + let root = stopwatch.start_root_section( + "host_export_root", + SpanGroup::HostFn, + || [test_attr("groups_gate_their_sections_boundary")], + &[], + ); + assert!(stopwatch.current_context().is_none()); + let child = stopwatch.start_section_with("child", || { + [test_attr("groups_gate_their_sections_boundary")] + }); + assert!(stopwatch.current_context().is_none()); + child.end(); + root.end(); + // Back in the outer trace. + assert_eq!( + stopwatch.current_context().unwrap().trace_id(), + outer_ctx.trace_id() + ); + let after = stopwatch.start_section("after"); + after.end(); + outer.end(); + let spans = handle.spans_with_attr("test", "groups_gate_their_sections_boundary"); + assert_eq!(names(&spans), ["process_block"]); + assert_eq!( + names(&handle.trace(&outer_ctx.trace_id())), + ["after", "process_block"] + ); + }, + ); + } + + #[test] + fn tracing_off_keeps_sections_working() { + in_subprocess( + "components::metrics::stopwatch::tests::tracing_off_keeps_sections_working", + &[], + || { + assert!(!telemetry::enabled()); + let stopwatch = stopwatch(); + let root = stopwatch.start_root_section( + "process_block", + SpanGroup::Block, + || [("block_number", 1.into())], + &[], + ); + assert!(stopwatch.current_context().is_none()); + let child = stopwatch.start_section_with("a", || [("k", "v".into())]); + stopwatch.set_attr("k2", "v2"); + { + let inner = stopwatch.inner.lock().unwrap(); + assert_eq!(inner.section_stack, ["unknown", "process_block", "a"]); + assert!( + inner + .spans + .iter() + .all(|span| matches!(span, SectionSpan::None)) + ); + } + child.end(); + root.end(); + let inner = stopwatch.inner.lock().unwrap(); + assert_eq!(inner.section_stack, ["unknown"]); + assert_eq!(inner.spans.len(), 1); + }, + ); + } +} diff --git a/graph/src/components/store/write.rs b/graph/src/components/store/write.rs index 1a29714e51e..d9b4d7c6688 100644 --- a/graph/src/components/store/write.rs +++ b/graph/src/components/store/write.rs @@ -6,6 +6,8 @@ use std::{ use slog::Logger; +use crate::telemetry::Context; + use crate::{ blockchain::{BlockPtr, BlockTime, block_stream::FirehoseCursor}, cheap_clone::CheapClone, @@ -783,9 +785,21 @@ pub struct Batch { /// the writer holds a lock, conflicts with appending to the batch and /// causes batches to be finished prematurely. indirect_weight: usize, + /// The traces of the blocks whose changes this batch contains, so that + /// the span for writing the batch can link back to them. Empty when + /// tracing is off. Capped at [`Batch::MAX_TRACE_CONTEXTS`]; the rest + /// are counted in `dropped_trace_contexts`. + pub trace_contexts: Vec, + /// How many block traces did not fit into `trace_contexts`. + pub dropped_trace_contexts: usize, } impl Batch { + /// How many block traces a batch links to at most. A batch rarely + /// spans more blocks than this, and a span with hundreds of links is + /// not useful. + pub const MAX_TRACE_CONTEXTS: usize = 32; + pub fn new( block_ptr: BlockPtr, block_time: BlockTime, @@ -831,6 +845,8 @@ impl Batch { error: None, is_non_fatal_errors_active, indirect_weight: 0, + trace_contexts: Vec::new(), + dropped_trace_contexts: 0, }; batch.weigh(); Ok(batch) @@ -853,9 +869,23 @@ impl Batch { self.deterministic_errors .append(&mut batch.deterministic_errors); self.offchain_to_remove.append(batch.offchain_to_remove); + self.dropped_trace_contexts += batch.dropped_trace_contexts; + self.push_trace_contexts(batch.trace_contexts); Ok(()) } + /// Remember the traces of the blocks in this batch, up to + /// [`Self::MAX_TRACE_CONTEXTS`], and count the ones that do not fit. + pub fn push_trace_contexts(&mut self, contexts: impl IntoIterator) { + for ctx in contexts { + if self.trace_contexts.len() < Self::MAX_TRACE_CONTEXTS { + self.trace_contexts.push(ctx); + } else { + self.dropped_trace_contexts += 1; + } + } + } + /// Append `batch` to `self` so that writing `self` afterwards has the /// same effect as writing `self` first and then `batch` in separate /// transactions. @@ -1384,3 +1414,50 @@ mod test { assert_eq!(group.row_count(), 2); } } + +#[cfg(test)] +mod trace_context_tests { + use super::Batch; + use crate::blockchain::block_stream::FirehoseCursor; + use crate::blockchain::{BlockPtr, BlockTime}; + use crate::prelude::alloy::primitives::B256; + use crate::telemetry::Span; + use crate::telemetry::testing::install; + + fn batch(block: u8) -> Batch { + let block_ptr = BlockPtr::from((B256::from([block; 32]), block as i32)); + let block_time = BlockTime::for_test(&block_ptr); + Batch::new( + block_ptr, + block_time, + FirehoseCursor::None, + Vec::new(), + Vec::new(), + Vec::new(), + Vec::new(), + false, + crate::log::discard(), + ) + .unwrap() + } + + #[test] + fn trace_contexts_are_capped_and_counted() { + install(); + let mut first = batch(1); + let contexts: Vec<_> = (0..Batch::MAX_TRACE_CONTEXTS + 5) + .map(|_| Span::root("block").context().unwrap()) + .collect(); + first.push_trace_contexts(contexts); + assert_eq!(first.trace_contexts.len(), Batch::MAX_TRACE_CONTEXTS); + assert_eq!(first.dropped_trace_contexts, 5); + + // Appending carries the count over and keeps capping. + let mut second = batch(2); + second.push_trace_contexts([Span::root("block").context().unwrap()]); + second.dropped_trace_contexts = 2; + first.append(second).unwrap(); + assert_eq!(first.trace_contexts.len(), Batch::MAX_TRACE_CONTEXTS); + assert_eq!(first.dropped_trace_contexts, 8); + } +} diff --git a/graph/src/data/query/query.rs b/graph/src/data/query/query.rs index ab069852374..c74e71827f2 100644 --- a/graph/src/data/query/query.rs +++ b/graph/src/data/query/query.rs @@ -9,6 +9,7 @@ use std::sync::Arc; use crate::{ data::graphql::shape_hash::shape_hash, prelude::{ApiVersion, DeploymentHash, ENV_VARS, SubgraphName, q, r}, + telemetry, }; fn deserialize_number<'de, D>(deserializer: D) -> Result @@ -135,6 +136,10 @@ pub struct Query { pub query_text: Arc, pub variables_text: Arc, pub trace: bool, + /// The tracing span of the request this query belongs to, if the + /// request is traced. Spans for work done on its behalf are started + /// under it. + pub trace_context: Option, } impl Query { @@ -162,8 +167,25 @@ impl Query { query_text: Arc::new(query_text), variables_text: Arc::new(variables_text), trace, + trace_context: None, } } + + pub fn with_trace_context(mut self, trace_context: Option) -> Self { + self.trace_context = trace_context; + self + } + + /// An identifier for this query that is stable across identical + /// requests. It is what query timing logs are tagged with; the texts it + /// hashes are only meaningful when the query is traced or query timing + /// is logged, see [`Query::new`]. + pub fn query_id(&self) -> String { + let mut hasher = DefaultHasher::new(); + self.query_text.hash(&mut hasher); + self.variables_text.hash(&mut hasher); + format!("{:x}-{:x}", self.shape_hash, hasher.finish()) + } } #[derive(Copy, Clone, Debug, Deserialize, Serialize)] diff --git a/graph/src/data_source/mod.rs b/graph/src/data_source/mod.rs index 016903ab55f..1d3af743bb5 100644 --- a/graph/src/data_source/mod.rs +++ b/graph/src/data_source/mod.rs @@ -11,6 +11,7 @@ pub use causality_region::CausalityRegion; #[cfg(test)] mod tests; +use crate::telemetry::AttrValue; use crate::{ blockchain::{ Block, BlockPtr, BlockTime, Blockchain, DataSource as _, DataSourceTemplate as _, @@ -660,6 +661,17 @@ impl MappingTrigger { Self::Subgraph(_) => None, // TODO(krishna) } } + + /// Attributes for the span of the handler that processes this trigger. + pub fn trace_attributes(&self) -> Vec<(&'static str, AttrValue)> { + let (kind, mut attrs) = match self { + Self::Onchain(trigger) => ("onchain", trigger.trace_attributes()), + Self::Offchain(_) => ("offchain", Vec::new()), + Self::Subgraph(_) => ("subgraph", Vec::new()), + }; + attrs.push(("trigger", kind.into())); + attrs + } } impl Clone for DataSource { diff --git a/graph/src/env/mod.rs b/graph/src/env/mod.rs index 54c585d5ab3..2af022d578e 100644 --- a/graph/src/env/mod.rs +++ b/graph/src/env/mod.rs @@ -17,6 +17,7 @@ use self::store::*; use crate::{ components::{store::BlockNumber, subgraph::SubgraphVersionSwitchingMode}, runtime::gas::CONST_MAX_GAS_PER_HANDLER, + telemetry::SpanGroups, }; pub use self::amp::AmpEnv; @@ -235,6 +236,13 @@ pub struct EnvVars { /// which must be an absolute path. This only has an effect in debug /// builds. Set with `GRAPH_SECTION_MAP`. Defaults to `None`. pub section_map: Option, + /// Export OpenTelemetry spans over OTLP. Set by the flag + /// `GRAPH_OTEL_TRACING`. Off by default. + pub otel_tracing: bool, + /// Which groups of spans to emit when tracing is on. Set by + /// `GRAPH_OTEL_SPANS` as a comma-separated list of group names; see + /// [`SpanGroups`]. Defaults to every group except `host_fn`. + pub otel_spans: SpanGroups, /// Set the maximum grpc decode size(in MB) for firehose BlockIngestor connections. /// Defaults to 25MB pub firehose_grpc_max_decode_size_mb: usize, @@ -372,6 +380,8 @@ impl EnvVars { .unwrap_or(2 * inner.reorg_threshold), dips_metrics_object_store_url: inner.dips_metrics_object_store_url, section_map: inner.section_map, + otel_tracing: inner.otel_tracing.0, + otel_spans: inner.otel_spans, firehose_grpc_max_decode_size_mb: inner.firehose_grpc_max_decode_size_mb, genesis_validation_enabled: inner.genesis_validation_enabled.0, disable_deployment_hash_validation: inner.disable_deployment_hash_validation.0, @@ -593,6 +603,10 @@ struct Inner { dips_metrics_object_store_url: Option, #[envconfig(from = "GRAPH_SECTION_MAP")] section_map: Option, + #[envconfig(from = "GRAPH_OTEL_TRACING", default = "false")] + otel_tracing: EnvVarBoolean, + #[envconfig(from = "GRAPH_OTEL_SPANS", default = "block,handler,store,graphql")] + otel_spans: SpanGroups, #[envconfig(from = "GRAPH_NODE_FIREHOSE_MAX_DECODE_SIZE", default = "25")] firehose_grpc_max_decode_size_mb: usize, #[envconfig(from = "GRAPH_NODE_GENESIS_VALIDATION_ENABLED", default = "false")] diff --git a/graph/src/lib.rs b/graph/src/lib.rs index 00f01f53551..96be47c6ab8 100644 --- a/graph/src/lib.rs +++ b/graph/src/lib.rs @@ -35,6 +35,9 @@ pub mod ipfs; pub mod tls; +/// Opt-in OpenTelemetry tracing. +pub mod telemetry; + pub mod abi; pub mod amp; diff --git a/graph/src/telemetry/mod.rs b/graph/src/telemetry/mod.rs new file mode 100644 index 00000000000..50806fc2a9b --- /dev/null +++ b/graph/src/telemetry/mod.rs @@ -0,0 +1,802 @@ +//! Opt-in OpenTelemetry tracing. +//! +//! This module is the only place in graph-node that uses the `opentelemetry` +//! crates. Everything else goes through the small API defined here, so that +//! span creation is a cheap no-op unless tracing was enabled at startup and +//! so that a change in the OpenTelemetry API only has to be dealt with in +//! one place. +//! +//! Spans are organised in [`SpanGroup`]s. The expensive groups, like one +//! span per WASM host function call, can be left off while blocks and +//! handlers are still traced. Which groups are on is decided once at +//! startup; see [`SpanGroups`]. +//! +//! The node installs a tracer provider at startup with [`init`], and tests +//! do so with [`testing::install`]. Until one of them runs, [`enabled`] is +//! `false` and every [`Span`] is a no-op. + +mod span; +pub mod testing; + +use std::fmt; +use std::str::FromStr; +use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; + +use anyhow::Context as _; +use opentelemetry::trace::{SpanContext, TraceContextExt}; +use opentelemetry::{Key, KeyValue, global}; +use opentelemetry_http::{HeaderExtractor, HeaderInjector}; +use opentelemetry_sdk::Resource; +use opentelemetry_sdk::propagation::TraceContextPropagator; +use opentelemetry_sdk::trace::SdkTracerProvider; +use slog::{Logger, info}; + +use crate::env::EnvVars; + +pub use span::{AttrValue, MAX_ERROR_MESSAGE_LEN, Span}; + +/// Whether a tracer provider has been installed. Checked before any span +/// work so that the disabled path costs one atomic load. +static ENABLED: AtomicBool = AtomicBool::new(false); + +/// The bitset of enabled [`SpanGroup`]s. Only meaningful when `ENABLED` +/// is set; it is written before `ENABLED` and read after it. +static GROUPS: AtomicU8 = AtomicU8::new(0); + +/// `true` once a tracer provider has been installed. +#[inline] +pub fn enabled() -> bool { + ENABLED.load(Ordering::Acquire) +} + +/// `true` if tracing is enabled and `group` was selected. +#[inline] +pub fn group_enabled(group: SpanGroup) -> bool { + enabled() && GROUPS.load(Ordering::Relaxed) & group.bit() != 0 +} + +/// Flip the global switch. Called by whoever installs a tracer provider. +fn set_enabled(groups: SpanGroups) { + GROUPS.store(groups.0, Ordering::Relaxed); + ENABLED.store(true, Ordering::Release); +} + +/// What [`init`] needs to know, taken from the environment. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct Config { + /// `GRAPH_OTEL_TRACING` + pub enabled: bool, + /// `GRAPH_OTEL_SPANS` + pub groups: SpanGroups, +} + +impl From<&EnvVars> for Config { + fn from(env: &EnvVars) -> Self { + Config { + enabled: env.otel_tracing, + groups: env.otel_spans, + } + } +} + +/// Keeps the tracer provider alive. Dropping it flushes buffered spans and +/// shuts the exporter down, so it must live until the process is about to +/// exit. +#[must_use = "dropping the guard shuts tracing down"] +pub struct Guard { + provider: Option, +} + +impl Guard { + /// `true` if this guard owns a live tracer provider. + pub fn is_enabled(&self) -> bool { + self.provider.is_some() + } + + /// Export the buffered spans and shut the exporter down. This blocks + /// until the export finished or timed out, so call it from a blocking + /// context. Dropping the guard does the same but discards the result. + pub fn shutdown(mut self) -> Result<(), anyhow::Error> { + match self.provider.take() { + Some(provider) => provider + .shutdown() + .context("failed to flush spans during shutdown"), + None => Ok(()), + } + } +} + +impl Drop for Guard { + fn drop(&mut self) { + if let Some(provider) = self.provider.take() { + // Shutdown drains the batch processor. An error here means + // spans were lost on the way out, and there is nobody left to + // tell. + let _ = provider.shutdown(); + } + } +} + +/// Install the OTLP exporter as the process-wide tracer provider if +/// `config.enabled` is set, and return a guard that keeps it alive. With +/// tracing off, nothing is installed and the guard does nothing. +/// +/// The exporter speaks OTLP over `http/protobuf` and is configured through +/// the standard `OTEL_EXPORTER_OTLP_*` environment variables, with +/// `http://localhost:4318` as the default endpoint. The sampler is +/// configured through `OTEL_TRACES_SAMPLER` and `OTEL_TRACES_SAMPLER_ARG`. +/// The service name comes from `OTEL_SERVICE_NAME` and defaults to +/// `graph-node`. +/// +/// Must be called at most once per process. +pub fn init(config: &Config, logger: &Logger) -> Result { + if !config.enabled { + return Ok(Guard { provider: None }); + } + if enabled() { + anyhow::bail!("OpenTelemetry tracing was already initialized"); + } + + let exporter = opentelemetry_otlp::SpanExporter::builder() + .with_http() + .build() + .context("failed to build the OTLP span exporter")?; + + let var = |name: &str| std::env::var(name).ok().filter(|v| !v.is_empty()); + let service_name = service_name(&Resource::builder().build(), var("OTEL_SERVICE_NAME")); + let resource = Resource::builder() + .with_service_name(service_name.clone()) + .with_attribute(KeyValue::new("service.version", env!("CARGO_PKG_VERSION"))) + .build(); + + let provider = SdkTracerProvider::builder() + .with_batch_exporter(exporter) + .with_resource(resource) + .build(); + global::set_tracer_provider(provider.clone()); + global::set_text_map_propagator(TraceContextPropagator::new()); + set_enabled(config.groups); + + info!(logger, "OpenTelemetry tracing enabled"; + "endpoint" => otlp_traces_endpoint( + var("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"), + var("OTEL_EXPORTER_OTLP_ENDPOINT")), + "service_name" => service_name, + "spans" => config.groups.to_string()); + + Ok(Guard { + provider: Some(provider), + }) +} + +/// The `service.name` to report, in the order the OpenTelemetry +/// specification gives: `OTEL_SERVICE_NAME`, then whatever the SDK +/// detected from `OTEL_RESOURCE_ATTRIBUTES`, then `graph-node`. The SDK's +/// own `unknown_service` fallback is never used, and its detectors do not +/// give `OTEL_SERVICE_NAME` priority on their own. +fn service_name(detected: &Resource, otel_service_name: Option) -> String { + otel_service_name + .or_else(|| { + detected + .get(&Key::from_static_str("service.name")) + .map(|value| value.to_string()) + .filter(|name| { + !name.is_empty() + && name != "unknown_service" + && !name.starts_with("unknown_service:") + }) + }) + .unwrap_or_else(|| "graph-node".to_string()) +} + +/// The endpoint the exporter will send traces to, for the startup log +/// line. Mirrors the lookup the OTLP crate does internally, given the +/// values of `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` and +/// `OTEL_EXPORTER_OTLP_ENDPOINT`. +fn otlp_traces_endpoint(traces_endpoint: Option, endpoint: Option) -> String { + traces_endpoint.unwrap_or_else(|| { + let base = endpoint.unwrap_or_else(|| "http://localhost:4318".to_string()); + format!("{}/v1/traces", base.trim_end_matches('/')) + }) +} + +/// A family of spans that can be turned on and off together. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum SpanGroup { + /// One root span per processed block and the runner sections under it. + Block, + /// `run_handler` and `match_and_decode`, one per trigger. + Handler, + /// One span per WASM host function call. Can be thousands per handler. + HostFn, + /// Write batches in the store and the SQL queries that serve GraphQL. + Store, + /// One root span per GraphQL request. + GraphQl, +} + +impl SpanGroup { + const ALL: [SpanGroup; 5] = [ + SpanGroup::Block, + SpanGroup::Handler, + SpanGroup::HostFn, + SpanGroup::Store, + SpanGroup::GraphQl, + ]; + + /// The name used in `GRAPH_OTEL_SPANS`. + pub fn name(self) -> &'static str { + match self { + SpanGroup::Block => "block", + SpanGroup::Handler => "handler", + SpanGroup::HostFn => "host_fn", + SpanGroup::Store => "store", + SpanGroup::GraphQl => "graphql", + } + } + + fn from_name(name: &str) -> Option { + Self::ALL.into_iter().find(|group| group.name() == name) + } + + fn bit(self) -> u8 { + 1 << (self as u8) + } +} + +/// A set of [`SpanGroup`]s, parsed from `GRAPH_OTEL_SPANS`. +/// +/// The variable is a comma-separated list of group names. `all` and `none` +/// are accepted as shorthands. Unknown names are an error so that a typo +/// does not silently disable a group. +#[derive(Clone, Copy, PartialEq, Eq)] +pub struct SpanGroups(u8); + +impl SpanGroups { + pub const NONE: SpanGroups = SpanGroups(0); + pub const ALL: SpanGroups = SpanGroups(0b11111); + + pub fn contains(self, group: SpanGroup) -> bool { + self.0 & group.bit() != 0 + } + + pub fn with(self, group: SpanGroup) -> SpanGroups { + SpanGroups(self.0 | group.bit()) + } + + pub fn iter(self) -> impl Iterator { + SpanGroup::ALL + .into_iter() + .filter(move |group| self.contains(*group)) + } +} + +impl Default for SpanGroups { + /// Everything except `host_fn`, which is too verbose to be on by default. + fn default() -> Self { + SpanGroups::NONE + .with(SpanGroup::Block) + .with(SpanGroup::Handler) + .with(SpanGroup::Store) + .with(SpanGroup::GraphQl) + } +} + +impl FromStr for SpanGroups { + type Err = String; + + fn from_str(s: &str) -> Result { + let mut groups = SpanGroups::NONE; + for name in s.split(',').map(str::trim).filter(|name| !name.is_empty()) { + match name { + "all" => groups = SpanGroups::ALL, + "none" => groups = SpanGroups::NONE, + name => match SpanGroup::from_name(name) { + Some(group) => groups = groups.with(group), + None => { + let known = SpanGroup::ALL + .iter() + .map(|group| group.name()) + .collect::>() + .join(", "); + return Err(format!( + "unknown span group `{name}`, expected one of {known}, all, none" + )); + } + }, + } + } + Ok(groups) + } +} + +impl fmt::Display for SpanGroups { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let names: Vec<_> = self.iter().map(SpanGroup::name).collect(); + write!(f, "{}", names.join(",")) + } +} + +impl fmt::Debug for SpanGroups { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "SpanGroups({self})") + } +} + +/// The identity of a span, used to start children of it and to carry a +/// trace across threads and tasks. +/// +/// A `Context` is only ever obtained from a live, sampled [`Span`] or from +/// an incoming request header, so holding one implies there is something +/// to attach to. +#[derive(Clone, Debug)] +pub struct Context(opentelemetry::Context); + +impl Context { + fn from_span_context(span_context: SpanContext) -> Context { + Context(opentelemetry::Context::new().with_remote_span_context(span_context)) + } + + /// The context attached to the current thread, if any. + /// + /// Only code that runs on one thread for the lifetime of a [`ContextGuard`] + /// should rely on this. Everywhere else, pass a `Context` explicitly. + pub fn current() -> Option { + if !enabled() { + return None; + } + let cx = opentelemetry::Context::current(); + if cx.span().span_context().is_valid() { + Some(Context(cx)) + } else { + None + } + } + + /// Make this the current context of the calling thread until the guard + /// is dropped. The guard must be dropped on the thread that created it + /// and must not be held across an `await`. + pub fn attach(self) -> ContextGuard { + ContextGuard(self.0.attach()) + } + + fn span_context(&self) -> SpanContext { + self.0.span().span_context().clone() + } + + /// The trace id as a hex string, for logging and tests. + pub fn trace_id(&self) -> String { + self.span_context().trace_id().to_string() + } +} + +/// Restores the previous thread-local context on drop. See +/// [`Context::attach`]. +#[must_use = "dropping the guard detaches the context"] +pub struct ContextGuard(#[allow(dead_code)] opentelemetry::ContextGuard); + +/// Read the W3C trace context headers from an incoming request. +/// +/// Returns `None` when tracing is off or the request carries no valid +/// `traceparent` header. +pub fn extract(headers: &http::HeaderMap) -> Option { + if !enabled() { + return None; + } + let cx = global::get_text_map_propagator(|propagator| { + propagator.extract_with_context(&opentelemetry::Context::new(), &HeaderExtractor(headers)) + }); + if cx.span().span_context().is_valid() { + Some(Context(cx)) + } else { + None + } +} + +/// Write W3C trace context headers for `ctx` into an outgoing request. +pub fn inject(ctx: &Context, headers: &mut http::HeaderMap) { + if !enabled() { + return; + } + global::get_text_map_propagator(|propagator| { + propagator.inject_context(&ctx.0, &mut HeaderInjector(headers)) + }); +} + +#[cfg(test)] +mod tests { + use http::{HeaderMap, HeaderValue}; + + use super::testing::{in_child_process, in_subprocess, install}; + use super::*; + + const TRACEPARENT_SAMPLED: &str = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"; + const TRACEPARENT_UNSAMPLED: &str = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-00"; + + #[test] + fn env_defaults_to_off() { + in_subprocess("telemetry::tests::env_defaults_to_off", &[], || { + let env = crate::env::EnvVars::from_env().unwrap(); + assert!(!env.otel_tracing); + assert_eq!(env.otel_spans, SpanGroups::default()); + }); + } + + #[test] + fn disabled_is_noop() { + in_subprocess("telemetry::tests::disabled_is_noop", &[], || { + assert!(!enabled()); + assert!(!group_enabled(SpanGroup::Block)); + + let mut span = Span::root("disabled"); + assert!(!span.is_recording()); + span.set_attr("k", "v"); + assert!(span.context().is_none()); + assert!(Context::current().is_none()); + + let mut headers = HeaderMap::new(); + headers.insert("traceparent", HeaderValue::from_static(TRACEPARENT_SAMPLED)); + assert!(extract(&headers).is_none()); + }); + } + + #[test] + fn config_from_env_defaults_to_off() { + in_subprocess( + "telemetry::tests::config_from_env_defaults_to_off", + &[], + || { + let env = crate::env::EnvVars::from_env().unwrap(); + assert_eq!(Config::from(&env), Config::default()); + assert!(!Config::default().enabled); + }, + ); + } + + #[test] + fn init_disabled_installs_nothing() { + let was_enabled = enabled(); + let guard = init(&Config::default(), &crate::log::discard()).unwrap(); + assert!(!guard.is_enabled()); + assert_eq!(enabled(), was_enabled); + guard.shutdown().unwrap(); + assert_eq!(enabled(), was_enabled); + } + + #[test] + fn otlp_endpoint_is_derived_like_the_sdk() { + let some = |s: &str| Some(s.to_string()); + assert_eq!( + otlp_traces_endpoint(None, None), + "http://localhost:4318/v1/traces" + ); + assert_eq!( + otlp_traces_endpoint(None, some("http://agent:4318/")), + "http://agent:4318/v1/traces" + ); + assert_eq!( + otlp_traces_endpoint(some("http://x:1/t"), some("http://agent:4318")), + "http://x:1/t" + ); + } + + #[test] + fn service_name_precedence() { + let detected = |name: &str| { + Resource::builder_empty() + .with_service_name(name.to_string()) + .build() + }; + let none = Resource::builder_empty().build(); + let some = |s: &str| Some(s.to_string()); + + assert_eq!(service_name(&none, None), "graph-node"); + assert_eq!(service_name(&detected(""), None), "graph-node"); + assert_eq!( + service_name(&detected("unknown_service"), None), + "graph-node" + ); + assert_eq!( + service_name(&detected("unknown_service:graph"), None), + "graph-node" + ); + assert_eq!( + service_name(&detected("unknown_service-indexer"), None), + "unknown_service-indexer" + ); + assert_eq!( + service_name(&detected("from-attributes"), None), + "from-attributes" + ); + assert_eq!(service_name(&none, some("indexer-7")), "indexer-7"); + assert_eq!( + service_name(&detected("from-attributes"), some("indexer-7")), + "indexer-7" + ); + } + + /// Starts a mock collector in the parent process, then runs `init` + /// against it in a subprocess. `expected_service_name` must show up in + /// the exported protobuf body. + fn export_to_mock_collector( + test_name: &str, + env: &[(&str, &str)], + expected_service_name: &str, + ) { + use wiremock::matchers::{header, method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + if in_child_process() { + let config = Config { + enabled: true, + groups: SpanGroups::default(), + }; + let guard = init(&config, &crate::log::discard()).unwrap(); + assert!(guard.is_enabled()); + assert!(enabled()); + assert!(group_enabled(SpanGroup::Block)); + assert!(!group_enabled(SpanGroup::HostFn)); + + let mut span = Span::root("exported"); + span.set_attr("test", test_name.to_string()); + span.end(); + + // Shutting down flushes the batch processor. + guard.shutdown().unwrap(); + return; + } + + let rt = tokio::runtime::Runtime::new().unwrap(); + let server = rt.block_on(async { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/v1/traces")) + .and(header("content-type", "application/x-protobuf")) + .respond_with(ResponseTemplate::new(200)) + .expect(1..) + .mount(&server) + .await; + server + }); + + let uri = server.uri(); + let mut child_env = vec![("OTEL_EXPORTER_OTLP_ENDPOINT", uri.as_str())]; + child_env.extend_from_slice(env); + in_subprocess(test_name, &child_env, || unreachable!()); + + let requests = rt.block_on(async { server.received_requests().await.unwrap() }); + assert!( + !requests.is_empty(), + "no export request reached the collector" + ); + // The resource name and the span name travel as plain strings + // inside the protobuf body. + let body = requests + .iter() + .flat_map(|req| req.body.iter().copied()) + .collect::>(); + let body = String::from_utf8_lossy(&body); + assert!( + body.contains(expected_service_name), + "service.name `{expected_service_name}` missing from export" + ); + assert!(body.contains("exported"), "span name missing from export"); + rt.block_on(async { drop(server) }); + } + + #[test] + fn init_exports_over_otlp_http() { + export_to_mock_collector( + "telemetry::tests::init_exports_over_otlp_http", + &[], + "graph-node", + ); + } + + #[test] + fn init_honours_otel_service_name() { + export_to_mock_collector( + "telemetry::tests::init_honours_otel_service_name", + &[ + ("OTEL_SERVICE_NAME", "indexer-7"), + ("OTEL_RESOURCE_ATTRIBUTES", "service.name=from-attributes"), + ], + "indexer-7", + ); + } + + #[test] + fn init_honours_resource_attributes_service_name() { + export_to_mock_collector( + "telemetry::tests::init_honours_resource_attributes_service_name", + &[("OTEL_RESOURCE_ATTRIBUTES", "service.name=from-attributes")], + "from-attributes", + ); + } + + #[test] + fn span_groups_parse() { + let groups: SpanGroups = "block, host_fn".parse().unwrap(); + assert!(groups.contains(SpanGroup::Block)); + assert!(groups.contains(SpanGroup::HostFn)); + assert!(!groups.contains(SpanGroup::Store)); + assert_eq!(groups.to_string(), "block,host_fn"); + + assert_eq!("all".parse::().unwrap(), SpanGroups::ALL); + assert_eq!("none".parse::().unwrap(), SpanGroups::NONE); + assert_eq!("".parse::().unwrap(), SpanGroups::NONE); + assert_eq!( + "block,handler,store,graphql".parse::().unwrap(), + SpanGroups::default() + ); + assert!(!SpanGroups::default().contains(SpanGroup::HostFn)); + + let err = "block,hostfn".parse::().unwrap_err(); + assert!(err.contains("hostfn"), "{err}"); + } + + #[test] + fn root_and_child_are_exported() { + let handle = install(); + assert!(enabled()); + assert!(group_enabled(SpanGroup::HostFn)); + + let mut root = Span::root("parent"); + root.set_attr("test", "root_and_child_are_exported"); + root.set_attr("number", 7i64); + let parent_ctx = root.context().expect("sampled root has a context"); + let mut child = Span::child("child", &parent_ctx); + child.set_error("boom"); + child.end(); + root.end(); + + let roots = handle.spans_with_attr("test", "root_and_child_are_exported"); + assert_eq!(roots.len(), 1); + let root = &roots[0]; + assert_eq!(root.name, "parent"); + assert_eq!(root.parent_span_id, None); + assert_eq!(root.attr("number"), Some("7")); + + let children = handle.children_of(root); + assert_eq!(children.len(), 1); + assert_eq!(children[0].name, "child"); + assert_eq!(children[0].trace_id, root.trace_id); + assert!(children[0].is_error); + } + + #[test] + fn error_message_is_bounded() { + let handle = install(); + let mut span = Span::root("failed"); + span.set_attr("test", "error_message_is_bounded"); + // Multi-byte characters, so that truncation must respect char + // boundaries to produce valid UTF-8. + span.set_error(&"ü".repeat(MAX_ERROR_MESSAGE_LEN)); + span.end(); + + let spans = handle.spans_with_attr("test", "error_message_is_bounded"); + assert_eq!(spans.len(), 1); + assert!(spans[0].is_error); + let message = spans[0].error_message.as_ref().unwrap(); + assert!(message.len() <= MAX_ERROR_MESSAGE_LEN); + assert!(message.len() > MAX_ERROR_MESSAGE_LEN - 2); + assert!(message.chars().all(|c| c == 'ü')); + } + + #[test] + fn dropped_span_is_exported() { + let handle = install(); + { + let mut span = Span::root("dropped"); + span.set_attr("test", "dropped_span_is_exported"); + } + assert_eq!( + handle + .spans_with_attr("test", "dropped_span_is_exported") + .len(), + 1 + ); + } + + #[test] + fn links_are_exported() { + let handle = install(); + + let mut a = Span::root("a"); + a.set_attr("test", "links_are_exported"); + let mut b = Span::root("b"); + b.set_attr("test", "links_are_exported"); + let links = [a.context().unwrap(), b.context().unwrap()]; + let mut linked = Span::root_with_links("linked", &links); + linked.set_attr("test", "links_are_exported"); + linked.end(); + a.end(); + b.end(); + + let spans = handle.spans_with_attr("test", "links_are_exported"); + let linked = spans.iter().find(|s| s.name == "linked").unwrap(); + let mut linked_to: Vec<_> = linked.links.iter().map(|l| l.0.as_str()).collect(); + linked_to.sort(); + let mut expected: Vec<_> = spans + .iter() + .filter(|s| s.name != "linked") + .map(|s| s.trace_id.as_str()) + .collect(); + expected.sort(); + assert_eq!(linked_to, expected); + } + + #[test] + fn extract_and_inject_roundtrip() { + install(); + + let mut headers = HeaderMap::new(); + assert!(extract(&headers).is_none()); + + headers.insert("traceparent", HeaderValue::from_static(TRACEPARENT_SAMPLED)); + let ctx = extract(&headers).expect("valid traceparent"); + assert_eq!(ctx.trace_id(), "0af7651916cd43dd8448eb211c80319c"); + + let child = Span::child("child", &ctx); + let child_ctx = child.context().unwrap(); + assert_eq!(child_ctx.trace_id(), ctx.trace_id()); + + let mut out = HeaderMap::new(); + inject(&child_ctx, &mut out); + let traceparent = out.get("traceparent").unwrap().to_str().unwrap(); + assert!( + traceparent.starts_with("00-0af7651916cd43dd8448eb211c80319c-"), + "{traceparent}" + ); + assert!(traceparent.ends_with("-01"), "{traceparent}"); + } + + #[test] + fn unsampled_parent_yields_noop_child() { + install(); + + let mut headers = HeaderMap::new(); + headers.insert( + "traceparent", + HeaderValue::from_static(TRACEPARENT_UNSAMPLED), + ); + let ctx = extract(&headers).expect("an unsampled traceparent is still valid"); + let child = Span::child("child", &ctx); + assert!(!child.is_recording()); + assert!(child.context().is_none()); + } + + #[test] + fn attach_makes_context_current() { + install(); + + assert!(Context::current().is_none()); + let span = Span::root("attached"); + let ctx = span.context().unwrap(); + { + let _guard = ctx.clone().attach(); + let current = Context::current().expect("attached context is current"); + assert_eq!(current.trace_id(), ctx.trace_id()); + } + assert!(Context::current().is_none()); + } + + #[test] + fn root_ignores_attached_context() { + let handle = install(); + + let outer = Span::root("outer"); + let outer_ctx = outer.context().unwrap(); + let _guard = outer_ctx.clone().attach(); + + let mut root = Span::root("inner"); + root.set_attr("test", "root_ignores_attached_context"); + let root_ctx = root.context().unwrap(); + assert_ne!(root_ctx.trace_id(), outer_ctx.trace_id()); + root.end(); + + let spans = handle.spans_with_attr("test", "root_ignores_attached_context"); + assert_eq!(spans.len(), 1); + assert_eq!(spans[0].parent_span_id, None); + } +} diff --git a/graph/src/telemetry/span.rs b/graph/src/telemetry/span.rs new file mode 100644 index 00000000000..14df8f569dd --- /dev/null +++ b/graph/src/telemetry/span.rs @@ -0,0 +1,201 @@ +use std::borrow::Cow; + +use opentelemetry::global::{self, BoxedSpan}; +use opentelemetry::trace::{Link, Span as _, SpanBuilder, Status, Tracer}; +use opentelemetry::{KeyValue, Value}; + +use super::{Context, enabled}; + +/// The instrumentation scope under which all graph-node spans are reported. +const TRACER_NAME: &str = "graph-node"; + +/// Error messages on spans are cut to this many bytes. Subgraph errors can +/// carry a WASM backtrace, which has no place in a span status. +pub const MAX_ERROR_MESSAGE_LEN: usize = 4096; + +/// A span attribute value. Mirrors the subset of OpenTelemetry value types +/// that graph-node uses. +#[derive(Clone, Debug, PartialEq)] +pub enum AttrValue { + Str(Cow<'static, str>), + I64(i64), + Bool(bool), +} + +impl From<&'static str> for AttrValue { + fn from(value: &'static str) -> Self { + AttrValue::Str(Cow::Borrowed(value)) + } +} + +impl From for AttrValue { + fn from(value: String) -> Self { + AttrValue::Str(Cow::Owned(value)) + } +} + +impl From> for AttrValue { + fn from(value: Cow<'static, str>) -> Self { + AttrValue::Str(value) + } +} + +impl From for AttrValue { + fn from(value: i64) -> Self { + AttrValue::I64(value) + } +} + +impl From for AttrValue { + fn from(value: i32) -> Self { + AttrValue::I64(value as i64) + } +} + +impl From for AttrValue { + fn from(value: u32) -> Self { + AttrValue::I64(value as i64) + } +} + +impl From for AttrValue { + fn from(value: usize) -> Self { + AttrValue::I64(i64::try_from(value).unwrap_or(i64::MAX)) + } +} + +impl From for AttrValue { + fn from(value: bool) -> Self { + AttrValue::Bool(value) + } +} + +impl From for Value { + fn from(value: AttrValue) -> Self { + match value { + AttrValue::Str(s) => Value::String(s.into()), + AttrValue::I64(i) => Value::I64(i), + AttrValue::Bool(b) => Value::Bool(b), + } + } +} + +/// A span that is being recorded, or a no-op. +/// +/// The span ends when `end` is called or when it is dropped, whichever +/// comes first. A `Span` is a no-op when tracing is disabled or when the +/// trace it belongs to was not sampled; a no-op span has no [`Context`], +/// so nothing can be parented to it and children are never built. +#[must_use = "a span ends when it is dropped"] +pub struct Span(Option); + +impl Span { + /// A span that records nothing. + pub fn noop() -> Span { + Span(None) + } + + /// Start a new trace. + pub fn root(name: impl Into>) -> Span { + Self::start(name.into(), None, &[]) + } + + /// Start a new trace that points at other traces without being part + /// of them, for work that was caused by several of them. + pub fn root_with_links(name: impl Into>, links: &[Context]) -> Span { + Self::start(name.into(), None, links) + } + + /// Start a span in the trace of `parent`. + pub fn child(name: impl Into>, parent: &Context) -> Span { + Self::start(name.into(), Some(parent), &[]) + } + + fn start(name: Cow<'static, str>, parent: Option<&Context>, links: &[Context]) -> Span { + if !enabled() { + return Span(None); + } + + let mut builder = SpanBuilder::from_name(name); + if !links.is_empty() { + let links = links + .iter() + .map(|ctx| Link::new(ctx.span_context(), Vec::new(), 0)) + .collect(); + builder = builder.with_links(links); + } + + let tracer = global::tracer(TRACER_NAME); + let span = match parent { + Some(parent) => tracer.build_with_context(builder, &parent.0), + // An explicit empty context, not the thread-local one, so that a + // root is always a root regardless of what is attached. + None => tracer.build_with_context(builder, &opentelemetry::Context::new()), + }; + + // Unsampled spans are not exported and must not become parents, so + // treat them as no-ops from the start. + if span.span_context().is_sampled() { + Span(Some(span)) + } else { + Span(None) + } + } + + /// `true` if this span will be exported. + pub fn is_recording(&self) -> bool { + self.0.is_some() + } + + pub fn set_attr(&mut self, key: &'static str, value: impl Into) { + if let Some(span) = self.0.as_mut() { + span.set_attribute(KeyValue::new(key, Value::from(value.into()))); + } + } + + pub fn set_attrs(&mut self, attrs: impl IntoIterator) { + if let Some(span) = self.0.as_mut() { + span.set_attributes( + attrs + .into_iter() + .map(|(key, value)| KeyValue::new(key, Value::from(value))), + ); + } + } + + /// Mark the span as failed. The message shows up as the span status, + /// cut to [`MAX_ERROR_MESSAGE_LEN`] bytes. + pub fn set_error(&mut self, message: &str) { + if let Some(span) = self.0.as_mut() { + let mut end = message.len().min(MAX_ERROR_MESSAGE_LEN); + while !message.is_char_boundary(end) { + end -= 1; + } + span.set_status(Status::error(message[..end].to_owned())); + } + } + + /// The context to start children under, or `None` for a no-op span. + pub fn context(&self) -> Option { + self.0 + .as_ref() + .map(|span| Context::from_span_context(span.span_context().clone())) + } + + /// End the span now. Dropping it has the same effect; this reads better + /// when the end is a deliberate point in the code. + pub fn end(mut self) { + if let Some(span) = self.0.as_mut() { + span.end(); + } + } +} + +impl std::fmt::Debug for Span { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match &self.0 { + Some(span) => write!(f, "Span({})", span.span_context().span_id()), + None => write!(f, "Span(noop)"), + } + } +} diff --git a/graph/src/telemetry/testing.rs b/graph/src/telemetry/testing.rs new file mode 100644 index 00000000000..d8483016fe7 --- /dev/null +++ b/graph/src/telemetry/testing.rs @@ -0,0 +1,217 @@ +//! An in-memory span exporter for tests. +//! +//! The tracer provider is global to the process, so a test binary installs +//! it once with [`install`] and every test in that binary shares it. Spans +//! are never cleared. Tests tell their own spans apart by putting a unique +//! attribute on them and using [`Handle::spans_with_attr`], or by following +//! a trace id they obtained from a [`Context`]. + +use std::future::{Future, ready}; +use std::process::Command; +use std::sync::{Arc, Mutex, OnceLock}; + +use opentelemetry::global; +use opentelemetry::trace::{SpanId, Status}; +use opentelemetry_sdk::error::OTelSdkResult; +use opentelemetry_sdk::propagation::TraceContextPropagator; +use opentelemetry_sdk::trace::{SdkTracerProvider, SpanData, SpanExporter}; + +use super::{SpanGroups, set_enabled}; + +/// What the exporter remembers about a span, in plain strings so that tests +/// in other crates do not need the OpenTelemetry types. +#[derive(Clone, Debug, PartialEq)] +pub struct ExportedSpan { + pub name: String, + pub trace_id: String, + pub span_id: String, + /// `None` for a root span. + pub parent_span_id: Option, + pub attributes: Vec<(String, String)>, + /// `(trace_id, span_id)` of each linked span. + pub links: Vec<(String, String)>, + pub is_error: bool, + /// The status description when `is_error`. + pub error_message: Option, +} + +impl ExportedSpan { + pub fn attr(&self, key: &str) -> Option<&str> { + self.attributes + .iter() + .find(|(k, _)| k == key) + .map(|(_, v)| v.as_str()) + } + + pub fn has_attr(&self, key: &str, value: &str) -> bool { + self.attr(key) == Some(value) + } +} + +impl From<&SpanData> for ExportedSpan { + fn from(data: &SpanData) -> Self { + let parent_span_id = if data.parent_span_id == SpanId::INVALID { + None + } else { + Some(data.parent_span_id.to_string()) + }; + ExportedSpan { + name: data.name.to_string(), + trace_id: data.span_context.trace_id().to_string(), + span_id: data.span_context.span_id().to_string(), + parent_span_id, + attributes: data + .attributes + .iter() + .map(|kv| (kv.key.to_string(), kv.value.to_string())) + .collect(), + links: data + .links + .iter() + .map(|link| { + ( + link.span_context.trace_id().to_string(), + link.span_context.span_id().to_string(), + ) + }) + .collect(), + is_error: matches!(data.status, Status::Error { .. }), + error_message: match &data.status { + Status::Error { description } => Some(description.to_string()), + _ => None, + }, + } + } +} + +#[derive(Clone, Debug, Default)] +struct Exporter(Arc>>); + +impl SpanExporter for Exporter { + fn export(&self, batch: Vec) -> impl Future + Send { + let mut spans = self.0.lock().unwrap(); + spans.extend(batch.iter().map(ExportedSpan::from)); + ready(Ok(())) + } +} + +/// Access to the spans exported so far in this process. +pub struct Handle { + exporter: Exporter, + provider: SdkTracerProvider, +} + +impl Handle { + /// Every span exported so far, in export order. + pub fn spans(&self) -> Vec { + // The simple processor exports on span end, so this is a formality. + let _ = self.provider.force_flush(); + self.exporter.0.lock().unwrap().clone() + } + + /// Spans carrying `key = value`. Give each test a unique value to pick + /// out its own spans. + pub fn spans_with_attr(&self, key: &str, value: &str) -> Vec { + self.spans() + .into_iter() + .filter(|span| span.has_attr(key, value)) + .collect() + } + + /// Every span of one trace, in export order. Children end before their + /// parents, so a root is the last span of its trace. + pub fn trace(&self, trace_id: &str) -> Vec { + self.spans() + .into_iter() + .filter(|span| span.trace_id == trace_id) + .collect() + } + + /// Direct children of `parent`, in export order. + pub fn children_of(&self, parent: &ExportedSpan) -> Vec { + self.spans() + .into_iter() + .filter(|span| span.parent_span_id.as_deref() == Some(parent.span_id.as_str())) + .collect() + } +} + +static HANDLE: OnceLock = OnceLock::new(); + +/// Install the in-memory exporter as the process-wide tracer provider with +/// every span group enabled, or return the one installed earlier. +pub fn install() -> &'static Handle { + install_with(SpanGroups::ALL) +} + +/// Like [`install`], with only `groups` enabled. The first call in a +/// process decides; later calls return the existing handle whatever their +/// argument, so a test that needs specific groups runs in a subprocess, +/// see [`in_subprocess`]. +pub fn install_with(groups: SpanGroups) -> &'static Handle { + HANDLE.get_or_init(|| { + let exporter = Exporter::default(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + global::set_tracer_provider(provider.clone()); + global::set_text_map_propagator(TraceContextPropagator::new()); + set_enabled(groups); + Handle { exporter, provider } + }) +} + +/// Marks a child process spawned by [`run_in_subprocess`]. +pub(super) const CHILD_MARKER: &str = "GRAPH_TELEMETRY_TEST_CHILD"; + +/// `true` inside a process started by [`run_in_subprocess`]. +pub fn in_child_process() -> bool { + std::env::var_os(CHILD_MARKER).is_some() +} + +/// Run `body` in a fresh process with every `OTEL_*` and `GRAPH_OTEL_*` +/// variable unset and `env` set. The tracer provider and the enabled flag +/// are process-global, so this is the only way to observe the disabled +/// state, or to install a provider with specific groups, without racing +/// other tests in the same binary. +/// +/// The parent re-executes the test binary for the single test named +/// `test_name` with a marker variable set; the child sees the marker and +/// runs `body` instead of spawning again. The child never mutates its +/// environment, so `body` may start threads freely. +/// +/// An `async` test cannot pass its body as a closure; it checks +/// [`in_child_process`] itself and calls [`run_in_subprocess`] when it is +/// the parent. +pub fn in_subprocess(test_name: &str, env: &[(&str, &str)], body: impl FnOnce()) { + if in_child_process() { + body(); + } else { + run_in_subprocess(test_name, env); + } +} + +/// The parent half of [`in_subprocess`]: re-execute this test binary for +/// `test_name` and assert that it passed. +pub fn run_in_subprocess(test_name: &str, env: &[(&str, &str)]) { + let exe = std::env::current_exe().unwrap(); + let mut command = Command::new(exe); + command.args(["--exact", test_name, "--test-threads=1"]); + for (name, _) in std::env::vars() { + if name.starts_with("OTEL_") || name.starts_with("GRAPH_OTEL_") { + command.env_remove(name); + } + } + command.env(CHILD_MARKER, "1"); + command.envs(env.iter().copied()); + + let output = command.output().unwrap(); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!( + output.status.success(), + "{test_name} failed in subprocess\n--- stdout\n{stdout}\n--- stderr\n{}", + String::from_utf8_lossy(&output.stderr) + ); + // Make sure the subprocess actually ran the test. + assert!(stdout.contains("1 passed"), "{stdout}"); +} diff --git a/graphql/src/execution/execution.rs b/graphql/src/execution/execution.rs index c3d116e6c64..7a096cd9515 100644 --- a/graphql/src/execution/execution.rs +++ b/graphql/src/execution/execution.rs @@ -21,6 +21,7 @@ use graph::data::query::CacheStatus; use graph::env::CachedSubgraphIds; use graph::prelude::*; use graph::schema::ast as sast; +use graph::telemetry; use graph::util::{lfu_cache::LfuCache, stable_hash_glue::impl_stable_hash}; use super::QueryHash; @@ -490,6 +491,18 @@ pub(crate) async fn execute_root_selection_set( let query_text = execute_ctx.query.query_text.cheap_clone(); let variables_text = execute_ctx.query.variables_text.cheap_clone(); match graph::spawn_blocking_allow_panic(move || { + // Make the request's trace visible to the store, which starts + // its query spans under the thread-local context. This is + // sound only because the future below is driven to completion + // on this thread; a `tokio::spawn` inside it would run on a + // thread without the context and its spans would not be + // parented. The test `query_spans_join_the_request_trace` in + // the store tests fails if that happens. + let _trace_guard = execute_ctx + .query + .trace_context + .clone() + .map(telemetry::Context::attach); let mut query_res = QueryResult::from( graph::block_on(execute_root_selection_set_uncached( &execute_ctx, diff --git a/graphql/src/execution/query.rs b/graphql/src/execution/query.rs index c12d68d7fa9..cf0005971cb 100644 --- a/graphql/src/execution/query.rs +++ b/graphql/src/execution/query.rs @@ -6,11 +6,10 @@ use graphql_tools::validation::rules::*; use graphql_tools::validation::validate::{ValidationPlan, validate}; use lazy_static::lazy_static; use std::collections::{BTreeMap, HashMap, HashSet}; -use std::hash::{Hash, Hasher}; +use std::convert::TryFrom; use std::iter::FromIterator; use std::sync::Arc; use std::time::Instant; -use std::{collections::hash_map::DefaultHasher, convert::TryFrom}; use graph::data::graphql::{ObjectOrInterface, ext::TypeExt}; use graph::data::query::{Query as GraphDataQuery, QueryVariables}; @@ -21,6 +20,7 @@ use graph::prelude::{ }; use graph::schema::ErrorPolicy; use graph::schema::ast::{self as sast}; +use graph::telemetry; use crate::execution::ast as a; use crate::execution::get_field; @@ -128,6 +128,8 @@ pub struct Query { pub query_text: Arc, pub variables_text: Arc, pub query_id: String, + /// See [`GraphDataQuery::trace_context`] + pub trace_context: Option, } fn validate_query( @@ -185,13 +187,7 @@ impl Query { max_depth: u8, metrics: Arc, ) -> Result, Vec> { - let query_hash = { - let mut hasher = DefaultHasher::new(); - query.query_text.hash(&mut hasher); - query.variables_text.hash(&mut hasher); - hasher.finish() - }; - let query_id = format!("{:x}-{:x}", query.shape_hash, query_hash); + let query_id = query.query_id(); let logger = logger.new(o!( "subgraph_id" => schema.id().clone(), "query_id" => query_id.clone() @@ -264,6 +260,7 @@ impl Query { query_text: query.query_text.cheap_clone(), variables_text: query.variables_text.cheap_clone(), query_id, + trace_context: query.trace_context, }; Ok(Arc::new(query)) diff --git a/node/src/main.rs b/node/src/main.rs index 652258c734d..cdb141aef64 100644 --- a/node/src/main.rs +++ b/node/src/main.rs @@ -38,6 +38,28 @@ async fn main_inner() { "Runtime configured with {} max blocking threads", *MAX_BLOCKING_THREADS ); + let telemetry = graph::telemetry::init(&graph::telemetry::Config::from(&*env_vars), &logger) + .unwrap_or_else(|err| panic!("Failed to initialize OpenTelemetry tracing: {err:#}")); + if telemetry.is_enabled() { + // graph-node has no shutdown sequence; it runs until it is killed. + // With tracing on, catch the termination signals so that buffered + // spans are flushed before the process exits. + let logger = logger.clone(); + graph::spawn(async move { + let code = wait_for_termination().await; + info!(logger, "Shutting down, flushing traces"); + // Shutdown blocks until the export finished or timed out. + let result = graph::spawn_blocking_allow_panic(move || telemetry.shutdown()) + .await + .map_err(graph::anyhow::Error::from) + .and_then(|result| result); + if let Err(err) = result { + error!(logger, "Failed to flush traces"; "error" => format!("{err:#}")); + } + std::process::exit(code); + }); + } + let (prometheus_registry, metrics_registry) = launcher::setup_metrics(&logger); let ipfs_client = graph::ipfs::new_ipfs_client(&opt.ipfs, &metrics_registry, &logger) @@ -66,3 +88,22 @@ async fn main_inner() { ) .await; } + +/// Wait for SIGINT or SIGTERM and return the exit code the default signal +/// disposition would have produced. +async fn wait_for_termination() -> i32 { + #[cfg(unix)] + { + use tokio::signal::unix::{SignalKind, signal}; + let mut terminate = signal(SignalKind::terminate()).expect("can listen for SIGTERM"); + tokio::select! { + _ = tokio::signal::ctrl_c() => 130, + _ = terminate.recv() => 143, + } + } + #[cfg(not(unix))] + { + let _ = tokio::signal::ctrl_c().await; + 130 + } +} diff --git a/runtime/wasm/src/mapping.rs b/runtime/wasm/src/mapping.rs index c185148bff0..e24a234aed6 100644 --- a/runtime/wasm/src/mapping.rs +++ b/runtime/wasm/src/mapping.rs @@ -154,7 +154,14 @@ where { let logger = logger.cheap_clone(); - let _section = host_metrics.stopwatch.start_section("run_handler"); + // The handler span names the handler and its trigger. + let _section = host_metrics + .stopwatch + .start_section_with("run_handler", || { + let mut attrs = vec![("handler", trigger.handler_name().to_owned().into())]; + attrs.extend(trigger.trigger.trace_attributes()); + attrs + }); if ENV_VARS.log_trigger_data { debug!(logger, "trigger data: {:?}", trigger); } diff --git a/runtime/wasm/src/module/context.rs b/runtime/wasm/src/module/context.rs index 70d65489304..f3a0cf690c3 100644 --- a/runtime/wasm/src/module/context.rs +++ b/runtime/wasm/src/module/context.rs @@ -260,12 +260,13 @@ impl WasmInstanceContext<'_> { ) -> Result<(), HostExportError> { let stopwatch = self.as_ref().host_metrics.stopwatch.cheap_clone(); let logger = self.as_ref().ctx.logger.cheap_clone(); - stopwatch.start_section("host_export_store_set__wasm_instance_context_store_set"); let entity: String = asc_get(self, entity_ptr, gas)?; let id: String = asc_get(self, id_ptr, gas)?; let data = asc_get(self, data_ptr, gas)?; + stopwatch.set_attrs_with(|| [("entity_type", entity.clone().into())]); + if self.as_ref().ctx.instrument { debug!(self.as_ref().ctx.logger, "store_set"; "type" => &entity, diff --git a/runtime/wasm/src/module/instance.rs b/runtime/wasm/src/module/instance.rs index 79017beadc1..0db36b246f0 100644 --- a/runtime/wasm/src/module/instance.rs +++ b/runtime/wasm/src/module/instance.rs @@ -297,7 +297,8 @@ fn link_chain_host_fn( let gas = caller.data().gas.cheap_clone(); let host_metrics = caller.data().host_metrics.cheap_clone(); let stopwatch = host_metrics.stopwatch.cheap_clone(); - let _section = stopwatch.start_section(§ion_name); + let _section = stopwatch + .start_section_with(§ion_name, || [("host_function", name.into())]); let ctx = HostFnCtx { logger: caller.data().ctx.logger.cheap_clone(), @@ -379,7 +380,10 @@ pub(crate) fn build_linker( Box::new(async move { let gas = caller.data().gas.cheap_clone(); let host_metrics = caller.data().host_metrics.cheap_clone(); - let _section = host_metrics.stopwatch.start_section($section); + let _section = host_metrics.stopwatch.start_section_with( + $section, + || [("host_function", $wasm_name.into())], + ); #[allow(unused_mut)] let mut ctx = std::pin::pin!(WasmInstanceContext::new(&mut caller)); @@ -426,7 +430,10 @@ pub(crate) fn build_linker( Box::new(async move { let gas = caller.data().gas.cheap_clone(); let host_metrics = caller.data().host_metrics.cheap_clone(); - let _section = host_metrics.stopwatch.start_section($section); + let _section = host_metrics.stopwatch.start_section_with( + $section, + || [("host_function", $wasm_name.into())], + ); #[allow(unused_mut)] let mut ctx = WasmInstanceContext::new(&mut caller); diff --git a/server/http/src/service.rs b/server/http/src/service.rs index 66a799caed3..365bbe7bb92 100644 --- a/server/http/src/service.rs +++ b/server/http/src/service.rs @@ -28,6 +28,7 @@ use graph::prelude::serde_json::json; use graph::semver::VersionReq; use graph::slog::Logger; use graph::slog::error; +use graph::telemetry::{self, Span, SpanGroup}; use graph::url::form_urlencoded; use graph::{components::server::query::ServerError, data::query::QueryTarget}; @@ -47,6 +48,46 @@ fn client_error(msg: impl Into) -> ServerResponse { .unwrap() } +/// Start the span for a GraphQL request, as a child of the trace the +/// request's W3C headers refer to, or as a new trace. The `graphql` span +/// group must be enabled, or the span is a no-op. +fn request_span(request: &Request, target: &QueryTarget) -> Span { + if !telemetry::group_enabled(SpanGroup::GraphQl) { + return Span::noop(); + } + let mut span = match telemetry::extract(request.headers()) { + Some(parent) => Span::child("graphql_query", &parent), + None => Span::root("graphql_query"), + }; + if span.is_recording() { + let (subgraph, version) = match target { + QueryTarget::Name(name, version) => (name.to_string(), version), + QueryTarget::Deployment(id, version) => (id.to_string(), version), + }; + span.set_attrs([ + ("subgraph", subgraph.into()), + ("api_version", version.version.to_string().into()), + ]); + } + span +} + +/// Join `errors` for a span status, stopping once the status is as long +/// as it can get anyway. +fn error_message(errors: &[QueryError]) -> String { + let mut message = String::new(); + for error in errors { + if message.len() >= telemetry::MAX_ERROR_MESSAGE_LEN { + break; + } + if !message.is_empty() { + message.push_str("; "); + } + message.push_str(&error.to_string()); + } + message +} + /// A Hyper Service that serves GraphQL over a POST / endpoint. #[derive(Debug)] pub struct GraphQLService { @@ -168,25 +209,50 @@ where }) .unwrap_or(false) }; - let body = request - .collect() - .await - .map_err(|_| ServerError::InternalError("Failed to read request body".into()))? - .to_bytes(); + let mut span = request_span(&request, &target); + let body = match request.collect().await { + Ok(body) => body.to_bytes(), + Err(_) => { + let err = ServerError::InternalError("Failed to read request body".into()); + span.set_error(&err.to_string()); + return Err(err); + } + }; let query = parse_graphql_request(&body, trace); let query_parsing_time = start.elapsed(); let mut result = match query { Ok(query) => { + if span.is_recording() { + // The same id that query timing logs carry, so the two + // can be matched up. + span.set_attr("query_id", query.query_id()); + } + let query = query.with_trace_context(span.context()); self.graphql_runner .cheap_clone() .run_query(query, target) .await } Err(ServerError::QueryError(e)) => QueryResult::from(e).into(), - Err(e) => return Err(e), + Err(e) => { + span.set_error(&e.to_string()); + return Err(e); + } }; + if span.is_recording() { + if let Some(deployment) = result.deployment_hash() { + span.set_attr("deployment", deployment.to_string()); + } + if result.has_errors() { + let errors = result.errors(); + span.set_attr("error_count", errors.len()); + span.set_error(&error_message(&errors)); + } + } + span.end(); + result.trace.query_parsing(query_parsing_time); self.graphql_runner .metrics() @@ -605,4 +671,105 @@ mod tests { .expect("Query result field \"name\" is not a string"); assert_eq!(name, "Jordi".to_string()); } + + mod spans { + use graph::http_body_util::Full; + use graph::hyper::body::Bytes; + use graph::hyper::header::{CONTENT_LENGTH, CONTENT_TYPE}; + use graph::hyper::{Method, Request}; + use graph::prelude::*; + use graph::telemetry::testing::{ + ExportedSpan, Handle, in_child_process, install_with, run_in_subprocess, + }; + use graph::telemetry::{SpanGroup, SpanGroups}; + + use super::{GraphQLService, TestGraphQlRunner}; + + const TRACEPARENT: &str = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"; + + /// The tests in this module share the process-global exporter; the + /// first `install_with` wins, so they all ask for the same groups. + /// Each uses its own subgraph name to find its span. + fn install() -> &'static Handle { + install_with(SpanGroups::NONE.with(SpanGroup::GraphQl)) + } + + async fn post(subgraph: &str, traceparent: Option<&str>, query: &str) { + let logger = Logger::root(slog::Discard, o!()); + let service = GraphQLService::new(logger, Arc::new(TestGraphQlRunner)); + let body = serde_json::to_string(&serde_json::json!({ "query": query })).unwrap(); + let mut request = Request::builder() + .method(Method::POST) + .header(CONTENT_TYPE, "application/json") + .header(CONTENT_LENGTH, body.len()) + .uri(format!("http://localhost:8000/subgraphs/name/{subgraph}")); + if let Some(traceparent) = traceparent { + request = request.header("traceparent", traceparent); + } + let request: Request> = request.body(Full::from(body)).unwrap(); + service.call(request).await; + } + + fn request_span(spans: &Handle, subgraph: &str) -> ExportedSpan { + let mut spans = spans.spans_with_attr("subgraph", subgraph); + assert_eq!(spans.len(), 1, "one span for {subgraph}: {spans:?}"); + let span = spans.pop().unwrap(); + assert_eq!(span.name, "graphql_query"); + assert!(span.attr("api_version").is_some()); + span + } + + #[graph::test] + async fn joins_the_trace_from_the_traceparent_header() { + let spans = install(); + post("spans/traceparent", Some(TRACEPARENT), "{ name }").await; + + let span = request_span(spans, "spans/traceparent"); + assert!(span.attr("query_id").is_some()); + assert_eq!(span.trace_id, "0af7651916cd43dd8448eb211c80319c"); + assert_eq!(span.parent_span_id.as_deref(), Some("b7ad6b7169203331")); + assert!(!span.is_error); + } + + #[graph::test] + async fn starts_a_trace_without_a_traceparent_header() { + let spans = install(); + post("spans/root", None, "{ name }").await; + + let span = request_span(spans, "spans/root"); + assert_eq!(span.parent_span_id, None); + assert!(!span.is_error); + } + + #[graph::test] + async fn records_query_errors() { + let spans = install(); + post("spans/error", None, "{ name").await; + + // The query did not parse, so there is no query id. + let span = request_span(spans, "spans/error"); + assert_eq!(span.attr("query_id"), None); + assert!(span.is_error); + assert!( + span.error_message + .as_deref() + .is_some_and(|message| message.contains("Unexpected")), + "{:?}", + span.error_message + ); + } + + #[graph::test] + async fn needs_the_graphql_group() { + // The enabled groups are process-global, so this test gets a + // process of its own. + if !in_child_process() { + return run_in_subprocess("service::tests::spans::needs_the_graphql_group", &[]); + } + let spans = install_with(SpanGroups::NONE.with(SpanGroup::Store)); + post("spans/off", Some(TRACEPARENT), "{ name }").await; + + assert!(spans.spans().is_empty(), "{:?}", spans.spans()); + } + } } diff --git a/store/postgres/src/relational.rs b/store/postgres/src/relational.rs index 4059e736635..84bd1ea0740 100644 --- a/store/postgres/src/relational.rs +++ b/store/postgres/src/relational.rs @@ -44,6 +44,7 @@ use graph::schema::{ InputSchema, }; use graph::slog::{debug, warn}; +use graph::telemetry::{self, Context, Span, SpanGroup}; use inflector::Inflector; use itertools::Itertools; use lazy_static::lazy_static; @@ -94,6 +95,74 @@ const DELETE_OPERATION_CHUNK_SIZE: usize = 1_000; pub const STRING_PREFIX_SIZE: usize = 256; pub const BYTE_ARRAY_PREFIX_SIZE: usize = 64; +/// SQL text in logs and spans is cut to this many bytes (20kB). +const SQL_TEXT_MAXLEN: usize = 20_480; + +/// Cut `text` to at most `max` bytes, on a char boundary and including +/// the marker that shows it was cut. +fn truncate_sql(text: &mut String, max: usize) { + const MARKER: &str = " ..."; + if text.len() <= max { + return; + } + // Leave room for the marker when it fits at all. + let mut end = if max >= MARKER.len() { + max - MARKER.len() + } else { + max + }; + while !text.is_char_boundary(end) { + end -= 1; + } + text.truncate(end); + if max >= MARKER.len() { + text.push_str(MARKER); + } +} + +#[cfg(test)] +mod truncate_sql_tests { + use super::truncate_sql; + + #[test] + fn keeps_short_text() { + let mut text = "select 1".to_string(); + truncate_sql(&mut text, 20); + assert_eq!(text, "select 1"); + assert!(text.len() <= 20); + } + + #[test] + fn stays_within_max_bytes() { + let mut text = "x".repeat(100); + truncate_sql(&mut text, 10); + assert_eq!(text, "xxxxxx ..."); + assert!(text.len() <= 10); + } + + #[test] + fn respects_char_boundaries() { + // Each `ü` is two bytes; a cut at 7 would split one. + let mut text = "ü".repeat(50); + truncate_sql(&mut text, 11); + assert_eq!(text, "üüü ..."); + assert!(text.len() <= 11); + } + + #[test] + fn max_smaller_than_marker() { + let mut text = "abcdef".to_string(); + truncate_sql(&mut text, 2); + assert_eq!(text, "ab"); + assert!(text.len() <= 2); + + let mut text = "üüü".to_string(); + truncate_sql(&mut text, 3); + assert_eq!(text, "ü"); + assert!(text.len() <= 3); + } +} + lazy_static! { pub(crate) static ref STATEMENT_TIMEOUT: Option = ENV_VARS .graphql @@ -851,8 +920,7 @@ impl Layout { entity_count: usize, trace: bool, ) -> Trace { - // 20kB - const MAXLEN: usize = 20_480; + const MAXLEN: usize = SQL_TEXT_MAXLEN; if !ENV_VARS.log_sql_timing() && !trace { return Trace::None; @@ -870,10 +938,7 @@ impl Layout { // If the query + bind variables is more than MAXLEN, truncate it; // this will happen when queries have very large bind variables // (e.g., long arrays of string ids) - if text.len() > MAXLEN { - text.truncate(MAXLEN); - text.push_str(" ..."); - } + truncate_sql(&mut text, MAXLEN); info!( logger, "Query timing (SQL)"; @@ -903,6 +968,24 @@ impl Layout { let query_clone = query.clone(); + // The query is a child of whatever trace is attached to this + // thread, which is the GraphQL request's when there is one. With + // nothing attached there is no span. + let mut span = match Context::current() { + Some(ctx) if telemetry::group_enabled(SpanGroup::Store) => { + Span::child("db_query", &ctx) + } + _ => Span::noop(), + }; + if span.is_recording() { + span.set_attr("subgraph_id", self.site.deployment.to_string()); + if ENV_VARS.log_sql_timing() { + let mut text = debug_query(&query_clone).to_string().replace('\n', "\t"); + truncate_sql(&mut text, SQL_TEXT_MAXLEN); + span.set_attr("sql", text); + } + } + let start = Instant::now(); let values = conn .transaction(async |conn| { @@ -913,6 +996,7 @@ impl Layout { }) .await .map_err(|e| { + span.set_error(&e.to_string()); use diesel::result::DatabaseErrorKind; use diesel::result::Error::*; // Sometimes `debug_query(..)` can't be turned into a @@ -938,6 +1022,8 @@ impl Layout { )), } })?; + span.set_attr("entity_count", values.len()); + span.end(); let trace = log_query_timing(logger, &query_clone, start.elapsed(), values.len(), trace); let parent_type = filter_collection.parent_type()?.map(ColumnType::from); diff --git a/store/postgres/src/writable.rs b/store/postgres/src/writable.rs index 633688aeb1a..fd8c3851330 100644 --- a/store/postgres/src/writable.rs +++ b/store/postgres/src/writable.rs @@ -21,6 +21,7 @@ use graph::prelude::{ }; use graph::schema::{EntityKey, EntityType, InputSchema}; use graph::slog::{debug, info, warn}; +use graph::telemetry::{self, SpanGroup}; use graph::util::bounded_queue::BoundedQueue; use graph::{ cheap_clone::CheapClone, @@ -817,10 +818,31 @@ impl Request { // will poison and shut down the queue return Err(err.clone()); } + // Writing a batch is its own trace, linked to the traces + // of the blocks whose changes it contains. Nesting it under + // one of them would be wrong, since a batch may hold several + // blocks and is written after they were processed. + let _section = stopwatch.start_root_section( + "write_batch", + SpanGroup::Store, + || { + [ + ("subgraph_id", store.site.deployment.to_string().into()), + ("first_block", batch.first_block.into()), + ("last_block", batch.block_ptr.number.into()), + ("entity_count", batch.entity_count().into()), + ("dropped_links", batch.dropped_trace_contexts.into()), + ] + }, + &batch.trace_contexts, + ); let res = store .transact_block_operations(&batch, stopwatch) .await .map(|()| ExecResult::Continue); + if let Err(err) = &res { + stopwatch.set_error_with(|| err.to_string()); + } info!(store.logger, "Committed write batch"; "block_number" => batch.block_ptr.number, "block_count" => batch.block_ptr.number - batch.first_block + 1, @@ -1130,9 +1152,16 @@ impl Queue { /// enough async fn push_write( &self, - batch: Batch, + mut batch: Batch, stopwatch: &StopwatchMetrics, ) -> Result<(), StoreError> { + // Remember the block's trace so the write can link back to it. + if telemetry::group_enabled(SpanGroup::Store) + && let Some(ctx) = stopwatch.current_context() + { + batch.push_trace_contexts([ctx]); + } + let batch = if ENV_VARS.store.write_batch_size == 0 || ENV_VARS.store.write_batch_duration.is_zero() || !self.batch_writes() @@ -1424,7 +1453,7 @@ impl Queue { fn deployment_synced(&self) { self.stop_batching(); - self.stopwatch.disable() + self.stopwatch.stop_recording() } fn batch_writes(&self) -> bool { diff --git a/store/test-store/tests/graphql/query.rs b/store/test-store/tests/graphql/query.rs index d2410d62b2e..effd722ab00 100644 --- a/store/test-store/tests/graphql/query.rs +++ b/store/test-store/tests/graphql/query.rs @@ -743,6 +743,62 @@ where }) } +/// The store starts its query spans under the context attached to the +/// thread it runs on, and query execution attaches the request's context +/// on the blocking thread that runs the query. Guards that hop. +#[test] +fn query_spans_join_the_request_trace() { + use graph::telemetry::testing::install_with; + use graph::telemetry::{Span, SpanGroup, SpanGroups}; + + let spans = install_with(SpanGroups::NONE.with(SpanGroup::Store)); + + run_test_sequentially(move |store| async move { + let deployment = setup_readonly(store.as_ref()).await; + + // What the HTTP server does for a traced request. The query is + // not one other tests run, so the result is not cached. + let root = Span::root("graphql_query"); + let ctx = root.context().expect("the exporter is installed"); + let query = q::parse_query("{ musicians(first: 3, orderBy: name) { id name } }") + .unwrap() + .into_static(); + let query = Query::new(query, None, false).with_trace_context(Some(ctx.clone())); + let runner = Arc::new(GraphQlRunner::new( + &LOGGER, + STORE.clone(), + LOAD_MANAGER.clone(), + METRICS_REGISTRY.clone(), + Arc::new(graph::components::log_store::NoOpLogStore), + )); + let target = QueryTarget::Deployment(deployment.hash.clone(), Default::default()); + let result = runner + .run_query_with_complexity(query, target, None, None, None, None) + .await; + assert!(!result.has_errors(), "{:?}", result.errors()); + root.end(); + + let trace = spans.trace(&ctx.trace_id()); + let root = trace + .iter() + .find(|span| span.name == "graphql_query") + .expect("the root span was exported"); + let queries: Vec<_> = trace + .iter() + .filter(|span| span.name == "db_query") + .collect(); + assert!(!queries.is_empty(), "no db_query span in {trace:?}"); + for query in queries { + assert_eq!(query.parent_span_id.as_deref(), Some(root.span_id.as_str())); + assert_eq!( + query.attr("subgraph_id"), + Some(deployment.hash.to_string().as_str()) + ); + assert!(query.attr("entity_count").is_some()); + } + }); +} + #[test] fn can_query_one_to_one_relationship() { const QUERY: &str = " diff --git a/store/test-store/tests/postgres/writable.rs b/store/test-store/tests/postgres/writable.rs index 7c17e8f201f..2ba67d7783b 100644 --- a/store/test-store/tests/postgres/writable.rs +++ b/store/test-store/tests/postgres/writable.rs @@ -526,3 +526,217 @@ fn read_range_pool_created_test() { } }) } + +/// Transact `ops` at `block` through `writable` with `stopwatch`, the way +/// the subgraph runner does, so that the block's trace is the one the +/// write gets linked to. +async fn transact_with_stopwatch( + writable: &Arc, + stopwatch: &StopwatchMetrics, + block: u8, + ops: Vec, +) { + let block_ptr = block_pointer(block); + let mut entity_cache = EntityCache::new( + Arc::new(writable.clone()), + SeqGenerator::new(block_ptr.number), + ); + entity_cache.append(ops); + let mods = entity_cache + .as_modifications(block_ptr.number, stopwatch) + .await + .expect("failed to convert to modifications") + .modifications; + let block_time = graph::blockchain::BlockTime::for_test(&block_ptr); + writable + .transact_block_operations( + block_ptr, + block_time, + FirehoseCursor::None, + mods, + stopwatch, + Vec::new(), + Vec::new(), + Vec::new(), + false, + false, + ) + .await + .unwrap(); +} + +// The tracer provider is process-global. The tests in this binary +// serialise their database access with an in-process lock, so a test +// cannot re-execute itself in a subprocess the way the runner tests do; it +// would race every other test for the database. The exporter is installed +// in-process instead, and the tracing tests only look at the spans of the +// traces they created themselves. + +/// The groups the tracing tests in this binary need. The first call to +/// `install_with` in the process decides, so both tests ask for the same. +fn tracing_groups() -> graph::telemetry::SpanGroups { + use graph::telemetry::{SpanGroup, SpanGroups}; + SpanGroups::NONE + .with(SpanGroup::Block) + .with(SpanGroup::Store) +} + +#[test] +fn write_batch_links_block_traces() { + use graph::telemetry::SpanGroup; + use graph::telemetry::testing::install_with; + + let spans = install_with(tracing_groups()); + + run_test(move |_, writable, _, deployment| async move { + let stopwatch = StopwatchMetrics::new( + LOGGER.clone(), + deployment.hash.clone(), + "process", + Arc::new(MetricsRegistry::mock()), + "primary".to_string(), + ); + + // Two blocks, each its own trace, as in the runner. + let mut block_traces = Vec::new(); + for block in 1..=2u8 { + let section = stopwatch.start_root_section( + "process_block", + SpanGroup::Block, + || [("block_number", (block as i32).into())], + &[], + ); + block_traces.push(stopwatch.current_context().unwrap().trace_id()); + let data = entity! { TEST_SUBGRAPH_SCHEMA => + id: &block.to_string(), + count: block as i32, + vid: block as i64, + }; + let key = COUNTER2_TYPE.parse_key(block.to_string()).unwrap(); + transact_with_stopwatch( + &writable, + &stopwatch, + block, + vec![EntityOperation::Set { key, data }], + ) + .await; + section.end(); + } + writable.flush().await.unwrap(); + + // The writes may have landed in one batch or two. Either way, every + // write links only to our blocks, and between them they link to + // both. + let writes: Vec<_> = spans + .spans() + .into_iter() + .filter(|span| { + span.name == "write_batch" + && span + .links + .iter() + .any(|(trace_id, _)| block_traces.contains(trace_id)) + }) + .collect(); + assert!( + !writes.is_empty(), + "no write_batch span linked to the blocks" + ); + let mut linked: Vec<&str> = writes + .iter() + .flat_map(|span| span.links.iter().map(|(trace_id, _)| trace_id.as_str())) + .collect(); + linked.sort(); + linked.dedup(); + let mut expected: Vec<&str> = block_traces.iter().map(String::as_str).collect(); + expected.sort(); + assert_eq!(linked, expected); + + for write in &writes { + assert_eq!(write.parent_span_id, None, "write_batch is a root"); + assert!( + !block_traces.contains(&write.trace_id), + "the write is its own trace" + ); + assert!(!write.is_error); + assert_eq!( + write.attr("subgraph_id"), + Some(deployment.hash.to_string().as_str()) + ); + assert!(write.attr("first_block").is_some()); + assert!(write.attr("last_block").is_some()); + assert!(write.attr("entity_count").is_some()); + let children: Vec = spans + .children_of(write) + .into_iter() + .map(|span| span.name) + .collect(); + assert!( + children + .iter() + .any(|name| name == "apply_entity_modifications"), + "store sections nest under write_batch: {children:?}" + ); + } + }); +} + +#[test] +fn db_query_spans_join_the_attached_trace() { + use graph::prelude::{AttributeNames, BLOCK_NUMBER_MAX, EntityCollection, EntityQuery}; + use graph::telemetry::Span; + use graph::telemetry::testing::install_with; + + let spans = install_with(tracing_groups()); + + run_test(move |store, _, _, deployment| async move { + let query = || { + EntityQuery::new( + deployment.hash.clone(), + BLOCK_NUMBER_MAX, + EntityCollection::All(vec![(COUNTER_TYPE.clone(), AttributeNames::All)]), + ) + }; + let subgraph_store = store.subgraph_store(); + + // With a trace attached to the thread that runs the query, the way + // GraphQL execution does it, the query becomes a child span. + let mut root = Span::root("graphql_query"); + root.set_attr("test", "db_query_spans_join_the_attached_trace"); + let ctx = root.context().unwrap(); + let (store_clone, query_clone) = (subgraph_store.clone(), query()); + graph::spawn_blocking_allow_panic(move || { + let _guard = ctx.attach(); + graph::block_on(store_clone.find(query_clone)) + }) + .await + .unwrap() + .unwrap(); + root.end(); + + let roots = spans.spans_with_attr("test", "db_query_spans_join_the_attached_trace"); + assert_eq!(roots.len(), 1); + let queries = spans.children_of(&roots[0]); + assert_eq!(queries.len(), 1, "one db_query under the request"); + let db_query = &queries[0]; + assert_eq!(db_query.name, "db_query"); + assert_eq!( + db_query.attr("subgraph_id"), + Some(deployment.hash.to_string().as_str()) + ); + assert_eq!(db_query.attr("entity_count"), Some("0")); + assert!(!db_query.is_error); + + // Without an attached trace the query joins nothing: the request + // still has its one child, and no `db_query` is a root. + subgraph_store.find(query()).await.unwrap(); + assert_eq!(spans.children_of(&roots[0]).len(), 1); + assert!( + spans + .spans() + .iter() + .all(|span| span.name != "db_query" || span.parent_span_id.is_some()), + "a db_query span without a parent was exported" + ); + }); +} diff --git a/tests/tests/runner_tests.rs b/tests/tests/runner_tests.rs index 1d44657c441..2b7506c4cdc 100644 --- a/tests/tests/runner_tests.rs +++ b/tests/tests/runner_tests.rs @@ -846,6 +846,153 @@ async fn file_data_sources() { } } +#[graph::test] +async fn block_spans() { + use graph::telemetry::testing::{in_child_process, install_with, run_in_subprocess}; + use graph::telemetry::{SpanGroup, SpanGroups}; + + // The tracer provider is process-global and would stay on for every + // other test in this binary, so this test runs in a process of its own. + // That also keeps `block_handlers`, which deploys the same subgraph + // and so shares its deployment hash, out of the exported spans. + if !in_child_process() { + return run_in_subprocess("block_spans", &[]); + } + let spans = install_with( + SpanGroups::NONE + .with(SpanGroup::Block) + .with(SpanGroup::Handler) + .with(SpanGroup::HostFn), + ); + + let RunnerTestRecipe { stores, test_info } = + RunnerTestRecipe::new("block_spans", "block-handlers").await; + + let blocks: Vec> = { + let block_0 = genesis(); + let block_1_to_3 = generate_empty_blocks_for_range(block_0.ptr(), 1, 3, 0); + let block_4 = { + let mut block = empty_block(block_1_to_3.last().unwrap().ptr(), test_ptr(4)); + push_test_polling_trigger(&mut block); + block + }; + vec![block_0] + .into_iter() + .chain(block_1_to_3) + .chain(vec![block_4]) + .collect() + }; + + let hashes: HashMap = blocks + .iter() + .map(|block| (block.ptr().number, block.ptr().hash_hex())) + .collect(); + + let stop_block = test_ptr(4); + let chain = chain(&test_info.test_name, blocks, &stores, None).await; + let ctx = fixture::setup(&test_info, &stores, &chain, None, None).await; + ctx.start_and_sync_to(stop_block).await; + + let roots = spans.spans_with_attr("subgraph_id", &test_info.hash.to_string()); + assert!(!roots.is_empty(), "no block spans were exported"); + assert!(roots.iter().all(|span| span.name == "process_block")); + assert!(roots.iter().all(|span| span.parent_span_id.is_none())); + assert!(roots.iter().all(|span| !span.is_error)); + + // Exactly one trace per processed block, with the block's identity on it. + let mut block_numbers: Vec = roots + .iter() + .map(|span| span.attr("block_number").unwrap().parse().unwrap()) + .collect(); + block_numbers.sort(); + assert_eq!(block_numbers, (0..=4).collect::>()); + for root in &roots { + let number: i32 = root.attr("block_number").unwrap().parse().unwrap(); + assert_eq!(root.attr("block_hash"), Some(hashes[&number].as_str())); + assert_eq!(root.attr("network"), Some("test")); + assert_eq!(root.attr("shard"), Some("primary")); + } + + // The block with the polling trigger ran a handler; every block was + // written to the store. + let block_4 = roots + .iter() + .find(|span| span.attr("block_number") == Some("4")) + .unwrap(); + let children = spans.children_of(block_4); + let names: Vec<&str> = children.iter().map(|span| span.name.as_str()).collect(); + for expected in [ + "process_triggers", + "handle_new_data_sources", + "update_proof_of_indexing", + "as_modifications", + "transact_block", + ] { + assert!( + names.contains(&expected), + "no {expected} under block 4: {names:?}" + ); + } + let transact = children + .iter() + .find(|span| span.name == "transact_block") + .unwrap_or_else(|| panic!("no transact_block span under block 4: {names:?}")); + assert!(transact.attr("entity_count").is_some()); + let triggers = children + .iter() + .find(|span| span.name == "process_triggers") + .unwrap(); + let handlers: Vec<_> = spans + .children_of(triggers) + .into_iter() + .filter(|span| span.name == "run_handler") + .collect(); + assert!(!handlers.is_empty(), "no handler ran for block 4"); + for handler in &handlers { + assert_eq!(handler.attr("handler"), Some("handleBlockPolling")); + assert_eq!(handler.attr("trigger"), Some("onchain")); + // A block handler has no transaction. + assert_eq!(handler.attr("tx_hash"), None); + + // The handler saves one entity; the host function span names its + // type. + let store_sets: Vec<_> = spans + .children_of(handler) + .into_iter() + .filter(|span| span.name == "host_export_store_set") + .collect(); + assert_eq!( + store_sets.len(), + 1, + "expected one store.set per handler run" + ); + assert_eq!( + store_sets[0].attr("entity_type"), + Some("BlockFromPollingHandler") + ); + assert_eq!(store_sets[0].attr("host_function"), Some("store.set")); + + // Every other host function call is a `host_export_other` span + // that names the import it was made for: the handler logs once + // and converts the block number to a string twice. + let mut others: Vec<_> = spans + .children_of(handler) + .into_iter() + .filter(|span| span.name == "host_export_other") + .map(|span| span.attr("host_function").unwrap_or("").to_owned()) + .collect(); + others.sort(); + assert_eq!( + others, + [ + "log.log", + "typeConversion.bigIntToString", + "typeConversion.bigIntToString" + ] + ); + } +} + #[graph::test] async fn block_handlers() { let RunnerTestRecipe { stores, test_info } =