| 30 | } |
| 31 | |
| 32 | func (cmd *dropTracesCmd) Run(opts *globalOptions) error { |
| 33 | var ( |
| 34 | logger = log.NewLogfmtLogger(os.Stdout) |
| 35 | ctx = context.Background() |
| 36 | ) |
| 37 | |
| 38 | level.Info(logger).Log("msg", "beginning process to drop traces", "traces", cmd.TraceIDs, "tenant", cmd.TenantID) |
| 39 | level.Warn(logger).Log("msg", "compaction must be disabled or a compactor may duplicate a block as this process is rewriting it") |
| 40 | if cmd.DropTrace { |
| 41 | level.Warn(logger).Log("msg", "this is not a dry run. blocks will be rewritten and marked compacted") |
| 42 | } |
| 43 | |
| 44 | r, w, c, err := loadBackend(&cmd.backendOptions, opts) |
| 45 | if err != nil { |
| 46 | return err |
| 47 | } |
| 48 | |
| 49 | // Group trace IDs by blocks |
| 50 | ids := strings.Split(cmd.TraceIDs, ",") |
| 51 | traceIDs := make([]common.ID, len(ids)) |
| 52 | for _, id := range ids { |
| 53 | traceID, err := util.HexStringToTraceID(id) |
| 54 | if err != nil { |
| 55 | return err |
| 56 | } |
| 57 | |
| 58 | traceIDs = append(traceIDs, traceID) |
| 59 | } |
| 60 | |
| 61 | // It might be significantly improved if common.BackendBlock supported bulk searches. |
| 62 | blocks, err := cmd.blocksWithAnyTraceID(ctx, r, logger, cmd.TenantID, traceIDs...) |
| 63 | if err != nil { |
| 64 | return err |
| 65 | } |
| 66 | |
| 67 | if len(blocks) == 0 { |
| 68 | level.Info(logger).Log("msg", "traces not found in any block", "traces", cmd.TraceIDs) |
| 69 | } |
| 70 | |
| 71 | // Remove traces from blocks |
| 72 | for _, block := range blocks { |
| 73 | if !cmd.DropTrace { |
| 74 | level.Warn(logger).Log("msg", "not dropping trace, use --drop-trace to actually drop") |
| 75 | continue |
| 76 | } |
| 77 | |
| 78 | level.Info(logger).Log("msg", "rewriting block", "block", block.BlockID, "size", block.Size_, "totalTraces", block.TotalObjects) |
| 79 | newMeta, err := rewriteBlock(ctx, r, w, block, traceIDs, logger) |
| 80 | if err != nil { |
| 81 | level.Error(logger).Log("msg", "error rewriting block", "block", block.BlockID, "err", err) |
| 82 | continue |
| 83 | } |
| 84 | if newMeta == nil { |
| 85 | level.Info(logger).Log("msg", "block removed", "block", block.BlockID) |
| 86 | } else { |
| 87 | level.Info(logger).Log("msg", "rewrote block", "block", block.BlockID, "newBlock", newMeta.BlockID) |
| 88 | } |
| 89 | |