mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-04 08:09:27 +00:00
fix: fetch annotations in parallel with queries (#8736)
### **PR Type**
Bug fix, Enhancement
___
### **Description**
- Fetch annotations in parallel with queries
- Update watcher to re-render on annotations
- Preserve error handling and tracing
- Cache saving remains asynchronous
___
### Diagram Walkthrough
```mermaid
flowchart LR
Q["Build and run queries"] -- "Promise.all" --> QR["Query results"]
A["Start annotations fetch"] -- "in parallel" --> AP["Annotations Promise"]
QR -- "update state.data/metadata" --> S["Update state"]
AP -- "await & assign" --> AN["state.annotations"]
AN -- "watcher triggers" --> R["Panel re-render"]
```
<details> <summary><h3> File Walkthrough</h3></summary>
<table><thead><tr><th></th><th align="left">Relevant
files</th></tr></thead><tbody><tr><td><strong>Enhancement</strong></td><td><table>
<tr>
<td>
<details>
<summary><strong>usePanelDataLoader.ts</strong><dd><code>Parallelize
annotations fetching across query paths</code>
</dd></summary>
<hr>
web/src/composables/dashboard/usePanelDataLoader.ts
<ul><li>Start annotations fetch in parallel with PromQL queries.<br>
<li> Start annotations fetch in parallel with SQL/search paths.<br> <li>
Await annotations after primary data ready; update state.<br> <li> Add
try/catch guards; keep tracing and caching behavior.</ul>
</details>
</td>
<td><a
href="https://github.com/openobserve/openobserve/pull/8736/files#diff-5a42e9493a7bcd722b88e61fd38883c4f61001bc9cafa399f65f8eafda3aace9">+188/-115</a></td>
</tr>
<tr>
<td>
<details>
<summary><strong>PanelSchemaRenderer.vue</strong><dd><code>Re-render
panel when annotations update</code>
</dd></summary>
<hr>
web/src/components/dashboards/PanelSchemaRenderer.vue
<ul><li>Add annotations to watch dependencies for re-render.<br> <li>
Ensure UI updates when annotations arrive.</ul>
</details>
</td>
<td><a
href="https://github.com/openobserve/openobserve/pull/8736/files#diff-e5376e7e77bf3aaf6590c2522d8c2dbffcc95a8006ae0d6dc0040a37825367d6">+1/-1</a>
</td>
</tr>
</table></td></tr></tr></tbody></table>
</details>
___
This commit is contained in:
@@ -767,7 +767,7 @@ export default defineComponent({
|
||||
);
|
||||
|
||||
watch(
|
||||
[data, store?.state],
|
||||
[data, store?.state, annotations],
|
||||
async () => {
|
||||
// emit vrl function field list
|
||||
if (data.value?.length && data.value[0] && data.value[0].length) {
|
||||
|
||||
@@ -1252,81 +1252,106 @@ export const usePanelDataLoader = (
|
||||
|
||||
// Check if the query type is "promql"
|
||||
if (panelSchema.value.queryType == "promql") {
|
||||
// Iterate through each query in the panel schema
|
||||
const queryPromises = panelSchema.value.queries?.map(
|
||||
async (it: any) => {
|
||||
const { query: query1, metadata: metadata1 } = replaceQueryValue(
|
||||
it.query,
|
||||
startISOTimestamp,
|
||||
endISOTimestamp,
|
||||
panelSchema.value.queryType,
|
||||
);
|
||||
|
||||
const { query: query2, metadata: metadata2 } =
|
||||
await applyDynamicVariables(query1, panelSchema.value.queryType);
|
||||
|
||||
const query = query2;
|
||||
const metadata = {
|
||||
originalQuery: it.query,
|
||||
query: query,
|
||||
startTime: startISOTimestamp,
|
||||
endTime: endISOTimestamp,
|
||||
queryType: panelSchema.value.queryType,
|
||||
variables: [...(metadata1 || []), ...(metadata2 || [])],
|
||||
};
|
||||
const { traceparent, traceId } = generateTraceContext();
|
||||
addTraceId(traceId);
|
||||
try {
|
||||
const res = await callWithAbortController(
|
||||
() =>
|
||||
queryService.metrics_query_range({
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: query,
|
||||
start_time: startISOTimestamp,
|
||||
end_time: endISOTimestamp,
|
||||
step: panelSchema.value.config.step_value ?? "0",
|
||||
dashboard_id: dashboardId?.value,
|
||||
dashboard_name: dashboardName?.value,
|
||||
folder_id: folderId?.value,
|
||||
folder_name: folderName?.value,
|
||||
panel_id: panelSchema.value.id,
|
||||
panel_name: panelSchema.value.title,
|
||||
run_id: runId?.value,
|
||||
tab_id: tabId?.value,
|
||||
tab_name: tabName?.value,
|
||||
}),
|
||||
abortController.signal,
|
||||
try {
|
||||
// Iterate through each query in the panel schema
|
||||
const queryPromises = panelSchema.value.queries?.map(
|
||||
async (it: any) => {
|
||||
const { query: query1, metadata: metadata1 } = replaceQueryValue(
|
||||
it.query,
|
||||
startISOTimestamp,
|
||||
endISOTimestamp,
|
||||
panelSchema.value.queryType,
|
||||
);
|
||||
|
||||
state.errorDetail = {
|
||||
message: "",
|
||||
code: "",
|
||||
const { query: query2, metadata: metadata2 } =
|
||||
await applyDynamicVariables(
|
||||
query1,
|
||||
panelSchema.value.queryType,
|
||||
);
|
||||
|
||||
const query = query2;
|
||||
const metadata = {
|
||||
originalQuery: it.query,
|
||||
query: query,
|
||||
startTime: startISOTimestamp,
|
||||
endTime: endISOTimestamp,
|
||||
queryType: panelSchema.value.queryType,
|
||||
variables: [...(metadata1 || []), ...(metadata2 || [])],
|
||||
};
|
||||
return { result: res.data.data, metadata: metadata };
|
||||
} catch (error) {
|
||||
processApiError(error, "promql");
|
||||
return { result: null, metadata: metadata };
|
||||
} finally {
|
||||
removeTraceId(traceId);
|
||||
const { traceparent, traceId } = generateTraceContext();
|
||||
addTraceId(traceId);
|
||||
try {
|
||||
const res = await callWithAbortController(
|
||||
() =>
|
||||
queryService.metrics_query_range({
|
||||
org_identifier:
|
||||
store.state.selectedOrganization.identifier,
|
||||
query: query,
|
||||
start_time: startISOTimestamp,
|
||||
end_time: endISOTimestamp,
|
||||
step: panelSchema.value.config.step_value ?? "0",
|
||||
dashboard_id: dashboardId?.value,
|
||||
dashboard_name: dashboardName?.value,
|
||||
folder_id: folderId?.value,
|
||||
folder_name: folderName?.value,
|
||||
panel_id: panelSchema.value.id,
|
||||
panel_name: panelSchema.value.title,
|
||||
run_id: runId?.value,
|
||||
tab_id: tabId?.value,
|
||||
tab_name: tabName?.value,
|
||||
}),
|
||||
abortController.signal,
|
||||
);
|
||||
|
||||
state.errorDetail = {
|
||||
message: "",
|
||||
code: "",
|
||||
};
|
||||
return { result: res.data.data, metadata: metadata };
|
||||
} catch (error) {
|
||||
processApiError(error, "promql");
|
||||
return { result: null, metadata: metadata };
|
||||
} finally {
|
||||
removeTraceId(traceId);
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
// Start fetching annotations in parallel with queries
|
||||
const annotationsPromise = (async () => {
|
||||
try {
|
||||
if (!shouldFetchAnnotations()) {
|
||||
return [];
|
||||
}
|
||||
const annotationList = await refreshAnnotations(
|
||||
startISOTimestamp,
|
||||
endISOTimestamp,
|
||||
);
|
||||
return annotationList || [];
|
||||
} catch (annotationError) {
|
||||
console.error("Failed to fetch annotations:", annotationError);
|
||||
return [];
|
||||
}
|
||||
},
|
||||
);
|
||||
// get annotations
|
||||
const annotationList = shouldFetchAnnotations()
|
||||
? await refreshAnnotations(startISOTimestamp, endISOTimestamp)
|
||||
: [];
|
||||
})();
|
||||
|
||||
// Wait for all query promises to resolve
|
||||
const queryResults: any = await Promise.all(queryPromises);
|
||||
state.loading = false;
|
||||
state.data = queryResults.map((it: any) => it?.result);
|
||||
state.metadata = {
|
||||
queries: queryResults.map((it: any) => it?.metadata),
|
||||
};
|
||||
state.annotations = annotationList || [];
|
||||
// Wait for all query promises to resolve
|
||||
const queryResults: any = await Promise.all(queryPromises);
|
||||
|
||||
// this is async task, which will be executed in background(await is not required)
|
||||
saveCurrentStateToCache();
|
||||
state.loading = false;
|
||||
state.data = queryResults.map((it: any) => it?.result);
|
||||
state.metadata = {
|
||||
queries: queryResults.map((it: any) => it?.metadata),
|
||||
};
|
||||
|
||||
// Wait for annotations to complete and update state
|
||||
// The watcher in PanelSchemaRenderer will trigger re-render when this updates
|
||||
state.annotations = await annotationsPromise;
|
||||
|
||||
// this is async task, which will be executed in background(await is not required)
|
||||
saveCurrentStateToCache();
|
||||
} catch (error) {
|
||||
state.loading = false;
|
||||
}
|
||||
} else {
|
||||
// copy of current abortController
|
||||
// which is used to check whether the current query has been aborted
|
||||
@@ -1435,6 +1460,26 @@ export const usePanelDataLoader = (
|
||||
}
|
||||
|
||||
try {
|
||||
// Start fetching annotations in parallel with search
|
||||
const annotationsPromise = (async () => {
|
||||
try {
|
||||
if (!shouldFetchAnnotations()) {
|
||||
return [];
|
||||
}
|
||||
const annotationList = await refreshAnnotations(
|
||||
startISOTimestamp,
|
||||
endISOTimestamp,
|
||||
);
|
||||
return annotationList || [];
|
||||
} catch (annotationError) {
|
||||
console.error(
|
||||
"Failed to fetch annotations:",
|
||||
annotationError,
|
||||
);
|
||||
return [];
|
||||
}
|
||||
})();
|
||||
|
||||
// get search queries
|
||||
const searchQueries = timeShiftQueries.map(
|
||||
(it: any) => it.searchRequestObj,
|
||||
@@ -1546,21 +1591,15 @@ export const usePanelDataLoader = (
|
||||
);
|
||||
}
|
||||
|
||||
// get annotations
|
||||
const annotationList = shouldFetchAnnotations()
|
||||
? await refreshAnnotations(
|
||||
startISOTimestamp,
|
||||
endISOTimestamp,
|
||||
)
|
||||
: [];
|
||||
state.annotations = annotationList;
|
||||
|
||||
// need to break the loop, save the cache
|
||||
// this is async task, which will be executed in background(await is not required)
|
||||
saveCurrentStateToCache();
|
||||
} finally {
|
||||
removeTraceId(traceId);
|
||||
}
|
||||
|
||||
// Wait for annotations to complete (started in parallel earlier)
|
||||
state.annotations = await annotationsPromise;
|
||||
} catch (error) {
|
||||
// Process API error for "sql"
|
||||
processApiError(error, "sql");
|
||||
@@ -1612,15 +1651,26 @@ export const usePanelDataLoader = (
|
||||
return;
|
||||
}
|
||||
state.metadata.queries[panelQueryIndex] = metadata;
|
||||
const annotations = shouldFetchAnnotations()
|
||||
? await refreshAnnotations(
|
||||
Number(startISOTimestamp),
|
||||
Number(endISOTimestamp),
|
||||
)
|
||||
: [];
|
||||
state.annotations = annotations;
|
||||
|
||||
let annotationsPromise: Promise<any> | null = null;
|
||||
|
||||
if (searchResponse?.value?.hits?.length > 0) {
|
||||
// Start fetching annotations in parallel
|
||||
annotationsPromise = (async () => {
|
||||
try {
|
||||
if (!shouldFetchAnnotations()) {
|
||||
return [];
|
||||
}
|
||||
const annotationList = await refreshAnnotations(
|
||||
Number(startISOTimestamp),
|
||||
Number(endISOTimestamp),
|
||||
);
|
||||
return annotationList || [];
|
||||
} catch (annotationError) {
|
||||
return [];
|
||||
}
|
||||
})();
|
||||
|
||||
// Add empty objects to state.resultMetaData for the results of this query
|
||||
state.data.push([]);
|
||||
state.resultMetaData.push({});
|
||||
@@ -1631,39 +1681,62 @@ export const usePanelDataLoader = (
|
||||
state.resultMetaData[currentQueryIndex] = searchResponse.value;
|
||||
// set loading to false
|
||||
state.loading = false;
|
||||
} else {
|
||||
// Start fetching annotations in parallel for ALL query types
|
||||
annotationsPromise = (async () => {
|
||||
try {
|
||||
if (!shouldFetchAnnotations()) {
|
||||
return [];
|
||||
}
|
||||
const annotationList = await refreshAnnotations(
|
||||
Number(startISOTimestamp),
|
||||
Number(endISOTimestamp),
|
||||
);
|
||||
return annotationList || [];
|
||||
} catch (annotationError) {
|
||||
console.error(
|
||||
"Failed to fetch annotations:",
|
||||
annotationError,
|
||||
);
|
||||
return [];
|
||||
}
|
||||
})();
|
||||
|
||||
return;
|
||||
if (isStreamingEnabled(store.state)) {
|
||||
await getDataThroughStreaming(
|
||||
query,
|
||||
it,
|
||||
startISOTimestamp,
|
||||
endISOTimestamp,
|
||||
pageType,
|
||||
panelQueryIndex,
|
||||
abortControllerRef,
|
||||
);
|
||||
} else if (isWebSocketEnabled(store.state)) {
|
||||
await getDataThroughWebSocket(
|
||||
query,
|
||||
it,
|
||||
startISOTimestamp,
|
||||
endISOTimestamp,
|
||||
pageType,
|
||||
panelQueryIndex,
|
||||
);
|
||||
} else {
|
||||
await getDataThroughPartitions(
|
||||
query,
|
||||
metadata,
|
||||
it,
|
||||
startISOTimestamp,
|
||||
endISOTimestamp,
|
||||
pageType,
|
||||
abortControllerRef,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if (isStreamingEnabled(store.state)) {
|
||||
await getDataThroughStreaming(
|
||||
query,
|
||||
it,
|
||||
startISOTimestamp,
|
||||
endISOTimestamp,
|
||||
pageType,
|
||||
panelQueryIndex,
|
||||
abortControllerRef,
|
||||
);
|
||||
} else if (isWebSocketEnabled(store.state)) {
|
||||
await getDataThroughWebSocket(
|
||||
query,
|
||||
it,
|
||||
startISOTimestamp,
|
||||
endISOTimestamp,
|
||||
pageType,
|
||||
panelQueryIndex,
|
||||
);
|
||||
} else {
|
||||
await getDataThroughPartitions(
|
||||
query,
|
||||
metadata,
|
||||
it,
|
||||
startISOTimestamp,
|
||||
endISOTimestamp,
|
||||
pageType,
|
||||
abortControllerRef,
|
||||
);
|
||||
// Wait for annotations if they were started
|
||||
if (annotationsPromise) {
|
||||
state.annotations = await annotationsPromise;
|
||||
}
|
||||
|
||||
// this is async task, which will be executed in background(await is not required)
|
||||
|
||||
Reference in New Issue
Block a user