Method AppendTrace

tempodb/encoding/vparquet3/wal_block.go:342–369

(id common.ID, trace *tempopb.Trace, start, end uint32, adjustIngestionSlack bool)

}

func (b *walBlock) AppendTrace(id common.ID, trace *tempopb.Trace, start, end uint32, adjustIngestionSlack bool) error {
	var connected bool
	b.buffer, connected = traceToParquet(b.meta, id, trace, b.buffer)
	if !connected {
		dataquality.WarnDisconnectedTrace(b.meta.TenantID, dataquality.PhaseTraceFlushedToWal)
	}
	if b.buffer != nil && b.buffer.RootSpanName == "" {
		dataquality.WarnRootlessTrace(b.meta.TenantID, dataquality.PhaseTraceFlushedToWal)
	}

	if adjustIngestionSlack {
		start, end = common.AdjustTimeRangeForSlack(b.meta.TenantID, b.ingestionSlack, start, end)
	}

	// add to current
	_, err := b.writer.Write([]*Trace{b.buffer})
	if err != nil {
		return fmt.Errorf("error writing row: %w", err)
	}

	b.mtx.Lock()
	defer b.mtx.Unlock()
	b.meta.ObjectAdded(start, end)
	b.ids.Set(id, int64(b.ids.Len())) // Next row number
	b.unflushedSize += int64(estimateMarshalledSizeFromTrace(b.buffer))

	return nil
}

func (b *walBlock) IngestionSlack() time.Duration {
	return b.ingestionSlack