[SWIP-17] [DISCUSS] Move the Native Trace Analysis from Java Listeners to TAL

[wu-sheng]

This SWIP moves SkyWalking's native trace analysis, the Java listeners in agent-analyzer, to TAL rules. It is a
port: no output change and no logic change, with one documented exception. This post is the general design; the first
comment holds the complete TAL rule files of the port. The full text, about twice this post, lands as
docs/en/swip/SWIP-17/SWIP.md with the rule files once the design is agreed.

Motivation

SkyWalking analyzes traces with two engines. Spans that arrive over OTLP are analyzed by
TAL rules, which an operator can read and edit. Segments that SkyWalking's own
agents send are analyzed by Java listeners in the agent-analyzer module. Those listeners turn each span into OAL
sources (Service, ServiceInstance, Endpoint, the relations, DatabaseAccess and others) for core.oal and the
other .oal files, write the trace itself, and hand GenAI spans to their analyzers. Changing how a native span becomes
a service, an endpoint or a relation, such as reading one more tag, is a Java change. The knowledge of what a span means
lives in two places, written in two styles.

This SWIP moves the whole native analysis into TAL rules hosted by the trace-analyzer module, and removes the
AnalysisListenerFactory SPI: every analysis of a native span is a rule file that the configuration activates. It is a
port and nothing else:

• No output change. For every segment, the rules send the same sources, with the same field values, in the same
  order, as the listeners do. They make the same stateful calls and side-effect calls in the same order, apart from the
  one exception below, and write the same log lines, debug included, under the script's logger rather than the
  listener's class name. OAL, storage, query, alarm and the UI are untouched, and so is every metric name.
• No logic change. Each rule reproduces one listener branch by branch, including the cases where the Java fails a
  segment on malformed input. A difference found while porting is a defect of the port, not a fix to make on the way.
  A behaviour worth changing is changed afterwards, in its own change, with its own changelog line.
• One exception, the evaluation sample. The listener samples a GenAI span for evaluation during the span loop. The
  rule writes the sample as a record in build {}, where a native rule makes all of its outputs, so the sample is
  taken after the segment's last span. A segment that analyzes completely makes the same sample calls, with the same
  contexts, in the same order, only later. A segment that fails no longer takes the samples of the GenAI spans before
  the failure, see
  Failure behaviour the port keeps.

TAL gains what the port needs:

• Native input: a sw-native input type, provided through the existing TALSourceTypeProvider extension point,
  whose input is the span, with its segment as span.segment.
• config: a read-only root: config.from(module, setting) reads any loaded module's setting, and a few named
  entries read the values a module parses or changes at runtime.
• Outputs in build {}: a native rule makes its outputs in one deferred block. source {} hands an existing OAL
  source to the OAL engine, documented as the adoption path for the existing source system; record writes what an
  output builder builds; metrics {} declares a new metric, as it does in every TAL rule.
• Lifecycle constructs, plain assignment, typed values and log: what a per-span rule needs to reproduce a
  listener exactly, and to read like one.
• Bridge functions: functions that call the caches, naming rules and lookups the listeners call. Other modules
  contribute their own functions through a new TALFunctionProvider extension point. Logic
  stays in the rules: a function reads a cache, a settings file or a lookup the rules cannot reach, or parses what TAL
  has no syntax for.

The complete ported rule files are in the first comment of this discussion. A
test module runs the old listeners and the new rules side by side and compares their complete, ordered behaviour, the
exception's sample calls as a list of their own, following the precedent of the MAL and LAL move from Groovy to the v2
compilers.

Scope

Today, in agent-analyzer and the modules plugged into it	After this SWIP, rule files in tal/
RPCAnalysisListener	native-rpc.yaml: rules native-rpc-entry, native-rpc-exit, native-rpc-logic-endpoint
EndpointDepFromCrossThreadAnalysisListener	native-endpoint-cross-thread.yaml
NetworkAddressAliasMappingListener	native-network-alias.yaml
SegmentAnalysisListener	native-segment.yaml, plus the parseSegment values the runner derives on span.segment before the spans
VirtualServiceAnalysisListener: VirtualCacheProcessor, VirtualDatabaseProcessor, VirtualMQProcessor	native-virtual-service.yaml: rules native-virtual-cache, native-virtual-database, native-virtual-mq
VirtualServiceAnalysisListener: VirtualGenAIProcessor (only when gen-ai-analyzer is loaded)	native-virtual-genai.yaml, loaded only when gen-ai-analyzer is loaded
GenAIEvaluationAnalysisListener, plugged in by ai-evaluation through the AnalysisListenerFactory SPI	native-ai-evaluation.yaml, loaded only when ai-evaluation is loaded
TraceAnalyzer, AnalysisListener, AnalysisListenerFactory, the listener interfaces, SegmentParserServiceImpl and the listener manager	removed; trace-analyzer runs the rules
ISegmentParserService, the segment entry point, in agent-analyzer	moved to trace-analyzer, whose runner implements it

Six listeners and four processors move. The RPC listener becomes three rules because its build() sends all entry
sources, then all exit sources, then all logic-endpoint sources; three rules, flushed in file order, send in that
order. See How trace-analyzer runs the native rules.

Architecture Graph

Today
  agent --segment--> trace receiver --> ISegmentParserService (agent-analyzer)
                                          '--> TraceAnalyzer, one per segment
                                                 |- RPCAnalysisListener
                                                 |- EndpointDepFromCrossThreadAnalysisListener
                                                 |- NetworkAddressAliasMappingListener   --> SourceReceiver --> OAL, records
                                                 |- SegmentAnalysisListener
                                                 |- VirtualServiceAnalysisListener (cache, database, MQ, GenAI)
                                                 '- AnalysisListenerFactory SPI (ai-evaluation)

After
  agent --segment--> trace receiver --> ISegmentParserService (trace-analyzer)
                                          '--> native rule runner, one run per segment
                                                 parseSegment's work, onto span.segment, then
                                                 rules from nativeTalFiles, in list order:
                                                 native-rpc, native-endpoint-cross-thread,
                                                 native-network-alias, native-segment,      --> SourceReceiver --> OAL, records
                                                 native-virtual-service, native-virtual-genai,
                                                 native-ai-evaluation                       --> record GenAIEvaluationSample
                                                                                                --> IAIEvaluationService queue
                                                 rules call: bridge functions, through TALFunctionProvider:
                                                   trace-analyzer's formatServiceName, findNetworkAlias,
                                                   tagList, clusterNameOrAddress, matchGenAIModel, ...
                                                 rules read: config.from(module, setting), the settings a
                                                   module serves through its TALConfigReader service,
                                                   and named entries for parsed or dynamic values

  test/script-cases/trace-analysis-listener-v1     the removed Java, test scope only
  test/script-cases/trace-analysis-v1-v2-checker   runs both on the same segments, compares ordered event logs

trace-analyzer already depends on agent-analyzer (oap-server/analyzer/trace-analyzer/pom.xml), so the runner
lives there with no new Maven dependency in either direction.

ISegmentParserService is internal: its only callers are the native trace receiver's gRPC and REST handlers, the Kafka
fetcher's segment handler and the data-generator tool, all in this repository. It moves from agent-analyzer to
trace-analyzer. TraceAnalyzerModule declares it, and its provider registers the runner as its implementation. The
callers add trace-analyzer to their required modules and look the service up there, see
Activation and startup.

Proposed Changes

The sw-native input type

SwNativeTALSourceTypeProvider implements the existing TALSourceTypeProvider extension point under the alias
sw-native, beside otlp. Its input is the span. A rule reads it from two places, and each value is in exactly one:

• span: what the agent sent for the span: operationName, peer, spanId, parentSpanId, startTime, endTime,
  componentId, isError, skipAnalysis, spanLayer (the constant's name), isEntry, isExit, isLocal, tags
  (in wire order, each with key and value), refs (reference views with refType, isCrossProcess, traceId,
  parentTraceSegmentId, parentSpanId, parentService, parentServiceInstance, parentEndpoint,
  networkAddressUsedAtPeer), ref (the first reference or null), refsOrNull and segment;
• span.segment: what the agent sent for the segment (service, serviceInstance, traceId, traceSegmentId,
  isSizeLimited, spans, bytes), and what the runner derives before the first rule runs, doing
  SegmentAnalysisListener.parseSegment's work with the same collaborators and log lines: startTime, endTime,
  latency, isError, tags (the searchable tags) and sampled.

span.segment is one read-only object per segment run, shared by every span view of it, as TraceAnalyzer hands every
listener the same SegmentObject. A rule cannot change a span or the segment: the Segment record stores the received
segment unchanged. Editing the stored trace is left to a proposal of its own.

What the listeners derive for one span alone, the rules write out where they need it:

• the service's layer, identifyServiceLayer: FAAS for a span whose spanLayer is FAAS, GENERAL for any other;
• the span's latency, (span.endTime - span.startTime) as Integer, Java's (int) cast;
• once per segment, as SegmentAnalysisListener.build() runs once: a segmentDef guard on the first span.

The otlp input's meta root and the statements that overwrite it are not part of the sw-native input. The runner
runs the sw-native rules on every span of every segment; a rule's layer key is GENERAL.

The config root

config is a read-only root. config.from("<module>", "<setting>") reads a setting that a loaded module chooses to
serve to TAL:

if (!(config.from("agent-analyzer", "traceAnalysis") as Boolean)) { abort {} }
def rate = config.from("ai-evaluation", "sampleRate") as Long

A module that lets TAL read its settings exposes a standard service, as it exposes any other: it lists
TALConfigReader (server-core) in its ModuleDefine.services(), and its provider registers an implementation over
the configuration it holds. The provider is never exposed, only the settings its reader serves.

public interface TALConfigReader extends Service {
    boolean has(String setting);   // asked when a rule compiles; an unknown setting fails the boot
    Object get(String setting);    // asked each time the rule runs; converted with `as`
}

The compiler takes the reader through moduleManager.find(<module>).provider().getService(TALConfigReader.class).
agent-analyzer serves traceAnalysis and maxSlowSQLLength; ai-evaluation serves sampleRate and active, the
decision its prepare() made from its judge configuration, and not the judge settings themselves. A rule file that
reads another module's settings lists it under requiredModules. agent-analyzer is loaded whenever a native file
compiles, so a native file reads its settings without listing it.

A few named entries stay for values a rule looks up with an argument, values agent-analyzer parses or changes at
runtime, and the runner's own settings: noUpstreamRealAddressAgents, slowDBAccessThreshold(type),
slowCacheReadThreshold(type), slowCacheWriteThreshold(type), searchableTracesTags,
segmentStatusAnalysisStrategy, forceSampleErrorSegment and traceSamplingPolicy, each read through the object that
owns it today. gateways.yml and server-clusters.yml stay behind the functions isAddressConfiguredAsGateway and
clusterNameOrAddress.

How trace-analyzer runs the native rules

The runner reproduces TraceAnalyzer's lifecycle callback for callback. TAL stays per span. A listener's per-segment
behaviour maps to two blocks and one kind of variable, each the counterpart of one Java mechanism:

TraceAnalyzer today	The runner
A segment without spans returns at once	Same
SegmentListener.parseSegment, before the spans	The derived span.segment values above, computed before the first rule's first run
FirstAnalysisListener.parseFirst, on every span whose id is 0, before any other callback on that span, whatever its type	first {} of every rule, in file order, on every span whose spanId is 0, before any rule's regular run on that span, whatever its type
The entry, exit or local callbacks of every listener, in listener order, on every span in wire order	Each rule's filter runs, in file order, on every span in wire order. A rule tests span.isEntry, span.isExit or span.isLocal; a span of an unknown type matches none, as no callback runs for it today, and the runner logs the same error for it under its own logger
A listener's fields, alive for one segment	segmentDef, a variable alive for one segment
build() of every listener, in listener order, after the last span	build {} blocks. A block is queued when a span's run reaches it. After the last span the runner runs the queued blocks rule by rule in file order, and within a rule in the order they were queued
NetworkAddressAliasMappingListener's alias row, sent during the span loop	writeNetworkAlias, sent at its statement
GenAIEvaluationAnalysisListener's sample, taken during the span loop	a record in build {}, taken in the build phase: the one place the port moves a send, see Failure behaviour the port keeps
A listener's log lines	log statements, under the script's logger org.apache.skywalking.trace.analysis.script.<script name>, and the functions that call the logging Java. The runner calls a compiled rule directly, not through TalDSL.evaluate, so the [TAL] rule=... debug line the OTLP path writes before each rule is not written for a native rule

Ordering follows from this. A Java builder is one queued build {} block: the parse callback fills it, build()
formats its names and sends its sources. A rule that formats names inside build {} therefore makes the
NamingControl calls, whose results depend on EndpointNameGrouping's per-service state and whose truncations log, at
the same point and in the same number as the Java does. A rule that sends from build {} sends in the Java order.

Failures follow the Java too. The runner catches nothing, neither around a rule nor around a record's init and
complete; the OTLP path catches around both. A rule that throws stops the segment at the same point a throwing
listener stops it today, and the exception reaches ISegmentParserService.send and the receiver unchanged:

• the gRPC handler counts and logs it;
• the REST handler counts it and rethrows;
• the Kafka fetcher's handler catches only InvalidProtocolBufferException, so the exception leaves the handler.

A throw before the build phase means no queued block runs, and so no evaluation sample of the segment is taken.
Everything already done stays done: alias rows sent, log lines written, and changes to the endpoint grouping's state. A
throw inside a build {} block, including one raised while the OAL engine dispatches a source, stops the remaining
blocks, and what was already sent stays sent.

The sw-native input has no retry dedup (dedupWindow does not apply), no network-alias pairing, and no persist {}
or tag: the segment's storage is the native-segment rule's Segment record. prepareNetworkAlias,
buildNetworkAlias, persist {}, tag, the outputType key and sink {} are compile errors in a sw-native rule,
and so are metrics {}, source {} and record outside build {}: a native rule makes its outputs there.

Activation and startup

trace-analyzer:
  default:
    talFiles: ${SW_TRACE_TAL_FILES:otlp-general}
    # TAL rule files under `tal/` for spans that SkyWalking agents send, without extension, comma separated, in the
    # order their rules run. The default reproduces the agent path.
    nativeTalFiles: ${SW_TRACE_NATIVE_TAL_FILES:native-rpc,native-endpoint-cross-thread,native-network-alias,native-segment,native-virtual-service,native-virtual-genai,native-ai-evaluation}

• Rule files sit in tal/ beside otlp-general.yaml; every native rule declares inputType: sw-native. A missing
  file or two native rules of one name fail the boot. A separate list keeps native analysis running for an operator who
  set SW_TRACE_TAL_FILES to their own OTLP files.
• requiredModules at a file's top level compiles and runs its rules only when the modules are loaded:
  native-virtual-genai lists gen-ai-analyzer, native-ai-evaluation lists ai-evaluation, the checks the listener
  factories make today.
• traceAnalysis keeps its meaning: the RPC, cross-thread and alias rules abort when it is false.
• Startup. trace-analyzer declares among its required modules, when each is loaded, agent-analyzer, the module
  each discovered TALFunctionProvider names, and every module a rule file lists under requiredModules, so every
  module a rule reaches has started first. It registers the runner as ISegmentParserService in prepare(), and binds
  the function providers and compiles the rules in start(). The native trace receiver, the Kafka fetcher and the
  data-generator tool require trace-analyzer beside agent-analyzer, so no segment reaches the runner before its
  rules are compiled.
• Without trace-analyzer. With SW_TRACE_ANALYZER=- while the native trace receiver (on by default) or the Kafka
  fetcher is loaded, the module manager fails the boot, naming the module; with neither loaded, removing it still works
  for OTLP alone. The data-generator tool's own application.yml gains a trace-analyzer section.
• Without agent-analyzer. The native trace receiver still requires it, as today. trace-analyzer alone, for
  OTLP, stays allowed and compiles no native file.

The order of nativeTalFiles is part of the contract. No rule reads another rule's output directly: the one
cross-file write, an alias row, reaches native-rpc-exit only through the alias cache, where it is not visible until
the cache reloads, and a reload can land while a segment is still being analyzed, as it can today. Order still decides
what reaches an order-sensitive merge first (EndpointRelationServerSideMetrics.combine and EndpointTraffic.combine keep
the first object), what is already sent when a segment fails, and the order of the stateful naming calls through
EndpointNameGrouping. Inside native-rpc the rules are entry, exit and logic endpoint, the listener's build grouping;
inside native-virtual-service, cache, database and MQ, with GenAI after, the processor order.

New TAL section: Source

This is the text proposed for the Source section of the native input's page, tal-sw-native.md, see
Documentation layout.

Source (existing system adoption)

source {} hands one source of the existing OAL system to the OAL engine: the same object the SkyWalking agent path
builds for core.oal and the other .oal files. It exists so that an analysis written in Java can move to TAL
without changing what OAL computes from it, and every metric named in .oal keeps its name, its function and its
number. A new metric is declared with metrics {} instead.

source <Name> {
  <field> <expr> [as <Type>]
  ...
}

• <Name> is the name OAL's from() gives the source, the name of its @ScopeDeclaration: Service,
  ServiceInstance, Endpoint, ServiceRelation, ServiceInstanceRelation, EndpointRelation, ServiceMeta,
  EndpointMeta, DatabaseAccess, DatabaseSlowStatement, ServiceDatabaseSlowStatement, CacheAccess,
  VirtualCacheSlowAccess, MQAccess, MQEndpointAccess, Segment, TagAutocomplete, GenAIProviderAccess,
  GenAIModelAccess, and any source a module declares. Only a concrete source class with a no-argument constructor
  qualifies; an unknown name is a compile error that lists the known ones.

• Each field statement calls the source's setter of that name, checked when the rule compiles, as an
  Output Type field is. Keywords such as name, status, layer
  and endpoint work as field names here, and set the source's field rather than meta. The value is converted to
  the setter's parameter type:

  • a number, a boolean, a string or a byte array as it is;
  • a Layer or an enum, such as RequestType or DetectPoint, as it is when the value already has that type
    (see Typed Layer and enum values), and from its name when it is a string. A
    literal name is checked when the rule compiles; any other string is converted when the rule runs, and a name that
    matches nothing fails the rule;
  • a List or Map field takes the list or map object as it is, such as
    tagList(), without copying it.

  A field the block does not set keeps the source's constructor default: null for most objects, an empty list for
  Segment's tags, a SideCar for Service's.

• timeBucket is required, minuteTimeBucket(...) for a metrics source and recordTimeBucket(...) for a record
  such as DatabaseSlowStatement, as the agent path sets them. A source {} without it is a compile error.

• Nothing is formatted implicitly. A name is stored as the expression gives it. A rule ported from the agent path
  calls formatServiceName, formatInstanceName and formatEndpointName where the Java called NamingControl.

• The block holds field statements only. A value that depends on a condition is computed into a def first.

• Where it is written. source {} belongs to the sw-native input and is written inside build {}: it
  is built and sent at its statement, so a later failure in the same block leaves it sent. In an otlp rule it is a
  compile error that fails the boot; an OTLP rule names its entities through metrics {}, whose entities
  MeterEntity formats.

A record field takes the same conversions, objects, null, Layer, enums, lists and maps included, and
so does an Output Type field.

source ServiceMeta {
  name db
  layer "VIRTUAL_DATABASE"
  timeBucket minuteTimeBucket(span.startTime)
}

Other syntax additions

Assignment

<name> = <expr> [as <Type>], on every input. def declares a variable and = assigns it; today a later def of the
same name assigns, which reads like a second declaration, and it keeps working. A variable that the branches of an if
assign is declared before them, with a typed null for a String, a Layer or an enum (def owner = null as String)
and a literal for a number or a boolean. The native rule files use = for every assignment and def only to declare.

foreach

foreach (<name> in <list>) {
  <the statements of the block foreach sits in>
}

<list> is a java.util.List, such as span.refs, span.tags, span.segment.tags or a list a function returns.
The loop variable is typed to the list's element type when the rule compiles, from the getter's or function's generic
return type, so ref.parentService and spanTag.key are compile-checked chains. A def inside the body declares a
variable of one iteration. It is declared again at each iteration, and it is not visible after the loop: that is how
the Java builds a fresh builder per reference. A variable declared before the loop and assigned inside it with = keeps
its value across iterations. foreach is allowed wherever if is. It replaces indexing
such as span.refs[0], which the compiler does not support on a List.

build

build {
  <def, assignment, if, foreach, abort, log, metrics, source, record>
}

sw-native only, inside extractor {}. The block is queued when the run reaches it and runs once after the segment's
last span, see How trace-analyzer runs the native rules.

• What it sees. The run's input, its span and its function context, as they were for that run, not the last
  span's. The def values the run had when it reached the block, including the loop variables of the foreach it sits
  in, captured by value. The segmentDef values as they are when the block runs.
• Its own variables. A captured value is the block's own copy. caller = formatServiceName(caller) reads the copy
  and assigns it, as the Java formats a builder's field in place; the run and the other blocks do not see the change.
  A def inside the block declares a variable of that block.
• Outputs. A sw-native rule makes its outputs here and nowhere else, each at its statement:
  • metrics {}, a new metric, as in every TAL rule. Its timestamp defaults to span.startTime, there being no
    meta, and a failure inside it skips that block, as it does everywhere: the first failure of a metric logs a
    warning, later ones log at debug. No shipped native rule
    declares one;
  • source {}, one source of the existing OAL system;
  • record, one record of an output builder.
• Abort. An abort {} later in the same run withdraws the blocks that run queued. An abort {} inside a block
  stops that block only.

first

first {
  <the statements of a filter>
}

sw-native only, at filter level. It runs on every span whose spanId is 0, before any rule's regular run on that
span, whatever the span's type, as FirstAnalysisListener.parseFirst does. It is a run of its own: it has its own
def values, an abort {} in it ends that run only, and it shares only segmentDef values with the rule's regular
run.

segmentDef

segmentDef <name> = <expr>

sw-native only. A variable of the rule that lives for one segment, as a listener's field does. It starts as null for
each segment. The segmentDef statement declares it and assigns it each time it runs; <name> = <expr> assigns it
too, anywhere in the rule, on any span of the segment; and the name reads it anywhere in the rule, before its first
assignment included. Its storage belongs to one run of the runner, never to the compiled
rule, so concurrent segments do not share it, and it is cleared when the segment ends, also when it fails.

writeNetworkAlias

writeNetworkAlias <address>

sw-native only. It sends, at its statement, a NetworkAddressAliasSetup with:

• address the given address;
• representService formatServiceName(span.segment.service);
• representServiceNormal true;
• representServiceInstance formatInstanceName(span.segment.serviceInstance);
• the minute of the span's start as its time bucket.

This is the row NetworkAddressAliasMappingListener sends from an entry span's cross-process reference. It sends the
source and lets the dispatcher and the cache reload do the rest, without touching the cache. It does not use the OTLP
path's alias registration, which refuses an empty address or service and writes no instance.

The SkyWalking agent propagates the address the caller dialed in the sw8 header, so the callee alone has every fact
of the row, and nothing waits for a caller's half as the OTLP pairing does. The source has no @ScopeDeclaration
name, so source {} cannot name it.

log

log <trace | debug | info | warn | error> "<message>" [, <expr>, ...]

Writes the message through SLF4J under the logger org.apache.skywalking.trace.analysis.script.<script name>, the
script being the rule file without its extension, such as org.apache.skywalking.trace.analysis.script.native-rpc; the
name is not configurable. Each {} is replaced by the next argument as SLF4J formats it. The arguments are evaluated
only when the level is enabled, as the Java's isDebugEnabled() guards do. It is allowed wherever a statement is. A
function that logs, such as parseStatusCode, logs under the calling rule's script logger; a function that delegates to
Java, such as the naming functions or isAddressConfiguredAsGateway, logs as that Java logs today, under its own class
name.

Typed Layer and enum values

as converts a string to a Layer or to an enum a source or a record takes, such as RequestType,
DetectPoint, MQOperation, VirtualCacheOperation, TagType and GenAITraceRefType:

def calleeLayer = "VIRTUAL_CACHE" as Layer
def type = "HTTP" as RequestType

Layer is a registry, not a Java enum: as Layer calls Layer.valueOf(name), which fails on an unknown name, and a
literal is checked when the rule compiles against the layers registered by then, those a rule file's
layerDefinitions register included. An enum literal is checked against the enum's constants. A def given a typed
value has that type, so a misspelt name is a compile error where it is written, not a failure when its branch runs.

A Layer or an enum value compared with a string literal compares by name, and the literal is checked when the rule
compiles: calleeLayer != "FAAS", operation == "Write".

Where each construct applies

The native constructs apply to sw-native rules only. The constructs the native input cannot support are refused in
sw-native rules, and every other input type, otlp and any input a TALSourceTypeProvider adds, keeps them as it
does today, so the restriction breaks no existing rule. The rules compile at boot, so a construct used under the wrong
inputType fails the boot, naming the file, the rule and the construct, as a rule on an unregistered layer does today.
A file that requiredModules leaves out is not compiled, and so not checked.

Applies to	Constructs
sw-native only	build {}, source {}, record, first {}, segmentDef, writeNetworkAlias, the functions for the sw-native input, and the native views: span as the native span, span.segment and the reference and tag views
every input except sw-native	the meta root and the statements that overwrite it (service, instance, endpoint, layer), the outputType key and sink {}, tag, persist {}, prepareNetworkAlias, buildNetworkAlias and pairing, and the functions that read data a native span does not carry or the pairing's rows: resource, scopeAttr, event, eventAttr, callerScopedAddress, resolveNetworkAlias, isGateway, findServerClusterName
every input	assignment (<name> = <expr>), foreach, log, the config root, typed Layer and enum values, metrics {} (inside build {} in a sw-native rule), requiredModules, def, if, abort {}, attr and hasAttr, urlHost, toJson, toJsonArray, component, and the functions for every input

New keywords

source, record, foreach, in, build, first, segmentDef, writeNetworkAlias, log and config. They still
work as field names and label keys, but no longer as def names. The shipped otlp-general rule uses none of them as a
def name. The existing keywords stay what they are: the native rules name their locals so that none is a keyword, such
as spanTag rather than tag and logicJson rather than value.

record

record <Name> {
  <field> <expr> [as <Type>]
  ...
}

sw-native only, inside build {}. It writes one record of an output builder, a TALOutputBuilder, at its
statement, as source {} sends one OAL source. <Name> is the builder's short name or class name; each field statement
calls the builder's setter, with the conversions of source {}. The runner creates a new builder, calls bindInput,
sets the fields, then calls init and complete, with nothing caught; on sw-native the metadata is null and the
input is the native span view. TALOutputBuilder gains default boolean supportsInputType(String inputType), true
for every input but sw-native, so existing builders such as SlowSQL keep working where they do and are refused in a
native rule. Any module can add a builder native rules write, by returning true for sw-native and listing it in
META-INF/services. A native rule has no outputType key and no sink {}: it decides with if and abort {}.

New TAL functions

Every function below calls the Java the listeners call, or reproduces a private method of it where the table says so,
so a value, a failure and a log line are the same as theirs.

Functions for every input

Function	Returns	The Java it calls
buildServiceId(name, normal)	the service id	IDManager.ServiceID.buildId
buildInstanceId(serviceId, name)	the instance id	IDManager.ServiceInstanceID.buildId
buildEndpointId(serviceId, name)	the endpoint id	IDManager.EndpointID.buildId
minuteTimeBucket(millis), recordTimeBucket(millis)	the minute and the record (second) time bucket, in the JVM's time zone	TimeBucket.getMinuteTimeBucket, TimeBucket.getRecordTimeBucket
round(double)	the long nearest the value	Math.round(double)
cut(text, length)	the text cut to at most length characters	StringUtil.cut
isBlank(text)	whether the text is null, or empty once trimmed	StringUtil.isBlank
emptyList(), emptyMap()	a new, empty, mutable ArrayList or HashMap	the containers an EndpointSourceBuilder starts with
parseJsonObject(text)	the text parsed as a JSON object, null for blank text; malformed or non-object text fails the rule. The getters that follow are Gson's, so their coercions and failures are Gson's. toJson() answers an empty object instead	Gson.fromJson(text, JsonObject.class)
parseLongOrZero(text), parseIntOrZero(text)	0 for null or empty text; the number; or 0 after the warning Failed to parse value to long: {} (to int for the second) under the script's logger, for text that does not parse	reproduces GenAIMeterAnalyzer's private parseSafeLong and parseSafeInt: Long.parseLong, Integer.parseInt
murmur3Hash(text)	the first 64 bits of the murmur3-128 hash of the text's UTF-8 bytes, as a long	Guava's Hashing.murmur3_128().hashString(text, UTF_8).asLong(), as DefaultAIEvaluationSamplingPolicy hashes a trace id
floorMod(x, y)	the floor modulus of two long values, never negative for a positive y	Math.floorMod(long, long)

Functions for the sw-native input

Contributed by trace-analyzer, which reaches agent-analyzer's configuration files and the core's caches. They read
the run's span where they need it. The naming functions are here and not for every input: an otlp rule names its
entities through metrics {}, whose entities MeterEntity already formats, so only the native rules, which must make
the listeners' exact NamingControl calls, need to call them.

Function	Returns	The Java it calls
formatServiceName(name)	the service name after the length limit; null stays null; a truncation logs	NamingControl.formatServiceName
formatInstanceName(name)	the instance name after the length limit; a truncation logs	NamingControl.formatInstanceName
formatEndpointName(service, endpoint)	the endpoint unchanged when either argument is empty; otherwise the name after the length limit and the endpoint grouping, which keeps per-service state	NamingControl.formatEndpointName
matchGenAIModel(model)	the match for the model's longest prefix in gen-ai-config.yml: provider, unknown when none or when the model is empty, and pricing, the matched model's inputEstimatedCostPerM and outputEstimatedCostPerM (double), null when no model matches. The first call loads gen-ai-config.yml	GenAIModelMatcher.getInstance().match(model) (library-util), which GenAIContextResolver.resolve asks for the provider and GenAIProviderPrefixMatcher.match for the prices
tagList()	the span's tags as key.trim() + ":" + value.trim() strings, in wire order, in a new list	EndpointSourceBuilder.setTag, the tags list
tagMapKeepingLast()	the span's raw tags as a new map, a repeated key keeping its last value	EndpointSourceBuilder.setTag, the originalTags map
tagMapKeepingFirst()	the same, keeping the first value	the toMap with (left, right) -> left in GenAIEvaluationAnalysisListener and GenAIMeterAnalyzer
tagMapRejectingRepeats()	the same; a repeated key fails the rule with the exception Collectors.toMap throws	the Collectors.toMap without a merge function in VirtualCacheProcessor
parseStatusCode(text)	the text as an Integer, or null after logging, under the script's logger, the warning span {} has illegal status code {} with the wire span and the text	Integer.parseInt in setPublicAttrs, its catch (NumberFormatException) and log.warn
clusterNameOrAddress(address)	the cluster server-clusters.yml lists the address under, or the address; nothing caught. The existing findServerClusterName answers empty instead, and catches its exceptions	ServerClusterConfig.clusterNameOrAddress
findNetworkAlias(address)	the alias row stored for the address, or null. Its service and instance are decoded from its ids, the service first and both always, so a malformed id fails even when the instance is not used. instance is _blank when the row names none	NetworkAddressAliasCache.get, and the IDManager decoding in RPCAnalysisListener.parseExit
isAddressConfiguredAsGateway(address)	whether gateways.yml lists the address, matched exactly, with that class's debug line; the existing isGateway also retries a bare host at port 80, which the agent path does not	UninstrumentedGatewaysConfig.isAddressConfiguredAsGateway

Contributed by other modules, and the extension points

The GenAI extraction and the evaluation's sampling policy are logic, so the rules write them out: the tokens, the time
to first token and the cost from the model's prices in native-virtual-genai, and in native-ai-evaluation the
decision from ai-evaluation's active and sampleRate, read through config.from, and a murmur3 hash of the trace
id. The OTLP and Zipkin paths keep calling the same Java, which reads the same settings.

• ai-evaluation contributes GenAIEvaluationSample
  (org.apache.skywalking.oap.server.ai.evaluation.listener.GenAIEvaluationSampleBuilder), a TALOutputBuilder that
  native-ai-evaluation writes with record. Its complete builds the AIEvaluationContext the listener builds and
  calls IAIEvaluationService.sample, which queues it for the LLM judge: a hand-off, not a stored record.
• TALFunctionProvider (server-core, found through ServiceLoader, bound in trace-analyzer's start()) lets a
  module contribute functions without depending on trace-analyzer. The shipped rules call trace-analyzer's own.
• TALConfigReader (server-core) is the module service behind config.from, see The config root.
• TALOutputBuilder.supportsInputType says which inputs may write a builder, every input but sw-native by default.

Failure behaviour the port keeps

These Java paths fail a segment on input the agent should not send. The port reproduces each, through the functions
above and the runner's no-catch rule:

Java	Input that fails	The rule
VirtualMQProcessor	a non-blank transmission.latency tag that Long.parseLong rejects, even when a later one is valid	spanTag.value as Long in an assignment, which calls Long.parseLong, after the same blank check
RPCAnalysisListener logic endpoint	an x-le tag that is blank (Gson answers null, then has throws), not a JSON object, malformed, or holds a value its getter cannot convert	parseJsonObject and the same Gson getters, in the same order; a tag without the keys is skipped, as today
VirtualCacheProcessor	a cache span with a repeated tag key, checked before the cache.type guard	tagMapRejectingRepeats(), called first
OAL dispatch during build()	a source the OAL engine fails on, such as a blank transmission.latency tag converted by core.oal's (str->long) cast	the same source sent at the same point of the same build {} order

Making any of them tolerant is a behaviour change, and so it is out of scope.

One failure path changes, by choice: when the evaluation sample is taken. GenAIEvaluationAnalysisListener samples a
GenAI span during the span loop. native-ai-evaluation writes the sample as a record in build {}, where
a sw-native rule makes its outputs, so it is taken in the build phase, after every earlier file's blocks. The
sampling decision and the tag reads stay in the span run, in their place.

• A segment that analyzes completely makes the same sample calls, with the same contexts, in the same order. The
  sampling decision is a hash of the trace id, and the evaluation queue shares no state with the OAL engine, so the
  analysis itself does not change. What the evaluation service does with a call depends on its consumer threads, today
  as after the port: whether a task of the same id is still pending, whether the bounded queue has room, and in which
  minute its counters move, which the OAP's self-observability rules read as per-minute increases. A call made later
  can meet a different state, so a duplicate task, such as one from a re-sent segment, or a call that meets a full
  queue, can be accepted or skipped differently, and a counter can move in the next minute.
• A segment that fails, on any span or in any earlier block, takes no sample. Today the GenAI spans before the
  failure have theirs taken.
• sample itself checks the context and counts a full queue rather than throwing, so taking the samples last
  stops nothing that would have run.

Verifying that the rules behave as the listeners did

When MAL, LAL and the hierarchy rules moved from Groovy to the ANTLR4 and Javassist compilers (#13723), the Groovy
runtime stayed as test-only modules and v1-v2 checkers compared both engines on the same scripts, until #13782 removed
them after the new engines shipped. This SWIP repeats that pattern with a stricter comparison: exact, ordered, and
covering stateful calls, side effects and log lines.

• Modules. test/script-cases/trace-analysis-listener-v1 holds the removed Java, moved with git mv and never
  packaged: TraceAnalyzer, SegmentParserServiceImpl, the listener manager and interfaces, the six listeners, the four
  processors, their builders and the listener assembly frozen at the base commit. test/script-cases/trace-analysis-v1-v2-checker
  holds NativeTraceComparisonTest, the expected-data generator and NativeTraceBenchmark.
• What is compared. Each side runs every fixture with its own instance of every stateful collaborator, and
  produces one ordered event log per segment: every source sent, rendered field by field by reflection; every
  NamingControl call; every sampler, searchable-tags, status-analyzer and alias-cache call; every log line, with the
  listener's class logger and the script's logger counted as one; and every exception. The two logs must be equal,
  event by event, and so must the EndpointNameGrouping states after each segment. The evaluation sample calls, which
  the port moves to the build phase, are compared as their own list. A second level runs the fixtures through a real
  DispatcherManager with the shipped OAL scripts and compares what reaches the stream processors and what the merges
  and TopN buffers keep.
• Input. Segments recorded from the e2e cases that run SkyWalking agents (Java, gateway, nginx-lua, Go, Python,
  Node.js, PHP, Kafka, virtual MQ, server clusters, meter, virtual GenAI, mixed GenAI), fixtures built from them for every
  branch and malformed input, several configuration profiles, and cold and warm alias-cache passes.
• Guards. Designated positive fixtures per rule, counts that fail on zero, mutation tests on the comparison, and
  JaCoCo branch coverage over the moved classes.
• Expected data, performance, boot and e2e. .data.yaml holds the v1 event logs, written by a separate generator.
  A JMH benchmark compares both sides, and a regression blocks the switch. Boot tests cover the module combinations, and
  every listed e2e case runs unchanged.
• Retirement. The listener module and the checker stay at least until a release has shipped the TAL analysis.

Steps

1. TAL: plain assignment, source {}, record, foreach, build {}, first {}, segmentDef,
   writeNetworkAlias, log, the config root and TALConfigReader, typed values, file-level requiredModules,
   TALFunctionProvider, supportsInputType and the new functions, with the compiler work they need: getters evaluated
   where they stand, setters for objects, enums, Layer, lists and maps, deferred build {} code, per-segment
   segmentDef storage, input-aware compile checks, ordered rule loading and no catch on the native path.
2. The port: the sw-native input and its views, the runner, ISegmentParserService moved to trace-analyzer,
   nativeTalFiles, the shipped rule files, the functions and GenAIEvaluationSample; the listeners and the
   AnalysisListenerFactory SPI move to the test module, and the checker lands. It lands only with the comparison
   passing on every fixture and profile.
3. Documentation, per Documentation layout below.
4. After a release: retirement.

Imported Dependencies libs and their licenses.

None in the distribution. The checker module uses the JaCoCo Maven plugin (EPL-2.0) at build time only; it is not
packaged.

Compatibility

• Outputs. The checker asserts identical ordered event logs on every fixture and profile: the same sources with
  the same fields in the same order, the same NamingControl calls, sampler calls, alias rows, log lines and failures.
  The evaluation sample calls are the same, with the same contexts and in the same order, for a segment that
  completes, though made after its last span, so what the evaluation service accepts can differ with its consumers'
  timing; a segment that fails makes none, see Failure behaviour the port keeps.
• Configuration.
  • One new setting, trace-analyzer.default.nativeTalFiles.
  • talFiles, traceAnalysis, the analyzer's thresholds and every other existing setting keep their meaning.
  • SW_TRACE_ANALYZER=- now stops the boot while the native trace receiver, which is on by default, or the Kafka
    fetcher is loaded, since trace-analyzer hosts the native analysis. An operator who removed the module to turn
    off OTLP analysis sets talFiles to empty instead, which the module already supports. With neither caller
    loaded, removing the module still works.
• Extension point. The AnalysisListenerFactory SPI is removed. A third-party listener moves to a rule file, plus
  a TALFunctionProvider for what TAL cannot read or call, and a TALOutputBuilder that supports sw-native for
  what it hands over that is not an OAL source. TALOutputBuilder gains supportsInputType as a default method, so
  existing implementations compile and behave unchanged. The module system does not change: a module that lets TAL
  read its settings declares the TALConfigReader service, and its provider registers it, as agent-analyzer and
  ai-evaluation do.
• TAL. Additive, except the ten new keywords, which a user's TAL rule can no longer use as def names. Plain
  assignment is additive too: a later def of a declared name keeps assigning it.
• Getters. On every input, a getter now runs where it stands rather than hoisted to the start of the rule. A chain
  that a hoisted getter dereferenced before its guard threw there, which on the OTLP path skips the rule for that span,
  with a warning the first time and at debug after; it now reaches the guard. The otlp input's chains read protobuf
  getters, which never return null and have no side effects, so the shipped otlp-general rules behave the same. A
  custom input whose getters can return null may now pass a guard where it used to fail before it.
• Log lines. The same levels, messages and arguments. The native analysis's lines move from the listeners' class
  loggers to org.apache.skywalking.trace.analysis.script.<script name> and the runner's own logger, so a logging
  configuration that sets a level for a listener class sets it for the script's logger instead. GenAIMeterAnalyzer's
  lines for native spans, its missing-model debug line and its parse warnings, move to the native-virtual-genai
  script's logger; it keeps logging under its own name for OTLP and Zipkin spans.
• Storage, protocols, query, UI. No change.
• Editing the rules. An operator who edits a native rule file, or reorders nativeTalFiles, changes the native
  analysis, as with any rule file. The shipped files and the default order are what the checker compares.
• Distributions. A distribution that precompiles TAL rules needs a compile-and-boot check of the native files, the
  new constructs, the function providers and the TALConfigReader services config.from reads before it ships them.

General usage docs

Documentation layout

The tracing documentation is organized by where a trace comes from. TAL comes first, as the language the native and
OpenTelemetry analyses are written in. Then each source has the same parts: its setup, its bundled analysis, its TAL
input reference, and best practices where there are some. Every existing page keeps its path; sections move only out of
today's tal.md, to the OTLP input page, and the Zipkin part of trace-sampling.md, to the Zipkin page.

Setup > Tracing
├── Overview                          setup/backend/tracing-overview.md            new
├── Trace Analysis Language           concepts-and-designs/tal.md                  the language, reorganized
├── SkyWalking Native Traces
│   ├── Setup                         setup/backend/native-trace.md                new
│   ├── Bundled Analysis              setup/backend/native-trace-analysis.md       new
│   │   ├── Trace Sampling            setup/backend/trace-sampling.md              existing, its Zipkin part moves out
│   │   ├── Slow Database Statement   setup/backend/slow-db-statement.md           existing
│   │   ├── Slow Cache Command        setup/backend/slow-cache-command.md          existing
│   │   └── Message Queue Performance setup/backend/mq.md                          existing
│   ├── TAL for Native Segments       concepts-and-designs/tal-sw-native.md        new
│   └── Best Practices                                                             added with the first practice
├── OpenTelemetry Traces
│   ├── Setup                         setup/backend/otlp-trace.md                  existing
│   ├── Bundled Analysis              setup/backend/trace-analyzer.md              existing
│   ├── TAL for OpenTelemetry Spans   concepts-and-designs/tal-otlp.md             new
│   └── Best Practices                setup/backend/otlp-trace-practices/*.md      existing, unchanged
├── Zipkin Traces                     setup/backend/zipkin-trace.md                existing, gains the Zipkin sampling part
└── Shared by the sources
    ├── Uninstrumented Gateways       setup/backend/uninstrumented-gateways.md     existing
    └── Server Clusters               setup/backend/server-clusters.md             existing

The comparison table on the language page:

	otlp	sw-native
Sent by	OpenTelemetry SDKs and collectors	SkyWalking agents, as trace segments over gRPC, HTTP or Kafka
Rule files	talFiles (SW_TRACE_TAL_FILES)	nativeTalFiles (SW_TRACE_NATIVE_TAL_FILES)
Which rules see a span	the rules declared for the span's service.layer	every rule of every listed file whose requiredModules are loaded, in list order, on every span of the segment, in wire order
What a rule reads	span, the OTLP Span, and meta, what the receiver derived from it	span, the native span, and span.segment, its segment
Runs	one per span and rule	first {} on a span 0, the span's run, and build {} after the segment's last span; segmentDef lives for the segment
Outputs	metrics {}; a record through outputType and sink {}; persist {} and tag for the stored span	inside build {}: metrics {}, source {} for OAL sources, record for an output builder
The trace itself	stored by the receiver when no span listener vetoes it and no rule's persist {} votes to drop it, then past the receiver's rate limit and head sampling. An enforcer {} skips head sampling only; with otlpTraceStorage: zipkin it marks the span debug, which the Zipkin path keeps past both its rate limit and its sample rate. Neither overrides a veto	stored as the native-segment rule's Segment source when the segment is sampled; no rule changes it
Network alias	learned from caller and callee pairing, prepareNetworkAlias and buildNetworkAlias	written by the callee alone, writeNetworkAlias, from the address the sw8 header carries
Entity names	formatted by the metrics {} entity constructors	formatted by formatServiceName, formatInstanceName and formatEndpointName, where the listeners called the naming rules
Retries	off by default; with dedupWindow above 0, a span sent again to the same node within the window, while the node still holds its id, is analyzed once	no deduplication, as today
A rule that throws	caught: the rule is skipped for that span, with a warning the first time and at debug after, and the other rules still run	not caught: the segment fails, as a throwing listener fails it today

[wu-sheng]

The complete TAL rule files of this proposal, one per listener or processor they port. Together they are the native
trace analysis that the Java listeners in agent-analyzer do today. The default nativeTalFiles order is the order
below; the design is in the discussion's first post. The comments in each file name the Java method each part
reproduces. The Apache license headers are left out here.

native-rpc.yaml: RPC listener: entry, exit and logic endpoints

# RPCAnalysisListener, as TAL rules for spans that SkyWalking agents send (SWIP-17).
#
# RPCAnalysisListener.build() sends every entry builder's sources, then every exit builder's, then every logic
# endpoint's. The listener is therefore three rules, whose build {} blocks the runner flushes in this order. One queued
# build {} is one builder: the span run fills it as parseEntry, parseExit or parseLocal does, and the block formats its
# names as RPCTrafficSourceBuilder.prepare() does and sends its sources as build() does, in the same order.
#
# The rules log under org.apache.skywalking.trace.analysis.script.native-rpc, parseStatusCode's warning included.

rules:
  # parseEntry: one builder per reference, whatever the reference type. Without a reference, one builder from an MQ
  # consumer's peer, or from `User`.
  - name: native-rpc-entry
    layer: GENERAL
    inputType: sw-native
    dsl: |
      filter {
        if (!(config.from("agent-analyzer", "traceAnalysis") as Boolean) || !span.isEntry || span.skipAnalysis) {
          abort {}
        }
        extractor {
          // identifyServiceLayer: FAAS for a FaaS span, GENERAL for any other
          def calleeLayer = "GENERAL" as Layer
          if (span.spanLayer == "FAAS") {
            calleeLayer = "FAAS" as Layer
          }

          // refsOrNull is the references, or one null when the span has none
          foreach (ref in span.refsOrNull) {
            // One builder per reference: its caller fields start null, as a new RPCTrafficSourceBuilder's do
            def caller = null as String
            def callerInstance = null as String
            def callerEndpoint = null as String
            def callerLayer = null as Layer
            def owner = null as String
            def ownerLayer = null as Layer
            if (ref != null) {
              callerEndpoint = "User"
              if (ref.parentEndpoint) {
                callerEndpoint = ref.parentEndpoint
              }
              if (span.spanLayer == "MQ" || isAddressConfiguredAsGateway(ref.networkAddressUsedAtPeer)) {
                // The broker, or the uninstrumented gateway, is the caller; the parent service owns the endpoint
                caller = ref.networkAddressUsedAtPeer
                callerLayer = "VIRTUAL_GATEWAY" as Layer
                if (span.spanLayer == "MQ") {
                  caller = clusterNameOrAddress(ref.networkAddressUsedAtPeer)
                  callerLayer = "VIRTUAL_MQ" as Layer
                }
                callerInstance = caller
                owner = ref.parentService
                ownerLayer = "GENERAL" as Layer
              } else {
                caller = ref.parentService
                callerInstance = ref.parentServiceInstance
                callerLayer = "GENERAL" as Layer
              }
            } else if (span.spanLayer == "MQ" && !isBlank(span.peer)) {
              // A consumer without a producer reference: the broker its peer names, and no caller endpoint
              caller = clusterNameOrAddress(span.peer)
              callerInstance = caller
              callerLayer = "VIRTUAL_MQ" as Layer
            } else {
              caller = "User"
              callerInstance = "User"
              callerEndpoint = "User"
              callerLayer = "UNDEFINED" as Layer
            }

            // setPublicAttrs, once per builder, after its caller is set
            def bucket = minuteTimeBucket(span.startTime)
            def latency = (span.endTime - span.startTime) as Integer
            def httpCode = 0
            def rpcCode = null as String
            def tags = tagList()
            def originalTags = tagMapKeepingLast()
            foreach (spanTag in span.tags) {
              if (spanTag.key == "status_code" || spanTag.key == "http.status_code") {
                // Each one is parsed; one that does not parse is logged and skipped, so the last valid one wins
                def parsed = parseStatusCode(spanTag.value)
                if (parsed != null) {
                  httpCode = parsed
                }
              } else if (spanTag.key == "rpc.status_code") {
                rpcCode = spanTag.value
              }
            }
            def ok = true
            if (span.isError) {
              ok = false
            }
            def type = "RPC" as RequestType
            if (span.spanLayer == "Http") {
              type = "HTTP" as RequestType
            } else if (span.spanLayer == "Database") {
              type = "DATABASE" as RequestType
            } else if (span.spanLayer == "MQ") {
              type = "MQ" as RequestType
            } else if (span.spanLayer == "GenAI") {
              type = "GenAI" as RequestType
            }

            build {
              // RPCTrafficSourceBuilder#prepare: the same calls in the same order. A captured value is this block's
              // own copy: assigning it here changes it for this block only. owner is null outside the MQ and gateway
              // branch, and formatServiceName(null) is null, as in the Java.
              caller = formatServiceName(caller)
              owner = formatServiceName(owner)
              callerInstance = formatInstanceName(callerInstance)
              callerEndpoint = formatEndpointName(caller, callerEndpoint)
              def callee = formatServiceName(span.segment.service)
              def calleeInstance = formatInstanceName(span.segment.serviceInstance)
              def calleeEndpoint = formatEndpointName(callee, span.operationName)

              // build(): each source is sent at its statement
              source Service {
                name callee
                serviceInstanceName calleeInstance
                endpointName calleeEndpoint
                layer calleeLayer
                latency latency
                status ok
                httpResponseStatusCode httpCode
                rpcStatusCode rpcCode
                type type
                tags tags
                originalTags originalTags
                timeBucket bucket
              }
              source ServiceInstance {
                name calleeInstance
                serviceName callee
                serviceLayer calleeLayer
                endpointName calleeEndpoint
                latency latency
                status ok
                httpResponseStatusCode httpCode
                rpcStatusCode rpcCode
                type type
                tags tags
                originalTags originalTags
                timeBucket bucket
              }
              source ServiceRelation {
                sourceServiceName caller
                sourceServiceInstanceName callerInstance
                sourceLayer callerLayer
                destServiceName callee
                destServiceInstanceName calleeInstance
                destLayer calleeLayer
                endpoint calleeEndpoint
                componentId span.componentId
                latency latency
                status ok
                httpResponseStatusCode httpCode
                rpcStatusCode rpcCode
                type type
                detectPoint "SERVER"
                timeBucket bucket
              }
              if (callerInstance && calleeInstance) {
                source ServiceInstanceRelation {
                  sourceServiceName caller
                  sourceServiceInstanceName callerInstance
                  sourceServiceLayer callerLayer
                  destServiceName callee
                  destServiceInstanceName calleeInstance
                  destServiceLayer calleeLayer
                  endpoint calleeEndpoint
                  componentId span.componentId
                  latency latency
                  status ok
                  httpResponseStatusCode httpCode
                  rpcStatusCode rpcCode
                  type type
                  detectPoint "SERVER"
                  timeBucket bucket
                }
              }
              // A FaaS function is its own endpoint: neither Endpoint nor EndpointRelation
              if (calleeLayer != "FAAS") {
                source Endpoint {
                  name calleeEndpoint
                  serviceName callee
                  serviceLayer calleeLayer
                  serviceInstanceName calleeInstance
                  latency latency
                  status ok
                  httpResponseStatusCode httpCode
                  rpcStatusCode rpcCode
                  type type
                  tags tags
                  originalTags originalTags
                  timeBucket bucket
                }
                if (callerEndpoint && calleeEndpoint) {
                  // The endpoint's owner when one was set, even an empty one; the caller otherwise
                  def relationService = caller
                  def relationLayer = callerLayer
                  if (owner != null) {
                    relationService = owner
                    relationLayer = ownerLayer
                  }
                  source EndpointRelation {
                    endpoint callerEndpoint
                    serviceName relationService
                    serviceLayer relationLayer
                    serviceInstanceName callerInstance
                    childEndpoint calleeEndpoint
                    childServiceName callee
                    childServiceLayer calleeLayer
                    childServiceInstanceName calleeInstance
                    componentId span.componentId
                    rpcLatency latency
                    status ok
                    httpResponseStatusCode httpCode
                    rpcStatusCode rpcCode
                    type type
                    detectPoint "SERVER"
                    timeBucket bucket
                  }
                }
              }
            }
          }
          // parseEntry ends with the span's logic endpoints: rule native-rpc-logic-endpoint
        }
      }

  # parseExit: one builder per exit span that names a peer.
  - name: native-rpc-exit
    layer: GENERAL
    inputType: sw-native
    dsl: |
      filter {
        if (!(config.from("agent-analyzer", "traceAnalysis") as Boolean) || !span.isExit || span.skipAnalysis) {
          abort {}
        }
        // An empty peer is skipped; a peer of spaces is not
        if (!span.peer) {
          abort {}
        }
        extractor {
          // identifyServiceLayer, for this service as the caller: FAAS for a FaaS span, GENERAL for any other
          def callerLayer = "GENERAL" as Layer
          if (span.spanLayer == "FAAS") {
            callerLayer = "FAAS" as Layer
          }
          // The alias row of the peer as reported; both of its ids are decoded, so a malformed one fails here
          def alias = findNetworkAlias(span.peer)
          def callee = null as String
          def calleeInstance = null as String
          def calleeLayer = null as Layer
          if (alias == null) {
            // A database, a cache or a broker no alias maps is named as the cluster server-clusters.yml lists
            callee = span.peer
            if (span.spanLayer == "Database" || span.spanLayer == "Cache" || span.spanLayer == "MQ") {
              callee = clusterNameOrAddress(span.peer)
            }
            calleeInstance = callee
            // identifyRemoteServiceLayer, from the peer as reported
            calleeLayer = "UNDEFINED" as Layer
            if (span.spanLayer == "Database") {
              calleeLayer = "VIRTUAL_DATABASE" as Layer
            } else if (span.spanLayer == "RPCFramework") {
              calleeLayer = "GENERAL" as Layer
            } else if (span.spanLayer == "Http") {
              calleeLayer = "GENERAL" as Layer
              if (isAddressConfiguredAsGateway(span.peer)) {
                calleeLayer = "VIRTUAL_GATEWAY" as Layer
              }
            } else if (span.spanLayer == "MQ") {
              calleeLayer = "VIRTUAL_MQ" as Layer
            } else if (span.spanLayer == "Cache") {
              calleeLayer = "VIRTUAL_CACHE" as Layer
            } else if (span.spanLayer == "FAAS") {
              calleeLayer = "FAAS" as Layer
            } else if (span.spanLayer == "GenAI") {
              calleeLayer = "VIRTUAL_GENAI" as Layer
            }
            // Unknown and UNRECOGNIZED stay UNDEFINED; the Java switch covers every SpanLayer constant
          } else {
            // The alias names a real service. Its instance only when the agent reports the upstream's real address
            // and the row names an instance; calleeInstance stays null otherwise, so no instance relation follows.
            callee = alias.service
            if (!config.noUpstreamRealAddressAgents.contains(span.componentId) && alias.instance != "_blank") {
              calleeInstance = alias.instance
            }
            calleeLayer = "GENERAL" as Layer
          }

          // setPublicAttrs. The exit builder's tag containers are built by the Java but read by none of its sources,
          // and building them neither logs nor keeps state, so this rule does not build them.
          def bucket = minuteTimeBucket(span.startTime)
          def latency = (span.endTime - span.startTime) as Integer
          def httpCode = 0
          def rpcCode = null as String
          foreach (spanTag in span.tags) {
            if (spanTag.key == "status_code" || spanTag.key == "http.status_code") {
              def parsed = parseStatusCode(spanTag.value)
              if (parsed != null) {
                httpCode = parsed
              }
            } else if (spanTag.key == "rpc.status_code") {
              rpcCode = spanTag.value
            }
          }
          def ok = true
          if (span.isError) {
            ok = false
          }
          def type = "RPC" as RequestType
          if (span.spanLayer == "Http") {
            type = "HTTP" as RequestType
          } else if (span.spanLayer == "Database") {
            type = "DATABASE" as RequestType
          } else if (span.spanLayer == "MQ") {
            type = "MQ" as RequestType
          } else if (span.spanLayer == "GenAI") {
            type = "GenAI" as RequestType
          }

          build {
            // RPCTrafficSourceBuilder#prepare, all seven calls. An exit builder sets no endpoint owner, caller endpoint
            // or callee endpoint, so those three calls get null and return it without reaching the endpoint grouping,
            // and the relations carry no endpoint.
            def caller = formatServiceName(span.segment.service)
            def owner = formatServiceName(null as String)
            def callerInstance = formatInstanceName(span.segment.serviceInstance)
            def callerEndpoint = formatEndpointName(caller, null as String)
            callee = formatServiceName(callee)
            calleeInstance = formatInstanceName(calleeInstance)
            def calleeEndpoint = formatEndpointName(callee, null as String)

            source ServiceRelation {
              sourceServiceName caller
              sourceServiceInstanceName callerInstance
              sourceLayer callerLayer
              destServiceName callee
              destServiceInstanceName calleeInstance
              destLayer calleeLayer
              componentId span.componentId
              latency latency
              status ok
              httpResponseStatusCode httpCode
              rpcStatusCode rpcCode
              type type
              detectPoint "CLIENT"
              timeBucket bucket
            }
            if (callerInstance && calleeInstance) {
              source ServiceInstanceRelation {
                sourceServiceName caller
                sourceServiceInstanceName callerInstance
                sourceServiceLayer callerLayer
                destServiceName callee
                destServiceInstanceName calleeInstance
                destServiceLayer calleeLayer
                componentId span.componentId
                latency latency
                status ok
                httpResponseStatusCode httpCode
                rpcStatusCode rpcCode
                type type
                detectPoint "CLIENT"
                timeBucket bucket
              }
            }
          }
        }
      }

  # parseLogicEndpoints, from parseEntry (after the entry's references) and from parseLocal: one Endpoint per accepted
  # `x-le` tag, in tag order.
  - name: native-rpc-logic-endpoint
    layer: GENERAL
    inputType: sw-native
    dsl: |
      filter {
        if (!(config.from("agent-analyzer", "traceAnalysis") as Boolean)) {
          abort {}
        }
        // parseEntry returns early for a span marked skipAnalysis; parseLocal reads its logic endpoints whatever the
        // mark
        if (!span.isLocal && !(span.isEntry && !span.skipAnalysis)) {
          abort {}
        }
        extractor {
          foreach (spanTag in span.tags) {
            if (spanTag.key == "x-le") {
              // Gson.fromJson(value, JsonObject.class): blank text gives null, and the has() below then throws;
              // text that is not a JSON object throws here. The getters are Gson's own, in the Java's order.
              def logicJson = parseJsonObject(spanTag.value)
              def accepted = false
              // Read only under if (accepted), where both branches set them
              def logicName = null as String
              def logicLatency = 0
              def logicOk = false
              if (span.isLocal && logicJson.has("logic-span") && logicJson.get("logic-span").getAsBoolean()) {
                accepted = true
                logicName = span.operationName
                logicLatency = (span.endTime - span.startTime) as Integer
                logicOk = true
                if (span.isError) {
                  logicOk = false
                }
              } else if (logicJson.has("name") && logicJson.has("latency") && logicJson.has("status")) {
                accepted = true
                logicName = logicJson.get("name").getAsString()
                logicLatency = logicJson.get("latency").getAsInt()
                logicOk = logicJson.get("status").getAsBoolean()
              }
              // A tag without the keys is skipped
              if (accepted) {
                // The builder starts with empty tag containers and is filled during the span loop, as the Java does
                def tags = emptyList()
                def originalTags = emptyMap()
                def bucket = minuteTimeBucket(span.startTime)
                build {
                  // EndpointSourceBuilder#prepare
                  def callee = formatServiceName(span.segment.service)
                  def calleeInstance = formatInstanceName(span.segment.serviceInstance)
                  def calleeEndpoint = formatEndpointName(callee, logicName)
                  // A logic endpoint carries none of the span's public attributes: HTTP 0, no RPC code, no tags
                  source Endpoint {
                    name calleeEndpoint
                    serviceName callee
                    serviceLayer "GENERAL"
                    serviceInstanceName calleeInstance
                    latency logicLatency
                    status logicOk
                    httpResponseStatusCode 0
                    type "LOGIC"
                    tags tags
                    originalTags originalTags
                    timeBucket bucket
                  }
                }
              }
            }
          }
        }
      }

native-endpoint-cross-thread.yaml: Endpoint dependencies across threads

# EndpointDepFromCrossThreadAnalysisListener, as a TAL rule for spans that SkyWalking agents send (SWIP-17).
#
# An exit or local span with references, such as the first span of a task another thread started, draws the endpoint
# dependency from each reference's endpoint to the span. One queued build {} is one EndpointDependencyBuilder.

rules:
  - name: native-endpoint-cross-thread
    layer: GENERAL
    inputType: sw-native
    dsl: |
      filter {
        if (!(config.from("agent-analyzer", "traceAnalysis") as Boolean) || !(span.isExit || span.isLocal)
            || span.skipAnalysis) {
          abort {}
        }
        extractor {
          // identifyServiceLayer: FAAS for a FaaS span, GENERAL for any other
          def calleeLayer = "GENERAL" as Layer
          if (span.spanLayer == "FAAS") {
            calleeLayer = "FAAS" as Layer
          }
          // Without a reference there is no builder; every reference type counts
          foreach (ref in span.refs) {
            // One builder per reference: the owner fields stay null outside the MQ and gateway branch
            def callerEndpoint = "User"
            def caller = null as String
            def callerInstance = null as String
            def callerLayer = null as Layer
            def owner = null as String
            def ownerLayer = null as Layer
            if (ref.parentEndpoint) {
              callerEndpoint = ref.parentEndpoint
            }
            if (span.spanLayer == "MQ" || isAddressConfiguredAsGateway(ref.networkAddressUsedAtPeer)) {
              caller = ref.networkAddressUsedAtPeer
              callerLayer = "VIRTUAL_GATEWAY" as Layer
              if (span.spanLayer == "MQ") {
                caller = clusterNameOrAddress(ref.networkAddressUsedAtPeer)
                callerLayer = "VIRTUAL_MQ" as Layer
              }
              callerInstance = caller
              owner = ref.parentService
              ownerLayer = "GENERAL" as Layer
            } else {
              caller = ref.parentService
              callerInstance = ref.parentServiceInstance
              callerLayer = "GENERAL" as Layer
            }

            // This listener's own setPublicAttrs: HTTP code NONE (0), every tag, the status. It sets no type and no
            // RPC status code, so both stay null even when the span carries status tags.
            def bucket = minuteTimeBucket(span.startTime)
            def latency = (span.endTime - span.startTime) as Integer
            def tags = tagList()
            def originalTags = tagMapKeepingLast()
            def ok = true
            if (span.isError) {
              ok = false
            }

            build {
              // RPCTrafficSourceBuilder#prepare, through EndpointDependencyBuilder#prepare
              caller = formatServiceName(caller)
              owner = formatServiceName(owner)
              callerInstance = formatInstanceName(callerInstance)
              callerEndpoint = formatEndpointName(caller, callerEndpoint)
              def callee = formatServiceName(span.segment.service)
              def calleeInstance = formatInstanceName(span.segment.serviceInstance)
              def calleeEndpoint = formatEndpointName(callee, span.operationName)

              // EndpointDependencyBuilder#toSourceEndpoint: the caller service, never the endpoint's owner
              source EndpointMeta {
                serviceName caller
                serviceNormal callerLayer.isNormal()
                endpoint callerEndpoint
                timeBucket bucket
              }
              // No FaaS exclusion here, unlike RPCAnalysisListener
              source Endpoint {
                name calleeEndpoint
                serviceName callee
                serviceLayer calleeLayer
                serviceInstanceName calleeInstance
                latency latency
                status ok
                httpResponseStatusCode 0
                tags tags
                originalTags originalTags
                timeBucket bucket
              }
              if (callerEndpoint && calleeEndpoint) {
                def relationService = caller
                def relationLayer = callerLayer
                if (owner != null) {
                  relationService = owner
                  relationLayer = ownerLayer
                }
                source EndpointRelation {
                  endpoint callerEndpoint
                  serviceName relationService
                  serviceLayer relationLayer
                  serviceInstanceName callerInstance
                  childEndpoint calleeEndpoint
                  childServiceName callee
                  childServiceLayer calleeLayer
                  childServiceInstanceName calleeInstance
                  componentId span.componentId
                  rpcLatency latency
                  status ok
                  httpResponseStatusCode 0
                  detectPoint "SERVER"
                  timeBucket bucket
                }
              }
            }
          }
        }
      }

native-network-alias.yaml: Network address alias from the sw8 reference

# NetworkAddressAliasMappingListener, as a TAL rule for spans that SkyWalking agents send (SWIP-17).
#
# The sw8 header carries the address the caller dialed, so an entry span alone names its own service as that
# address. The row is sent during the span loop, as the listener sends it, not in the build phase.

rules:
  - name: native-network-alias
    layer: GENERAL
    inputType: sw-native
    dsl: |
      filter {
        if (!(config.from("agent-analyzer", "traceAnalysis") as Boolean) || !span.isEntry || span.skipAnalysis) {
          abort {}
        }
        // Logged for every entry span the listener does not skip, before its MQ check
        log debug "service instance mapping listener parse reference"
        if (span.spanLayer == "MQ") {
          abort {}
        }
        extractor {
          foreach (ref in span.refs) {
            if (ref.isCrossProcess) {
              // Formatted as a service name before the gateway check, as the listener does
              def address = formatServiceName(ref.networkAddressUsedAtPeer)
              if (!isAddressConfiguredAsGateway(address)) {
                // NetworkAddressAliasSetup: this address, formatServiceName(span.segment.service),
                // formatInstanceName(span.segment.serviceInstance), representServiceNormal true, the minute of the
                // span's start; sent at this statement
                writeNetworkAlias address
              }
            }
          }
        }
      }

native-segment.yaml: The Segment record and tag autocomplete

# SegmentAnalysisListener, as a TAL rule for spans that SkyWalking agents send (SWIP-17): the trace itself, one Segment
# record per sampled segment, and a TagAutocomplete row per stored tag.
#
# parseSegment runs before every span callback. The runner does that work before any rule runs, in its order, and
# exposes the results on the segment: span.segment.startTime and span.segment.latency (the fold over every span),
# span.segment.isError (the error fold under segmentStatusAnalysisStrategy), span.segment.tags (the tags
# config.searchableTracesTags picks, which the record stores), and span.segment.sampled (the sampling policy's
# decision, or an error segment under forceSampleErrorSegment). The segmentDef variables are the listener's fields.

rules:
  - name: native-segment
    layer: GENERAL
    inputType: sw-native
    dsl: |
      filter {
        // parseFirst: on every span whose id is 0, before any rule's regular run on that span, whatever its type
        first {
          if (span.segment.sampled) {
            // The service name and id are computed once per segment, by the first callback that needs them
            if (serviceId == null) {
              segmentDef serviceName = formatServiceName(span.segment.service)
              segmentDef serviceId = buildServiceId(serviceName, true)
            }
            // The record's fields, set in parseFirst's order; a later span 0 sets them again
            segmentDef recordBucket = recordTimeBucket(span.segment.startTime)
            segmentDef segmentId = span.segment.traceSegmentId
            segmentDef serviceInstanceId = buildInstanceId(serviceId, formatInstanceName(span.segment.serviceInstance))
            segmentDef errorValue = 0
            if (span.segment.isError) {
              errorValue = 1
            }
            segmentDef dataBinary = span.segment.bytes
            segmentDef endpointId = buildEndpointId(serviceId, formatEndpointName(serviceName, span.operationName))
          }
        }

        // parseEntry: every entry span names the segment's endpoint, sampled or not, and skipAnalysis or not. A span
        // 0 later in wire order renames it in first {}.
        if (span.isEntry) {
          extractor {
            // The segment variables first {} declares, assigned again
            if (serviceId == null) {
              serviceName = formatServiceName(span.segment.service)
              serviceId = buildServiceId(serviceName, true)
            }
            endpointId = buildEndpointId(serviceId, formatEndpointName(serviceName, span.operationName))
          }
        }

        // build(): queued once per segment, by the run on its first span in wire order, whatever the span's type. The
        // block reads the segmentDef values as they are after the last span.
        if (buildQueued == null) {
          segmentDef buildQueued = true
          extractor {
            build {
              if (!span.segment.sampled) {
                log debug "segment ignored, trace id: {}", span.segment.traceId
              } else {
                // segmentId is null when no span 0 ran parseFirst, and the Java logs it as null too
                log debug "segment listener build, segment id: {}", segmentId
                def recordEndpointId = endpointId
                if (recordEndpointId == null) {
                  recordEndpointId = ""
                }
                // The record's start time: the fold's start when parseFirst ran, the field's default 0 otherwise
                def recordStart = 0L
                if (segmentId != null) {
                  recordStart = span.segment.startTime
                  source Segment {
                    segmentId segmentId
                    traceId span.segment.traceId
                    serviceId serviceId
                    serviceInstanceId serviceInstanceId
                    endpointId recordEndpointId
                    latency span.segment.latency
                    startTime span.segment.startTime
                    timeBucket recordBucket
                    isError errorValue
                    dataBinary dataBinary
                    tags span.segment.tags
                  }
                } else {
                  // No span 0: parseFirst never ran, so only what parseSegment and parseEntry set; the other fields
                  // keep the record's defaults, the time bucket 0 among them
                  source Segment {
                    traceId span.segment.traceId
                    endpointId recordEndpointId
                    timeBucket 0
                    tags span.segment.tags
                  }
                }
                // addAutocompleteTags: the minute of the record's start time, minuteTimeBucket(0) without span 0
                foreach (recordTag in span.segment.tags) {
                  source TagAutocomplete {
                    tagKey recordTag.key
                    tagValue recordTag.value
                    tagType "TRACE"
                    timeBucket minuteTimeBucket(recordStart)
                  }
                }
              }
            }
          }
        }
      }

native-virtual-service.yaml: Virtual cache, database and MQ

# VirtualServiceAnalysisListener's cache, database and MQ processors, as TAL rules for spans that SkyWalking agents
# send (SWIP-17), in the listener's processor order. A processor prepares its sources during the span loop and the
# listener sends them in build(), so each rule prepares every value in the span run, in the processor's order, and
# only sends from build {}.
#
# The listener is called for entry, exit and local spans, with no skipAnalysis or traceAnalysis check.

rules:
  # VirtualCacheProcessor
  - name: native-virtual-cache
    layer: GENERAL
    inputType: sw-native
    dsl: |
      filter {
        if (!(span.isEntry || span.isExit || span.isLocal) || span.spanLayer != "Cache") {
          abort {}
        }
        extractor {
          // The tag map is built first, so a repeated key fails the segment even when cache.type is missing
          def cacheTags = tagMapRejectingRepeats()
          def cacheType = cacheTags.get("cache.type") as String
          if (isBlank(cacheType)) {
            abort {}
          }
          // The thresholds are looked up by the type in the JVM's default-locale lower case
          def thresholdType = cacheType.toLowerCase()
          def peer = span.peer
          if (isBlank(peer)) {
            // A local cache, named after its type as the span spelled it, with no server-clusters lookup
            peer = cacheType + "-local"
          } else {
            peer = clusterNameOrAddress(span.peer)
          }
          def bucket = minuteTimeBucket(span.startTime)
          def cacheService = formatServiceName(peer)
          def latency = (span.endTime - span.startTime) as Integer
          def ok = true
          if (span.isError) {
            ok = false
          }
          def opTag = cacheTags.get("cache.op") as String
          def operation = "Others" as VirtualCacheOperation
          if (opTag == "write") {
            operation = "Write" as VirtualCacheOperation
          } else if (opTag == "read") {
            operation = "Read" as VirtualCacheOperation
          }
          def slow = false
          if (operation == "Write" && latency > config.slowCacheWriteThreshold(thresholdType)) {
            slow = true
          } else if (operation == "Read" && latency > config.slowCacheReadThreshold(thresholdType)) {
            slow = true
          }
          if (slow) {
            // The slow record's values, prepared in the processor's order; a missing command or key stays null
            def slowServiceId = buildServiceId(cacheService, false)
            def slowRecordId = span.segment.traceSegmentId + "-" + span.spanId
            def command = cacheTags.get("cache.cmd") as String
            def cacheKey = cacheTags.get("cache.key") as String
            def recordBucket = recordTimeBucket(span.startTime)
          }

          build {
            source ServiceMeta {
              name cacheService
              layer "VIRTUAL_CACHE"
              timeBucket bucket
            }
            if (slow) {
              source VirtualCacheSlowAccess {
                cacheServiceId slowServiceId
                latency latency
                id slowRecordId
                status ok
                traceId span.segment.traceId
                command command
                key cacheKey
                timeBucket recordBucket
                timestamp span.startTime
                operation operation
              }
            }
            source CacheAccess {
              cacheTypeId span.componentId
              latency latency
              name cacheService
              status ok
              timeBucket bucket
              operation operation
            }
          }
        }
      }

  # VirtualDatabaseProcessor
  - name: native-virtual-database
    layer: GENERAL
    inputType: sw-native
    dsl: |
      filter {
        if (!(span.isEntry || span.isExit || span.isLocal) || span.spanLayer != "Database") {
          abort {}
        }
        extractor {
          def bucket = minuteTimeBucket(span.startTime)
          // A database reached at an address server-clusters.yml lists is named as its cluster; no blank-peer guard
          def db = formatServiceName(clusterNameOrAddress(span.peer))
          def latency = (span.endTime - span.startTime) as Integer
          def ok = true
          if (span.isError) {
            ok = false
          }
          // readStatementIfSlow: every statement is cut and the last one wins; every db.type looks up its threshold,
          // and any one exceeded makes the call slow
          def statement = null as String
          def slow = false
          foreach (spanTag in span.tags) {
            if (spanTag.key == "db.statement") {
              statement = cut(spanTag.value, config.from("agent-analyzer", "maxSlowSQLLength") as Integer)
            } else if (spanTag.key == "db.type" && latency > config.slowDBAccessThreshold(spanTag.value)) {
              slow = true
            }
          }
          // The two slow records' values, prepared in the processor's order
          def writeSlow = false
          if (slow && !isBlank(statement)) {
            writeSlow = true
            def dbRecordId = span.segment.traceSegmentId + "-" + span.spanId
            def dbServiceId = buildServiceId(db, false)
            def dbRecordBucket = recordTimeBucket(span.startTime)
            def callerRecordId = span.segment.traceSegmentId + "-" + span.spanId
            // The calling service's id from the segment's service as the agent sent it, not formatted
            def callerServiceId = buildServiceId(span.segment.service, true)
            def callerRecordBucket = recordTimeBucket(span.startTime)
          }

          build {
            source ServiceMeta {
              name db
              layer "VIRTUAL_DATABASE"
              timeBucket bucket
            }
            source DatabaseAccess {
              databaseTypeId span.componentId
              latency latency
              name db
              status ok
              timeBucket bucket
            }
            if (writeSlow) {
              source DatabaseSlowStatement {
                id dbRecordId
                traceId span.segment.traceId
                databaseServiceId dbServiceId
                statement statement
                latency latency
                timeBucket dbRecordBucket
                timestamp span.startTime
              }
              source ServiceDatabaseSlowStatement {
                id callerRecordId
                traceId span.segment.traceId
                serviceId callerServiceId
                statement statement
                latency latency
                timeBucket callerRecordBucket
                timestamp span.startTime
              }
            }
          }
        }
      }

  # VirtualMQProcessor
  - name: native-virtual-mq
    layer: GENERAL
    inputType: sw-native
    dsl: |
      filter {
        if (!(span.isEntry || span.isExit) || span.spanLayer != "MQ") {
          abort {}
        }
        extractor {
          def operation = null as MQOperation
          def peer = null as String
          if (span.isEntry) {
            // A consumer: the first reference's address when it is not blank, the span's peer otherwise; a later
            // reference is never read
            operation = "Consume" as MQOperation
            peer = span.peer
            if (span.ref != null && !isBlank(span.ref.networkAddressUsedAtPeer)) {
              peer = span.ref.networkAddressUsedAtPeer
            }
          } else {
            operation = "Produce" as MQOperation
            peer = span.peer
          }
          // A blank broker returns before any tag is read
          if (isBlank(peer)) {
            abort {}
          }
          // collectTags: the last topic and queue win; every transmission.latency is parsed in turn, blank as 0, and
          // one Long.parseLong rejects fails the segment, even when a later one is valid
          def topic = null as String
          def queue = null as String
          def transmissionLatency = 0L
          foreach (spanTag in span.tags) {
            if (spanTag.key == "mq.topic") {
              topic = spanTag.value
            } else if (spanTag.key == "mq.queue") {
              queue = spanTag.value
            } else if (spanTag.key == "transmission.latency") {
              transmissionLatency = 0L
              if (!isBlank(spanTag.value)) {
                transmissionLatency = spanTag.value as Long
              }
            }
          }
          def broker = formatServiceName(clusterNameOrAddress(peer))
          def bucket = minuteTimeBucket(span.startTime)
          def ok = true
          if (span.isError) {
            ok = false
          }
          // buildEndpointName: the non-blank of topic and queue, joined by "/", raw
          def mqEndpoint = ""
          if (!isBlank(topic) && !isBlank(queue)) {
            mqEndpoint = topic + "/" + queue
          } else if (!isBlank(topic)) {
            mqEndpoint = topic
          } else if (!isBlank(queue)) {
            mqEndpoint = queue
          }
          if (mqEndpoint) {
            // Named during the span loop, as the processor names it
            def mqEndpointName = formatEndpointName(broker, mqEndpoint)
          }

          build {
            source ServiceMeta {
              name broker
              layer "VIRTUAL_MQ"
              timeBucket bucket
            }
            source MQAccess {
              typeId span.componentId
              transmissionLatency transmissionLatency
              name broker
              status ok
              timeBucket bucket
              operation operation
            }
            if (mqEndpoint) {
              source EndpointMeta {
                serviceName broker
                serviceNormal false
                endpoint mqEndpointName
                timeBucket bucket
              }
              source MQEndpointAccess {
                typeId span.componentId
                transmissionLatency transmissionLatency
                status ok
                timeBucket bucket
                operation operation
                serviceName broker
                endpoint mqEndpointName
              }
            }
          }
        }
      }

native-virtual-genai.yaml: Virtual GenAI

# VirtualServiceAnalysisListener's GenAI processor, as a TAL rule for spans that SkyWalking agents send (SWIP-17).
# Loaded only with the gen-ai-analyzer module, as the listener's factory adds the processor only then.
#
# The processor extracts the GenAI facts (GenAIMeterAnalyzer.extractMetricsFromSWSpan, written out below) and prepares
# the sources (GenAIMeterAnalyzer.transferToSources) during the span loop, and the listener sends them in build(), after
# the cache, database and MQ sources. The outputs stay the four OAL sources that virtual-gen-ai.oal reads.

requiredModules:
  - gen-ai-analyzer

rules:
  - name: native-virtual-genai
    layer: GENERAL
    inputType: sw-native
    dsl: |
      filter {
        if (!(span.isEntry || span.isExit || span.isLocal) || span.spanLayer != "GenAI") {
          abort {}
        }
        extractor {
          // extractMetricsFromSWSpan. The raw tags, the first value of a repeated key winning
          def genAITags = tagMapKeepingFirst()
          // GenAIContextResolver.resolve: the declared provider, else the legacy gen_ai.system, else the provider
          // gen-ai-config.yml names for the model; the model lookup runs first, whichever branch is taken
          def modelName = genAITags.get("gen_ai.response.model") as String
          def declaredProvider = genAITags.get("gen_ai.provider.name") as String
          def legacySystem = genAITags.get("gen_ai.system") as String
          def providerName = matchGenAIModel(modelName).provider
          if (!isBlank(declaredProvider)) {
            providerName = declaredProvider
          } else if (!isBlank(legacySystem)) {
            providerName = legacySystem
          }
          if (isBlank(modelName)) {
            log debug "Model name is missing in span [{}], skipping GenAI analysis", span.operationName
            abort {}
          }
          // GenAIProviderPrefixMatcher.match: the same lookup, for the model's prices; null when no model is priced
          def pricing = matchGenAIModel(modelName).pricing
          // parseSafeLong: 0 for no value, and 0 with a warning for one that does not parse
          def inputTokens = parseLongOrZero(genAITags.get("gen_ai.usage.input_tokens") as String)
          def outputTokens = parseLongOrZero(genAITags.get("gen_ai.usage.output_tokens") as String)
          // calculateTotalCost
          def totalCost = 0.0
          if (pricing != null) {
            if (pricing.inputEstimatedCostPerM > 0) {
              totalCost = totalCost + inputTokens * pricing.inputEstimatedCostPerM
            }
            if (pricing.outputEstimatedCostPerM > 0) {
              totalCost = totalCost + outputTokens * pricing.outputEstimatedCostPerM
            }
          }
          // The Java also builds the provider's service id here, which nothing reads
          def timeToFirstToken = parseIntOrZero(genAITags.get("gen_ai.server.time_to_first_token") as String)
          def latency = span.endTime - span.startTime
          def ok = true
          if (span.isError) {
            ok = false
          }
          def bucket = minuteTimeBucket(span.startTime)

          // transferToSources, during the span loop: the same six naming calls and two roundings, in its order
          def metaService = formatServiceName(providerName)
          def instanceName = formatInstanceName(modelName)
          def instanceService = formatServiceName(providerName)
          def providerService = formatServiceName(providerName)
          def providerCost = round(totalCost)
          def modelService = formatServiceName(providerName)
          def modelInstanceName = formatInstanceName(modelName)
          def modelCost = round(totalCost)

          build {
            source ServiceMeta {
              name metaService
              layer "VIRTUAL_GENAI"
              timeBucket bucket
            }
            // Only these four fields; the others keep the source's defaults
            source ServiceInstance {
              timeBucket bucket
              name instanceName
              serviceLayer "VIRTUAL_GENAI"
              serviceName instanceService
            }
            source GenAIProviderAccess {
              name providerService
              inputTokens inputTokens
              outputTokens outputTokens
              totalEstimatedCost providerCost
              latency latency
              status ok
              timeBucket bucket
            }
            source GenAIModelAccess {
              serviceName modelService
              modelName modelInstanceName
              inputTokens inputTokens
              outputTokens outputTokens
              totalEstimatedCost modelCost
              timeToFirstToken timeToFirstToken
              latency latency
              status ok
              timeBucket bucket
            }
          }
        }
      }

native-ai-evaluation.yaml: GenAI evaluation sampling

# GenAIEvaluationAnalysisListener, as a TAL rule for spans that SkyWalking agents send (SWIP-17). Loaded only with the
# ai-evaluation module, which the listener's factory required.
#
# The sampling decision and the tag reads run on the span, where the listener makes them. The sample itself is a record
# of the GenAIEvaluationSample builder, written in build {} as a native rule makes all of its outputs, so it is taken
# after the segment's last span, once the earlier files' blocks have run. A segment that analyzes completely makes the
# same sample calls, with the same contexts, in the same order as the listener; a segment that fails makes none
# (SWIP-17, "Failure behaviour the port keeps"). The builder's complete() builds the AIEvaluationContext from the fields
# below and calls IAIEvaluationService.sample, with nothing caught; it hands the context to the evaluation queue and
# stores nothing.

requiredModules:
  - ai-evaluation

rules:
  - name: native-ai-evaluation
    layer: GENERAL
    inputType: sw-native
    dsl: |
      filter {
        if (!(span.isEntry || span.isExit || span.isLocal) || span.skipAnalysis || span.spanLayer != "GenAI") {
          abort {}
        }
        // IAIEvaluationService.shouldSample, decided before the tags are read, so an unsampled trace reads none.
        // ai-evaluation is inactive, and samples nothing, when its judge configuration is incomplete
        if (!(config.from("ai-evaluation", "active") as Boolean)) {
          abort {}
        }
        // DefaultAIEvaluationSamplingPolicy: a non-empty trace id whose murmur3-128 hash, taken as a long, falls below
        // sampleRate in a space of 1,000,000
        if (span.segment.traceId == ""
            || floorMod(murmur3Hash(span.segment.traceId), 1000000L)
                >= config.from("ai-evaluation", "sampleRate") as Long) {
          abort {}
        }
        extractor {
          // The raw tags, the first value of a repeated key winning
          def genAITags = tagMapKeepingFirst()
          if (isBlank(genAITags.get("gen_ai.response.model") as String)) {
            abort {}
          }
          // GenAIContextResolver.resolve: the declared provider, else the legacy gen_ai.system, else the provider
          // gen-ai-config.yml names for the model; the model lookup runs first, whichever branch is taken
          def responseModel = genAITags.get("gen_ai.response.model") as String
          def declaredProvider = genAITags.get("gen_ai.provider.name") as String
          def legacySystem = genAITags.get("gen_ai.system") as String
          def providerName = matchGenAIModel(responseModel).provider
          if (!isBlank(declaredProvider)) {
            providerName = declaredProvider
          } else if (!isBlank(legacySystem)) {
            providerName = legacySystem
          }
          build {
            record GenAIEvaluationSample {
              traceId span.segment.traceId
              traceRefType "SKYWALKING_NATIVE"
              segmentId span.segment.traceSegmentId
              // The span's id, not its position in the segment
              spanIndex span.spanId
              serviceName span.segment.service
              serviceInstanceName span.segment.serviceInstance
              operationName span.operationName
              providerName providerName
              modelName responseModel
              startTimeMillis span.startTime
              endTimeMillis span.endTime
              error span.isError
              tags genAITags
            }
          }
        }
      }