mudhorn/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2025-02-21 03:16:00 -08:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

Use page writer in compaction

Browse source
This commit is contained in:
Fabian Reinartz 2017-01-09 18:34:29 +01:00
parent 89d8467f5c
commit 0dffd52238
3 changed files with 21 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
head.go
View file

@ -409,6 +409,7 @@ func (h *HeadBlock) updateMapping() {
h.mtx.RLock() h.mtx.RLock()
if h.mapper.sortable != nil && h.mapper.Len() == len(h.descs) { if h.mapper.sortable != nil && h.mapper.Len() == len(h.descs) {
h.mtx.RUnlock()
return return
} }

2
wal.go
View file

@ -174,7 +174,7 @@ const (
// walPageBytes is the alignment for flushing records to the backing Writer. // walPageBytes is the alignment for flushing records to the backing Writer.
// It should be a multiple of the minimum sector size so that WAL can safely // It should be a multiple of the minimum sector size so that WAL can safely
// distinguish between torn writes and ordinary data corruption. // distinguish between torn writes and ordinary data corruption.
walPageBytes = 32 * minSectorSize walPageBytes = 16 * minSectorSize
) )
func newWALEncoder(f *os.File) (*walEncoder, error) { func newWALEncoder(f *os.File) (*walEncoder, error) {

28
writer.go
View file

@ -9,6 +9,7 @@ import (
"strings" "strings"
"github.com/bradfitz/slice" "github.com/bradfitz/slice"
"github.com/coreos/etcd/pkg/ioutil"
"github.com/fabxc/tsdb/chunks" "github.com/fabxc/tsdb/chunks"
"github.com/fabxc/tsdb/labels" "github.com/fabxc/tsdb/labels"
"github.com/pkg/errors" "github.com/pkg/errors"
@ -22,6 +23,8 @@ const (
MagicIndex = 0xBAAAD700 MagicIndex = 0xBAAAD700
) )
const compactionPageBytes = minSectorSize * 64
// SeriesWriter serializes a time block of chunked series data. // SeriesWriter serializes a time block of chunked series data.
type SeriesWriter interface { type SeriesWriter interface {
// WriteSeries writes the time series data chunks for a single series. // WriteSeries writes the time series data chunks for a single series.
@ -40,16 +43,18 @@ type SeriesWriter interface {
// seriesWriter implements the SeriesWriter interface for the standard // seriesWriter implements the SeriesWriter interface for the standard
// serialization format. // serialization format.
type seriesWriter struct { type seriesWriter struct {
w io.Writer ow io.Writer
n int64 w *ioutil.PageWriter
c int n int64
c int
index IndexWriter index IndexWriter
} }
func newSeriesWriter(w io.Writer, index IndexWriter) *seriesWriter { func newSeriesWriter(w io.Writer, index IndexWriter) *seriesWriter {
return &seriesWriter{ return &seriesWriter{
w: w, ow: w,
w: ioutil.NewPageWriter(w, compactionPageBytes, 0),
n: 0, n: 0,
index: index, index: index,
} }
@ -128,7 +133,7 @@ func (w *seriesWriter) Size() int64 {
} }
func (w *seriesWriter) Close() error { func (w *seriesWriter) Close() error {
return nil return w.w.Flush()
} }
// ChunkMeta holds information about a chunk of data. // ChunkMeta holds information about a chunk of data.
@ -178,8 +183,9 @@ type indexWriterSeries struct {
// indexWriter implements the IndexWriter interface for the standard // indexWriter implements the IndexWriter interface for the standard
// serialization format. // serialization format.
type indexWriter struct { type indexWriter struct {
w io.Writer ow io.Writer
n int64 w *ioutil.PageWriter
n int64
series map[uint32]*indexWriterSeries series map[uint32]*indexWriterSeries
@ -190,7 +196,8 @@ type indexWriter struct {
func newIndexWriter(w io.Writer) *indexWriter { func newIndexWriter(w io.Writer) *indexWriter {
return &indexWriter{ return &indexWriter{
w: w, w: ioutil.NewPageWriter(w, compactionPageBytes, 0),
ow: w,
n: 0, n: 0,
symbols: make(map[string]uint32, 4096), symbols: make(map[string]uint32, 4096),
series: make(map[uint32]*indexWriterSeries, 4096), series: make(map[uint32]*indexWriterSeries, 4096),
@ -489,5 +496,8 @@ func (w *indexWriter) finalize() error {
} }
func (w *indexWriter) Close() error { func (w *indexWriter) Close() error {
return w.finalize() if err := w.finalize(); err != nil {
return err
}
return w.w.Flush()
} }