-
Notifications
You must be signed in to change notification settings - Fork 2.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
tools: Added thanos bucket tool rewrite (for now: allowing block seri…
…es deletions). Signed-off-by: Bartlomiej Plotka <bwplotka@gmail.com>
- Loading branch information
Showing
8 changed files
with
991 additions
and
7 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,176 @@ | ||
| package block | ||
|
|
||
| import ( | ||
| "context" | ||
| "io" | ||
| "os" | ||
| "path/filepath" | ||
|
|
||
| "github.com/go-kit/kit/log" | ||
| "github.com/go-kit/kit/log/level" | ||
| "github.com/pkg/errors" | ||
| "github.com/prometheus/prometheus/pkg/labels" | ||
| "github.com/prometheus/prometheus/tsdb" | ||
| "github.com/prometheus/prometheus/tsdb/chunks" | ||
| tsdb_errors "github.com/prometheus/prometheus/tsdb/errors" | ||
| "github.com/prometheus/prometheus/tsdb/fileutil" | ||
| "github.com/prometheus/prometheus/tsdb/index" | ||
| ) | ||
|
|
||
| // Reader is like tsdb.BlockReader but without tombstones and size methods. | ||
| type Reader interface { | ||
| // Index returns an IndexReader over the block's data. | ||
| Index() (tsdb.IndexReader, error) | ||
|
|
||
| // Chunks returns a ChunkReader over the block's data. | ||
| Chunks() (tsdb.ChunkReader, error) | ||
|
|
||
| Meta() tsdb.BlockMeta | ||
| } | ||
|
|
||
| // SeriesWriter is interface for writing series into one or multiple Blocks. | ||
| // Statistics has to be counted by implementation. | ||
| type SeriesWriter interface { | ||
| tsdb.IndexWriter | ||
| tsdb.ChunkWriter | ||
| } | ||
|
|
||
| // Writer is interface for creating block(s). | ||
| type Writer interface { | ||
| SeriesWriter | ||
|
|
||
| Flush() (tsdb.BlockStats, error) | ||
| } | ||
|
|
||
| type DiskWriter struct { | ||
| statsGatheringSeriesWriter | ||
|
|
||
| bTmp, bDir string | ||
| logger log.Logger | ||
| closers []io.Closer | ||
| } | ||
|
|
||
| const tmpForCreationBlockDirSuffix = ".tmp-for-creation" | ||
|
|
||
| // NewDiskWriter allows to write single TSDB block to disk and returns statistics. | ||
| func NewDiskWriter(ctx context.Context, logger log.Logger, bDir string) (_ *DiskWriter, err error) { | ||
| bTmp := bDir + tmpForCreationBlockDirSuffix | ||
|
|
||
| d := &DiskWriter{ | ||
| bTmp: bTmp, | ||
| bDir: bDir, | ||
| logger: logger, | ||
| } | ||
| defer func() { | ||
| if err != nil { | ||
| err = tsdb_errors.NewMulti(err, tsdb_errors.CloseAll(d.closers)).Err() | ||
| if err := os.RemoveAll(bTmp); err != nil { | ||
| level.Error(logger).Log("msg", "removed tmp folder after failed compaction", "err", err.Error()) | ||
| } | ||
| } | ||
| }() | ||
|
|
||
| if err = os.RemoveAll(bTmp); err != nil { | ||
| return nil, err | ||
| } | ||
| if err = os.MkdirAll(bTmp, 0777); err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| chunkw, err := chunks.NewWriter(filepath.Join(bTmp, ChunksDirname)) | ||
| if err != nil { | ||
| return nil, errors.Wrap(err, "open chunk writer") | ||
| } | ||
| d.closers = append(d.closers, chunkw) | ||
|
|
||
| // TODO(bwplotka): Setup instrumentedChunkWriter if we want to upstream this code. | ||
|
|
||
| indexw, err := index.NewWriter(ctx, filepath.Join(bTmp, IndexFilename)) | ||
| if err != nil { | ||
| return nil, errors.Wrap(err, "open index writer") | ||
| } | ||
| d.closers = append(d.closers, indexw) | ||
| d.statsGatheringSeriesWriter = statsGatheringSeriesWriter{iw: indexw, cw: chunkw} | ||
| return d, nil | ||
| } | ||
|
|
||
| func (d *DiskWriter) Flush() (_ tsdb.BlockStats, err error) { | ||
| defer func() { | ||
| if err != nil { | ||
| err = tsdb_errors.NewMulti(err, tsdb_errors.CloseAll(d.closers)).Err() | ||
| if err := os.RemoveAll(d.bTmp); err != nil { | ||
| level.Error(d.logger).Log("msg", "removed tmp folder failed after block(s) write", "err", err.Error()) | ||
| } | ||
| } | ||
| }() | ||
| df, err := fileutil.OpenDir(d.bTmp) | ||
| if err != nil { | ||
| return tsdb.BlockStats{}, errors.Wrap(err, "open temporary block dir") | ||
| } | ||
| defer func() { | ||
| if df != nil { | ||
| err = tsdb_errors.NewMulti(err, df.Close()).Err() | ||
| } | ||
| }() | ||
|
|
||
| if err := df.Sync(); err != nil { | ||
| return tsdb.BlockStats{}, errors.Wrap(err, "sync temporary dir file") | ||
| } | ||
|
|
||
| // Close temp dir before rename block dir (for windows platform). | ||
| if err = df.Close(); err != nil { | ||
| return tsdb.BlockStats{}, errors.Wrap(err, "close temporary dir") | ||
| } | ||
| df = nil | ||
|
|
||
| if err := tsdb_errors.CloseAll(d.closers); err != nil { | ||
| d.closers = nil | ||
| return tsdb.BlockStats{}, err | ||
| } | ||
| d.closers = nil | ||
|
|
||
| // Block successfully written, make it visible in destination dir by moving it from tmp one. | ||
| if err := fileutil.Replace(d.bTmp, d.bDir); err != nil { | ||
| return tsdb.BlockStats{}, errors.Wrap(err, "rename block dir") | ||
| } | ||
| return d.stats, nil | ||
| } | ||
|
|
||
| type statsGatheringSeriesWriter struct { | ||
| iw tsdb.IndexWriter | ||
| cw tsdb.ChunkWriter | ||
|
|
||
| stats tsdb.BlockStats | ||
| symbols int64 | ||
| } | ||
|
|
||
| func (s *statsGatheringSeriesWriter) AddSymbol(sym string) error { | ||
| if err := s.iw.AddSymbol(sym); err != nil { | ||
| return err | ||
| } | ||
| s.symbols++ | ||
| return nil | ||
| } | ||
|
|
||
| func (s *statsGatheringSeriesWriter) AddSeries(ref uint64, l labels.Labels, chks ...chunks.Meta) error { | ||
| if err := s.iw.AddSeries(ref, l, chks...); err != nil { | ||
| return err | ||
| } | ||
| s.stats.NumSeries++ | ||
| return nil | ||
| } | ||
|
|
||
| func (s *statsGatheringSeriesWriter) WriteChunks(chks ...chunks.Meta) error { | ||
| if err := s.cw.WriteChunks(chks...); err != nil { | ||
| return err | ||
| } | ||
| s.stats.NumChunks += uint64(len(chks)) | ||
| for _, chk := range chks { | ||
| s.stats.NumSamples += uint64(chk.Chunk.NumSamples()) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (s statsGatheringSeriesWriter) Close() error { | ||
| return tsdb_errors.NewMulti(s.iw.Close(), s.cw.Close()).Err() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,105 @@ | ||
| package block | ||
|
|
||
| import ( | ||
| "math" | ||
|
|
||
| "github.com/prometheus/prometheus/pkg/labels" | ||
| "github.com/prometheus/prometheus/storage" | ||
| "github.com/prometheus/prometheus/tsdb/chunks" | ||
| "github.com/prometheus/prometheus/tsdb/index" | ||
| "github.com/prometheus/prometheus/tsdb/tombstones" | ||
| ) | ||
|
|
||
| type Modifier interface { | ||
| Modify(sym index.StringIter, set storage.ChunkSeriesSet, log printChangeLog) (index.StringIter, storage.ChunkSeriesSet) | ||
| } | ||
|
|
||
| type DeletionModifier struct { | ||
| deletions []DeleteRequest | ||
| } | ||
|
|
||
| func WithDeletionModifier(deletions []DeleteRequest) *DeletionModifier { | ||
| return &DeletionModifier{deletions: deletions} | ||
| } | ||
|
|
||
| func (d *DeletionModifier) Modify(sym index.StringIter, set storage.ChunkSeriesSet, log printChangeLog) (index.StringIter, storage.ChunkSeriesSet) { | ||
| return sym, &delModifierSeriesSet{ | ||
| d: d, | ||
|
|
||
| ChunkSeriesSet: set, | ||
| log: log, | ||
| } | ||
| } | ||
|
|
||
| type delModifierSeriesSet struct { | ||
| storage.ChunkSeriesSet | ||
|
|
||
| d *DeletionModifier | ||
| log printChangeLog | ||
|
|
||
| err error | ||
| } | ||
|
|
||
| func (d *delModifierSeriesSet) Next() bool { | ||
| for d.ChunkSeriesSet.Next() { | ||
| s := d.ChunkSeriesSet.At() | ||
| lbls := s.Labels() | ||
|
|
||
| var intervals tombstones.Intervals | ||
| for _, deletions := range d.d.deletions { | ||
| for _, m := range deletions.Matchers { | ||
| v := lbls.Get(m.Name) | ||
| if v == "" { | ||
| continue | ||
| } | ||
|
|
||
| if m.Matches(v) { | ||
| continue | ||
| } | ||
| for _, in := range deletions.intervals { | ||
| intervals = intervals.Add(in) | ||
| } | ||
| break | ||
| } | ||
| } | ||
|
|
||
| if (tombstones.Interval{Mint: math.MinInt64, Maxt: math.MaxInt64}.IsSubrange(intervals)) { | ||
| // Quick path for skipping series completely. | ||
| chksIter := d.ChunkSeriesSet.At().Iterator() | ||
| var chks []chunks.Meta | ||
| for chksIter.Next() { | ||
| chks = append(chks, chksIter.At()) | ||
| } | ||
| d.err = chksIter.Err() | ||
| if d.err != nil { | ||
| return false | ||
| } | ||
|
|
||
| deleted := tombstones.Intervals{} | ||
| if len(chks) > 0 { | ||
| deleted.Add(tombstones.Interval{Mint: chks[0].MinTime, Maxt: chks[len(chks)].MaxTime}) | ||
| } | ||
| d.log.DeleteSeries(lbls, deleted) | ||
| continue | ||
| } | ||
| } | ||
| return false | ||
| } | ||
| func (d *delModifierSeriesSet) At() storage.ChunkSeries { | ||
|
|
||
| } | ||
|
|
||
| func (d *delModifierSeriesSet) Err() error { | ||
| panic("implement me") | ||
| } | ||
|
|
||
| func (d *delModifierSeriesSet) Warnings() storage.Warnings { | ||
| panic("implement me") | ||
| } | ||
|
|
||
| // TODO(bwplotka): Add relabelling. | ||
|
|
||
| type DeleteRequest struct { | ||
| Matchers []*labels.Matcher | ||
| intervals tombstones.Intervals | ||
| } |
Oops, something went wrong.