This is an automated email from the ASF dual-hosted git repository. wusheng pushed a commit to branch feature/lal-logmetadata-refactoring in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit 76bd5845cb2869ae299f720dca0c00acd4b2e652 Author: Wu Sheng <[email protected]> AuthorDate: Thu Mar 12 21:17:50 2026 +0800 LAL: introduce LogMetadata to decouple metadata from LogData in pipeline entry Separate metadata (service, layer, timestamp, trace context) from input data (LogData body/tags, typed protos) in the LAL pipeline. Receivers now construct a LogMetadata POJO instead of relying on LogData for metadata fields, and Envoy ALS no longer needs a fake LogData. The ExecutionContext carries metadata via ctx.metadata() and the raw input via ctx.input(), removing extraLog() indirection and LogData type awareness from runtime. The test tool (LogTestQuery) is separated from production sink via dryRun flag instead of capture/logContainer mixing. --- .../listener/DatabaseSlowStatementBuilder.java | 18 ++-- .../trace/parser/listener/SampledTraceBuilder.java | 22 +++-- oap-server/analyzer/log-analyzer/CLAUDE.md | 27 +++--- .../log/analyzer/v2/compiler/LALBlockCodegen.java | 78 ++++++++++++---- .../analyzer/v2/compiler/LALClassGenerator.java | 2 +- .../log/analyzer/v2/compiler/LALCodegenHelper.java | 34 ++++--- .../analyzer/v2/compiler/rt/LalRuntimeHelper.java | 15 ++- .../skywalking/oap/log/analyzer/v2/dsl/DSL.java | 10 +- .../oap/log/analyzer/v2/dsl/ExecutionContext.java | 66 ++++++-------- .../analyzer/v2/dsl/spec/filter/FilterSpec.java | 67 ++++++-------- .../v2/dsl/spec/parser/TextParserSpec.java | 2 +- .../v2/provider/log/ILogAnalyzerService.java | 14 +-- .../log/analyzer/v2/provider/log/LogAnalyzer.java | 29 +++--- .../v2/provider/log/LogAnalyzerServiceImpl.java | 7 +- .../provider/log/listener/LogAnalysisListener.java | 12 +-- .../provider/log/listener/LogFilterListener.java | 10 +- .../v2/provider/log/listener/LogSinkListener.java | 16 +--- .../provider/log/listener/RecordSinkListener.java | 20 +--- .../provider/log/listener/TrafficSinkListener.java | 25 ++--- .../log/analyzer/v2/spi/LALSourceTypeProvider.java | 6 +- .../compiler/LALClassGeneratorExtractorTest.java | 14 +-- .../v2/compiler/LALExpressionExecutionTest.java | 26 +++--- .../oap/server/core/source/LALOutputBuilder.java | 11 +-- .../oap/server/core/source/LogBuilder.java | 101 +++++++++++++-------- .../oap/server/core/source/LogMetadata.java | 46 ++++++++++ .../oap/server/core/source/LogMetadataUtils.java | 61 +++++++++++++ .../agent/kafka/provider/handler/LogHandler.java | 6 +- .../oap/query/graphql/resolver/LogTestQuery.java | 18 ++-- .../envoy/persistence/EnvoyAccessLogBuilder.java | 18 ++-- .../envoy/persistence/LogsPersistence.java | 30 +++--- .../envoy/persistence/TCPLogsPersistence.java | 30 +++--- .../persistence/EnvoyAccessLogBuilderTest.java | 37 ++++---- .../envoy/persistence/EnvoyAlsLalTest.java | 17 ++-- .../otel/otlp/OpenTelemetryLogHandler.java | 21 ++--- .../handler/grpc/LogReportServiceGrpcHandler.java | 4 +- .../handler/rest/LogReportServiceHTTPHandler.java | 7 +- .../oap/server/checker/lal/LalBenchmark.java | 12 ++- .../oap/server/checker/lal/LalComparisonTest.java | 44 ++++++--- 38 files changed, 568 insertions(+), 415 deletions(-) diff --git a/oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/parser/listener/DatabaseSlowStatementBuilder.java b/oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/parser/listener/DatabaseSlowStatementBuilder.java index db33d33e2f..1cd2c92f96 100644 --- a/oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/parser/listener/DatabaseSlowStatementBuilder.java +++ b/oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/parser/listener/DatabaseSlowStatementBuilder.java @@ -18,11 +18,9 @@ package org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener; -import java.util.Optional; import lombok.Getter; import lombok.Setter; import lombok.extern.slf4j.Slf4j; -import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.oap.server.core.CoreModule; import org.apache.skywalking.oap.server.core.analysis.DownSampling; import org.apache.skywalking.oap.server.core.analysis.IDManager; @@ -31,6 +29,7 @@ import org.apache.skywalking.oap.server.core.analysis.TimeBucket; import org.apache.skywalking.oap.server.core.config.NamingControl; import org.apache.skywalking.oap.server.core.source.DatabaseSlowStatement; import org.apache.skywalking.oap.server.core.source.LALOutputBuilder; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.server.core.source.SourceReceiver; import org.apache.skywalking.oap.server.library.module.ModuleManager; @@ -70,7 +69,7 @@ public class DatabaseSlowStatementBuilder implements LALOutputBuilder { } /** - * Constructor for v1 (Groovy) path which doesn't use {@link #init}. + * Constructor for v1 (Groovy) path where NamingControl is resolved eagerly. */ public DatabaseSlowStatementBuilder(final NamingControl namingControl) { NAMING_CONTROL = namingControl; @@ -83,7 +82,7 @@ public class DatabaseSlowStatementBuilder implements LALOutputBuilder { } @Override - public void init(final LogData logData, final Optional<Object> extraLog, + public void init(final LogMetadata metadata, final Object input, final ModuleManager moduleManager) { if (!INITIALIZED) { NAMING_CONTROL = moduleManager.find(CoreModule.NAME) @@ -93,17 +92,20 @@ public class DatabaseSlowStatementBuilder implements LALOutputBuilder { } // Only populate fields not already set by the LAL extractor. if (this.serviceName == null) { - this.serviceName = logData.getService(); + this.serviceName = metadata.getService(); } if (this.traceId == null) { - this.traceId = logData.getTraceContext().getTraceId(); + final LogMetadata.TraceContext tc = metadata.getTraceContext(); + if (tc != null) { + this.traceId = tc.getTraceId(); + } } if (this.timestamp == 0) { - this.timestamp = logData.getTimestamp(); + this.timestamp = metadata.getTimestamp(); } if (this.timeBucket == 0) { this.timeBucket = TimeBucket.getTimeBucket( - this.timestamp > 0 ? this.timestamp : logData.getTimestamp(), + this.timestamp > 0 ? this.timestamp : metadata.getTimestamp(), DownSampling.Second); } } diff --git a/oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/parser/listener/SampledTraceBuilder.java b/oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/parser/listener/SampledTraceBuilder.java index 1605a7ca10..5c7068cb74 100644 --- a/oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/parser/listener/SampledTraceBuilder.java +++ b/oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/parser/listener/SampledTraceBuilder.java @@ -20,11 +20,9 @@ package org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener import com.google.common.base.Preconditions; import com.google.common.base.Strings; -import java.util.Optional; import lombok.Getter; import lombok.Setter; import lombok.extern.slf4j.Slf4j; -import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.oap.server.core.analysis.DownSampling; import org.apache.skywalking.oap.server.core.analysis.IDManager; import org.apache.skywalking.oap.server.core.analysis.Layer; @@ -40,6 +38,7 @@ import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine; import org.apache.skywalking.oap.server.core.source.DetectPoint; import org.apache.skywalking.oap.server.core.source.ISource; import org.apache.skywalking.oap.server.core.source.LALOutputBuilder; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.server.core.source.ProcessRelation; import org.apache.skywalking.oap.server.core.source.SourceReceiver; import org.apache.skywalking.oap.server.library.module.ModuleManager; @@ -94,7 +93,7 @@ public class SampledTraceBuilder implements LALOutputBuilder { } /** - * Constructor for v1 (Groovy) path which doesn't use {@link #init}. + * Constructor for v1 (Groovy) path where NamingControl is resolved eagerly. */ public SampledTraceBuilder(final NamingControl namingControl) { NAMING_CONTROL = namingControl; @@ -107,7 +106,7 @@ public class SampledTraceBuilder implements LALOutputBuilder { } @Override - public void init(final LogData logData, final Optional<Object> extraLog, + public void init(final LogMetadata metadata, final Object input, final ModuleManager moduleManager) { if (!INITIALIZED) { NAMING_CONTROL = moduleManager.find(CoreModule.NAME) @@ -117,19 +116,22 @@ public class SampledTraceBuilder implements LALOutputBuilder { } // Only populate fields not already set by the LAL extractor. if (this.traceId == null) { - this.traceId = logData.getTraceContext().getTraceId(); + final LogMetadata.TraceContext tc = metadata.getTraceContext(); + if (tc != null) { + this.traceId = tc.getTraceId(); + } } if (this.serviceName == null) { - this.serviceName = logData.getService(); + this.serviceName = metadata.getService(); } if (this.serviceInstanceName == null) { - this.serviceInstanceName = logData.getServiceInstance(); + this.serviceInstanceName = metadata.getServiceInstance(); } - if (this.layer == null && !logData.getLayer().isEmpty()) { - this.layer = logData.getLayer(); + if (this.layer == null && metadata.getLayer() != null && !metadata.getLayer().isEmpty()) { + this.layer = metadata.getLayer(); } if (this.timestamp == 0) { - this.timestamp = logData.getTimestamp(); + this.timestamp = metadata.getTimestamp(); } } diff --git a/oap-server/analyzer/log-analyzer/CLAUDE.md b/oap-server/analyzer/log-analyzer/CLAUDE.md index 3a0f6b0d13..5e46858ce1 100644 --- a/oap-server/analyzer/log-analyzer/CLAUDE.md +++ b/oap-server/analyzer/log-analyzer/CLAUDE.md @@ -41,7 +41,7 @@ oap-server/analyzer/log-analyzer/ src/main/java/.../dsl/ LalExpression.java — Functional interface: execute(FilterSpec, ExecutionContext) - ExecutionContext.java — Per-log execution state (log, parsed, flags) + ExecutionContext.java — Per-log execution state (metadata, input, parsed, flags) DSL.java — Wraps compiled expression + FilterSpec spec/filter/FilterSpec.java — Top-level filter spec (all methods take ctx explicitly) spec/extractor/MetricExtractor.java — Handles LAL metrics {} blocks (prepare/submit samples to MAL) @@ -182,10 +182,11 @@ The generator detects the parser type from the AST at compile time and generates | JSON/YAML | `parsed.service` | `h.mapVal("service")` | | JSON/YAML nested | `parsed.a.b` | `h.mapVal("a", "b")` | | TEXT (regexp) | `parsed.level` | `h.group("level")` | -| NONE + inputType | `parsed.response.code` | `((ExtraLogType) h.ctx().extraLog()).getResponse().getCode()` | -| NONE + no inputType | `parsed.service` | `h.ctx().log().getService()` (LogData.Builder fallback) | -| log fields | `log.service` | `h.ctx().log().getService()` | -| log trace | `log.traceContext.traceId` | `h.ctx().log().getTraceContext().getTraceId()` | +| NONE + inputType | `parsed.response.code` | `((InputType) h.ctx().input()).getResponse().getCode()` | +| NONE + no inputType | `parsed.service` | `h.ctx().metadata().getService()` (LogMetadata fallback) | +| log fields (metadata) | `log.service` | `h.ctx().metadata().getService()` | +| log fields (LogData) | `log.body` | `h.ctx().log().getBody()` | +| log trace | `log.traceContext.traceId` | `h.ctx().metadata().getTraceContext().getTraceId()` | | tags | `tag("KEY")` | `h.tagValue("KEY")` | ### inputType and LALSourceTypeProvider SPI @@ -195,7 +196,7 @@ For LAL rules with no DSL parser (`json{}`/`yaml{}`/`text{}`), the compiler need 1. **DSL parser** (`json{}`, `yaml{}`, `text{}`) — parser wins, inputType is ignored 2. **Explicit `inputType`** in YAML rule config — FQCN string, resolved via `Class.forName()` 3. **`LALSourceTypeProvider` SPI** — default inputType for a layer, discovered via `ServiceLoader` -4. **`LogData.Builder` fallback** — if none of the above, `parsed.*` generates getter chains on `LogData.Builder` with compile-time reflection validation. Fields not found on `LogData.Builder` cause `IllegalArgumentException` at boot. +4. **`LogMetadata` fallback** — if none of the above, `parsed.*` generates getter chains on `LogMetadata` with compile-time reflection validation. Fields not found on `LogMetadata` cause `IllegalArgumentException` at boot. The SPI interface is in `org.apache.skywalking.oap.log.analyzer.v2.spi.LALSourceTypeProvider`. Receiver plugins implement it and register in `META-INF/services/`. Example: `EnvoyHTTPLALSourceTypeProvider` registers `HTTPAccessLogEntry` for `Layer.MESH`. @@ -238,7 +239,7 @@ If no setter is found, compilation fails with an `IllegalArgumentException` at b **Runtime dispatch**: `RecordSinkListener.parse()` reads the output object from `ExecutionContext.output()` (already populated by generated code), calls -`init(logData, extraLog, moduleManager)` to populate standard fields and resolve +`init(metadata, input, moduleManager)` to populate standard fields and resolve services (e.g., `NamingControl`, `ConfigService`) from `ModuleManager`, then `build()` dispatches via `complete(sourceReceiver)`. Each builder caches resolved services in static fields so `ModuleManager` lookups only happen once. @@ -276,7 +277,7 @@ Instance-based helper created at the start of `execute()`, holds the `ExecutionC - `mapVal(key)`, `mapVal(k1, k2)`, `mapVal(k1, k2, k3)` — JSON/YAML map access - `group(name)` — text regexp named group - `tagValue(key)` — log tag lookup -- `ctx()` — access to ExecutionContext (for `h.ctx().log()` proto getters) +- `ctx()` — access to ExecutionContext (for `h.ctx().metadata()` and `h.ctx().log()` getters) **Type conversion:** `toStr()`, `toLong()`, `toInt()`, `toBool()` @@ -284,11 +285,11 @@ Instance-based helper created at the start of `execute()`, holds the `ExecutionC **Safe navigation:** `toString()`, `trim()` -## JSON/YAML LogData Field Population +## JSON/YAML Metadata Field Population -When `json{}` or `yaml{}` parses the log body, `FilterSpec` also adds LogData proto fields +When `json{}` or `yaml{}` parses the log body, `FilterSpec` also adds `LogMetadata` fields (`service`, `serviceInstance`, `endpoint`, `layer`, `timestamp`) to the parsed map via -`putIfAbsent`. Body-parsed values take priority; proto fields serve as fallback. This matches +`putIfAbsent`. Body-parsed values take priority; metadata fields serve as fallback. This matches v1 Groovy `Binding.Parsed.getAt(key)` behavior where `parsed.service` falls back to `LogData.getService()` when the JSON body doesn't contain a `service` key. @@ -342,7 +343,7 @@ rule-name: ### Principles -1. **`body-type` determines parsing**: `json` → `json{}` block, `text` → `text{}` block, `none` → proto extraLog or raw LogData access. +1. **`body-type` determines parsing**: `json` → `json{}` block, `text` → `text{}` block, `none` → typed proto input or LogMetadata fallback. 2. **`extra-log` for proto types**: When rules access `parsed.*` on protobuf types (e.g., `HTTPAccessLogEntry`), provide `proto-class` and `proto-json`. The test harness parses via `JsonFormat`. 3. **`expect` section is mandatory**: Every entry must have `expect` with at least `save` and `abort`. 4. **Tag assertions**: `tag.KEY` in expect asserts extracted tag values (e.g., `tag.status.code: "500"`). @@ -380,7 +381,7 @@ Test utilities from `org.apache.skywalking.oap.server.testing.dsl`: - `LalRuleLoader.loadAllRules(Path)` — loads all LAL rules with companion `.input.data` or `.data.yaml` - `LalLogDataBuilder.buildLogData(Map)` — builds `LogData.Builder` from test input map - `LalLogDataBuilder.buildSyntheticLogData(String)` — builds synthetic LogData from DSL string -- `LalLogDataBuilder.buildExtraLog(Map)` — builds proto Message for extraLog from input map +- `LalLogDataBuilder.buildExtraLog(Map)` — builds proto Message for typed input from input map - `DslRuleLoader.findScriptsDir(String...)` — resolves scripts directory from candidates - `DslRuleLoader.findRuleLine(String[], String, int)` — finds 1-based line number of rule in YAML diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALBlockCodegen.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALBlockCodegen.java index 629edf7f4e..aec41cf37c 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALBlockCodegen.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALBlockCodegen.java @@ -23,7 +23,7 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; -import org.apache.skywalking.apm.network.logging.v3.LogData; +import org.apache.skywalking.oap.server.core.source.LogMetadata; /** * Static code-generation methods for LAL extractor, sink, condition, and @@ -43,6 +43,8 @@ final class LALBlockCodegen { "org.apache.skywalking.oap.log.analyzer.v2.compiler.rt.LalRuntimeHelper"; private static final String PROCESS_REGISTRY = "org.apache.skywalking.oap.meter.analyzer.v2.dsl.registry.ProcessRegistry"; + private static final String LOGDATA_BUILDER_CAST = + "((org.apache.skywalking.apm.network.logging.v3.LogData.Builder) h.ctx().input())"; // Built-in function registry for def variable type inference. // Maps DSL function name → [runtime helper method, return type]. @@ -87,7 +89,7 @@ final class LALBlockCodegen { if (genCtx.inputType != null) { final String elTypeName = genCtx.inputType.getName(); body.append(" ").append(elTypeName).append(" _p = (") - .append(elTypeName).append(") h.ctx().extraLog();\n"); + .append(elTypeName).append(") h.ctx().input();\n"); lvtVars.add(new String[]{"_p", "L" + elTypeName.replace('.', '/') + ";"}); } @@ -460,7 +462,7 @@ final class LALBlockCodegen { if (genCtx.inputType != null) { final String elTypeName = genCtx.inputType.getName(); body.append(" ").append(elTypeName).append(" _p = (") - .append(elTypeName).append(") h.ctx().extraLog();\n"); + .append(elTypeName).append(") h.ctx().input();\n"); lvtVars.add(new String[]{"_p", "L" + elTypeName.replace('.', '/') + ";"}); } @@ -929,36 +931,68 @@ final class LALBlockCodegen { // ==================== Log access (direct proto getters) ==================== + /** + * Generate code for {@code log.xxx} field access in LAL scripts. + * <ul> + * <li>Metadata fields (service, endpoint, layer, timestamp, traceContext) + * → {@code h.ctx().metadata().getXxx()}</li> + * <li>LogData-only fields (body, tags) + * → {@code ((LogData.Builder) h.ctx().input()).getXxx()}</li> + * </ul> + */ static void generateLogAccess(final StringBuilder sb, final List<LALScriptModel.ValueAccessSegment> chain) { if (chain.isEmpty()) { - sb.append("h.ctx().log()"); + sb.append("h.ctx().input()"); return; } - String current = "h.ctx().log()"; + String current; boolean needsBoxing = false; String boxType = null; + // Determine root based on first field + final LALScriptModel.ValueAccessSegment first = chain.get(0); + if (!(first instanceof LALScriptModel.FieldSegment)) { + current = LOGDATA_BUILDER_CAST; + } else { + final String firstName = ((LALScriptModel.FieldSegment) first).getName(); + if (LALCodegenHelper.METADATA_GETTERS.containsKey(firstName)) { + current = "h.ctx().metadata()"; + } else if (LALCodegenHelper.LOG_GETTERS.containsKey(firstName)) { + current = LOGDATA_BUILDER_CAST; + } else { + throw new IllegalArgumentException( + "Unknown log field: log." + firstName + + ". Supported metadata fields: " + + LALCodegenHelper.METADATA_GETTERS.keySet() + + ", LogData fields: " + + LALCodegenHelper.LOG_GETTERS.keySet()); + } + } + for (int i = 0; i < chain.size(); i++) { final LALScriptModel.ValueAccessSegment seg = chain.get(i); if (seg instanceof LALScriptModel.FieldSegment) { final String name = ((LALScriptModel.FieldSegment) seg).getName(); - if (i == 0 && LALCodegenHelper.LOG_GETTERS.containsKey(name)) { + if (i == 0 && LALCodegenHelper.METADATA_GETTERS.containsKey(name)) { if ("traceContext".equals(name)) { current = current + ".getTraceContext()"; } else { current = current + "." - + LALCodegenHelper.LOG_GETTERS.get(name) + "()"; + + LALCodegenHelper.METADATA_GETTERS.get(name) + "()"; if (LALCodegenHelper.LONG_FIELDS.contains(name)) { needsBoxing = true; boxType = "Long"; } } + } else if (i == 0 && LALCodegenHelper.LOG_GETTERS.containsKey(name)) { + current = current + "." + + LALCodegenHelper.LOG_GETTERS.get(name) + "()"; } else if (i == 1 && current.endsWith(".getTraceContext()") - && LALCodegenHelper.TRACE_CONTEXT_GETTERS.containsKey(name)) { + && LALCodegenHelper.METADATA_TRACE_GETTERS.containsKey(name)) { current = current + "." - + LALCodegenHelper.TRACE_CONTEXT_GETTERS.get(name) + "()"; + + LALCodegenHelper.METADATA_TRACE_GETTERS.get(name) + "()"; if (LALCodegenHelper.INT_FIELDS.contains(name)) { needsBoxing = true; boxType = "Integer"; @@ -966,10 +1000,12 @@ final class LALBlockCodegen { } else { throw new IllegalArgumentException( "Unknown log field: log." + name - + ". Supported fields: " - + LALCodegenHelper.LOG_GETTERS.keySet() + + ". Supported metadata fields: " + + LALCodegenHelper.METADATA_GETTERS.keySet() + ", traceContext." - + LALCodegenHelper.TRACE_CONTEXT_GETTERS.keySet()); + + LALCodegenHelper.METADATA_TRACE_GETTERS.keySet() + + ", LogData fields: " + + LALCodegenHelper.LOG_GETTERS.keySet()); } } else if (seg instanceof LALScriptModel.MethodSegment) { current = appendMethodSegment(current, @@ -1032,9 +1068,9 @@ final class LALBlockCodegen { current = generateExtraLogAccess(fieldSegments, genCtx.inputType, "_p", true, genCtx); } else { - // No parser and no inputType — fall back to LogData proto - current = generateExtraLogAccess(fieldSegments, LogData.Builder.class, - "h.ctx().log()", false, genCtx); + // No parser and no inputType — fall back to LogMetadata + current = generateExtraLogAccess(fieldSegments, LogMetadata.class, + "h.ctx().metadata()", false, genCtx); } break; default: @@ -1083,9 +1119,19 @@ final class LALBlockCodegen { for (int i = 0; i < fieldSegments.size(); i++) { final LALScriptModel.FieldSegment seg = fieldSegments.get(i); final String field = seg.getName(); - final String getterName = "get" + Character.toUpperCase(field.charAt(0)) + String getterName = "get" + Character.toUpperCase(field.charAt(0)) + field.substring(1); + // Apply getter aliases (e.g., traceSegmentId → segmentId on LogMetadata) + final String alias = LALCodegenHelper.METADATA_GETTER_ALIASES.get(getterName); + if (alias != null) { + try { + currentType.getMethod(getterName); + } catch (NoSuchMethodException ignored) { + getterName = alias; + } + } + final java.lang.reflect.Method getter; try { getter = currentType.getMethod(getterName); diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALClassGenerator.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALClassGenerator.java index 3a1cab3655..fad68e4901 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALClassGenerator.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALClassGenerator.java @@ -543,7 +543,7 @@ public final class LALClassGenerator { if (genCtx.inputType != null) { final String elTypeName = genCtx.inputType.getName(); sb.append(" ").append(elTypeName).append(" _p = (") - .append(elTypeName).append(") h.ctx().extraLog();\n"); + .append(elTypeName).append(") h.ctx().input();\n"); } sb.append(genCtx.protoVarDecls); } diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALCodegenHelper.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALCodegenHelper.java index bb890a65fe..1a11be387d 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALCodegenHelper.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALCodegenHelper.java @@ -29,27 +29,39 @@ import java.util.Set; */ final class LALCodegenHelper { + /** Metadata fields — available on {@code LogMetadata} (always present). */ + static final Map<String, String> METADATA_GETTERS = new HashMap<>(); + /** TraceContext sub-fields on {@code LogMetadata.TraceContext}. */ + static final Map<String, String> METADATA_TRACE_GETTERS = new HashMap<>(); + /** LogData-only fields — require the input to be {@code LogData}. */ static final Map<String, String> LOG_GETTERS = new HashMap<>(); - static final Map<String, String> TRACE_CONTEXT_GETTERS = new HashMap<>(); + /** + * Getter-name overrides for reflection-based access. + * Applied by {@link LALBlockCodegen#generateExtraLogAccess} when a getter + * is not found on the current type — tries the alias before failing. + */ + static final Map<String, String> METADATA_GETTER_ALIASES = Map.of(); static final Set<String> LONG_FIELDS = new HashSet<>(); static final Set<String> INT_FIELDS = new HashSet<>(); static { - LOG_GETTERS.put("service", "getService"); - LOG_GETTERS.put("serviceInstance", "getServiceInstance"); - LOG_GETTERS.put("endpoint", "getEndpoint"); - LOG_GETTERS.put("timestamp", "getTimestamp"); + METADATA_GETTERS.put("service", "getService"); + METADATA_GETTERS.put("serviceInstance", "getServiceInstance"); + METADATA_GETTERS.put("endpoint", "getEndpoint"); + METADATA_GETTERS.put("layer", "getLayer"); + METADATA_GETTERS.put("timestamp", "getTimestamp"); + METADATA_GETTERS.put("traceContext", "getTraceContext"); + + METADATA_TRACE_GETTERS.put("traceId", "getTraceId"); + METADATA_TRACE_GETTERS.put("traceSegmentId", "getTraceSegmentId"); + METADATA_TRACE_GETTERS.put("spanId", "getSpanId"); + LOG_GETTERS.put("body", "getBody"); - LOG_GETTERS.put("traceContext", "getTraceContext"); LOG_GETTERS.put("tags", "getTags"); - LOG_GETTERS.put("layer", "getLayer"); - - TRACE_CONTEXT_GETTERS.put("traceId", "getTraceId"); - TRACE_CONTEXT_GETTERS.put("traceSegmentId", "getTraceSegmentId"); - TRACE_CONTEXT_GETTERS.put("spanId", "getSpanId"); LONG_FIELDS.add("timestamp"); INT_FIELDS.add("spanId"); + } private LALCodegenHelper() { diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/rt/LalRuntimeHelper.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/rt/LalRuntimeHelper.java index 56ad4b706a..4ffd24ae1d 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/rt/LalRuntimeHelper.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/rt/LalRuntimeHelper.java @@ -31,6 +31,7 @@ import java.util.List; import java.util.Map; import org.apache.commons.lang3.StringUtils; import org.apache.skywalking.apm.network.common.v3.KeyStringValuePair; +import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.oap.log.analyzer.v2.dsl.ExecutionContext; import org.apache.skywalking.oap.server.library.util.StringUtil; @@ -67,13 +68,17 @@ import org.apache.skywalking.oap.server.library.util.StringUtil; * // Generated: h.tagValue("LOG_KIND") * </pre> * - * <p><b>4. Log proto data</b> — direct access to {@code LogData.Builder} fields. + * <p><b>4. Metadata</b> — direct access to {@code LogMetadata} fields. * Not accessed through this helper; the compiler generates direct getter chains - * like {@code h.ctx().log().getService()}. + * like {@code h.ctx().metadata().getService()}. * - * <p><b>5. ExtraLog proto data</b> — direct access to typed protobuf extraLog. + * <p><b>5. LogData body/tags</b> — direct access to {@code LogData.Builder} fields. * Not accessed through this helper; the compiler generates typed cast + getter - * chains like {@code ((HTTPAccessLogEntry) h.ctx().extraLog()).getResponse()}. + * chains like {@code ((LogData.Builder) h.ctx().input()).getBody()}. + * + * <p><b>6. Typed input proto data</b> — direct access to typed protobuf input. + * Not accessed through this helper; the compiler generates typed cast + getter + * chains like {@code ((HTTPAccessLogEntry) h.ctx().input()).getResponse()}. * * <h2>Type Conversion Methods</h2> * @@ -178,7 +183,7 @@ public final class LalRuntimeHelper { * </pre> */ public String tagValue(final String key) { - final List dl = ctx.log().getTags().getDataList(); + final List dl = ((LogData.Builder) ctx.input()).getTags().getDataList(); for (int i = 0; i < dl.size(); i++) { final KeyStringValuePair kv = (KeyStringValuePair) dl.get(i); if (key.equals(kv.getKey())) { diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/DSL.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/DSL.java index 265b674458..b279744a4c 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/DSL.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/DSL.java @@ -20,10 +20,10 @@ package org.apache.skywalking.oap.log.analyzer.v2.dsl; import lombok.AccessLevel; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.oap.log.analyzer.v2.compiler.LALClassGenerator; import org.apache.skywalking.oap.log.analyzer.v2.dsl.spec.filter.FilterSpec; import org.apache.skywalking.oap.log.analyzer.v2.provider.LogAnalyzerModuleConfig; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.library.module.ModuleStartException; @@ -82,11 +82,11 @@ public class DSL { public void evaluate(final ExecutionContext ctx) { if (log.isDebugEnabled()) { - final LogData.Builder logData = ctx.log(); - log.debug("[LAL] rule={}, class={}, service={}, instance={}, endpoint={}, bodyType={}", + final LogMetadata metadata = ctx.metadata(); + log.debug("[LAL] rule={}, class={}, service={}, instance={}, endpoint={}", ruleName, expression.getClass().getName(), - logData.getService(), logData.getServiceInstance(), - logData.getEndpoint(), logData.getBody().getContentCase()); + metadata.getService(), metadata.getServiceInstance(), + metadata.getEndpoint()); } expression.execute(filterSpec, ctx); } diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/ExecutionContext.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/ExecutionContext.java index 992d797720..04154077a2 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/ExecutionContext.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/ExecutionContext.java @@ -23,10 +23,9 @@ import java.util.Map; import java.util.Optional; import java.util.regex.Matcher; import lombok.Getter; -import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.oap.meter.analyzer.v2.dsl.SampleFamily; import org.apache.skywalking.oap.server.core.source.LALOutputBuilder; -import org.apache.skywalking.oap.server.core.source.Log; +import org.apache.skywalking.oap.server.core.source.LogMetadata; /** * Mutable property storage for a single LAL script execution cycle. @@ -34,21 +33,22 @@ import org.apache.skywalking.oap.server.core.source.Log; * <p>A new ExecutionContext is created for each incoming log. It carries all * per-log state through the compiled LAL pipeline: * <ul> - * <li>{@code log} — the incoming {@code LogData.Builder}</li> + * <li>{@code metadata} — the {@link LogMetadata} (service, layer, timestamp, trace context)</li> + * <li>{@code input} — the raw input object (LogData.Builder for standard logs, typed proto for ALS)</li> * <li>{@code parsed} — structured data extracted by json/text/yaml parsers</li> * <li>{@code save}/{@code abort} — control flags set by extractor/sink logic</li> * <li>{@code metrics_container} — optional list for LAL-extracted metrics (log-MAL)</li> - * <li>{@code log_container} — optional container for the built {@code Log} source object</li> + * <li>{@code dry_run} — when true, sink skips persistence (used by the LAL test tool)</li> * </ul> */ public class ExecutionContext { - public static final String KEY_LOG = "log"; + public static final String KEY_INPUT = "input"; + public static final String KEY_METADATA = "metadata"; public static final String KEY_PARSED = "parsed"; public static final String KEY_SAVE = "save"; public static final String KEY_ABORT = "abort"; public static final String KEY_METRICS_CONTAINER = "metrics_container"; - public static final String KEY_CAPTURE_LOG = "capture_log"; - public static final String KEY_LOG_CONTAINER = "log_container"; + public static final String KEY_DRY_RUN = "dry_run"; public static final String KEY_OUTPUT = "output"; private final Map<String, Object> properties = new HashMap<>(); @@ -65,32 +65,30 @@ public class ExecutionContext { return properties.get(name); } - public ExecutionContext log(final LogData.Builder log) { - setProperty(KEY_LOG, log); + /** + * Initialize from metadata + input. + */ + public ExecutionContext init(final LogMetadata metadata, final Object input) { + setProperty(KEY_METADATA, metadata); + setProperty(KEY_INPUT, input); setProperty(KEY_SAVE, true); setProperty(KEY_ABORT, false); setProperty(KEY_METRICS_CONTAINER, null); - setProperty(KEY_CAPTURE_LOG, false); - setProperty(KEY_LOG_CONTAINER, null); + setProperty(KEY_DRY_RUN, false); setProperty(KEY_OUTPUT, null); return this; } - public ExecutionContext log(final LogData log) { - return log(log.toBuilder()); + public LogMetadata metadata() { + return (LogMetadata) getProperty(KEY_METADATA); } - public LogData.Builder log() { - return (LogData.Builder) getProperty(KEY_LOG); - } - - public ExecutionContext extraLog(final Object extraLog) { - parsed().extraLog = extraLog; - return this; - } - - public Object extraLog() { - return parsed().getExtraLog(); + /** + * Returns the raw input object. For standard logs this is a {@code LogData.Builder}; + * for ALS it is the typed proto (e.g., {@code HTTPAccessLogEntry}). + */ + public Object input() { + return getProperty(KEY_INPUT); } public ExecutionContext parsed(final Matcher parsed) { @@ -140,22 +138,13 @@ public class ExecutionContext { return Optional.ofNullable((List<SampleFamily>) getProperty(KEY_METRICS_CONTAINER)); } - public ExecutionContext captureLog(final boolean capture) { - setProperty(KEY_CAPTURE_LOG, capture); - return this; - } - - public boolean shouldCaptureLog() { - return (boolean) getProperty(KEY_CAPTURE_LOG); - } - - public ExecutionContext logContainer(final Log container) { - setProperty(KEY_LOG_CONTAINER, container); + public ExecutionContext dryRun(final boolean dryRun) { + setProperty(KEY_DRY_RUN, dryRun); return this; } - public Optional<Log> logContainer() { - return Optional.ofNullable((Log) getProperty(KEY_LOG_CONTAINER)); + public boolean isDryRun() { + return (boolean) getProperty(KEY_DRY_RUN); } public void setOutput(final Object output) { @@ -176,8 +165,5 @@ public class ExecutionContext { @Getter private Map<String, Object> map; - - @Getter - private Object extraLog; } } diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/filter/FilterSpec.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/filter/FilterSpec.java index 476a9802f4..dcd72fa052 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/filter/FilterSpec.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/filter/FilterSpec.java @@ -19,11 +19,9 @@ package org.apache.skywalking.oap.log.analyzer.v2.dsl.spec.filter; import com.fasterxml.jackson.core.type.TypeReference; -import com.google.protobuf.TextFormat; import java.util.Arrays; import java.util.List; import java.util.Map; -import java.util.Optional; import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.oap.log.analyzer.v2.dsl.ExecutionContext; import org.apache.skywalking.oap.log.analyzer.v2.dsl.spec.AbstractSpec; @@ -37,7 +35,7 @@ import org.apache.skywalking.oap.log.analyzer.v2.provider.LogAnalyzerModuleConfi import org.apache.skywalking.oap.log.analyzer.v2.provider.log.listener.LogSinkListenerFactory; import org.apache.skywalking.oap.log.analyzer.v2.provider.log.listener.RecordSinkListener; import org.apache.skywalking.oap.log.analyzer.v2.provider.log.listener.TrafficSinkListener; -import org.apache.skywalking.oap.server.core.source.LogBuilder; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.library.module.ModuleStartException; import org.slf4j.Logger; @@ -113,21 +111,21 @@ public class FilterSpec extends AbstractSpec { /** * LAL {@code json {}} — parses {@code LogData.body.json.json} into a * {@code Map<String, Object>} and stores it in {@code ctx.parsed()}. - * LogData proto fields (service, serviceInstance, endpoint, layer, timestamp) + * Metadata fields (service, serviceInstance, endpoint, layer, timestamp) * are also added to the map via {@code putIfAbsent}, so body values take - * priority while proto fields serve as fallback — matching v1 Groovy + * priority while metadata fields serve as fallback — matching v1 Groovy * {@code Binding.Parsed.getAt(key)} behavior. */ public void json(final ExecutionContext ctx) { if (ctx.shouldAbort()) { return; } - final LogData.Builder logData = ctx.log(); try { + final LogData.Builder logData = (LogData.Builder) ctx.input(); final Map<String, Object> parsed = jsonParser.create().readValue( logData.getBody().getJson().getJson(), parsedType ); - addLogDataFields(parsed, logData); + addMetadataFields(parsed, ctx.metadata()); ctx.parsed(parsed); } catch (final Exception e) { if (jsonParser.abortOnFailure()) { @@ -139,18 +137,18 @@ public class FilterSpec extends AbstractSpec { /** * LAL {@code yaml {}} — parses {@code LogData.body.yaml.yaml} into a * {@code Map<String, Object>} and stores it in {@code ctx.parsed()}. - * LogData proto fields are added the same way as {@link #json(ExecutionContext)}. + * Metadata fields are added the same way as {@link #json(ExecutionContext)}. */ public void yaml(final ExecutionContext ctx) { if (ctx.shouldAbort()) { return; } - final LogData.Builder logData = ctx.log(); try { + final LogData.Builder logData = (LogData.Builder) ctx.input(); final Map<String, Object> parsed = yamlParser.create().load( logData.getBody().getYaml().getYaml() ); - addLogDataFields(parsed, logData); + addMetadataFields(parsed, ctx.metadata()); ctx.parsed(parsed); } catch (final Exception e) { if (yamlParser.abortOnFailure()) { @@ -171,32 +169,24 @@ public class FilterSpec extends AbstractSpec { } private void doSink(final ExecutionContext ctx) { - final LogData.Builder logData = ctx.log(); - final Optional<Object> extraLog = Optional.ofNullable(ctx.extraLog()); + if (ctx.isDryRun()) { + return; + } + + final LogMetadata metadata = ctx.metadata(); + final Object input = ctx.input(); if (!ctx.shouldSave()) { if (LOGGER.isDebugEnabled()) { - LOGGER.debug("Log is dropped: {}", TextFormat.shortDebugString(logData)); + LOGGER.debug("Log is dropped: service={}, layer={}", + metadata.getService(), metadata.getLayer()); } return; } - if (ctx.shouldCaptureLog()) { - sinkListenerFactories.stream() - .map(LogSinkListenerFactory::create) - .filter(it -> it instanceof RecordSinkListener) - .map(it -> it.parse(logData, extraLog, ctx)) - .map(it -> (RecordSinkListener) it) - .map(RecordSinkListener::getBuilder) - .filter(it -> it instanceof LogBuilder) - .map(it -> ((LogBuilder) it).toLog()) - .findFirst() - .ifPresent(log -> ctx.logContainer(log)); - } else { - sinkListenerFactories.stream() - .map(LogSinkListenerFactory::create) - .forEach(it -> it.parse(logData, extraLog, ctx).build()); - } + sinkListenerFactories.stream() + .map(LogSinkListenerFactory::create) + .forEach(it -> it.parse(metadata, input, ctx).build()); } // ==================== Direct-access APIs for flattened generated code ==================== @@ -229,18 +219,21 @@ public class FilterSpec extends AbstractSpec { } /** - * Add LogData proto fields to the parsed map so that {@code parsed.service}, + * Add metadata fields to the parsed map so that {@code parsed.service}, * {@code parsed.serviceInstance}, etc. resolve correctly — matching v1 Groovy * {@code Binding.Parsed.getAt(key)} fallback behavior. * Uses {@code putIfAbsent} so body-parsed values take priority. */ - private static void addLogDataFields(final Map<String, Object> parsed, - final LogData.Builder logData) { - putIfNotEmpty(parsed, "service", logData.getService()); - putIfNotEmpty(parsed, "serviceInstance", logData.getServiceInstance()); - putIfNotEmpty(parsed, "endpoint", logData.getEndpoint()); - putIfNotEmpty(parsed, "layer", logData.getLayer()); - final long ts = logData.getTimestamp(); + private static void addMetadataFields(final Map<String, Object> parsed, + final LogMetadata metadata) { + if (metadata == null) { + return; + } + putIfNotEmpty(parsed, "service", metadata.getService()); + putIfNotEmpty(parsed, "serviceInstance", metadata.getServiceInstance()); + putIfNotEmpty(parsed, "endpoint", metadata.getEndpoint()); + putIfNotEmpty(parsed, "layer", metadata.getLayer()); + final long ts = metadata.getTimestamp(); if (ts > 0) { parsed.putIfAbsent("timestamp", ts); } diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/parser/TextParserSpec.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/parser/TextParserSpec.java index 8d57a0c2ad..a0f07926c2 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/parser/TextParserSpec.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/dsl/spec/parser/TextParserSpec.java @@ -39,7 +39,7 @@ public class TextParserSpec extends AbstractParserSpec { if (ctx.shouldAbort()) { return; } - final LogData.Builder log = ctx.log(); + final LogData.Builder log = (LogData.Builder) ctx.input(); final Matcher matcher = pattern.matcher(log.getBody().getText().getText()); final boolean matched = matcher.find(); if (matched) { diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/ILogAnalyzerService.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/ILogAnalyzerService.java index ec951fba50..6aae340848 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/ILogAnalyzerService.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/ILogAnalyzerService.java @@ -17,8 +17,7 @@ package org.apache.skywalking.oap.log.analyzer.v2.provider.log; -import java.util.Optional; -import org.apache.skywalking.apm.network.logging.v3.LogData; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.server.library.module.Service; /** @@ -26,10 +25,11 @@ import org.apache.skywalking.oap.server.library.module.Service; */ public interface ILogAnalyzerService extends Service { - void doAnalysis(LogData.Builder log, Optional<Object> extraLog); - - default void doAnalysis(LogData logData, Optional<Object> extraLog) { - doAnalysis(logData.toBuilder(), extraLog); - } + /** + * @param metadata uniform metadata (service, layer, timestamp, trace context) + * @param input source-specific input object (LogData for standard logs, + * HTTPAccessLogEntry for envoy ALS, etc.) + */ + void doAnalysis(LogMetadata metadata, Object input); } diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/LogAnalyzer.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/LogAnalyzer.java index 08f84a9156..19d99541bd 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/LogAnalyzer.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/LogAnalyzer.java @@ -20,12 +20,11 @@ package org.apache.skywalking.oap.log.analyzer.v2.provider.log; import java.util.ArrayList; import java.util.List; import java.util.Objects; -import java.util.Optional; import lombok.extern.slf4j.Slf4j; -import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.oap.log.analyzer.v2.provider.log.listener.LogAnalysisListener; import org.apache.skywalking.oap.server.core.UnexpectedException; import org.apache.skywalking.oap.server.core.analysis.Layer; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.server.library.util.StringUtil; /** @@ -40,9 +39,9 @@ import org.apache.skywalking.oap.server.library.util.StringUtil; * {@link org.apache.skywalking.oap.log.analyzer.v2.provider.log.listener.LogFilterListener.Factory}, * which returns a listener wrapping all compiled {@link org.apache.skywalking.oap.log.analyzer.v2.dsl.DSL} * instances for that layer.</li> - * <li>{@code notifyAnalysisListener(builder, extraLog)} — calls + * <li>{@code notifyAnalysisListener(metadata, input)} — calls * {@link org.apache.skywalking.oap.log.analyzer.v2.provider.log.listener.LogAnalysisListener#parse} - * on each listener, which binds the log data to the compiled LAL scripts.</li> + * on each listener, which binds the metadata and input to the compiled LAL scripts.</li> * <li>{@code notifyAnalysisListenerToBuild()} — calls * {@link org.apache.skywalking.oap.log.analyzer.v2.provider.log.listener.LogAnalysisListener#build} * on each listener, which evaluates the compiled LAL scripts (extractors, sinks).</li> @@ -58,36 +57,34 @@ public class LogAnalyzer { private final List<LogAnalysisListener> listeners = new ArrayList<>(); - public void doAnalysis(LogData.Builder builder, Optional<Object> extraLog) { - if (StringUtil.isEmpty(builder.getService())) { - // If the service name is empty, the log will be ignored. + public void doAnalysis(LogMetadata metadata, final Object input) { + if (StringUtil.isEmpty(metadata.getService())) { log.debug("The log is ignored because the Service name is empty"); return; } Layer layer; - if ("".equals(builder.getLayer())) { + if (metadata.getLayer() == null || metadata.getLayer().isEmpty()) { layer = Layer.GENERAL; } else { try { - layer = Layer.nameOf(builder.getLayer()); + layer = Layer.nameOf(metadata.getLayer()); } catch (UnexpectedException e) { - log.warn("The Layer {} is not found, abandon the log.", builder.getLayer()); + log.warn("The Layer {} is not found, abandon the log.", metadata.getLayer()); return; } } createAnalysisListeners(layer); - if (builder.getTimestamp() == 0) { - // If no timestamp, OAP server would use the received timestamp as log's timestamp - builder.setTimestamp(System.currentTimeMillis()); + if (metadata.getTimestamp() == 0) { + metadata.setTimestamp(System.currentTimeMillis()); } - notifyAnalysisListener(builder, extraLog); + notifyAnalysisListener(metadata, input); notifyAnalysisListenerToBuild(); } - private void notifyAnalysisListener(LogData.Builder builder, final Optional<Object> extraLog) { - listeners.forEach(listener -> listener.parse(builder, extraLog)); + private void notifyAnalysisListener(final LogMetadata metadata, final Object input) { + listeners.forEach(listener -> listener.parse(metadata, input)); } private void notifyAnalysisListenerToBuild() { diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/LogAnalyzerServiceImpl.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/LogAnalyzerServiceImpl.java index 8e1c746bfc..581a2a337d 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/LogAnalyzerServiceImpl.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/LogAnalyzerServiceImpl.java @@ -19,11 +19,10 @@ package org.apache.skywalking.oap.log.analyzer.v2.provider.log; import java.util.ArrayList; import java.util.List; -import java.util.Optional; import lombok.RequiredArgsConstructor; -import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.oap.log.analyzer.v2.provider.LogAnalyzerModuleConfig; import org.apache.skywalking.oap.log.analyzer.v2.provider.log.listener.LogAnalysisListenerFactory; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.server.library.module.ModuleManager; @RequiredArgsConstructor @@ -33,9 +32,9 @@ public class LogAnalyzerServiceImpl implements ILogAnalyzerService, ILogAnalysis private final List<LogAnalysisListenerFactory> analysisListenerFactories = new ArrayList<>(); @Override - public void doAnalysis(final LogData.Builder log, Optional<Object> extraLog) { + public void doAnalysis(final LogMetadata metadata, final Object input) { LogAnalyzer analyzer = new LogAnalyzer(this); - analyzer.doAnalysis(log, extraLog); + analyzer.doAnalysis(metadata, input); } @Override diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/LogAnalysisListener.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/LogAnalysisListener.java index fe299835e9..0154ee76c6 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/LogAnalysisListener.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/LogAnalysisListener.java @@ -17,8 +17,7 @@ package org.apache.skywalking.oap.log.analyzer.v2.provider.log.listener; -import java.util.Optional; -import org.apache.skywalking.apm.network.logging.v3.LogData; +import org.apache.skywalking.oap.server.core.source.LogMetadata; /** * LogAnalysisListener represents the callback when OAP does the log data analysis. @@ -31,11 +30,10 @@ public interface LogAnalysisListener { /** * Parse the raw data from the probe. - * @param logData log metadata (service, layer, timestamp, etc.) - * @param extraLog the actual input object whose type matches - * {@code LALSourceTypeProvider#inputType()} — may be {@code null} - * for standard logs where LogData is the sole input + * @param metadata uniform metadata (service, layer, timestamp, trace context) + * @param input source-specific input object (LogData for standard logs, + * HTTPAccessLogEntry for envoy ALS, etc.) * @return {@code this} for chaining. */ - LogAnalysisListener parse(LogData.Builder logData, Optional<Object> extraLog); + LogAnalysisListener parse(LogMetadata metadata, Object input); } diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/LogFilterListener.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/LogFilterListener.java index 0d1873ac91..ab16b1f540 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/LogFilterListener.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/LogFilterListener.java @@ -22,13 +22,11 @@ import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.Map; -import java.util.Optional; import java.util.ServiceLoader; import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; import java.util.HashMap; -import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.oap.log.analyzer.v2.dsl.ExecutionContext; import org.apache.skywalking.oap.log.analyzer.v2.dsl.DSL; import org.apache.skywalking.oap.log.analyzer.v2.provider.LALConfig; @@ -38,6 +36,7 @@ import org.apache.skywalking.oap.log.analyzer.v2.spi.LALSourceTypeProvider; import org.apache.skywalking.oap.server.core.analysis.Layer; import org.apache.skywalking.oap.server.core.source.LALOutputBuilder; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.library.module.ModuleStartException; @@ -76,12 +75,11 @@ public class LogFilterListener implements LogAnalysisListener { } @Override - public LogAnalysisListener parse(final LogData.Builder logData, - final Optional<Object> extraLog) { - final LogData logDataSnapshot = logData.build(); + public LogAnalysisListener parse(final LogMetadata metadata, + final Object input) { contexts = new ArrayList<>(dsls.size()); for (int i = 0; i < dsls.size(); i++) { - contexts.add(new ExecutionContext().log(logDataSnapshot).extraLog(extraLog.orElse(null))); + contexts.add(new ExecutionContext().init(metadata, input)); } return this; } diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/LogSinkListener.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/LogSinkListener.java index afda6f0671..d282156505 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/LogSinkListener.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/LogSinkListener.java @@ -17,9 +17,8 @@ package org.apache.skywalking.oap.log.analyzer.v2.provider.log.listener; -import java.util.Optional; -import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.oap.log.analyzer.v2.dsl.ExecutionContext; +import org.apache.skywalking.oap.server.core.source.LogMetadata; public interface LogSinkListener { /** @@ -32,16 +31,5 @@ public interface LogSinkListener { * Parse the raw data from the probe. * @return {@code this} for chaining. */ - LogSinkListener parse(LogData.Builder logData, Optional<Object> extraLog); - - /** - * Parse the raw data from the probe with access to the execution context. - * Implementations can use the context to apply output fields or other - * per-execution state to the sink output. - * @return {@code this} for chaining. - */ - default LogSinkListener parse(final LogData.Builder logData, final Optional<Object> extraLog, - final ExecutionContext ctx) { - return parse(logData, extraLog); - } + LogSinkListener parse(LogMetadata metadata, Object input, ExecutionContext ctx); } diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/RecordSinkListener.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/RecordSinkListener.java index 11e741db7d..4d6e465e2b 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/RecordSinkListener.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/RecordSinkListener.java @@ -17,14 +17,13 @@ package org.apache.skywalking.oap.log.analyzer.v2.provider.log.listener; -import java.util.Optional; import lombok.SneakyThrows; -import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.oap.log.analyzer.v2.dsl.ExecutionContext; import org.apache.skywalking.oap.log.analyzer.v2.provider.LogAnalyzerModuleConfig; import org.apache.skywalking.oap.server.core.CoreModule; import org.apache.skywalking.oap.server.core.source.LALOutputBuilder; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.server.core.source.SourceReceiver; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.slf4j.Logger; @@ -34,7 +33,7 @@ import org.slf4j.LoggerFactory; * RecordSinkListener forwards LAL output to the persistence layer. * * <p>All LAL rules produce an {@link LALOutputBuilder} in the {@link ExecutionContext}. - * This listener calls {@code init()} to populate standard fields from LogData, + * This listener calls {@code init()} to populate standard fields from metadata, * then {@code complete()} to dispatch the final source(s). */ public class RecordSinkListener implements LogSinkListener { @@ -64,23 +63,14 @@ public class RecordSinkListener implements LogSinkListener { @Override @SneakyThrows - public LogSinkListener parse(final LogData.Builder logData, - final Optional<Object> extraLog) { - return this; - } - - @Override - @SneakyThrows - public LogSinkListener parse(final LogData.Builder logData, - final Optional<Object> extraLog, + public LogSinkListener parse(final LogMetadata metadata, + final Object input, final ExecutionContext ctx) { if (ctx == null || !(ctx.output() instanceof LALOutputBuilder)) { return this; } builder = ctx.outputAsBuilder(); - // Pass the input data matching the declared inputType: - // extraLog (e.g., HTTPAccessLogEntry) when present, otherwise LogData. - builder.init(logData.build(), extraLog, moduleManager); + builder.init(metadata, input, moduleManager); return this; } diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/TrafficSinkListener.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/TrafficSinkListener.java index 8a9ab2b6f2..9679a3354f 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/TrafficSinkListener.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/provider/log/listener/TrafficSinkListener.java @@ -17,10 +17,10 @@ package org.apache.skywalking.oap.log.analyzer.v2.provider.log.listener; -import java.util.Optional; import lombok.RequiredArgsConstructor; -import org.apache.skywalking.apm.network.logging.v3.LogData; +import org.apache.skywalking.oap.log.analyzer.v2.dsl.ExecutionContext; import org.apache.skywalking.oap.server.core.analysis.Layer; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.server.library.util.StringUtil; import org.apache.skywalking.oap.log.analyzer.v2.provider.LogAnalyzerModuleConfig; import org.apache.skywalking.oap.server.core.CoreModule; @@ -62,36 +62,37 @@ public class TrafficSinkListener implements LogSinkListener { } @Override - public LogSinkListener parse(final LogData.Builder logData, - final Optional<Object> extraLog) { + public LogSinkListener parse(final LogMetadata metadata, + final Object input, + final ExecutionContext ctx) { Layer layer; - if (StringUtil.isNotEmpty(logData.getLayer())) { - layer = Layer.valueOf(logData.getLayer()); + if (StringUtil.isNotEmpty(metadata.getLayer())) { + layer = Layer.valueOf(metadata.getLayer()); } else { layer = Layer.GENERAL; } final long timeBucket = TimeBucket.getTimeBucket(System.currentTimeMillis(), DownSampling.Minute); // to service traffic - String serviceName = namingControl.formatServiceName(logData.getService()); + String serviceName = namingControl.formatServiceName(metadata.getService()); String serviceId = IDManager.ServiceID.buildId(serviceName, layer.isNormal()); serviceMeta = new ServiceMeta(); - serviceMeta.setName(namingControl.formatServiceName(logData.getService())); + serviceMeta.setName(namingControl.formatServiceName(metadata.getService())); serviceMeta.setLayer(layer); serviceMeta.setTimeBucket(timeBucket); // to service instance traffic - if (StringUtil.isNotEmpty(logData.getServiceInstance())) { + if (StringUtil.isNotEmpty(metadata.getServiceInstance())) { instanceMeta = new ServiceInstanceUpdate(); instanceMeta.setServiceId(serviceId); - instanceMeta.setName(namingControl.formatInstanceName(logData.getServiceInstance())); + instanceMeta.setName(namingControl.formatInstanceName(metadata.getServiceInstance())); instanceMeta.setTimeBucket(timeBucket); } // to endpoint traffic - if (StringUtil.isNotEmpty(logData.getEndpoint())) { + if (StringUtil.isNotEmpty(metadata.getEndpoint())) { endpointMeta = new EndpointMeta(); endpointMeta.setServiceName(serviceName); endpointMeta.setServiceNormal(true); - endpointMeta.setEndpoint(namingControl.formatEndpointName(serviceName, logData.getEndpoint())); + endpointMeta.setEndpoint(namingControl.formatEndpointName(serviceName, metadata.getEndpoint())); endpointMeta.setTimeBucket(timeBucket); } return this; diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/spi/LALSourceTypeProvider.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/spi/LALSourceTypeProvider.java index d843b9159c..d9c4b7c818 100644 --- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/spi/LALSourceTypeProvider.java +++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/v2/spi/LALSourceTypeProvider.java @@ -23,8 +23,8 @@ import org.apache.skywalking.oap.server.core.analysis.Layer; * SPI for receiver plugins to declare the input and default output types for * LAL rules on a specific {@link Layer}. * - * <p><b>Input type</b> ({@link #inputType()}): The Java type of the {@code extraLog} - * passed to LAL via {@code ILogAnalyzerService.doAnalysis(LogData, Message)}. + * <p><b>Input type</b> ({@link #inputType()}): The Java type of the {@code input} + * passed to LAL via {@code ILogAnalyzerService.doAnalysis(LogMetadata, Object)}. * The LAL compiler uses this at compile time to generate optimized direct * getter calls for {@code parsed.*} field access. This is per-layer because * all rules for a layer share the same input proto type. @@ -62,7 +62,7 @@ public interface LALSourceTypeProvider { Layer layer(); /** - * The Java type passed as {@code extraLog} by the receiver plugin for + * The Java type passed as {@code input} by the receiver plugin for * this layer. The compiler resolves getter chains on this type at * compile time. */ diff --git a/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALClassGeneratorExtractorTest.java b/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALClassGeneratorExtractorTest.java index 8fa30a6adc..8fa8640ed8 100644 --- a/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALClassGeneratorExtractorTest.java +++ b/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALClassGeneratorExtractorTest.java @@ -117,7 +117,7 @@ class LALClassGeneratorExtractorTest extends LALClassGeneratorTestBase { } @Test - void compileNoParserFallsBackToLogDataProto() throws Exception { + void compileNoParserFallsBackToLogMetadata() throws Exception { final String dsl = "filter {\n" + " extractor {\n" @@ -127,12 +127,12 @@ class LALClassGeneratorExtractorTest extends LALClassGeneratorTestBase { + " sink {}\n" + "}"; final String source = generator.generateSource(dsl); - assertTrue(source.contains("h.ctx().log().getService()"), - "Expected h.ctx().log().getService() but got: " + source); - assertTrue(source.contains("h.ctx().log().getServiceInstance()"), - "Expected h.ctx().log().getServiceInstance() but got: " + source); + assertTrue(source.contains("h.ctx().metadata().getService()"), + "Expected h.ctx().metadata().getService() but got: " + source); + assertTrue(source.contains("h.ctx().metadata().getServiceInstance()"), + "Expected h.ctx().metadata().getServiceInstance() but got: " + source); assertFalse(source.contains("_p"), - "Should NOT have _p variable for LogData fallback but got: " + source); + "Should NOT have _p variable for LogMetadata fallback but got: " + source); compileAndAssert(dsl); } @@ -158,7 +158,7 @@ class LALClassGeneratorExtractorTest extends LALClassGeneratorTestBase { final String fqcn = "io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry"; assertTrue(source.contains( - fqcn + " _p = (" + fqcn + ") h.ctx().extraLog()"), + fqcn + " _p = (" + fqcn + ") h.ctx().input()"), "Expected _p local variable for inputType cast but got: " + source); assertTrue(source.contains("_p.getResponse()"), "Expected _p.getResponse() via cached variable but got: " + source); diff --git a/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALExpressionExecutionTest.java b/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALExpressionExecutionTest.java index 38e51069d5..07004a4d6f 100644 --- a/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALExpressionExecutionTest.java +++ b/oap-server/analyzer/log-analyzer/src/test/java/org/apache/skywalking/oap/log/analyzer/v2/compiler/LALExpressionExecutionTest.java @@ -39,6 +39,8 @@ import org.apache.skywalking.oap.log.analyzer.v2.module.LogAnalyzerModule; import org.apache.skywalking.oap.log.analyzer.v2.provider.LogAnalyzerModuleConfig; import org.apache.skywalking.oap.log.analyzer.v2.provider.LogAnalyzerModuleProvider; import org.apache.skywalking.oap.server.core.CoreModule; +import org.apache.skywalking.oap.server.core.source.LogMetadata; +import org.apache.skywalking.oap.server.core.source.LogMetadataUtils; import org.apache.skywalking.oap.server.core.config.NamingControl; import org.apache.skywalking.oap.server.core.config.group.EndpointNameGrouping; import org.apache.skywalking.oap.server.core.config.ConfigService; @@ -173,13 +175,13 @@ class LALExpressionExecutionTest { if (ruleLayer != null) { logData.setLayer(ruleLayer); } - final ExecutionContext ctx = new ExecutionContext(); - ctx.log(logData); + final LogMetadata metadata = LogMetadataUtils.fromLogData(logData); final Message extraLog = LalLogDataBuilder.buildExtraLog(input); - if (extraLog != null) { - ctx.extraLog(extraLog); - } + final Object lalInput = extraLog != null ? extraLog : logData; + + final ExecutionContext ctx = new ExecutionContext(); + ctx.init(metadata, lalInput); expr.execute(filterSpec, ctx); @@ -208,28 +210,28 @@ class LALExpressionExecutionTest { ruleName + ": shouldAbort mismatch"); break; case "service": - assertOutputField(ruleName, outputObj, "service", expected, ctx.log().getService()); + assertOutputField(ruleName, outputObj, "service", expected, ctx.metadata().getService()); break; case "instance": - assertOutputField(ruleName, outputObj, "serviceInstance", expected, ctx.log().getServiceInstance()); + assertOutputField(ruleName, outputObj, "serviceInstance", expected, ctx.metadata().getServiceInstance()); break; case "endpoint": - assertOutputField(ruleName, outputObj, "endpoint", expected, ctx.log().getEndpoint()); + assertOutputField(ruleName, outputObj, "endpoint", expected, ctx.metadata().getEndpoint()); break; case "layer": - assertOutputField(ruleName, outputObj, "layer", expected, ctx.log().getLayer()); + assertOutputField(ruleName, outputObj, "layer", expected, ctx.metadata().getLayer()); break; case "timestamp": - assertOutputField(ruleName, outputObj, "timestamp", expected, String.valueOf(ctx.log().getTimestamp())); + assertOutputField(ruleName, outputObj, "timestamp", expected, String.valueOf(ctx.metadata().getTimestamp())); break; default: if (key.startsWith("tag.")) { final String tagKey = key.substring(4); if (outputObj instanceof org.apache.skywalking.oap.server.core.source.LogBuilder) { assertLalTag(ruleName, outputObj, tagKey, expected); - } else { + } else if (ctx.input() instanceof LogData.Builder) { final List<KeyStringValuePair> tags = - ctx.log().getTags().getDataList(); + ((LogData.Builder) ctx.input()).getTags().getDataList(); assertTrue(tags.stream().anyMatch( t -> tagKey.equals(t.getKey()) && expected.equals(t.getValue())), diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LALOutputBuilder.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LALOutputBuilder.java index fad33dd9a8..e88ab85368 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LALOutputBuilder.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LALOutputBuilder.java @@ -18,8 +18,6 @@ package org.apache.skywalking.oap.server.core.source; -import java.util.Optional; -import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.oap.server.library.module.ModuleManager; /** @@ -53,13 +51,12 @@ public interface LALOutputBuilder { * Pre-populate standard fields before custom output fields are applied. * Called once per log entry. * - * @param logData log metadata (service, layer, timestamp, trace context, etc.) - * @param extraLog optional extra input whose type matches - * {@code LALSourceTypeProvider#inputType()} for the layer - * (e.g., {@code HTTPAccessLogEntry} for envoy access logs) + * @param metadata uniform metadata (service, layer, timestamp, trace context, etc.) + * @param input source-specific input object ({@code LogData} for standard logs, + * {@code HTTPAccessLogEntry} for envoy access logs, etc.) * @param moduleManager module manager for resolving services (e.g., NamingControl) */ - void init(LogData logData, Optional<Object> extraLog, ModuleManager moduleManager); + void init(LogMetadata metadata, Object input, ModuleManager moduleManager); /** * Validate the builder state and dispatch the final output source(s). diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogBuilder.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogBuilder.java index 96f7f7a294..a49ffc83cb 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogBuilder.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogBuilder.java @@ -23,7 +23,6 @@ import java.util.Arrays; import java.util.Collection; import java.util.HashSet; import java.util.List; -import java.util.Optional; import java.util.UUID; import lombok.Getter; import lombok.Setter; @@ -31,7 +30,6 @@ import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.apm.network.logging.v3.LogDataBody; -import org.apache.skywalking.apm.network.logging.v3.TraceContext; import org.apache.skywalking.oap.server.core.Const; import org.apache.skywalking.oap.server.core.CoreModule; import org.apache.skywalking.oap.server.core.analysis.IDManager; @@ -95,8 +93,21 @@ public class LogBuilder implements LALOutputBuilder { } @Override - public void init(final LogData logData, final Optional<Object> extraLog, + public void init(final LogMetadata metadata, final Object input, final ModuleManager moduleManager) { + ensureInitialized(moduleManager); + if (input instanceof LogData) { + this.logData = (LogData) input; + } else if (input instanceof LogData.Builder) { + this.logData = ((LogData.Builder) input).build(); + } + initFromMetadata(metadata); + } + + /** + * Initialize static services from ModuleManager (once per JVM). + */ + protected void ensureInitialized(final ModuleManager moduleManager) { if (!INITIALIZED) { NAMING_CONTROL = moduleManager.find(CoreModule.NAME) .provider() @@ -108,33 +119,39 @@ public class LogBuilder implements LALOutputBuilder { configService.getSearchableLogsTags().split(Const.COMMA)); INITIALIZED = true; } - this.logData = logData; - // Only populate fields that were NOT already set by the LAL extractor. - // The extractor runs before init(), so extractor values take priority. + } + + /** + * Populate fields from metadata. Only sets fields not already set by + * the LAL extractor (extractor runs before init, so its values take priority). + */ + protected void initFromMetadata(final LogMetadata metadata) { if (this.service == null) { - this.service = logData.getService(); + this.service = metadata.getService(); } if (this.serviceInstance == null) { - this.serviceInstance = logData.getServiceInstance(); + this.serviceInstance = metadata.getServiceInstance(); } if (this.endpoint == null) { - this.endpoint = logData.getEndpoint(); + this.endpoint = metadata.getEndpoint(); } if (this.layer == null) { - this.layer = logData.getLayer(); - } - final TraceContext tc = logData.getTraceContext(); - if (this.traceId == null) { - this.traceId = tc.getTraceId(); - } - if (this.segmentId == null) { - this.segmentId = tc.getTraceSegmentId(); + this.layer = metadata.getLayer(); } - if (this.spanId < 0) { - this.spanId = tc.getSpanId(); + final LogMetadata.TraceContext tc = metadata.getTraceContext(); + if (tc != null) { + if (this.traceId == null) { + this.traceId = tc.getTraceId(); + } + if (this.segmentId == null) { + this.segmentId = tc.getTraceSegmentId(); + } + if (this.spanId < 0) { + this.spanId = tc.getSpanId(); + } } if (this.timestamp == 0) { - this.timestamp = logData.getTimestamp(); + this.timestamp = metadata.getTimestamp(); } } @@ -179,21 +196,23 @@ public class LogBuilder implements LALOutputBuilder { log.setSpanId(spanId); } } - // content - final LogDataBody body = logData.getBody(); - if (body.hasText()) { - log.setContentType(ContentType.TEXT); - log.setContent(body.getText().getText()); - } else if (body.hasYaml()) { - log.setContentType(ContentType.YAML); - log.setContent(body.getYaml().getYaml()); - } else if (body.hasJson()) { - log.setContentType(ContentType.JSON); - log.setContent(body.getJson().getJson()); - } - // raw tags from original LogData - if (logData.getTags().getDataCount() > 0) { - log.setTagsRawData(logData.getTags().toByteArray()); + // content (only when input is LogData) + if (logData != null) { + final LogDataBody body = logData.getBody(); + if (body.hasText()) { + log.setContentType(ContentType.TEXT); + log.setContent(body.getText().getText()); + } else if (body.hasYaml()) { + log.setContentType(ContentType.YAML); + log.setContent(body.getYaml().getYaml()); + } else if (body.hasJson()) { + log.setContentType(ContentType.JSON); + log.setContent(body.getJson().getJson()); + } + // raw tags from original LogData + if (logData.getTags().getDataCount() > 0) { + log.setTagsRawData(logData.getTags().toByteArray()); + } } // searchable tags from LogData + LAL-added tags log.getTags().addAll(collectSearchableTags()); @@ -205,11 +224,13 @@ public class LogBuilder implements LALOutputBuilder { final HashSet<Tag> result = new HashSet<>(); if (SEARCHABLE_TAG_KEYS != null) { // Tags from original LogData - logData.getTags().getDataList().forEach(kv -> { - if (SEARCHABLE_TAG_KEYS.contains(kv.getKey())) { - addSearchableTag(result, kv.getKey(), kv.getValue()); - } - }); + if (logData != null) { + logData.getTags().getDataList().forEach(kv -> { + if (SEARCHABLE_TAG_KEYS.contains(kv.getKey())) { + addSearchableTag(result, kv.getKey(), kv.getValue()); + } + }); + } // Tags added by LAL extractor for (final String[] kv : lalTags) { if (SEARCHABLE_TAG_KEYS.contains(kv[0])) { diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogMetadata.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogMetadata.java new file mode 100644 index 0000000000..f8aeae49a2 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogMetadata.java @@ -0,0 +1,46 @@ +/* + * 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.skywalking.oap.server.core.source; + +import lombok.Builder; +import lombok.Data; + +/** + * Uniform metadata carrier for log analysis. Receivers build this from their + * native source (LogData proto for standard logs, ALS context for envoy, etc.). + * + * <p>For building from {@code LogData} protobuf, see {@link LogMetadataUtils}. + */ +@Data +@Builder +public class LogMetadata { + private String service; + private String serviceInstance; + private String endpoint; + private String layer; + private long timestamp; + private TraceContext traceContext; + + @Data + @Builder + public static class TraceContext { + private String traceId; + private String traceSegmentId; + private int spanId; + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogMetadataUtils.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogMetadataUtils.java new file mode 100644 index 0000000000..db40a38ba7 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/LogMetadataUtils.java @@ -0,0 +1,61 @@ +/* + * 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.skywalking.oap.server.core.source; + +import org.apache.skywalking.apm.network.logging.v3.LogData; +import org.apache.skywalking.apm.network.logging.v3.TraceContext; + +/** + * Utility for building {@link LogMetadata} from protobuf {@link LogData}. + */ +public final class LogMetadataUtils { + + private LogMetadataUtils() { + } + + /** + * Build LogMetadata from a LogData protobuf message. + */ + public static LogMetadata fromLogData(final LogData logData) { + LogMetadata.TraceContext tc = null; + if (logData.hasTraceContext()) { + final TraceContext ctxProto = + logData.getTraceContext(); + tc = LogMetadata.TraceContext.builder() + .traceId(ctxProto.getTraceId()) + .traceSegmentId(ctxProto.getTraceSegmentId()) + .spanId(ctxProto.getSpanId()) + .build(); + } + return LogMetadata.builder() + .service(logData.getService()) + .serviceInstance(logData.getServiceInstance()) + .endpoint(logData.getEndpoint()) + .layer(logData.getLayer()) + .timestamp(logData.getTimestamp()) + .traceContext(tc) + .build(); + } + + /** + * Build LogMetadata from a LogData.Builder. + */ + public static LogMetadata fromLogData(final LogData.Builder builder) { + return fromLogData(builder.build()); + } +} diff --git a/oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/provider/handler/LogHandler.java b/oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/provider/handler/LogHandler.java index c39925fad5..18e1ca6351 100644 --- a/oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/provider/handler/LogHandler.java +++ b/oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/provider/handler/LogHandler.java @@ -20,8 +20,8 @@ package org.apache.skywalking.oap.server.analyzer.agent.kafka.provider.handler; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.common.utils.Bytes; -import java.util.Optional; import org.apache.skywalking.apm.network.logging.v3.LogData; +import org.apache.skywalking.oap.server.core.source.LogMetadataUtils; import org.apache.skywalking.oap.log.analyzer.v2.module.LogAnalyzerModule; import org.apache.skywalking.oap.log.analyzer.v2.provider.log.ILogAnalyzerService; import org.apache.skywalking.oap.server.analyzer.agent.kafka.module.KafkaFetcherConfig; @@ -71,8 +71,8 @@ public class LogHandler extends AbstractKafkaHandler { @Override public void handle(final ConsumerRecord<String, Bytes> record) { try (HistogramMetrics.Timer ignore = histogram.createTimer()) { - LogData logData = parseConsumerRecord(record); - logAnalyzerService.doAnalysis(logData, Optional.empty()); + LogData.Builder builder = parseConsumerRecord(record).toBuilder(); + logAnalyzerService.doAnalysis(LogMetadataUtils.fromLogData(builder), builder); } catch (Exception e) { errorCounter.inc(); log.error(e.getMessage(), e); diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/LogTestQuery.java b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/LogTestQuery.java index 8e3b481208..461fd62080 100644 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/LogTestQuery.java +++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/LogTestQuery.java @@ -29,6 +29,7 @@ import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.apm.network.logging.v3.LogTags; import org.apache.skywalking.oap.log.analyzer.v2.dsl.ExecutionContext; import org.apache.skywalking.oap.log.analyzer.v2.dsl.DSL; +import org.apache.skywalking.oap.server.core.source.LogMetadataUtils; import org.apache.skywalking.oap.log.analyzer.v2.module.LogAnalyzerModule; import org.apache.skywalking.oap.log.analyzer.v2.provider.LogAnalyzerModuleConfig; import org.apache.skywalking.oap.log.analyzer.v2.provider.LogAnalyzerModuleProvider; @@ -39,6 +40,8 @@ import org.apache.skywalking.oap.query.graphql.type.Metrics; import org.apache.skywalking.oap.server.core.analysis.IDManager; import org.apache.skywalking.oap.server.core.query.type.KeyValue; import org.apache.skywalking.oap.server.core.query.type.Log; +import org.apache.skywalking.oap.server.core.source.LogBuilder; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.library.util.ProtoBufJsonUtils; @@ -72,17 +75,21 @@ public class LogTestQuery implements GraphQLQueryResolver { final LogData.Builder log = LogData.newBuilder(); ProtoBufJsonUtils.fromJSON(request.getLog(), log); - ctx.log(log); + final LogMetadata metadata = LogMetadataUtils.fromLogData(log); + ctx.init(metadata, log); - ctx.captureLog(true); + ctx.dryRun(true); ctx.metricsContainer(new ArrayList<>()); dsl.evaluate(ctx); final LogTestResponse.LogTestResponseBuilder builder = LogTestResponse.builder(); - ctx.logContainer().ifPresent(it -> { - final Log l = new Log(); + if (ctx.shouldSave() && ctx.outputAsBuilder() instanceof LogBuilder) { + final LogBuilder logBuilder = (LogBuilder) ctx.outputAsBuilder(); + logBuilder.init(metadata, log, moduleManager); + final org.apache.skywalking.oap.server.core.source.Log it = logBuilder.toLog(); + final Log l = new Log(); if (isNotBlank(it.getServiceId())) { l.setServiceName(IDManager.ServiceID.analysisId(it.getServiceId()).getName()); } @@ -113,9 +120,8 @@ public class LogTestQuery implements GraphQLQueryResolver { // ignore } } - builder.log(l); - }); + } ctx.metricsContainer().ifPresent(it -> { final List<Metrics> samples = it.stream() diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/EnvoyAccessLogBuilder.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/EnvoyAccessLogBuilder.java index 60ba08618c..9478939481 100644 --- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/EnvoyAccessLogBuilder.java +++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/EnvoyAccessLogBuilder.java @@ -19,12 +19,11 @@ package org.apache.skywalking.oap.server.receiver.envoy.persistence; import com.google.protobuf.Message; -import java.util.Optional; import lombok.SneakyThrows; -import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.oap.server.core.query.type.ContentType; import org.apache.skywalking.oap.server.core.source.Log; import org.apache.skywalking.oap.server.core.source.LogBuilder; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.library.util.ProtoBufJsonUtils; @@ -35,9 +34,9 @@ import org.apache.skywalking.oap.server.library.util.ProtoBufJsonUtils; * as JSON content. The serialization is deferred to {@link #toLog()} so that * it only happens when the log is actually persisted (after LAL filtering). * - * <p>The {@link #init} method stores the extra log entry (e.g., - * {@code HTTPAccessLogEntry}) for JSON serialization, then delegates to - * the base class for standard field population from {@code LogData}. + * <p>The {@link #init} method stores the access log entry for JSON + * serialization, then delegates to the base class for metadata-only + * field population (no LogData in the envoy path). */ public class EnvoyAccessLogBuilder extends LogBuilder { public static final String NAME = "EnvoyAccessLog"; @@ -50,10 +49,13 @@ public class EnvoyAccessLogBuilder extends LogBuilder { } @Override - public void init(final LogData logData, final Optional<Object> extraLog, + public void init(final LogMetadata metadata, final Object input, final ModuleManager moduleManager) { - extraLog.ifPresent(entry -> this.accessLogEntry = entry); - super.init(logData, extraLog, moduleManager); + if (input != null) { + this.accessLogEntry = input; + } + ensureInitialized(moduleManager); + initFromMetadata(metadata); } @Override diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/LogsPersistence.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/LogsPersistence.java index 09437cbe3b..f2ac92c15f 100644 --- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/LogsPersistence.java +++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/LogsPersistence.java @@ -20,13 +20,12 @@ package org.apache.skywalking.oap.server.receiver.envoy.persistence; import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry; import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage; -import java.util.Optional; import lombok.extern.slf4j.Slf4j; -import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetric; import org.apache.skywalking.oap.log.analyzer.v2.module.LogAnalyzerModule; import org.apache.skywalking.oap.log.analyzer.v2.provider.log.ILogAnalyzerService; import org.apache.skywalking.oap.server.core.analysis.Layer; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.library.module.ModuleStartException; import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig; @@ -66,8 +65,16 @@ public class LogsPersistence implements ALSHTTPAnalysis { return result; } - final LogData logData = convertToLogData(entry, result); - logAnalyzerService.doAnalysis(logData, Optional.of(entry)); + final ServiceMetaInfo service = result.getService(); + final HTTPServiceMeshMetric.Builder metrics = + new LogEntry2MetricsAdapter(entry, null, null).adaptCommonPart(); + final LogMetadata metadata = LogMetadata.builder() + .service(service.getServiceName()) + .serviceInstance(service.getServiceInstanceName()) + .timestamp(metrics.getEndTime()) + .layer(Layer.MESH.name()) + .build(); + logAnalyzerService.doAnalysis(metadata, entry); } catch (final Exception e) { log.error("Failed to persist Envoy access log", e); } @@ -78,19 +85,4 @@ public class LogsPersistence implements ALSHTTPAnalysis { public Role identify(final StreamAccessLogsMessage.Identifier alsIdentifier, final Role prev) { return prev; } - - public LogData convertToLogData(final HTTPAccessLogEntry logEntry, final Result result) { - final ServiceMetaInfo service = result.getService(); - - final HTTPServiceMeshMetric.Builder metrics = - new LogEntry2MetricsAdapter(logEntry, null, null).adaptCommonPart(); - - return LogData - .newBuilder() - .setService(service.getServiceName()) - .setServiceInstance(service.getServiceInstanceName()) - .setTimestamp(metrics.getEndTime()) - .setLayer(Layer.MESH.name()) - .build(); - } } diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/TCPLogsPersistence.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/TCPLogsPersistence.java index ac66002d1f..be942164d8 100644 --- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/TCPLogsPersistence.java +++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/TCPLogsPersistence.java @@ -18,12 +18,11 @@ package org.apache.skywalking.oap.server.receiver.envoy.persistence; -import java.util.Optional; -import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.apm.network.servicemesh.v3.TCPServiceMeshMetric; import org.apache.skywalking.oap.log.analyzer.v2.module.LogAnalyzerModule; import org.apache.skywalking.oap.log.analyzer.v2.provider.log.ILogAnalyzerService; import org.apache.skywalking.oap.server.core.analysis.Layer; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.library.module.ModuleStartException; import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig; @@ -66,8 +65,16 @@ public class TCPLogsPersistence implements TCPAccessLogAnalyzer { return result; } - final LogData logData = convertToLogData(entry, result); - logAnalyzerService.doAnalysis(logData, Optional.of(entry)); + final ServiceMetaInfo service = result.getService(); + final TCPServiceMeshMetric.Builder metrics = + new TCPLogEntry2MetricsAdapter(entry, null, null).adaptCommonPart(); + final LogMetadata metadata = LogMetadata.builder() + .service(service.getServiceName()) + .serviceInstance(service.getServiceInstanceName()) + .timestamp(metrics.getEndTime()) + .layer(Layer.MESH.name()) + .build(); + logAnalyzerService.doAnalysis(metadata, entry); } catch (final Exception e) { log.error("Failed to persist Envoy access log", e); } @@ -78,19 +85,4 @@ public class TCPLogsPersistence implements TCPAccessLogAnalyzer { public Role identify(final StreamAccessLogsMessage.Identifier alsIdentifier, final Role prev) { return prev; } - - public LogData convertToLogData(final TCPAccessLogEntry logEntry, final Result result) { - final ServiceMetaInfo service = result.getService(); - - final TCPServiceMeshMetric.Builder metrics = - new TCPLogEntry2MetricsAdapter(logEntry, null, null).adaptCommonPart(); - - return LogData - .newBuilder() - .setService(service.getServiceName()) - .setServiceInstance(service.getServiceInstanceName()) - .setTimestamp(metrics.getEndTime()) - .setLayer(Layer.MESH.name()) - .build(); - } } diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/EnvoyAccessLogBuilderTest.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/EnvoyAccessLogBuilderTest.java index 7bc3b0f89d..2c3b39536f 100644 --- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/EnvoyAccessLogBuilderTest.java +++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/EnvoyAccessLogBuilderTest.java @@ -23,8 +23,7 @@ import io.envoyproxy.envoy.data.accesslog.v3.AccessLogCommon; import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry; import io.envoyproxy.envoy.data.accesslog.v3.HTTPResponseProperties; import java.util.Collections; -import java.util.Optional; -import org.apache.skywalking.apm.network.logging.v3.LogData; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.log.analyzer.v2.compiler.LALClassGenerator; import org.apache.skywalking.oap.log.analyzer.v2.dsl.ExecutionContext; import org.apache.skywalking.oap.log.analyzer.v2.dsl.LalExpression; @@ -87,11 +86,11 @@ class EnvoyAccessLogBuilderTest { builder.setService("test-svc"); builder.setTimestamp(1609459200000L); - final LogData logData = LogData.newBuilder() - .setService("test-svc") - .setTimestamp(1609459200000L) + final LogMetadata metadata = LogMetadata.builder() + .service("test-svc") + .timestamp(1609459200000L) .build(); - builder.init(logData, Optional.of(entry), moduleManager); + builder.init(metadata, entry, moduleManager); final Log log = builder.toLog(); @@ -109,12 +108,12 @@ class EnvoyAccessLogBuilderTest { builder.setService("test-svc"); builder.setTimestamp(1609459200000L); - final LogData logData = LogData.newBuilder() - .setService("test-svc") - .setTimestamp(1609459200000L) + final LogMetadata metadata = LogMetadata.builder() + .service("test-svc") + .timestamp(1609459200000L) .build(); // Pass a default (empty) entry — no response code, so toLog() serializes empty JSON - builder.init(logData, Optional.of(HTTPAccessLogEntry.getDefaultInstance()), moduleManager); + builder.init(metadata, HTTPAccessLogEntry.getDefaultInstance(), moduleManager); final Log log = builder.toLog(); @@ -159,12 +158,13 @@ class EnvoyAccessLogBuilderTest { final LalExpression expr = generator.compile(dsl); - // Build LogData with service/instance (simulates what LogsPersistence creates) - final LogData.Builder logData = LogData.newBuilder() - .setService("envoy-test-svc") - .setServiceInstance("envoy-test-instance") - .setTimestamp(1609459200000L) - .setLayer("MESH"); + // Build LogMetadata (simulates what LogsPersistence creates) + final LogMetadata metadata = LogMetadata.builder() + .service("envoy-test-svc") + .serviceInstance("envoy-test-instance") + .timestamp(1609459200000L) + .layer("MESH") + .build(); // Build HTTPAccessLogEntry with response code 503 final HTTPAccessLogEntry entry = HTTPAccessLogEntry.newBuilder() @@ -177,8 +177,7 @@ class EnvoyAccessLogBuilderTest { // Execute final FilterSpec filterSpec = buildFilterSpec(); final ExecutionContext ctx = new ExecutionContext(); - ctx.log(logData); - ctx.extraLog(entry); + ctx.init(metadata, entry); expr.execute(filterSpec, ctx); // Verify output is EnvoyAccessLogBuilder @@ -196,7 +195,7 @@ class EnvoyAccessLogBuilderTest { when(cs.getSearchableLogsTags()).thenReturn("status.code,svc"); // Reset static initialized flag so the new config takes effect resetLogBuilderState(); - output.init(logData.build(), Optional.of(entry), testMm); + output.init(metadata, entry, testMm); final Log log = output.toLog(); assertTrue(log.getTags().stream().anyMatch( diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/EnvoyAlsLalTest.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/EnvoyAlsLalTest.java index 9a9cb6a96e..050fdae597 100644 --- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/EnvoyAlsLalTest.java +++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/EnvoyAlsLalTest.java @@ -26,9 +26,8 @@ import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry; import io.envoyproxy.envoy.data.accesslog.v3.HTTPResponseProperties; import java.lang.reflect.Field; import java.util.Collections; -import java.util.Optional; import javassist.ClassPool; -import org.apache.skywalking.apm.network.logging.v3.LogData; +import org.apache.skywalking.oap.server.core.source.LogMetadata; import org.apache.skywalking.oap.log.analyzer.v2.compiler.LALClassGenerator; import org.apache.skywalking.oap.log.analyzer.v2.dsl.ExecutionContext; import org.apache.skywalking.oap.log.analyzer.v2.dsl.LalExpression; @@ -306,15 +305,15 @@ class EnvoyAlsLalTest { private ExecutionContext execute( final LalExpression expr, final HTTPAccessLogEntry entry) throws Exception { - final LogData.Builder logData = LogData.newBuilder() - .setService("als-test-svc") - .setTimestamp(1609459200000L) - .setLayer("MESH"); + final LogMetadata metadata = LogMetadata.builder() + .service("als-test-svc") + .timestamp(1609459200000L) + .layer("MESH") + .build(); final FilterSpec filterSpec = buildFilterSpec(); final ExecutionContext ctx = new ExecutionContext(); - ctx.log(logData); - ctx.extraLog(entry); + ctx.init(metadata, entry); expr.execute(filterSpec, ctx); return ctx; } @@ -332,7 +331,7 @@ class EnvoyAlsLalTest { resetLogBuilderState(); final ModuleManager tagMm = buildCoreModuleManager( String.join(",", searchableTagKeys)); - output.init(ctx.log().build(), Optional.of(entry), tagMm); + output.init(ctx.metadata(), entry, tagMm); return output.toLog(); } diff --git a/oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHandler.java b/oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHandler.java index b2b526eef8..9bebac1077 100644 --- a/oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHandler.java +++ b/oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryLogHandler.java @@ -25,7 +25,6 @@ import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceResponse; import io.opentelemetry.proto.collector.logs.v1.LogsServiceGrpc; import io.opentelemetry.proto.common.v1.KeyValue; import io.opentelemetry.proto.logs.v1.LogRecord; -import java.util.Optional; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import lombok.Getter; @@ -34,6 +33,7 @@ import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.apm.network.logging.v3.LogDataBody; import org.apache.skywalking.apm.network.logging.v3.LogTags; import org.apache.skywalking.apm.network.logging.v3.TextLog; +import org.apache.skywalking.oap.server.core.source.LogMetadataUtils; import org.apache.skywalking.oap.log.analyzer.v2.module.LogAnalyzerModule; import org.apache.skywalking.oap.log.analyzer.v2.provider.log.ILogAnalyzerService; import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegister; @@ -122,16 +122,15 @@ public class OpenTelemetryLogHandler private void doAnalysisQuietly(String service, String layer, String serviceInstance, LogRecord logRecord) { try { - logAnalyzerService().doAnalysis( - LogData - .newBuilder() - .setService(service) - .setServiceInstance(serviceInstance) - .setTimestamp(logRecord.getTimeUnixNano() / 1_000_000) - .setTags(buildTags(logRecord)) - .setBody(buildBody(logRecord)) - .setLayer(layer), - Optional.empty()); + final LogData.Builder builder = LogData + .newBuilder() + .setService(service) + .setServiceInstance(serviceInstance) + .setTimestamp(logRecord.getTimeUnixNano() / 1_000_000) + .setTags(buildTags(logRecord)) + .setBody(buildBody(logRecord)) + .setLayer(layer); + logAnalyzerService().doAnalysis(LogMetadataUtils.fromLogData(builder), builder); } catch (Exception e) { log.error("Failed to analyze logs", e); } diff --git a/oap-server/server-receiver-plugin/skywalking-log-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/log/provider/handler/grpc/LogReportServiceGrpcHandler.java b/oap-server/server-receiver-plugin/skywalking-log-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/log/provider/handler/grpc/LogReportServiceGrpcHandler.java index d243142b60..4d7563fd09 100644 --- a/oap-server/server-receiver-plugin/skywalking-log-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/log/provider/handler/grpc/LogReportServiceGrpcHandler.java +++ b/oap-server/server-receiver-plugin/skywalking-log-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/log/provider/handler/grpc/LogReportServiceGrpcHandler.java @@ -18,11 +18,11 @@ package org.apache.skywalking.oap.server.receiver.log.provider.handler.grpc; import io.grpc.stub.StreamObserver; -import java.util.Optional; import lombok.extern.slf4j.Slf4j; import org.apache.skywalking.apm.network.common.v3.Commands; import org.apache.skywalking.apm.network.logging.v3.LogData; import org.apache.skywalking.apm.network.logging.v3.LogReportServiceGrpc; +import org.apache.skywalking.oap.server.core.source.LogMetadataUtils; import org.apache.skywalking.oap.server.library.util.StringUtil; import org.apache.skywalking.oap.log.analyzer.v2.module.LogAnalyzerModule; import org.apache.skywalking.oap.log.analyzer.v2.provider.log.ILogAnalyzerService; @@ -93,7 +93,7 @@ public class LogReportServiceGrpcHandler extends LogReportServiceGrpc.LogReportS try { LogData.Builder builder = logData.toBuilder(); setServiceName(builder); - logAnalyzerService.doAnalysis(builder, Optional.empty()); + logAnalyzerService.doAnalysis(LogMetadataUtils.fromLogData(builder), builder); } catch (Exception e) { errorCounter.inc(); log.error(e.getMessage(), e); diff --git a/oap-server/server-receiver-plugin/skywalking-log-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/log/provider/handler/rest/LogReportServiceHTTPHandler.java b/oap-server/server-receiver-plugin/skywalking-log-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/log/provider/handler/rest/LogReportServiceHTTPHandler.java index 32a3a761fc..bcb64282ec 100644 --- a/oap-server/server-receiver-plugin/skywalking-log-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/log/provider/handler/rest/LogReportServiceHTTPHandler.java +++ b/oap-server/server-receiver-plugin/skywalking-log-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/log/provider/handler/rest/LogReportServiceHTTPHandler.java @@ -19,10 +19,10 @@ package org.apache.skywalking.oap.server.receiver.log.provider.handler.rest; import com.linecorp.armeria.server.annotation.Post; import java.util.List; -import java.util.Optional; import lombok.extern.slf4j.Slf4j; import org.apache.skywalking.apm.network.common.v3.Commands; import org.apache.skywalking.apm.network.logging.v3.LogData; +import org.apache.skywalking.oap.server.core.source.LogMetadataUtils; import org.apache.skywalking.oap.log.analyzer.v2.module.LogAnalyzerModule; import org.apache.skywalking.oap.log.analyzer.v2.provider.log.ILogAnalyzerService; import org.apache.skywalking.oap.server.library.module.ModuleManager; @@ -64,7 +64,10 @@ public class LogReportServiceHTTPHandler { @Post("/v3/logs") public Commands collectLogs(final List<LogData> logs) { try (final HistogramMetrics.Timer ignored = histogram.createTimer()) { - logs.forEach(it -> logAnalyzerService.doAnalysis(it, Optional.empty())); + logs.forEach(it -> { + final LogData.Builder builder = it.toBuilder(); + logAnalyzerService.doAnalysis(LogMetadataUtils.fromLogData(builder), builder); + }); return Commands.newBuilder().build(); } catch (final Throwable e) { errorCounter.inc(); diff --git a/test/script-cases/script-runtime-with-groovy/mal-lal-v1-v2-checker/src/test/java/org/apache/skywalking/oap/server/checker/lal/LalBenchmark.java b/test/script-cases/script-runtime-with-groovy/mal-lal-v1-v2-checker/src/test/java/org/apache/skywalking/oap/server/checker/lal/LalBenchmark.java index b8178394b2..c077a17db0 100644 --- a/test/script-cases/script-runtime-with-groovy/mal-lal-v1-v2-checker/src/test/java/org/apache/skywalking/oap/server/checker/lal/LalBenchmark.java +++ b/test/script-cases/script-runtime-with-groovy/mal-lal-v1-v2-checker/src/test/java/org/apache/skywalking/oap/server/checker/lal/LalBenchmark.java @@ -239,12 +239,14 @@ public class LalBenchmark { public void executeV2(final Blackhole bh) { for (int i = 0; i < v2Exprs.size(); i++) { try { + final LogData logData = testLogs.get(i); + final Message extraLog = extraLogs.get(i); + final org.apache.skywalking.oap.server.core.source.LogMetadata metadata = + org.apache.skywalking.oap.server.core.source.LogMetadataUtils.fromLogData(logData); + final Object input = extraLog != null ? extraLog : logData; final org.apache.skywalking.oap.log.analyzer.v2.dsl.ExecutionContext ctx = - new org.apache.skywalking.oap.log.analyzer.v2.dsl.ExecutionContext() - .log(testLogs.get(i)); - if (extraLogs.get(i) != null) { - ctx.extraLog(extraLogs.get(i)); - } + new org.apache.skywalking.oap.log.analyzer.v2.dsl.ExecutionContext(); + ctx.init(metadata, input); v2Exprs.get(i).execute(v2FilterSpec, ctx); bh.consume(ctx); } catch (Exception ignored) { diff --git a/test/script-cases/script-runtime-with-groovy/mal-lal-v1-v2-checker/src/test/java/org/apache/skywalking/oap/server/checker/lal/LalComparisonTest.java b/test/script-cases/script-runtime-with-groovy/mal-lal-v1-v2-checker/src/test/java/org/apache/skywalking/oap/server/checker/lal/LalComparisonTest.java index 70b6a104d4..437dfdd54b 100644 --- a/test/script-cases/script-runtime-with-groovy/mal-lal-v1-v2-checker/src/test/java/org/apache/skywalking/oap/server/checker/lal/LalComparisonTest.java +++ b/test/script-cases/script-runtime-with-groovy/mal-lal-v1-v2-checker/src/test/java/org/apache/skywalking/oap/server/checker/lal/LalComparisonTest.java @@ -165,12 +165,13 @@ class LalComparisonTest { final String v2CompileError, final boolean v2Only) throws Exception { - final LogData testLog; + final LogData.Builder testLogBuilder; if (inputData != null) { - testLog = LalLogDataBuilder.buildLogData(inputData).build(); + testLogBuilder = LalLogDataBuilder.buildLogData(inputData); } else { - testLog = LalLogDataBuilder.buildSyntheticLogData(dsl); + testLogBuilder = LalLogDataBuilder.buildSyntheticLogData(dsl).toBuilder(); } + final LogData testLog = testLogBuilder.build(); // Build proto extraLog from input data if available final Message extraLog = inputData != null @@ -215,10 +216,11 @@ class LalComparisonTest { v2Manager, new LogAnalyzerModuleConfig()); disableSinkListenersOnSpec(v2FilterSpec); - v2Ctx = new org.apache.skywalking.oap.log.analyzer.v2.dsl.ExecutionContext().log(testLog); - if (extraLog != null) { - v2Ctx.extraLog(extraLog); - } + final org.apache.skywalking.oap.server.core.source.LogMetadata v2Metadata = + org.apache.skywalking.oap.server.core.source.LogMetadataUtils.fromLogData(testLog); + final Object v2Input = extraLog != null ? extraLog : testLogBuilder; + v2Ctx = new org.apache.skywalking.oap.log.analyzer.v2.dsl.ExecutionContext(); + v2Ctx.init(v2Metadata, v2Input); v2Expr.execute(v2FilterSpec, v2Ctx); } catch (Exception e) { @@ -247,25 +249,26 @@ class LalComparisonTest { // If the extractor sets a field, v1 puts it on LogData, v2 puts it on LogBuilder. // If the extractor doesn't set a field, both keep the original LogData value. final Object v2Output = v2Ctx.output(); + final org.apache.skywalking.oap.server.core.source.LogMetadata v2Meta = v2Ctx.metadata(); if (v2Output instanceof org.apache.skywalking.oap.server.core.source.LogBuilder) { final org.apache.skywalking.oap.server.core.source.LogBuilder v2Builder = (org.apache.skywalking.oap.server.core.source.LogBuilder) v2Output; assertV2Field(testName, "service", v1Log.getService(), - getFieldValue(v2Builder, "service", null), v2Ctx.log().getService()); + getFieldValue(v2Builder, "service", null), v2Meta.getService()); assertV2Field(testName, "serviceInstance", v1Log.getServiceInstance(), getFieldValue(v2Builder, "serviceInstance", null), - v2Ctx.log().getServiceInstance()); + v2Meta.getServiceInstance()); assertV2Field(testName, "endpoint", v1Log.getEndpoint(), - getFieldValue(v2Builder, "endpoint", null), v2Ctx.log().getEndpoint()); + getFieldValue(v2Builder, "endpoint", null), v2Meta.getEndpoint()); assertV2Field(testName, "layer", v1Log.getLayer(), - getFieldValue(v2Builder, "layer", null), v2Ctx.log().getLayer()); + getFieldValue(v2Builder, "layer", null), v2Meta.getLayer()); final long v2Ts = v2Builder.getTimestamp(); - assertEquals(v1Log.getTimestamp(), v2Ts != 0 ? v2Ts : v2Ctx.log().getTimestamp(), + assertEquals(v1Log.getTimestamp(), v2Ts != 0 ? v2Ts : v2Meta.getTimestamp(), testName + ": timestamp mismatch"); // Compare tags: v1 stores in LogData.tags, v2 stores in LogBuilder.lalTags compareV1TagsToV2Builder(testName, v1Log, v2Builder); - } else { - final LogData.Builder v2Log = v2Ctx.log(); + } else if (v2Ctx.input() instanceof LogData.Builder) { + final LogData.Builder v2Log = (LogData.Builder) v2Ctx.input(); assertEquals(v1Log.getService(), v2Log.getService(), testName + ": service mismatch"); assertEquals(v1Log.getServiceInstance(), v2Log.getServiceInstance(), @@ -278,6 +281,16 @@ class LalComparisonTest { testName + ": timestamp mismatch"); assertEquals(v1Log.getTags(), v2Log.getTags(), testName + ": tags mismatch"); + } else { + // ALS case: no LogData, compare metadata + assertEquals(v1Log.getService(), v2Meta.getService(), + testName + ": service mismatch"); + assertEquals(v1Log.getServiceInstance(), v2Meta.getServiceInstance(), + testName + ": serviceInstance mismatch"); + assertEquals(v1Log.getLayer(), v2Meta.getLayer(), + testName + ": layer mismatch"); + assertEquals(v1Log.getTimestamp(), v2Meta.getTimestamp(), + testName + ": timestamp mismatch"); } } @@ -287,7 +300,8 @@ class LalComparisonTest { final Map<String, Object> expect = (Map<String, Object>) inputData.get("expect"); if (expect != null) { - final LogData.Builder v2Log = v2Ctx.log(); + final LogData.Builder v2Log = v2Ctx.input() instanceof LogData.Builder + ? (LogData.Builder) v2Ctx.input() : null; validateExpected(testName, v2Ctx, v2Log, expect); } }
