Method Flush

modules/blockbuilder/tenant_store.go:104–201

(ctx context.Context, r tempodb.Reader, w tempodb.Writer, c tempodb.Compactor)

}

func (s *tenantStore) Flush(ctx context.Context, r tempodb.Reader, w tempodb.Writer, c tempodb.Compactor) error {
	ctx, span := tracer.Start(ctx, "tenantStore.Flush", trace.WithAttributes(attribute.String("tenant", s.tenantID)))
	defer span.End()

	if s.liveTraces.Len() == 0 {
		span.AddEvent("no live_traces to flush")
		// This can happen if the tenant instance was created but
		// no live traces were successfully pushed. i.e. all exceeded max trace size.
		return nil
	}

	blockID, existingBlocksToBeCompacted, err := s.determineBlockIDs(ctx, r)
	if err != nil {
		return err
	}

	// TODO - Check if the existing block can be reused and exit early

	if len(existingBlocksToBeCompacted) > 0 {
		for _, blockID := range existingBlocksToBeCompacted {
			level.Warn(s.logger).Log(
				"msg", "Marking existing block compacted",
				"tenant", s.tenantID,
				"blockid", blockID,
			)
			span.AddEvent("marking existing block compacted", trace.WithAttributes(attribute.String("block_id", blockID.String())))
			if err := c.MarkBlockCompacted(s.tenantID, blockID); err != nil {
				return err
			}
		}
	}

	// Initial meta for creating the block
	meta := backend.NewBlockMeta(s.tenantID, (uuid.UUID)(blockID), s.enc.Version())
	meta.DedicatedColumns = s.getDedicatedColumns()
	meta.ReplicationFactor = 1
	meta.TotalObjects = int64(s.liveTraces.Len())

	var (
		st     = time.Now()
		l      = s.wal.LocalBackend()
		reader = backend.NewReader(l)
		writer = backend.NewWriter(l)
		iter   = newLiveTracesIter(s.liveTraces)
	)

	level.Info(s.logger).Log(
		"msg", "Flushing block",
		"tenant", s.tenantID,
		"blockid", meta.BlockID,
		"meta", meta,
	)

	newMeta, err := s.enc.CreateBlock(ctx, &s.cfg.BlockConfig, meta, iter, reader, writer)
	if err != nil {
		return err
	}