Adding instrumentation/tracing points on main calls

crates/contextforge-data-plane-cpex/src/cmf.rs

@@ -8,7 +8,7 @@ use cpex::cpex_core::{
 };
 use rmcp::{ErrorData, model::ErrorCode};
 
-#[derive(Clone, Copy)]
+#[derive(Clone, Debug, Copy)]
 pub(crate) enum Operation {
     Tool,
     Prompt,

crates/contextforge-data-plane-cpex/src/resources/mod.rs

@@ -128,7 +128,7 @@ impl GatewayPluginRuntimeHandle {
             .current()?
             .before(
                 Operation::Resource,
-                "",
+                resource_uri,
                 |id| resource_request_payload(resource_uri, id),
                 |payload, _| {
                     let [ContentPart::ResourceRef { content }] = payload.message.content.as_slice() else {

crates/contextforge-data-plane-cpex/src/runtime.rs

@@ -14,6 +14,7 @@ use cpex::cpex_core::{
     manager::PluginManager,
 };
 use rmcp::ErrorData;
+use tracing::instrument;
 
 use crate::{
     cmf::{CmfResponse, Operation, modified_message_payload, plugin_denied_error},
@@ -61,6 +62,7 @@ impl GatewayPluginRuntime {
         Ok(Self { manager, hooks })
     }
 
+    #[instrument(name = "cmf_plugin_before", level = "info", skip(self, payload, update))]
     pub(crate) async fn before<U: Default>(
         self: &Arc<Self>,
         operation: Operation,
@@ -92,6 +94,7 @@ impl GatewayPluginRuntime {
         Ok((update, state))
     }
 
+    #[instrument(name = "cmf_plugin_invoke", level = "info", skip(self, payload, context_table))]
     async fn invoke(
         &self,
         hook: &'static str,

crates/contextforge-data-plane-lib/src/authorization/jwks/jwks_authorization.rs

@@ -6,6 +6,7 @@ use std::fmt;
 use std::net::IpAddr;
 use std::path::{Path, PathBuf};
 use std::time::Duration;
+use tracing::instrument;
 use url::Url;
 
 const JWKS_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
@@ -50,6 +51,7 @@ impl fmt::Debug for JwtAuthorizationService {
 
 #[async_trait]
 impl AuthorizationService for JwtAuthorizationService {
+    #[instrument(name = "jwt_authorization_service", level = "info", skip_all)]
     async fn authorize(&self, authorization_token: &http::HeaderValue) -> Option<AuthorizationClaims> {
         let token = authorization_token.as_bytes().strip_prefix(b"Bearer ")?;
         let token = str::from_utf8(token).ok()?;

crates/contextforge-data-plane-lib/src/authorization/principal_extractor/cel_principal_extractor.rs

@@ -5,7 +5,7 @@ use std::sync::Arc;
 use cel::{Context, Program, objects::Key};
 use serde_json::Value as JsonValue;
 use thiserror::Error;
-use tracing::debug;
+use tracing::{debug, instrument};
 
 use crate::authorization::{AuthorizedPrincipal, PrincipalExtractor};
 
@@ -51,7 +51,7 @@ pub enum CelPrincipalExtractorError {
 ///   "scopes": []
 /// }
 /// ```
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct CelPrincipalExtractor {
     program: Arc<Program>,
 }
@@ -71,6 +71,7 @@ impl CelPrincipalExtractor {
 }
 
 impl PrincipalExtractor for CelPrincipalExtractor {
+    #[instrument(name = "principal_extract", level = "info", skip_all)]
     fn extract(
         &self,
         claims: &serde_json::Value,

crates/contextforge-data-plane-lib/src/authorization/principal_extractor/default_principal_extractor.rs

@@ -1,9 +1,12 @@
+use tracing::instrument;
+
 use crate::authorization::{AuthorizedPrincipal, PrincipalExtractor};
 
 #[derive(Debug, Clone)]
 pub struct DefaultPrincipalExtractor {}
 
 impl PrincipalExtractor for DefaultPrincipalExtractor {
+    #[instrument(name = "principal_extract", level = "info", skip_all)]
     fn extract(
         &self,
         claims: &serde_json::Value,

crates/contextforge-data-plane-lib/src/gateway/mcp_service/mod.rs

@@ -24,6 +24,12 @@ pub struct McpService {
     plugin_runtime: Option<GatewayPluginRuntimeHandle>,
 }
 
+#[allow(clippy::missing_fields_in_debug)]
+impl std::fmt::Debug for McpService {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("McpService").field("http_client", &self.http_client).finish()
+    }
+}
 #[allow(clippy::unused_async_trait_impl)]
 impl ServerHandler for McpService {
     async fn initialize(

crates/contextforge-data-plane-lib/src/gateway/mcp_service/prompts.rs

@@ -4,14 +4,15 @@ use rmcp::{
     model::{ErrorCode, GetPromptRequestParams, GetPromptResponse},
     service::RequestContext,
 };
-use tracing::info;
+use tracing::{info, instrument};
 
 use super::McpService;
 use crate::gateway::{
     mcp_call_validator::AuthorizedCallValidator, mcp_service::initialization::connect_backend_for_request,
     routing_error::backend_forward_error,
 };
 
+#[instrument(name = "get_prompt", level = "info", skip_all)]
 pub(super) async fn get_prompt(
     mcp_service: &McpService,
     request: GetPromptRequestParams,

crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs

@@ -5,7 +5,7 @@ use rmcp::{
     model::{CallToolRequestParams, CallToolResponse, ErrorCode, ProtocolVersion},
     service::RequestContext,
 };
-use tracing::{info, warn};
+use tracing::{info, instrument, warn};
 
 use super::McpService;
 use crate::gateway::{
@@ -14,6 +14,7 @@ use crate::gateway::{
 };
 use crate::mcp_standard_headers;
 
+#[instrument(name = "call_tool", level = "info", skip_all)]
 pub(super) async fn call_tool(
     mcp_service: &McpService,
     request: CallToolRequestParams,

crates/contextforge-data-plane-lib/src/layers/principal_extractor.rs

@@ -5,19 +5,6 @@ use tracing::debug;
 
 use crate::{AuthorizationClaims, authorization::PrincipalExtractor, errors::unauthorized_response};
 
-// pub async fn principal_extractor_layer(request: http::Request<axum::body::Body>, next: Next) -> Response {
-//     let maybe_claims = request.extensions().get::<AuthorizationClaims>();
-//     let Some(Ok(authorized_principal)) = maybe_claims.map(|claims| {
-//         AuthorizedPrincipal::try_from(claims).inspect_err(|e| debug!("Can't extract the principal {e:?}"))
-//     }) else {
-//         return unauthorized_response("Invalid token. Unable to extract the principal from claims");
-//     };
-//     let (mut parts, body) = request.into_parts();
-//     parts.extensions.insert(authorized_principal);
-//     let request = Request::from_parts(parts, body);
-//     next.run(request).await
-// }
-
 use std::task::{Context, Poll};
 use tower::{Layer, Service};
 

crates/contextforge-data-plane-lib/src/user_config_store/redis_config_store.rs

@@ -12,7 +12,7 @@ use redis::{
     cmd,
 };
 use tokio::sync::Mutex;
-use tracing::{debug, warn};
+use tracing::{debug, instrument, warn};
 
 use super::{ConfigStoreError, UserConfigStore};
 use crate::{
@@ -67,6 +67,7 @@ impl RedisUserConfigStore {
 
 #[async_trait]
 impl UserConfigStore for RedisUserConfigStore {
+    #[instrument(name = "user_config_store_get_config", level = "info", skip(self))]
     async fn get_config<'a>(&self, user_key: &'a User) -> Result<UserConfig, ConfigStoreError> {
         let subject = user_key.key();
 
@@ -136,6 +137,7 @@ impl UserConfigStore for RedisUserConfigStore {
         Ok(user_config)
     }
 
+    #[instrument(name = "user_config_store_set_config", level = "info", skip(self, config))]
     async fn set_config<'a>(&self, user_key: &'a User, config: &'a UserConfig) -> Result<(), ConfigStoreError> {
         let subject = user_key.key();
 

crates/contextforge-data-plane/Cargo.toml

@@ -21,15 +21,13 @@ cpex-tool-namespace = { git = "https://github.com/contextforge-gateway-rs/cpex-p
 cpex-secrets-detection = { workspace = true, optional = true }
 clap.workspace = true
 tracing.workspace = true
-tracing-appender = "0.2.3"
 tracing-opentelemetry.workspace = true
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 tokio.workspace = true
 opentelemetry.workspace = true
-opentelemetry-otlp = { version = "0.32", features = ["grpc-tonic", "http-proto", "reqwest-blocking-client", "metrics"] }
+opentelemetry-otlp = { version = "0.32", features = ["grpc-tonic", "http-proto", "reqwest-blocking-client", "metrics","tls-aws-lc"] }
 opentelemetry_sdk.workspace = true
 tonic = "0.14"
-num_cpus = "1.17.0"
 rmcp.workspace = true
 tikv-jemallocator = "0.7.0"
 rustls.workspace = true