Issue #12: implement DataSourceWithLogsContextSupport interface

CONTRIBUTING.md

@@ -14,14 +14,20 @@ You need:
 
 ### Building
 
+### All the stack
+
+```shell
+./build_and_start.sh
+```
+
 #### Frontend
 
 ```bash
-$ yarn install
-$ yarn build
+$ npm install
+$ npm run build
 ```
 
-When developing the front, use `yarn dev`.
+When developing the front, use `npm run dev`.
 
 #### Backend
 
@@ -49,7 +55,6 @@ $ npm run test
 $ go test -v ./pkg/...
 ```
 
-
 ## Release
 
 TODO

build_and_start.sh

@@ -0,0 +1,6 @@
+#!/usr/bin/env bash
+
+npm install
+npm run build
+mage -v
+docker-compose up --build --force-recreate

docker-compose.yaml

@@ -12,7 +12,9 @@ services:
     volumes:
       - ./:/var/lib/grafana/plugins/grafana-quickwit-datasource
       - ./provisioning:/etc/grafana/provisioning
-    # - ./grafana/storage:/var/lib/grafana
-    # - ./grafana/grafana.ini:/etc/grafana/grafana.ini
+      - gquickwit:/var/lib/grafana
     extra_hosts:
       - "host.docker.internal:host-gateway"
+
+volumes:
+  gquickwit:

pkg/quickwit/client/search_request.go

@@ -101,11 +101,11 @@ func (b *SearchRequestBuilder) Sort(order SortOrder, field string, unmappedType
 	return b
 }
 
-func (b *SearchRequestBuilder) AddSearchAfter(value interface{}) *SearchRequestBuilder {
+func (b *SearchRequestBuilder) AddSearchAfter(value any) *SearchRequestBuilder {
 	if b.customProps["search_after"] == nil {
-		b.customProps["search_after"] = []interface{}{value}
+		b.customProps["search_after"] = []any{value}
 	} else {
-		b.customProps["search_after"] = append(b.customProps["search_after"].([]interface{}), value)
+		b.customProps["search_after"] = append(b.customProps["search_after"].([]any), value)
 	}
 
 	return b

pkg/quickwit/data_query.go

@@ -340,6 +340,13 @@ func processLogsQuery(q *Query, b *es.SearchRequestBuilder, from, to int64, defa
 	b.Size(stringToIntWithDefaultValue(metric.Settings.Get("limit").MustString(), defaultSize))
 	// TODO when hightlight is supported in quickwit
 	// b.AddHighlight()
+
+	// This is currently used only for log context query to get
+	// log lines before and after the selected log line
+	searchAfter := metric.Settings.Get("searchAfter").MustArray()
+	for _, value := range searchAfter {
+		b.AddSearchAfter(value)
+	}
 }
 
 func processDocumentQuery(q *Query, b *es.SearchRequestBuilder, from, to int64, defaultTimeField string) {

src/IntervalMap.ts

@@ -0,0 +1,18 @@
+import { DurationUnit } from '@grafana/data';
+import { Interval } from './types';
+
+type IntervalMap = Record<
+  Interval,
+  {
+    startOf: DurationUnit;
+    amount: DurationUnit;
+  }
+>;
+
+export const intervalMap: IntervalMap = {
+  Hourly: { startOf: 'hour', amount: 'hours' },
+  Daily: { startOf: 'day', amount: 'days' },
+  Weekly: { startOf: 'isoWeek', amount: 'weeks' },
+  Monthly: { startOf: 'month', amount: 'months' },
+  Yearly: { startOf: 'year', amount: 'years' },
+};

src/datasource.ts

@@ -1,35 +1,40 @@
 import { cloneDeep, first as _first, map as _map, groupBy } from 'lodash';
 import { Observable, lastValueFrom, from, isObservable, of } from 'rxjs';
 import { catchError, mergeMap, map } from 'rxjs/operators';
+import { intervalMap } from './IntervalMap';
 
 import {
   AbstractQuery,
+  CoreApp,
   DataFrame,
+  DataQueryError,
   DataQueryRequest,
   DataQueryResponse,
   DataSourceApi,
   DataSourceInstanceSettings,
   DataSourceJsonData,
+  DataSourceWithLogsContextSupport,
   DataSourceWithQueryImportSupport,
   DataSourceWithSupplementaryQueriesSupport,
+  dateTime,
   FieldColorModeId,
   FieldType,
   getDefaultTimeRange,
   LoadingState,
   LogLevel,
+  LogRowModel,
   LogsVolumeCustomMetaData,
   LogsVolumeType,
   MetricFindValue,
   QueryFixAction,
+  rangeUtil,
   ScopedVars,
   SupplementaryQueryType,
   TimeRange,
 } from '@grafana/data';
-import { BucketAggregation, DataLinkConfig, ElasticsearchQuery, Field, FieldMapping, IndexMetadata, TermsQuery } from './types';
-import {
-  DataSourceWithBackend, getTemplateSrv, TemplateSrv,
-} from '@grafana/runtime';
-import { QuickwitOptions } from 'quickwit';
+import { BucketAggregation, DataLinkConfig, ElasticsearchQuery, Field, FieldMapping, IndexMetadata, Logs, TermsQuery, Interval } from './types';
+import { DataSourceWithBackend, getTemplateSrv, TemplateSrv } from '@grafana/runtime';
+import { LogRowContextOptions, LogRowContextQueryDirection, QuickwitOptions } from 'quickwit';
 import { ElasticQueryBuilder } from 'QueryBuilder';
 import { colors } from '@grafana/ui';
 
@@ -39,7 +44,7 @@ import { isMetricAggregationWithField } from 'components/QueryEditor/MetricAggre
 import { bucketAggregationConfig } from 'components/QueryEditor/BucketAggregationsEditor/utils';
 import { isBucketAggregationWithField } from 'components/QueryEditor/BucketAggregationsEditor/aggregations';
 import ElasticsearchLanguageProvider from 'LanguageProvider';
-
+import { ReactNode } from 'react';
 
 export const REF_ID_STARTER_LOG_VOLUME = 'log-volume-';
 
@@ -48,6 +53,7 @@ export type ElasticDatasource = QuickwitDataSource;
 export class QuickwitDataSource
   extends DataSourceWithBackend<ElasticsearchQuery, QuickwitOptions>
   implements
+    DataSourceWithLogsContextSupport,
     DataSourceWithSupplementaryQueriesSupport<ElasticsearchQuery>,
     DataSourceWithQueryImportSupport<ElasticsearchQuery>
 {
@@ -59,6 +65,7 @@ export class QuickwitDataSource
   queryBuilder: ElasticQueryBuilder;
   dataLinks: DataLinkConfig[];
   languageProvider: ElasticsearchLanguageProvider;
+  intervalPattern?: Interval;
 
   constructor(
     instanceSettings: DataSourceInstanceSettings<QuickwitOptions>,
@@ -427,6 +434,78 @@ export class QuickwitDataSource
     return text;
   }
 
+  private makeLogContextDataRequest = (row: LogRowModel, options?: LogRowContextOptions) => {
+    const direction = options?.direction || LogRowContextQueryDirection.Backward;
+    const searchAfterNs = row.dataFrame.fields.find((f) => f.name === 'sort')?.values.get(row.rowIndex) ?? [row.timeEpochNs]
+    const searchAfterMs = [Math.round(searchAfterNs[0]/1000000)]
+
+    const logQuery: Logs = {
+      type: 'logs',
+      id: '1',
+      settings: {
+        limit: options?.limit ? options?.limit.toString() : '10',
+        // Sorting of results in the context query
+        sortDirection: direction === LogRowContextQueryDirection.Backward ? 'desc' : 'asc',
+        // Used to get the next log lines before/after the current log line using sort field of selected log line
+        searchAfter: searchAfterMs,
+      },
+    };
+
+    const query: ElasticsearchQuery = {
+      refId: `log-context-${row.dataFrame.refId}-${direction}`,
+      metrics: [logQuery],
+      query: '',
+    };
+
+    const timeRange = createContextTimeRange(row.timeEpochMs, direction, this.intervalPattern);
+    const range = {
+      from: timeRange.from,
+      to: timeRange.to,
+      raw: timeRange,
+    };
+
+    const interval = rangeUtil.calculateInterval(range, 1);
+
+    const contextRequest: DataQueryRequest<ElasticsearchQuery> = {
+      requestId: `log-context-request-${row.dataFrame.refId}-${options?.direction}`,
+      targets: [query],
+      interval: interval.interval,
+      intervalMs: interval.intervalMs,
+      range,
+      scopedVars: {},
+      timezone: 'UTC',
+      app: CoreApp.Explore,
+      startTime: Date.now(),
+      hideFromInspector: true,
+    };
+    return contextRequest;
+  };
+
+  getLogRowContext = async (row: LogRowModel, options?: LogRowContextOptions): Promise<{ data: DataFrame[] }> => {
+    const contextRequest = this.makeLogContextDataRequest(row, options);
+
+    return lastValueFrom(
+      this.query(contextRequest).pipe(
+        catchError((err) => {
+          const error: DataQueryError = {
+            message: 'Error during context query. Please check JS console logs.',
+            status: err.status,
+            statusText: err.statusText,
+          };
+          throw error;
+        })
+      )
+    );
+  };
+
+  showContextToggle(row?: LogRowModel | undefined): boolean {
+    return true;
+  }
+
+  getLogRowContextUi?(row: LogRowModel, runContextQuery?: (() => void) | undefined): ReactNode {
+    return true;
+  }
+
   /**
    * Returns false if the query should be skipped
    */
@@ -740,3 +819,36 @@ function luceneEscape(value: string) {
 
   return value.replace(/([\!\*\+\-\=<>\s\&\|\(\)\[\]\{\}\^\~\?\:\\/"])/g, '\\$1');
 }
+
+function createContextTimeRange(rowTimeEpochMs: number, direction: string, intervalPattern: Interval | undefined) {
+  const offset = 7;
+  // For log context, we want to request data from 7 subsequent/previous indices
+  if (intervalPattern) {
+    const intervalInfo = intervalMap[intervalPattern];
+    if (direction === LogRowContextQueryDirection.Forward) {
+      return {
+        from: dateTime(rowTimeEpochMs).utc(),
+        to: dateTime(rowTimeEpochMs).add(offset, intervalInfo.amount).utc().startOf(intervalInfo.startOf),
+      };
+    } else {
+      return {
+        from: dateTime(rowTimeEpochMs).subtract(offset, intervalInfo.amount).utc().startOf(intervalInfo.startOf),
+        to: dateTime(rowTimeEpochMs).utc(),
+      };
+    }
+    // If we don't have an interval pattern, we can't do this, so we just request data from 7h before/after
+  } else {
+    if (direction === LogRowContextQueryDirection.Forward) {
+      return {
+        from: dateTime(rowTimeEpochMs).utc(),
+        to: dateTime(rowTimeEpochMs).add(offset, 'hours').utc(),
+      };
+    } else {
+      return {
+        from: dateTime(rowTimeEpochMs).subtract(offset, 'hours').utc(),
+        to: dateTime(rowTimeEpochMs).utc(),
+      };
+    }
+  }
+}
+