diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 1355a4dbbce..f2c6c0a40c0 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -291,6 +291,14 @@ 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; 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 run: | sbt "DAO/jacoco" \ "Auth/jacoco" \ @@ -330,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:` @@ -609,6 +621,14 @@ 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; 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 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..d252be934e6 100644 --- a/amber/src/main/python/core/runnables/main_loop.py +++ b/amber/src/main/python/core/runnables/main_loop.py @@ -444,8 +444,11 @@ 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( - f"receive channel ECM from {channel_id}, id = {ecm.id}, cmd = {command}" + logger.debug( + "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( @@ -453,8 +456,11 @@ def _process_ecm(self, ecm_element: ECMElement): ) if self.context.ecm_manager.is_ecm_aligned(channel_id, ecm): - logger.info( - f"process channel ECM from {channel_id}, id = {ecm.id}, cmd = {command}" + logger.debug( + "process channel ECM from {}, id = {}, cmd = {}", + channel_id, + ecm.id, + command, ) if command is not None: @@ -470,9 +476,11 @@ 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( - f"send ECM to {active_channel_id}," - f" id = {ecm.id}, cmd = {command}" + logger.debug( + "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 2c085b9dc89..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,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"removed actor ref for $id. 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 } }