Add OTLP-only observability controls

Support disabling stdout logging while keeping OTLP export active, and document the TPM cert reissue behavior in the sidecar connection flow.

Author: sl5-guy

Cargo.lock

@@ -32,7 +32,7 @@ Typical first-time flow:
 2. `./examples/connect.sh --gateway 127.0.0.1:8443 --gateway-ca certs/server-ca.pem` creates `machine-client-ca.pem` under `~/.local/share/agent-gateway/` (or `$XDG_DATA_HOME`). Append that file to `client_ca_path` (for example, `cat ~/.local/share/agent-gateway/machine-client-ca.pem >> certs/client-ca-bundle.pem`).
 3. Start the gateway (`cargo run -- --config config.toml`), then return to the terminal running `connect.sh` and press Enter to start the sidecar and Claude.
 
-On later runs, start the gateway first, run `connect.sh`, and press Enter after confirming the machine CA is still registered. `--regenerate-certs` creates a fresh simulated TPM state and machine client CA, so the new CA must be appended to `client_ca_path`.
+On later runs, start the gateway first, run `connect.sh`, and press Enter after confirming the machine CA is still registered. If `connect.sh` finds an existing simulated TPM key but the saved `machine-client.pem` was issued for a different public key, it reissues `machine-client.pem` for the current TPM key using the existing machine client CA. `--regenerate-certs` creates a fresh simulated TPM state and machine client CA, so the new CA must be appended to `client_ca_path`.
 
 Pass a custom policy extension value: `connect.sh ... --extension-value agent-beta`.
 
@@ -71,6 +71,14 @@ Copy `config.example.toml` to `config.toml` and edit it. Key sections:
 | `log_level` | yes | `tracing` filter (e.g. `info`, `debug`, `agent_gateway=debug`) |
 | `otlp_endpoint` | no | OTLP gRPC endpoint for distributed tracing |
 
+Set `AGENT_GATEWAY_LOG_STDOUT=false` to disable stdout/stderr formatting while
+leaving OTLP export enabled. The sidecar reads observability settings from
+environment variables. Use `RUST_LOG` for its tracing filter and set
+`OTEL_EXPORTER_OTLP_ENDPOINT` (for example, `http://localhost:4317`) to export
+sidecar spans over OTLP. When enabled, the sidecar injects W3C trace-context
+headers into the CONNECT request it sends to the gateway, and the gateway
+continues the same trace.
+
 ## Running
 
 ```

README.md

@@ -179,12 +179,17 @@ ensure_tpm_key() {
     rm -f "$CERT_DIR/tpm-signing-key.ctx"
 }
 
-verify_cert_matches_tpm_key() {
+cert_matches_tpm_key() {
+    [[ -f "$CERT_DIR/machine-client.pem" ]] || return 1
     openssl pkey -pubin -in "$CERT_DIR/machine-client-public.pem" -outform DER \
         > "$CERT_DIR/machine-client-public.der"
     openssl x509 -in "$CERT_DIR/machine-client.pem" -pubkey -noout \
         | openssl pkey -pubin -outform DER > "$CERT_DIR/machine-client-cert-public.der"
-    if ! cmp -s "$CERT_DIR/machine-client-public.der" "$CERT_DIR/machine-client-cert-public.der"; then
+    cmp -s "$CERT_DIR/machine-client-public.der" "$CERT_DIR/machine-client-cert-public.der"
+}
+
+verify_cert_matches_tpm_key() {
+    if ! cert_matches_tpm_key; then
         echo "error: machine-client.pem public key does not match simulated TPM key $TPM_HANDLE" >&2
         exit 1
     fi
@@ -195,6 +200,9 @@ ensure_tpm_key
 
 if needs_certs; then
     generate_certs
+elif ! cert_matches_tpm_key; then
+    echo "==> Reissuing machine client certificate for simulated TPM key $TPM_HANDLE"
+    generate_certs
 fi
 verify_cert_matches_tpm_key
 

examples/connect.sh

@@ -20,4 +20,8 @@ bytes = "1"
 clap = { version = "4", features = ["derive"] }
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+tracing-opentelemetry = "0.32"
+opentelemetry = "0.31"
+opentelemetry_sdk = { version = "0.31", features = ["rt-tokio"] }
+opentelemetry-otlp = { version = "0.31", features = ["grpc-tonic"] }
 anyhow = "1"

sidecar/Cargo.toml

@@ -6,19 +6,23 @@ use std::sync::Arc;
 
 use anyhow::Context;
 use bytes::Bytes;
+use http::{HeaderMap, HeaderName, HeaderValue};
 use http_body_util::combinators::BoxBody;
 use http_body_util::{BodyExt, Empty, Full};
 use hyper::body::Incoming;
 use hyper::client::conn::http2 as h2_client;
 use hyper::service::Service;
 use hyper::{Method, Request, Response, StatusCode};
 use hyper_util::rt::{TokioExecutor, TokioIo};
+use opentelemetry::global;
+use opentelemetry::propagation::Injector;
 use rustls::ClientConfig;
 use rustls_pki_types::ServerName;
 use tokio::io::copy_bidirectional;
 use tokio::net::TcpStream;
 use tokio::sync::Mutex;
-use tracing::{error, info, warn};
+use tracing::{Instrument, error, info, warn};
+use tracing_opentelemetry::OpenTelemetrySpanExt;
 
 type Body = BoxBody<Bytes, Infallible>;
 
@@ -198,14 +202,34 @@ fn build_connect_request(authority: &str) -> anyhow::Result<Request<Body>> {
         .build()
         .context("building CONNECT URI")?;
 
-    let req = Request::builder()
+    let mut req = Request::builder()
         .method(Method::CONNECT)
         .uri(uri)
         .body(Empty::<Bytes>::new().boxed())
         .context("building CONNECT request")?;
+    inject_current_trace_context(req.headers_mut());
     Ok(req)
 }
 
+struct HeaderInjector<'a>(&'a mut HeaderMap);
+
+impl Injector for HeaderInjector<'_> {
+    fn set(&mut self, key: &str, value: String) {
+        if let Ok(name) = HeaderName::from_bytes(key.as_bytes())
+            && let Ok(value) = HeaderValue::from_str(&value)
+        {
+            self.0.insert(name, value);
+        }
+    }
+}
+
+fn inject_current_trace_context(headers: &mut HeaderMap) {
+    let context = tracing::Span::current().context();
+    global::get_text_map_propagator(|propagator| {
+        propagator.inject_context(&context, &mut HeaderInjector(headers))
+    });
+}
+
 #[derive(Clone)]
 struct SidecarService {
     connector: Arc<GatewayConnector>,
@@ -238,106 +262,122 @@ async fn handle(
         None => return response(StatusCode::BAD_REQUEST, "missing authority"),
     };
 
-    info!(
+    let span = tracing::info_span!(
+        "sidecar CONNECT",
         source_identity = %connector.source_identity(),
         source_peer_addr = %source_peer_addr,
         dest_authority = %authority,
         gateway_endpoint = %connector.gateway_endpoint(),
-        "CONNECT request"
     );
 
-    let gw_response = match connector.send_connect(&authority, source_peer_addr).await {
-        Ok(r) => r,
-        Err(e) => {
-            error!(
-                source_identity = %connector.source_identity(),
-                source_peer_addr = %source_peer_addr,
-                dest_authority = %authority,
-                gateway_endpoint = %connector.gateway_endpoint(),
-                error = %e,
-                "gateway connect failed"
-            );
-            return response(StatusCode::BAD_GATEWAY, "gateway unreachable");
-        }
-    };
-
-    let status = gw_response.status();
-    if status != StatusCode::OK {
-        warn!(
+    async move {
+        info!(
             source_identity = %connector.source_identity(),
             source_peer_addr = %source_peer_addr,
             dest_authority = %authority,
             gateway_endpoint = %connector.gateway_endpoint(),
-            %status,
-            "gateway rejected CONNECT"
+            "CONNECT request"
         );
-        return response(status, "gateway denied request");
-    }
 
-    let gw_upgraded = match hyper::upgrade::on(gw_response).await {
-        Ok(u) => u,
-        Err(e) => {
-            error!(
+        let gw_response = match connector.send_connect(&authority, source_peer_addr).await {
+            Ok(r) => r,
+            Err(e) => {
+                error!(
+                    source_identity = %connector.source_identity(),
+                    source_peer_addr = %source_peer_addr,
+                    dest_authority = %authority,
+                    gateway_endpoint = %connector.gateway_endpoint(),
+                    error = %e,
+                    "gateway connect failed"
+                );
+                return response(StatusCode::BAD_GATEWAY, "gateway unreachable");
+            }
+        };
+
+        let status = gw_response.status();
+        if status != StatusCode::OK {
+            warn!(
                 source_identity = %connector.source_identity(),
                 source_peer_addr = %source_peer_addr,
                 dest_authority = %authority,
                 gateway_endpoint = %connector.gateway_endpoint(),
-                error = %e,
-                "gateway upgrade failed"
+                %status,
+                "gateway rejected CONNECT"
             );
-            return response(StatusCode::BAD_GATEWAY, "gateway tunnel failed");
+            return response(status, "gateway denied request");
         }
-    };
-
-    let client_upgrade = hyper::upgrade::on(req);
-    let source_identity = connector.source_identity().to_owned();
-    let gateway_endpoint = connector.gateway_endpoint().to_owned();
 
-    tokio::spawn(async move {
-        let client_upgraded = match client_upgrade.await {
+        let gw_upgraded = match hyper::upgrade::on(gw_response).await {
             Ok(u) => u,
             Err(e) => {
-                warn!(
-                    source_identity = %source_identity,
+                error!(
+                    source_identity = %connector.source_identity(),
                     source_peer_addr = %source_peer_addr,
                     dest_authority = %authority,
-                    gateway_endpoint = %gateway_endpoint,
+                    gateway_endpoint = %connector.gateway_endpoint(),
                     error = %e,
-                    "client upgrade failed"
+                    "gateway upgrade failed"
                 );
-                return;
+                return response(StatusCode::BAD_GATEWAY, "gateway tunnel failed");
             }
         };
 
-        let mut client_io = TokioIo::new(client_upgraded);
-        let mut gw_io = TokioIo::new(gw_upgraded);
+        let client_upgrade = hyper::upgrade::on(req);
+        let source_identity = connector.source_identity().to_owned();
+        let gateway_endpoint = connector.gateway_endpoint().to_owned();
 
-        match copy_bidirectional(&mut client_io, &mut gw_io).await {
-            Ok((up, down)) => {
-                info!(
-                    source_identity = %source_identity,
-                    source_peer_addr = %source_peer_addr,
-                    dest_authority = %authority,
-                    gateway_endpoint = %gateway_endpoint,
-                    bytes_client_to_dest = up,
-                    bytes_dest_to_client = down,
-                    "tunnel closed"
-                );
-            }
-            Err(e) => {
-                error!(
-                    source_identity = %source_identity,
-                    source_peer_addr = %source_peer_addr,
-                    dest_authority = %authority,
-                    gateway_endpoint = %gateway_endpoint,
-                    error = %e,
-                    "tunnel error"
-                );
+        let tunnel_span = tracing::Span::current();
+        tokio::spawn(
+            async move {
+                let client_upgraded = match client_upgrade.await {
+                    Ok(u) => u,
+                    Err(e) => {
+                        warn!(
+                            source_identity = %source_identity,
+                            source_peer_addr = %source_peer_addr,
+                            dest_authority = %authority,
+                            gateway_endpoint = %gateway_endpoint,
+                            error = %e,
+                            "client upgrade failed"
+                        );
+                        return;
+                    }
+                };
+
+                let mut client_io = TokioIo::new(client_upgraded);
+                let mut gw_io = TokioIo::new(gw_upgraded);
+
+                match copy_bidirectional(&mut client_io, &mut gw_io).await {
+                    Ok((up, down)) => {
+                        info!(
+                            source_identity = %source_identity,
+                            source_peer_addr = %source_peer_addr,
+                            dest_authority = %authority,
+                            gateway_endpoint = %gateway_endpoint,
+                            bytes_client_to_dest = up,
+                            bytes_dest_to_client = down,
+                            "tunnel closed"
+                        );
+                    }
+                    Err(e) => {
+                        error!(
+                            source_identity = %source_identity,
+                            source_peer_addr = %source_peer_addr,
+                            dest_authority = %authority,
+                            gateway_endpoint = %gateway_endpoint,
+                            error = %e,
+                            "tunnel error"
+                        );
+                    }
+                }
             }
-        }
-    });
+            .instrument(tunnel_span),
+        );
 
-    response(StatusCode::OK, "")
+        response(StatusCode::OK, "")
+    }
+    .instrument(span)
+    .await
 }
 
 fn response(status: StatusCode, message: &str) -> Response<Body> {
@@ -478,6 +518,38 @@ mod tests {
         assert_eq!(req.uri().authority().unwrap().as_str(), "[::1]:8443");
     }
 
+    #[test]
+    fn connect_request_injects_trace_context_headers() {
+        use opentelemetry::trace::TracerProvider as _;
+        use opentelemetry_sdk::propagation::TraceContextPropagator;
+        use opentelemetry_sdk::trace::SdkTracerProvider;
+        use tracing_subscriber::layer::SubscriberExt;
+
+        opentelemetry::global::set_text_map_propagator(TraceContextPropagator::new());
+
+        let provider = SdkTracerProvider::builder().build();
+        let tracer = provider.tracer("test");
+        let subscriber =
+            tracing_subscriber::registry().with(tracing_opentelemetry::layer().with_tracer(tracer));
+
+        tracing::subscriber::with_default(subscriber, || {
+            let span = tracing::info_span!("sidecar test span");
+            let _guard = span.enter();
+
+            let req = build_connect_request("example.com:443").unwrap();
+            let traceparent = req
+                .headers()
+                .get("traceparent")
+                .and_then(|value| value.to_str().ok())
+                .expect("traceparent should be injected");
+
+            assert!(traceparent.starts_with("00-"));
+            assert_eq!(traceparent.len(), 55);
+        });
+
+        let _ = provider.shutdown();
+    }
+
     #[test]
     fn response_ok_has_empty_body() {
         let resp = response(StatusCode::OK, "");