Method ListBlocks

tempodb/backend/s3/s3.go:396–502

(
	ctx context.Context,
	tenant string,
)

}

func (rw *readerWriter) ListBlocks(
	ctx context.Context,
	tenant string,
) ([]uuid.UUID, []uuid.UUID, error) {
	ctx, span := tracer.Start(ctx, "readerWriter.ListBlocks")
	defer span.End()

	var (
		wg                = sync.WaitGroup{}
		mtx               = sync.Mutex{}
		bb                = blockboundary.CreateBlockBoundaries(rw.cfg.ListBlocksConcurrency)
		errChan           = make(chan error, len(bb))
		keypath           = backend.KeyPathWithPrefix(backend.KeyPath{tenant}, rw.cfg.Prefix)
		minID             uuid.UUID
		maxID             uuid.UUID
		blockIDs          = make([]uuid.UUID, 0, 1000)
		compactedBlockIDs = make([]uuid.UUID, 0, 1000)
	)

	prefix := path.Join(keypath...)
	if len(prefix) > 0 {
		prefix += "/"
	}

	for i := 0; i < len(bb)-1; i++ {
		minID = uuid.UUID(bb[i])
		maxID = uuid.UUID(bb[i+1])

		wg.Add(1)
		go func(minUUID, maxUUID uuid.UUID) {
			defer wg.Done()

			var (
				err        error
				res        minio.ListBucketV2Result
				startAfter = prefix + minUUID.String()
			)

			for res.IsTruncated = true; res.IsTruncated; {
				if ctx.Err() != nil {
					return
				}

				res, err = rw.core.ListObjectsV2(rw.cfg.Bucket, prefix, startAfter, res.NextContinuationToken, "", 0)
				if err != nil {
					errChan <- fmt.Errorf("error finding objects in s3 bucket, bucket: %s: %w", rw.cfg.Bucket, err)
					return
				}

				for _, c := range res.Contents {
					// i.e: <blockID>/meta
					parts := strings.Split(strings.TrimPrefix(c.Key, prefix), "/")
					if len(parts) != 2 {
						continue
					}

					switch parts[1] {
					case backend.MetaName: