Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(shardSplitting): improve error handling #873

Merged
merged 3 commits into from
Nov 5, 2024
Merged

feat(shardSplitting): improve error handling #873

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
65cadb2
feat(shardSplitting): improve error handling
matyax Nov 5, 2024
e3ba150
chore: spelling
matyax Nov 5, 2024
2305946
fix(shardQuerySplitting): stop on parse errors
matyax Nov 5, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
feat(shardSplitting): improve error handling
  • Loading branch information
@matyax
matyax committed Nov 5, 2024
commit 65cadb2118b19e4e2c60bcbff5f0c7db07079015
14 changes: 13 additions & 1 deletion src/services/shardQuerySplitting.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ const originalWarn = console.warn;
beforeAll(() => {
jest.spyOn(console, 'log').mockImplementation(() => {});
jest.spyOn(console, 'warn').mockImplementation(() => {});
jest.spyOn(console, 'error').mockImplementation(() => {});
});
afterAll(() => {
console.log = originalLog;
Expand Down Expand Up @@ -156,7 +157,7 @@ describe('runShardSplitQuery()', () => {
jest
// @ts-expect-error
.spyOn(datasource, 'runQuery')
.mockReturnValueOnce(of({ state: LoadingState.Error, error: { refId: 'A', message: 'Error' }, data: [] }));
.mockReturnValueOnce(of({ state: LoadingState.Error, error: { refId: 'A', message: 'timeout' }, data: [] }));
// @ts-expect-error
jest.spyOn(global, 'setTimeout').mockImplementationOnce((callback) => {
callback();
Expand All @@ -169,6 +170,17 @@ describe('runShardSplitQuery()', () => {
});
});

test('Failed requests have loading state Error', async () => {
jest.mocked(datasource.languageProvider.fetchLabelValues).mockResolvedValue(['1']);
jest
// @ts-expect-error
.spyOn(datasource, 'runQuery')
.mockReturnValue(of({ state: LoadingState.Error, error: { refId: 'A', message: 'max entries' }, data: [] }));
await expect(runShardSplitQuery(datasource, request)).toEmitValuesWith((response: DataQueryResponse[]) => {
expect(response[0].state).toBe(LoadingState.Error);
});
});

test('Adjusts the group size based on errors and execution time', async () => {
const request = createRequest([{ expr: 'count_over_time($SELECTOR[1m])', refId: 'A' }], {
range: {
Expand Down
42 changes: 30 additions & 12 deletions src/services/shardQuerySplitting.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,17 +81,17 @@
subquerySubscription = null;
}

if (shouldStop) {
subscriber.complete();
return;
}

const done = () => {
mergedResponse.state = LoadingState.Done;
mergedResponse.state = shouldStop ? LoadingState.Error : LoadingState.Done;
subscriber.next(mergedResponse);
subscriber.complete();
};

if (shouldStop) {
done();
return;
}

const nextRequest = () => {
const nextCycle = Math.min(cycle + groupSize, shards.length);
if (cycle < shards.length && nextCycle <= shards.length) {
Expand All @@ -102,11 +102,12 @@
};

const retry = (errorResponse?: DataQueryResponse) => {
if (errorResponse?.errors && errorResponse.errors[0].message?.includes('maximum of series')) {
logger.info(`Maximum series reached, skipping retry`);
return false;
} else if (errorResponse?.errors && errorResponse.errors[0].message?.includes('parse error')) {
logger.info(`Parse error, skipping retry`);
try {
if (errorResponse && !isRetriableError(errorResponse)) {
return false;
}
} catch (e) {
logger.error(e);
shouldStop = true;
return false;
}
Expand Down Expand Up @@ -149,7 +150,7 @@
// @ts-expect-error
subquerySubscription = datasource.runQuery(subRequest).subscribe({
next: (partialResponse: DataQueryResponse) => {
if ((partialResponse.errors ?? []).length > 0 || partialResponse.error != null) {

Check warning on line 153 in src/services/shardQuerySplitting.ts

View workflow job for this annotation

GitHub Actions / build 

'error' is deprecated. use errors instead -- will be removed in Grafana 10+
if (retry(partialResponse)) {
return;
}
Expand Down Expand Up @@ -204,7 +205,7 @@
const selector = getSelectorForShardValues(splittingTargets[0].expr);

if (!isValidQuery(selector)) {
console.log(`Skipping invalid selector: ${selector}`);
debug(`Skipping invalid selector: ${selector}`);
subscriber.complete();
return;
}
Expand Down Expand Up @@ -296,6 +297,23 @@
return Math.floor(Math.sqrt(shards.length));
}

function isRetriableError(errorResponse: DataQueryResponse) {
const message = errorResponse.errors
? (errorResponse.errors[0].message ?? '').toLowerCase()
: errorResponse.error?.message ?? '';

Check warning on line 303 in src/services/shardQuerySplitting.ts

View workflow job for this annotation

GitHub Actions / build 

'error' is deprecated. use errors instead -- will be removed in Grafana 10+
if (message.includes('timeout')) {
return true;
} else if (
message.includes('parse error') ||
message.includes('max entries') ||
message.includes('maximum of series')
) {
// If the error is a parse error, we want to signal to stop querying.
throw new Error(message);
}
return false;
}

// Enable to output debugging logs
const DEBUG_ENABLED = Boolean(localStorage.getItem(`${pluginJson.id}.sharding_debug_enabled`));
function debug(message: string) {
Expand Down
Loading