cleanUserPartialBlocks delete partial blocks which are safe to be deleted. The provided partials map and index are updated accordingly.
(ctx context.Context, userID string, partials map[ulid.ULID]error, idx *bucketindex.Index, userBucket objstore.InstrumentedBucket, userLogger log.Logger)
| 918 | // cleanUserPartialBlocks delete partial blocks which are safe to be deleted. The provided partials map |
| 919 | // and index are updated accordingly. |
| 920 | func (c *BlocksCleaner) cleanUserPartialBlocks(ctx context.Context, userID string, partials map[ulid.ULID]error, idx *bucketindex.Index, userBucket objstore.InstrumentedBucket, userLogger log.Logger) { |
| 921 | // Collect all blocks with missing meta.json into buffered channel. |
| 922 | blocks := make([]any, 0, len(partials)) |
| 923 | |
| 924 | for blockID, blockErr := range partials { |
| 925 | // We can safely delete only blocks which are partial because the meta.json is missing. |
| 926 | if !errors.Is(blockErr, bucketindex.ErrBlockMetaNotFound) { |
| 927 | continue |
| 928 | } |
| 929 | blocks = append(blocks, blockID) |
| 930 | } |
| 931 | |
| 932 | var mux sync.Mutex |
| 933 | |
| 934 | _ = concurrency.ForEach(ctx, blocks, defaultDeleteBlocksConcurrency, func(ctx context.Context, job any) error { |
| 935 | blockID := job.(ulid.ULID) |
| 936 | // We can safely delete only partial blocks with a deletion mark. |
| 937 | err := metadata.ReadMarker(ctx, userLogger, userBucket, blockID.String(), &metadata.DeletionMark{}) |
| 938 | if errors.Is(err, metadata.ErrorMarkerNotFound) { |
| 939 | //If only visit marker exists in the block, we can safely delete it. |
| 940 | isEmpty := true |
| 941 | notVisitMarkerError := userBucket.ReaderWithExpectedErrs(IsNotBlockVisitMarkerError).Iter(ctx, blockID.String(), func(file string) error { |
| 942 | isEmpty = false |
| 943 | if !IsBlockVisitMarker(file) { |
| 944 | // return error here to fail iteration fast |
| 945 | // to avoid going through all files |
| 946 | return ErrorNotBlockVisitMarker |
| 947 | } |
| 948 | return nil |
| 949 | }) |
| 950 | if isEmpty || notVisitMarkerError != nil { |
| 951 | // skip deleting partial block if block directory |
| 952 | // is empty or non visit marker file exists |
| 953 | return nil |
| 954 | } |
| 955 | } else if err != nil { |
| 956 | level.Warn(userLogger).Log("msg", "error reading partial block deletion mark", "block", blockID, "err", err) |
| 957 | return nil |
| 958 | } |
| 959 | |
| 960 | // Hard-delete partial blocks having a deletion mark, even if the deletion threshold has not |
| 961 | // been reached yet. |
| 962 | if err := block.Delete(ctx, userLogger, userBucket, blockID); err != nil { |
| 963 | c.blocksFailedTotal.Inc() |
| 964 | level.Warn(userLogger).Log("msg", "error deleting partial block marked for deletion", "block", blockID, "err", err) |
| 965 | return nil |
| 966 | } |
| 967 | |
| 968 | // Remove the block from the bucket index too. |
| 969 | mux.Lock() |
| 970 | idx.RemoveBlock(blockID) |
| 971 | delete(partials, blockID) |
| 972 | mux.Unlock() |
| 973 | |
| 974 | c.blocksCleanedTotal.Inc() |
| 975 | c.tenantBlocksCleanedTotal.WithLabelValues(userID).Inc() |
| 976 | level.Info(userLogger).Log("msg", "deleted partial block marked for deletion", "block", blockID) |
| 977 | return nil |
no test coverage detected