parent
921f82cfc1
commit
9874377ead
|
@ -15,7 +15,6 @@ package tsdb
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
|
||||||
"io/ioutil"
|
"io/ioutil"
|
||||||
"math/rand"
|
"math/rand"
|
||||||
"os"
|
"os"
|
||||||
|
@ -70,22 +69,20 @@ func createBlock(tb testing.TB, dir string, nSeries int, mint, maxt int64) strin
|
||||||
|
|
||||||
lbls, err := labels.ReadLabels(filepath.Join("testdata", "20kseries.json"), nSeries)
|
lbls, err := labels.ReadLabels(filepath.Join("testdata", "20kseries.json"), nSeries)
|
||||||
testutil.Ok(tb, err)
|
testutil.Ok(tb, err)
|
||||||
var ref uint64
|
refs := make([]uint64, nSeries)
|
||||||
|
|
||||||
fmt.Println(len(lbls))
|
|
||||||
|
|
||||||
for ts := mint; ts <= maxt; ts++ {
|
for ts := mint; ts <= maxt; ts++ {
|
||||||
app := head.Appender()
|
app := head.Appender()
|
||||||
for i, lbl := range lbls {
|
for i, lbl := range lbls {
|
||||||
if i > 0 && lbl.String() == lbls[i-1].String() {
|
if refs[i] != 0 {
|
||||||
err := app.AddFast(ref, ts, rand.Float64())
|
err := app.AddFast(refs[i], ts, rand.Float64())
|
||||||
if err == nil {
|
if err == nil {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
ref, err = app.Add(lbl, int64(ts), rand.Float64())
|
ref, err := app.Add(lbl, int64(ts), rand.Float64())
|
||||||
testutil.Ok(tb, err)
|
testutil.Ok(tb, err)
|
||||||
|
refs[i] = ref
|
||||||
}
|
}
|
||||||
err := app.Commit()
|
err := app.Commit()
|
||||||
testutil.Ok(tb, err)
|
testutil.Ok(tb, err)
|
||||||
|
|
|
@ -617,7 +617,6 @@ func (c *LeveledCompactor) populateBlock(blocks []BlockReader, meta *BlockMeta,
|
||||||
|
|
||||||
c.metrics.populatingBlocks.Inc()
|
c.metrics.populatingBlocks.Inc()
|
||||||
|
|
||||||
fmt.Println(blocks)
|
|
||||||
for i, b := range blocks {
|
for i, b := range blocks {
|
||||||
select {
|
select {
|
||||||
case <-c.ctx.Done():
|
case <-c.ctx.Done():
|
||||||
|
@ -625,8 +624,6 @@ func (c *LeveledCompactor) populateBlock(blocks []BlockReader, meta *BlockMeta,
|
||||||
default:
|
default:
|
||||||
}
|
}
|
||||||
|
|
||||||
fmt.Println("next block")
|
|
||||||
|
|
||||||
indexr, err := b.Index()
|
indexr, err := b.Index()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return errors.Wrapf(err, "open index reader for block %s", b)
|
return errors.Wrapf(err, "open index reader for block %s", b)
|
||||||
|
@ -689,7 +686,6 @@ func (c *LeveledCompactor) populateBlock(blocks []BlockReader, meta *BlockMeta,
|
||||||
default:
|
default:
|
||||||
}
|
}
|
||||||
|
|
||||||
// fmt.Println("next set")
|
|
||||||
lset, chks, dranges := set.At() // The chunks here are not fully deleted.
|
lset, chks, dranges := set.At() // The chunks here are not fully deleted.
|
||||||
|
|
||||||
// Skip the series with all deleted chunks.
|
// Skip the series with all deleted chunks.
|
||||||
|
|
|
@ -15,7 +15,6 @@ package tsdb
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
|
||||||
"io/ioutil"
|
"io/ioutil"
|
||||||
"math"
|
"math"
|
||||||
"os"
|
"os"
|
||||||
|
@ -753,8 +752,8 @@ func TestCancelCompactions(t *testing.T) {
|
||||||
defer os.RemoveAll(tmpdir)
|
defer os.RemoveAll(tmpdir)
|
||||||
|
|
||||||
// Create some blocks to fall within the compaction range.
|
// Create some blocks to fall within the compaction range.
|
||||||
createBlock(t, tmpdir, 1000, 0, 1000)
|
createBlock(t, tmpdir, 3000, 0, 1000)
|
||||||
createBlock(t, tmpdir, 1000, 1000, 2000)
|
createBlock(t, tmpdir, 3000, 1000, 2000)
|
||||||
createBlock(t, tmpdir, 1, 2000, 2001) // The most recent block is ignored so can be e small one.
|
createBlock(t, tmpdir, 1, 2000, 2001) // The most recent block is ignored so can be e small one.
|
||||||
|
|
||||||
db, err := Open(tmpdir, log.NewNopLogger(), nil, &Options{BlockRanges: []int64{1, 2000}})
|
db, err := Open(tmpdir, log.NewNopLogger(), nil, &Options{BlockRanges: []int64{1, 2000}})
|
||||||
|
@ -764,8 +763,7 @@ func TestCancelCompactions(t *testing.T) {
|
||||||
dbClosed := make(chan struct{})
|
dbClosed := make(chan struct{})
|
||||||
for {
|
for {
|
||||||
if prom_testutil.ToFloat64(db.compactor.(*LeveledCompactor).metrics.populatingBlocks) > 0 {
|
if prom_testutil.ToFloat64(db.compactor.(*LeveledCompactor).metrics.populatingBlocks) > 0 {
|
||||||
fmt.Println("populating started.")
|
time.Sleep(3 * time.Millisecond)
|
||||||
time.Sleep(2 * time.Millisecond)
|
|
||||||
go func() {
|
go func() {
|
||||||
testutil.Ok(t, db.Close())
|
testutil.Ok(t, db.Close())
|
||||||
close(dbClosed)
|
close(dbClosed)
|
||||||
|
|
4
db.go
4
db.go
|
@ -819,9 +819,7 @@ func (db *DB) Head() *Head {
|
||||||
// Close the partition.
|
// Close the partition.
|
||||||
func (db *DB) Close() error {
|
func (db *DB) Close() error {
|
||||||
close(db.stopc)
|
close(db.stopc)
|
||||||
// fmt.Println("closing")
|
db.compactCnl()
|
||||||
// db.compactCnl()
|
|
||||||
// fmt.Println("closed")
|
|
||||||
<-db.donec
|
<-db.donec
|
||||||
|
|
||||||
db.mtx.Lock()
|
db.mtx.Lock()
|
||||||
|
|
Loading…
Reference in New Issue