chore(amber): reduce engine log verbosity and pin the CI log-level backstop

.github/workflows/build.yml

@@ -291,6 +291,19 @@ 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. 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. 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: |
           sbt "DAO/jacoco" \
               "Auth/jacoco" \
@@ -330,6 +343,12 @@ 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: 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:`
@@ -609,6 +628,19 @@ 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. 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. 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: |
           sbt scalafmtCheckAll \
               "scalafixAll --check" \
@@ -638,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
@@ -651,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
@@ -740,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
@@ -925,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:
@@ -1013,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
@@ -1084,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()

amber/src/main/python/core/runnables/main_loop.py

@@ -444,17 +444,23 @@ 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(
                 PauseType.ECM_PAUSE, channel_id
             )
 
         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)
 

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. 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)
     }

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))

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 =>

amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DPThread.scala

@@ -99,15 +99,15 @@ 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 {
             runDPThreadMainLogic()
           } 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) => {

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)

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)

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
   }
 }

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}
 

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")))
+    )
+  }
 }

common/config/src/test/scala/org/apache/texera/common/config/PekkoConfigSpec.scala

@@ -76,4 +76,44 @@ class PekkoConfigSpec extends AnyFlatSpec with Matchers {
     }
     config.getString("pekko.stdout-loglevel") shouldBe "INFO"
   }
+
+  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.
+    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))
+    }
+  }
+
+  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"
+    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"
+  }
 }