test(lokie2e): add explain query test

go-faster · Jul 8, 2024 · e40c2d9 · e40c2d9
1 parent 7b9af32
commit e40c2d9
Show file tree

Hide file tree

Showing 3 changed files with 86 additions and 6 deletions.
diff --git a/integration/lokie2e/common_test.go b/integration/lokie2e/common_test.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/http/httptest"
 	"os"
+	"regexp"
 	"slices"
 	"testing"
 	"time"
@@ -522,6 +523,86 @@ func runTest(
 			})
 		}
 	})
+	t.Run("Explain", func(t *testing.T) {
+		for i, tt := range []struct {
+			query    string
+			contains []string
+		}{
+			{
+				`{http_method=~".+"} |= "HEAD"`,
+				[]string{
+					`Offloading line filters.+|=`,
+					`Pipeline could be fully offloaded to Clickhouse`,
+				},
+			},
+			{
+				`{http_method=~".+"} | http_method = "GET"`,
+				[]string{
+					`Offloading pipeline label filters.+http_method=`,
+					`Pipeline could be fully offloaded to Clickhouse`,
+				},
+			},
+			{
+				`{http_method=~".+"} |= "HEAD" | http_method = "GET" | json | status != 200`,
+				[]string{
+					`Offloading line filters.+|=`,
+					`Offloading pipeline label filters.+http_method=`,
+				},
+			},
+
+			{
+				`sum by (http_method) ( count_over_time({http_method=~".+"} [30s]) )`,
+				[]string{
+					`Sampling could be offloaded to Clickhouse`,
+				},
+			},
+			{
+				`sum by (http_method) ( count_over_time({http_method=~".+"} |= "HEAD" [30s]) )`,
+				[]string{
+					`Offloading line filters.+|=`,
+					`Pipeline could be fully offloaded to Clickhouse`,
+					`Sampling could be offloaded to Clickhouse`,
+				},
+			},
+		} {
+			tt := tt
+			t.Run(fmt.Sprintf("Test%d", i+1), func(t *testing.T) {
+				t.Parallel()
+
+				resp, err := c.QueryRange(ctx, lokiapi.QueryRangeParams{
+					// Expected result is empty
+					Query: "@explain\n" + tt.query,
+					// Query all data in a one step.
+					Start: lokiapi.NewOptLokiTime(asLokiTime(set.End)),
+					End:   lokiapi.NewOptLokiTime(asLokiTime(set.End + otelstorage.Timestamp(10*time.Second))),
+					Step:  lokiapi.NewOptPrometheusDuration("30s"),
+					Limit: lokiapi.NewOptInt(1000),
+				})
+				require.NoError(t, err)
+
+				data, ok := resp.Data.GetStreamsResult()
+				require.True(t, ok)
+				streams := data.Result
+				require.Len(t, streams, 1)
+				entries := streams[0].Values
+				require.NotEmpty(t, streams, entries)
+
+				for _, pattern := range tt.contains {
+					re, err := regexp.Compile(pattern)
+					require.NoError(t, err)
+
+					require.True(t,
+						slices.ContainsFunc(entries, func(s lokiapi.LogEntry) bool {
+							return re.MatchString(s.V)
+						}),
+						"There is should be at least one log entry that matches %q in %#v",
+						pattern,
+						entries,
+					)
+				}
+			})
+		}
+	})
 	t.Run("MetricQueries", func(t *testing.T) {
 		t.Run("EmptySampleQuery", func(t *testing.T) {
 			resp, err := c.QueryRange(ctx, lokiapi.QueryRangeParams{

diff --git a/internal/chstorage/querier.go b/internal/chstorage/querier.go
@@ -119,7 +119,6 @@ func (q *Querier) do(ctx context.Context, s selectQuery) error {
 			zap.String("query_type", s.Type),
 			zap.String("table", s.Table),
 			zap.String("signal", s.Signal),
-			zap.String("body", query.Body),
 			zap.Duration("took", took),
 			errField,
 		)

diff --git a/internal/chstorage/querier_logs_optimizer.go b/internal/chstorage/querier_logs_optimizer.go
@@ -103,7 +103,7 @@ func (o *ClickhouseOptimizer) buildRangeAggregationSampling(n *logqlengine.Range
 		return n
 	}
 
-	if ce := lg.Check(zap.DebugLevel, "Offloading sampling"); ce != nil {
+	if ce := lg.Check(zap.DebugLevel, "Sampling could be offloaded to Clickhouse"); ce != nil {
 		ce.Write(
 			zap.Stringer("sampling_op", samplingOp),
 			zap.Stringers("grouping_labels", grouping),
@@ -153,16 +153,16 @@ func (o *ClickhouseOptimizer) optimizePipeline(n logqlengine.PipelineNode, lg *z
 	}
 
 	sn.Sel.Line = o.offloadLineFilters(pn.Pipeline)
-	if len(sn.Sel.Line) > 0 {
+	if f := sn.Sel.Line; len(f) > 0 {
 		if ce := lg.Check(zap.DebugLevel, "Offloading line filters"); ce != nil {
-			ce.Write(zap.Stringers("line_filters", sn.Sel.Line))
+			ce.Write(zap.Stringers("line_filters", f))
 		}
 	}
 
 	sn.Sel.PipelineLabels = o.offloadLabelFilters(pn.Pipeline)
-	if len(sn.Sel.PipelineLabels) > 0 {
+	if f := sn.Sel.PipelineLabels; len(f) > 0 {
 		if ce := lg.Check(zap.DebugLevel, "Offloading pipeline label filters"); ce != nil {
-			ce.Write(zap.Stringers("pipeline_labels", sn.Sel.Line))
+			ce.Write(zap.Stringers("pipeline_labels", f))
 		}
 	}