Ma77Ball commented on code in PR #5375: URL: https://github.com/apache/texera/pull/5375#discussion_r4358258318
########## amber/src/main/scala/org/apache/texera/web/observability/RequestContextMdcFilter.scala: ########## @@ -0,0 +1,165 @@ +/* + * 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 org.slf4j.MDC + +import javax.servlet._ +import javax.servlet.http.HttpServletRequest + +/** + * Servlet filter that pushes request-scoped IDs into the SLF4J MDC + * so every log record emitted while handling a request carries + * `texera.workflow.id`, `texera.execution.id`, and + * `texera.computing_unit.id` when those are visible in the URL or + * request headers. + * + * Without this, the OTel log appender bridges every Logback event + * with no per-request context, so the dashboard's CU/workflow + * filters match nothing live: the only records carrying those keys + * are seed data pushed via the VictoriaLogs ingest API. + * + * Sources of IDs, in priority order: + * 1. HTTP headers `X-Texera-Workflow-Id`, `X-Texera-Execution-Id`, Review Comment: Removed the filters, so the unverified note about the Angular client sending those headers is gone. ########## amber/src/main/scala/org/apache/texera/web/observability/RequestContextMdcFilter.scala: ########## @@ -0,0 +1,165 @@ +/* + * 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 org.slf4j.MDC + +import javax.servlet._ +import javax.servlet.http.HttpServletRequest + +/** + * Servlet filter that pushes request-scoped IDs into the SLF4J MDC + * so every log record emitted while handling a request carries + * `texera.workflow.id`, `texera.execution.id`, and + * `texera.computing_unit.id` when those are visible in the URL or + * request headers. + * + * Without this, the OTel log appender bridges every Logback event + * with no per-request context, so the dashboard's CU/workflow + * filters match nothing live: the only records carrying those keys + * are seed data pushed via the VictoriaLogs ingest API. + * + * Sources of IDs, in priority order: + * 1. HTTP headers `X-Texera-Workflow-Id`, `X-Texera-Execution-Id`, + * `X-Texera-Computing-Unit-Id`: set explicitly by the + * Angular client when known. + * 2. URL path segments matching `/workflow/<digits>`, + * `/execution/<digits>`, `/computing-unit/<digits>` (also the + * `/{wid}` / `/{cuid}` patterns used by some resources). + * + * MDC is ALWAYS cleared in a `finally` block so a thread reused for + * a different request doesn't leak the previous request's labels. + * + * Keys here must stay out of [[LogSanitizer.DeniedMdcKeys]]: the + * sanitizer is default-allow, so any key added here is exported unless + * it is denied there, in which case the OTel appender drops it. + */ +class RequestContextMdcFilter extends Filter { + + override def init(filterConfig: FilterConfig): Unit = () + override def destroy(): Unit = () + + override def doFilter( + request: ServletRequest, + response: ServletResponse, + chain: FilterChain + ): Unit = { + val pushed = scala.collection.mutable.ArrayBuffer.empty[String] + request match { + case http: HttpServletRequest => + // Header overrides come first: an explicit client header is + // the most precise signal we have. + applyHeader(http, "X-Texera-Workflow-Id", "texera.workflow.id", pushed) + applyHeader(http, "X-Texera-Execution-Id", "texera.execution.id", pushed) + applyHeader(http, "X-Texera-Computing-Unit-Id", "texera.computing_unit.id", pushed) + + // URL extraction: only fill keys that weren't already set + // by headers. + val path = Option(http.getRequestURI).getOrElse("") + applyPath(path, RequestContextMdcFilter.WorkflowPattern, "texera.workflow.id", pushed) + applyPath(path, RequestContextMdcFilter.ExecutionPattern, "texera.execution.id", pushed) + applyPath(path, RequestContextMdcFilter.CuPattern, "texera.computing_unit.id", pushed) + + // Query-parameter extraction: the WebSocket upgrade URL is + // `/wsapi/workflow-websocket?wid=<w>&cuid=<c>` and Texera + // also accepts `?cuid=N` on the REST PVE endpoints. Filling + // the MDC here means every record emitted during the WS + // handshake AND the long-running session that follows + // carries the ids the client supplied. + applyParam(http, "wid", "texera.workflow.id", pushed) + applyParam(http, "eid", "texera.execution.id", pushed) + applyParam(http, "cuid", "texera.computing_unit.id", pushed) + case _ => + // Non-HTTP request (websocket upgrades fall through here on Review Comment: Removed the filters, so the inaccurate note about websocket requests is gone. ########## amber/src/main/scala/org/apache/texera/web/observability/RequestContextMdcFilter.scala: ########## @@ -0,0 +1,165 @@ +/* + * 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 org.slf4j.MDC + +import javax.servlet._ +import javax.servlet.http.HttpServletRequest + +/** + * Servlet filter that pushes request-scoped IDs into the SLF4J MDC + * so every log record emitted while handling a request carries + * `texera.workflow.id`, `texera.execution.id`, and + * `texera.computing_unit.id` when those are visible in the URL or + * request headers. + * + * Without this, the OTel log appender bridges every Logback event + * with no per-request context, so the dashboard's CU/workflow + * filters match nothing live: the only records carrying those keys + * are seed data pushed via the VictoriaLogs ingest API. + * + * Sources of IDs, in priority order: + * 1. HTTP headers `X-Texera-Workflow-Id`, `X-Texera-Execution-Id`, + * `X-Texera-Computing-Unit-Id`: set explicitly by the + * Angular client when known. + * 2. URL path segments matching `/workflow/<digits>`, + * `/execution/<digits>`, `/computing-unit/<digits>` (also the + * `/{wid}` / `/{cuid}` patterns used by some resources). + * + * MDC is ALWAYS cleared in a `finally` block so a thread reused for + * a different request doesn't leak the previous request's labels. + * + * Keys here must stay out of [[LogSanitizer.DeniedMdcKeys]]: the + * sanitizer is default-allow, so any key added here is exported unless + * it is denied there, in which case the OTel appender drops it. + */ +class RequestContextMdcFilter extends Filter { + + override def init(filterConfig: FilterConfig): Unit = () + override def destroy(): Unit = () + + override def doFilter( + request: ServletRequest, + response: ServletResponse, + chain: FilterChain + ): Unit = { + val pushed = scala.collection.mutable.ArrayBuffer.empty[String] + request match { + case http: HttpServletRequest => + // Header overrides come first: an explicit client header is + // the most precise signal we have. + applyHeader(http, "X-Texera-Workflow-Id", "texera.workflow.id", pushed) + applyHeader(http, "X-Texera-Execution-Id", "texera.execution.id", pushed) + applyHeader(http, "X-Texera-Computing-Unit-Id", "texera.computing_unit.id", pushed) + + // URL extraction: only fill keys that weren't already set + // by headers. + val path = Option(http.getRequestURI).getOrElse("") + applyPath(path, RequestContextMdcFilter.WorkflowPattern, "texera.workflow.id", pushed) + applyPath(path, RequestContextMdcFilter.ExecutionPattern, "texera.execution.id", pushed) + applyPath(path, RequestContextMdcFilter.CuPattern, "texera.computing_unit.id", pushed) + + // Query-parameter extraction: the WebSocket upgrade URL is + // `/wsapi/workflow-websocket?wid=<w>&cuid=<c>` and Texera + // also accepts `?cuid=N` on the REST PVE endpoints. Filling + // the MDC here means every record emitted during the WS + // handshake AND the long-running session that follows + // carries the ids the client supplied. + applyParam(http, "wid", "texera.workflow.id", pushed) + applyParam(http, "eid", "texera.execution.id", pushed) + applyParam(http, "cuid", "texera.computing_unit.id", pushed) + case _ => + // Non-HTTP request (websocket upgrades fall through here on + // some Servlet versions). No MDC context to add. + } + try { + chain.doFilter(request, response) + } finally { + // Defence in depth: clear only what THIS filter set so we don't + // step on MDC populated by other middleware. + pushed.foreach(MDC.remove) + } + } + + private def applyHeader( + req: HttpServletRequest, + headerName: String, + mdcKey: String, + pushed: scala.collection.mutable.ArrayBuffer[String] + ): Unit = { + val raw = Option(req.getHeader(headerName)).map(_.trim).filter(_.nonEmpty) + raw.foreach { value => + // Cheap allowlist: only digits get through. Prevents log + // forging via a CRLF in the header value. + if (value.forall(_.isDigit) && value.length <= 19) { + MDC.put(mdcKey, value) + pushed += mdcKey + } + } + } + + private def applyPath( + path: String, + pattern: scala.util.matching.Regex, + mdcKey: String, + pushed: scala.collection.mutable.ArrayBuffer[String] + ): Unit = { + if (MDC.get(mdcKey) == null) { + pattern.findFirstMatchIn(path).foreach { m => + val v = m.group(1) + MDC.put(mdcKey, v) + pushed += mdcKey + } + } + } + + /** Read a query parameter, validate it's a positive integer, push + * to MDC. Same shape as applyHeader: numeric-only allowlist and a + * length cap so a forged value can't inject newlines into the + * rendered log line. + */ + private def applyParam( + req: HttpServletRequest, + paramName: String, + mdcKey: String, + pushed: scala.collection.mutable.ArrayBuffer[String] + ): Unit = { + if (MDC.get(mdcKey) == null) { + val raw = Option(req.getParameter(paramName)).map(_.trim).filter(_.nonEmpty) + raw.foreach { value => + if (value.forall(_.isDigit) && value.length <= 19) { + MDC.put(mdcKey, value) + pushed += mdcKey + } + } + } + } +} + +object RequestContextMdcFilter { Review Comment: Removed the filters, so the repeated key strings are gone. ########## common/observability/src/main/scala/org/apache/texera/observability/OtelInit.scala: ########## @@ -0,0 +1,418 @@ +/* + * 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.observability + +import com.typesafe.scalalogging.LazyLogging +import org.apache.texera.common.config.{EnvironmentalVariable, ObservabilityConfig} +import io.opentelemetry.api.{GlobalOpenTelemetry, OpenTelemetry} +import io.opentelemetry.api.common.{AttributeKey, Attributes} +import io.opentelemetry.exporter.otlp.logs.OtlpGrpcLogRecordExporter +import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporter +import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter +import io.opentelemetry.sdk.OpenTelemetrySdk +import io.opentelemetry.sdk.logs.SdkLoggerProvider +import io.opentelemetry.sdk.logs.`export`.{BatchLogRecordProcessor, LogRecordExporter} +import io.opentelemetry.sdk.metrics.SdkMeterProvider +import io.opentelemetry.sdk.metrics.`export`.{MetricExporter, PeriodicMetricReader} +import io.opentelemetry.sdk.resources.Resource +import io.opentelemetry.sdk.trace.SdkTracerProvider +import io.opentelemetry.sdk.trace.`export`.{BatchSpanProcessor, SpanExporter} + +import java.net.URI +import java.time.Duration +import scala.util.{Failure, Success, Try} + +/** + * Bootstraps the OpenTelemetry SDK for a Texera service. + * + * Disabled by default; set OTEL_SDK_DISABLED=false to enable it. Reads its + * settings from observability.conf (each defaulted, each OTEL_*-overridable), + * validates the endpoint against an allowlist, builds tracer/log/metric + * providers, and attaches a Logback appender. Returns None when disabled or + * misconfigured; never throws. + */ +object OtelInit extends LazyLogging { + + /** Endpoint schemes we accept. OTLP-over-gRPC uses http/https endpoints; + * the exporter rejects a `grpc://` scheme outright, so it is not allowed. + */ + private[observability] val AllowedSchemes: Set[String] = Set("http", "https") + + /** Hosts we accept for the OTLP endpoint by default. */ + private[observability] val DefaultAllowedHosts: Set[String] = Set( + "localhost", + "127.0.0.1", + "[::1]" + ) + + /** Default endpoint. 127.0.0.1 (not "localhost") to force IPv4 so a + * natively-run service reaches the collector on dual-stack hosts. + */ + private val DefaultEndpoint = "http://127.0.0.1:4317"; + + /** Metric export interval bounds; out-of-range values fall back to the + * default (see clampIntervalMs). + */ + private[observability] val MinMetricIntervalMs: Long = 1000L + private[observability] val MaxMetricIntervalMs: Long = 10L * 60L * 1000L + private[observability] val DefaultMetricIntervalMs: Long = 30L * 1000L + + // Idempotency guard: init() is a no-op after the first call. + @volatile private var initialized: Option[OpenTelemetry] = None + + /** + * Initialize the SDK for the given service name. Returns Some on + * success, None when disabled or misconfigured. When enabled, also + * attaches a [[TexeraOtelLogAppender]] to the Logback ROOT logger. + */ + def init(serviceName: String): Option[OpenTelemetry] = + synchronized { + if (initialized.isDefined) return initialized + + // Guard the whole body: reading observability.conf (ObservabilityConfig's + // eager vals) and building the SDK are the fallible steps. A missing or + // malformed config would otherwise escape as ExceptionInInitializerError, + // even when telemetry is disabled, breaking init()'s "never throws" + // contract. On any failure, disable telemetry with one WARN and return None. + Try { + // Source the OTEL_* settings from observability.conf (HOCON defaults + // already merged with any env override); fall back to the raw environment + // for anything else. + val env = (key: String) => + key match { + case EnvironmentalVariable.ENV_OTEL_SDK_DISABLED => + Some(ObservabilityConfig.sdkDisabled) + case EnvironmentalVariable.ENV_OTEL_EXPORTER_OTLP_ENDPOINT => + Some(ObservabilityConfig.endpoint) + case EnvironmentalVariable.ENV_OTEL_RESOURCE_ATTRIBUTES => + Some(ObservabilityConfig.resourceAttributes) + case EnvironmentalVariable.ENV_TEXERA_OTEL_ALLOWED_HOSTS => + Some(ObservabilityConfig.allowedHosts) + case EnvironmentalVariable.ENV_OTEL_METRIC_EXPORT_INTERVAL => + Some(ObservabilityConfig.metricExportIntervalMs) + case other => Option(System.getenv(other)) + } + val result = initInternal( + serviceName = serviceName, + envProvider = env, + spanExporterFactory = buildOtlpSpanExporter, + logExporterFactory = endpoint => Some(buildOtlpLogExporter(endpoint)), + metricExporterFactory = endpoint => Some(buildOtlpMetricExporter(endpoint)), + logbackAttacher = LogbackBinder.attach + ) + // Register globally so OTel-aware code can use GlobalOpenTelemetry + // without threading the SDK through callsites. set() throws on a + // second call; wrap defensively. + result.foreach { sdk => + Try(GlobalOpenTelemetry.set(sdk)).failed.foreach { t => + logger.warn( + s"GlobalOpenTelemetry already set; using the existing instance: ${t.getMessage}" + ) + } + } + result + } match { + case Success(result) => result + case Failure(t) => + logger.warn( + s"OpenTelemetry SDK initialization failed; continuing without telemetry: ${t.getMessage}" + ) + None + } + } + + /** + * Test-only entry point: injects an env-var map and exporters so the + * SDK makes no network connection. Does not attach the Logback appender. + */ + private[observability] def initForTest( + serviceName: String, + envOverride: Map[String, String], + exporter: SpanExporter, + metricExporter: Option[MetricExporter] = None + ): Option[OpenTelemetry] = + synchronized { + initInternal( + serviceName = serviceName, + envProvider = envOverride.get, + spanExporterFactory = _ => exporter, + logExporterFactory = _ => None, + metricExporterFactory = _ => metricExporter, + logbackAttacher = (_, _) => () // no-op in tests + ) + } + + /** Test-only: forget any previously-installed SDK. Does not unregister + * shutdown hooks (the previous SDK is closed instead). + */ + private[observability] def resetForTest(): Unit = + synchronized { + initialized.foreach { + case sdk: OpenTelemetrySdk => + Try(sdk.getSdkTracerProvider.close()) + Try(sdk.getSdkLoggerProvider.close()) + Try(sdk.getSdkMeterProvider.close()) + case _ => () + } + initialized = None + } + + private def initInternal( + serviceName: String, + envProvider: String => Option[String], + spanExporterFactory: String => SpanExporter, + logExporterFactory: String => Option[LogRecordExporter], + metricExporterFactory: String => Option[MetricExporter], + logbackAttacher: (String, OpenTelemetry) => Unit + ): Option[OpenTelemetry] = { + if (initialized.isDefined) return initialized + + // Disabled by default (issue #5367): stay inert unless OTEL_SDK_DISABLED is + // explicitly false. An unreachable endpoint drops records without crashing. + val disabled = envProvider(EnvironmentalVariable.ENV_OTEL_SDK_DISABLED).getOrElse("true") + if (!disabled.equalsIgnoreCase("false")) { + logger.info( + "OpenTelemetry SDK disabled (OTEL_SDK_DISABLED not false). No telemetry will be emitted." + ) + return None + } + + val endpoint = + envProvider(EnvironmentalVariable.ENV_OTEL_EXPORTER_OTLP_ENDPOINT).getOrElse(DefaultEndpoint) + val extraAllowed = envProvider(EnvironmentalVariable.ENV_TEXERA_OTEL_ALLOWED_HOSTS) + .map(_.split(',').iterator.map(_.trim.toLowerCase).filter(_.nonEmpty).toSet) + .getOrElse(Set.empty) + val allowedHosts = DefaultAllowedHosts ++ extraAllowed + + validateEndpoint(endpoint, allowedHosts) match { + case Left(reason) => + // One WARN; no telemetry is emitted. + logger.warn( + s"OpenTelemetry SDK disabled: invalid OTEL_EXPORTER_OTLP_ENDPOINT — $reason. " + + "Set TEXERA_OTEL_ALLOWED_HOSTS to extend the allowlist." + ) + return None + case Right(_) => // ok + } + + val rawAttrs = envProvider(EnvironmentalVariable.ENV_OTEL_RESOURCE_ATTRIBUTES).getOrElse("") + val resource = buildResource(serviceName, rawAttrs) + + val spanExporter = spanExporterFactory(endpoint) + val tracerProvider = SdkTracerProvider + .builder() + .setResource(resource) + .addSpanProcessor(BatchSpanProcessor.builder(spanExporter).build()) + .build() + + val sdkBuilder = OpenTelemetrySdk.builder().setTracerProvider(tracerProvider) + + // Logger provider is optional; the factory returns None in tests. + val loggerProviderOpt = logExporterFactory(endpoint).map { logExporter => + val lp = SdkLoggerProvider + .builder() + .setResource(resource) + .addLogRecordProcessor(BatchLogRecordProcessor.builder(logExporter).build()) + .build() + sdkBuilder.setLoggerProvider(lp) + lp + } + + // Meter provider is optional too; interval falls back to the default + // when out of range. + val intervalMs = + clampIntervalMs(envProvider(EnvironmentalVariable.ENV_OTEL_METRIC_EXPORT_INTERVAL)) + val meterProviderOpt = metricExporterFactory(endpoint).map { metricExporter => + val reader = PeriodicMetricReader + .builder(metricExporter) + .setInterval(Duration.ofMillis(intervalMs)) + .build() + val mp = SdkMeterProvider + .builder() + .setResource(resource) + .registerMetricReader(reader) + .build() + sdkBuilder.setMeterProvider(mp) + mp + } + + val sdk = sdkBuilder.build() + + // One startup span carrying only service.name. + val span = sdk.getTracer("texera.bootstrap").spanBuilder("service.start").startSpan() + Try(span.setAttribute("service.name", serviceName)) + span.end() + + // Wire the Logback appender; failure here must not crash the service. + Try(logbackAttacher(serviceName, sdk)).failed.foreach { t => + logger.warn(s"Failed to attach OTel Logback appender (logs not exported): ${t.getMessage}") + } + + // Flush providers on shutdown. Added after the SDK is fully built. + Runtime.getRuntime.addShutdownHook( + new Thread( + () => { + Try(tracerProvider.close()) + loggerProviderOpt.foreach(lp => Try(lp.close())) + meterProviderOpt.foreach(mp => Try(mp.close())) + () + }, + "otel-shutdown" + ) + ) + + initialized = Some(sdk) + logger.info(s"OpenTelemetry SDK initialized for service '$serviceName' (endpoint=$endpoint).") + initialized + } + + /** + * Validate the endpoint is parseable and uses an allowlisted scheme + * and host. Pure function. + */ + private[observability] def validateEndpoint( + endpoint: String, + allowedHosts: Set[String] + ): Either[String, Unit] = { + Try(URI.create(endpoint)) match { + case Failure(e) => + Left(s"unparseable URI (${e.getClass.getSimpleName})") + case Success(uri) => + val scheme = Option(uri.getScheme).map(_.toLowerCase).getOrElse("") + if (scheme.isEmpty) { + Left("missing scheme") + } else if (!AllowedSchemes.contains(scheme)) { + Left( + s"scheme '$scheme' not in allowlist ${AllowedSchemes.toSeq.sorted.mkString("{", ",", "}")}" + ) + } else { + val host = Option(uri.getHost).map(_.toLowerCase).getOrElse("") + if (host.isEmpty) { + Left("missing host") + } else if (!allowedHosts.contains(host)) { + Left(s"host '$host' not in allowlist") + } else { + Right(()) + } + } + } + } + + /** + * Build a Resource from the service name and OTEL_RESOURCE_ATTRIBUTES. + * Every parsed attribute is applied so new resource fields need no edit + * here; the one exception is service.name, which the argument controls + * and env cannot override. + */ + private[observability] def buildResource(serviceName: String, rawAttrs: String): Resource = { + val builder = Attributes.builder() + builder.put(AttributeKey.stringKey("service.name"), serviceName) Review Comment: Added a unique id for each process (a random value by default that can be set through OTEL_RESOURCE_ATTRIBUTES), so computing units that share the same service name no longer blur together in logs, metrics, and traces, with two tests covering it. The other attributes you listed (namespace, version, and the Kubernetes fields) are a good next step and will come where the computing-unit resource is built. ########## amber/src/main/scala/org/apache/texera/web/observability/RequestContextMdcFilter.scala: ########## @@ -0,0 +1,165 @@ +/* + * 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 org.slf4j.MDC + +import javax.servlet._ +import javax.servlet.http.HttpServletRequest + +/** + * Servlet filter that pushes request-scoped IDs into the SLF4J MDC + * so every log record emitted while handling a request carries + * `texera.workflow.id`, `texera.execution.id`, and + * `texera.computing_unit.id` when those are visible in the URL or + * request headers. + * + * Without this, the OTel log appender bridges every Logback event + * with no per-request context, so the dashboard's CU/workflow + * filters match nothing live: the only records carrying those keys + * are seed data pushed via the VictoriaLogs ingest API. + * + * Sources of IDs, in priority order: + * 1. HTTP headers `X-Texera-Workflow-Id`, `X-Texera-Execution-Id`, + * `X-Texera-Computing-Unit-Id`: set explicitly by the + * Angular client when known. + * 2. URL path segments matching `/workflow/<digits>`, + * `/execution/<digits>`, `/computing-unit/<digits>` (also the + * `/{wid}` / `/{cuid}` patterns used by some resources). + * + * MDC is ALWAYS cleared in a `finally` block so a thread reused for + * a different request doesn't leak the previous request's labels. + * + * Keys here must stay out of [[LogSanitizer.DeniedMdcKeys]]: the + * sanitizer is default-allow, so any key added here is exported unless + * it is denied there, in which case the OTel appender drops it. + */ +class RequestContextMdcFilter extends Filter { + + override def init(filterConfig: FilterConfig): Unit = () + override def destroy(): Unit = () + + override def doFilter( + request: ServletRequest, + response: ServletResponse, + chain: FilterChain + ): Unit = { + val pushed = scala.collection.mutable.ArrayBuffer.empty[String] + request match { + case http: HttpServletRequest => + // Header overrides come first: an explicit client header is + // the most precise signal we have. + applyHeader(http, "X-Texera-Workflow-Id", "texera.workflow.id", pushed) + applyHeader(http, "X-Texera-Execution-Id", "texera.execution.id", pushed) + applyHeader(http, "X-Texera-Computing-Unit-Id", "texera.computing_unit.id", pushed) + + // URL extraction: only fill keys that weren't already set + // by headers. + val path = Option(http.getRequestURI).getOrElse("") + applyPath(path, RequestContextMdcFilter.WorkflowPattern, "texera.workflow.id", pushed) + applyPath(path, RequestContextMdcFilter.ExecutionPattern, "texera.execution.id", pushed) + applyPath(path, RequestContextMdcFilter.CuPattern, "texera.computing_unit.id", pushed) + + // Query-parameter extraction: the WebSocket upgrade URL is + // `/wsapi/workflow-websocket?wid=<w>&cuid=<c>` and Texera + // also accepts `?cuid=N` on the REST PVE endpoints. Filling + // the MDC here means every record emitted during the WS Review Comment: Removed both filters. The workflow and execution ids need to follow the run into its later websocket messages and background callbacks, which a request filter cannot do, so that will be handled where the execution runs in a later change. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
