dekaf: Remove log forwarding from the registry

We never ended up setting a task name so no logs got sent, and it's best if we don't allow external data like from HTTP requests to get into the ops logs

Author: jshearer

crates/dekaf/src/lib.rs

@@ -243,7 +243,8 @@ impl App {
             // We only set this after successfully validating that this task exists and is a Dekaf
             // materialization. Otherwise we will either log auth errors attempting to append to
             // a journal that doesn't exist, or possibly log confusing errors to a different task's logs entirely.
-            logging::get_log_forwarder().set_task_name(username.clone(), labels.build.clone());
+            logging::get_log_forwarder()
+                .map(|f| f.set_task_name(username.clone(), labels.build.clone()));
 
             // 3. Validate that the provided password matches the task's bearer token
             if password != config.token {
@@ -261,8 +262,8 @@ impl App {
             )?))
         } else if username.contains("{") {
             // Since we don't have a task, we also don't have a logs journal to write to,
-            // so we should isable log forwarding for this session.
-            logging::get_log_forwarder().shutdown();
+            // so we should disable log forwarding for this session.
+            logging::get_log_forwarder().map(|f| f.shutdown());
 
             let raw_token = String::from_utf8(base64::decode(password)?.to_vec())?;
             let refresh: RefreshToken = serde_json::from_str(raw_token.as_str())?;

crates/dekaf/src/logging.rs

@@ -131,8 +131,8 @@ where
     )
 }
 
-pub fn get_log_forwarder() -> TaskForwarder<GazetteWriter> {
-    TASK_FORWARDER.get()
+pub fn get_log_forwarder() -> Option<TaskForwarder<GazetteWriter>> {
+    TASK_FORWARDER.try_with(|v| v.clone()).ok()
 }
 
 pub fn set_log_level(level: ops::LogLevel) {

crates/dekaf/src/read.rs

@@ -354,22 +354,24 @@ impl Read {
         // Right: Input documents from journal. Left: Input docs from destination. Out: Right Keys ⋃ Left Keys
         // Dekaf reads docs from journals, so it emits "right". It doesn't do reduction with a destination system,
         // so it does not emit "left". And right now, it does not reduce at all, so "out" is the same as "right".
-        logging::get_log_forwarder().send_stats(
-            self.collection_name.to_owned(),
-            ops::stats::Binding {
-                right: Some(ops::stats::DocsAndBytes {
-                    docs_total: stats_records,
-                    bytes_total: stats_bytes,
-                }),
-                out: Some(ops::stats::DocsAndBytes {
-                    docs_total: stats_records,
-                    bytes_total: stats_bytes,
-                }),
-                left: None,
-                last_source_published_at: last_source_published_at
-                    .and_then(|c| c.to_pb_json_timestamp()),
-            },
-        );
+        logging::get_log_forwarder().map(|f| {
+            f.send_stats(
+                self.collection_name.to_owned(),
+                ops::stats::Binding {
+                    right: Some(ops::stats::DocsAndBytes {
+                        docs_total: stats_records,
+                        bytes_total: stats_bytes,
+                    }),
+                    out: Some(ops::stats::DocsAndBytes {
+                        docs_total: stats_records,
+                        bytes_total: stats_bytes,
+                    }),
+                    left: None,
+                    last_source_published_at: last_source_published_at
+                        .and_then(|c| c.to_pb_json_timestamp()),
+                },
+            );
+        });
 
         let frozen = buf.freeze();
 

crates/dekaf/src/registry.rs

@@ -27,42 +27,12 @@ pub fn build_router(app: Arc<App>) -> axum::Router<()> {
             get(get_subject_latest),
         )
         .route("/schemas/ids/:id", get(get_schema_by_id))
-        .route_layer(axum::middleware::from_fn_with_state(
-            app.clone(),
-            attach_logger,
-        ))
         .layer(tower_http::trace::TraceLayer::new_for_http())
         .with_state(app);
 
     schema_router
 }
 
-async fn attach_logger(
-    axum::extract::State(app): axum::extract::State<Arc<App>>,
-    request: axum::extract::Request,
-    next: axum::middleware::Next,
-) -> Response {
-    let writer = GazetteWriter::new(app);
-
-    logging::forward_logs(
-        writer,
-        tokio_util::sync::CancellationToken::new(),
-        async move {
-            if let Some(user_agent) = request.headers().get(USER_AGENT) {
-                match user_agent.to_str() {
-                    Ok(user_agent) => {
-                        tracing::Span::current()
-                            .record_hierarchical(SESSION_CLIENT_ID_FIELD_MARKER, user_agent);
-                    }
-                    Err(e) => tracing::warn!(?e, "Got bad user agent header"),
-                };
-            }
-            next.run(request).await
-        },
-    )
-    .await
-}
-
 // List all collections as "subjects", which are generally Kafka topics in the ecosystem.
 #[tracing::instrument(skip_all)]
 async fn all_subjects(