From 89289d359484960c04265a261639008eb853d689 Mon Sep 17 00:00:00 2001 From: Xinyuan Lin Date: Wed, 22 Jul 2026 03:44:13 -0700 Subject: [PATCH 1/7] chore(amber): reduce CI log verbosity from per-worker/per-message engine logs Amber engine loggers are named "] [" by AmberLogging so the pattern renders both the actor id and the class. A side effect is that these logger names start with the actor id, not "org.apache", so the `org.apache=WARN` rule in logback.xml never matches them and every one of these statements emits at the root INFO level. In loop/integration tests workers are recreated each iteration, so per-ECM, per-worker, per-region lines fire dozens of times and flood the CI console. Demote the chatty per-message / per-worker-lifecycle / per-region logs to DEBUG so they no longer print at the default INFO level (CI, local, and prod), and demote three misleading WARNs that fire on normal operation: * receive/process/send ECM (DataProcessor.scala, main_loop.py) * register->Actor / "unknown identifier" / "might have crashed" (PekkoActorRefMappingService.scala) * replay-log conf, DP thread start/exit, "Starting the worker", "Region N successfully terminated", operator completion * ClientActor "cannot handle" fallthrough Genuine fault paths are untouched (DP-thread ERROR, "Failed to fetch actorRef" WARN, "Error when terminating region" WARN). As a CI backstop, pin TEXERA_SERVICE_LOG_LEVEL and UDF_PYTHON_LOG_STREAMHANDLER_LEVEL to WARN on the amber unit and integration test steps. --- .github/workflows/build.yml | 10 ++++++++++ amber/src/main/python/core/runnables/main_loop.py | 6 +++--- .../common/PekkoActorRefMappingService.scala | 6 +++--- .../engine/architecture/common/WorkflowActor.scala | 2 +- .../scheduling/RegionExecutionManager.scala | 2 +- .../amber/engine/architecture/worker/DPThread.scala | 4 ++-- .../engine/architecture/worker/DataProcessor.scala | 8 ++++---- .../worker/promisehandlers/StartHandler.scala | 2 +- .../amber/engine/common/client/ClientActor.scala | 2 +- 9 files changed, 26 insertions(+), 16 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 1e0092fa4b6..f2fe80decab 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -291,6 +291,11 @@ jobs: AMBER_TEST_FILTER: skip-integration # unit job uses its provisioned postgres catalog; default (rest) needs a Lakekeeper not run here STORAGE_ICEBERG_CATALOG_TYPE: postgres + # Backstop for CI log volume: the chatty per-worker/per-message engine + # logs are DEBUG in source, but pin the JVM root and the spawned Python + # UDF workers to WARN so nothing at INFO leaks back into the CI console. + TEXERA_SERVICE_LOG_LEVEL: WARN + UDF_PYTHON_LOG_STREAMHANDLER_LEVEL: WARN run: | sbt "DAO/jacoco" \ "Auth/jacoco" \ @@ -609,6 +614,11 @@ jobs: # by amber's unit-test coverage. env: AMBER_TEST_FILTER: integration-only + # Backstop for CI log volume: the chatty per-worker/per-message engine + # logs are DEBUG in source, but pin the JVM root and the spawned Python + # UDF workers to WARN so nothing at INFO leaks back into the CI console. + TEXERA_SERVICE_LOG_LEVEL: WARN + UDF_PYTHON_LOG_STREAMHANDLER_LEVEL: WARN run: | sbt scalafmtCheckAll \ "scalafixAll --check" \ diff --git a/amber/src/main/python/core/runnables/main_loop.py b/amber/src/main/python/core/runnables/main_loop.py index 29cfc2e622a..e1f97af0d48 100644 --- a/amber/src/main/python/core/runnables/main_loop.py +++ b/amber/src/main/python/core/runnables/main_loop.py @@ -444,7 +444,7 @@ def _process_ecm(self, ecm_element: ECMElement): ecm = ecm_element.payload command = ecm.command_mapping.get(self.context.worker_id) channel_id = self.context.current_input_channel_id - logger.info( + logger.debug( f"receive channel ECM from {channel_id}, id = {ecm.id}, cmd = {command}" ) if ecm.ecm_type != EmbeddedControlMessageType.NO_ALIGNMENT: @@ -453,7 +453,7 @@ def _process_ecm(self, ecm_element: ECMElement): ) if self.context.ecm_manager.is_ecm_aligned(channel_id, ecm): - logger.info( + logger.debug( f"process channel ECM from {channel_id}, id = {ecm.id}, cmd = {command}" ) @@ -470,7 +470,7 @@ def _process_ecm(self, ecm_element: ECMElement): active_channel_id ) in self.context.output_manager.get_output_channel_ids(): if active_channel_id in downstream_channels_in_scope: - logger.info( + logger.debug( f"send ECM to {active_channel_id}," f" id = {ecm.id}, cmd = {command}" ) diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala index 2c085b9dc89..15df4791fa3 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala @@ -76,13 +76,13 @@ class PekkoActorRefMappingService(actorService: PekkoActorService) extends Amber def removeActorRef(id: ActorVirtualIdentity): Unit = { if (actorRefMapping.contains(id)) { val ref = actorRefMapping.remove(id).get - logger.warn(s"actor $id is not reachable anymore, it might have crashed. old ref = $ref") + logger.debug(s"actor $id is not reachable anymore, it might have crashed. old ref = $ref") } } def registerActorRef(id: ActorVirtualIdentity, ref: ActorRef): Unit = { if (!actorRefMapping.contains(id)) { - logger.info(s"register ${VirtualIdentityUtils.toShorterString(id)} -> $ref") + logger.debug(s"register ${VirtualIdentityUtils.toShorterString(id)} -> $ref") actorRefMapping(id) = ref if (messageStash.contains(id)) { val stash = messageStash(id) @@ -119,7 +119,7 @@ class PekkoActorRefMappingService(actorService: PekkoActorService) extends Amber } } else { // on coordinator, wait for actor ref registration. - logger.warn(s"unknown identifier: ${VirtualIdentityUtils.toShorterString(id)}") + logger.debug(s"unknown identifier: ${VirtualIdentityUtils.toShorterString(id)}") val toNotifySet = toNotifyOnRegistration.getOrElseUpdate(id, mutable.HashSet[ActorRef]()) replyTo.foreach(toNotifySet.add) } diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/WorkflowActor.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/WorkflowActor.scala index 0f744dca264..ba2a539d11d 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/WorkflowActor.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/WorkflowActor.scala @@ -102,7 +102,7 @@ abstract class WorkflowActor( val transferService: PekkoMessageTransferService = new PekkoMessageTransferService(actorService, actorRefMappingService, handleBackpressure) - logger.info(s"worker replay log writing conf: $replayLogConfOpt") + logger.debug(s"worker replay log writing conf: $replayLogConfOpt") val logStorage: SequentialRecordStorage[ReplayLogRecord] = SequentialRecordStorage.getStorage(replayLogConfOpt.map(_.writeTo)) diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala index fd239b59d58..586d55ca256 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala @@ -210,7 +210,7 @@ class RegionExecutionManager( // 3. Log whether the kills were successful gracefulStopRequests.transform { case Return(_) => - logger.info(s"Region ${region.id.id} successfully terminated.") + logger.debug(s"Region ${region.id.id} successfully terminated.") regionExecution.getAllOperatorExecutions.foreach { case (_, opExec) => opExec.getWorkerIds.foreach { workerId => diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DPThread.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DPThread.scala index f845836843d..829eaf1a77e 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DPThread.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DPThread.scala @@ -99,7 +99,7 @@ class DPThread( // so create() can append a suffix without the execution id. Once per thread, // assuming a thread serves one execution. LargeBinaryManager.setCurrentBaseUri(largeBinaryBaseUri) - logger.info("DP thread started") + logger.debug("DP thread started") startFuture.complete(()) dp.statisticsManager.initializeWorkerStartTime(System.nanoTime()) try { @@ -107,7 +107,7 @@ class DPThread( } catch safely { case _: InterruptedException => // dp thread will stop here - logger.info("DP Thread exits") + logger.debug("DP Thread exits") case err: Throwable => logger.error("DP Thread exists unexpectedly", err) dp.outputHandler(Left(MainThreadDelegateMessage((worker) => { diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala index a86d16c9262..feec522a9d4 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala @@ -171,7 +171,7 @@ class DataProcessor( executor.close() adaptiveBatchingMonitor.stopAdaptiveBatching() stateManager.transitTo(COMPLETED) - logger.info( + logger.debug( s"$executor completed, # of input ports = ${inputManager.getAllPorts.size}, " + s"input tuple count = ${statisticsManager.getInputTupleCount}, " + s"output tuple count = ${statisticsManager.getOutputTupleCount}" @@ -242,14 +242,14 @@ class DataProcessor( ): Unit = { inputManager.currentChannelId = channelId val command = ecm.commandMapping.get(actorId.name) - logger.info(s"receive ECM from $channelId, id = ${ecm.id}, cmd = $command") + logger.debug(s"receive ECM from $channelId, id = ${ecm.id}, cmd = $command") if (ecm.ecmType != NO_ALIGNMENT) { pauseManager.pauseInputChannel(ECMPause(ecm.id), List(channelId)) } if (ecmManager.isECMAligned(channelId, ecm)) { logManager.markAsReplayDestination(ecm.id) // invoke the control command carried with the ECM - logger.info(s"process ECM from $channelId, id = ${ecm.id}, cmd = $command") + logger.debug(s"process ECM from $channelId, id = ${ecm.id}, cmd = $command") if (command.isDefined) { // The reply must go back to the actor that originated the invocation // (recorded in command.context.sender), not to channelId.fromWorkerId. @@ -268,7 +268,7 @@ class DataProcessor( outputManager.flush(Some(downstreamChannelsInScope)) outputGateway.getActiveChannels.foreach { activeChannelId => if (downstreamChannelsInScope.contains(activeChannelId)) { - logger.info( + logger.debug( s"send ECM to $activeChannelId, id = ${ecm.id}, cmd = $command" ) outputGateway.sendTo(activeChannelId, ecm) diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartHandler.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartHandler.scala index 5d1bf8ccbde..a9e3f4ed466 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartHandler.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartHandler.scala @@ -39,7 +39,7 @@ trait StartHandler { request: EmptyRequest, ctx: AsyncRPCContext ): Future[WorkerStateResponse] = { - logger.info("Starting the worker.") + logger.debug("Starting the worker.") if (dp.executor.isInstanceOf[SourceOperatorExecutor]) { val channelId = ChannelIdentity(ActorVirtualIdentity("SOURCE_STARTER"), actorId, isControl = false) diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/common/client/ClientActor.scala b/amber/src/main/scala/org/apache/texera/amber/engine/common/client/ClientActor.scala index 6cff3282969..9a2bc1ff628 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/common/client/ClientActor.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/common/client/ClientActor.scala @@ -156,6 +156,6 @@ private[client] class ClientActor extends Actor with AmberLogging { sender() ! Ack coordinator ! x case other => - logger.warn("client actor cannot handle " + other) //skip + logger.debug("client actor cannot handle " + other) //skip } } From 6812804910bd94b35aad6d636a990df36d627ace Mon Sep 17 00:00:00 2001 From: Xinyuan Lin Date: Wed, 22 Jul 2026 18:40:21 -0700 Subject: [PATCH 2/7] chore(amber): address review feedback on log demotion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - main_loop.py: switch the three per-ECM debug logs from eager f-strings to loguru's lazy `{}`-placeholder form. loguru's _log short-circuits on core.min_level before formatting, so the ChannelIdentity/command string building is now skipped entirely when DEBUG is disabled (the default). - PekkoActorRefMappingService: reword the demoted teardown log to a neutral "removed actor ref for " — removeActorRef also runs on normal graceful region teardown, so the old "might have crashed" wording was misleading. The Scala interpolated debug logs are left as-is: com.typesafe.scala-logging 3.9.6 Logger methods are macros that guard the message with isDebugEnabled, so `s"..."` is not built when the level is disabled. --- .../src/main/python/core/runnables/main_loop.py | 16 ++++++++++++---- .../common/PekkoActorRefMappingService.scala | 2 +- 2 files changed, 13 insertions(+), 5 deletions(-) diff --git a/amber/src/main/python/core/runnables/main_loop.py b/amber/src/main/python/core/runnables/main_loop.py index e1f97af0d48..d252be934e6 100644 --- a/amber/src/main/python/core/runnables/main_loop.py +++ b/amber/src/main/python/core/runnables/main_loop.py @@ -445,7 +445,10 @@ def _process_ecm(self, ecm_element: ECMElement): command = ecm.command_mapping.get(self.context.worker_id) channel_id = self.context.current_input_channel_id logger.debug( - f"receive channel ECM from {channel_id}, id = {ecm.id}, cmd = {command}" + "receive channel ECM from {}, id = {}, cmd = {}", + channel_id, + ecm.id, + command, ) if ecm.ecm_type != EmbeddedControlMessageType.NO_ALIGNMENT: self.context.pause_manager.pause_input_channel( @@ -454,7 +457,10 @@ def _process_ecm(self, ecm_element: ECMElement): if self.context.ecm_manager.is_ecm_aligned(channel_id, ecm): logger.debug( - f"process channel ECM from {channel_id}, id = {ecm.id}, cmd = {command}" + "process channel ECM from {}, id = {}, cmd = {}", + channel_id, + ecm.id, + command, ) if command is not None: @@ -471,8 +477,10 @@ def _process_ecm(self, ecm_element: ECMElement): ) in self.context.output_manager.get_output_channel_ids(): if active_channel_id in downstream_channels_in_scope: logger.debug( - f"send ECM to {active_channel_id}," - f" id = {ecm.id}, cmd = {command}" + "send ECM to {}, id = {}, cmd = {}", + active_channel_id, + ecm.id, + command, ) self._send_ecm_to_channel(active_channel_id, ecm) diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala index 15df4791fa3..6a1eef86e8c 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala @@ -76,7 +76,7 @@ class PekkoActorRefMappingService(actorService: PekkoActorService) extends Amber def removeActorRef(id: ActorVirtualIdentity): Unit = { if (actorRefMapping.contains(id)) { val ref = actorRefMapping.remove(id).get - logger.debug(s"actor $id is not reachable anymore, it might have crashed. old ref = $ref") + logger.debug(s"removed actor ref for $id. old ref = $ref") } } From 35203dc9505b5b7520c4e09ac6cb45ae91a9b6e9 Mon Sep 17 00:00:00 2001 From: Xinyuan Lin Date: Wed, 22 Jul 2026 21:43:34 -0700 Subject: [PATCH 3/7] fix(ci): pin Python UDF worker log level to WARNING, not WARN loguru has no WARN level (only WARNING), so UDF_PYTHON_LOG_STREAMHANDLER_LEVEL=WARN made every spawned Python UDF worker raise `ValueError: Level 'WARN' does not exist` at startup, before it connected back to the JVM. PythonProxyClient then blocked forever on the port promise (Await.result, no timeout), so the amber-integration job hung until GitHub's 6h cap. logback (TEXERA_SERVICE_LOG_LEVEL) keeps WARN, which is its correct spelling. Also add a 40-min timeout to the amber-integration job so a worker-startup failure fails fast instead of burning a 6h runner. --- .github/workflows/build.yml | 22 ++++++++++++++++------ 1 file changed, 16 insertions(+), 6 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index d38c46c4e6f..f2c6c0a40c0 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -292,10 +292,13 @@ jobs: # unit job uses its provisioned postgres catalog; default (rest) needs a Lakekeeper not run here STORAGE_ICEBERG_CATALOG_TYPE: postgres # Backstop for CI log volume: the chatty per-worker/per-message engine - # logs are DEBUG in source, but pin the JVM root and the spawned Python - # UDF workers to WARN so nothing at INFO leaks back into the CI console. + # logs are DEBUG in source; pin the JVM root and the spawned Python UDF + # workers so nothing at INFO leaks into the CI console. Mind the spelling: + # logback's level is WARN, but loguru (the Python worker) only knows + # WARNING and raises ValueError on "WARN" — which crashes the worker at + # startup before it hands its port back, hanging the whole job. TEXERA_SERVICE_LOG_LEVEL: WARN - UDF_PYTHON_LOG_STREAMHANDLER_LEVEL: WARN + UDF_PYTHON_LOG_STREAMHANDLER_LEVEL: WARNING run: | sbt "DAO/jacoco" \ "Auth/jacoco" \ @@ -335,6 +338,10 @@ jobs: # cutting lints (scalafmt / scalafix) and the amber dist + binary # license check stay in `amber`; this job is tests-only. if: ${{ inputs.run_amber_integration }} + # A Python UDF worker that fails to start (e.g. a bad log level) leaves the + # JVM blocked on the proxy handshake with no internal timeout, so the whole + # job would otherwise hang until GitHub's 6h cap. Fail fast instead. + timeout-minutes: 40 strategy: # macOS provisions postgres / minio / lakekeeper natively because # GitHub-hosted macOS runners have no Docker (and `services:` @@ -615,10 +622,13 @@ jobs: env: AMBER_TEST_FILTER: integration-only # Backstop for CI log volume: the chatty per-worker/per-message engine - # logs are DEBUG in source, but pin the JVM root and the spawned Python - # UDF workers to WARN so nothing at INFO leaks back into the CI console. + # logs are DEBUG in source; pin the JVM root and the spawned Python UDF + # workers so nothing at INFO leaks into the CI console. Mind the spelling: + # logback's level is WARN, but loguru (the Python worker) only knows + # WARNING and raises ValueError on "WARN" — which crashes the worker at + # startup before it hands its port back, hanging the whole job. TEXERA_SERVICE_LOG_LEVEL: WARN - UDF_PYTHON_LOG_STREAMHANDLER_LEVEL: WARN + UDF_PYTHON_LOG_STREAMHANDLER_LEVEL: WARNING run: | sbt scalafmtCheckAll \ "scalafixAll --check" \ From 02e917cefff8d76d745e10a6ef5a569ebe5d5e4d Mon Sep 17 00:00:00 2001 From: Xinyuan Lin Date: Fri, 24 Jul 2026 19:32:27 -0700 Subject: [PATCH 4/7] chore(amber): wire CI log levels to runner.debug, address review - build.yml: the WARN/WARNING pins on the two amber test steps now flip to DEBUG when the run has "Enable debug logging" set (runner.debug == '1'), so the demoted engine logs come back on demand without editing the workflow. - build.yml: lower the amber-integration job timeout from 40 to 25 minutes; the last 38 green runs on main took 9.0-16.4 min (avg 12.1), so 25 keeps cold-cache headroom while still failing ~14x faster than the 6h cap. - PekkoActorRefMappingService: reword the teardown log to "actor is not reachable anymore. old ref = " per review - the mapping service cannot tell a graceful stop from a crash, so the message now infers neither. --- .github/workflows/build.yml | 36 +++++++++++-------- .../common/PekkoActorRefMappingService.scala | 2 +- 2 files changed, 23 insertions(+), 15 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index f2c6c0a40c0..fffbdcdcc37 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -293,12 +293,15 @@ jobs: STORAGE_ICEBERG_CATALOG_TYPE: postgres # Backstop for CI log volume: the chatty per-worker/per-message engine # logs are DEBUG in source; pin the JVM root and the spawned Python UDF - # workers so nothing at INFO leaks into the CI console. Mind the spelling: - # logback's level is WARN, but loguru (the Python worker) only knows - # WARNING and raises ValueError on "WARN" — which crashes the worker at - # startup before it hands its port back, hanging the whole job. - TEXERA_SERVICE_LOG_LEVEL: WARN - UDF_PYTHON_LOG_STREAMHANDLER_LEVEL: WARNING + # workers so nothing at INFO leaks into the CI console. Re-running the + # job with "Enable debug logging" (which sets runner.debug to '1') + # flips both to DEBUG, bringing the full engine trace back on demand. + # Mind the spelling: logback's level is WARN, but loguru (the Python + # worker) only knows WARNING and raises ValueError on "WARN" — which + # crashes the worker at startup before it hands its port back, + # hanging the whole job. + TEXERA_SERVICE_LOG_LEVEL: ${{ runner.debug == '1' && 'DEBUG' || 'WARN' }} + UDF_PYTHON_LOG_STREAMHANDLER_LEVEL: ${{ runner.debug == '1' && 'DEBUG' || 'WARNING' }} run: | sbt "DAO/jacoco" \ "Auth/jacoco" \ @@ -340,8 +343,10 @@ jobs: if: ${{ inputs.run_amber_integration }} # A Python UDF worker that fails to start (e.g. a bad log level) leaves the # JVM blocked on the proxy handshake with no internal timeout, so the whole - # job would otherwise hang until GitHub's 6h cap. Fail fast instead. - timeout-minutes: 40 + # job would otherwise hang until GitHub's 6h cap. Fail fast instead: recent + # green runs take 9-17 minutes, so 25 leaves headroom for a cold cache or a + # slow macOS runner without masking a real hang. + timeout-minutes: 25 strategy: # macOS provisions postgres / minio / lakekeeper natively because # GitHub-hosted macOS runners have no Docker (and `services:` @@ -623,12 +628,15 @@ jobs: AMBER_TEST_FILTER: integration-only # Backstop for CI log volume: the chatty per-worker/per-message engine # logs are DEBUG in source; pin the JVM root and the spawned Python UDF - # workers so nothing at INFO leaks into the CI console. Mind the spelling: - # logback's level is WARN, but loguru (the Python worker) only knows - # WARNING and raises ValueError on "WARN" — which crashes the worker at - # startup before it hands its port back, hanging the whole job. - TEXERA_SERVICE_LOG_LEVEL: WARN - UDF_PYTHON_LOG_STREAMHANDLER_LEVEL: WARNING + # workers so nothing at INFO leaks into the CI console. Re-running the + # job with "Enable debug logging" (which sets runner.debug to '1') + # flips both to DEBUG, bringing the full engine trace back on demand. + # Mind the spelling: logback's level is WARN, but loguru (the Python + # worker) only knows WARNING and raises ValueError on "WARN" — which + # crashes the worker at startup before it hands its port back, + # hanging the whole job. + TEXERA_SERVICE_LOG_LEVEL: ${{ runner.debug == '1' && 'DEBUG' || 'WARN' }} + UDF_PYTHON_LOG_STREAMHANDLER_LEVEL: ${{ runner.debug == '1' && 'DEBUG' || 'WARNING' }} run: | sbt scalafmtCheckAll \ "scalafixAll --check" \ diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala index 6a1eef86e8c..f3fa7f3b928 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala @@ -76,7 +76,7 @@ class PekkoActorRefMappingService(actorService: PekkoActorService) extends Amber def removeActorRef(id: ActorVirtualIdentity): Unit = { if (actorRefMapping.contains(id)) { val ref = actorRefMapping.remove(id).get - logger.debug(s"removed actor ref for $id. old ref = $ref") + logger.debug(s"actor $id is not reachable anymore. old ref = $ref") } } From b67bf3acd7ba924229fd68b87bd122b8cf98044d Mon Sep 17 00:00:00 2001 From: Xinyuan Lin Date: Fri, 24 Jul 2026 19:50:23 -0700 Subject: [PATCH 5/7] fix(amber): translate logback log-level spellings for pekko.loglevel The CI backstop sets TEXERA_SERVICE_LOG_LEVEL=WARN, which cluster.conf forwards into pekko.loglevel. Pekko only accepts OFF/ERROR/WARNING/INFO/ DEBUG (Logging.levelFor), so every ActorSystem creation printed "unknown pekko.loglevel WARN" with a LoggerException stack trace and fell back to ERROR - one per spec in the amber unit and integration jobs, the very noise this PR is meant to remove. The env knob cannot simply switch to WARNING: the same variable drives the logback root level in logback.xml, and logback silently falls back to DEBUG on names it does not know - flooding instead of quieting. Normalize at the single choke point instead: PekkoConfig.pekkoConfig (which every ActorSystem, prod and test, receives via AmberRuntime.pekkoConfig) now rewrites pekko.loglevel through normalizePekkoLogLevel (WARN -> WARNING, TRACE/ALL -> DEBUG), so one env knob drives both systems in either spelling. Covered by a new invariant test - the resolved loglevel must be pekko-valid whatever the env holds, red on WARN before this fix - plus unit tests for the mapping. --- .github/workflows/build.yml | 8 ++++-- common/config/src/main/resources/cluster.conf | 3 ++ .../texera/common/config/PekkoConfig.scala | 28 +++++++++++++++++-- .../common/config/PekkoConfigSpec.scala | 24 ++++++++++++++++ 4 files changed, 59 insertions(+), 4 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index fffbdcdcc37..d6388041d60 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -299,7 +299,9 @@ jobs: # Mind the spelling: logback's level is WARN, but loguru (the Python # worker) only knows WARNING and raises ValueError on "WARN" — which # crashes the worker at startup before it hands its port back, - # hanging the whole job. + # hanging the whole job. pekko.loglevel shares loguru's vocabulary; + # PekkoConfig.normalizePekkoLogLevel translates the logback spelling + # so the JVM knob can stay WARN here. TEXERA_SERVICE_LOG_LEVEL: ${{ runner.debug == '1' && 'DEBUG' || 'WARN' }} UDF_PYTHON_LOG_STREAMHANDLER_LEVEL: ${{ runner.debug == '1' && 'DEBUG' || 'WARNING' }} run: | @@ -634,7 +636,9 @@ jobs: # Mind the spelling: logback's level is WARN, but loguru (the Python # worker) only knows WARNING and raises ValueError on "WARN" — which # crashes the worker at startup before it hands its port back, - # hanging the whole job. + # hanging the whole job. pekko.loglevel shares loguru's vocabulary; + # PekkoConfig.normalizePekkoLogLevel translates the logback spelling + # so the JVM knob can stay WARN here. TEXERA_SERVICE_LOG_LEVEL: ${{ runner.debug == '1' && 'DEBUG' || 'WARN' }} UDF_PYTHON_LOG_STREAMHANDLER_LEVEL: ${{ runner.debug == '1' && 'DEBUG' || 'WARNING' }} run: | diff --git a/common/config/src/main/resources/cluster.conf b/common/config/src/main/resources/cluster.conf index 524cd0216d3..60feaa6ab4b 100644 --- a/common/config/src/main/resources/cluster.conf +++ b/common/config/src/main/resources/cluster.conf @@ -23,6 +23,9 @@ pekko { # Log level used by the configured loggers (see "loggers") as soon # as they have been started; before that, see "stdout-loglevel" # Options: OFF, ERROR, WARNING, INFO, DEBUG + # The env override may arrive in logback spelling (the same variable drives + # the logback root level in logback.xml); PekkoConfig.normalizePekkoLogLevel + # translates e.g. WARN -> WARNING before this reaches any ActorSystem. loglevel = "INFO" loglevel = ${?TEXERA_SERVICE_LOG_LEVEL} diff --git a/common/config/src/main/scala/org/apache/texera/common/config/PekkoConfig.scala b/common/config/src/main/scala/org/apache/texera/common/config/PekkoConfig.scala index 449df5ef9d4..d1159be70a5 100644 --- a/common/config/src/main/scala/org/apache/texera/common/config/PekkoConfig.scala +++ b/common/config/src/main/scala/org/apache/texera/common/config/PekkoConfig.scala @@ -18,13 +18,37 @@ package org.apache.texera.common.config -import com.typesafe.config.{Config, ConfigFactory} +import com.typesafe.config.{Config, ConfigFactory, ConfigValueFactory} object PekkoConfig { // Load configuration private val conf: Config = ConfigFactory.parseResources("cluster.conf").resolve() + /** + * Translate a logback level spelling into one pekko accepts. + * + * cluster.conf forwards ${?TEXERA_SERVICE_LOG_LEVEL} into pekko.loglevel, but that env + * var's vocabulary belongs to logback: the same variable drives the logback root level + * in logback.xml, and logback cannot translate on its side (it silently falls back to + * DEBUG on names it does not know, such as WARNING). Pekko in turn only accepts OFF, + * ERROR, WARNING, INFO and DEBUG, and prints a LoggerException on every ActorSystem + * creation before falling back to ERROR when handed a logback-only spelling such as + * WARN. Translating here lets the single env knob drive both systems. + */ + private[config] def normalizePekkoLogLevel(level: String): String = + level.toUpperCase match { + case "WARN" => "WARNING" + case "TRACE" | "ALL" => "DEBUG" + case other => other + } + // Return the complete Pekko configuration with fallback to default application config - def pekkoConfig: Config = conf.withFallback(ConfigFactory.defaultApplication()).resolve() + def pekkoConfig: Config = { + val resolved = conf.withFallback(ConfigFactory.defaultApplication()).resolve() + resolved.withValue( + "pekko.loglevel", + ConfigValueFactory.fromAnyRef(normalizePekkoLogLevel(resolved.getString("pekko.loglevel"))) + ) + } } diff --git a/common/config/src/test/scala/org/apache/texera/common/config/PekkoConfigSpec.scala b/common/config/src/test/scala/org/apache/texera/common/config/PekkoConfigSpec.scala index 13282f286da..cc734b7e9ae 100644 --- a/common/config/src/test/scala/org/apache/texera/common/config/PekkoConfigSpec.scala +++ b/common/config/src/test/scala/org/apache/texera/common/config/PekkoConfigSpec.scala @@ -76,4 +76,28 @@ class PekkoConfigSpec extends AnyFlatSpec with Matchers { } config.getString("pekko.stdout-loglevel") shouldBe "INFO" } + + it should "always expose a loglevel pekko accepts, whatever spelling the env used" in { + // Pekko only knows these level names (Logging.levelFor). Anything else — such as + // logback's WARN arriving through ${?TEXERA_SERVICE_LOG_LEVEL} — makes every + // ActorSystem creation print a LoggerException and fall back to ERROR. + Set("OFF", "ERROR", "WARNING", "INFO", "DEBUG") should contain( + PekkoConfig.pekkoConfig.getString("pekko.loglevel") + ) + } + + "PekkoConfig.normalizePekkoLogLevel" should "translate logback-only spellings to pekko's" in { + PekkoConfig.normalizePekkoLogLevel("WARN") shouldBe "WARNING" + PekkoConfig.normalizePekkoLogLevel("warn") shouldBe "WARNING" + PekkoConfig.normalizePekkoLogLevel("TRACE") shouldBe "DEBUG" + PekkoConfig.normalizePekkoLogLevel("ALL") shouldBe "DEBUG" + } + + it should "pass levels pekko already accepts through unchanged" in { + Seq("OFF", "ERROR", "WARNING", "INFO", "DEBUG").foreach { level => + PekkoConfig.normalizePekkoLogLevel(level) shouldBe level + } + // pekko's levelFor is case-insensitive, so uppercasing valid input is harmless + PekkoConfig.normalizePekkoLogLevel("info") shouldBe "INFO" + } } From d7982b020c8f98be3ee1655a233d23bf1b5ebe5f Mon Sep 17 00:00:00 2001 From: Xinyuan Lin Date: Sat, 25 Jul 2026 00:26:22 -0700 Subject: [PATCH 6/7] ci: extend the log-level backstop to every service with a level knob Review follow-up on the amber-only scoping: the WARN backstop wired to "Enable debug logging" now covers every CI step that runs or boots a component with a log-level configuration, keeping one consistent level across the board: - platform test step: all six platform services read TEXERA_SERVICE_LOG_LEVEL into their Dropwizard logging config (logback vocabulary, WARN); - platform-integration + amber-integration boot smoke steps: same env for the packaged services; safe because smoke-boot's verdict is LISTEN-based, never log-scraping (#6332); - pyamber unit + Python integration pytest steps: LOGURU_LEVEL=WARNING (loguru vocabulary) for loguru's default stderr sink, which prints straight into the CI log because pytest runs with -s; test-local sinks with an explicit level are unaffected; - agent-service unit tests: TEXERA_SERVICE_LOG_LEVEL=WARN - env.ts validates ["ERROR","WARN","INFO","DEBUG"] before mapping to pino's lowercase levels, so the logback spelling is correct there too. Components with no log-level configuration (frontend, infra, pyright-language-service) are left alone. --- .github/workflows/build.yml | 35 +++++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index d6388041d60..b2feaad21d9 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -670,6 +670,10 @@ jobs: # root (the default) that resolves to ./amber and reads # amber/src/main/resources/web-config.yml (port 8080). if: matrix.os == 'ubuntu-latest' + env: + # Quiet boot logs, same wiring as the test steps above. Safe here: + # smoke-boot's verdict is LISTEN-based, never log-scraping (#6332). + TEXERA_SERVICE_LOG_LEVEL: ${{ runner.debug == '1' && 'DEBUG' || 'WARN' }} run: .github/scripts/smoke-boot.sh "/tmp/dists/amber-*/bin/texera-web-application" 8080 - name: Smoke-test computing-unit-master boots # computing-unit-master boots on postgres only, like texera-web: main @@ -683,9 +687,19 @@ jobs: # amber/src/main/resources/computing-unit-master-config.yml (port 8085). # Reuses the amber dist built + unzipped above for the texera-web boot. if: matrix.os == 'ubuntu-latest' + env: + # Quiet boot logs, same wiring as the test steps above. Safe here: + # smoke-boot's verdict is LISTEN-based, never log-scraping (#6332). + TEXERA_SERVICE_LOG_LEVEL: ${{ runner.debug == '1' && 'DEBUG' || 'WARN' }} run: .github/scripts/smoke-boot.sh "/tmp/dists/amber-*/bin/computing-unit-master" 8085 - name: Run Python integration tests # --junit-xml feeds the Test Analytics upload below. + env: + # Pytest runs with -s, so loguru's default stderr sink prints + # straight into the CI log; pin it to WARNING (loguru has no WARN) + # and flip to DEBUG on debug re-runs. Sinks that tests add with an + # explicit level are unaffected. + LOGURU_LEVEL: ${{ runner.debug == '1' && 'DEBUG' || 'WARNING' }} run: | cd amber && pytest -m integration --junit-xml=junit-integration.xml -sv - name: Upload amber integration test results to Codecov @@ -772,6 +786,12 @@ jobs: - name: Build dist and run ${{ matrix.service }} tests with coverage # Single sbt invocation so dist + test share compiled state. Use # `jacoco` so the codecov upload step has a report to pick up. + env: + # CI log-volume backstop, same wiring as the amber jobs: every + # platform service reads this into its Dropwizard logging config + # (logback vocabulary, hence WARN); a re-run with "Enable debug + # logging" flips it to DEBUG. + TEXERA_SERVICE_LOG_LEVEL: ${{ runner.debug == '1' && 'DEBUG' || 'WARN' }} run: sbt "${{ matrix.sbt_project }}/dist" "${{ matrix.sbt_project }}/jacoco" - name: Unzip ${{ matrix.service }} dist and check binary LICENSE + NOTICE # Each platform service has its own LICENSE-binary / NOTICE-binary at @@ -957,6 +977,9 @@ jobs: # without a runtime classpath/linkage crash (#6220). env: TEXERA_HOME: ${{ github.workspace }} + # Quiet boot logs, same wiring as the amber jobs. Safe here: + # smoke-boot's verdict is LISTEN-based, never log-scraping (#6332). + TEXERA_SERVICE_LOG_LEVEL: ${{ runner.debug == '1' && 'DEBUG' || 'WARN' }} run: .github/scripts/smoke-boot.sh "/tmp/dists/${{ matrix.service }}-*/bin/${{ matrix.service }}" "${{ matrix.port }}" pyamber: @@ -1045,6 +1068,12 @@ jobs: # --junit-xml emits a JUnit-XML report alongside the coverage XML # so the Test Analytics upload below can feed Codecov's failing- # test PR comments and flaky-test detection on main. + env: + # Pytest runs with -s, so loguru's default stderr sink prints + # straight into the CI log; pin it to WARNING (loguru has no WARN) + # and flip to DEBUG on debug re-runs. Sinks that tests add with an + # explicit level are unaffected. + LOGURU_LEVEL: ${{ runner.debug == '1' && 'DEBUG' || 'WARNING' }} run: | cd amber && pytest -m "not integration" --cov=src/main/python --cov-report=xml --junit-xml=junit.xml -sv - name: Upload pyamber coverage to Codecov @@ -1116,6 +1145,12 @@ jobs: # --reporter=junit emits a JUnit-XML alongside coverage so the # Test Analytics upload below can feed Codecov's failing-test PR # comments and flaky-test detection on main. + env: + # CI log-volume backstop, same wiring as the JVM jobs. env.ts + # validates against ["ERROR","WARN","INFO","DEBUG"] before mapping + # to pino's lowercase levels, so the logback spelling WARN is the + # right one here too. + TEXERA_SERVICE_LOG_LEVEL: ${{ runner.debug == '1' && 'DEBUG' || 'WARN' }} run: bun test --coverage --coverage-reporter=lcov --reporter=junit --reporter-outfile=junit.xml - name: Upload agent-service coverage to Codecov if: matrix.os == 'ubuntu-latest' && always() From b06b189730aa983d8d05decee8511036a2399860 Mon Sep 17 00:00:00 2001 From: Xinyuan Lin Date: Sat, 25 Jul 2026 00:55:47 -0700 Subject: [PATCH 7/7] test(config): make the pekko loglevel invariant test env-insensitive Review follow-up: the invariant test asserted the RESOLVED pekko.loglevel is in pekko's accepted set, which fails for any env spelling outside the translated vocabulary (e.g. TEXERA_SERVICE_LOG_LEVEL=FATAL passes through normalizePekkoLogLevel unchanged) - an env-sensitive assertion in a file whose convention is to guard exactly that. Replace it with: - the pure-function invariant: every spelling in the combined logback + pekko vocabulary (OFF/ERROR/WARN/WARNING/INFO/DEBUG/ TRACE/ALL) normalizes into pekko's accepted set; - an env-insensitive plumbing check: whatever the env supplied, the exposed pekko.loglevel equals its normalized form (identity for unknown spellings, so it can never false-positive) - exercised for real by CI, which sets WARN. Verified locally: 8/8 with the env unset, with WARN (the CI condition), and with FATAL (the reviewer's counterexample, which failed before this change). --- .../common/config/PekkoConfigSpec.scala | 30 ++++++++++++++----- 1 file changed, 23 insertions(+), 7 deletions(-) diff --git a/common/config/src/test/scala/org/apache/texera/common/config/PekkoConfigSpec.scala b/common/config/src/test/scala/org/apache/texera/common/config/PekkoConfigSpec.scala index cc734b7e9ae..18dca80e877 100644 --- a/common/config/src/test/scala/org/apache/texera/common/config/PekkoConfigSpec.scala +++ b/common/config/src/test/scala/org/apache/texera/common/config/PekkoConfigSpec.scala @@ -77,16 +77,32 @@ class PekkoConfigSpec extends AnyFlatSpec with Matchers { config.getString("pekko.stdout-loglevel") shouldBe "INFO" } - it should "always expose a loglevel pekko accepts, whatever spelling the env used" in { - // Pekko only knows these level names (Logging.levelFor). Anything else — such as - // logback's WARN arriving through ${?TEXERA_SERVICE_LOG_LEVEL} — makes every + it should "expose exactly the normalized form of whatever the env supplied" in { + // Env-insensitive by construction: for ANY env value the exposed level must + // equal its normalized form (which is the identity for spellings the + // translation doesn't know). This pins the plumbing — pekkoConfig actually + // routes pekko.loglevel through normalizePekkoLogLevel — and is exercised + // for real by CI, which sets the logback spelling WARN. + sys.env + .get("TEXERA_SERVICE_LOG_LEVEL") + .orElse(sys.props.get("TEXERA_SERVICE_LOG_LEVEL")) + .foreach { raw => + PekkoConfig.pekkoConfig.getString("pekko.loglevel") shouldBe + PekkoConfig.normalizePekkoLogLevel(raw) + } + } + + "PekkoConfig.normalizePekkoLogLevel" should "map known spellings into pekko's set" in { + // Every spelling from the combined logback + pekko vocabulary must land in + // pekko's accepted set (Logging.levelFor); anything else makes every // ActorSystem creation print a LoggerException and fall back to ERROR. - Set("OFF", "ERROR", "WARNING", "INFO", "DEBUG") should contain( - PekkoConfig.pekkoConfig.getString("pekko.loglevel") - ) + val pekkoAccepted = Set("OFF", "ERROR", "WARNING", "INFO", "DEBUG") + Seq("OFF", "ERROR", "WARN", "WARNING", "INFO", "DEBUG", "TRACE", "ALL").foreach { spelling => + pekkoAccepted should contain(PekkoConfig.normalizePekkoLogLevel(spelling)) + } } - "PekkoConfig.normalizePekkoLogLevel" should "translate logback-only spellings to pekko's" in { + it should "translate logback-only spellings to pekko's" in { PekkoConfig.normalizePekkoLogLevel("WARN") shouldBe "WARNING" PekkoConfig.normalizePekkoLogLevel("warn") shouldBe "WARNING" PekkoConfig.normalizePekkoLogLevel("TRACE") shouldBe "DEBUG"