adding tests / hiding documentation for now

Signed-off-by: alanprot <[email protected]>

Author: alanprot

docs/configuration/config-file-reference.md

@@ -152,96 +152,6 @@ api:
 # The compactor_config configures the compactor for the blocks storage.
 [compactor: <compactor_config>]
 
-parquet_converter:
-  # Comma separated list of tenants that can be converted. If specified, only
-  # these tenants will be converted, otherwise all tenants can be converted.
-  # CLI flag: -parquet-converter.enabled-tenants
-  [enabled_tenants: <string> | default = ""]
-
-  # Comma separated list of tenants that cannot converted.
-  # CLI flag: -parquet-converter.disabled-tenants
-  [disabled_tenants: <string> | default = ""]
-
-  ring:
-    kvstore:
-      # Backend storage to use for the ring. Supported values are: consul, etcd,
-      # inmemory, memberlist, multi.
-      # CLI flag: -parquet-converter.ring.store
-      [store: <string> | default = "consul"]
-
-      # The prefix for the keys in the store. Should end with a /.
-      # CLI flag: -parquet-converter.ring.prefix
-      [prefix: <string> | default = "collectors/"]
-
-      dynamodb:
-        # Region to access dynamodb.
-        # CLI flag: -parquet-converter.ring.dynamodb.region
-        [region: <string> | default = ""]
-
-        # Table name to use on dynamodb.
-        # CLI flag: -parquet-converter.ring.dynamodb.table-name
-        [table_name: <string> | default = ""]
-
-        # Time to expire items on dynamodb.
-        # CLI flag: -parquet-converter.ring.dynamodb.ttl-time
-        [ttl: <duration> | default = 0s]
-
-        # Time to refresh local ring with information on dynamodb.
-        # CLI flag: -parquet-converter.ring.dynamodb.puller-sync-time
-        [puller_sync_time: <duration> | default = 1m]
-
-        # Maximum number of retries for DDB KV CAS.
-        # CLI flag: -parquet-converter.ring.dynamodb.max-cas-retries
-        [max_cas_retries: <int> | default = 10]
-
-        # Timeout of dynamoDbClient requests. Default is 2m.
-        # CLI flag: -parquet-converter.ring.dynamodb.timeout
-        [timeout: <duration> | default = 2m]
-
-      # The consul_config configures the consul client.
-      # The CLI flags prefix for this block config is: parquet-converter.ring
-      [consul: <consul_config>]
-
-      # The etcd_config configures the etcd client.
-      # The CLI flags prefix for this block config is: parquet-converter.ring
-      [etcd: <etcd_config>]
-
-      multi:
-        # Primary backend storage used by multi-client.
-        # CLI flag: -parquet-converter.ring.multi.primary
-        [primary: <string> | default = ""]
-
-        # Secondary backend storage used by multi-client.
-        # CLI flag: -parquet-converter.ring.multi.secondary
-        [secondary: <string> | default = ""]
-
-        # Mirror writes to secondary store.
-        # CLI flag: -parquet-converter.ring.multi.mirror-enabled
-        [mirror_enabled: <boolean> | default = false]
-
-        # Timeout for storing value to secondary store.
-        # CLI flag: -parquet-converter.ring.multi.mirror-timeout
-        [mirror_timeout: <duration> | default = 2s]
-
-    # Period at which to heartbeat to the ring. 0 = disabled.
-    # CLI flag: -parquet-converter.ring.heartbeat-period
-    [heartbeat_period: <duration> | default = 5s]
-
-    # The heartbeat timeout after which parquet-converter are considered
-    # unhealthy within the ring. 0 = never (timeout disabled).
-    # CLI flag: -parquet-converter.ring.heartbeat-timeout
-    [heartbeat_timeout: <duration> | default = 1m]
-
-    # Time since last heartbeat before parquet-converter will be removed from
-    # ring. 0 to disable
-    # CLI flag: -parquet-converter.auto-forget-delay
-    [auto_forget_delay: <duration> | default = 2m]
-
-    # File path where tokens are stored. If empty, tokens are not stored at
-    # shutdown and restored at startup.
-    # CLI flag: -parquet-converter.ring.tokens-file-path
-    [tokens_file_path: <string> | default = ""]
-
 # The store_gateway_config configures the store-gateway service used by the
 # blocks storage.
 [store_gateway: <store_gateway_config>]
@@ -2589,7 +2499,6 @@ The `consul_config` configures the consul client. The supported CLI flags `<pref
 - `compactor.ring`
 - `distributor.ha-tracker`
 - `distributor.ring`
-- `parquet-converter.ring`
 - `ruler.ring`
 - `store-gateway.sharding-ring`
 
@@ -2906,7 +2815,6 @@ The `etcd_config` configures the etcd client. The supported CLI flags `<prefix>`
 - `compactor.ring`
 - `distributor.ha-tracker`
 - `distributor.ring`
-- `parquet-converter.ring`
 - `ruler.ring`
 - `store-gateway.sharding-ring`
 
@@ -3803,12 +3711,6 @@ query_rejection:
 # CLI flag: -compactor.partition-series-count
 [compactor_partition_series_count: <int> | default = 0]
 
-# The default tenant's shard size when the shuffle-sharding strategy is used by
-# the parquet converter. When this setting is specified in the per-tenant
-# overrides, a value of 0 disables shuffle sharding for the tenant.
-# CLI flag: -parquet-converter.tenant-shard-size
-[parquet_converter_tenant_shard_size: <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.

pkg/cortex/cortex.go

@@ -114,7 +114,7 @@ type Config struct {
 	QueryRange       queryrange.Config               `yaml:"query_range"`
 	BlocksStorage    tsdb.BlocksStorageConfig        `yaml:"blocks_storage"`
 	Compactor        compactor.Config                `yaml:"compactor"`
-	ParquetConverter parquetconverter.Config         `yaml:"parquet_converter"`
+	ParquetConverter parquetconverter.Config         `yaml:"parquet_converter" doc:"hidden"`
 	StoreGateway     storegateway.Config             `yaml:"store_gateway"`
 	TenantFederation tenantfederation.Config         `yaml:"tenant_federation"`
 

pkg/cortex/modules.go

@@ -696,8 +696,7 @@ func (t *Cortex) initAlertManager() (serv services.Service, err error) {
 
 func (t *Cortex) initParquetConverter() (serv services.Service, err error) {
 	t.Cfg.ParquetConverter.Ring.ListenPort = t.Cfg.Server.GRPCListenPort
-	t.Parquetconverter = parquetconverter.NewConverter(t.Cfg.ParquetConverter, t.Cfg.BlocksStorage, t.Cfg.Compactor.BlockRanges.ToMilliseconds(), util_log.Logger, prometheus.DefaultRegisterer, t.Overrides)
-	return t.Parquetconverter, nil
+	return parquetconverter.NewConverter(t.Cfg.ParquetConverter, t.Cfg.BlocksStorage, t.Cfg.Compactor.BlockRanges.ToMilliseconds(), util_log.Logger, prometheus.DefaultRegisterer, t.Overrides)
 }
 
 func (t *Cortex) initCompactor() (serv services.Service, err error) {

pkg/parquetconverter/converter.go

@@ -47,6 +47,7 @@ type Config struct {
 	EnabledTenants      flagext.StringSliceCSV `yaml:"enabled_tenants"`
 	DisabledTenants     flagext.StringSliceCSV `yaml:"disabled_tenants"`
 	MetaSyncConcurrency int                    `yaml:"meta_sync_concurrency"`
+	ConversionInterval  time.Duration          `yaml:"conversion_interval"`
 
 	DataDir string `yaml:"data_dir"`
 
@@ -70,7 +71,7 @@ type Converter struct {
 	ringSubservices        *services.Manager
 	ringSubservicesWatcher *services.FailureWatcher
 
-	bkt objstore.InstrumentedBucket
+	bkt objstore.Bucket
 
 	// chunk pool
 	pool chunkenc.Pool
@@ -88,9 +89,17 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.Var(&cfg.DisabledTenants, "parquet-converter.disabled-tenants", "Comma separated list of tenants that cannot converted.")
 	f.StringVar(&cfg.DataDir, "parquet-converter.data-dir", "./data", "Data directory in which to cache blocks and process conversions.")
 	f.IntVar(&cfg.MetaSyncConcurrency, "parquet-converter.meta-sync-concurrency", 20, "Number of Go routines to use when syncing block meta files from the long term storage.")
+	f.DurationVar(&cfg.ConversionInterval, "parquet-converter.conversion-interval", time.Minute, "The frequency at which the conversion job runs.")
 }
 
-func NewConverter(cfg Config, storageCfg cortex_tsdb.BlocksStorageConfig, blockRanges []int64, logger log.Logger, registerer prometheus.Registerer, limits *validation.Overrides) *Converter {
+func NewConverter(cfg Config, storageCfg cortex_tsdb.BlocksStorageConfig, blockRanges []int64, logger log.Logger, registerer prometheus.Registerer, limits *validation.Overrides) (*Converter, error) {
+	bkt, err := bucket.NewClient(context.Background(), storageCfg.Bucket, nil, "parquet-converter", logger, registerer)
+
+	return newConverter(cfg, bkt, storageCfg, blockRanges, logger, registerer, limits), err
+}
+
+func newConverter(cfg Config, bkt objstore.InstrumentedBucket, storageCfg cortex_tsdb.BlocksStorageConfig, blockRanges []int64, logger log.Logger, registerer prometheus.Registerer, limits *validation.Overrides) *Converter {
+	bkt = bucketindex.BucketWithGlobalMarkers(bkt)
 	c := &Converter{
 		cfg:            cfg,
 		reg:            registerer,
@@ -101,24 +110,17 @@ func NewConverter(cfg Config, storageCfg cortex_tsdb.BlocksStorageConfig, blockR
 		pool:           chunkenc.NewPool(),
 		blockRanges:    blockRanges,
 		fetcherMetrics: block.NewFetcherMetrics(registerer, nil, nil),
+		bkt:            bkt,
 	}
 
 	c.Service = services.NewBasicService(c.starting, c.running, c.stopping)
 	return c
 }
 
 func (c *Converter) starting(ctx context.Context) error {
-	bkt, err := bucket.NewClient(ctx, c.storageCfg.Bucket, nil, "parquet-converter", c.logger, c.reg)
-	bkt = bucketindex.BucketWithGlobalMarkers(bkt)
-	
-	if err != nil {
-		return err
-	}
-
-	c.bkt = bkt
 	lifecyclerCfg := c.cfg.Ring.ToLifecyclerConfig()
+	var err error
 	c.ringLifecycler, err = ring.NewLifecycler(lifecyclerCfg, ring.NewNoopFlushTransferer(), "parquet-converter", ringKey, true, false, c.logger, prometheus.WrapRegistererWithPrefix("cortex_", c.reg))
-
 	if err != nil {
 		return errors.Wrap(err, "unable to initialize converter ring lifecycler")
 	}
@@ -152,7 +154,7 @@ func (c *Converter) starting(ctx context.Context) error {
 
 func (c *Converter) running(ctx context.Context) error {
 	level.Info(c.logger).Log("msg", "parquet-converter started")
-	t := time.NewTicker(time.Second * 10)
+	t := time.NewTicker(c.cfg.ConversionInterval)
 	defer t.Stop()
 	for {
 		select {
@@ -187,7 +189,6 @@ func (c *Converter) running(ctx context.Context) error {
 				level.Info(userLogger).Log("msg", "scanned user", "user", userID)
 
 				err = c.convertUser(ctx, userLogger, ring, userID)
-
 				if err != nil {
 					level.Error(userLogger).Log("msg", "failed to convert user", "user", userID, "err", err)
 				}
@@ -216,7 +217,6 @@ func (c *Converter) discoverUsers(ctx context.Context) ([]string, error) {
 }
 
 func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring ring.ReadRing, userID string) error {
-
 	uBucket := bucket.NewUserBucketClient(userID, c.bkt, c.limits)
 
 	var blockLister block.Lister
@@ -253,7 +253,6 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin
 		c.fetcherMetrics,
 		[]block.MetadataFilter{ignoreDeletionMarkFilter},
 	)
-
 	if err != nil {
 		return errors.Wrap(err, "error creating block fetcher")
 	}
@@ -275,7 +274,6 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin
 		}
 
 		marker, err := cortex_parquet.ReadConverterMark(ctx, b.ULID, uBucket, logger)
-
 		if err != nil {
 			level.Error(logger).Log("msg", "failed to read marker", "block", b.ULID.String(), "err", err)
 			continue
@@ -304,7 +302,6 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin
 		}
 
 		tsdbBlock, err := tsdb.OpenBlock(logutil.GoKitLogToSlog(logger), bdir, c.pool, tsdb.DefaultPostingsDecoderFactory)
-
 		if err != nil {
 			level.Error(logger).Log("msg", "Error opening block", "err", err)
 			continue

pkg/parquetconverter/converter_test.go

@@ -0,0 +1,136 @@
+package parquetconverter
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"testing"
+	"time"
+
+	"github.com/cortexproject/cortex/integration/e2e"
+	"github.com/cortexproject/cortex/pkg/ring"
+	"github.com/cortexproject/cortex/pkg/ring/kv/consul"
+	"github.com/cortexproject/cortex/pkg/storage/bucket"
+	"github.com/cortexproject/cortex/pkg/storage/parquet"
+	cortex_tsdb "github.com/cortexproject/cortex/pkg/storage/tsdb"
+	"github.com/cortexproject/cortex/pkg/util/concurrency"
+	"github.com/cortexproject/cortex/pkg/util/flagext"
+	"github.com/cortexproject/cortex/pkg/util/services"
+	"github.com/cortexproject/cortex/pkg/util/test"
+	"github.com/cortexproject/cortex/pkg/util/validation"
+	"github.com/go-kit/log"
+	"github.com/oklog/ulid"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/tsdb"
+	"github.com/stretchr/testify/assert"
+	"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"
+)
+
+func TestConverter(t *testing.T) {
+	cfg := prepareConfig()
+	user := "user"
+	ringStore, closer := consul.NewInMemoryClient(ring.GetCodec(), log.NewNopLogger(), nil)
+	t.Cleanup(func() { assert.NoError(t, closer.Close()) })
+	dir := t.TempDir()
+
+	cfg.Ring.InstanceID = "compactor-1"
+	cfg.Ring.InstanceAddr = "1.2.3.4"
+	cfg.Ring.KVStore.Mock = ringStore
+	bucketClient := objstore.NewInMemBucket()
+	userBucket := bucket.NewPrefixedBucketClient(bucketClient, user)
+
+	c, logger, _ := prepare(t, cfg, objstore.WithNoopInstr(bucketClient), nil)
+
+	ctx := context.Background()
+
+	lbls := labels.Labels{labels.Label{
+		Name:  "__name__",
+		Value: "test",
+	}}
+
+	blocks := []ulid.ULID{}
+	// Create blocks
+	for _, duration := range []time.Duration{2 * time.Hour, 24 * time.Hour} {
+		rnd := rand.New(rand.NewSource(time.Now().Unix()))
+		id, err := e2e.CreateBlock(ctx, rnd, dir, []labels.Labels{lbls}, 2, 0, duration.Milliseconds(), time.Minute.Milliseconds(), 10)
+		require.NoError(t, err)
+		blocks = append(blocks, id)
+	}
+
+	for _, bIds := range blocks {
+		blockDir := fmt.Sprintf("%s/%s", dir, bIds.String())
+		b, err := tsdb.OpenBlock(nil, blockDir, nil, nil)
+		require.NoError(t, err)
+		fmt.Println(b.Dir())
+		err = block.Upload(ctx, logger, userBucket, b.Dir(), metadata.NoneFunc)
+		require.NoError(t, err)
+	}
+
+	// Try to start the compactor with a bad consul kv-store. The
+	err := services.StartAndAwaitRunning(context.Background(), c)
+	require.NoError(t, err)
+
+	blocksConverted := []ulid.ULID{}
+
+	test.Poll(t, 3*time.Minute, 1, func() interface{} {
+		blocksConverted = blocksConverted[:0]
+		for _, bIds := range blocks {
+			m, err := parquet.ReadConverterMark(ctx, bIds, userBucket, logger)
+			require.NoError(t, err)
+			if m.Version == parquet.CurrentVersion {
+				blocksConverted = append(blocksConverted, bIds)
+			}
+		}
+		return len(blocksConverted)
+	})
+
+	// Verify all files are there
+	for _, block := range blocksConverted {
+		for _, file := range []string{
+			fmt.Sprintf("%s/parquet-converter-mark.json", block.String()),
+			fmt.Sprintf("markers/%s-parquet-converter-mark.json", block.String()),
+			fmt.Sprintf("%s/0.chunks.parquet", block.String()),
+			fmt.Sprintf("%s/0.labels.parquet", block.String()),
+		} {
+			ok, err := userBucket.Exists(ctx, file)
+			require.NoError(t, err)
+			require.True(t, ok)
+		}
+	}
+}
+
+func prepareConfig() Config {
+	cfg := Config{}
+	flagext.DefaultValues(&cfg)
+	cfg.ConversionInterval = time.Second
+	return cfg
+}
+
+func prepare(t *testing.T, cfg Config, bucketClient objstore.InstrumentedBucket, limits *validation.Limits) (*Converter, log.Logger, prometheus.Gatherer) {
+	storageCfg := cortex_tsdb.BlocksStorageConfig{}
+	blockRanges := cortex_tsdb.DurationList{2 * time.Hour, 12 * time.Hour, 24 * time.Hour}
+	flagext.DefaultValues(&storageCfg)
+	storageCfg.BucketStore.BlockDiscoveryStrategy = string(cortex_tsdb.RecursiveDiscovery)
+
+	// Create a temporary directory for compactor data.
+	cfg.DataDir = t.TempDir()
+
+	logs := &concurrency.SyncBuffer{}
+	logger := log.NewLogfmtLogger(logs)
+	registry := prometheus.NewRegistry()
+
+	if limits == nil {
+		limits = &validation.Limits{}
+		flagext.DefaultValues(limits)
+	}
+
+	overrides, err := validation.NewOverrides(*limits, nil)
+	require.NoError(t, err)
+
+	c := newConverter(cfg, bucketClient, storageCfg, blockRanges.ToMilliseconds(), logger, registry, overrides)
+	return c, logger, registry
+}