(ctx context.Context, userID string)
| 995 | } |
| 996 | |
| 997 | func (c *Compactor) compactUser(ctx context.Context, userID string) error { |
| 998 | bucket := bucket.NewUserBucketClient(userID, c.bucketClient, c.limits) |
| 999 | |
| 1000 | reg := prometheus.NewRegistry() |
| 1001 | |
| 1002 | ulogger := util_log.WithUserID(userID, c.logger) |
| 1003 | ulogger = util_log.WithExecutionID(ulid.MustNew(ulid.Now(), crypto_rand.Reader).String(), ulogger) |
| 1004 | |
| 1005 | // Filters out duplicate blocks that can be formed from two or more overlapping |
| 1006 | // blocks that fully submatches the source blocks of the older blocks. |
| 1007 | var deduplicateBlocksFilter CortexMetadataFilter |
| 1008 | if c.compactorCfg.ShardingStrategy == util.ShardingStrategyShuffle && c.compactorCfg.CompactionStrategy == util.CompactionStrategyPartitioning { |
| 1009 | deduplicateBlocksFilter = &disabledDeduplicateFilter{} |
| 1010 | } else { |
| 1011 | deduplicateBlocksFilter = block.NewDeduplicateFilter(c.compactorCfg.BlockSyncConcurrency) |
| 1012 | } |
| 1013 | |
| 1014 | // While fetching blocks, we filter out blocks that were marked for deletion by using IgnoreDeletionMarkFilter. |
| 1015 | // No delay is used -- all blocks with deletion marker are ignored, and not considered for compaction. |
| 1016 | ignoreDeletionMarkFilter := block.NewIgnoreDeletionMarkFilter( |
| 1017 | ulogger, |
| 1018 | bucket, |
| 1019 | 0, |
| 1020 | c.compactorCfg.MetaSyncConcurrency) |
| 1021 | |
| 1022 | // Filters out blocks with no compaction maker; blocks can be marked as no compaction for reasons like |
| 1023 | // out of order chunks or index file too big. |
| 1024 | noCompactMarkerFilter := compact.NewGatherNoCompactionMarkFilter(ulogger, bucket, c.compactorCfg.MetaSyncConcurrency) |
| 1025 | |
| 1026 | var blockLister block.Lister |
| 1027 | blockDiscoveryStrategy := cortex_tsdb.BlockDiscoveryStrategy(c.storageCfg.BucketStore.BlockDiscoveryStrategy) |
| 1028 | switch blockDiscoveryStrategy { |
| 1029 | case cortex_tsdb.ConcurrentDiscovery: |
| 1030 | blockLister = block.NewConcurrentLister(ulogger, bucket) |
| 1031 | case cortex_tsdb.RecursiveDiscovery: |
| 1032 | blockLister = block.NewRecursiveLister(ulogger, bucket) |
| 1033 | case cortex_tsdb.BucketIndexDiscovery: |
| 1034 | if !c.storageCfg.BucketStore.BucketIndex.Enabled { |
| 1035 | return cortex_tsdb.ErrInvalidBucketIndexBlockDiscoveryStrategy |
| 1036 | } |
| 1037 | blockLister = bucketindex.NewBlockLister(ulogger, c.bucketClient, userID, c.limits) |
| 1038 | default: |
| 1039 | return cortex_tsdb.ErrBlockDiscoveryStrategy |
| 1040 | } |
| 1041 | |
| 1042 | // List of filters to apply (order matters). |
| 1043 | filterList := []block.MetadataFilter{ |
| 1044 | // Remove the ingester ID because we don't shard blocks anymore, while still |
| 1045 | // honoring the shard ID if sharding was done in the past. |
| 1046 | NewLabelRemoverFilter([]string{cortex_tsdb.IngesterIDExternalLabel}), |
| 1047 | block.NewConsistencyDelayMetaFilter(ulogger, c.compactorCfg.ConsistencyDelay, reg), |
| 1048 | } |
| 1049 | |
| 1050 | // Add ignoreDeletionMarkFilter only when not using bucket index discovery. |
| 1051 | if blockDiscoveryStrategy != cortex_tsdb.BucketIndexDiscovery { |
| 1052 | filterList = append(filterList, ignoreDeletionMarkFilter) |
| 1053 | } |
| 1054 |
no test coverage detected