From 4498e49d1a49922046c61ff9fcb977cf2b79b3ac Mon Sep 17 00:00:00 2001 From: eduardscaueru Date: Sun, 20 Oct 2024 15:49:34 +0300 Subject: [PATCH] Added new implementation that makes the distributor accept multiple HA pairs (cluster, replica) in the same requets/batch. This can be enabled with a new flag, accept_mixed_ha_samples, an will take effect only if accept_ha_samples is set to true. Fixed test by reducing the number of ingesters to 2 and replication factor to 2. Added config reference. Do not remove replica label if cluster label is not present. Added more HA mixed replicas tests with no cluster and replica labels and with cluster label only. Added e2e test for mixed HA samples in the same request. Refactored distributor mixed HA samples logic. Added experimental flag for accept_mixed_ha_samples. Handled ReplicasNotMatchError TooManyReplicaGroupsError differently. Remove replica label at validation if it is too large. Signed-off-by: eduardscaueru --- CHANGELOG.md | 1 + docs/configuration/config-file-reference.md | 6 + .../distributor_mixed_ha_samples_test.go | 141 +++++++++++++ pkg/distributor/distributor.go | 29 ++- pkg/distributor/distributor_test.go | 198 +++++++++++++++++- pkg/util/validation/limits.go | 7 + 6 files changed, 380 insertions(+), 2 deletions(-) create mode 100644 integration/distributor_mixed_ha_samples_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 82f0a9367e..904d1abc3a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ * [FEATURE] Ruler: Minimize chances of missed rule group evaluations that can occur due to OOM kills, bad underlying nodes, or due to an unhealthy ruler that appears in the ring as healthy. This feature is enabled via `-ruler.enable-ha-evaluation` flag. #6129 * [FEATURE] Store Gateway: Add an in-memory chunk cache. #6245 * [FEATURE] Chunk Cache: Support multi level cache and add metrics. #6249 +* [FEATURE] Distributor: Accept multiple HA Tracker pairs in the same request. #6256 * [ENHANCEMENT] Ingester: Add `blocks-storage.tsdb.wal-compression-type` to support zstd wal compression type. #6232 * [ENHANCEMENT] Query Frontend: Add info field to query response. #6207 * [ENHANCEMENT] Query Frontend: Add peakSample in query stats response. #6188 diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md index a841909e5c..a38c86bd0d 100644 --- a/docs/configuration/config-file-reference.md +++ b/docs/configuration/config-file-reference.md @@ -3163,6 +3163,12 @@ The `limits_config` configures default and per-tenant limits imposed by Cortex s # CLI flag: -distributor.ha-tracker.enable-for-all-users [accept_ha_samples: | default = false] +# [Experimental] Flag to enable handling of samples with mixed external labels +# identifying replicas in an HA Prometheus setup. Supported only if +# -distributor.ha-tracker.enable-for-all-users is true. +# CLI flag: -experimental.distributor.ha-tracker.mixed-ha-samples +[accept_mixed_ha_samples: | default = false] + # Prometheus label to look for in samples to identify a Prometheus HA cluster. # CLI flag: -distributor.ha-tracker.cluster [ha_cluster_label: | default = "cluster"] diff --git a/integration/distributor_mixed_ha_samples_test.go b/integration/distributor_mixed_ha_samples_test.go new file mode 100644 index 0000000000..839419916b --- /dev/null +++ b/integration/distributor_mixed_ha_samples_test.go @@ -0,0 +1,141 @@ +//go:build requires_docker +// +build requires_docker + +package integration + +import ( + "fmt" + "testing" + "time" + + "github.com/prometheus/common/model" + "github.com/prometheus/prometheus/prompb" + "github.com/stretchr/testify/require" + + "github.com/cortexproject/cortex/integration/e2e" + e2edb "github.com/cortexproject/cortex/integration/e2e/db" + "github.com/cortexproject/cortex/integration/e2ecortex" +) + +func TestDistriubtorAcceptMixedHASamplesRunningInMicroservicesMode(t *testing.T) { + const blockRangePeriod = 5 * time.Minute + + t.Run(fmt.Sprintf("%s", "Distriubtor accept mixed HA samples in the same request"), func(t *testing.T) { + s, err := e2e.NewScenario(networkName) + require.NoError(t, err) + defer s.Close() + + // Start dependencies. + consul := e2edb.NewConsul() + etcd := e2edb.NewETCD() + minio := e2edb.NewMinio(9000, bucketName) + require.NoError(t, s.StartAndWaitReady(consul, etcd, minio)) + + // Configure the querier to only look in ingester + // and enbale distributor ha tracker with mixed samples. + distributorFlags := map[string]string{ + "-distributor.ha-tracker.enable": "true", + "-distributor.ha-tracker.enable-for-all-users": "true", + "-experimental.distributor.ha-tracker.mixed-ha-samples": "true", + "-distributor.ha-tracker.cluster": "cluster", + "-distributor.ha-tracker.replica": "__replica__", + "-distributor.ha-tracker.store": "etcd", + "-distributor.ha-tracker.etcd.endpoints": "etcd:2379", + } + querierFlags := mergeFlags(BlocksStorageFlags(), map[string]string{ + "-querier.query-store-after": (1 * time.Hour).String(), + }) + flags := mergeFlags(BlocksStorageFlags(), map[string]string{ + "-blocks-storage.tsdb.block-ranges-period": blockRangePeriod.String(), + "-blocks-storage.tsdb.ship-interval": "5s", + "-blocks-storage.tsdb.retention-period": ((blockRangePeriod * 2) - 1).String(), + "-blocks-storage.bucket-store.max-chunk-pool-bytes": "1", + }) + + // Start Cortex components. + distributor := e2ecortex.NewDistributor("distributor", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), distributorFlags, "") + ingester := e2ecortex.NewIngester("ingester", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "") + require.NoError(t, s.StartAndWaitReady(distributor, ingester)) + + // Wait until both the distributor and ingester have updated the ring. + require.NoError(t, distributor.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total")) + + distributorClient, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), "", "", "", "user-1") + require.NoError(t, err) + + // Push some series to Cortex. + series1Timestamp := time.Now() + series2Timestamp := series1Timestamp.Add(-2 * time.Second) + series3Timestamp := series1Timestamp.Add(-4 * time.Second) + series4Timestamp := series1Timestamp.Add(-6 * time.Second) + series5Timestamp := series1Timestamp.Add(-8 * time.Second) + series6Timestamp := series1Timestamp.Add(-10 * time.Second) + series7Timestamp := series1Timestamp.Add(-12 * time.Second) + series1, _ := generateSeries("foo", series1Timestamp, prompb.Label{Name: "__replica__", Value: "replica0"}, prompb.Label{Name: "cluster", Value: "cluster0"}) + series2, _ := generateSeries("foo", series2Timestamp, prompb.Label{Name: "__replica__", Value: "replica1"}, prompb.Label{Name: "cluster", Value: "cluster0"}) + series3, _ := generateSeries("foo", series3Timestamp, prompb.Label{Name: "__replica__", Value: "replica0"}, prompb.Label{Name: "cluster", Value: "cluster1"}) + series4, _ := generateSeries("foo", series4Timestamp, prompb.Label{Name: "__replica__", Value: "replica1"}, prompb.Label{Name: "cluster", Value: "cluster1"}) + series5, _ := generateSeries("foo", series5Timestamp, prompb.Label{Name: "__replica__", Value: "replicaNoCluster"}) + series6, _ := generateSeries("foo", series6Timestamp, prompb.Label{Name: "cluster", Value: "clusterNoReplica"}) + series7, _ := generateSeries("foo", series7Timestamp, prompb.Label{Name: "other", Value: "label"}) + + res, err := distributorClient.Push([]prompb.TimeSeries{series1[0], series2[0], series3[0], series4[0], series5[0], series6[0], series7[0]}) + require.NoError(t, err) + require.Equal(t, 200, res.StatusCode) + + // Wait until the samples have been deduped. + require.NoError(t, distributor.WaitSumMetrics(e2e.Equals(2), "cortex_distributor_deduped_samples_total")) + require.NoError(t, distributor.WaitSumMetrics(e2e.Equals(3), "cortex_distributor_non_ha_samples_received_total")) + + // Start the querier and store-gateway, and configure them to frequently sync blocks fast enough to trigger consistency check. + storeGateway := e2ecortex.NewStoreGateway("store-gateway", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "") + querier := e2ecortex.NewQuerier("querier", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), mergeFlags(querierFlags, flags), "") + require.NoError(t, s.StartAndWaitReady(querier, storeGateway)) + + // Wait until the querier and store-gateway have updated the ring, and wait until the blocks are old enough for consistency check + require.NoError(t, querier.WaitSumMetrics(e2e.Equals(512*2), "cortex_ring_tokens_total")) + require.NoError(t, storeGateway.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total")) + + // Query back the series. + querierClient, err := e2ecortex.NewClient("", querier.HTTPEndpoint(), "", "", "user-1") + require.NoError(t, err) + + // Query back the series (only in the ingesters). + result, err := querierClient.Query("foo[5m]", series1Timestamp) + require.NoError(t, err) + + require.Equal(t, model.ValMatrix, result.Type()) + m := result.(model.Matrix) + require.Equal(t, 5, m.Len()) + numValidHA := 0 + numNonHA := 0 + for _, ss := range m { + replicaLabel, okReplica := ss.Metric["__replica__"] + if okReplica { + require.Equal(t, string(replicaLabel), "replicaNoCluster") + } + clusterLabel, okCluster := ss.Metric["cluster"] + if okCluster { + require.Equal(t, string(clusterLabel) == "cluster1" || string(clusterLabel) == "cluster0" || string(clusterLabel) == "clusterNoReplica", true) + if clusterLabel == "cluster1" || clusterLabel == "cluster0" { + numValidHA++ + } + } + if (okReplica && !okCluster && replicaLabel == "replicaNoCluster") || (okCluster && !okReplica && clusterLabel == "clusterNoReplica") || (!okCluster && !okReplica) { + numNonHA++ + } + require.NotEmpty(t, ss.Values) + for _, v := range ss.Values { + require.NotEmpty(t, v) + } + } + require.Equal(t, numNonHA, 3) + require.Equal(t, numValidHA, 2) + + // Ensure no service-specific metrics prefix is used by the wrong service. + assertServiceMetricsPrefixes(t, Distributor, distributor) + assertServiceMetricsPrefixes(t, Ingester, ingester) + assertServiceMetricsPrefixes(t, StoreGateway, storeGateway) + assertServiceMetricsPrefixes(t, Querier, querier) + }) +} diff --git a/pkg/distributor/distributor.go b/pkg/distributor/distributor.go index 699035e672..6af8296f5e 100644 --- a/pkg/distributor/distributor.go +++ b/pkg/distributor/distributor.go @@ -652,7 +652,7 @@ func (d *Distributor) Push(ctx context.Context, req *cortexpb.WriteRequest) (*co // Cache user limit with overrides so we spend less CPU doing locking. See issue #4904 limits := d.limits.GetOverridesForUser(userID) - if limits.AcceptHASamples && len(req.Timeseries) > 0 { + if limits.AcceptHASamples && len(req.Timeseries) > 0 && !limits.AcceptMixedHASamples { cluster, replica := findHALabels(limits.HAReplicaLabel, limits.HAClusterLabel, req.Timeseries[0].Labels) removeReplica, err = d.checkSample(ctx, userID, cluster, replica, limits) if err != nil { @@ -859,6 +859,33 @@ func (d *Distributor) prepareSeriesKeys(ctx context.Context, req *cortexpb.Write // check each sample and discard if outside limits. skipLabelNameValidation := d.cfg.SkipLabelNameValidation || req.GetSkipLabelNameValidation() for _, ts := range req.Timeseries { + if limits.AcceptHASamples && limits.AcceptMixedHASamples { + cluster, replica := findHALabels(limits.HAReplicaLabel, limits.HAClusterLabel, ts.Labels) + if cluster != "" && replica != "" { + removeReplicaTs, err := d.checkSample(ctx, userID, cluster, replica, limits) + if err != nil { + // discard sample + if errors.Is(err, ha.ReplicasNotMatchError{}) { + // These samples have been deduped. + d.dedupedSamples.WithLabelValues(userID, cluster).Add(float64(len(ts.Samples) + len(ts.Histograms))) + } + if errors.Is(err, ha.TooManyReplicaGroupsError{}) { + d.validateMetrics.DiscardedSamples.WithLabelValues(validation.TooManyHAClusters, userID).Add(float64(len(ts.Samples) + len(ts.Histograms))) + } + + continue + } + if !removeReplicaTs { + removeReplica = false // valid HA sample, but replica is too large. Remove it at validation step. + } else { + removeReplica = true // valid HA sample + } + } else { + removeReplica = false // non HA sample + d.nonHASamples.WithLabelValues(userID).Add(float64(len(ts.Samples) + len(ts.Histograms))) + } + } + // Use timestamp of latest sample in the series. If samples for series are not ordered, metric for user may be wrong. if len(ts.Samples) > 0 { latestSampleTimestampMs = max(latestSampleTimestampMs, ts.Samples[len(ts.Samples)-1].TimestampMs) diff --git a/pkg/distributor/distributor_test.go b/pkg/distributor/distributor_test.go index 28cf8b1dc6..6d49f24933 100644 --- a/pkg/distributor/distributor_test.go +++ b/pkg/distributor/distributor_test.go @@ -1031,6 +1031,102 @@ func TestDistributor_PushHAInstances(t *testing.T) { } } +func TestDistributor_PushMixedHAInstances(t *testing.T) { + t.Parallel() + ctx := user.InjectOrgID(context.Background(), "user") + + for i, tc := range []struct { + enableTracker bool + acceptMixedHASamples bool + samples int + expectedResponse *cortexpb.WriteResponse + expectedCode int32 + }{ + { + enableTracker: true, + acceptMixedHASamples: true, + samples: 5, + expectedResponse: emptyResponse, + expectedCode: 202, + }, + } { + for _, shardByAllLabels := range []bool{true} { + tc := tc + shardByAllLabels := shardByAllLabels + for _, enableHistogram := range []bool{false} { + enableHistogram := enableHistogram + t.Run(fmt.Sprintf("[%d](shardByAllLabels=%v, histogram=%v)", i, shardByAllLabels, enableHistogram), func(t *testing.T) { + t.Parallel() + var limits validation.Limits + flagext.DefaultValues(&limits) + limits.AcceptHASamples = true + limits.AcceptMixedHASamples = tc.acceptMixedHASamples + limits.MaxLabelValueLength = 25 + + ds, ingesters, _, _ := prepare(t, prepConfig{ + numIngesters: 2, + happyIngesters: 2, + numDistributors: 1, + replicationFactor: 2, + shardByAllLabels: shardByAllLabels, + limits: &limits, + enableTracker: tc.enableTracker, + }) + + d := ds[0] + + request := makeWriteRequestHAMixedSamples(tc.samples, enableHistogram) + response, _ := d.Push(ctx, request) + assert.Equal(t, tc.expectedResponse, response) + + for i := range ingesters { + timeseries := ingesters[i].series() + assert.Equal(t, 5, len(timeseries)) + clusters := make(map[string]int) + replicas := make(map[string]int) + for _, v := range timeseries { + replicaLabel := "" + clusterLabel := "" + for _, label := range v.Labels { + if label.Name == "__replica__" { + replicaLabel = label.Value + _, ok := replicas[label.Value] + if !ok { + replicas[label.Value] = 1 + } else { + assert.Fail(t, fmt.Sprintf("Two timeseries with same replica label, %s, were found, but only one should be present", label.Value)) + } + } + if label.Name == "cluster" { + clusterLabel = label.Value + _, ok := clusters[label.Value] + if !ok { + clusters[label.Value] = 1 + } else { + assert.Fail(t, fmt.Sprintf("Two timeseries with same cluster label, %s, were found, but only one should be present", label.Value)) + } + } + } + if clusterLabel == "" && replicaLabel != "" { + assert.Equal(t, "replicaNoCluster", replicaLabel) + } + assert.Equal(t, tc.samples, len(v.Samples)) + } + assert.Equal(t, 3, len(clusters)) + for _, nr := range clusters { + assert.Equal(t, true, nr == 1) + } + assert.Equal(t, 1, len(replicas)) + for _, nr := range clusters { + assert.Equal(t, true, nr == 1) + } + } + }) + } + } + } +} + func TestDistributor_PushQuery(t *testing.T) { t.Parallel() const shuffleShardSize = 5 @@ -2831,7 +2927,7 @@ func prepare(tb testing.TB, cfg prepConfig) ([]*Distributor, []*mockIngester, [] EnableHATracker: true, KVStore: kv.Config{Mock: mock}, UpdateTimeout: 100 * time.Millisecond, - FailoverTimeout: time.Second, + FailoverTimeout: time.Hour, } cfg.limits.HAMaxClusters = 100 } @@ -2950,6 +3046,106 @@ func makeWriteRequestHA(samples int, replica, cluster string, histogram bool) *c return request } +func makeWriteRequestHAMixedSamples(samples int, histogram bool) *cortexpb.WriteRequest { + request := &cortexpb.WriteRequest{} + + for _, haPair := range []struct { + cluster string + replica string + }{ + { + cluster: "cluster0", + replica: "replica0", + }, + { + cluster: "cluster0", + replica: "replica1", + }, + { + cluster: "cluster1", + replica: "replica0", + }, + { + cluster: "cluster1", + replica: "replica1", + }, + { + cluster: "", + replica: "replicaNoCluster", + }, + { + cluster: "clusterNoReplica", + replica: "", + }, + { + cluster: "", + replica: "", + }, + } { + cluster := haPair.cluster + replica := haPair.replica + var ts cortexpb.PreallocTimeseries + if cluster == "" && replica == "" { + ts = cortexpb.PreallocTimeseries{ + TimeSeries: &cortexpb.TimeSeries{ + Labels: []cortexpb.LabelAdapter{ + {Name: "__name__", Value: "foo"}, + {Name: "bar", Value: "baz"}, + }, + }, + } + } else if cluster == "" && replica != "" { + ts = cortexpb.PreallocTimeseries{ + TimeSeries: &cortexpb.TimeSeries{ + Labels: []cortexpb.LabelAdapter{ + {Name: "__name__", Value: "foo"}, + {Name: "__replica__", Value: replica}, + {Name: "bar", Value: "baz"}, + }, + }, + } + } else if cluster != "" && replica == "" { + ts = cortexpb.PreallocTimeseries{ + TimeSeries: &cortexpb.TimeSeries{ + Labels: []cortexpb.LabelAdapter{ + {Name: "__name__", Value: "foo"}, + {Name: "bar", Value: "baz"}, + {Name: "cluster", Value: cluster}, + }, + }, + } + } else { + ts = cortexpb.PreallocTimeseries{ + TimeSeries: &cortexpb.TimeSeries{ + Labels: []cortexpb.LabelAdapter{ + {Name: "__name__", Value: "foo"}, + {Name: "__replica__", Value: replica}, + {Name: "bar", Value: "baz"}, + {Name: "cluster", Value: cluster}, + }, + }, + } + } + if histogram { + ts.Histograms = []cortexpb.Histogram{ + cortexpb.HistogramToHistogramProto(int64(samples), tsdbutil.GenerateTestHistogram(samples)), + } + } else { + var s = make([]cortexpb.Sample, 0) + for i := 0; i < samples; i++ { + sample := cortexpb.Sample{ + Value: float64(i), + TimestampMs: int64(i), + } + s = append(s, sample) + } + ts.Samples = s + } + request.Timeseries = append(request.Timeseries, ts) + } + return request +} + func makeWriteRequestExemplar(seriesLabels []string, timestamp int64, exemplarLabels []string) *cortexpb.WriteRequest { return &cortexpb.WriteRequest{ Timeseries: []cortexpb.PreallocTimeseries{ diff --git a/pkg/util/validation/limits.go b/pkg/util/validation/limits.go index 87173abb13..42560ccaed 100644 --- a/pkg/util/validation/limits.go +++ b/pkg/util/validation/limits.go @@ -119,6 +119,7 @@ type Limits struct { IngestionRateStrategy string `yaml:"ingestion_rate_strategy" json:"ingestion_rate_strategy"` IngestionBurstSize int `yaml:"ingestion_burst_size" json:"ingestion_burst_size"` AcceptHASamples bool `yaml:"accept_ha_samples" json:"accept_ha_samples"` + AcceptMixedHASamples bool `yaml:"accept_mixed_ha_samples" json:"accept_mixed_ha_samples"` HAClusterLabel string `yaml:"ha_cluster_label" json:"ha_cluster_label"` HAReplicaLabel string `yaml:"ha_replica_label" json:"ha_replica_label"` HAMaxClusters int `yaml:"ha_max_clusters" json:"ha_max_clusters"` @@ -220,6 +221,7 @@ func (l *Limits) RegisterFlags(f *flag.FlagSet) { f.StringVar(&l.IngestionRateStrategy, "distributor.ingestion-rate-limit-strategy", "local", "Whether the ingestion rate limit should be applied individually to each distributor instance (local), or evenly shared across the cluster (global).") f.IntVar(&l.IngestionBurstSize, "distributor.ingestion-burst-size", 50000, "Per-user allowed ingestion burst size (in number of samples).") f.BoolVar(&l.AcceptHASamples, "distributor.ha-tracker.enable-for-all-users", false, "Flag to enable, for all users, handling of samples with external labels identifying replicas in an HA Prometheus setup.") + f.BoolVar(&l.AcceptMixedHASamples, "experimental.distributor.ha-tracker.mixed-ha-samples", false, "Flag to enable handling of samples with mixed external labels identifying replicas in an HA Prometheus setup. Supported only if -distributor.ha-tracker.enable-for-all-users is true.") f.StringVar(&l.HAClusterLabel, "distributor.ha-tracker.cluster", "cluster", "Prometheus label to look for in samples to identify a Prometheus HA cluster.") f.StringVar(&l.HAReplicaLabel, "distributor.ha-tracker.replica", "__replica__", "Prometheus label to look for in samples to identify a Prometheus HA replica.") f.IntVar(&l.HAMaxClusters, "distributor.ha-tracker.max-clusters", 0, "Maximum number of clusters that HA tracker will keep track of for single user. 0 to disable the limit.") @@ -547,6 +549,11 @@ func (o *Overrides) AcceptHASamples(userID string) bool { return o.GetOverridesForUser(userID).AcceptHASamples } +// AcceptMixedHASamples returns whether the distributor should track and accept mixed samples from HA replicas for this user. +func (o *Overrides) AcceptMixedHASamples(userID string) bool { + return o.GetOverridesForUser(userID).AcceptMixedHASamples +} + // HAClusterLabel returns the cluster label to look for when deciding whether to accept a sample from a Prometheus HA replica. func (o *Overrides) HAClusterLabel(userID string) string { return o.GetOverridesForUser(userID).HAClusterLabel