(t *testing.T)
| 19 | ) |
| 20 | |
| 21 | func TestGRPCCompression(t *testing.T) { |
| 22 | compressions := []string{"snappy", "snappy-block", "gzip", "zstd"} |
| 23 | |
| 24 | for _, compression := range compressions { |
| 25 | t.Run(fmt.Sprintf("ingester client/%s", compression), func(t *testing.T) { |
| 26 | s, err := e2e.NewScenario(networkName) |
| 27 | require.NoError(t, err) |
| 28 | defer s.Close() |
| 29 | |
| 30 | flags := mergeFlags(BlocksStorageFlags(), map[string]string{ |
| 31 | "-distributor.replication-factor": "1", |
| 32 | "-ingester.client.grpc-compression": compression, |
| 33 | }) |
| 34 | |
| 35 | consul := e2edb.NewConsul() |
| 36 | minio := e2edb.NewMinio(9000, flags["-blocks-storage.s3.bucket-name"]) |
| 37 | require.NoError(t, s.StartAndWaitReady(consul, minio)) |
| 38 | |
| 39 | distributor := e2ecortex.NewDistributor("distributor", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "") |
| 40 | ingester := e2ecortex.NewIngester("ingester-1", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "") |
| 41 | querier := e2ecortex.NewQuerier("querier", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "") |
| 42 | require.NoError(t, s.StartAndWaitReady(distributor, ingester, querier)) |
| 43 | |
| 44 | require.NoError(t, distributor.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_ring_members"}, e2e.WithLabelMatchers( |
| 45 | labels.MustNewMatcher(labels.MatchEqual, "name", "ingester"), |
| 46 | labels.MustNewMatcher(labels.MatchEqual, "state", "ACTIVE")))) |
| 47 | |
| 48 | c, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), querier.HTTPEndpoint(), "", "", userID) |
| 49 | require.NoError(t, err) |
| 50 | |
| 51 | now := time.Now() |
| 52 | series, expectedVector := generateSeries("series_1", now, prompb.Label{Name: "foo", Value: "bar"}) |
| 53 | |
| 54 | res, err := c.Push(series) |
| 55 | require.NoError(t, err) |
| 56 | require.Equal(t, 200, res.StatusCode) |
| 57 | |
| 58 | result, err := c.Query("series_1", now) |
| 59 | require.NoError(t, err) |
| 60 | require.Equal(t, model.ValVector, result.Type()) |
| 61 | assert.Equal(t, expectedVector, result.(model.Vector)) |
| 62 | }) |
| 63 | |
| 64 | t.Run(fmt.Sprintf("store gateway client/%s", compression), func(t *testing.T) { |
| 65 | const blockRangePeriod = 5 * time.Second |
| 66 | |
| 67 | s, err := e2e.NewScenario(networkName) |
| 68 | require.NoError(t, err) |
| 69 | defer s.Close() |
| 70 | |
| 71 | flags := mergeFlags(BlocksStorageFlags(), map[string]string{ |
| 72 | "-distributor.replication-factor": "1", |
| 73 | "-blocks-storage.tsdb.block-ranges-period": blockRangePeriod.String(), |
| 74 | "-blocks-storage.tsdb.ship-interval": "1s", |
| 75 | "-blocks-storage.bucket-store.sync-interval": "1s", |
| 76 | "-blocks-storage.tsdb.retention-period": ((blockRangePeriod * 2) - 1).String(), |
| 77 | "-store-gateway.sharding-enabled": "false", |
| 78 | "-querier.store-gateway-client.grpc-compression": compression, |
nothing calls this directly
no test coverage detected