Function TestQuerierDistributedExecution

integration/querier_test.go:875–948

(t *testing.T)

}

func TestQuerierDistributedExecution(t *testing.T) {
	// e2e test setup
	s, err := e2e.NewScenario(networkName)
	require.NoError(t, err)
	defer s.Close()

	// initialize the flags
	flags := mergeFlags(
		BlocksStorageFlags(),
		map[string]string{
			"-blocks-storage.tsdb.block-ranges-period": (5 * time.Second).String(),
			"-blocks-storage.tsdb.ship-interval":       "1s",
			"-blocks-storage.tsdb.retention-period":    ((5 * time.Second * 2) - 1).String(),
			"-querier.thanos-engine":                   "true",
			// enable distributed execution (logical plan execution)
			"-querier.distributed-exec-enabled": "true",
		},
	)

	minio := e2edb.NewMinio(9000, flags["-blocks-storage.s3.bucket-name"])
	consul := e2edb.NewConsul()
	require.NoError(t, s.StartAndWaitReady(consul, minio))

	// start services
	distributor := e2ecortex.NewDistributor("distributor", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
	ingester := e2ecortex.NewIngester("ingester", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
	queryScheduler := e2ecortex.NewQueryScheduler("query-scheduler", flags, "")
	storeGateway := e2ecortex.NewStoreGateway("store-gateway", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
	require.NoError(t, s.StartAndWaitReady(queryScheduler, distributor, ingester, storeGateway))
	flags = mergeFlags(flags, map[string]string{
		"-querier.store-gateway-addresses": strings.Join([]string{storeGateway.NetworkGRPCEndpoint()}, ","),
	})

	queryFrontend := e2ecortex.NewQueryFrontend("query-frontend", mergeFlags(flags, map[string]string{
		"-frontend.scheduler-address": queryScheduler.NetworkGRPCEndpoint(),
	}), "")
	require.NoError(t, s.Start(queryFrontend))

	querier := e2ecortex.NewQuerier("querier", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), mergeFlags(flags, map[string]string{
		"-querier.scheduler-address": queryScheduler.NetworkGRPCEndpoint(),
	}), "")
	require.NoError(t, s.StartAndWaitReady(querier))

	// wait until the distributor and querier has updated the ring.
	require.NoError(t, distributor.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total"))
	require.NoError(t, querier.WaitSumMetrics(e2e.Equals(2*512), "cortex_ring_tokens_total"))

	c, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), queryFrontend.HTTPEndpoint(), "", "", "user-1")
	require.NoError(t, err)

	series1Timestamp := time.Now()
	series2Timestamp := series1Timestamp.Add(time.Minute * 1)
	series1, expectedVector1 := generateSeries("series_1", series1Timestamp, prompb.Label{Name: "series_1", Value: "series_1"})
	series2, expectedVector2 := generateSeries("series_2", series2Timestamp, prompb.Label{Name: "series_2", Value: "series_2"})

	res, err := c.Push(series1)
	require.NoError(t, err)
	require.Equal(t, 200, res.StatusCode)