feat(lokicompliance): add package to test LogQL engine compliance

internal/lokicompliance/compare.go

@@ -0,0 +1,137 @@
+// Package lokicompliance provides utilities for Loki/LogQL compliance testing.
+package lokicompliance
+
+import (
+	"context"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/go-faster/errors"
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+
+	"github.com/go-faster/oteldb/internal/lokiapi"
+)
+
+const (
+	defaultFraction = 0.00001
+	defaultMargin   = 0.0
+)
+
+var _ LokiAPI = (*lokiapi.Client)(nil)
+
+// LokiAPI represents LogQL API.
+type LokiAPI interface {
+	Query(ctx context.Context, params lokiapi.QueryParams) (*lokiapi.QueryResponse, error)
+	QueryRange(ctx context.Context, params lokiapi.QueryRangeParams) (*lokiapi.QueryResponse, error)
+}
+
+// TestCase represents a fully expanded query to be tested.
+type TestCase struct {
+	Query          string            `json:"query"`
+	SkipComparison bool              `json:"skipComparison"`
+	ShouldFail     bool              `json:"shouldFail"`
+	Start          time.Time         `json:"start"`
+	End            time.Time         `json:"end"`
+	Step           time.Duration     `json:"step"`
+	Limit          int               `json:"limit"`
+	Direction      lokiapi.Direction `json:"direction"`
+}
+
+// A Comparer allows comparing query results for test cases between a reference API and a test API.
+type Comparer struct {
+	refAPI         LokiAPI
+	testAPI        LokiAPI
+	compareOptions cmp.Options
+}
+
+// New returns a new Comparer.
+func New(refAPI, testAPI LokiAPI) *Comparer {
+	var options cmp.Options
+	addFloatCompareOptions(&options)
+	return &Comparer{
+		refAPI:         refAPI,
+		testAPI:        testAPI,
+		compareOptions: options,
+	}
+}
+
+// Result tracks a single test case's query comparison result.
+type Result struct {
+	TestCase          *TestCase `json:"testCase"`
+	Diff              string    `json:"diff"`
+	UnexpectedFailure string    `json:"unexpectedFailure"`
+	UnexpectedSuccess bool      `json:"unexpectedSuccess"`
+	Unsupported       bool      `json:"unsupported"`
+}
+
+// Success returns true if the comparison result was successful.
+func (r *Result) Success() bool {
+	return r.Diff == "" && !r.UnexpectedSuccess && r.UnexpectedFailure == ""
+}
+
+func getLokiTime(t time.Time) lokiapi.LokiTime {
+	ts := strconv.FormatInt(t.UnixNano(), 10)
+	return lokiapi.LokiTime(ts)
+}
+
+func getLokiDuration(t time.Duration) lokiapi.PrometheusDuration {
+	return lokiapi.PrometheusDuration(t.String())
+}
+
+// Compare runs a test case query against the reference API and the test API and compares the results.
+func (c *Comparer) Compare(tc *TestCase) (*Result, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	params := lokiapi.QueryRangeParams{
+		Query:     tc.Query,
+		Start:     lokiapi.NewOptLokiTime(getLokiTime(tc.Start)),
+		End:       lokiapi.NewOptLokiTime(getLokiTime(tc.End)),
+		Step:      lokiapi.NewOptPrometheusDuration(getLokiDuration(tc.Step)),
+		Direction: lokiapi.NewOptDirection(tc.Direction),
+		Limit:     lokiapi.NewOptInt(tc.Limit),
+	}
+
+	refResult, refErr := c.refAPI.QueryRange(ctx, params)
+	testResult, testErr := c.testAPI.QueryRange(ctx, params)
+
+	if (refErr != nil) != tc.ShouldFail {
+		if refErr != nil {
+			return nil, errors.Wrapf(refErr, "querying reference API for %q", tc.Query)
+		}
+		return nil, errors.Errorf("expected reference API query %q to fail, but succeeded", tc.Query)
+	}
+
+	if (testErr != nil) != tc.ShouldFail {
+		if testErr != nil {
+			return &Result{
+				TestCase:          tc,
+				UnexpectedFailure: testErr.Error(),
+				Unsupported:       strings.Contains(testErr.Error(), "501"),
+			}, nil
+		}
+		return &Result{TestCase: tc, UnexpectedSuccess: true}, nil
+	}
+
+	if tc.SkipComparison || tc.ShouldFail {
+		return &Result{TestCase: tc}, nil
+	}
+
+	return &Result{
+		TestCase: tc,
+		Diff:     cmp.Diff(refResult, testResult, c.compareOptions),
+	}, nil
+}
+
+func addFloatCompareOptions(options *cmp.Options) {
+	fraction := defaultFraction
+	margin := defaultMargin
+	*options = append(
+		*options,
+		cmpopts.EquateApprox(fraction, margin),
+		// A NaN is usually not treated as equal to another NaN, but we want to treat it as such here.
+		cmpopts.EquateNaNs(),
+	)
+}

internal/lokicompliance/config.go

@@ -0,0 +1,69 @@
+package lokicompliance
+
+import (
+	"bytes"
+	"os"
+
+	"github.com/go-faster/errors"
+	"github.com/go-faster/yaml"
+
+	"github.com/go-faster/oteldb/internal/lokiapi"
+)
+
+// Config models the main configuration file.
+type Config struct {
+	ReferenceTargetConfig TargetConfig       `yaml:"reference_target_config"`
+	TestTargetConfig      TargetConfig       `yaml:"test_target_config"`
+	TestCases             []*TestCasePattern `yaml:"test_cases"`
+	QueryParameters       QueryParameters    `yaml:"query_parameters"`
+}
+
+// TestCase represents a given query (pattern) to be tested.
+type TestCasePattern struct {
+	Query          string   `yaml:"query"`
+	VariantArgs    []string `yaml:"variant_args,omitempty"`
+	SkipComparison bool     `yaml:"skip_comparison,omitempty"`
+	ShouldFail     bool     `yaml:"should_fail,omitempty"`
+}
+
+type QueryParameters struct {
+	EndTime        string            `yaml:"end_time"`
+	RangeInSeconds float64           `yaml:"range_in_seconds"`
+	StepInSeconds  float64           `yaml:"step_in_seconds"`
+	Direction      lokiapi.Direction `yaml:"direction"`
+	Limit          *int              `yaml:"limit"`
+}
+
+// TargetConfig represents the configuration of a single Prometheus API endpoint.
+type TargetConfig struct {
+	QueryURL string `yaml:"query_url"`
+}
+
+// LoadFromFiles parses the given YAML files into a Config.
+func LoadFromFiles(filenames []string) (*Config, error) {
+	var buf bytes.Buffer
+	for _, f := range filenames {
+		content, err := os.ReadFile(f) // #nosec G304
+		if err != nil {
+			return nil, errors.Wrapf(err, "reading config file %s", f)
+		}
+		if _, err := buf.Write(content); err != nil {
+			return nil, errors.Wrapf(err, "appending config file %s to buffer", f)
+		}
+	}
+	cfg, err := Load(buf.Bytes())
+	if err != nil {
+		return nil, errors.Wrapf(err, "parsing YAML files %s", filenames)
+	}
+	return cfg, nil
+}
+
+// Load parses the YAML input into a Config.
+func Load(content []byte) (*Config, error) {
+	cfg := &Config{}
+	err := yaml.Unmarshal(content, cfg)
+	if err != nil {
+		return nil, err
+	}
+	return cfg, nil
+}

internal/lokicompliance/expand.go

@@ -0,0 +1,121 @@
+package lokicompliance
+
+import (
+	"slices"
+	"strings"
+	"text/template"
+	"time"
+
+	"github.com/go-faster/errors"
+
+	"github.com/go-faster/oteldb/internal/lokiapi"
+)
+
+var testVariantArgs = map[string][]string{
+	"range":        {"1s", "15s", "1m", "5m", "15m", "1h"},
+	"offset":       {"1m", "5m", "10m"},
+	"simpleAggrOp": {"sum", "avg", "max", "min", "count", "stddev", "stdvar"},
+	"topBottomOp":  {"topk", "bottomk"},
+	"quantile": {
+		"-0.5",
+		"0.1",
+		"0.5",
+		"0.75",
+		"0.95",
+		"0.90",
+		"0.99",
+		"1",
+		"1.5",
+	},
+	"lineFilterOp": {"|=", "!=", "~=", "!~"},
+	"arithBinOp":   {"+", "-", "*", "/", "%", "^"},
+	"logicBinOp":   {"and", "or", "unless"},
+}
+
+func execTemplateToString(t *template.Template, data any) (string, error) {
+	var sb strings.Builder
+	if err := t.Execute(&sb, data); err != nil {
+		return "", err
+	}
+	return sb.String(), nil
+}
+
+func getQueries(
+	t *template.Template,
+	variantArgs []string,
+	args map[string]string,
+	add func(string),
+) error {
+	if len(variantArgs) == 0 {
+		q, err := execTemplateToString(t, args)
+		if err != nil {
+			return err
+		}
+		add(q)
+		return nil
+	}
+
+	arg := variantArgs[0]
+	values, ok := testVariantArgs[arg]
+	if !ok {
+		return errors.Errorf("unknown arg %q", arg)
+	}
+	variantArgs = variantArgs[1:]
+
+	for _, val := range values {
+		args[arg] = val
+		if err := getQueries(t, variantArgs, args, add); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// ExpandQuery expands given test case.
+func ExpandQuery(cfg *Config, start, end time.Time, step time.Duration) (r []*TestCase, _ error) {
+	var (
+		params    = cfg.QueryParameters
+		limit     = 10000
+		direction = lokiapi.DirectionForward
+	)
+	if l := params.Limit; l != nil {
+		limit = *l
+	}
+	if d := params.Direction; d != "" {
+		direction = d
+	}
+
+	for _, tc := range cfg.TestCases {
+		templ, err := template.New("query").Parse(tc.Query)
+		if err != nil {
+			return nil, errors.Wrapf(err, "parse query template %q", tc.Query)
+		}
+
+		// Sort and deduplicate args.
+		args := tc.VariantArgs
+		slices.Sort(args)
+		args = slices.Compact(args)
+
+		if err := getQueries(
+			templ,
+			tc.VariantArgs,
+			make(map[string]string, len(args)),
+			func(query string) {
+				r = append(r, &TestCase{
+					Query:          query,
+					SkipComparison: tc.SkipComparison,
+					ShouldFail:     tc.ShouldFail,
+					Start:          start,
+					End:            end,
+					Step:           step,
+					Limit:          limit,
+					Direction:      direction,
+				})
+			},
+		); err != nil {
+			return nil, errors.Wrapf(err, "expand query %q", tc.Query)
+		}
+	}
+
+	return r, nil
+}

internal/lokicompliance/expand_test.go

@@ -0,0 +1,30 @@
+package lokicompliance
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestExpandQuery(t *testing.T) {
+	arg := "offset"
+	cfg := &Config{
+		TestCases: []*TestCasePattern{
+			{
+				Query:       "{{ ." + arg + " }}",
+				VariantArgs: []string{arg},
+			},
+		},
+	}
+
+	ts := time.Time{}
+	tcs, err := ExpandQuery(cfg, ts, ts, 0)
+	require.NoError(t, err)
+
+	var queries []string
+	for _, tc := range tcs {
+		queries = append(queries, tc.Query)
+	}
+	require.ElementsMatch(t, queries, testVariantArgs[arg])
+}

internal/lokicompliance/lokicompliance.go

@@ -0,0 +1,2 @@
+// Packagel lokicompliance provides utilities for Loki/LogQL compliance testing.
+package lokicompliance