Parquet: Skip parquet conversion for blocks with too many labels

docs/configuration/config-file-reference.md

@@ -4779,6 +4779,12 @@ query_rejection:
 # CLI flag: -parquet-converter.sort-columns
 [parquet_converter_sort_columns: <list of string> | default = []]
 
+# [Experimental] Maximum number of distinct label names allowed in a TSDB block
+# for parquet conversion. If exceeded, the converter writes a no-convert marker.
+# 0 to disable.
+# CLI flag: -parquet-converter.max-block-label-names
+[parquet_converter_max_block_label_names: <int> | default = 0]
+
 # S3 server-side encryption type. Required to enable server-side encryption
 # overrides for a specific tenant. If not set, the default S3 client settings
 # are used.

docs/configuration/v1-guarantees.md

@@ -9,9 +9,9 @@ For the v1.0 release, we want to provide the following guarantees:
 
 ## Flags, Config and minor version upgrades
 
-Upgrading cortex from one minor version to the next should "just work"; that being said, we don't want to bump the major version every time we remove a flag, so we will will keep deprecated flags around for 2 minor release.  There is a metric (`deprecated_flags_inuse_total`) you can alert on to find out if you're using a deprecated  flag.
+Upgrading cortex from one minor version to the next should "just work"; that being said, we don't want to bump the major version every time we remove a flag, so we will will keep deprecated flags around for 2 minor release. There is a metric (`deprecated_flags_inuse_total`) you can alert on to find out if you're using a deprecated flag.
 
-Similarly to flags, minor version upgrades using config files should "just work".  If we do need to change config, we will keep the old way working for two minor version.  There will be a metric you can alert on for this too.
+Similarly to flags, minor version upgrades using config files should "just work". If we do need to change config, we will keep the old way working for two minor version. There will be a metric you can alert on for this too.
 
 These guarantees don't apply for [experimental features](#experimental-features).
 
@@ -25,11 +25,11 @@ Cortex strives to be 100% API compatible with Prometheus (under `/prometheus/*`
 
 - Additional API endpoints for creating, removing and modifying alerts and recording rules.
 - Additional API around pushing metrics (under `/api/push`).
-- Additional API endpoints for management of Cortex itself, such as the ring.  These APIs are not part of the any compatibility guarantees.
+- Additional API endpoints for management of Cortex itself, such as the ring. These APIs are not part of the any compatibility guarantees.
 
 ## Experimental features
 
-Cortex is an actively developed project and we want to encourage the introduction of new features and capability.  As such, not everything in each release of Cortex is considered "production-ready". We don't provide any backwards compatibility guarantees on these and the config and flags might break.
+Cortex is an actively developed project and we want to encourage the introduction of new features and capability. As such, not everything in each release of Cortex is considered "production-ready". We don't provide any backwards compatibility guarantees on these and the config and flags might break.
 
 Currently experimental features are:
 
@@ -150,3 +150,4 @@ Currently experimental features are:
   - `-ingester.head-queried-series-metrics-windows` time windows to report (default: 2h)
   - `-ingester.head-queried-series-metrics-window-duration` HLL sub-window size
   - `-ingester.head-queried-series-metrics-sample-rate` query sampling rate
+- Parquet Converter: `-parquet-converter.max-block-label-names` (int) - If enabled, adds a no-convert mark and skips blocks with too many labels.

integration/parquet_converter_test.go

@@ -0,0 +1,142 @@
+//go:build integration
+
+package integration
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/stretchr/testify/require"
+	"github.com/thanos-io/objstore"
+	"github.com/thanos-io/thanos/pkg/block"
+	"github.com/thanos-io/thanos/pkg/block/metadata"
+
+	"github.com/cortexproject/cortex/integration/e2e"
+	e2ecache "github.com/cortexproject/cortex/integration/e2e/cache"
+	e2edb "github.com/cortexproject/cortex/integration/e2e/db"
+	"github.com/cortexproject/cortex/integration/e2ecortex"
+	"github.com/cortexproject/cortex/pkg/storage/bucket"
+	"github.com/cortexproject/cortex/pkg/storage/tsdb"
+	"github.com/cortexproject/cortex/pkg/util/log"
+	cortex_testutil "github.com/cortexproject/cortex/pkg/util/test"
+)
+
+func TestParquetConverter_NoConvertMarkWithTooManyLabels(t *testing.T) {
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	consul := e2edb.NewConsulWithName("consul")
+	memcached := e2ecache.NewMemcached()
+	require.NoError(t, s.StartAndWaitReady(consul, memcached))
+
+	baseFlags := mergeFlags(AlertmanagerLocalFlags(), BlocksStorageFlags())
+	flags := mergeFlags(
+		baseFlags,
+		map[string]string{
+			"-target": "all,parquet-converter",
+			"-blocks-storage.tsdb.block-ranges-period":                             "1m,24h",
+			"-blocks-storage.tsdb.ship-interval":                                   "1s",
+			"-blocks-storage.bucket-store.sync-interval":                           "1s",
+			"-blocks-storage.bucket-store.metadata-cache.bucket-index-content-ttl": "1s",
+			"-blocks-storage.bucket-store.bucket-index.idle-timeout":               "1s",
+			"-blocks-storage.bucket-store.bucket-index.enabled":                    "true",
+			"-blocks-storage.bucket-store.index-cache.backend":                     tsdb.IndexCacheBackendInMemory,
+			// compactor
+			"-compactor.cleanup-interval": "1s",
+			// Ingester.
+			"-ring.store":      "consul",
+			"-consul.hostname": consul.NetworkHTTPEndpoint(),
+			// Distributor.
+			"-distributor.replication-factor": "1",
+			// Store-gateway.
+			"-store-gateway.sharding-enabled":   "false",
+			"--querier.store-gateway-addresses": "nonExistent", // Make sure we do not call Store gateways
+			// alert manager
+			"-alertmanager.web.external-url": "http://localhost/alertmanager",
+			// Enable vertical sharding.
+			"-frontend.query-vertical-shard-size": "3",
+			"-frontend.max-cache-freshness":       "1m",
+			// enable experimental promQL funcs
+			"-querier.enable-promql-experimental-functions": "true",
+			// parquet-converter
+			"-parquet-converter.ring.consul.hostname":  consul.NetworkHTTPEndpoint(),
+			"-parquet-converter.conversion-interval":   "1s",
+			"-parquet-converter.enabled":               "true",
+			"-parquet-converter.max-block-label-names": "1",
+			// Querier
+			"-querier.enable-parquet-queryable": "true",
+			// Enable cache for parquet labels and chunks
+			"-blocks-storage.bucket-store.parquet-labels-cache.backend":             "inmemory,memcached",
+			"-blocks-storage.bucket-store.parquet-labels-cache.memcached.addresses": "dns+" + memcached.NetworkEndpoint(e2ecache.MemcachedPort),
+			"-blocks-storage.bucket-store.chunks-cache.backend":                     "inmemory,memcached",
+			"-blocks-storage.bucket-store.chunks-cache.memcached.addresses":         "dns+" + memcached.NetworkEndpoint(e2ecache.MemcachedPort),
+		},
+	)
+
+	// make alert manager config dir
+	require.NoError(t, writeFileToSharedDir(s, "alertmanager_configs", []byte{}))
+
+	ctx := context.Background()
+	rnd := rand.New(rand.NewSource(time.Now().Unix()))
+	dir := filepath.Join(s.SharedDir(), "data")
+	lbls := []labels.Labels{
+		labels.FromStrings("__name__", "test_series_a", "job", "test"),
+	}
+
+	numSamples := 60
+	scrapeInterval := time.Minute
+	now := time.Now()
+	start := now.Add(-time.Hour * 24)
+	end := now.Add(-time.Hour)
+
+	minio := e2edb.NewMinio(9000, flags["-blocks-storage.s3.bucket-name"])
+	require.NoError(t, s.StartAndWaitReady(minio))
+
+	cortex := e2ecortex.NewSingleBinary("cortex", flags, "")
+	require.NoError(t, s.StartAndWaitReady(cortex))
+	storage, err := e2ecortex.NewS3ClientForMinio(minio, flags["-blocks-storage.s3.bucket-name"])
+	require.NoError(t, err)
+	bkt := bucket.NewUserBucketClient("user-1", storage.GetBucket(), nil)
+
+	id, err := e2e.CreateBlock(ctx, rnd, dir, lbls, numSamples,
+		start.UnixMilli(),
+		end.UnixMilli(),
+		scrapeInterval.Milliseconds(), 10,
+	)
+	require.NoError(t, err)
+
+	err = block.Upload(ctx, log.Logger, bkt, filepath.Join(dir, id.String()), metadata.NoneFunc)
+	require.NoError(t, err)
+
+	// Wait for the converter to write the no-convert marker
+	cortex_testutil.Poll(t, 30*time.Second, true, func() interface{} {
+		noConvertMarkerPath := fmt.Sprintf("%s/parquet-no-convert-mark.json", id.String())
+		found := false
+		err := bkt.Iter(ctx, "", func(name string) error {
+			if name == noConvertMarkerPath {
+				found = true
+			}
+			return nil
+		}, objstore.WithRecursiveIter())
+		require.NoError(t, err)
+		return found
+	})
+
+	// confirm the conversion did not happen (check both paths)
+	blockID := id.String()
+	markerPaths := []string{
+		fmt.Sprintf("%s/parquet-converter-mark.json", blockID),
+		fmt.Sprintf("parquet-markers/%s-parquet-converter-mark.json", blockID),
+	}
+	for _, markerPath := range markerPaths {
+		exists, err := bkt.Exists(ctx, markerPath)
+		require.NoError(t, err)
+		require.False(t, exists, "converter mark should not exist at %s", markerPath)
+	}
+}

pkg/parquetconverter/converter.go

@@ -49,6 +49,9 @@ const (
 	ringKey = "parquet-converter"
 
 	converterMetaPrefix = "converter-meta-"
+
+	parquetConverterDataColumnDuration = time.Hour * 8
+	parquetConverterSystemColumnCount  = 2 // s_col_indexes and s_series_hash.
 )
 
 var RingOp = ring.NewOp([]ring.InstanceState{ring.ACTIVE}, nil)
@@ -139,7 +142,7 @@ func newConverter(cfg Config, bkt objstore.InstrumentedBucket, storageCfg cortex
 		metrics:        newMetrics(registerer),
 		bkt:            bkt,
 		baseConverterOptions: []convert.ConvertOption{
-			convert.WithColDuration(time.Hour * 8),
+			convert.WithColDuration(parquetConverterDataColumnDuration),
 			convert.WithRowGroupSize(cfg.MaxRowsPerRowGroup),
 		},
 	}
@@ -388,14 +391,32 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin
 		}
 
 		// We don't convert blocks again if they already have a valid converter mark.
-		if cortex_parquet.ValidConverterMarkVersion(marker.Version) {
+		if cortex_parquet.ValidNoConvertMarkVersion(marker.Version) {
+			level.Debug(logger).Log("msg", "skipping block, no-convert marker already exists", "block", b.ULID.String())
+			c.metrics.skippedBlocks.WithLabelValues(userID, cortex_parquet.NoConvertReasonTooManyLabels).Inc()
 			continue
 		}
 
 		if !cortex_parquet.ShouldConvertBlockToParquet(b.MinTime, b.MaxTime, c.blockRanges) {
 			continue
 		}
 
+		configuredMaxBlockLabelNames := c.limits.ParquetConverterMaxBlockLabelNames(userID)
+		maxBlockLabelNames := effectiveMaxBlockLabelNames(configuredMaxBlockLabelNames, b.MinTime, b.MaxTime)
+
+		// If the threshold is enabled, check for no-convert mark
+		if configuredMaxBlockLabelNames > 0 {
+
+			noConvertMark, err := cortex_parquet.ReadNoConvertMark(ctx, b.ULID, uBucket, logger)
+			if err != nil {
+				level.Error(logger).Log("msg", "failed to read parquet no-convert marker", "block", b.ULID.String(), "err", err)
+				continue
+			}
+			if cortex_parquet.ValidNoConvertMarkVersion(noConvertMark.Version) {
+				continue
+			}
+		}
+
 		if err := os.RemoveAll(c.compactRootDir()); err != nil {
 			level.Error(logger).Log("msg", "failed to remove work directory", "path", c.compactRootDir(), "err", err)
 			if c.checkConvertError(userID, err) {
@@ -425,6 +446,33 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin
 			continue
 		}
 
+		if configuredMaxBlockLabelNames > 0 {
+			labelNames, err := tsdbBlock.LabelNames(ctx)
+			if err != nil {
+				_ = tsdbBlock.Close()
+				level.Error(logger).Log("msg", "failed to get label names", "block", b.ULID.String(), "err", err)
+				if c.checkConvertError(userID, err) {
+					return err
+				}
+				continue
+			}
+			labelNamesCount := len(labelNames)
+			if labelNamesCount > maxBlockLabelNames {
+				if err := cortex_parquet.WriteNoConvertMark(ctx, b.ULID, uBucket, labelNamesCount, maxBlockLabelNames); err != nil {
+					_ = tsdbBlock.Close()
+					level.Error(logger).Log("msg", "failed to write parquet no-convert marker", "block", b.ULID.String(), "err", err)
+					if c.checkConvertError(userID, err) {
+						return err
+					}
+					continue
+				}
+				level.Debug(logger).Log("msg", "skipping parquet conversion for block with too many label names", "block", b.ULID.String(), "label_names", labelNamesCount, "limit", maxBlockLabelNames)
+				c.metrics.skippedBlocks.WithLabelValues(userID, cortex_parquet.NoConvertReasonTooManyLabels).Inc()
+				_ = tsdbBlock.Close()
+				continue
+			}
+		}
+
 		level.Info(logger).Log("msg", "converting block", "block", b.ULID.String(), "dir", bdir)
 		start := time.Now()
 
@@ -486,6 +534,25 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin
 	return nil
 }
 
+func effectiveMaxBlockLabelNames(configuredMaxBlockLabelNames int, mint, maxt int64) int {
+	if configuredMaxBlockLabelNames <= 0 {
+		return configuredMaxBlockLabelNames
+	}
+
+	dataColumnCount := 0
+	if maxt >= mint {
+		dataColumnCount = int((maxt-mint)/parquetConverterDataColumnDuration.Milliseconds()) + 1
+	}
+
+	// Reserve for s_col_indexes, s_series_hash, and generated s_data_* columns.
+	maxBlockLabelNames := max(parquet.MaxColumnIndex-parquetConverterSystemColumnCount-dataColumnCount, 0)
+
+	if configuredMaxBlockLabelNames > maxBlockLabelNames {
+		return maxBlockLabelNames
+	}
+	return configuredMaxBlockLabelNames
+}
+
 func (c *Converter) checkConvertError(userID string, err error) (terminate bool) {
 	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || c.isCausedByPermissionDenied(err) {
 		terminate = true