This is an automated email from the ASF dual-hosted git repository.
vogievetsky pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 34049ff8389 fix: better query manager and allow setting cookie for dev
forward (#19692)
34049ff8389 is described below
commit 34049ff83891ae93ca85090b369186347381d492
Author: Vadim Ogievetsky <[email protected]>
AuthorDate: Mon Jul 20 11:49:53 2026 -0700
fix: better query manager and allow setting cookie for dev forward (#19692)
* better query manager and allow setting cookie for dev forward
* feedback fixes
---
.../compaction-status/compaction-status.ts | 2 +-
.../supervisor-status/supervisor-status.ts | 12 +
.../src/utils/query-manager/query-manager.spec.ts | 370 +++++++++++++++++++++
.../src/utils/query-manager/query-manager.ts | 22 +-
.../views/supervisors-view/supervisors-view.tsx | 19 +-
web-console/webpack.config.mjs | 13 +-
6 files changed, 426 insertions(+), 12 deletions(-)
diff --git
a/web-console/src/druid-models/compaction-status/compaction-status.ts
b/web-console/src/druid-models/compaction-status/compaction-status.ts
index 494918182d4..5712eb958b0 100644
--- a/web-console/src/druid-models/compaction-status/compaction-status.ts
+++ b/web-console/src/druid-models/compaction-status/compaction-status.ts
@@ -77,7 +77,7 @@ export function formatCompactionInfo(compaction:
CompactionInfo) {
return 'Fully compacted';
}
} else {
- return capitalizeFirst(status.scheduleStatus);
+ return capitalizeFirst(status.scheduleStatus.replace(/_/g, ' '));
}
} else {
return 'Awaiting first run';
diff --git
a/web-console/src/druid-models/supervisor-status/supervisor-status.ts
b/web-console/src/druid-models/supervisor-status/supervisor-status.ts
index 3004cf350e7..feb42a0a7f7 100644
--- a/web-console/src/druid-models/supervisor-status/supervisor-status.ts
+++ b/web-console/src/druid-models/supervisor-status/supervisor-status.ts
@@ -43,6 +43,18 @@ export interface SupervisorStatus {
state: string;
detailedState: string;
recentErrors: SupervisorError[];
+
+ // autocompact
+ scheduleStatus: string;
+ bytesAwaitingCompaction: number;
+ bytesCompacted: number;
+ bytesSkipped: number;
+ segmentCountAwaitingCompaction: number;
+ segmentCountCompacted: number;
+ segmentCountSkipped: number;
+ intervalCountAwaitingCompaction: number;
+ intervalCountCompacted: number;
+ intervalCountSkipped: number;
};
}
diff --git a/web-console/src/utils/query-manager/query-manager.spec.ts
b/web-console/src/utils/query-manager/query-manager.spec.ts
new file mode 100644
index 00000000000..05568a9c1a0
--- /dev/null
+++ b/web-console/src/utils/query-manager/query-manager.spec.ts
@@ -0,0 +1,370 @@
+/*
+ * 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.
+ */
+
+import { IntermediateQueryState } from './intermediate-query-state';
+import type { QueryManagerOptions } from './query-manager';
+import { QueryManager } from './query-manager';
+import { QueryState } from './query-state';
+import { ResultWithAuxiliaryWork } from './result-with-auxiliary-work';
+
+// jsdom's AbortSignal (used as the global in the jest environment) predates
`throwIfAborted`,
+// which the QueryManager relies on. Polyfill it so the tests exercise real
browser behavior.
+beforeAll(() => {
+ if (typeof AbortSignal.prototype.throwIfAborted !== 'function') {
+ AbortSignal.prototype.throwIfAborted = function throwIfAborted(this:
AbortSignal) {
+ if (this.aborted) throw (this as any).reason ?? new Error('Aborted');
+ };
+ }
+});
+
+// All debounces and background delays are set to 0 so that `run()` executes
synchronously
+// (the constructor wires the runWhen* helpers directly to `run` when the
debounce is 0, and
+// every `wait(delay)` is guarded by `if (delay)`). That makes the manager
fully driven by the
+// microtask queue, which we flush with `Promise.resolve()`.
+const ZERO_TIMING = {
+ debounceInit: 0,
+ debounceIdle: 0,
+ debounceLoading: 0,
+ backgroundStatusCheckInitDelay: 0,
+ backgroundStatusCheckDelay: 0,
+} as const;
+
+// Promise-returning helpers keep the test callbacks free of unnecessary
`async` (which the
+// require-await lint rule rejects) while still satisfying the
Promise-returning signatures.
+const resolved = <T>(value: T): Promise<T> => Promise.resolve(value);
+const rejected = (error: Error): Promise<never> => Promise.reject(error);
+
+interface Harness<Q, R, I = never, E extends Error = Error> {
+ manager: QueryManager<Q, R, I, E>;
+ states: QueryState<R, E, I>[];
+}
+
+function makeManager<Q, R, I = never, E extends Error = Error>(
+ options: Omit<QueryManagerOptions<Q, R, I, E>, 'onStateChange'> &
+ Partial<Pick<QueryManagerOptions<Q, R, I, E>, 'onStateChange'>>,
+): Harness<Q, R, I, E> {
+ const states: QueryState<R, E, I>[] = [];
+ const manager = new QueryManager<Q, R, I, E>({
+ ...ZERO_TIMING,
+ ...options,
+ onStateChange: state => states.push(state),
+ });
+ return { manager, states };
+}
+
+// Flushes microtasks until the latest recorded state past `fromIndex` is
settled (data without
+// auxiliaryLoading, or an error). Returns that final state.
+async function settleAfter<R, E extends Error, I>(
+ states: QueryState<R, E, I>[],
+ fromIndex: number,
+): Promise<QueryState<R, E, I>> {
+ for (let i = 0; i < 500; i++) {
+ await Promise.resolve();
+ const last = states[states.length - 1];
+ if (
+ states.length > fromIndex &&
+ last &&
+ (last.isError() || (last.state === 'data' && !last.auxiliaryLoading))
+ ) {
+ return last;
+ }
+ }
+ throw new Error('query manager did not settle');
+}
+
+describe('QueryManager', () => {
+ describe('basic query lifecycle', () => {
+ it('transitions from loading to data', async () => {
+ const { manager, states } = makeManager<string, string>({
+ processQuery: query => resolved(`result:${query}`),
+ });
+
+ manager.runQuery('hello');
+ const final = await settleAfter(states, 0);
+
+ expect(states[0].isLoading()).toEqual(true);
+ expect(final.state).toEqual('data');
+ expect(final.data).toEqual('result:hello');
+ expect(manager.getState().data).toEqual('result:hello');
+ expect(manager.getLastQuery()).toEqual('hello');
+ });
+
+ it('passes the query through to processQuery', async () => {
+ const seen: string[] = [];
+ const { manager, states } = makeManager<string, number>({
+ processQuery: query => {
+ seen.push(query);
+ return resolved(query.length);
+ },
+ });
+
+ manager.runQuery('abcd');
+ await settleAfter(states, 0);
+
+ expect(seen).toEqual(['abcd']);
+ expect(manager.getState().data).toEqual(4);
+ });
+
+ it('captures errors thrown by processQuery into an error state', async ()
=> {
+ const { manager, states } = makeManager<string, string>({
+ processQuery: () => rejected(new Error('boom')),
+ });
+
+ manager.runQuery('x');
+ const final = await settleAfter(states, 0);
+
+ expect(final.isError()).toEqual(true);
+ expect(final.getErrorMessage()).toEqual('boom');
+ });
+
+ it('keeps prior data as lastData on a subsequent run', async () => {
+ const { manager, states } = makeManager<string, string>({
+ processQuery: query => resolved(`result:${query}`),
+ });
+
+ manager.runQuery('one');
+ await settleAfter(states, 0);
+
+ const start = states.length;
+ manager.runQuery('two');
+ // While loading the second query, the previous data is exposed as
lastData.
+ expect(states[start].isLoading()).toEqual(true);
+ expect(states[start].lastData).toEqual('result:one');
+
+ const final = await settleAfter(states, start);
+ expect(final.data).toEqual('result:two');
+ });
+ });
+
+ describe('query superseding', () => {
+ it('keeps only the newest query result when a second query is issued
mid-flight', async () => {
+ let resolveFirst: ((value: string) => void) | undefined;
+ const { manager, states } = makeManager<string, string>({
+ processQuery: query =>
+ query === 'first'
+ ? new Promise<string>(resolve => {
+ resolveFirst = resolve;
+ })
+ : Promise.resolve(`result:${query}`),
+ });
+
+ manager.runQuery('first');
+ manager.runQuery('second');
+
+ const final = await settleAfter(states, 0);
+ expect(final.data).toEqual('result:second');
+
+ // Even if the stale first query resolves later, it must not overwrite
the state.
+ resolveFirst?.('result:first');
+ await Promise.resolve();
+ await Promise.resolve();
+ expect(manager.getState().data).toEqual('result:second');
+ });
+ });
+
+ describe('intermediate query state (backgroundStatusCheck)', () => {
+ it('loops through intermediate states until a final result is produced',
async () => {
+ // delay: 0 on the IntermediateQueryState skips the real setTimeout wait
(the
+ // backgroundStatusCheck*Delay defaults can't be zeroed since `0 || 500`
is 500).
+ const { manager, states } = makeManager<string, string, string>({
+ processQuery: () => resolved(new IntermediateQueryState('pending', 0)),
+ backgroundStatusCheck: state =>
+ resolved(state === 'pending' ? new IntermediateQueryState('almost',
0) : 'done'),
+ });
+
+ manager.runQuery('go');
+ const final = await settleAfter(states, 0);
+
+ const intermediates = states
+ .filter(s => s.isLoading())
+ .map(s => s.intermediate)
+ .filter(Boolean);
+ expect(intermediates).toEqual(['pending', 'almost']);
+ expect(final.data).toEqual('done');
+ });
+
+ it('errors when an intermediate state is returned without a
backgroundStatusCheck', async () => {
+ const { manager, states } = makeManager<string, string, string>({
+ processQuery: () => resolved(new IntermediateQueryState('pending')),
+ });
+
+ manager.runQuery('go');
+ const final = await settleAfter(states, 0);
+
+ expect(final.isError()).toEqual(true);
+ expect(final.getErrorMessage()).toContain('backgroundStatusCheck must be
set');
+ });
+
+ it('swallows recoverable background errors and surfaces them as
intermediateError', async () => {
+ let checks = 0;
+ const { manager, states } = makeManager<string, string, string>({
+ processQuery: () => resolved(new IntermediateQueryState('pending', 0)),
+ backgroundStatusCheck: () => {
+ checks++;
+ return checks === 1 ? rejected(new Error('transient')) :
resolved('done');
+ },
+ swallowBackgroundError: e => e.message === 'transient',
+ });
+
+ manager.runQuery('go');
+ const final = await settleAfter(states, 0);
+
+ const withError = states.find(s => s.intermediateError);
+ expect(withError?.intermediateError?.message).toEqual('transient');
+ expect(final.data).toEqual('done');
+ });
+ });
+
+ describe('ResultWithAuxiliaryWork', () => {
+ interface Enriched {
+ base: number;
+ a?: number;
+ b?: number;
+ }
+
+ const makeAuxProcessQuery = () => ():
Promise<ResultWithAuxiliaryWork<Enriched>> =>
+ resolved(
+ new ResultWithAuxiliaryWork<Enriched>({ base: 1 }, [
+ result => resolved({ ...result, a: 2 }),
+ result => resolved({ ...result, b: 3 }),
+ ]),
+ );
+
+ it('unwraps a result that has no auxiliary queries', async () => {
+ const { manager, states } = makeManager<string, Enriched>({
+ processQuery: () => resolved(new ResultWithAuxiliaryWork<Enriched>({
base: 1 }, [])),
+ });
+
+ manager.runQuery('go');
+ const final = await settleAfter(states, 0);
+
+ expect(final.data).toEqual({ base: 1 });
+ expect(states.some(s => s.auxiliaryLoading)).toEqual(false);
+ });
+
+ it('enriches incrementally on a foreground run', async () => {
+ const { manager, states } = makeManager<string, Enriched>({
+ processQuery: makeAuxProcessQuery(),
+ });
+
+ manager.runQuery('go');
+ const final = await settleAfter(states, 0);
+
+ // The base result and the first auxiliary result are published with
auxiliaryLoading: true.
+ const auxStates = states.filter(s => s.auxiliaryLoading);
+ expect(auxStates.map(s => s.data)).toEqual([{ base: 1 }, { base: 1, a: 2
}]);
+
+ // The final state is fully enriched and no longer auxiliary-loading.
+ expect(final.auxiliaryLoading).toBeFalsy();
+ expect(final.data).toEqual({ base: 1, a: 2, b: 3 });
+ });
+
+ it('does NOT enrich incrementally on a background run', async () => {
+ const { manager, states } = makeManager<string, Enriched>({
+ processQuery: makeAuxProcessQuery(),
+ });
+
+ // Initial foreground run so there is a lastQuery to rerun.
+ manager.runQuery('go');
+ await settleAfter(states, 0);
+
+ const start = states.length;
+ manager.rerunLastQuery(true);
+ const final = await settleAfter(states, start);
+
+ // The background run emits exactly one state: the final, fully
enriched, non-loading result.
+ const backgroundStates = states.slice(start);
+ expect(backgroundStates).toHaveLength(1);
+ expect(backgroundStates[0].auxiliaryLoading).toBeFalsy();
+ expect(final.data).toEqual({ base: 1, a: 2, b: 3 });
+ });
+
+ it('enriches incrementally on a foreground rerun', async () => {
+ const { manager, states } = makeManager<string, Enriched>({
+ processQuery: makeAuxProcessQuery(),
+ });
+
+ manager.runQuery('go');
+ await settleAfter(states, 0);
+
+ const start = states.length;
+ manager.rerunLastQuery(); // foreground (default)
+ await settleAfter(states, start);
+
+ const auxStates = states.slice(start).filter(s => s.auxiliaryLoading);
+ expect(auxStates.map(s => s.data)).toEqual([{ base: 1 }, { base: 1, a: 2
}]);
+ });
+ });
+
+ describe('intermediate query tracking', () => {
+ it('records the last intermediate query set via the extra callback', async
() => {
+ const { manager, states } = makeManager<string, string>({
+ processQuery: (query, _signal, extra) => {
+ extra.setIntermediateQuery(`intermediate:${query}`);
+ return resolved(`result:${query}`);
+ },
+ });
+
+ manager.runQuery('go');
+ await settleAfter(states, 0);
+
+ expect(manager.getLastIntermediateQuery()).toEqual('intermediate:go');
+ });
+ });
+
+ describe('control methods', () => {
+ it('reset returns the manager to the init state', async () => {
+ const { manager, states } = makeManager<string, string>({
+ processQuery: query => resolved(`result:${query}`),
+ });
+
+ manager.runQuery('go');
+ await settleAfter(states, 0);
+
+ manager.reset();
+ expect(manager.getState().isInit()).toEqual(true);
+ });
+
+ it('terminate stops further state changes and ignores new queries', async
() => {
+ const { manager, states } = makeManager<string, string>({
+ processQuery: query => resolved(`result:${query}`),
+ });
+
+ manager.runQuery('go');
+ await settleAfter(states, 0);
+
+ const countBefore = states.length;
+ manager.terminate();
+ expect(manager.isTerminated()).toEqual(true);
+
+ manager.runQuery('again');
+ await Promise.resolve();
+ await Promise.resolve();
+ expect(states.length).toEqual(countBefore);
+ });
+
+ it('initialises from a provided initState', () => {
+ const { manager } = makeManager<string, string>({
+ initState: new QueryState<string>({ data: 'seed' }),
+ processQuery: query => resolved(`result:${query}`),
+ });
+
+ expect(manager.getState().data).toEqual('seed');
+ });
+ });
+});
diff --git a/web-console/src/utils/query-manager/query-manager.ts
b/web-console/src/utils/query-manager/query-manager.ts
index c802b8014ce..fc1dba726d1 100644
--- a/web-console/src/utils/query-manager/query-manager.ts
+++ b/web-console/src/utils/query-manager/query-manager.ts
@@ -78,6 +78,7 @@ export class QueryManager<Q, R, I = never, E extends Error =
Error> {
private readonly swallowBackgroundError?: (e: Error) => boolean;
private terminated = false;
+ private runInBackground = false;
private nextQuery: Q | undefined;
private lastQuery: Q | undefined;
private lastIntermediateQuery: any;
@@ -124,6 +125,7 @@ export class QueryManager<Q, R, I = never, E extends Error
= Error> {
private async run(): Promise<void> {
this.lastQuery = this.nextQuery;
if (typeof this.lastQuery === 'undefined') return;
+ const background = this.runInBackground;
this.currentQueryId++;
const myQueryId = this.currentQueryId;
@@ -266,13 +268,15 @@ export class QueryManager<Q, R, I = never, E extends
Error = Error> {
const numAuxiliaryQueries = auxiliaryQueries.length;
data = data.result;
- this.setState(
- new QueryState<R, E>({
- data,
- auxiliaryLoading: true,
- lastData,
- }),
- );
+ if (!background) {
+ this.setState(
+ new QueryState<R, E>({
+ data,
+ auxiliaryLoading: true,
+ lastData,
+ }),
+ );
+ }
try {
for (let i = 0; i < numAuxiliaryQueries; i++) {
@@ -282,7 +286,7 @@ export class QueryManager<Q, R, I = never, E extends Error
= Error> {
data = await auxiliaryQueries[i](data, signal);
if (this.currentQueryId !== myQueryId) return;
- if (i < numAuxiliaryQueries - 1) {
+ if (!background && i < numAuxiliaryQueries - 1) {
// Update data in intermediate state
this.setState(
new QueryState<R, E>({
@@ -307,6 +311,7 @@ export class QueryManager<Q, R, I = never, E extends Error
= Error> {
}
private trigger() {
+ this.runInBackground = false;
if (this.currentRunCancelFn && !this.state.auxiliaryLoading) {
// Currently loading main query
void this.runWhenLoading();
@@ -337,6 +342,7 @@ export class QueryManager<Q, R, I = never, E extends Error
= Error> {
if (runInBackground && this.currentRunCancelFn) return;
this.nextQuery = this.lastQuery;
if (runInBackground) {
+ this.runInBackground = true;
void this.runWhenIdle();
} else {
this.trigger();
diff --git a/web-console/src/views/supervisors-view/supervisors-view.tsx
b/web-console/src/views/supervisors-view/supervisors-view.tsx
index 31a64f1bf0c..544b1b99f18 100644
--- a/web-console/src/views/supervisors-view/supervisors-view.tsx
+++ b/web-console/src/views/supervisors-view/supervisors-view.tsx
@@ -48,6 +48,7 @@ import {
TaskGroupHandoffDialog,
} from '../../dialogs';
import type {
+ CompactionConfig,
ConsoleViewId,
IngestionSpec,
QueryWithContext,
@@ -56,7 +57,11 @@ import type {
SupervisorStatus,
SupervisorStatusTask,
} from '../../druid-models';
-import { getConsoleViewIcon, getTotalSupervisorStats } from
'../../druid-models';
+import {
+ formatCompactionInfo,
+ getConsoleViewIcon,
+ getTotalSupervisorStats,
+} from '../../druid-models';
import type { Capabilities } from '../../helpers';
import {
SMALL_TABLE_PAGE_SIZE,
@@ -357,7 +362,10 @@ export class SupervisorsView extends React.PureComponent<
return {
supervisor_id: deepGet(sup, 'id'),
datasource: deepGet(sup, 'dataSource'),
- type: deepGet(sup, 'spec.tuningConfig.type'),
+ type:
+ deepGet(sup, 'spec.type') ||
+ deepGet(sup, 'spec.ioConfig.type') ||
+ deepGet(sup, 'spec.tuningConfig.type'),
source:
deepGet(sup, 'spec.ioConfig.topic') ||
deepGet(sup, 'spec.ioConfig.stream') ||
@@ -1060,6 +1068,13 @@ export class SupervisorsView extends React.PureComponent<
aggregateLag,
)}`}</span>
) : null;
+ } else if (original.type === 'autocompact') {
+ const compactionConfig: CompactionConfig | undefined =
original.spec?.spec;
+ if (!supervisorStatusPayload || !compactionConfig) return null;
+ return formatCompactionInfo({
+ status: supervisorStatusPayload,
+ config: compactionConfig,
+ });
} else {
return null;
}
diff --git a/web-console/webpack.config.mjs b/web-console/webpack.config.mjs
index a1108f807bb..17cfb57a64a 100644
--- a/web-console/webpack.config.mjs
+++ b/web-console/webpack.config.mjs
@@ -43,10 +43,13 @@ export default env => {
if (!druidUrl.startsWith('http')) {
druidUrl = (druidUrl.endsWith(':9088') ? 'https://' : 'http://') +
druidUrl;
}
- if (!/:\d+$/.test(druidUrl)) {
+ if (druidUrl.endsWith('localhost')) {
druidUrl += druidUrl.startsWith('https://') ? ':9088' : ':8888';
}
+ // Optional cookie to add to all proxied requests, e.g.
druid_cookie='name=value'
+ const druidCookie = (env || {}).druid_cookie || process.env.druid_cookie;
+
const mode = process.env.NODE_ENV === 'production' ? 'production' :
'development';
// eslint-disable-next-line no-undef
@@ -111,6 +114,14 @@ export default env => {
context: ['/status', '/druid', '/proxy'],
target: druidUrl,
secure: false,
+ changeOrigin: true,
+ onProxyReq: (proxyReq, _req) => {
+ if (druidCookie) {
+ proxyReq.setHeader('Cookie', druidCookie);
+ }
+ // To debug use:
+ // console.log(`[proxy] ${req.method} ${req.url} ->
${proxyReq.path}`);
+ },
},
],
},
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
