Merge pull request #84 from quickwit-oss/ddelemeny/split-datasource-process-response

Extract datasource query processing

Author: ddelemeny

src/LogContext/components/LogContextUI.tsx

@@ -8,8 +8,8 @@ import { Button } from "@grafana/ui";
 import { useQueryBuilder } from '@/QueryBuilder/lucene';
 import { LogContextQueryBuilderSidebar } from "./LogContextQueryBuilderSidebar";
 import { DatasourceContext } from "components/QueryEditor/ElasticsearchQueryContext";
-import { QuickwitDataSource } from "datasource";
-import { useDatasourceFields } from "datasource.utils";
+import { QuickwitDataSource } from "@/datasource";
+import { useDatasourceFields } from "@/datasource/utils";
 
 const logContextUiStyle = css`
   display: flex;

src/components/QueryEditor/index.tsx

@@ -22,7 +22,7 @@ import { QueryTypeSelector } from './QueryTypeSelector';
 
 import { getHook } from 'utils/context';
 import { LuceneQueryEditor } from '@/components/LuceneQueryEditor';
-import { useDatasourceFields } from 'datasource.utils';
+import { useDatasourceFields } from '@/datasource/utils';
 
 export type ElasticQueryEditorProps = QueryEditorProps<ElasticDatasource, ElasticsearchQuery, QuickwitOptions>;
 

src/datasource.ts renamed to src/datasource/index.ts

@@ -7,7 +7,6 @@ import {
   AdHocVariableFilter,
   CoreApp,
   DataFrame,
-  DataLink,
   DataQueryRequest,
   DataQueryResponse,
   DataSourceApi,
@@ -30,12 +29,11 @@ import {
   SupplementaryQueryType,
   TimeRange,
 } from '@grafana/data';
-import { BucketAggregation, DataLinkConfig, ElasticsearchQuery, TermsQuery, FieldCapabilitiesResponse } from './types';
+import { BucketAggregation, DataLinkConfig, ElasticsearchQuery, TermsQuery, FieldCapabilitiesResponse } from '@/types';
 import { 
   DataSourceWithBackend, 
   getTemplateSrv, 
-  TemplateSrv,
-  getDataSourceSrv } from '@grafana/runtime';
+  TemplateSrv } from '@grafana/runtime';
 import { QuickwitOptions } from 'quickwit';
 import { getDataQuery } from 'QueryBuilder/elastic';
 import { colors } from '@grafana/ui';
@@ -49,7 +47,8 @@ import ElasticsearchLanguageProvider from 'LanguageProvider';
 import { ReactNode } from 'react';
 import { fieldTypeMap } from 'utils';
 import { addAddHocFilter } from 'modifyQuery';
-import { LogContextProvider, LogRowContextOptions } from './LogContext/LogContextProvider';
+import { LogContextProvider, LogRowContextOptions } from '@/LogContext/LogContextProvider';
+import { getQueryResponseProcessor } from 'datasource/processResponse';
 
 export const REF_ID_STARTER_LOG_VOLUME = 'log-volume-';
 
@@ -94,16 +93,8 @@ export class QuickwitDataSource
   }
 
   query(request: DataQueryRequest<ElasticsearchQuery>): Observable<DataQueryResponse> {
-     return super.query(request)
-       .pipe(map((response) => {
-          response.data.forEach((dataFrame) => {
-            const metrics = request.targets[0]!.metrics
-            if (metrics && metrics[0].type === 'logs'){
-              enhanceDataFrameWithDataLinks(dataFrame, this.dataLinks, this.logMessageField);
-            }
-          });
-         return response;
-       }));
+    const queryProcessor = getQueryResponseProcessor(this, request)
+     return super.query(request) .pipe(map(queryProcessor.processResponse));
   }
 
     /**
@@ -740,80 +731,3 @@ function luceneEscape(value: string) {
 
   return value.replace(/([\!\*\+\-\=<>\s\&\|\(\)\[\]\{\}\^\~\?\:\\/"])/g, '\\$1');
 }
-
-export function enhanceDataFrameWithDataLinks(dataFrame: DataFrame, dataLinks: DataLinkConfig[], logMessageField: string | undefined) {
-  // Ignore log volume dataframe, no need to add links or a displayed message field.
-  if (!dataFrame.refId || dataFrame.refId.startsWith('log-volume')) {
-    return;
-  }
-  if (logMessageField) {
-    const messageFields = logMessageField.split(',');
-    let field_idx_list = [];
-    for (const messageField of messageFields) {
-      const field_idx = dataFrame.fields.findIndex((field) => field.name === messageField);
-      if (field_idx !== -1) {
-        field_idx_list.push(field_idx);
-      }
-    }
-    const displayedMessages = Array(dataFrame.length);
-    for (let idx = 0; idx < dataFrame.length; idx++) {
-      let displayedMessage = "";
-      // If we have only one field, we assume the field name is obvious for the user and we don't need to show it.
-      if (field_idx_list.length === 1) {
-        displayedMessage = `${dataFrame.fields[field_idx_list[0]].values[idx]}`;
-      } else {
-        for (const field_idx of field_idx_list) {
-          displayedMessage += ` ${dataFrame.fields[field_idx].name}=${dataFrame.fields[field_idx].values[idx]}`;
-        }
-      }
-      displayedMessages[idx] = displayedMessage.trim();
-    }
-
-    const newField = {
-        name: 'message',
-        type: FieldType.string,
-        config: {},
-        values: displayedMessages,
-    }
-    const [timestamp, ...rest] = dataFrame.fields;
-    dataFrame.fields = [timestamp, newField, ...rest];
-  }
-  
-  if (!dataLinks.length) {
-    return;
-  }
-
-  for (const field of dataFrame.fields) {
-    const linksToApply = dataLinks.filter((dataLink) => dataLink.field === field.name);
-
-    if (linksToApply.length === 0) {
-      continue;
-    }
-
-    field.config = field.config || {};
-    field.config.links = [...(field.config.links || [], linksToApply.map(generateDataLink))];
-  }
-}
-
-function generateDataLink(linkConfig: DataLinkConfig): DataLink {
-  const dataSourceSrv = getDataSourceSrv();
-
-  if (linkConfig.datasourceUid) {
-    const dsSettings = dataSourceSrv.getInstanceSettings(linkConfig.datasourceUid);
-
-    return {
-      title: linkConfig.urlDisplayLabel || '',
-      url: '',
-      internal: {
-        query: { query: linkConfig.url },
-        datasourceUid: linkConfig.datasourceUid,
-        datasourceName: dsSettings?.name ?? 'Data source not found',
-      },
-    };
-  } else {
-    return {
-      title: linkConfig.urlDisplayLabel || '',
-      url: linkConfig.url,
-    };
-  }
-}

src/datasource/processResponse.ts

@@ -0,0 +1,94 @@
+import { DataFrame, DataLink, DataQueryRequest, DataQueryResponse, FieldType } from "@grafana/data";
+import { getDataSourceSrv } from "@grafana/runtime";
+import { QuickwitDataSource } from 'datasource';
+import { DataLinkConfig, ElasticsearchQuery } from "../types";
+
+export function getQueryResponseProcessor(datasource: QuickwitDataSource, request: DataQueryRequest<ElasticsearchQuery>) {
+  return {
+    processResponse: (response: DataQueryResponse) => {
+      response.data.forEach((dataFrame) => {
+        const metrics = request.targets[0].metrics;
+        if (metrics && metrics[0].type === 'logs') {
+          processLogsDataFrame(datasource, dataFrame);
+        }
+      });
+      return response;
+    }
+  };
+}
+function getCustomFieldName(fieldname: string) { return `$qw_${fieldname}`; }
+export function processLogsDataFrame(datasource: QuickwitDataSource, dataFrame: DataFrame) {
+  // Ignore log volume dataframe, no need to add links or a displayed message field.
+  if (!dataFrame.refId || dataFrame.refId.startsWith('log-volume')) {
+    return;
+  }
+  if (datasource.logMessageField) {
+    const messageFields = datasource.logMessageField.split(',');
+    let field_idx_list = [];
+    for (const messageField of messageFields) {
+      const field_idx = dataFrame.fields.findIndex((field) => field.name === messageField);
+      if (field_idx !== -1) {
+        field_idx_list.push(field_idx);
+      }
+    }
+    const displayedMessages = Array(dataFrame.length);
+    for (let idx = 0; idx < dataFrame.length; idx++) {
+      let displayedMessage = "";
+      // If we have only one field, we assume the field name is obvious for the user and we don't need to show it.
+      if (field_idx_list.length === 1) {
+        displayedMessage = `${dataFrame.fields[field_idx_list[0]].values[idx]}`;
+      } else {
+        for (const field_idx of field_idx_list) {
+          displayedMessage += ` ${dataFrame.fields[field_idx].name}=${dataFrame.fields[field_idx].values[idx]}`;
+        }
+      }
+      displayedMessages[idx] = displayedMessage.trim();
+    }
+
+    const newField = {
+      name: getCustomFieldName('message'),
+      type: FieldType.string,
+      config: {},
+      values: displayedMessages,
+    };
+    const [timestamp, ...rest] = dataFrame.fields;
+    dataFrame.fields = [timestamp, newField, ...rest];
+  }
+
+  if (!datasource.dataLinks.length) {
+    return;
+  }
+
+  for (const field of dataFrame.fields) {
+    const linksToApply = datasource.dataLinks.filter((dataLink) => dataLink.field === field.name);
+
+    if (linksToApply.length === 0) {
+      continue;
+    }
+
+    field.config = field.config || {};
+    field.config.links = [...(field.config.links || [], linksToApply.map(generateDataLink))];
+  }
+}
+function generateDataLink(linkConfig: DataLinkConfig): DataLink {
+  const dataSourceSrv = getDataSourceSrv();
+
+  if (linkConfig.datasourceUid) {
+    const dsSettings = dataSourceSrv.getInstanceSettings(linkConfig.datasourceUid);
+
+    return {
+      title: linkConfig.urlDisplayLabel || '',
+      url: '',
+      internal: {
+        query: { query: linkConfig.url },
+        datasourceUid: linkConfig.datasourceUid,
+        datasourceName: dsSettings?.name ?? 'Data source not found',
+      },
+    };
+  } else {
+    return {
+      title: linkConfig.urlDisplayLabel || '',
+      url: linkConfig.url,
+    };
+  }
+}

src/datasource.utils.ts renamed to src/datasource/utils.ts

@@ -1,4 +1,4 @@
-import { QuickwitDataSource } from "datasource";
+import { QuickwitDataSource } from "@/datasource";
 import { useState, useEffect, useCallback } from "react";
 import{ MetricFindValue } from '@grafana/data';