Function TestIngest

integration/remote_write_v2_test.go:273–394

(t *testing.T)

}

func TestIngest(t *testing.T) {
	const blockRangePeriod = 5 * time.Second

	s, err := e2e.NewScenario(networkName)
	require.NoError(t, err)
	defer s.Close()

	// Start dependencies.
	consul := e2edb.NewConsulWithName("consul")
	require.NoError(t, s.StartAndWaitReady(consul))

	flags := mergeFlags(
		AlertmanagerLocalFlags(),
		map[string]string{
			"-store.engine":                                    blocksStorageEngine,
			"-blocks-storage.backend":                          "filesystem",
			"-blocks-storage.tsdb.head-compaction-interval":    "4m",
			"-blocks-storage.bucket-store.sync-interval":       "15m",
			"-blocks-storage.bucket-store.index-cache.backend": tsdb.IndexCacheBackendInMemory,
			"-blocks-storage.tsdb.block-ranges-period":         blockRangePeriod.String(),
			"-blocks-storage.tsdb.ship-interval":               "1s",
			"-blocks-storage.tsdb.retention-period":            ((blockRangePeriod * 2) - 1).String(),
			"-blocks-storage.tsdb.enable-native-histograms":    "true",
			// Ingester.
			"-ring.store":      "consul",
			"-consul.hostname": consul.NetworkHTTPEndpoint(),
			// Distributor.
			"-distributor.replication-factor":     "1",
			"-distributor.remote-writev2-enabled": "true",
			// Store-gateway.
			"-store-gateway.sharding-enabled": "false",
			// alert manager
			"-alertmanager.web.external-url": "http://localhost/alertmanager",
		},
	)

	// make alert manager config dir
	require.NoError(t, writeFileToSharedDir(s, "alertmanager_configs", []byte{}))

	path := path.Join(s.SharedDir(), "cortex-1")

	flags = mergeFlags(flags, map[string]string{"-blocks-storage.filesystem.dir": path})
	// Start Cortex replicas.
	cortex := e2ecortex.NewSingleBinary("cortex", flags, "")
	require.NoError(t, s.StartAndWaitReady(cortex))

	// Wait until Cortex replicas have updated the ring state.
	require.NoError(t, cortex.WaitSumMetrics(e2e.Equals(float64(512)), "cortex_ring_tokens_total"))

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

	now := time.Now()

	// series push
	symbols1, series, expectedVector := e2e.GenerateSeriesV2("test_series", now, prompb.Label{Name: "job", Value: "test"}, prompb.Label{Name: "foo", Value: "bar"})
	writeStats, err := c.PushV2(symbols1, series)
	require.NoError(t, err)