Method processCompactionJob

modules/backendworker/backendworker.go:276–327

(ctx context.Context, resp *tempopb.NextJobResponse)

}

func (w *BackendWorker) processCompactionJob(ctx context.Context, resp *tempopb.NextJobResponse) error {
	if resp.Detail.Tenant == "" {
		metricWorkerBadJobsReceived.WithLabelValues("no_tenant").Inc()
		return w.failJob(ctx, resp.JobId, "received compaction job with empty tenant")
	}

	level.Debug(log.Logger).Log("msg", "received job", "job_id", resp.JobId, "tenant", resp.Detail.Tenant)

	blockMetas := w.store.BlockMetas(resp.Detail.Tenant)

	// Collect the metas which match the IDs in the job
	var sourceMetas []*backend.BlockMeta
	for _, blockMeta := range blockMetas {
		for _, blockID := range resp.Detail.Compaction.Input {
			if blockMeta.BlockID.String() == blockID {
				sourceMetas = append(sourceMetas, blockMeta)
			}
		}
	}

	// Execute compaction using existing logic
	newCompacted, err := w.compact(ctx, sourceMetas, resp.Detail.Tenant)
	if err != nil {
		return w.failJob(ctx, resp.JobId, fmt.Sprintf("error compacting blocks: %v", err))
	}

	var newIDs []string
	for _, blockMeta := range newCompacted {
		newIDs = append(newIDs, blockMeta.BlockID.String())
	}

	// Mark job as complete
	err = w.callSchedulerWithBackoff(ctx, func(ctx context.Context) error {
		_, err = w.backendScheduler.UpdateJob(ctx, &tempopb.UpdateJobStatusRequest{
			JobId:  resp.JobId,
			Status: tempopb.JobStatus_JOB_STATUS_SUCCEEDED,
			Compaction: &tempopb.CompactionDetail{
				Output: newIDs,
			},
		})
		if err != nil {
			return fmt.Errorf("failed marking job %q as complete: %w", resp.JobId, err)
		}

		return nil
	})
	if err != nil {
		return w.failJob(ctx, resp.JobId, fmt.Sprintf("error marking job as complete: %v", err))
	}

	return nil
}

func (w *BackendWorker) processRetentionJob(ctx context.Context, resp *tempopb.NextJobResponse) error {
	tenantID := resp.Detail.Tenant