fix(logql-compliance-tester): wait until instance gets some data

cmd/logql-compliance-tester/lokiapi.go

@@ -27,17 +27,44 @@ func newLokiAPI(ctx context.Context, cfg lokicompliance.TargetConfig) (lokicompl
 }
 
 func waitForLoki(ctx context.Context, c *lokiapi.Client, cfg lokicompliance.TargetConfig) error {
+	check := func(ctx context.Context) error {
+		q := cfg.ReadyQuery
+		if q == "" {
+			q = `{job="varlogs"}`
+		}
+
+		resp, err := c.Query(ctx, lokiapi.QueryParams{
+			Query: q,
+		})
+		if err != nil {
+			if cerr := ctx.Err(); cerr != nil {
+				return backoff.Permanent(cerr)
+			}
+			return err
+		}
+
+		streams, ok := resp.Data.GetStreamsResult()
+		if !ok {
+			// Ready query should be exactly a log query.
+			err := errors.Errorf("unexpected result type %q", resp.Data.Type)
+			return backoff.Permanent(err)
+		}
+
+		for _, s := range streams.Result {
+			if len(s.Values) > 0 {
+				return nil
+			}
+		}
+		return errors.New("empty result")
+	}
+
 	var (
 		b   = backoff.NewConstantBackOff(5 * time.Second)
 		log = zctx.From(ctx)
 	)
 	if err := backoff.RetryNotify(
 		func() error {
-			_, err := c.Labels(ctx, lokiapi.LabelsParams{})
-			if cerr := ctx.Err(); cerr != nil {
-				return backoff.Permanent(cerr)
-			}
-			return err
+			return check(ctx)
 		},
 		b,
 		func(err error, d time.Duration) {
@@ -46,6 +73,6 @@ func waitForLoki(ctx context.Context, c *lokiapi.Client, cfg lokicompliance.Targ
 	); err != nil {
 		return err
 	}
-	log.Info("Loki is ready", zap.String("target", cfg.QueryURL))
+	log.Info("Target is ready", zap.String("target", cfg.QueryURL))
 	return nil
 }

cmd/logql-compliance-tester/main.go

@@ -16,6 +16,8 @@ import (
 	"github.com/cheggaaa/pb/v3"
 	"github.com/fatih/color"
 	"github.com/go-faster/errors"
+	"github.com/go-faster/sdk/zctx"
+	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/go-faster/oteldb/internal/lokicompliance"
@@ -113,17 +115,28 @@ func run(ctx context.Context) error {
 		rangeDuration    = flag.Duration("range", 10*time.Minute, "The duration of the query range.")
 		stepDuration     = flag.Duration("step", 10*time.Second, "The step of the query.")
 
+		wait = flag.Duration("wait", time.Minute, "The amonunt of time to wait until storages are filled up.")
+
 		outputFile        = flag.String("output-file", "", "Path to output file")
 		outputFormat      = flag.String("output-format", "text", "The comparison output format. Valid values: [json]")
 		outputPassing     = flag.Bool("output-passing", false, "Whether to also include passing test cases in the output.")
 		minimumPercentage = flag.Float64("target", math.NaN(), "Minimum compliance percentage")
 	)
-	flag.Parse()
 
+	flag.Parse()
 	if format := *outputFormat; format != "json" {
 		return errors.Errorf("unknown output format %q", format)
 	}
 
+	log, err := zap.NewDevelopment()
+	if err != nil {
+		return errors.Wrap(err, "create logger")
+	}
+	defer func() {
+		_ = log.Sync()
+	}()
+	ctx = zctx.Base(ctx, log)
+
 	cfg, err := lokicompliance.LoadFromFiles(configFile)
 	if err != nil {
 		return errors.Wrap(err, "loading configuration file")
@@ -153,6 +166,19 @@ func run(ctx context.Context) error {
 		return errors.Wrap(err, "expand test cases")
 	}
 
+	{
+		log.Info("Waiting for data", zap.Duration("wait", *wait))
+
+		waitTimer := time.NewTimer(*wait)
+		defer waitTimer.Stop()
+
+		select {
+		case <-waitTimer.C:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+
 	var (
 		results     = make([]*lokicompliance.Result, len(testCases))
 		progressBar = pb.StartNew(len(results))

internal/lokicompliance/config.go

@@ -36,7 +36,9 @@ type QueryParameters struct {
 
 // TargetConfig represents the configuration of a single Prometheus API endpoint.
 type TargetConfig struct {
-	QueryURL string `yaml:"query_url"`
+	// ReadyQuery is a log query to check instance readiness.
+	ReadyQuery string `yaml:"ready_query"`
+	QueryURL   string `yaml:"query_url"`
 }
 
 // LoadFromFiles parses the given YAML files into a Config.