fix(observability): log recovered stream failures

Author: CoderYellow

README.md

@@ -131,6 +131,23 @@ You can replace `SPRING_CONFIG_ADDITIONAL_LOCATION` at runtime if a different
 location is required. `JAVA_TOOL_OPTIONS` is also honored by the JVM for options
 such as trust stores, Kerberos diagnostics, or Java agents.
 
+Container logging defaults to `INFO` for this project and `WARN` for Hadoop.
+Both can be changed independently, while standard Spring Boot logging variables
+remain available:
+
+```bash
+docker run --rm -p 8080:8080 \
+  -e YARN_LOG_API_LOG_LEVEL=DEBUG \
+  -e YARN_LOG_API_HADOOP_LOG_LEVEL=INFO \
+  -e LOGGING_LEVEL_REACTOR_NETTY_HTTP_CLIENT=DEBUG \
+  ghcr.io/openprojectx/yarn-log-api:latest
+```
+
+Recovered stream failures are logged with their full stack trace and contextual
+application/container information before an `ERROR` or `WARNING` event is sent
+to the client. Log contents and SPNEGO cookies are never written by these
+diagnostic messages.
+
 The image classpath contains `/etc/hadoop/conf` and `/app/extensions/*`. Mount
 extra JARs into the extension directory when adding a filesystem provider,
 Spring auto-configuration, metrics exporter, or another runtime integration:

app/src/main/resources/application.yaml

@@ -18,6 +18,11 @@ management:
         enabled: true
       show-details: never
 
+logging:
+  level:
+    org.openprojectx.hadoop.yarn.log.api: ${YARN_LOG_API_LOG_LEVEL:INFO}
+    org.apache.hadoop: ${YARN_LOG_API_HADOOP_LOG_LEVEL:WARN}
+
 yarn-log-api:
   poll-interval: 1s
   initial-tail-bytes: 64KB

autoconfigure/src/main/kotlin/org/openprojectx/hadoop/yarn/log/api/autoconfigure/YarnLogSseController.kt

@@ -3,6 +3,7 @@ package org.openprojectx.hadoop.yarn.log.api.autoconfigure
 import org.openprojectx.hadoop.yarn.log.api.YarnLogEvent
 import org.openprojectx.hadoop.yarn.log.api.YarnLogStreamRequest
 import org.openprojectx.hadoop.yarn.log.api.YarnLogStreamService
+import org.slf4j.LoggerFactory
 import org.springframework.http.MediaType
 import org.springframework.http.codec.ServerSentEvent
 import org.springframework.web.bind.annotation.GetMapping
@@ -45,11 +46,30 @@ class YarnLogSseController(
             tailBytes = tailBytes ?: properties.initialTailBytes.toBytes(),
             pollInterval = interval,
         )
-        return service.stream(request).map { event ->
-            ServerSentEvent.builder(event)
-                .id(event.sequence.toString())
-                .event(event.type.name.lowercase())
-                .build()
-        }
+        logger.info(
+            "Opening YARN log SSE stream: applicationId={}, requester={}, follow={}, logFiles={}, " +
+                "containers={}, tailBytes={}, pollInterval={}",
+            request.applicationId,
+            request.requester ?: "anonymous",
+            request.follow,
+            request.logFiles,
+            request.containerIds,
+            request.tailBytes,
+            request.pollInterval,
+        )
+        return service.stream(request)
+            .doOnComplete { logger.info("Completed YARN log SSE stream: applicationId={}", applicationId) }
+            .doOnCancel { logger.info("Client cancelled YARN log SSE stream: applicationId={}", applicationId) }
+            .doOnError { error -> logger.error("YARN log SSE stream terminated unexpectedly: applicationId={}", applicationId, error) }
+            .map { event ->
+                ServerSentEvent.builder(event)
+                    .id(event.sequence.toString())
+                    .event(event.type.name.lowercase())
+                    .build()
+            }
+    }
+
+    private companion object {
+        val logger = LoggerFactory.getLogger(YarnLogSseController::class.java)
     }
 }

autoconfigure/src/main/kotlin/org/openprojectx/hadoop/yarn/log/api/autoconfigure/YarnLogWebSocketHandler.kt

@@ -3,6 +3,7 @@ package org.openprojectx.hadoop.yarn.log.api.autoconfigure
 import tools.jackson.databind.ObjectMapper
 import org.openprojectx.hadoop.yarn.log.api.YarnLogStreamRequest
 import org.openprojectx.hadoop.yarn.log.api.YarnLogStreamService
+import org.slf4j.LoggerFactory
 import org.springframework.web.reactive.socket.WebSocketHandler
 import org.springframework.web.reactive.socket.WebSocketSession
 import reactor.core.publisher.Mono
@@ -23,21 +24,35 @@ class YarnLogWebSocketHandler(
                     require(interval >= properties.minimumPollInterval) {
                         "pollInterval must be at least ${properties.minimumPollInterval}"
                     }
-                    service.stream(
-                        YarnLogStreamRequest(
-                            applicationId = requireNotNull(subscription.applicationId) { "applicationId is required" },
-                            requester = principal.ifEmpty { null },
-                            follow = subscription.follow,
-                            logFiles = subscription.logFiles.toSet().ifEmpty { setOf("stdout", "stderr") },
-                            containerIds = subscription.containers.toSet(),
-                            tailBytes = subscription.tailBytes ?: properties.initialTailBytes.toBytes(),
-                            pollInterval = interval,
-                        ),
+                    val request = YarnLogStreamRequest(
+                        applicationId = requireNotNull(subscription.applicationId) { "applicationId is required" },
+                        requester = principal.ifEmpty { null },
+                        follow = subscription.follow,
+                        logFiles = subscription.logFiles.toSet().ifEmpty { setOf("stdout", "stderr") },
+                        containerIds = subscription.containers.toSet(),
+                        tailBytes = subscription.tailBytes ?: properties.initialTailBytes.toBytes(),
+                        pollInterval = interval,
                     )
+                    logger.info(
+                        "Opening YARN log WebSocket stream: sessionId={}, applicationId={}, requester={}, " +
+                            "follow={}, logFiles={}, containers={}, tailBytes={}, pollInterval={}",
+                        session.id,
+                        request.applicationId,
+                        request.requester ?: "anonymous",
+                        request.follow,
+                        request.logFiles,
+                        request.containerIds,
+                        request.tailBytes,
+                        request.pollInterval,
+                    )
+                    service.stream(request)
                 }
             }
             .map { event -> session.textMessage(objectMapper.writeValueAsString(event)) }
         return session.send(outgoing)
+            .doOnSuccess { logger.info("Completed YARN log WebSocket session: sessionId={}", session.id) }
+            .doOnCancel { logger.info("Client cancelled YARN log WebSocket session: sessionId={}", session.id) }
+            .doOnError { error -> logger.error("YARN log WebSocket session failed: sessionId={}", session.id, error) }
     }
 
     class Subscription {
@@ -48,4 +63,8 @@ class YarnLogWebSocketHandler(
         var tailBytes: Long? = null
         var pollIntervalMs: Long? = null
     }
+
+    private companion object {
+        val logger = LoggerFactory.getLogger(YarnLogWebSocketHandler::class.java)
+    }
 }

core/build.gradle.kts

@@ -8,6 +8,7 @@ dependencies {
 
     api("io.projectreactor:reactor-core")
     implementation("org.springframework:spring-webflux")
+    implementation("org.slf4j:slf4j-api")
 
     implementation(libs.hadoopYarnClient) {
         exclude(group = "com.sun.jersey")

core/src/main/kotlin/org/openprojectx/hadoop/yarn/log/api/engine/DefaultYarnLogStreamService.kt

@@ -6,6 +6,7 @@ import org.openprojectx.hadoop.yarn.log.api.YarnLogEventType
 import org.openprojectx.hadoop.yarn.log.api.YarnLogSource
 import org.openprojectx.hadoop.yarn.log.api.YarnLogStreamRequest
 import org.openprojectx.hadoop.yarn.log.api.YarnLogStreamService
+import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.util.retry.Retry
@@ -31,6 +32,14 @@ class DefaultYarnLogStreamService(
             Mono.just(session.event(YarnLogEventType.OPEN)),
             firstSnapshot(session),
         ).onErrorResume { error ->
+            logger.error(
+                "YARN log stream failed: applicationId={}, follow={}, logFiles={}, containers={}",
+                request.applicationId,
+                request.follow,
+                request.logFiles,
+                request.containerIds,
+                error,
+            )
             Flux.just(
                 session.event(
                     type = YarnLogEventType.ERROR,
@@ -57,12 +66,24 @@ class DefaultYarnLogStreamService(
     private fun cycle(session: Session, snapshot: YarnApplicationSnapshot): Flux<YarnLogEvent> {
         val events = mutableListOf<YarnLogEvent>()
         if (session.lastApplicationState != snapshot.state.name) {
+            logger.info(
+                "YARN application state changed: applicationId={}, state={}",
+                snapshot.applicationId,
+                snapshot.state.name,
+            )
             session.lastApplicationState = snapshot.state.name
             events += session.event(YarnLogEventType.APPLICATION_STATE, state = snapshot.state.name)
         }
         snapshot.containers.forEach { container ->
             session.containers[container.containerId] = container
             if (session.discoveredContainers.add(container.containerId)) {
+                logger.debug(
+                    "Discovered YARN container: applicationId={}, containerId={}, nodeId={}, state={}",
+                    snapshot.applicationId,
+                    container.containerId,
+                    container.nodeId,
+                    container.state,
+                )
                 events += session.event(
                     type = YarnLogEventType.CONTAINER_DISCOVERED,
                     container = container,
@@ -108,6 +129,17 @@ class DefaultYarnLogStreamService(
         }
         return fetchAtWindow(session, container, logType, cursor)
             .onErrorResume { error ->
+                logger.warn(
+                    "Unable to read NodeManager log; the stream will continue: applicationId={}, " +
+                        "containerId={}, nodeId={}, logType={}, cursorOffset={}, windowBytes={}",
+                    session.request.applicationId,
+                    container.containerId,
+                    container.nodeId,
+                    logType,
+                    cursor.offset,
+                    cursor.windowBytes,
+                    error,
+                )
                 Mono.just(
                     session.event(
                         type = YarnLogEventType.WARNING,
@@ -140,6 +172,15 @@ class DefaultYarnLogStreamService(
             }
             response.fileLength < cursor.offset -> {
                 val oldOffset = cursor.offset
+                logger.warn(
+                    "NodeManager log was truncated: applicationId={}, containerId={}, logType={}, " +
+                        "oldOffset={}, newLength={}",
+                    session.request.applicationId,
+                    container.containerId,
+                    logType,
+                    oldOffset,
+                    response.fileLength,
+                )
                 cursor.generation++
                 cursor.offset = 0
                 cursor.initialized = false
@@ -162,6 +203,15 @@ class DefaultYarnLogStreamService(
                 if (expanded <= cursor.windowBytes) {
                     Mono.error(IllegalStateException("Log growth exceeded the maximum tail window"))
                 } else {
+                    logger.debug(
+                        "Expanding NodeManager tail window: applicationId={}, containerId={}, logType={}, " +
+                            "oldWindowBytes={}, newWindowBytes={}",
+                        session.request.applicationId,
+                        container.containerId,
+                        logType,
+                        cursor.windowBytes,
+                        expanded,
+                    )
                     cursor.windowBytes = expanded
                     fetchAtWindow(session, container, logType, cursor)
                 }
@@ -216,6 +266,15 @@ class DefaultYarnLogStreamService(
         }
         return aggregated
             .onErrorResume { error ->
+                logger.warn(
+                    "Unable to read aggregated logs after application completion: applicationId={}, " +
+                        "owner={}, logFiles={}, containers={}",
+                    snapshot.applicationId,
+                    snapshot.owner,
+                    session.request.logFiles,
+                    session.request.containerIds,
+                    error,
+                )
                 Flux.just(
                     session.event(
                         type = YarnLogEventType.WARNING,
@@ -309,4 +368,8 @@ class DefaultYarnLogStreamService(
             message = message,
         )
     }
+
+    private companion object {
+        val logger = LoggerFactory.getLogger(DefaultYarnLogStreamService::class.java)
+    }
 }

core/src/main/kotlin/org/openprojectx/hadoop/yarn/log/api/engine/HadoopYarnGateway.kt

@@ -2,6 +2,7 @@ package org.openprojectx.hadoop.yarn.log.api.engine
 
 import org.apache.hadoop.yarn.api.records.ApplicationId
 import org.apache.hadoop.yarn.client.api.YarnClient
+import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
 import reactor.core.scheduler.Scheduler
 
@@ -38,7 +39,20 @@ class HadoopYarnGateway(
                 containers = containers,
             )
         }.subscribeOn(hadoopScheduler)
+            .doOnSubscribe { logger.debug("Fetching YARN application snapshot: applicationId={}", applicationId) }
+            .doOnNext { snapshot ->
+                logger.debug(
+                    "Fetched YARN application snapshot: applicationId={}, state={}, containers={}",
+                    snapshot.applicationId,
+                    snapshot.state.name,
+                    snapshot.containers.size,
+                )
+            }
 
     private fun isFinal(state: String): Boolean =
         state == "FINISHED" || state == "FAILED" || state == "KILLED"
+
+    private companion object {
+        val logger = LoggerFactory.getLogger(HadoopYarnGateway::class.java)
+    }
 }

core/src/main/kotlin/org/openprojectx/hadoop/yarn/log/api/engine/NodeManagerLogClient.kt

@@ -1,6 +1,7 @@
 package org.openprojectx.hadoop.yarn.log.api.engine
 
 import org.apache.hadoop.security.authentication.client.AuthenticatedURL
+import org.slf4j.LoggerFactory
 import org.springframework.http.HttpHeaders
 import org.springframework.http.HttpStatus
 import org.springframework.http.MediaType
@@ -22,6 +23,13 @@ class NodeManagerLogClient(
         windowBytes: Long,
     ): Mono<NodeManagerLogResponse> {
         val uri = buildUri(nodeHttpAddress, containerId, logType, windowBytes)
+        logger.debug(
+            "Fetching NodeManager log tail: containerId={}, logType={}, windowBytes={}, uri={}",
+            containerId,
+            logType,
+            windowBytes,
+            uri,
+        )
         return exchange(uri, allowAuthenticationRetry = true)
             .map { parser.parse(it, windowBytes) }
     }
@@ -42,6 +50,7 @@ class NodeManagerLogClient(
                         when {
                             response.statusCode().is2xxSuccessful -> response.bodyToMono(ByteArray::class.java)
                             response.statusCode() == HttpStatus.UNAUTHORIZED && allowAuthenticationRetry -> {
+                                logger.warn("NodeManager returned 401; refreshing SPNEGO cookie and retrying: uri={}", uri)
                                 response.releaseBody().then(
                                     Mono.defer {
                                         cookieProvider.invalidate(uri)
@@ -68,4 +77,8 @@ class NodeManagerLogClient(
             .encode()
             .toUri()
     }
+
+    private companion object {
+        val logger = LoggerFactory.getLogger(NodeManagerLogClient::class.java)
+    }
 }