@@ -47,6 +47,7 @@ type Config struct {
4747 EnabledTenants flagext.StringSliceCSV `yaml:"enabled_tenants"`
4848 DisabledTenants flagext.StringSliceCSV `yaml:"disabled_tenants"`
4949 MetaSyncConcurrency int `yaml:"meta_sync_concurrency"`
50+ ConversionInterval time.Duration `yaml:"conversion_interval"`
5051
5152 DataDir string `yaml:"data_dir"`
5253
@@ -70,7 +71,7 @@ type Converter struct {
7071 ringSubservices * services.Manager
7172 ringSubservicesWatcher * services.FailureWatcher
7273
73- bkt objstore.InstrumentedBucket
74+ bkt objstore.Bucket
7475
7576 // chunk pool
7677 pool chunkenc.Pool
@@ -88,9 +89,17 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
8889 f .Var (& cfg .DisabledTenants , "parquet-converter.disabled-tenants" , "Comma separated list of tenants that cannot converted." )
8990 f .StringVar (& cfg .DataDir , "parquet-converter.data-dir" , "./data" , "Data directory in which to cache blocks and process conversions." )
9091 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." )
92+ f .DurationVar (& cfg .ConversionInterval , "parquet-converter.conversion-interval" , time .Minute , "The frequency at which the conversion job runs." )
9193}
9294
93- func NewConverter (cfg Config , storageCfg cortex_tsdb.BlocksStorageConfig , blockRanges []int64 , logger log.Logger , registerer prometheus.Registerer , limits * validation.Overrides ) * Converter {
95+ func NewConverter (cfg Config , storageCfg cortex_tsdb.BlocksStorageConfig , blockRanges []int64 , logger log.Logger , registerer prometheus.Registerer , limits * validation.Overrides ) (* Converter , error ) {
96+ bkt , err := bucket .NewClient (context .Background (), storageCfg .Bucket , nil , "parquet-converter" , logger , registerer )
97+
98+ return newConverter (cfg , bkt , storageCfg , blockRanges , logger , registerer , limits ), err
99+ }
100+
101+ func newConverter (cfg Config , bkt objstore.InstrumentedBucket , storageCfg cortex_tsdb.BlocksStorageConfig , blockRanges []int64 , logger log.Logger , registerer prometheus.Registerer , limits * validation.Overrides ) * Converter {
102+ bkt = bucketindex .BucketWithGlobalMarkers (bkt )
94103 c := & Converter {
95104 cfg : cfg ,
96105 reg : registerer ,
@@ -101,24 +110,17 @@ func NewConverter(cfg Config, storageCfg cortex_tsdb.BlocksStorageConfig, blockR
101110 pool : chunkenc .NewPool (),
102111 blockRanges : blockRanges ,
103112 fetcherMetrics : block .NewFetcherMetrics (registerer , nil , nil ),
113+ bkt : bkt ,
104114 }
105115
106116 c .Service = services .NewBasicService (c .starting , c .running , c .stopping )
107117 return c
108118}
109119
110120func (c * Converter ) starting (ctx context.Context ) error {
111- bkt , err := bucket .NewClient (ctx , c .storageCfg .Bucket , nil , "parquet-converter" , c .logger , c .reg )
112- bkt = bucketindex .BucketWithGlobalMarkers (bkt )
113-
114- if err != nil {
115- return err
116- }
117-
118- c .bkt = bkt
119121 lifecyclerCfg := c .cfg .Ring .ToLifecyclerConfig ()
122+ var err error
120123 c .ringLifecycler , err = ring .NewLifecycler (lifecyclerCfg , ring .NewNoopFlushTransferer (), "parquet-converter" , ringKey , true , false , c .logger , prometheus .WrapRegistererWithPrefix ("cortex_" , c .reg ))
121-
122124 if err != nil {
123125 return errors .Wrap (err , "unable to initialize converter ring lifecycler" )
124126 }
@@ -152,7 +154,7 @@ func (c *Converter) starting(ctx context.Context) error {
152154
153155func (c * Converter ) running (ctx context.Context ) error {
154156 level .Info (c .logger ).Log ("msg" , "parquet-converter started" )
155- t := time .NewTicker (time . Second * 10 )
157+ t := time .NewTicker (c . cfg . ConversionInterval )
156158 defer t .Stop ()
157159 for {
158160 select {
@@ -187,7 +189,6 @@ func (c *Converter) running(ctx context.Context) error {
187189 level .Info (userLogger ).Log ("msg" , "scanned user" , "user" , userID )
188190
189191 err = c .convertUser (ctx , userLogger , ring , userID )
190-
191192 if err != nil {
192193 level .Error (userLogger ).Log ("msg" , "failed to convert user" , "user" , userID , "err" , err )
193194 }
@@ -216,7 +217,6 @@ func (c *Converter) discoverUsers(ctx context.Context) ([]string, error) {
216217}
217218
218219func (c * Converter ) convertUser (ctx context.Context , logger log.Logger , ring ring.ReadRing , userID string ) error {
219-
220220 uBucket := bucket .NewUserBucketClient (userID , c .bkt , c .limits )
221221
222222 var blockLister block.Lister
@@ -253,7 +253,6 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin
253253 c .fetcherMetrics ,
254254 []block.MetadataFilter {ignoreDeletionMarkFilter },
255255 )
256-
257256 if err != nil {
258257 return errors .Wrap (err , "error creating block fetcher" )
259258 }
@@ -275,7 +274,6 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin
275274 }
276275
277276 marker , err := cortex_parquet .ReadConverterMark (ctx , b .ULID , uBucket , logger )
278-
279277 if err != nil {
280278 level .Error (logger ).Log ("msg" , "failed to read marker" , "block" , b .ULID .String (), "err" , err )
281279 continue
@@ -304,7 +302,6 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin
304302 }
305303
306304 tsdbBlock , err := tsdb .OpenBlock (logutil .GoKitLogToSlog (logger ), bdir , c .pool , tsdb .DefaultPostingsDecoderFactory )
307-
308305 if err != nil {
309306 level .Error (logger ).Log ("msg" , "Error opening block" , "err" , err )
310307 continue
0 commit comments