diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md index ae1830496ddc..19c78382c8ed 100644 --- a/docs/en/changes/changes.md +++ b/docs/en/changes/changes.md @@ -12,7 +12,7 @@ * Add BanyanDB trace tail sampling metrics to the BanyanDB self-observability layer, in a new `otel-rules/banyandb/banyandb-trace-sampling.yaml` rule file. It covers the whole `banyandb_trace_pipeline_*` / `banyandb_trace_tst_pipeline_*` catalog a sampler plugin chain emits — pipeline reconciliation, per-plugin `Decide` execution rate and latency, chain batching, the trace-level evaluated / retained / dropped / immature outcomes, every fail-open guard and bounded-retention counter, drop-set capacity and finalization state, the plugin telemetry-host safety bounds, and the first-party `sw-trace-sampler` / `zipkin-trace-sampler` decision and row metrics. The plugin chain is optional, and the metrics follow it: on a cluster with no sampler configured the wire families are never registered, so every metric here stays absent rather than reading zero. Modeled at Service scope with `group` kept as a metric label rather than at Endpoint scope, so one cluster-wide page can render per-group series and cluster totals alike — OAP does no cross-scope rollup, so an Endpoint-scope metric could not have been aggregated back up to the cluster. * Fix a second `CounterWindow` key collision in the v2 MAL engine, this time ACROSS rules. `rate()` / `increase()` / `irate()` resolve their lower bound from a process-wide window keyed on the counter's own name plus its post-`.sum(...)` label set, with nothing identifying the rule doing the evaluation. Two rules that read one wire family, tell their streams apart with `tagEqual(...)`, and then `.sum(...)` away the label they filtered on therefore collapse onto one window slot and difference against each other's values. The queue is ordered by (timestamp, value), so the smaller counter wins the lower-bound lookup and still reads correctly while its partner is inflated by the gap between them — which is why this went unnoticed. A collision needs the discriminating label to be DROPPED by the `.sum(...)`: where it survives, the rules' label values differ and the window keeps them apart. Auditing the shipped rules on that basis gives 10 colliding keys over ~25 rules — `meter_activemq_cluster_gc_parallel_young_collection_count` reported ~9000/min of young-gen collections from a completely idle broker (differencing against the old-gen counter); MySQL `commands_*` / `tps` rate against each other; so do the GenAI gateway input/output token rates, four Envoy `cluster_*` counters, APISIX matched/unmatched instance bandwidth, and BanyanDB's own `network_recv` / `network_sent`, which drop the `kind` label that separates bytes-received from bytes-sent on one interface. Measured against two live scrapes of the demo cluster's FODC proxy, that last pair was wrong on every interface: `network_sent` read a flat 0 B/s and `network_recv` read large negative values (down to -778 MB/s) from differencing against the sent counter, where both now match the byte delta exactly. No rule changes were needed for any of these -- each rule already reduces to the labels it should; only the window key was wrong. The window is now keyed by (owning rule, counter name, labels). This is the complement of the within-rule collision fixed earlier by keying on the counter's own name: neither name alone is sufficient, because the two collisions are independent. `RunningContext.metricName` — written on every rule evaluation and read by nobody since that earlier fix — is what supplies the rule identity, so no code generation or MAL syntax changes. Note the whole-rule-set comparison suite could not have caught this: it resets the shared window before every rule, the one condition under which the collision cannot appear. * Fix `meter_rabbitmq_node_outgoing_messages_total` double-counting one of its terms. The rule summed six delivery-rate terms but `rabbitmq_global_messages_delivered_get_auto_ack_total` appeared twice, so auto-ack `basic.get` deliveries were counted once more than the other four delivery paths and the reported outgoing rate ran high whenever polling consumers were in use. The duplicate term is removed, leaving the five distinct families (redelivered, consume auto/manual ack, get auto/manual ack). -* Add AI agent conversations landed by the AI Sessionizer: the `AI_AGENT` layer, the bundled `lal/ai-agent.yaml` rule with the `ConversationFile` output builder that verifies and stores Session Data and Session Flow files, the `ai_agent_session_data` and `ai_agent_session_flow` models in a new BanyanDB group `recordsAIAgent`, the `ai-agent-conversation` module that folds a conversation into one `asz.view` document, and the `listConversations` (with optional `conversation` and `title` conditions) / `getConversationRawFiles` GraphQL queries and the streamed `GET /ai-agent/conversations/{conversation}/v1/view` route that serves the document. A round from before the list attributes existed lands and lists with zero counts, and the view shows as much as landed: the chain resumes after a missing, unreadable or refused round, and the absent rounds and files are named once as ranges. A file over `maxFileBytes`, 15 MiB by default, is rejected at ingest and counted under the reason `size`, because one file over BanyanDB's 16 MiB gRPC message limit fails the bulk write it travels in and every record behind it; on MySQL the body column is `LONGTEXT`, since a body that size outgrows `MEDIUMTEXT` as Base64. Each window read is capped at `maxResponseBytes`, 100 MiB by default, as a per-call option on the BanyanDB client in place of its 50 MB default, so the module's reads are bounded by its own settings and nothing else's read changes. The document also carries the session's workspace changes: the Sessionizer's Claude Code plugin lands a Session Data file of kind `changes` beside a stream's transcript, `streams//changes--.sd`, with one change record per observed tool call, stored like any other file, and the runtime's own patch for its editing tools travels as the second data part of the call's result record; `workspace_changes` lists every record joined to its step by tool-use id with the record's fields as `changes/1` lists them, `summary.changes` counts them, and a tool step names its records under `changes`. A round's header counts, `changes`, `lines_added`, `lines_removed`, `llm_calls`, `subagents` and `bash_runs`, land on `ai_agent_session_flow` and reach the conversation list as `ConversationRow` fields, absent rather than zero when the round did not carry them. The layer also takes the agent runtime's own metrics, in the `otel-rules/ai-agent/` rule set enabled by default: Claude Code's `claude_code.token.usage`, as its own exporter sends it or as the Sessionizer derives it from the landed transcripts, per service and per sender, by type, model and query source with `session.id` summed away, plus the exporter's cost, active time, sessions, lines of code, commits, pull requests and edit decisions when they arrive. +* Add AI agent conversations landed by the AI Sessionizer: the `AI_AGENT` layer, the bundled `lal/ai-agent.yaml` rule with the `ConversationFile` output builder that verifies and stores Session Data and Session Flow files, the `ai_agent_session_data` and `ai_agent_session_flow` models in a new BanyanDB group `recordsAIAgent`, the `ai-agent-conversation` module that folds a conversation into one `asz.view` document, and the `listConversations` (with optional `conversation` and `title` conditions) / `getConversationRawFiles` GraphQL queries and the streamed `GET /ai-agent/conversations/{conversation}/v1/view` route that serves the document. The view and raw-file export leave BanyanDB's stage selection unspecified by default and select only the cold stage when the caller explicitly sets `coldStage` to true. A round from before the list attributes existed lands and lists with zero counts, and the view shows as much as landed: the chain resumes after a missing, unreadable or refused round, and the absent rounds and files are named once as ranges. A file over `maxFileBytes`, 15 MiB by default, is rejected at ingest and counted under the reason `size`, because one file over BanyanDB's 16 MiB gRPC message limit fails the bulk write it travels in and every record behind it; on MySQL the body column is `LONGTEXT`, since a body that size outgrows `MEDIUMTEXT` as Base64. Each window read is capped at `maxResponseBytes`, 100 MiB by default, as a per-call option on the BanyanDB client in place of its 50 MB default, so the module's reads are bounded by its own settings and nothing else's read changes. The document also carries the session's workspace changes: the Sessionizer's Claude Code plugin lands a Session Data file of kind `changes` beside a stream's transcript, `streams//changes--.sd`, with one change record per observed tool call, stored like any other file, and the runtime's own patch for its editing tools travels as the second data part of the call's result record; `workspace_changes` lists every record joined to its step by tool-use id with the record's fields as `changes/1` lists them, `summary.changes` counts them, and a tool step names its records under `changes`. A round's header counts, `changes`, `lines_added`, `lines_removed`, `llm_calls`, `subagents` and `bash_runs`, land on `ai_agent_session_flow` and reach the conversation list as `ConversationRow` fields, absent rather than zero when the round did not carry them. The layer also takes the agent runtime's own metrics, in the `otel-rules/ai-agent/` rule set enabled by default: Claude Code's `claude_code.token.usage`, as its own exporter sends it or as the Sessionizer derives it from the landed transcripts, per service and per sender, by type, model and query source with `session.id` summed away, plus the exporter's cost, active time, sessions, lines of code, commits, pull requests and edit decisions when they arrive. * Support querying Zipkin traces from the BanyanDB cold stage through the Zipkin HTTP query API. `/api/v2/traces` takes an optional `coldStage` parameter, and `/api/v2/trace/{traceId}` and `/api/v2/traceMany` take optional `coldStage`, `endTs` and `lookback` parameters; the two by-id lookups had no time range at all before. All of them are SkyWalking additions to the Zipkin API and default to the previous behavior, so existing Zipkin clients such as the Lens UI keep working unchanged, and storages other than BanyanDB ignore `coldStage`. The admin debugging endpoints `/debugging/query/zipkin/api/v2/traces` and `/debugging/query/zipkin/api/v2/trace` accept the same parameters. The BanyanDB stages e2e now seeds Zipkin traces into the generated cold data and verifies the three endpoints against it. * BanyanDB: a query without a time range now covers everything the group's hot/warm stages retain instead of only the last 24 hours. The hard-coded day was a fallback added with the trace model, which requires a time range; it made `queryTrace(traceId)` without a `duration`, the Zipkin `/api/v2/trace/{traceId}` and `/api/v2/traceMany` lookups, and the TraceQL by-id lookup return "not found" for any trace older than a day even though it was still retained. The unbounded range is now bound instead, and BanyanDB's own retention limits the scan. `queryTraces` now rejects a condition with neither `queryDuration` nor `traceId` instead of silently searching the last day; a `traceId` lookup without `queryDuration` searches everything the hot/warm stages retain and reports a 0-to-now `retrievedTimeRange`. * Fix the JDBC storages, MySQL, PostgreSQL and H2, reading only the first day's table of a time range shorter than a day that crosses midnight. `TableHelper` walked the range in 24-hour steps from its first instant, so a range that began late on one day and ended early on the next never reached the second day; every time-ranged read, metrics, traces, logs, alarms, records and the AI agent conversation list, was missing the newer day's rows for the length of the range after each midnight, the last thirty minutes read at 00:10 among them. The range is now walked by calendar day from the start of its first day. diff --git a/docs/en/setup/backend/ai-agent-conversation.md b/docs/en/setup/backend/ai-agent-conversation.md index 1e871ea43281..dd16f441c231 100644 --- a/docs/en/setup/backend/ai-agent-conversation.md +++ b/docs/en/setup/backend/ai-agent-conversation.md @@ -85,14 +85,16 @@ HTTP route on the same server, because its document is as large as the conversat most `limit` (default 1000), then folded to one row per conversation. An optional `conversation` narrows the read to one conversation by id, and an optional `title` keeps only the rows whose title contains the text, case-insensitively — matched after folding, on the newest round's title, so it never widens the rounds read. + On BanyanDB, `duration.coldStage: true` selects the cold stage; otherwise the query uses the default hot/warm stages. - `getConversationRawFiles(condition, files)` lists every landed file and round of a conversation with its id, digest and size; selecting `body` returns the files verbatim, which is the export path. The optional `files` - argument narrows the read to named files. + argument narrows the read to named files. On BanyanDB, `condition.coldStage: true` selects the cold stage, + defaulting to the hot/warm stages when omitted or false. ### The conversation view route ``` -GET /ai-agent/conversations/{conversation}/v1/view?service={serviceName}[&instance={instanceName}] +GET /ai-agent/conversations/{conversation}/v1/view?service={serviceName}[&instance={instanceName}][&coldStage=true] ``` It answers with the whole conversation, once, as one `asz.view` version 1.0 document, the document the @@ -115,6 +117,7 @@ and nothing is cached. |---|---| | `service` / `serviceId` | the service by name, or by id; one of them is required | | `instance` | optional, the sender's instance name from the list row; with it, every storage read is a full series lookup | +| `coldStage` | optional, false by default. On BanyanDB, true selects only the cold stage; otherwise the read uses the default hot/warm stages. The UI passes its selected stage when opening a conversation. Other storages ignore it. | | `Accept` | `application/vnd.skywalking.asz.view+yaml`, or any type naming `yaml`, for YAML; anything else, JSON, as `asz conversation -json` prints it | | `Content-Type` | names the document and its version, the HTTP way: `application/vnd.skywalking.asz.view+json; version=1.0` or `application/vnd.skywalking.asz.view+yaml; version=1.0`. The document's own first two keys, `format` and `version`, say the same | | `Accept-Encoding` | the body is compressed when the client allows; a document is repetitive text and shrinks several times over | @@ -205,9 +208,10 @@ dropped for want of a record worker; drop `ai-agent` from `SW_LOG_LAL_FILES` as with up to `maxResponseBytes`, 100 MiB by default, as a call option on the shared client in place of its 50 MB default, which every other read keeps; the window times `maxFileBytes` must stay under it. Elasticsearch answers at most 10,000 hits to one search. -- A read that is not bound to a duration, the view and the export, covers every retained stage: on BanyanDB the - default stages and, when the group keeps one, the cold stage. A conversation the list found in cold storage - is served, and one that spans stages is served whole. +- The view and the export read over the retention window of the caller's selected stages. On BanyanDB, the + default is hot/warm; cold is queried only when the caller explicitly sets `coldStage: true`. Every round and + file read uses that same selection. A conversation spanning stages can therefore report missing rounds or + files that are outside the selected stages. - When the caller names no sender, the view and the export read across every sender of the service and keep one copy of a file or round two senders both pushed, so a Sessionizer renamed between pushes still yields the whole conversation. diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/ConversationQueryService.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/ConversationQueryService.java index c7fdb55f20d3..be5765412f70 100644 --- a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/ConversationQueryService.java +++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/ConversationQueryService.java @@ -56,8 +56,8 @@ * The read side. A conversation is read in two storage passes and never a read per file: its rounds by * conversation over the whole retention window, then, for each session the head round names, the * files by series id over the range the head round carries, in seq windows that keep one storage response under - * the BanyanDB client's inbound cap. The fold, the chain check and the view are built once per head digest and - * cached. + * the BanyanDB client's inbound cap. The fold, the chain check and the view are built on every call, using + * only the storage stages the caller selected. */ @Slf4j public class ConversationQueryService implements IConversationQueryService { @@ -153,13 +153,11 @@ private static String instanceName(final String instanceId) { @Nullable public Map buildConversationView(final String serviceId, @Nullable final String serviceInstanceId, - final String conversation) throws IOException { - final List heads = - dao().queryRoundsDebuggable(serviceId, serviceInstanceId, conversation, null, 1, false); - if (heads.isEmpty()) { + final String conversation, final boolean coldStage) throws IOException { + final Chain chain = readChain(serviceId, serviceInstanceId, conversation, coldStage); + if (chain.roundInputs.isEmpty()) { return null; } - final Chain chain = readChain(serviceId, serviceInstanceId, conversation); return new ConversationViewBuilder(chain.fold, chain.roundInputs, chain.files, chain.problems).build(); } @@ -168,7 +166,7 @@ public ConversationRawFiles getConversationRawFiles(final String serviceId, @Nullable final String serviceInstanceId, final String conversation, @Nullable final List files, - final boolean includeBody) throws IOException { + final boolean includeBody, final boolean coldStage) throws IOException { final Set wanted = new LinkedHashSet<>(); if (files != null) { for (final String id : files) { @@ -178,7 +176,7 @@ public ConversationRawFiles getConversationRawFiles(final String serviceId, } } } - final List rounds = readRounds(serviceId, serviceInstanceId, conversation); + final List rounds = readRounds(serviceId, serviceInstanceId, conversation, coldStage); final ConversationRawFiles out = new ConversationRawFiles(); if (rounds.isEmpty()) { out.setErrorReason("no round of conversation " + conversation + " is stored for this service"); @@ -227,7 +225,7 @@ public ConversationRawFiles getConversationRawFiles(final String serviceId, : seqs.stream().mapToLong(Long::longValue).max().orElse(0); final long fromSeq = all ? 1 : seqs.stream().mapToLong(Long::longValue).min().orElse(1); final Set seen = new HashSet<>(); - for (final AIAgentSessionDataRecord f : readFiles(serviceId, instance, session, from, to, fromSeq, throughSeq)) { + for (final AIAgentSessionDataRecord f : readFiles(serviceId, instance, session, from, to, fromSeq, throughSeq, coldStage)) { if (!all && !seqs.contains(f.getSeq()) || !seen.add(f.getSeq())) { continue; } @@ -301,8 +299,8 @@ private static final class Chain { * senders or by a redelivery, is kept once, the first copy. */ private List readRounds(final String serviceId, @Nullable final String instance, - final String conversation) throws IOException { - final long headRound = dao().queryHeadRoundDebuggable(serviceId, instance, conversation); + final String conversation, final boolean coldStage) throws IOException { + final long headRound = dao().queryHeadRoundDebuggable(serviceId, instance, conversation, coldStage); if (headRound == 0) { return new ArrayList<>(); } @@ -311,7 +309,7 @@ private List readRounds(final String serviceId, @Nulla for (long start = 1; start <= headRound; start += window) { final long end = Math.min(headRound, start + window - 1); for (final AIAgentSessionFlowRecord r : dao().queryRoundsByNumberDebuggable( - serviceId, instance, conversation, start, end, config.getMaxResponseBytes())) { + serviceId, instance, conversation, start, end, config.getMaxResponseBytes(), coldStage)) { byRound.putIfAbsent(r.getRound(), r); } } @@ -325,9 +323,9 @@ private List readRounds(final String serviceId, @Nulla * Every stored round is listed, readable or not. */ private Chain readChain(final String serviceId, @Nullable final String serviceInstanceId, - final String conversation) throws IOException { + final String conversation, final boolean coldStage) throws IOException { final Chain chain = new Chain(); - final List rounds = readRounds(serviceId, serviceInstanceId, conversation); + final List rounds = readRounds(serviceId, serviceInstanceId, conversation, coldStage); final String instance = StringUtil.isNotEmpty(serviceInstanceId) ? serviceInstanceId : null; long throughSeq = 0; AIAgentSessionFlowRecord headRow = null; @@ -375,7 +373,7 @@ private Chain readChain(final String serviceId, @Nullable final String serviceIn sessions.add(id.startsWith("session/") ? id.substring("session/".length()) : id); } for (final String session : sessions) { - for (final AIAgentSessionDataRecord f : readFiles(serviceId, instance, session, from, to, 1, throughSeq)) { + for (final AIAgentSessionDataRecord f : readFiles(serviceId, instance, session, from, to, 1, throughSeq, coldStage)) { if (chain.files.containsKey(f.getSeq())) { // the same file under two senders; the chain check judges the copy that was kept continue; @@ -399,13 +397,14 @@ private static String first12(@Nullable final String s) { private List readFiles(final String serviceId, @Nullable final String instance, final String session, final long from, final long to, - final long fromSeq, final long throughSeq) throws IOException { + final long fromSeq, final long throughSeq, + final boolean coldStage) throws IOException { final List out = new ArrayList<>(); final int window = config.getFileReadWindow(); for (long start = fromSeq; start <= throughSeq; start += window) { final long end = Math.min(throughSeq, start + window - 1); out.addAll(dao().queryFilesDebuggable( - serviceId, instance, session, from, to, start, end, config.getMaxResponseBytes())); + serviceId, instance, session, from, to, start, end, config.getMaxResponseBytes(), coldStage)); } return out; } diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/IConversationQueryService.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/IConversationQueryService.java index 8922febf51ea..248a3fde63a8 100644 --- a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/IConversationQueryService.java +++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/IConversationQueryService.java @@ -54,12 +54,13 @@ ConversationList listConversations(String serviceId, @Nullable String serviceIns * @param serviceId the service * @param serviceInstanceId the sender, or null * @param conversation the conversation + * @param coldStage whether the caller explicitly selected BanyanDB's cold stage * @return the document as ordered maps, or null when the service stores no round of the conversation * @throws IOException on a storage failure */ @Nullable Map buildConversationView(String serviceId, @Nullable String serviceInstanceId, - String conversation) throws IOException; + String conversation, boolean coldStage) throws IOException; /** * Every landed file and round of a conversation as stored, or only the named ones. @@ -69,10 +70,11 @@ Map buildConversationView(String serviceId, @Nullable String ser * @param conversation the conversation * @param files only these file ids, or null for every file * @param includeBody whether the caller selected the body field + * @param coldStage whether the caller explicitly selected BanyanDB's cold stage * @return the files * @throws IOException on a storage failure */ ConversationRawFiles getConversationRawFiles(String serviceId, @Nullable String serviceInstanceId, String conversation, @Nullable List files, - boolean includeBody) throws IOException; + boolean includeBody, boolean coldStage) throws IOException; } diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/NoneConversationQueryService.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/NoneConversationQueryService.java index 0ba89754647a..477fe96c34c2 100644 --- a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/NoneConversationQueryService.java +++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/NoneConversationQueryService.java @@ -50,7 +50,7 @@ public ConversationList listConversations(final String serviceId, @Override public Map buildConversationView(final String serviceId, @Nullable final String serviceInstanceId, - final String conversation) { + final String conversation, final boolean coldStage) { return null; } @@ -59,7 +59,7 @@ public ConversationRawFiles getConversationRawFiles(final String serviceId, @Nullable final String serviceInstanceId, final String conversation, @Nullable final List files, - final boolean includeBody) { + final boolean includeBody, final boolean coldStage) { final ConversationRawFiles rawFiles = new ConversationRawFiles(); rawFiles.setErrorReason(DISABLED); return rawFiles; diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/http/ConversationViewHandler.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/http/ConversationViewHandler.java index a05c97a4a8c0..ae7640602f79 100644 --- a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/http/ConversationViewHandler.java +++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/http/ConversationViewHandler.java @@ -28,6 +28,7 @@ import com.linecorp.armeria.common.util.TimeoutMode; import com.linecorp.armeria.server.ServiceRequestContext; import com.linecorp.armeria.server.annotation.Decorator; +import com.linecorp.armeria.server.annotation.Default; import com.linecorp.armeria.server.annotation.Get; import com.linecorp.armeria.server.annotation.Header; import com.linecorp.armeria.server.annotation.Param; @@ -55,7 +56,8 @@ * the client allows, and given its own request timeout in place of the server's default. * *

Query parameters: service, the service name, or serviceId; optionally - * instance, the sender's instance name. What the body is, the HTTP layer says: the media type + * instance, the sender's instance name, and coldStage, false by default, to query + * BanyanDB's cold stage. What the body is, the HTTP layer says: the media type * names the document format and its version, application/vnd.skywalking.asz.view+json; version=1.0, or the * +yaml twin when Accept asks for YAML. The document's own first two keys repeat it. * @@ -91,6 +93,7 @@ public HttpResponse view(final ServiceRequestContext ctx, @Param("service") @Nullable final String serviceName, @Param("serviceId") @Nullable final String serviceIdParam, @Param("instance") @Nullable final String instanceName, + @Param("coldStage") @Default("false") final boolean coldStage, @Header("Accept") @Nullable final String accept) { final String serviceId; if (StringUtil.isNotEmpty(serviceIdParam)) { @@ -106,15 +109,15 @@ public HttpResponse view(final ServiceRequestContext ctx, ctx.setRequestTimeout(TimeoutMode.SET_FROM_NOW, timeout); final HttpResponseWriter res = HttpResponse.streaming(); - ctx.blockingTaskExecutor().execute(() -> stream(res, serviceId, instanceId, conversation, yaml)); + ctx.blockingTaskExecutor().execute(() -> stream(res, serviceId, instanceId, conversation, yaml, coldStage)); return res; } private void stream(final HttpResponseWriter res, final String serviceId, @Nullable final String instanceId, - final String conversation, final boolean yaml) { + final String conversation, final boolean yaml, final boolean coldStage) { final Map doc; try { - doc = service.buildConversationView(serviceId, instanceId, conversation); + doc = service.buildConversationView(serviceId, instanceId, conversation, coldStage); } catch (final Exception e) { // a storage client can surface a checked failure it never declared; whatever it is, the response // must say so, or the caller waits for the request timeout diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/input/ConversationCondition.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/input/ConversationCondition.java index 5804d5caf660..74b6d83399b4 100644 --- a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/input/ConversationCondition.java +++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/input/ConversationCondition.java @@ -27,4 +27,5 @@ public class ConversationCondition { private ServiceCondition service; private String conversation; private InstanceCondition instance; + private boolean coldStage; } diff --git a/oap-server/analyzer/ai-agent-conversation/src/test/java/org/apache/skywalking/oap/server/ai/agent/conversation/NoneAIAgentConversationProviderTest.java b/oap-server/analyzer/ai-agent-conversation/src/test/java/org/apache/skywalking/oap/server/ai/agent/conversation/NoneAIAgentConversationProviderTest.java index 7558d0b8216a..87e88004f232 100644 --- a/oap-server/analyzer/ai-agent-conversation/src/test/java/org/apache/skywalking/oap/server/ai/agent/conversation/NoneAIAgentConversationProviderTest.java +++ b/oap-server/analyzer/ai-agent-conversation/src/test/java/org/apache/skywalking/oap/server/ai/agent/conversation/NoneAIAgentConversationProviderTest.java @@ -53,8 +53,8 @@ public void testTheDisabledModuleStoresAndAnswersNothing() throws Exception { .getConversations() .isEmpty()); assertNotNull(service.listConversations("1", null, null, null, new Duration(), null).getErrorReason()); - assertNull(service.buildConversationView("1", null, "c")); - assertTrue(service.getConversationRawFiles("1", null, "c", Collections.emptyList(), true) + assertNull(service.buildConversationView("1", null, "c", false)); + assertTrue(service.getConversationRawFiles("1", null, "c", Collections.emptyList(), true, false) .getFiles() .isEmpty()); } diff --git a/oap-server/analyzer/ai-agent-conversation/src/test/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/ConversationQueryServiceTest.java b/oap-server/analyzer/ai-agent-conversation/src/test/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/ConversationQueryServiceTest.java index 22d2a7373793..f9f4c6fbc4a9 100644 --- a/oap-server/analyzer/ai-agent-conversation/src/test/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/ConversationQueryServiceTest.java +++ b/oap-server/analyzer/ai-agent-conversation/src/test/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/ConversationQueryServiceTest.java @@ -39,8 +39,12 @@ import org.apache.skywalking.oap.server.library.module.ModuleProviderHolder; import org.apache.skywalking.oap.server.library.module.ModuleServiceHolder; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; @@ -51,6 +55,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; import static org.mockito.Mockito.when; /** @@ -110,17 +115,17 @@ public void theHeadIsTheHighestRoundStoredNotTheNewestOfAListPage() throws Excep final String conversation = first.getHeader().getConversation(); final byte[] second = Fixtures.emptyRound(first, 2, first.getCommitDigest(), 4, 4, first.getHeader().getParser()); final IAIAgentConversationQueryDAO dao = mock(IAIAgentConversationQueryDAO.class); - when(dao.queryHeadRoundDebuggable(eq(SERVICE), any(), eq(conversation))).thenReturn(2L); - when(dao.queryRoundsByNumberDebuggable(eq(SERVICE), any(), eq(conversation), anyLong(), anyLong(), anyInt())) + when(dao.queryHeadRoundDebuggable(eq(SERVICE), any(), eq(conversation), eq(false))).thenReturn(2L); + when(dao.queryRoundsByNumberDebuggable(eq(SERVICE), any(), eq(conversation), anyLong(), anyLong(), anyInt(), eq(false))) .thenReturn(Arrays.asList( storedRound(conversation, 1, Fixtures.bytes(Fixtures.ROUND_FILE)), storedRound(conversation, 2, second))); - when(dao.queryFilesDebuggable(anyString(), any(), anyString(), anyLong(), anyLong(), anyLong(), anyLong(), anyInt())) + when(dao.queryFilesDebuggable(anyString(), any(), anyString(), anyLong(), anyLong(), anyLong(), anyLong(), anyInt(), eq(false))) .thenReturn(storedFiles()); final AIAgentConversationConfig config = new AIAgentConversationConfig(); config.setMaxListLimit(1); final ConversationRawFiles out = service(dao, config) - .getConversationRawFiles(SERVICE, null, conversation, null, false); + .getConversationRawFiles(SERVICE, null, conversation, null, false, false); assertNull(out.getErrorReason()); final List rounds = new ArrayList<>(); @@ -145,26 +150,81 @@ public void anUnreadableNewestRoundDoesNotBlockTheExport() throws Exception { final byte[] truncated = "{\"t\":\"header\",\"schema\":\"sf/1\"".getBytes(StandardCharsets.UTF_8); final AIAgentSessionFlowRecord broken = storedRound(conversation, 2, truncated); final IAIAgentConversationQueryDAO dao = mock(IAIAgentConversationQueryDAO.class); - when(dao.queryHeadRoundDebuggable(eq(SERVICE), any(), eq(conversation))).thenReturn(2L); - when(dao.queryRoundsByNumberDebuggable(eq(SERVICE), any(), eq(conversation), anyLong(), anyLong(), anyInt())) + when(dao.queryHeadRoundDebuggable(eq(SERVICE), any(), eq(conversation), eq(false))).thenReturn(2L); + when(dao.queryRoundsByNumberDebuggable(eq(SERVICE), any(), eq(conversation), anyLong(), anyLong(), anyInt(), eq(false))) .thenReturn(Arrays.asList(storedRound(conversation, 1, Fixtures.bytes(Fixtures.ROUND_FILE)), broken)); - when(dao.queryFilesDebuggable(anyString(), any(), anyString(), anyLong(), anyLong(), anyLong(), anyLong(), anyInt())) + when(dao.queryFilesDebuggable(anyString(), any(), anyString(), anyLong(), anyLong(), anyLong(), anyLong(), anyInt(), eq(false))) .thenReturn(storedFiles()); final ConversationQueryService service = service(dao, new AIAgentConversationConfig()); final String firstId = FileNames.roundFile(conversation, 1, first.getCommitDigest()); final ConversationRawFiles selected = service.getConversationRawFiles( - SERVICE, null, conversation, Collections.singletonList(firstId), true); + SERVICE, null, conversation, Collections.singletonList(firstId), true, false); assertNull(selected.getErrorReason()); assertEquals(1, selected.getFiles().size()); assertEquals(firstId, selected.getFiles().get(0).getId()); assertEquals(new String(Fixtures.bytes(Fixtures.ROUND_FILE), StandardCharsets.UTF_8), selected.getFiles().get(0).getBody()); - final ConversationRawFiles all = service.getConversationRawFiles(SERVICE, null, conversation, null, false); + final ConversationRawFiles all = service.getConversationRawFiles(SERVICE, null, conversation, null, false, false); assertNull(all.getErrorReason()); assertEquals(Fixtures.DATA_FILES.length + 2, all.getFiles().size()); final ConversationRawFile last = all.getFiles().get(all.getFiles().size() - 1); assertEquals(2, last.getRound()); assertTrue(last.getId().endsWith(broken.getDigest().substring(0, 12) + ".sf"), last.getId()); } + + @ParameterizedTest + @CsvSource({"false, false", "false, true", "true, false", "true, true"}) + public void everyViewAndExportWindowUsesTheRequestedStage(final boolean export, final boolean coldStage) throws Exception { + final SessionFlowRound first = Fixtures.round(); + final String conversation = first.getHeader().getConversation(); + final byte[] second = Fixtures.emptyRound(first, 2, first.getCommitDigest(), 4, 4, first.getHeader().getParser()); + final List files = storedFiles(); + final IAIAgentConversationQueryDAO dao = mock(IAIAgentConversationQueryDAO.class); + final AIAgentConversationConfig config = new AIAgentConversationConfig(); + config.setRoundReadWindow(1); + config.setFileReadWindow(2); + when(dao.queryHeadRoundDebuggable(SERVICE, null, conversation, coldStage)).thenReturn(2L); + when(dao.queryRoundsByNumberDebuggable(SERVICE, null, conversation, 1, 1, config.getMaxResponseBytes(), coldStage)) + .thenReturn(Collections.singletonList(storedRound(conversation, 1, Fixtures.bytes(Fixtures.ROUND_FILE)))); + when(dao.queryRoundsByNumberDebuggable(SERVICE, null, conversation, 2, 2, config.getMaxResponseBytes(), coldStage)) + .thenReturn(Collections.singletonList(storedRound(conversation, 2, second))); + when(dao.queryFilesDebuggable(eq(SERVICE), any(), eq(Fixtures.SESSION), anyLong(), anyLong(), + eq(1L), eq(2L), eq(config.getMaxResponseBytes()), eq(coldStage))) + .thenReturn(files.subList(0, 2)); + when(dao.queryFilesDebuggable(eq(SERVICE), any(), eq(Fixtures.SESSION), anyLong(), anyLong(), + eq(3L), eq(4L), eq(config.getMaxResponseBytes()), eq(coldStage))) + .thenReturn(files.subList(2, 4)); + final ConversationQueryService service = service(dao, config); + + if (export) { + final ConversationRawFiles result = service.getConversationRawFiles( + SERVICE, null, conversation, null, false, coldStage); + assertNull(result.getErrorReason()); + assertEquals(files.size() + 2, result.getFiles().size()); + } else { + assertNotNull(service.buildConversationView(SERVICE, null, conversation, coldStage)); + } + + verify(dao).queryHeadRoundDebuggable(SERVICE, null, conversation, coldStage); + verify(dao).queryRoundsByNumberDebuggable(SERVICE, null, conversation, 1, 1, config.getMaxResponseBytes(), coldStage); + verify(dao).queryRoundsByNumberDebuggable(SERVICE, null, conversation, 2, 2, config.getMaxResponseBytes(), coldStage); + verify(dao).queryFilesDebuggable(eq(SERVICE), any(), eq(Fixtures.SESSION), anyLong(), anyLong(), + eq(1L), eq(2L), eq(config.getMaxResponseBytes()), eq(coldStage)); + verify(dao).queryFilesDebuggable(eq(SERVICE), any(), eq(Fixtures.SESSION), anyLong(), anyLong(), + eq(3L), eq(4L), eq(config.getMaxResponseBytes()), eq(coldStage)); + verifyNoMoreInteractions(dao); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void aMissingConversationDoesNotFallBackToAnotherStage(final boolean coldStage) throws Exception { + final IAIAgentConversationQueryDAO dao = mock(IAIAgentConversationQueryDAO.class); + + assertNull(service(dao, new AIAgentConversationConfig()).buildConversationView( + SERVICE, null, Fixtures.SESSION, coldStage)); + + verify(dao).queryHeadRoundDebuggable(SERVICE, null, Fixtures.SESSION, coldStage); + verifyNoMoreInteractions(dao); + } } diff --git a/oap-server/analyzer/ai-agent-conversation/src/test/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/http/ConversationViewHandlerTest.java b/oap-server/analyzer/ai-agent-conversation/src/test/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/http/ConversationViewHandlerTest.java index f04cc8d8b78d..196acd4c9662 100644 --- a/oap-server/analyzer/ai-agent-conversation/src/test/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/http/ConversationViewHandlerTest.java +++ b/oap-server/analyzer/ai-agent-conversation/src/test/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/http/ConversationViewHandlerTest.java @@ -62,6 +62,8 @@ import org.apache.skywalking.oap.server.core.query.input.Duration; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; import org.yaml.snakeyaml.Yaml; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -77,6 +79,7 @@ public class ConversationViewHandlerTest { private static final String SERVICE_ID = IDManager.ServiceID.buildId(SERVICE, true); private static final Map DOC = fixtureDocument(); private static final AtomicReference LAST_INSTANCE_ID = new AtomicReference<>(); + private static final AtomicReference LAST_COLD_STAGE = new AtomicReference<>(); private static final IConversationQueryService STUB = new IConversationQueryService() { @Override @@ -90,8 +93,10 @@ public ConversationList listConversations(final String serviceId, @Nullable fina @Nullable public Map buildConversationView(final String serviceId, @Nullable final String serviceInstanceId, - final String conversation) throws IOException { + final String conversation, + final boolean coldStage) throws IOException { LAST_INSTANCE_ID.set(serviceInstanceId); + LAST_COLD_STAGE.set(coldStage); if ("broken".equals(conversation)) { throw new IOException("storage is down"); } @@ -103,7 +108,8 @@ public ConversationRawFiles getConversationRawFiles(final String serviceId, @Nullable final String serviceInstanceId, final String conversation, @Nullable final List files, - final boolean includeBody) { + final boolean includeBody, + final boolean coldStage) { throw new UnsupportedOperationException(); } }; @@ -238,6 +244,14 @@ public void theInstanceParameterNamesTheSender() { assertEquals(IDManager.ServiceInstanceID.buildId(SERVICE_ID, "sender-1"), LAST_INSTANCE_ID.get()); } + @ParameterizedTest + @CsvSource({"'', false", "&coldStage=false, false", "&coldStage=true, true"}) + public void coldStageRequiresAnExplicitRequest(final String parameter, final boolean expectedColdStage) { + LAST_COLD_STAGE.set(null); + assertEquals(200, get(path(Fixtures.SESSION) + parameter).status().code()); + assertEquals(expectedColdStage, LAST_COLD_STAGE.get()); + } + @Test public void statusesOfTheErrorPaths() { assertEquals(404, get(path("no-such-conversation")).status().code()); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IAIAgentConversationQueryDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IAIAgentConversationQueryDAO.java index 187e426f3d86..62b700ffebf9 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IAIAgentConversationQueryDAO.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IAIAgentConversationQueryDAO.java @@ -60,16 +60,17 @@ default List queryRoundsDebuggable(String serviceId, default long queryHeadRoundDebuggable(String serviceId, @Nullable String serviceInstanceId, - String conversation) throws IOException { + String conversation, + boolean coldStage) throws IOException { final DebuggingTraceContext traceContext = TRACE_CONTEXT.get(); DebuggingSpan span = null; try { if (traceContext != null) { span = traceContext.createSpan("Query Dao: queryHeadRound"); span.setMsg("ServiceId: " + serviceId + ", ServiceInstanceId: " + serviceInstanceId - + ", Conversation: " + conversation); + + ", Conversation: " + conversation + ", ColdStage: " + coldStage); } - return queryHeadRound(serviceId, serviceInstanceId, conversation); + return queryHeadRound(serviceId, serviceInstanceId, conversation, coldStage); } finally { if (traceContext != null && span != null) { traceContext.stopSpan(span); @@ -82,16 +83,19 @@ default List queryRoundsByNumberDebuggable(String serv String conversation, long fromRound, long throughRound, - int maxResponseBytes) throws IOException { + int maxResponseBytes, + boolean coldStage) throws IOException { final DebuggingTraceContext traceContext = TRACE_CONTEXT.get(); DebuggingSpan span = null; try { if (traceContext != null) { span = traceContext.createSpan("Query Dao: queryRoundsByNumber"); span.setMsg("ServiceId: " + serviceId + ", ServiceInstanceId: " + serviceInstanceId - + ", Conversation: " + conversation + ", Round: " + fromRound + ".." + throughRound); + + ", Conversation: " + conversation + ", Round: " + fromRound + ".." + throughRound + + ", ColdStage: " + coldStage); } - return queryRoundsByNumber(serviceId, serviceInstanceId, conversation, fromRound, throughRound, maxResponseBytes); + return queryRoundsByNumber( + serviceId, serviceInstanceId, conversation, fromRound, throughRound, maxResponseBytes, coldStage); } finally { if (traceContext != null && span != null) { traceContext.stopSpan(span); @@ -106,7 +110,8 @@ default List queryFilesDebuggable(String serviceId, long toTimestamp, long fromSeq, long throughSeq, - int maxResponseBytes) throws IOException { + int maxResponseBytes, + boolean coldStage) throws IOException { final DebuggingTraceContext traceContext = TRACE_CONTEXT.get(); DebuggingSpan span = null; try { @@ -114,10 +119,11 @@ default List queryFilesDebuggable(String serviceId, span = traceContext.createSpan("Query Dao: queryFiles"); span.setMsg("ServiceId: " + serviceId + ", ServiceInstanceId: " + serviceInstanceId + ", Session: " + session + ", From: " + fromTimestamp + ", To: " + toTimestamp - + ", Seq: " + fromSeq + ".." + throughSeq); + + ", Seq: " + fromSeq + ".." + throughSeq + ", ColdStage: " + coldStage); } return queryFiles( - serviceId, serviceInstanceId, session, fromTimestamp, toTimestamp, fromSeq, throughSeq, maxResponseBytes); + serviceId, serviceInstanceId, session, fromTimestamp, toTimestamp, fromSeq, throughSeq, + maxResponseBytes, coldStage); } finally { if (traceContext != null && span != null) { traceContext.stopSpan(span); @@ -131,7 +137,8 @@ default List queryFilesDebuggable(String serviceId, * @param serviceId the service * @param serviceInstanceId the sender, or null for every sender of the service * @param conversation one conversation, or null for every conversation - * @param duration the time window, or null for the whole retention window + * @param duration the time window and stage selection, or null for the whole retention window in + * the default stages; cold storage is queried only when explicitly selected * @param limit at most this many rows * @param includeBody whether to read the body column; the list page does not * @return the rounds, newest first @@ -145,21 +152,24 @@ List queryRounds(String serviceId, boolean includeBody) throws IOException; /** - * The highest round number stored for a conversation, over every retained stage, or 0 when none is. That is + * The highest round number stored for a conversation in the selected stages, or 0 when none is. That is * the chain's head; it is not the newest row by time, because the Sessionizer can write a later round that * carries no later activity, and it is read directly rather than off a page of rows. * * @param serviceId the service * @param serviceInstanceId the sender, or null for every sender of the service * @param conversation the conversation + * @param coldStage true to query only cold storage, false for the default stages; ignored by storage + * implementations without stage selection * @return the highest round number, or 0 * @throws IOException on a storage failure */ - long queryHeadRound(String serviceId, @Nullable String serviceInstanceId, String conversation) throws IOException; + long queryHeadRound(String serviceId, @Nullable String serviceInstanceId, String conversation, + boolean coldStage) throws IOException; /** - * The rounds of one conversation whose numbers lie in a window, bodies included, in round order, over every - * retained stage. A round is up to 2 MiB, so the caller reads a long chain window by window. + * The rounds of one conversation whose numbers lie in a window, bodies included, in round order, in the + * selected stages. A round is up to 2 MiB, so the caller reads a long chain window by window. * * @param serviceId the service * @param serviceInstanceId the sender, or null for every sender of the service @@ -168,12 +178,15 @@ List queryRounds(String serviceId, * @param throughRound the last round number of the window * @param maxResponseBytes the most bytes one storage response may carry; a storage that caps a response per * call applies it to this read, another ignores it + * @param coldStage true to query only cold storage, false for the default stages; ignored by storage + * implementations without stage selection * @return the rounds of the window, in round order * @throws IOException on a storage failure */ List queryRoundsByNumber(String serviceId, @Nullable String serviceInstanceId, String conversation, long fromRound, - long throughRound, int maxResponseBytes) throws IOException; + long throughRound, int maxResponseBytes, + boolean coldStage) throws IOException; /** * The files of one session whose seq is within the window and whose timestamp is within the range, with @@ -188,6 +201,8 @@ List queryRoundsByNumber(String serviceId, @Nullable S * @param throughSeq inclusive last seq * @param maxResponseBytes the most bytes one storage response may carry; a storage that caps a response per * call applies it to this read, another ignores it + * @param coldStage true to query only cold storage, false for the default stages; ignored by storage + * implementations without stage selection * @return the files, seq ascending * @throws IOException on a storage failure */ @@ -198,5 +213,6 @@ List queryFiles(String serviceId, long toTimestamp, long fromSeq, long throughSeq, - int maxResponseBytes) throws IOException; + int maxResponseBytes, + boolean coldStage) throws IOException; } diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/AIAgentConversationQuery.java b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/AIAgentConversationQuery.java index 8d6def069f17..56e43bc2c663 100644 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/AIAgentConversationQuery.java +++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/AIAgentConversationQuery.java @@ -106,7 +106,8 @@ public CompletableFuture getConversationRawFiles(final Con instanceId(condition.getInstance()), condition.getConversation(), files, - includeBody + includeBody, + condition.isColdStage() ); if (debug) { raw.setDebuggingTrace(traceContext.getExecTrace()); diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol b/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol index 9570bbd81f43..35ae8f2a7662 160000 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol +++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol @@ -1 +1 @@ -Subproject commit 9570bbd81f43c78c805638eaec5e9f0ef7019481 +Subproject commit 35ae8f2a7662df3d0b9681f635f24bf4e3f23d28 diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBAIAgentConversationQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBAIAgentConversationQueryDAO.java index cd26755f7efb..7cdbc8b069a4 100644 --- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBAIAgentConversationQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBAIAgentConversationQueryDAO.java @@ -33,15 +33,14 @@ import org.apache.skywalking.oap.server.core.storage.query.IAIAgentConversationQueryDAO; import org.apache.skywalking.oap.server.library.util.StringUtil; import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageClient; -import org.apache.skywalking.oap.server.storage.plugin.banyandb.MetadataRegistry; /** * Both reads are series lookups: the rounds by (service, instance) with conversation as an * indexed tag, the files by (service, instance, session) with a range on the indexed seq. * A missing instance is a partial series match, the way the log query works by service alone. * - *

A read that is not bound to a duration covers every retained stage: the default stages and, when the group - * has one, the cold stage, in two queries merged here. The list page alone follows its duration's stage. + *

Each read targets either the default stages or the explicitly selected cold stage. A duration-free round + * query uses the default stages. */ public class BanyanDBAIAgentConversationQueryDAO extends AbstractBanyanDBDAO implements IAIAgentConversationQueryDAO { private static final Set ROUND_TAGS = ImmutableSet.of( @@ -105,14 +104,10 @@ public List queryRounds(final String serviceId, } where.orderByDesc().limit(limit); final Set tags = includeBody ? ROUND_TAGS_WITH_BODY : ROUND_TAGS; - final List rows; - if (duration == null) { - rows = everyStage(AIAgentSessionFlowRecord.INDEX_NAME, tags, everythingRetained(), where, 0); - } else { - rows = queryDebuggable( - duration.isColdStage(), AIAgentSessionFlowRecord.INDEX_NAME, tags, getTimestampRange(duration), where - ).getElements(); - } + final List rows = queryDebuggable( + duration != null && duration.isColdStage(), AIAgentSessionFlowRecord.INDEX_NAME, tags, + duration == null ? everythingRetained() : getTimestampRange(duration), where + ).getElements(); final List rounds = rounds(rows, includeBody); rounds.sort((a, b) -> Long.compare(b.getTimestamp(), a.getTimestamp())); return rounds.size() > limit ? new ArrayList<>(rounds.subList(0, limit)) : rounds; @@ -120,7 +115,7 @@ public List queryRounds(final String serviceId, @Override public long queryHeadRound(final String serviceId, @Nullable final String serviceInstanceId, - final String conversation) throws IOException { + final String conversation, final boolean coldStage) throws IOException { final Conditions where = Conditions.create(); where.eq(AIAgentSessionFlowRecord.SERVICE_ID, serviceId); if (StringUtil.isNotEmpty(serviceInstanceId)) { @@ -129,9 +124,9 @@ public long queryHeadRound(final String serviceId, @Nullable final String servic where.eq(AIAgentSessionFlowRecord.CONVERSATION, conversation); where.orderByDesc(AIAgentSessionFlowRecord.ROUND).limit(1); long head = 0; - for (final RowEntity row : everyStage( - AIAgentSessionFlowRecord.INDEX_NAME, ImmutableSet.of(AIAgentSessionFlowRecord.ROUND), everythingRetained(), - where, 0)) { + for (final RowEntity row : queryDebuggable( + coldStage, AIAgentSessionFlowRecord.INDEX_NAME, ImmutableSet.of(AIAgentSessionFlowRecord.ROUND), + everythingRetained(), where).getElements()) { head = Math.max(head, longOf(row.getTagValue(AIAgentSessionFlowRecord.ROUND))); } return head; @@ -143,7 +138,8 @@ public List queryRoundsByNumber(final String serviceId final String conversation, final long fromRound, final long throughRound, - final int maxResponseBytes) throws IOException { + final int maxResponseBytes, + final boolean coldStage) throws IOException { final Conditions where = Conditions.create(); where.eq(AIAgentSessionFlowRecord.SERVICE_ID, serviceId); if (StringUtil.isNotEmpty(serviceInstanceId)) { @@ -155,8 +151,8 @@ public List queryRoundsByNumber(final String serviceId // every row of the window, up to the client's result window: a round two senders both pushed is there twice where.orderByAsc(); final List rounds = rounds( - everyStage(AIAgentSessionFlowRecord.INDEX_NAME, ROUND_TAGS_WITH_BODY, everythingRetained(), where, - maxResponseBytes), true); + queryDebuggable(coldStage, AIAgentSessionFlowRecord.INDEX_NAME, ROUND_TAGS_WITH_BODY, everythingRetained(), + where, maxResponseBytes).getElements(), true); rounds.sort((a, b) -> Long.compare(a.getRound(), b.getRound())); return rounds; } @@ -202,7 +198,8 @@ public List queryFiles(final String serviceId, final long toTimestamp, final long fromSeq, final long throughSeq, - final int maxResponseBytes) throws IOException { + final int maxResponseBytes, + final boolean coldStage) throws IOException { final Conditions where = Conditions.create(); where.eq(AIAgentSessionDataRecord.SERVICE_ID, serviceId); if (StringUtil.isNotEmpty(serviceInstanceId)) { @@ -212,9 +209,9 @@ public List queryFiles(final String serviceId, where.gte(AIAgentSessionDataRecord.SEQ, fromSeq); where.lte(AIAgentSessionDataRecord.SEQ, throughSeq); where.orderByAsc(); - final List rows = everyStage( - AIAgentSessionDataRecord.INDEX_NAME, FILE_TAGS, - new TimestampRange(Math.max(0, fromTimestamp - 1), toTimestamp + 1), where, maxResponseBytes); + final List rows = queryDebuggable( + coldStage, AIAgentSessionDataRecord.INDEX_NAME, FILE_TAGS, + new TimestampRange(Math.max(0, fromTimestamp - 1), toTimestamp + 1), where, maxResponseBytes).getElements(); final List files = new ArrayList<>(rows.size()); for (final RowEntity row : rows) { final AIAgentSessionDataRecord record = new AIAgentSessionDataRecord(); @@ -245,19 +242,4 @@ private static Long nullableLongOf(final Object value) { private static TimestampRange everythingRetained() { return new TimestampRange(0, System.currentTimeMillis() + CLOCK_SKEW_MILLIS); } - - /** - * The rows of the default stages and, when the group keeps one, of the cold stage: a conversation can span - * the two, and its list row may come from either. - */ - private List everyStage(final String model, final Set tags, final TimestampRange range, - final Conditions where, final int maxResponseBytes) throws IOException { - final List rows = new ArrayList<>( - queryDebuggable(false, model, tags, range, where, maxResponseBytes).getElements()); - final MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findRecordMetadata(model); - if (schema != null && schema.getMetadata().getResource().isEnableColdStage()) { - rows.addAll(queryDebuggable(true, model, tags, range, where, maxResponseBytes).getElements()); - } - return rows; - } } diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBAIAgentConversationQueryDAOTest.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBAIAgentConversationQueryDAOTest.java new file mode 100644 index 000000000000..b9103ed04e33 --- /dev/null +++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBAIAgentConversationQueryDAOTest.java @@ -0,0 +1,127 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.storage.plugin.banyandb.stream; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import org.apache.skywalking.library.banyandb.v1.client.StreamQueryResponse; +import org.apache.skywalking.library.banyandb.v1.client.metadata.Serializable; +import org.apache.skywalking.oap.server.core.analysis.DownSampling; +import org.apache.skywalking.oap.server.core.analysis.manual.aiagent.AIAgentSessionDataRecord; +import org.apache.skywalking.oap.server.core.analysis.manual.aiagent.AIAgentSessionFlowRecord; +import org.apache.skywalking.oap.server.core.query.input.Duration; +import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine; +import org.apache.skywalking.oap.server.core.storage.annotation.Storage; +import org.apache.skywalking.oap.server.core.storage.model.Model; +import org.apache.skywalking.oap.server.core.storage.model.StorageManipulationOpt; +import org.apache.skywalking.oap.server.core.storage.model.StorageModels; +import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageClient; +import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageConfig; +import org.apache.skywalking.oap.server.storage.plugin.banyandb.MetadataRegistry; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.mockito.MockedStatic; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.when; + +class BanyanDBAIAgentConversationQueryDAOTest { + private final List emitted = new ArrayList<>(); + private final List registeredModels = new ArrayList<>(); + private BanyanDBAIAgentConversationQueryDAO dao; + + @BeforeEach + void setUp() throws Exception { + final BanyanDBStorageConfig config = new BanyanDBStorageConfig(); + config.getRecordsAIAgent().setEnableColdStage(true); + final StorageModels models = new StorageModels(); + try (MockedStatic scopes = mockStatic(DefaultScopeDefine.class)) { + scopes.when(() -> DefaultScopeDefine.nameOf(DefaultScopeDefine.AI_AGENT_SESSION_FLOW)) + .thenReturn("AIAgentSessionFlow"); + scopes.when(() -> DefaultScopeDefine.nameOf(DefaultScopeDefine.AI_AGENT_SESSION_DATA)) + .thenReturn("AIAgentSessionData"); + registeredModels.add(models.add( + AIAgentSessionFlowRecord.class, DefaultScopeDefine.AI_AGENT_SESSION_FLOW, + new Storage(AIAgentSessionFlowRecord.INDEX_NAME, true, DownSampling.Second), + StorageManipulationOpt.withSchemaChange())); + registeredModels.add(models.add( + AIAgentSessionDataRecord.class, DefaultScopeDefine.AI_AGENT_SESSION_DATA, + new Storage(AIAgentSessionDataRecord.INDEX_NAME, true, DownSampling.Second), + StorageManipulationOpt.withSchemaChange())); + } + registeredModels.forEach(model -> MetadataRegistry.INSTANCE.registerStreamModel(model, config)); + final BanyanDBStorageClient client = mock(BanyanDBStorageClient.class); + when(client.getResultWindowMaxSize()).thenReturn(10000); + final StreamQueryResponse response = mock(StreamQueryResponse.class); + when(response.getElements()).thenReturn(Collections.emptyList()); + when(client.queryStream(anyInt(), anyString(), any(Serializable[].class))).thenAnswer(invocation -> { + emitted.add(invocation.getArgument(1)); + return response; + }); + dao = new BanyanDBAIAgentConversationQueryDAO(client); + } + + @AfterEach + void tearDown() { + registeredModels.forEach(MetadataRegistry.INSTANCE::evict); + } + + @Test + void roundsWithoutDurationDoNotQueryColdStorageEvenWhenEnabled() throws IOException { + dao.queryRoundsDebuggable("service", null, "conversation", null, 100, true); + + assertEquals(1, emitted.size()); + assertFalse(emitted.get(0).contains(" ON ")); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void eachReadQueriesOnlyTheRequestedStages(final boolean coldStage) throws IOException { + final Duration duration = mock(Duration.class); + when(duration.isColdStage()).thenReturn(coldStage); + when(duration.getStartTimestamp()).thenReturn(1000L); + when(duration.getEndTimestamp()).thenReturn(2000L); + + dao.queryRoundsDebuggable("service", null, null, duration, 100, false); + dao.queryHeadRoundDebuggable("service", null, "conversation", coldStage); + dao.queryRoundsByNumberDebuggable("service", null, "conversation", 1, 5, 1024, coldStage); + dao.queryFilesDebuggable("service", null, "session", 1000, 2000, 1, 5, 1024, coldStage); + + assertEquals(4, emitted.size(), "each read must issue exactly one storage query"); + for (final String query : emitted) { + assertTrue(query.contains("recordsAIAgent"), query); + assertEquals(coldStage, query.contains(" ON cold STAGES"), query); + if (!coldStage) { + assertFalse(query.contains(" ON "), query); + } + } + } +} diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AIAgentConversationQueryEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AIAgentConversationQueryEsDAO.java index 2d61ebeabc6e..2779244639f5 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AIAgentConversationQueryEsDAO.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AIAgentConversationQueryEsDAO.java @@ -124,7 +124,7 @@ public List queryRounds(final String serviceId, @Override public long queryHeadRound(final String serviceId, @Nullable final String serviceInstanceId, - final String conversation) throws IOException { + final String conversation, final boolean coldStage) throws IOException { final BoolQueryBuilder query = Query.bool(); if (IndexController.LogicIndicesRegister.isMergedTable(AIAgentSessionFlowRecord.INDEX_NAME)) { query.must(Query.term( @@ -158,7 +158,8 @@ public List queryRoundsByNumber(final String serviceId final String conversation, final long fromRound, final long throughRound, - final int maxResponseBytes) throws IOException { + final int maxResponseBytes, + final boolean coldStage) throws IOException { final BoolQueryBuilder query = Query.bool(); if (IndexController.LogicIndicesRegister.isMergedTable(AIAgentSessionFlowRecord.INDEX_NAME)) { query.must(Query.term( @@ -226,7 +227,8 @@ public List queryFiles(final String serviceId, final long toTimestamp, final long fromSeq, final long throughSeq, - final int maxResponseBytes) throws IOException { + final int maxResponseBytes, + final boolean coldStage) throws IOException { final BoolQueryBuilder query = Query.bool(); if (IndexController.LogicIndicesRegister.isMergedTable(AIAgentSessionDataRecord.INDEX_NAME)) { query.must(Query.term( diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCAIAgentConversationQueryDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCAIAgentConversationQueryDAO.java index f489f12c4e3c..1d6232bde612 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCAIAgentConversationQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCAIAgentConversationQueryDAO.java @@ -142,7 +142,7 @@ public List queryRounds(final String serviceId, @Override @SneakyThrows public long queryHeadRound(final String serviceId, @Nullable final String serviceInstanceId, - final String conversation) { + final String conversation, final boolean coldStage) { long head = 0; for (final String table : tableHelper.getTablesWithinTTL(AIAgentSessionFlowRecord.INDEX_NAME)) { final StringBuilder sql = new StringBuilder("select "); @@ -179,7 +179,8 @@ public List queryRoundsByNumber(final String serviceId final String conversation, final long fromRound, final long throughRound, - final int maxResponseBytes) { + final int maxResponseBytes, + final boolean coldStage) { final List columns = new ArrayList<>(ROUND_COLUMNS); columns.add(AIAgentSessionFlowRecord.BODY); final List rounds = new ArrayList<>(); @@ -225,7 +226,8 @@ public List queryFiles(final String serviceId, final long toTimestamp, final long fromSeq, final long throughSeq, - final int maxResponseBytes) { + final int maxResponseBytes, + final boolean coldStage) { final List tables = tableHelper.getTablesForRead( AIAgentSessionDataRecord.INDEX_NAME, TimeBucket.getTimeBucket(fromTimestamp, DownSampling.Day),