Yicong-Huang commented on code in PR #5375: URL: https://github.com/apache/texera/pull/5375#discussion_r3741176508
########## common/observability/src/main/scala/org/apache/texera/observability/TexeraOtelLogAppender.scala: ########## @@ -0,0 +1,127 @@ +/* + * 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 ch.qos.logback.classic.Level +import ch.qos.logback.classic.spi.{ILoggingEvent, IThrowableProxy, ThrowableProxyUtil} +import ch.qos.logback.core.UnsynchronizedAppenderBase +import io.opentelemetry.api.OpenTelemetry +import io.opentelemetry.api.common.AttributeKey +import io.opentelemetry.api.logs.{Logger, Severity} +import io.opentelemetry.api.trace.Span +import io.opentelemetry.context.Context + +import java.util.concurrent.TimeUnit + +/** + * Logback appender that sanitizes each event via [[LogSanitizer]] and + * emits it as an OTel LogRecord. [[append]] is a no-op until [[bind]] + * is called and after [[stop]]. + * + * This is internal plumbing, not the developer logging API. Code logs + * through the normal SLF4J / scala-logging interface and adds correlation + * ids via MDC; [[OtelInit.init]] attaches this appender to the ROOT logger + * so those records also reach OTel: + * + * {{{ + * class Foo extends LazyLogging { + * MDC.put("workflowId", id) // forwarded as an OTel log attribute + * try logger.info("started") // body + severity + trace context + * finally MDC.remove("workflowId") + * } + * }}} + */ +class TexeraOtelLogAppender extends UnsynchronizedAppenderBase[ILoggingEvent] { + + // @volatile so a late bind() is visible to appender threads. + @volatile private var otelLogger: Option[Logger] = None + + def bind(otel: OpenTelemetry): Unit = { + otelLogger = Some(otel.getLogsBridge.get("texera.logback")) + } + + override def stop(): Unit = { + otelLogger = None + super.stop() + } + + override def append(event: ILoggingEvent): Unit = { + otelLogger match { + case None => () // not bound + case Some(logger) => + try { + emit(logger, event) + } catch { + // An appender must not throw into the calling thread. + case t: Throwable => + addError("OTel log emission failed", t) + } + } + } + + private def emit(logger: Logger, event: ILoggingEvent): Unit = { + // Append the stack trace to the body when a throwable is attached. + val baseBody = LogSanitizer.sanitize(event.getFormattedMessage) + val body = Option(event.getThrowableProxy) match { + case Some(proxy) => + // Skip the C0 strip so trace newlines survive, but still cap. + LogSanitizer.truncate(baseBody + "\n" + formatThrowable(proxy)) Review Comment: The stack trace never sees secret redaction. `truncate` only length-caps; the `SecretPatterns` regexes live inside `sanitize` (LogSanitizer.scala:71). A `Bearer` token or a `password=` JDBC URL in an exception message reaches the collector verbatim, though the same string in a plain log line is redacted. The comment above justifies skipping the C0 strip; redaction is dropped with it. Splitting `sanitize` into `redactSecrets` and `stripControlChars` would let you redact the whole body and control-strip only the message. ########## common/observability/src/main/scala/org/apache/texera/observability/OtelInit.scala: ########## @@ -0,0 +1,382 @@ +/* + * 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 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. + * + * Enabled by default; set OTEL_SDK_DISABLED=true to turn it off. Reads + * OTEL_* env vars, 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. */ + private[observability] val AllowedSchemes: Set[String] = Set("http", "https", "grpc") + + /** Hosts we accept for the OTLP endpoint by default. */ + private[observability] val DefaultAllowedHosts: Set[String] = Set( + "localhost", + "127.0.0.1", + "::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 clamp 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 + + val env = (key: String) => Option(System.getenv(key)) + 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 + } + + /** + * 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 + + // Enabled by default; OTEL_SDK_DISABLED=true opts out. An + // unreachable endpoint drops records without crashing the service. + val disabled = envProvider("OTEL_SDK_DISABLED").getOrElse("false") Review Comment: Defaulting to enabled means every deployment starts exporters, yet none of the five OTEL_* knobs reaches a place an operator would look. Compare `storage.jdbc.url`: HOCON default (storage.conf:156), `${?STORAGE_JDBC_URL}` override (:157), name constant (EnvironmentalVariable.scala:44), plus values in `bin/single-node/.env` and `bin/k8s/values.yaml`. Grepping `OTEL_`/`TEXERA_OTEL_` across all `.conf`, `.yaml`, `.env` and dockerfiles returns nothing outside `common/observability` — so a shipped deployment exports to `127.0.0.1:4317` with no collector and no way to silence it. I'd route these through HOCON with `${?...}` overrides and add them to both deploy files. ########## common/observability/src/main/scala/org/apache/texera/observability/OtelInit.scala: ########## @@ -0,0 +1,382 @@ +/* + * 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 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. + * + * Enabled by default; set OTEL_SDK_DISABLED=true to turn it off. Reads + * OTEL_* env vars, 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. */ + private[observability] val AllowedSchemes: Set[String] = Set("http", "https", "grpc") Review Comment: `grpc` here aborts service startup. In the pinned 1.50.0 jars, `OtlpGrpcSpanExporterBuilder.setEndpoint` delegates to `ExporterBuilderUtil.validateEndpoint`, which throws `IllegalArgumentException` for any scheme other than http/https. The builder call on line 184 sits outside any `Try`, so a `grpc://` endpoint escapes `init()` into Dropwizard's `run()` — contradicting the "never throws" contract on line 47. OTLP over gRPC uses `http://` endpoints anyway. `OtelInitSpec.scala:45` asserts this scheme is valid and needs updating too. ```suggestion private[observability] val AllowedSchemes: Set[String] = Set("http", "https") ``` ########## common/observability/src/test/scala/org/apache/texera/observability/TexeraOtelLogAppenderSpec.scala: ########## @@ -0,0 +1,173 @@ +/* + * 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 ch.qos.logback.classic.{Level, Logger, LoggerContext} +import ch.qos.logback.classic.spi.LoggingEvent +import io.opentelemetry.api.OpenTelemetry +import io.opentelemetry.api.logs.Severity +import io.opentelemetry.sdk.OpenTelemetrySdk +import io.opentelemetry.sdk.logs.SdkLoggerProvider +import io.opentelemetry.sdk.logs.`export`.SimpleLogRecordProcessor +import io.opentelemetry.sdk.testing.exporter.InMemoryLogRecordExporter +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import org.slf4j.LoggerFactory + +import scala.jdk.CollectionConverters._ + +class TexeraOtelLogAppenderSpec extends AnyFlatSpec with Matchers { + + /** Build an OpenTelemetry SDK whose LoggerProvider drains to the + * given in-memory exporter via the synchronous SimpleLogRecordProcessor, + * so tests don't depend on batch timing. + */ + private def newFixture(): (OpenTelemetry, InMemoryLogRecordExporter, TexeraOtelLogAppender) = { + val exporter = InMemoryLogRecordExporter.create() + val lp = SdkLoggerProvider + .builder() + .addLogRecordProcessor(SimpleLogRecordProcessor.create(exporter)) + .build() + val sdk = OpenTelemetrySdk.builder().setLoggerProvider(lp).build() + val appender = new TexeraOtelLogAppender() + appender.setContext(LoggerFactory.getILoggerFactory.asInstanceOf[LoggerContext]) + appender.bind(sdk) + appender.start() + (sdk, exporter, appender) + } + + private def makeEvent( + message: String, + level: Level = Level.INFO, + mdc: Map[String, String] = Map.empty + ): LoggingEvent = { + val ctx = LoggerFactory.getILoggerFactory.asInstanceOf[LoggerContext] + val logger = ctx.getLogger("test.logger").asInstanceOf[Logger] + val ev = new LoggingEvent("fqcn", logger, level, message, null, null) + if (mdc.nonEmpty) ev.setMDCPropertyMap(mdc.asJava) + ev + } + + // ----- positive paths ------------------------------------------------- + + "TexeraOtelLogAppender" should "emit an INFO record with body + severity" in { + val (_, exporter, appender) = newFixture() + appender.doAppend(makeEvent("hello world")) + + val records = exporter.getFinishedLogRecordItems.asScala + records should have size 1 + records.head.getBodyValue.asString shouldBe "hello world" + records.head.getSeverity shouldBe Severity.INFO + records.head.getSeverityText shouldBe "INFO" + } + + it should "map every log level to a distinct OTel severity" in { + val (_, exporter, appender) = newFixture() + Seq(Level.TRACE, Level.DEBUG, Level.INFO, Level.WARN, Level.ERROR).foreach { lvl => + appender.doAppend(makeEvent(s"msg-$lvl", lvl)) + } + val severities = exporter.getFinishedLogRecordItems.asScala.map(_.getSeverity).toSet + severities shouldBe Set( + Severity.TRACE, + Severity.DEBUG, + Severity.INFO, + Severity.WARN, + Severity.ERROR + ) + } + + // ----- security: sanitisation happens at the boundary ----------------- + + it should "strip CRLF from a forged log-injection payload before emission" in { + val (_, exporter, appender) = newFixture() + appender.doAppend(makeEvent("hello\r\nFAKE LOG LINE\r\nworld")) + + val body = exporter.getFinishedLogRecordItems.asScala.head.getBodyValue.asString + body shouldBe "helloFAKE LOG LINEworld" + body should not include "\n" + body should not include "\r" + } + + it should "redact Bearer tokens at emission time" in { + val (_, exporter, appender) = newFixture() + appender.doAppend(makeEvent("Authorization: Bearer abc123XYZ.foo")) + + val body = exporter.getFinishedLogRecordItems.asScala.head.getBodyValue.asString + body should include("[REDACTED]") + body should not include "abc123XYZ" + } + + it should "truncate a 1 MiB body to MaxBodyBytes with the marker" in { + val (_, exporter, appender) = newFixture() + val oversize = "x" * (1024 * 1024) + appender.doAppend(makeEvent(oversize)) + + val body = exporter.getFinishedLogRecordItems.asScala.head.getBodyValue.asString + body.length shouldBe LogSanitizer.MaxBodyBytes + body should endWith(LogSanitizer.TruncatedMarker) + } + + // ----- security: MDC allowlist ---------------------------------------- + + it should "forward only allowlisted MDC keys as log attributes" in { + val (_, exporter, appender) = newFixture() + appender.doAppend( + makeEvent( + "msg", + mdc = Map( + "trace_id" -> "abc", + "texera.workflow.id" -> "42", + "secret" -> "should-not-leak", + "password" -> "p4ssw0rd" + ) + ) + ) + + val record = exporter.getFinishedLogRecordItems.asScala.head + val attrs = record.getAttributes.asMap.asScala.iterator.map { + case (k, v) => k.getKey -> v.toString + }.toMap + + attrs.keySet should contain allOf ("trace_id", "texera.workflow.id") + attrs.keySet should not contain ("secret") Review Comment: This test fails on the current head. I ran it: `Set("secret", ..., "password", ...) contained element "secret"` — 7 passed, 1 failed. The allowlist-to-deny-list switch updated `LogSanitizerSpec` but not this one. Worth not fixing by relaxing the assertion. `filterMdc` (LogSanitizer.scala:87) matches keys against five Pekko names only. Redaction inspects value text, so `MDC.put("password", pw)` is exported intact — the opposite of LogSanitizer's Scaladoc. A key-name check in `filterMdc` fixes the behavior and makes this test pass as written. ########## common/observability/src/main/scala/org/apache/texera/observability/OtelInit.scala: ########## @@ -0,0 +1,382 @@ +/* + * 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 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. + * + * Enabled by default; set OTEL_SDK_DISABLED=true to turn it off. Reads + * OTEL_* env vars, 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. */ + private[observability] val AllowedSchemes: Set[String] = Set("http", "https", "grpc") + + /** Hosts we accept for the OTLP endpoint by default. */ + private[observability] val DefaultAllowedHosts: Set[String] = Set( + "localhost", + "127.0.0.1", + "::1", Review Comment: Unreachable, so worth deleting. `URI.getHost` returns `[::1]` for the bracketed form and `null` for `http://::1:4317` (verified under JDK 17), so this entry can never match the value compared on line 273. The `[::1]` on the next line already covers IPv6 loopback. ########## common/observability/src/main/scala/org/apache/texera/observability/OtelInit.scala: ########## @@ -0,0 +1,382 @@ +/* + * 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 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. + * + * Enabled by default; set OTEL_SDK_DISABLED=true to turn it off. Reads + * OTEL_* env vars, 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. */ + private[observability] val AllowedSchemes: Set[String] = Set("http", "https", "grpc") + + /** Hosts we accept for the OTLP endpoint by default. */ + private[observability] val DefaultAllowedHosts: Set[String] = Set( + "localhost", + "127.0.0.1", + "::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 clamp to the Review Comment: "clamp" reads as coercion to the nearest bound, but `clampIntervalMs` discards the value and substitutes the default (:341-347) — as its own comment on :328 already says. The same wording is on :204. ```suggestion /** Metric export interval bounds; out-of-range values fall back to the ``` ########## common/observability/src/main/scala/org/apache/texera/observability/LogSanitizer.scala: ########## @@ -0,0 +1,90 @@ +/* + * 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 scala.jdk.CollectionConverters._ + +/** + * Pure functions that sanitize log bodies and MDC before export: + * strip control characters, redact secrets, cap body size, and + * filter MDC down by dropping denied keys. + */ +object LogSanitizer { + + /** Per-record body byte cap. */ Review Comment: This counts UTF-16 chars, not bytes — `truncate` compares `body.length` (:79). For non-ASCII bodies the encoded size can be several times the stated 16 KiB budget. ```suggestion /** Per-record body length cap, in chars. */ ``` -- 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]
