Adding context_id support to selectors

.changesets/feat_context_id_selector.md

@@ -0,0 +1,31 @@
+### Add `context_id` selector for telemetry to expose unique per-request identifier ([#8899](https://github.com/apollographql/router/pull/8899))
+
+A new `context_id` selector is now available for router, supergraph, subgraph, and connector telemetry instrumentation. This selector exposes the unique per-request context ID that can be used to reliably correlate and debug requests in traces, logs, and custom events.
+
+Previously, while the context ID was accessible in Rhai scripts as `request.id`, there was no telemetry selector to expose it. With this change, users can now include `context_id: true` in their telemetry configuration to add the context ID to spans, logs, and custom events.
+
+Example configuration:
+
+```yaml
+telemetry:
+  instrumentation:
+    spans:
+      router:
+        attributes:
+          "request.id":
+            context_id: true
+      supergraph:
+        attributes:
+          "request.id":
+            context_id: true
+      subgraph:
+        attributes:
+          "request.id":
+            context_id: true
+      connector:
+        attributes:
+          "request.id":
+            context_id: true
+```
+
+By [@BobaFetters](https://github.com/BobaFetters) in https://github.com/apollographql/router/pull/8899

apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__schema_generation.snap

@@ -2564,6 +2564,20 @@ expression: "&schema"
             "connector_on_response_error"
           ],
           "type": "object"
+        },
+        {
+          "additionalProperties": false,
+          "description": "The context ID of the request (unique per request).",
+          "properties": {
+            "context_id": {
+              "description": "The context ID",
+              "type": "boolean"
+            }
+          },
+          "required": [
+            "context_id"
+          ],
+          "type": "object"
         }
       ]
     },
@@ -8658,6 +8672,20 @@ expression: "&schema"
             "trace_id"
           ],
           "type": "object"
+        },
+        {
+          "additionalProperties": false,
+          "description": "The context ID of the request (unique per request).",
+          "properties": {
+            "context_id": {
+              "description": "The context ID",
+              "type": "boolean"
+            }
+          },
+          "required": [
+            "context_id"
+          ],
+          "type": "object"
         }
       ]
     },
@@ -10559,6 +10587,20 @@ expression: "&schema"
             "response_cache_control"
           ],
           "type": "object"
+        },
+        {
+          "additionalProperties": false,
+          "description": "The context ID of the request (unique per request).",
+          "properties": {
+            "context_id": {
+              "description": "The context ID",
+              "type": "boolean"
+            }
+          },
+          "required": [
+            "context_id"
+          ],
+          "type": "object"
         }
       ]
     },
@@ -11513,6 +11555,20 @@ expression: "&schema"
             "is_primary_response"
           ],
           "type": "object"
+        },
+        {
+          "additionalProperties": false,
+          "description": "The context ID of the request (unique per request).",
+          "properties": {
+            "context_id": {
+              "description": "The context ID",
+              "type": "boolean"
+            }
+          },
+          "required": [
+            "context_id"
+          ],
+          "type": "object"
         }
       ]
     },

apollo-router/src/plugin/test/mock/connector.rs

@@ -112,7 +112,13 @@ impl Service<ConnectorRequest> for MockConnector {
             let data = json!(response);
             let headers = self.headers.clone();
 
-            ConnectorResponse::test_new(response_key, Default::default(), data, Some(headers))
+            ConnectorResponse::test_new(
+                req.context.clone(),
+                response_key,
+                vec![],
+                data,
+                Some(headers),
+            )
         } else {
             let error_message = format!(
                 "couldn't find mock for query {}",
@@ -122,7 +128,13 @@ impl Service<ConnectorRequest> for MockConnector {
             let data = json!(error_message);
             let headers = self.headers.clone();
 
-            ConnectorResponse::test_new(response_key, Default::default(), data, Some(headers))
+            ConnectorResponse::test_new(
+                req.context.clone(),
+                response_key,
+                vec![],
+                data,
+                Some(headers),
+            )
         };
         future::ok(response)
     }

apollo-router/src/plugins/connectors/handle_responses.rs

@@ -199,6 +199,7 @@ pub(crate) async fn process_response<T: HttpBody>(
     }
 
     connector::request_service::Response {
+        context: context.clone(),
         transport_result: result,
         mapped_response,
     }
@@ -266,6 +267,7 @@ fn log_connectors_event(
             // connectors events.
 
             let response = connector::request_service::Response {
+                context: context.clone(),
                 transport_result: Ok(TransportResponse::Http(HttpResponse {
                     inner: parts.clone(),
                 })),

apollo-router/src/plugins/coprocessor/connector.rs

@@ -326,6 +326,7 @@ where
         }
 
         let res = request_service::Response {
+            context: request.context.clone(),
             transport_result: Err(ConnectorError::TransportFailure(message)),
             mapped_response: MappedResponse::Error {
                 error: runtime_error,

apollo-router/src/plugins/coprocessor/test.rs

@@ -5628,6 +5628,7 @@ mod tests {
                         .collect();
 
                     let response = request_service::Response::test_new(
+                        req.context.clone(),
                         req.key,
                         Default::default(),
                         serde_json_bytes::json!("ok"),
@@ -6240,8 +6241,9 @@ mod tests {
         fn create_error_connector_service()
         -> tower::util::BoxService<request_service::Request, request_service::Response, BoxError>
         {
-            tower::service_fn(|_req: request_service::Request| async {
+            tower::service_fn(|req: request_service::Request| async {
                 Ok(request_service::Response {
+                    context: req.context,
                     transport_result: Err(
                         apollo_federation::connectors::runtime::errors::Error::TransportFailure(
                             "original error".to_string(),

apollo-router/src/plugins/headers/mod.rs

@@ -1580,14 +1580,15 @@ mod test {
     }
 
     fn example_connector_response(
-        _req: connector::request_service::Request,
+        req: connector::request_service::Request,
     ) -> Result<connector::request_service::Response, BoxError> {
         let key = ResponseKey::RootField {
             name: "hello".to_string(),
             inputs: Default::default(),
             selection: Arc::new(JSONSelection::parse("$.data").unwrap()),
         };
         Ok(connector::request_service::Response::test_new(
+            req.context.clone(),
             key,
             Vec::new(),
             json!(""),

apollo-router/src/plugins/telemetry/config_new/connector/events.rs

@@ -215,6 +215,7 @@ mod tests {
             };
             test_harness
                 .call_connector_request_service(connector_request, |request| Response {
+                    context: request.context.clone(),
                     transport_result: Ok(TransportResponse::Http(HttpResponse {
                         inner: http::Response::builder()
                             .status(200)
@@ -301,6 +302,7 @@ mod tests {
             };
             test_harness
                 .call_connector_request_service(connector_request, |request| Response {
+                    context: request.context.clone(),
                     transport_result: Ok(TransportResponse::Http(HttpResponse {
                         inner: http::Response::builder()
                             .status(200)

apollo-router/src/plugins/telemetry/config_new/connector/selectors.rs

@@ -158,6 +158,11 @@ pub(crate) enum ConnectorSelector {
         /// set, returns true when the response contains a non-200 status code
         connector_on_response_error: bool,
     },
+    /// The context ID of the request (unique per request).
+    ContextId {
+        /// The context ID
+        context_id: bool,
+    },
 }
 
 impl Selector for ConnectorSelector {
@@ -256,6 +261,9 @@ impl Selector for ConnectorSelector {
                 .ok()
                 .flatten()
                 .map(opentelemetry::Value::from),
+            ConnectorSelector::ContextId { context_id } if *context_id => {
+                Some(opentelemetry::Value::from(request.context.id.clone()))
+            }
             _ => None,
         }
     }
@@ -334,14 +342,20 @@ impl Selector for ConnectorSelector {
             } if *connector_on_response_error => {
                 Some(matches!(response.mapped_response, MappedResponse::Error { .. }).into())
             }
+            ConnectorSelector::ContextId { context_id } if *context_id => {
+                Some(opentelemetry::Value::from(response.context.id.clone()))
+            }
             _ => None,
         }
     }
 
-    fn on_error(&self, error: &BoxError, _ctx: &Context) -> Option<Value> {
+    fn on_error(&self, error: &BoxError, ctx: &Context) -> Option<Value> {
         match self {
             ConnectorSelector::Error { .. } => Some(error.to_string().into()),
             ConnectorSelector::StaticField { r#static } => Some(r#static.clone().into()),
+            ConnectorSelector::ContextId { context_id } if *context_id => {
+                Some(opentelemetry::Value::from(ctx.id.clone()))
+            }
             _ => None,
         }
     }
@@ -367,6 +381,7 @@ impl Selector for ConnectorSelector {
                     | ConnectorSelector::RequestContext { .. }
                     | ConnectorSelector::SupergraphOperationName { .. }
                     | ConnectorSelector::SupergraphOperationKind { .. }
+                    | ConnectorSelector::ContextId { .. }
             ),
             Stage::Response => matches!(
                 self,
@@ -380,6 +395,7 @@ impl Selector for ConnectorSelector {
                     | ConnectorSelector::StaticField { .. }
                     | ConnectorSelector::ResponseMappingProblems { .. }
                     | ConnectorSelector::OnResponseError { .. }
+                    | ConnectorSelector::ContextId { .. }
             ),
             Stage::ResponseEvent => false,
             Stage::ResponseField => false,
@@ -391,6 +407,7 @@ impl Selector for ConnectorSelector {
                     | ConnectorSelector::ConnectorHttpMethod { .. }
                     | ConnectorSelector::ConnectorUrlTemplate { .. }
                     | ConnectorSelector::StaticField { .. }
+                    | ConnectorSelector::ContextId { .. }
             ),
             Stage::Drop => matches!(self, ConnectorSelector::StaticField { .. }),
         }
@@ -424,6 +441,7 @@ mod tests {
     use opentelemetry::Array;
     use opentelemetry::StringValue;
     use opentelemetry::Value;
+    use tower::BoxError;
 
     use super::ConnectorSelector;
     use super::ConnectorSource;
@@ -527,6 +545,7 @@ mod tests {
         mapping_problems: Vec<Problem>,
     ) -> Response {
         Response {
+            context: Context::new(),
             transport_result: Ok(TransportResponse::Http(HttpResponse {
                 inner: http::Response::builder()
                     .status(status_code)
@@ -547,6 +566,7 @@ mod tests {
 
     fn connector_response_with_mapped_error(status_code: StatusCode) -> Response {
         Response {
+            context: Context::new(),
             transport_result: Ok(TransportResponse::Http(HttpResponse {
                 inner: http::Response::builder()
                     .status(status_code)
@@ -565,6 +585,7 @@ mod tests {
 
     fn connector_response_with_header() -> Response {
         Response {
+            context: Context::new(),
             transport_result: Ok(TransportResponse::Http(HttpResponse {
                 inner: http::Response::builder()
                     .status(200)
@@ -999,4 +1020,44 @@ mod tests {
             Value::String("invalid digit found in string".into())
         );
     }
+
+    #[test]
+    fn connector_context_id() {
+        let selector = ConnectorSelector::ContextId { context_id: true };
+        let context = Context::new();
+        let expected_id: Value = context.id.clone().into();
+
+        // Test on_request
+        assert_eq!(
+            selector
+                .on_request(&connector_request(
+                    http_request(),
+                    Some(context.clone()),
+                    None
+                ))
+                .unwrap(),
+            expected_id
+        );
+
+        // Test on_response
+        let mut response = connector_response(StatusCode::OK);
+        response.context = context.clone();
+        assert_eq!(selector.on_response(&response).unwrap(), expected_id);
+
+        // Test on_error
+        assert_eq!(
+            selector
+                .on_error(&BoxError::from("test error".to_string()), &context)
+                .unwrap(),
+            expected_id
+        );
+
+        // Test that context_id: false returns None
+        let selector_disabled = ConnectorSelector::ContextId { context_id: false };
+        assert!(
+            selector_disabled
+                .on_request(&connector_request(http_request(), Some(context), None))
+                .is_none()
+        );
+    }
 }

apollo-router/src/plugins/telemetry/config_new/instruments.rs

@@ -3241,6 +3241,7 @@ mod tests {
                                         .unwrap();
                                     *http_response.headers_mut() = convert_http_headers(headers);
                                     let response = Response {
+                                        context: context.clone(),
                                         transport_result: Ok(TransportResponse::Http(
                                             HttpResponse {
                                                 inner: http_response.into_parts().0,