e2e benchmarking Thanos run up.

e2e_store_and_querier.test

// Copyright (c) The Thanos Authors.
// Licensed under the Apache License 2.0.

package e2e_test

import (
	"context"
	"fmt"
	"net"
	"net/http"
	"path"
	"sync"
	"testing"

	"github.com/cortexproject/cortex/integration/e2e"
	e2edb "github.com/cortexproject/cortex/integration/e2e/db"
	"github.com/go-kit/kit/log"
	"github.com/prometheus/common/model"
	"github.com/prometheus/prometheus/pkg/relabel"
	"github.com/thanos-io/thanos/pkg/objstore"
	"github.com/thanos-io/thanos/pkg/objstore/client"
	"github.com/thanos-io/thanos/pkg/objstore/s3"
	"github.com/thanos-io/thanos/pkg/testutil"
	"github.com/thanos-io/thanos/test/e2e/e2ethanos"
)

func httpPrompt(t *testing.T) {
	t.Log("Running until you hit \"localhost:9191\"")

	wg := sync.WaitGroup{}
	wg.Add(1)
	o := sync.Once{}

	l, err := net.Listen("tcp", "localhost:9191")
	testutil.Ok(t, err)

	go func() {
		_ = http.Serve(l, http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) {
			o.Do(func() {
				wg.Done()
			})
			_, _ = w.Write([]byte("Finishing"))
		}))
	}()
	wg.Wait()
	testutil.Ok(t, l.Close())

	t.Log("Finishing")
}

func TestStoreGateway_Manual(t *testing.T) {
	s, err := e2e.NewScenario("e2e_test_store_gateway")
	testutil.Ok(t, err)
	t.Cleanup(e2ethanos.CleanScenario(t, s))

	m := e2edb.NewMinio(8080, "thanos")
	testutil.Ok(t, s.StartAndWaitReady(m))

	s1, err := e2ethanos.NewStoreGW(s.SharedDir(), "1", client.BucketConfig{
		Type: client.S3,
		Config: s3.Config{
			Bucket:    "thanos",
			AccessKey: e2edb.MinioAccessKey,
			SecretKey: e2edb.MinioSecretKey,
			Endpoint:  m.NetworkHTTPEndpoint(),
			Insecure:  true,
		},
	}, relabel.Config{
		Action:       relabel.Drop,
		Regex:        relabel.MustNewRegexp("value2"),
		SourceLabels: model.LabelNames{"ext1"},
	})
	testutil.Ok(t, err)
	testutil.Ok(t, s.StartAndWaitReady(s1))
	// Ensure bucket UI.
	r, err := http.Get("http://" + path.Join(s1.HTTPEndpoint(), "loaded"))
	testutil.Ok(t, err)
	testutil.Equals(t, http.StatusOK, r.StatusCode)

	q, err := e2ethanos.NewQuerier(s.SharedDir(), "1", []string{s1.GRPCNetworkEndpoint()}, nil, nil, "", "")
	testutil.Ok(t, err)
	testutil.Ok(t, s.StartAndWaitReady(q))

	l := log.NewNopLogger()
	bkt, err := s3.NewBucketWithConfig(l, s3.Config{
		Bucket:    "thanos",
		AccessKey: e2edb.MinioAccessKey,
		SecretKey: e2edb.MinioSecretKey,
		Endpoint:  m.HTTPEndpoint(), // We need separate client config, when connecting to minio from outside.
		Insecure:  true,
	}, "test-feed")
	testutil.Ok(t, err)

	testutil.Ok(t, objstore.UploadDir(context.Background(), l, bkt, "../../_dev/datahub/01DN3SK96XDAEKRB1AN30AAW6E", "01DN3SK96XDAEKRB1AN30AAW6E"))

	// Wait for store to sync blocks.
	testutil.Ok(t, s1.WaitSumMetrics(e2e.Equals(1), "thanos_blocks_meta_synced"))
	testutil.Ok(t, s1.WaitSumMetrics(e2e.Equals(0), "thanos_blocks_meta_sync_failures_total"))

	testutil.Ok(t, s1.WaitSumMetrics(e2e.Equals(1), "thanos_bucket_store_blocks_loaded"))
	testutil.Ok(t, s1.WaitSumMetrics(e2e.Equals(0), "thanos_bucket_store_block_drops_total"))
	testutil.Ok(t, s1.WaitSumMetrics(e2e.Equals(0), "thanos_bucket_store_block_load_failures_total"))

	t.Log("Done, waiting. query:", fmt.Sprintf("http://%v/graph?g0.range_input=15m&g0.end_input=2019-09-18%2018%3A20&g0.max_source_resolution=0s&g0.expr=subscription_sync_total&g0.tab=0", q.HTTPEndpoint()))
	httpPrompt(t)
	t.Log("Finished.")
}