Added hashmod support for store sharding.

Signed-off-by: Bartlomiej Plotka <[email protected]>

Author: bwplotka

pkg/block/fetcher.go

@@ -401,14 +401,25 @@ func NewLabelShardedMetaFilter(relabelConfig []*relabel.Config) *LabelShardedMet
 	return &LabelShardedMetaFilter{relabelConfig: relabelConfig}
 }
 
+const blockIDLabel = "__block_id"
+
 // Filter filters out blocks that filters blocks that have no labels after relabelling.
 func (f *LabelShardedMetaFilter) Filter(metas map[ulid.ULID]*metadata.Meta, synced GaugeLabeled, _ bool) {
+	lbls := make(labels.Labels, 1)
 	for id, m := range metas {
-		if processedLabels := relabel.Process(labels.FromMap(m.Thanos.Labels), f.relabelConfig...); processedLabels != nil {
-			continue
+		lbls[0] = labels.Label{Name: blockIDLabel, Value: id.String()}
+		lbls = lbls[:1]
+		if len(lbls) < len(m.Thanos.Labels) {
+			lbls = make([]labels.Label, 0, len(m.Thanos.Labels))
+		}
+		for k, v := range m.Thanos.Labels {
+			lbls = append(lbls, labels.Label{Name: k, Value: v})
+		}
+
+		if processedLabels := relabel.Process(lbls, f.relabelConfig...); processedLabels == nil {
+			synced.WithLabelValues(labelExcludedMeta).Inc()
+			delete(metas, id)
 		}
-		synced.WithLabelValues(labelExcludedMeta).Inc()
-		delete(metas, id)
 	}
 }
 

pkg/block/fetcher_test.go

@@ -278,7 +278,7 @@ func TestMetaFetcher_Fetch(t *testing.T) {
 	})
 }
 
-func TestLabelShardedMetaFilter_Filter(t *testing.T) {
+func TestLabelShardedMetaFilter_Filter_Basic(t *testing.T) {
 	relabelContentYaml := `
     - action: drop
       regex: "A"
@@ -340,6 +340,72 @@ func TestLabelShardedMetaFilter_Filter(t *testing.T) {
 
 }
 
+func TestLabelShardedMetaFilter_Filter_Hashmod(t *testing.T) {
+	relabelContentYamlFmt := `
+    - action: hashmod
+      source_labels: ["ff%s"]
+      target_label: shard
+      modulus: 3
+    - action: keep
+      source_labels: ["shard"]
+      regex: %d
+`
+	for i := 0; i < 3; i++ {
+		t.Run(fmt.Sprintf("%v", i), func(t *testing.T) {
+			var relabelConfig []*relabel.Config
+			testutil.Ok(t, yaml.Unmarshal([]byte(fmt.Sprintf(relabelContentYamlFmt, blockIDLabel, i)), &relabelConfig))
+
+			f := NewLabelShardedMetaFilter(relabelConfig)
+
+			input := map[ulid.ULID]*metadata.Meta{
+				ULID(1): {
+					Thanos: metadata.Thanos{
+						Labels: map[string]string{"cluster": "B", "message": "keepme"},
+					},
+				},
+				ULID(2): {
+					Thanos: metadata.Thanos{
+						Labels: map[string]string{"something": "A", "message": "keepme"},
+					},
+				},
+				ULID(3): {
+					Thanos: metadata.Thanos{
+						Labels: map[string]string{"cluster": "A", "message": "keepme"},
+					},
+				},
+				ULID(4): {
+					Thanos: metadata.Thanos{
+						Labels: map[string]string{"cluster": "A", "something": "B", "message": "keepme"},
+					},
+				},
+				ULID(5): {
+					Thanos: metadata.Thanos{
+						Labels: map[string]string{"cluster": "B"},
+					},
+				},
+				ULID(6): {
+					Thanos: metadata.Thanos{
+						Labels: map[string]string{"cluster": "B", "message": "keepme"},
+					},
+				},
+				ULID(7): {},
+				ULID(8): {}, ULID(8): {},
+				ULID(9): {},
+			}
+			expected := map[ulid.ULID]*metadata.Meta{
+				// ?
+			}
+
+			synced := prometheus.NewGaugeVec(prometheus.GaugeOpts{}, []string{"state"})
+			f.Filter(input, synced, false)
+
+			testutil.Equals(t, 3.0, promtest.ToFloat64(synced.WithLabelValues(labelExcludedMeta)))
+			testutil.Equals(t, expected, input)
+		})
+
+	}
+}
+
 func TestTimePartitionMetaFilter_Filter(t *testing.T) {
 	mint := time.Unix(0, 1*time.Millisecond.Nanoseconds())
 	maxt := time.Unix(0, 10*time.Millisecond.Nanoseconds())