feat(observability): deployment (OTel collector + Parca/eBPF)

.github/workflows/create-release-candidate.yml

@@ -174,7 +174,7 @@ jobs:
 
           # Export the single-node directory from the tagged source
           mkdir -p "$TEMP_DIR/_raw"
-          git archive --format=tar "$TAG_NAME" -- bin/single-node/ sql/ NOTICE licenses/LICENSE-BSD-3-Clause.txt | tar -x -C "$TEMP_DIR/_raw"
+          git archive --format=tar "$TAG_NAME" -- bin/single-node/ bin/observability/ sql/ NOTICE licenses/LICENSE-BSD-3-Clause.txt | tar -x -C "$TEMP_DIR/_raw"
 
           # Copy deployment files
           cp "$TEMP_DIR/_raw/bin/single-node/docker-compose.yml" "$BUNDLE_DIR/"
@@ -187,10 +187,19 @@ jobs:
           cp "$TEMP_DIR/_raw/licenses/LICENSE-BSD-3-Clause.txt" "$BUNDLE_DIR/licenses/"
           cp -r "$TEMP_DIR/_raw/sql" "$BUNDLE_DIR/"
 
+          # Observability configs (OTel collector + Parca) referenced by the
+          # opt-in observability compose profiles. Shipped next to the compose
+          # file so the ../observability mounts resolve in the flattened bundle.
+          cp -r "$TEMP_DIR/_raw/bin/observability" "$BUNDLE_DIR/observability"
+
           # Patch the SQL mount path for the self-contained bundle layout
           # In the repo it's ../../sql (relative to bin/single-node/), in the bundle it's ./sql
           sed -i 's|../../sql|./sql|g' "$BUNDLE_DIR/docker-compose.yml"
 
+          # Patch the observability mount paths: ../observability (relative to
+          # bin/single-node/ in the repo) becomes ./observability in the bundle.
+          sed -i 's|../observability|./observability|g' "$BUNDLE_DIR/docker-compose.yml"
+
           # Generate a release-pinned .env file with the version tag
           # Start from the source .env and ensure IMAGE_REGISTRY and IMAGE_TAG are set
           cp "$TEMP_DIR/_raw/bin/single-node/.env" "$BUNDLE_DIR/.env"

.licenserc.yaml

@@ -37,6 +37,8 @@ header:
     - '**/.env.example'
     - '**/.prettierrc'
     - '**/bun.lock'
+    # Helm's generated dependency lock file (no header; like yarn.lock/bun.lock)
+    - '**/Chart.lock'
     # Third-party code with MIT license - see LICENSE file for attribution
     - 'common/workflow-operator/src/main/scala/com/kjetland/**'
     # TypeFox monaco-languageclient derived files (MIT License)

amber/src/main/scala/org/apache/texera/web/ComputingUnitMaster.scala

@@ -139,6 +139,7 @@ class ComputingUnitMaster extends io.dropwizard.Application[Configuration] with
 
   override def run(configuration: Configuration, environment: Environment): Unit = {
     org.apache.texera.observability.OtelInit.init("computing-unit-master")
+    org.apache.texera.web.observability.WorkflowMetricsRecorder.init()
     ObjectMapperUtils.warmupObjectMapperForOperatorsSerde()
 
     SqlServer.initConnection(

amber/src/main/scala/org/apache/texera/web/observability/WorkflowMetricsRecorder.scala

@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.web.observability
+
+import com.typesafe.scalalogging.LazyLogging
+import org.apache.texera.amber.core.virtualidentity.ExecutionIdentity
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState.{
+  COMPLETED,
+  FAILED,
+  KILLED,
+  PAUSED,
+  PAUSING,
+  RESUMING,
+  RUNNING
+}
+import org.apache.texera.observability.WorkflowMetrics
+import org.apache.texera.observability.WorkflowMetrics.WorkflowKind
+import org.apache.texera.web.service.WorkflowService
+
+import java.util.concurrent.ConcurrentHashMap
+
+/**
+  * Drives [[WorkflowMetrics]] from amber's execution lifecycle. The metric
+  * instruments live in common/config; this object is the single place
+  * that records them, so the lifecycle code only needs one-line calls.
+  *
+  *  - start / terminal counters and the duration histogram are recorded
+  *    from [[onStart]] and [[onStateChange]];
+  *  - the always-polled `texera.workflow.active` gauge is sourced from the
+  *    live WorkflowService registry via the supplier registered in [[init]].
+  */
+object WorkflowMetricsRecorder extends LazyLogging {
+
+  // Start time + kind per in-flight run, so a terminal transition can emit
+  // a duration and attribute the outcome. Keyed by the execution identity.
+  private val inFlight = new ConcurrentHashMap[ExecutionIdentity, (Long, WorkflowKind)]()
+
+  private val ActiveStates: Set[WorkflowAggregatedState] = Set(RUNNING, PAUSING, PAUSED, RESUMING)
+  private val TerminalStates: Set[WorkflowAggregatedState] = Set(COMPLETED, FAILED, KILLED)
+
+  /** Register the active-executions gauge supplier and bind the instruments.
+    *  Call once at startup, after OtelInit.init. The supplier is polled on
+    *  every metric collection, so it must never throw.
+    */
+  def init(): Unit = {
+    WorkflowMetrics.setActiveExecutionsSupplier(() =>
+      try {
+        WorkflowService.getAllWorkflowServices.iterator
+          .flatMap(s => Option(s.executionService.getValue))
+          .map(_.executionStateStore.metadataStore.getState.state)
+          .count(ActiveStates.contains)
+          .toLong
+      } catch {
+        case _: Throwable => 0L
+      }
+    )
+    WorkflowMetrics.ensureBound()
+  }
+
+  /** Record that a run started. */
+  def onStart(
+      executionId: ExecutionIdentity,
+      kind: WorkflowKind = WorkflowKind.Interactive
+  ): Unit = {
+    inFlight.put(executionId, (System.currentTimeMillis(), kind))
+    WorkflowMetrics.recordStart(kind)
+  }
+
+  /** Record terminal counters + duration exactly once, on the first
+    *  transition from a non-terminal into a terminal state. Safe to call on
+    *  every state change; non-terminal and repeat-terminal calls are no-ops.
+    */
+  def onStateChange(
+      executionId: ExecutionIdentity,
+      oldState: WorkflowAggregatedState,
+      newState: WorkflowAggregatedState
+  ): Unit = {
+    if (!TerminalStates.contains(newState) || TerminalStates.contains(oldState)) return
+    val entry = Option(inFlight.remove(executionId))
+    val kind = entry.map(_._2).getOrElse(WorkflowKind.Interactive)
+    val durationSec = entry.map(e => (System.currentTimeMillis() - e._1) / 1000.0).getOrElse(0.0)
+    newState match {
+      case COMPLETED => WorkflowMetrics.recordCompletion(kind, durationSec)
+      case FAILED    => WorkflowMetrics.recordFailure(kind, durationSec)
+      case KILLED    => WorkflowMetrics.recordCancellation(kind)
+      case _         => ()
+    }
+  }
+}

amber/src/main/scala/org/apache/texera/web/service/WorkflowExecutionService.scala

@@ -108,7 +108,12 @@ class WorkflowExecutionService(
   var executionRuntimeService: ExecutionRuntimeService = _
   var executionConsoleService: ExecutionConsoleService = _
 
-  def executeWorkflow(): Unit = {
+  // Returns Some(err) when setup fails synchronously (e.g. compilation) after the
+  // failure has already been surfaced to the UI via errorHandler, so the caller can
+  // mark its still-open setup span; None once the run has started, after which
+  // runtime failures arrive asynchronously through errorHandler and belong to later
+  // execution work rather than this setup.
+  def executeWorkflow(): Option[Throwable] = {
     try {
       val compilationResult = new WorkflowCompiler(workflowContext)
         .compile(request.logicalPlan, CompilationErrorHandling.Strict)
@@ -118,7 +123,7 @@ class WorkflowExecutionService(
         // stop here: `workflow` is still null, so falling through would NPE
         // below and mask the reported compilation error
         errorHandler(err)
-        return
+        return Some(err)
     }
 
     client = ComputingUnitMaster.createAmberRuntime(
@@ -172,6 +177,7 @@ class WorkflowExecutionService(
           }
         )
       )
+    None
   }
 
   override def unsubscribeAll(): Unit = {