feat: updated full time range logs volume histogram (#5)

* fix: added logs model for building logs volume graph

* fix: By default added fields to data frame

* fix: update logs volume dataframe

* fix: updated folder name

* fix: removed console logs

* fix: optimised search api calls

Author: omkarK06

.DS_Store

@@ -52,15 +52,16 @@
     "typescript": "^4.4.0",
     "webpack": "^5.69.1",
     "webpack-cli": "^4.9.2",
-    "webpack-livereload-plugin": "^3.0.2"
+    "webpack-livereload-plugin": "^3.0.2",
+    "rxjs": "7.8.0"
   },
   "engines": {
     "node": ">=16"
   },
   "dependencies": {
     "@emotion/css": "^11.1.3",
-    "@grafana/data": "9.3.8",
-    "@grafana/runtime": "9.3.8",
+    "@grafana/data": "9.5.2",
+    "@grafana/runtime": "9.5.2",
     "@grafana/ui": "9.3.8",
     "monaco-editor": "^0.34.0",
     "react": "17.0.2",

package-lock.json

@@ -15,6 +15,7 @@ export const QueryEditor = ({ query, onChange, onRunQuery, datasource }: Props)
   const [streamDetails, setStreamDetails]: any = useState({});
   const [streamOptions, setStreamOptions]: any = useState([]);
   const [orgOptions, setOrgOptions]: any = useState([]);
+  const [isMounted, setIsMounted]: any = useState(false);
 
   useEffect(() => {
     getOrganizations({ url: datasource.url, page_num: 0, page_size: 1000, sort_by: 'id' })
@@ -26,36 +27,38 @@ export const QueryEditor = ({ query, onChange, onRunQuery, datasource }: Props)
           })),
         ]);
         setupStreams(orgs.data[0].name).then((streams: any) => {
-          onChange({
-            ...query,
-            stream: streams[0].name,
-            organization: orgs.data[0].name,
-            sqlMode: false,
-          });
           datasource.updateStreamFields(streams[0].schema);
           setStreamOptions([
             ...Object.values(streams).map((stream: any) => ({
               label: stream.name,
               value: stream.name,
             })),
           ]);
-          onRunQuery();
+          if (!(query.organization && query.stream && query.hasOwnProperty('sqlMode'))) {
+            onChange({
+              ...query,
+              stream: streams[0].name,
+              organization: orgs.data[0].name,
+              sqlMode: false,
+            });
+          }
+          setIsMounted(true);
         });
       })
       .catch((err) => console.log(err));
     // eslint-disable-next-line react-hooks/exhaustive-deps
   }, []);
 
   useEffect(() => {
-    if (query.sqlMode !== undefined && query.stream) {
+    if (query.sqlMode !== undefined && query.stream && isMounted) {
       updateQuery();
     }
 
     // eslint-disable-next-line react-hooks/exhaustive-deps
   }, [query.sqlMode, query.organization, query.stream]);
 
   useEffect(() => {
-    if (query.stream && query.organization) {
+    if (query.stream && query.organization && isMounted) {
       updateQuery();
       onRunQuery();
     }
@@ -117,7 +120,7 @@ export const QueryEditor = ({ query, onChange, onRunQuery, datasource }: Props)
         stream: streams[0].name,
         organization: organization.value,
       });
-      datasource.updateStreamFields(cloneDeep(streamDetails[streams[0].name].schema));
+      datasource.updateStreamFields(cloneDeep(streams[0].schema));
       setStreamOptions([
         ...streams.map((stream: any) => ({
           label: stream.name,

package.json

@@ -1,6 +1,7 @@
 import { MyDataSourceOptions } from 'types';
 import { DataSource } from './datasource';
 import { DataSourceInstanceSettings, PluginSignatureStatus, PluginType } from '@grafana/data';
+import { buildQuery } from 'features/query/queryBuilder';
 
 let DateTime = {
   add: jest.fn(),
@@ -28,6 +29,12 @@ let DateTime = {
   minute: jest.fn(),
 };
 
+jest.mock('rxjs', () => {
+  return {
+    Observable: jest.fn(),
+  };
+});
+
 jest.mock('@grafana/runtime', () => ({
   ...jest.requireActual('@grafana/runtime'),
   getBackendSrv: () => {
@@ -120,21 +127,6 @@ describe('DataSource', () => {
     ds = new DataSource(instanceSettings);
   });
 
-  describe('getConsumableTime', () => {
-    it('should convert range to microsecond format', () => {
-      const range = {
-        from: new Date('2023-05-16T00:00:00Z'),
-        to: new Date('2023-05-16T01:00:00Z'),
-      };
-      const result = ds.getConsumableTime(range);
-
-      expect(result).toEqual({
-        startTimeInMicro: range.from.getTime() * 1000,
-        endTimeInMirco: range.to.getTime() * 1000,
-      });
-    });
-  });
-
   describe('testDatasource', () => {
     it('should return success status', async () => {
       const result = await ds.testDatasource();
@@ -405,7 +397,7 @@ describe('DataSource', () => {
       encoding: 'base64',
     };
     beforeEach(async () => {
-      result = ds.buildQuery(queryData, timestamps);
+      result = buildQuery(queryData, timestamps, queryData.streamFields);
     });
     it('should return query request data', () => {
       expect(JSON.stringify(result)).toMatch(JSON.stringify(expectedReq));

src/.DS_Store

@@ -3,17 +3,28 @@ import {
   DataQueryResponse,
   DataSourceApi,
   DataSourceInstanceSettings,
-  MutableDataFrame,
-  FieldType,
   QueryFixAction,
+  DataSourceWithSupplementaryQueriesSupport,
+  SupplementaryQueryType,
+  LogLevel,
 } from '@grafana/data';
-
+import { Observable } from 'rxjs';
 import { getBackendSrv } from '@grafana/runtime';
+import { queryLogsVolume } from './features/log/LogsModel';
 
-import { MyQuery, MyDataSourceOptions, TimeRange } from './types';
-import { b64EncodeUnicode, logsErrorMessage } from 'utils/zincutils';
+import { MyQuery, MyDataSourceOptions } from './types';
+import { logsErrorMessage, getConsumableTime } from 'utils/zincutils';
 import { getOrganizations } from 'services/organizations';
-export class DataSource extends DataSourceApi<MyQuery, MyDataSourceOptions> {
+import { cloneDeep } from 'lodash';
+import { getGraphDataFrame, getLogsDataFrame } from 'features/log/queryResponseBuilder';
+import { buildQuery } from './features/query/queryBuilder';
+
+const REF_ID_STARTER_LOG_VOLUME = 'log-volume-';
+
+export class DataSource
+  extends DataSourceApi<MyQuery, MyDataSourceOptions>
+  implements DataSourceWithSupplementaryQueriesSupport<MyQuery>
+{
   instanceSettings?: DataSourceInstanceSettings<MyDataSourceOptions>;
   url: string;
   streamFields: any[];
@@ -25,70 +36,34 @@ export class DataSource extends DataSourceApi<MyQuery, MyDataSourceOptions> {
     this.streamFields = [];
   }
 
-  getConsumableTime(range: any) {
-    const startTimeInMicro: any = new Date(new Date(range!.from.valueOf()).toISOString()).getTime() * 1000;
-    const endTimeInMirco: any = new Date(new Date(range!.to.valueOf()).toISOString()).getTime() * 1000;
-    return {
-      startTimeInMicro,
-      endTimeInMirco,
-    };
-  }
-
-  buildLogsDataFrame(logs: any[], target: MyQuery) {
-    const fieldsMapping: { [key: string]: FieldType } = {
-      Utf8: FieldType.string,
-      Int64: FieldType.number,
-      timestamp: FieldType.time,
-    };
-
-    const fields = [
-      { name: 'Time', type: FieldType.time },
-      { name: 'Content', type: FieldType.string },
-    ];
-
-    this.streamFields.forEach((field: any) => {
-      fields.push({
-        name: field.name,
-        type: fieldsMapping[field.type],
-      });
-    });
-
-    const frame = new MutableDataFrame({
-      refId: target.refId,
-      meta: {
-        preferredVisualisationType: 'logs',
-      },
-      fields,
-    });
-
-    logs.forEach((log: any) => {
-      frame.add({ ...log, Content: JSON.stringify(log) });
-    });
-
-    return frame;
-  }
-
   async query(options: DataQueryRequest<MyQuery>): Promise<DataQueryResponse> {
-    const timestamps = this.getConsumableTime(options.range);
+    const timestamps = getConsumableTime(options.range);
     const promises = options.targets.map((target) => {
-      const reqData = this.buildQuery(target, timestamps);
+      const reqData = buildQuery(target, timestamps, this.streamFields);
       return this.doRequest(target, reqData)
         .then((response) => {
-          return this.buildLogsDataFrame(response.hits, target);
+          if (target?.refId?.includes(REF_ID_STARTER_LOG_VOLUME)) {
+            return getGraphDataFrame(response, target);
+          }
+          return getLogsDataFrame(response, target, this.streamFields);
         })
         .catch((err) => {
-          let error = '';
-          if (err.response !== undefined) {
-            error = err.response.data.error;
+          let error = {
+            message: '',
+            detail: '',
+          };
+          if (err.data) {
+            error.message = err.data?.message;
+            error.detail = err.data?.error_detail;
           } else {
-            error = err.message;
+            error.message = err.statusText;
           }
 
-          const customMessage = logsErrorMessage(err.response.data.code);
+          const customMessage = logsErrorMessage(err.data.code);
           if (customMessage) {
-            error = customMessage;
+            error.message = customMessage;
           }
-          throw new Error(error);
+          throw new Error(error.message + (error.detail ? ` ( ${error.detail} ) ` : ''));
         });
     });
 
@@ -146,68 +121,54 @@ export class DataSource extends DataSourceApi<MyQuery, MyDataSourceOptions> {
     return { ...query, query: expression };
   }
 
-  buildQuery(queryData: MyQuery, timestamps: TimeRange) {
-    try {
-      let query: string = queryData.query || '';
-
-      let req: any = {
-        query: {
-          sql: 'select * from "[INDEX_NAME]" [WHERE_CLAUSE]',
-          start_time: timestamps.startTimeInMicro,
-          end_time: timestamps.endTimeInMirco,
-          size: 150,
-          sql_mode: 'full',
-        },
-      };
-
-      if (queryData.sqlMode) {
-        req.query.sql = queryData.query;
-      }
+  updateStreamFields(streamFields: any[]) {
+    this.streamFields = [...streamFields];
+  }
 
-      if (!queryData.sqlMode) {
-        let whereClause = query;
-
-        if (query.trim().length) {
-          whereClause = whereClause
-            .replace(/=(?=(?:[^"']*"[^"']*"')*[^"']*$)/g, ' =')
-            .replace(/>(?=(?:[^"']*"[^"']*"')*[^"']*$)/g, ' >')
-            .replace(/<(?=(?:[^"']*"[^"']*"')*[^"']*$)/g, ' <');
-
-          whereClause = whereClause
-            .replace(/!=(?=(?:[^"']*"[^"']*"')*[^"']*$)/g, ' !=')
-            .replace(/! =(?=(?:[^"']*"[^"']*"')*[^"']*$)/g, ' !=')
-            .replace(/< =(?=(?:[^"']*"[^"']*"')*[^"']*$)/g, ' <=')
-            .replace(/> =(?=(?:[^"']*"[^"']*"')*[^"']*$)/g, ' >=');
-
-          const parsedSQL = whereClause.split(' ');
-          this.streamFields.forEach((field: any) => {
-            parsedSQL.forEach((node: any, index: any) => {
-              if (node === field.name) {
-                node = node.replaceAll('"', '');
-                parsedSQL[index] = '"' + node + '"';
-              }
-            });
-          });
-
-          whereClause = parsedSQL.join(' ');
-
-          req.query.sql = req.query.sql.replace('[WHERE_CLAUSE]', ' WHERE ' + whereClause);
-        } else {
-          req.query.sql = req.query.sql.replace('[WHERE_CLAUSE]', '');
-        }
+  getDataProvider(
+    type: SupplementaryQueryType,
+    request: DataQueryRequest<MyQuery>
+  ): Observable<DataQueryResponse> | undefined {
+    if (!this.getSupportedSupplementaryQueryTypes().includes(type)) {
+      return undefined;
+    }
 
-        req.query.sql = req.query.sql.replace('[INDEX_NAME]', queryData.stream);
-      }
+    switch (type) {
+      case SupplementaryQueryType.LogsVolume:
+        return this.getLogsVolumeDataProvider(request);
+      default:
+        return undefined;
+    }
+  }
 
-      req['encoding'] = 'base64';
-      req.query.sql = b64EncodeUnicode(req.query.sql);
+  getSupportedSupplementaryQueryTypes(): SupplementaryQueryType[] {
+    return [SupplementaryQueryType.LogsVolume];
+    // return [SupplementaryQueryType.LogsVolume, SupplementaryQueryType.LogsSample];
+  }
 
-      return req;
-    } catch (e) {
-      console.log('error in building query:', e);
-    }
+  getSupplementaryQuery(type: SupplementaryQueryType, query: MyQuery): MyQuery | undefined {
+    return undefined;
   }
-  updateStreamFields(streamFields: any[]) {
-    this.streamFields = [...streamFields];
+
+  getLogsVolumeDataProvider(request: DataQueryRequest<MyQuery>): Observable<DataQueryResponse> | undefined {
+    const logsVolumeRequest = cloneDeep(request);
+    const targets = logsVolumeRequest.targets.map((target) => {
+      target['refId'] = REF_ID_STARTER_LOG_VOLUME + target.refId;
+      return target;
+    });
+
+    if (!targets.length) {
+      return undefined;
+    }
+
+    return queryLogsVolume(
+      this,
+      { ...logsVolumeRequest, targets },
+      {
+        extractLevel: () => LogLevel.unknown,
+        range: logsVolumeRequest.range,
+        targets: logsVolumeRequest.targets,
+      }
+    );
   }
 }