Tom Wilkie
7 years ago
committed by
Tom Wilkie
parent
adf5307470
commit
92fcf375b0
@ -0,0 +1,223 @@ |
||||
// Copyright 2018 The Prometheus Authors
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
// Package promauto provides constructors for the usual Prometheus metrics that
|
||||
// return them already registered with the global registry
|
||||
// (prometheus.DefaultRegisterer). This allows very compact code, avoiding any
|
||||
// references to the registry altogether, but all the constructors in this
|
||||
// package will panic if the registration fails.
|
||||
//
|
||||
// The following example is a complete program to create a histogram of normally
|
||||
// distributed random numbers from the math/rand package:
|
||||
//
|
||||
// package main
|
||||
//
|
||||
// import (
|
||||
// "math/rand"
|
||||
// "net/http"
|
||||
//
|
||||
// "github.com/prometheus/client_golang/prometheus"
|
||||
// "github.com/prometheus/client_golang/prometheus/promauto"
|
||||
// "github.com/prometheus/client_golang/prometheus/promhttp"
|
||||
// )
|
||||
//
|
||||
// var histogram = promauto.NewHistogram(prometheus.HistogramOpts{
|
||||
// Name: "random_numbers",
|
||||
// Help: "A histogram of normally distributed random numbers.",
|
||||
// Buckets: prometheus.LinearBuckets(-3, .1, 61),
|
||||
// })
|
||||
//
|
||||
// func Random() {
|
||||
// for {
|
||||
// histogram.Observe(rand.NormFloat64())
|
||||
// }
|
||||
// }
|
||||
//
|
||||
// func main() {
|
||||
// go Random()
|
||||
// http.Handle("/metrics", promhttp.Handler())
|
||||
// http.ListenAndServe(":1971", nil)
|
||||
// }
|
||||
//
|
||||
// Prometheus's version of a minimal hello-world program:
|
||||
//
|
||||
// package main
|
||||
//
|
||||
// import (
|
||||
// "fmt"
|
||||
// "net/http"
|
||||
//
|
||||
// "github.com/prometheus/client_golang/prometheus"
|
||||
// "github.com/prometheus/client_golang/prometheus/promauto"
|
||||
// "github.com/prometheus/client_golang/prometheus/promhttp"
|
||||
// )
|
||||
//
|
||||
// func main() {
|
||||
// http.Handle("/", promhttp.InstrumentHandlerCounter(
|
||||
// promauto.NewCounterVec(
|
||||
// prometheus.CounterOpts{
|
||||
// Name: "hello_requests_total",
|
||||
// Help: "Total number of hello-world requests by HTTP code.",
|
||||
// },
|
||||
// []string{"code"},
|
||||
// ),
|
||||
// http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
// fmt.Fprint(w, "Hello, world!")
|
||||
// }),
|
||||
// ))
|
||||
// http.Handle("/metrics", promhttp.Handler())
|
||||
// http.ListenAndServe(":1971", nil)
|
||||
// }
|
||||
//
|
||||
// This appears very handy. So why are these constructors locked away in a
|
||||
// separate package? There are two caveats:
|
||||
//
|
||||
// First, in more complex programs, global state is often quite problematic.
|
||||
// That's the reason why the metrics constructors in the prometheus package do
|
||||
// not interact with the global prometheus.DefaultRegisterer on their own. You
|
||||
// are free to use the Register or MustRegister functions to register them with
|
||||
// the global prometheus.DefaultRegisterer, but you could as well choose a local
|
||||
// Registerer (usually created with prometheus.NewRegistry, but there are other
|
||||
// scenarios, e.g. testing).
|
||||
//
|
||||
// The second issue is that registration may fail, e.g. if a metric inconsistent
|
||||
// with the newly to be registered one is already registered. But how to signal
|
||||
// and handle a panic in the automatic registration with the default registry?
|
||||
// The only way is panicking. While panicking on invalid input provided by the
|
||||
// programmer is certainly fine, things are a bit more subtle in this case: You
|
||||
// might just add another package to the program, and that package (in its init
|
||||
// function) happens to register a metric with the same name as your code. Now,
|
||||
// all of a sudden, either your code or the code of the newly imported package
|
||||
// panics, depending on initialization order, without any opportunity to handle
|
||||
// the case gracefully. Even worse is a scenario where registration happens
|
||||
// later during the runtime (e.g. upon loading some kind of plugin), where the
|
||||
// panic could be triggered long after the code has been deployed to
|
||||
// production. A possibility to panic should be explicitly called out by the
|
||||
// Must… idiom, cf. prometheus.MustRegister. But adding a separate set of
|
||||
// constructors in the prometheus package called MustRegisterNewCounterVec or
|
||||
// similar would be quite unwieldy. Adding an extra MustRegister method to each
|
||||
// metric, returning the registered metric, would result in nice code for those
|
||||
// using the method, but would pollute every single metric interface for
|
||||
// everybody avoiding the global registry.
|
||||
//
|
||||
// To address both issues, the problematic auto-registering and possibly
|
||||
// panicking constructors are all in this package with a clear warning
|
||||
// ahead. And whoever cares about avoiding global state and possibly panicking
|
||||
// function calls can simply ignore the existence of the promauto package
|
||||
// altogether.
|
||||
//
|
||||
// A final note: There is a similar case in the net/http package of the standard
|
||||
// library. It has DefaultServeMux as a global instance of ServeMux, and the
|
||||
// Handle function acts on it, panicking if a handler for the same pattern has
|
||||
// already been registered. However, one might argue that the whole HTTP routing
|
||||
// is usually set up closely together in the same package or file, while
|
||||
// Prometheus metrics tend to be spread widely over the codebase, increasing the
|
||||
// chance of surprising registration failures. Furthermore, the use of global
|
||||
// state in net/http has been criticized widely, and some avoid it altogether.
|
||||
package promauto |
||||
|
||||
import "github.com/prometheus/client_golang/prometheus" |
||||
|
||||
// NewCounter works like the function of the same name in the prometheus package
|
||||
// but it automatically registers the Counter with the
|
||||
// prometheus.DefaultRegisterer. If the registration fails, NewCounter panics.
|
||||
func NewCounter(opts prometheus.CounterOpts) prometheus.Counter { |
||||
c := prometheus.NewCounter(opts) |
||||
prometheus.MustRegister(c) |
||||
return c |
||||
} |
||||
|
||||
// NewCounterVec works like the function of the same name in the prometheus
|
||||
// package but it automatically registers the CounterVec with the
|
||||
// prometheus.DefaultRegisterer. If the registration fails, NewCounterVec
|
||||
// panics.
|
||||
func NewCounterVec(opts prometheus.CounterOpts, labelNames []string) *prometheus.CounterVec { |
||||
c := prometheus.NewCounterVec(opts, labelNames) |
||||
prometheus.MustRegister(c) |
||||
return c |
||||
} |
||||
|
||||
// NewCounterFunc works like the function of the same name in the prometheus
|
||||
// package but it automatically registers the CounterFunc with the
|
||||
// prometheus.DefaultRegisterer. If the registration fails, NewCounterFunc
|
||||
// panics.
|
||||
func NewCounterFunc(opts prometheus.CounterOpts, function func() float64) prometheus.CounterFunc { |
||||
g := prometheus.NewCounterFunc(opts, function) |
||||
prometheus.MustRegister(g) |
||||
return g |
||||
} |
||||
|
||||
// NewGauge works like the function of the same name in the prometheus package
|
||||
// but it automatically registers the Gauge with the
|
||||
// prometheus.DefaultRegisterer. If the registration fails, NewGauge panics.
|
||||
func NewGauge(opts prometheus.GaugeOpts) prometheus.Gauge { |
||||
g := prometheus.NewGauge(opts) |
||||
prometheus.MustRegister(g) |
||||
return g |
||||
} |
||||
|
||||
// NewGaugeVec works like the function of the same name in the prometheus
|
||||
// package but it automatically registers the GaugeVec with the
|
||||
// prometheus.DefaultRegisterer. If the registration fails, NewGaugeVec panics.
|
||||
func NewGaugeVec(opts prometheus.GaugeOpts, labelNames []string) *prometheus.GaugeVec { |
||||
g := prometheus.NewGaugeVec(opts, labelNames) |
||||
prometheus.MustRegister(g) |
||||
return g |
||||
} |
||||
|
||||
// NewGaugeFunc works like the function of the same name in the prometheus
|
||||
// package but it automatically registers the GaugeFunc with the
|
||||
// prometheus.DefaultRegisterer. If the registration fails, NewGaugeFunc panics.
|
||||
func NewGaugeFunc(opts prometheus.GaugeOpts, function func() float64) prometheus.GaugeFunc { |
||||
g := prometheus.NewGaugeFunc(opts, function) |
||||
prometheus.MustRegister(g) |
||||
return g |
||||
} |
||||
|
||||
// NewSummary works like the function of the same name in the prometheus package
|
||||
// but it automatically registers the Summary with the
|
||||
// prometheus.DefaultRegisterer. If the registration fails, NewSummary panics.
|
||||
func NewSummary(opts prometheus.SummaryOpts) prometheus.Summary { |
||||
s := prometheus.NewSummary(opts) |
||||
prometheus.MustRegister(s) |
||||
return s |
||||
} |
||||
|
||||
// NewSummaryVec works like the function of the same name in the prometheus
|
||||
// package but it automatically registers the SummaryVec with the
|
||||
// prometheus.DefaultRegisterer. If the registration fails, NewSummaryVec
|
||||
// panics.
|
||||
func NewSummaryVec(opts prometheus.SummaryOpts, labelNames []string) *prometheus.SummaryVec { |
||||
s := prometheus.NewSummaryVec(opts, labelNames) |
||||
prometheus.MustRegister(s) |
||||
return s |
||||
} |
||||
|
||||
// NewHistogram works like the function of the same name in the prometheus
|
||||
// package but it automatically registers the Histogram with the
|
||||
// prometheus.DefaultRegisterer. If the registration fails, NewHistogram panics.
|
||||
func NewHistogram(opts prometheus.HistogramOpts) prometheus.Histogram { |
||||
h := prometheus.NewHistogram(opts) |
||||
prometheus.MustRegister(h) |
||||
return h |
||||
} |
||||
|
||||
// NewHistogramVec works like the function of the same name in the prometheus
|
||||
// package but it automatically registers the HistogramVec with the
|
||||
// prometheus.DefaultRegisterer. If the registration fails, NewHistogramVec
|
||||
// panics.
|
||||
func NewHistogramVec(opts prometheus.HistogramOpts, labelNames []string) *prometheus.HistogramVec { |
||||
h := prometheus.NewHistogramVec(opts, labelNames) |
||||
prometheus.MustRegister(h) |
||||
return h |
||||
} |
||||
@ -1,2 +0,0 @@ |
||||
# Enable only "legacy" staticcheck verifications. |
||||
checks = [ "SA*" ] |
||||
@ -0,0 +1,284 @@ |
||||
// Copyright 2019 The Prometheus Authors
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package wal |
||||
|
||||
import ( |
||||
"encoding/binary" |
||||
"fmt" |
||||
"hash/crc32" |
||||
"io" |
||||
|
||||
"github.com/go-kit/kit/log" |
||||
"github.com/go-kit/kit/log/level" |
||||
"github.com/pkg/errors" |
||||
"github.com/prometheus/client_golang/prometheus" |
||||
"github.com/prometheus/client_golang/prometheus/promauto" |
||||
) |
||||
|
||||
var ( |
||||
readerCorruptionErrors = promauto.NewCounterVec(prometheus.CounterOpts{ |
||||
Name: "prometheus_tsdb_wal_reader_corruption_errors", |
||||
Help: "Errors encountered when reading the WAL.", |
||||
}, []string{"error"}) |
||||
) |
||||
|
||||
// NewLiveReader returns a new live reader.
|
||||
func NewLiveReader(logger log.Logger, r io.Reader) *LiveReader { |
||||
return &LiveReader{ |
||||
logger: logger, |
||||
rdr: r, |
||||
|
||||
// Until we understand how they come about, make readers permissive
|
||||
// to records spanning pages.
|
||||
permissive: true, |
||||
} |
||||
} |
||||
|
||||
// LiveReader reads WAL records from an io.Reader. It allows reading of WALs
|
||||
// that are still in the process of being written, and returns records as soon
|
||||
// as they can be read.
|
||||
type LiveReader struct { |
||||
logger log.Logger |
||||
rdr io.Reader |
||||
err error |
||||
rec []byte |
||||
hdr [recordHeaderSize]byte |
||||
buf [pageSize]byte |
||||
readIndex int // Index in buf to start at for next read.
|
||||
writeIndex int // Index in buf to start at for next write.
|
||||
total int64 // Total bytes processed during reading in calls to Next().
|
||||
index int // Used to track partial records, should be 0 at the start of every new record.
|
||||
|
||||
// For testing, we can treat EOF as a non-error.
|
||||
eofNonErr bool |
||||
|
||||
// We sometime see records span page boundaries. Should never happen, but it
|
||||
// does. Until we track down why, set permissive to true to tolerate it.
|
||||
// NB the non-ive Reader implementation allows for this.
|
||||
permissive bool |
||||
} |
||||
|
||||
// Err returns any errors encountered reading the WAL. io.EOFs are not terminal
|
||||
// and Next can be tried again. Non-EOFs are terminal, and the reader should
|
||||
// not be used again. It is up to the user to decide when to stop trying should
|
||||
// io.EOF be returned.
|
||||
func (r *LiveReader) Err() error { |
||||
if r.eofNonErr && r.err == io.EOF { |
||||
return nil |
||||
} |
||||
return r.err |
||||
} |
||||
|
||||
// Offset returns the number of bytes consumed from this segment.
|
||||
func (r *LiveReader) Offset() int64 { |
||||
return r.total |
||||
} |
||||
|
||||
func (r *LiveReader) fillBuffer() (int, error) { |
||||
n, err := r.rdr.Read(r.buf[r.writeIndex:len(r.buf)]) |
||||
r.writeIndex += n |
||||
return n, err |
||||
} |
||||
|
||||
// Next returns true if Record() will contain a full record.
|
||||
// If Next returns false, you should always checked the contents of Error().
|
||||
// Return false guarantees there are no more records if the segment is closed
|
||||
// and not corrupt, otherwise if Err() == io.EOF you should try again when more
|
||||
// data has been written.
|
||||
func (r *LiveReader) Next() bool { |
||||
for { |
||||
// If buildRecord returns a non-EOF error, its game up - the segment is
|
||||
// corrupt. If buildRecord returns an EOF, we try and read more in
|
||||
// fillBuffer later on. If that fails to read anything (n=0 && err=EOF),
|
||||
// we return EOF and the user can try again later. If we have a full
|
||||
// page, buildRecord is guaranteed to return a record or a non-EOF; it
|
||||
// has checks the records fit in pages.
|
||||
if ok, err := r.buildRecord(); ok { |
||||
return true |
||||
} else if err != nil && err != io.EOF { |
||||
r.err = err |
||||
return false |
||||
} |
||||
|
||||
// If we've filled the page and not found a record, this
|
||||
// means records have started to span pages. Shouldn't happen
|
||||
// but does and until we found out why, we need to deal with this.
|
||||
if r.permissive && r.writeIndex == pageSize && r.readIndex > 0 { |
||||
copy(r.buf[:], r.buf[r.readIndex:]) |
||||
r.writeIndex -= r.readIndex |
||||
r.readIndex = 0 |
||||
continue |
||||
} |
||||
|
||||
if r.readIndex == pageSize { |
||||
r.writeIndex = 0 |
||||
r.readIndex = 0 |
||||
} |
||||
|
||||
if r.writeIndex != pageSize { |
||||
n, err := r.fillBuffer() |
||||
if n == 0 || (err != nil && err != io.EOF) { |
||||
r.err = err |
||||
return false |
||||
} |
||||
} |
||||
} |
||||
} |
||||
|
||||
// Record returns the current record.
|
||||
// The returned byte slice is only valid until the next call to Next.
|
||||
func (r *LiveReader) Record() []byte { |
||||
return r.rec |
||||
} |
||||
|
||||
// Rebuild a full record from potentially partial records. Returns false
|
||||
// if there was an error or if we weren't able to read a record for any reason.
|
||||
// Returns true if we read a full record. Any record data is appended to
|
||||
// LiveReader.rec
|
||||
func (r *LiveReader) buildRecord() (bool, error) { |
||||
for { |
||||
// Check that we have data in the internal buffer to read.
|
||||
if r.writeIndex <= r.readIndex { |
||||
return false, nil |
||||
} |
||||
|
||||
// Attempt to read a record, partial or otherwise.
|
||||
temp, n, err := r.readRecord() |
||||
if err != nil { |
||||
return false, err |
||||
} |
||||
|
||||
r.readIndex += n |
||||
r.total += int64(n) |
||||
if temp == nil { |
||||
return false, nil |
||||
} |
||||
|
||||
rt := recType(r.hdr[0]) |
||||
if rt == recFirst || rt == recFull { |
||||
r.rec = r.rec[:0] |
||||
} |
||||
r.rec = append(r.rec, temp...) |
||||
|
||||
if err := validateRecord(rt, r.index); err != nil { |
||||
r.index = 0 |
||||
return false, err |
||||
} |
||||
if rt == recLast || rt == recFull { |
||||
r.index = 0 |
||||
return true, nil |
||||
} |
||||
// Only increment i for non-zero records since we use it
|
||||
// to determine valid content record sequences.
|
||||
r.index++ |
||||
} |
||||
} |
||||
|
||||
// Returns an error if the recType and i indicate an invalid record sequence.
|
||||
// As an example, if i is > 0 because we've read some amount of a partial record
|
||||
// (recFirst, recMiddle, etc. but not recLast) and then we get another recFirst or recFull
|
||||
// instead of a recLast or recMiddle we would have an invalid record.
|
||||
func validateRecord(typ recType, i int) error { |
||||
switch typ { |
||||
case recFull: |
||||
if i != 0 { |
||||
return errors.New("unexpected full record") |
||||
} |
||||
return nil |
||||
case recFirst: |
||||
if i != 0 { |
||||
return errors.New("unexpected first record, dropping buffer") |
||||
} |
||||
return nil |
||||
case recMiddle: |
||||
if i == 0 { |
||||
return errors.New("unexpected middle record, dropping buffer") |
||||
} |
||||
return nil |
||||
case recLast: |
||||
if i == 0 { |
||||
return errors.New("unexpected last record, dropping buffer") |
||||
} |
||||
return nil |
||||
default: |
||||
return errors.Errorf("unexpected record type %d", typ) |
||||
} |
||||
} |
||||
|
||||
// Read a sub-record (see recType) from the buffer. It could potentially
|
||||
// be a full record (recFull) if the record fits within the bounds of a single page.
|
||||
// Returns a byte slice of the record data read, the number of bytes read, and an error
|
||||
// if there's a non-zero byte in a page term record or the record checksum fails.
|
||||
// This is a non-method function to make it clear it does not mutate the reader.
|
||||
func (r *LiveReader) readRecord() ([]byte, int, error) { |
||||
// Special case: for recPageTerm, check that are all zeros to end of page,
|
||||
// consume them but don't return them.
|
||||
if r.buf[r.readIndex] == byte(recPageTerm) { |
||||
// End of page won't necessarily be end of buffer, as we may have
|
||||
// got misaligned by records spanning page boundaries.
|
||||
// r.total % pageSize is the offset into the current page
|
||||
// that r.readIndex points to in buf. Therefore
|
||||
// pageSize - (r.total % pageSize) is the amount left to read of
|
||||
// the current page.
|
||||
remaining := int(pageSize - (r.total % pageSize)) |
||||
if r.readIndex+remaining > r.writeIndex { |
||||
return nil, 0, io.EOF |
||||
} |
||||
|
||||
for i := r.readIndex; i < r.readIndex+remaining; i++ { |
||||
if r.buf[i] != 0 { |
||||
return nil, 0, errors.New("unexpected non-zero byte in page term bytes") |
||||
} |
||||
} |
||||
|
||||
return nil, remaining, nil |
||||
} |
||||
|
||||
// Not a recPageTerm; read the record and check the checksum.
|
||||
if r.writeIndex-r.readIndex < recordHeaderSize { |
||||
return nil, 0, io.EOF |
||||
} |
||||
|
||||
copy(r.hdr[:], r.buf[r.readIndex:r.readIndex+recordHeaderSize]) |
||||
length := int(binary.BigEndian.Uint16(r.hdr[1:])) |
||||
crc := binary.BigEndian.Uint32(r.hdr[3:]) |
||||
if r.readIndex+recordHeaderSize+length > pageSize { |
||||
if !r.permissive { |
||||
return nil, 0, fmt.Errorf("record would overflow current page: %d > %d", r.readIndex+recordHeaderSize+length, pageSize) |
||||
} |
||||
readerCorruptionErrors.WithLabelValues("record_span_page").Inc() |
||||
level.Warn(r.logger).Log("msg", "record spans page boundaries", "start", r.readIndex, "end", recordHeaderSize+length, "pageSize", pageSize) |
||||
} |
||||
if recordHeaderSize+length > pageSize { |
||||
return nil, 0, fmt.Errorf("record length greater than a single page: %d > %d", recordHeaderSize+length, pageSize) |
||||
} |
||||
if r.readIndex+recordHeaderSize+length > r.writeIndex { |
||||
return nil, 0, io.EOF |
||||
} |
||||
|
||||
rec := r.buf[r.readIndex+recordHeaderSize : r.readIndex+recordHeaderSize+length] |
||||
if c := crc32.Checksum(rec, castagnoliTable); c != crc { |
||||
return nil, 0, errors.Errorf("unexpected checksum %x, expected %x", c, crc) |
||||
} |
||||
|
||||
return rec, length + recordHeaderSize, nil |
||||
} |
||||
|
||||
func min(i, j int) int { |
||||
if i < j { |
||||
return i |
||||
} |
||||
return j |
||||
} |
||||
@ -0,0 +1,183 @@ |
||||
// Copyright 2019 The Prometheus Authors
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package wal |
||||
|
||||
import ( |
||||
"encoding/binary" |
||||
"hash/crc32" |
||||
"io" |
||||
|
||||
"github.com/pkg/errors" |
||||
) |
||||
|
||||
// Reader reads WAL records from an io.Reader.
|
||||
type Reader struct { |
||||
rdr io.Reader |
||||
err error |
||||
rec []byte |
||||
buf [pageSize]byte |
||||
total int64 // Total bytes processed.
|
||||
curRecTyp recType // Used for checking that the last record is not torn.
|
||||
} |
||||
|
||||
// NewReader returns a new reader.
|
||||
func NewReader(r io.Reader) *Reader { |
||||
return &Reader{rdr: r} |
||||
} |
||||
|
||||
// Next advances the reader to the next records and returns true if it exists.
|
||||
// It must not be called again after it returned false.
|
||||
func (r *Reader) Next() bool { |
||||
err := r.next() |
||||
if errors.Cause(err) == io.EOF { |
||||
// The last WAL segment record shouldn't be torn(should be full or last).
|
||||
// The last record would be torn after a crash just before
|
||||
// the last record part could be persisted to disk.
|
||||
if recType(r.curRecTyp) == recFirst || recType(r.curRecTyp) == recMiddle { |
||||
r.err = errors.New("last record is torn") |
||||
} |
||||
return false |
||||
} |
||||
r.err = err |
||||
return r.err == nil |
||||
} |
||||
|
||||
func (r *Reader) next() (err error) { |
||||
// We have to use r.buf since allocating byte arrays here fails escape
|
||||
// analysis and ends up on the heap, even though it seemingly should not.
|
||||
hdr := r.buf[:recordHeaderSize] |
||||
buf := r.buf[recordHeaderSize:] |
||||
|
||||
r.rec = r.rec[:0] |
||||
|
||||
i := 0 |
||||
for { |
||||
if _, err = io.ReadFull(r.rdr, hdr[:1]); err != nil { |
||||
return errors.Wrap(err, "read first header byte") |
||||
} |
||||
r.total++ |
||||
r.curRecTyp = recType(hdr[0]) |
||||
|
||||
// Gobble up zero bytes.
|
||||
if r.curRecTyp == recPageTerm { |
||||
// recPageTerm is a single byte that indicates the rest of the page is padded.
|
||||
// If it's the first byte in a page, buf is too small and
|
||||
// needs to be resized to fit pageSize-1 bytes.
|
||||
buf = r.buf[1:] |
||||
|
||||
// We are pedantic and check whether the zeros are actually up
|
||||
// to a page boundary.
|
||||
// It's not strictly necessary but may catch sketchy state early.
|
||||
k := pageSize - (r.total % pageSize) |
||||
if k == pageSize { |
||||
continue // Initial 0 byte was last page byte.
|
||||
} |
||||
n, err := io.ReadFull(r.rdr, buf[:k]) |
||||
if err != nil { |
||||
return errors.Wrap(err, "read remaining zeros") |
||||
} |
||||
r.total += int64(n) |
||||
|
||||
for _, c := range buf[:k] { |
||||
if c != 0 { |
||||
return errors.New("unexpected non-zero byte in padded page") |
||||
} |
||||
} |
||||
continue |
||||
} |
||||
n, err := io.ReadFull(r.rdr, hdr[1:]) |
||||
if err != nil { |
||||
return errors.Wrap(err, "read remaining header") |
||||
} |
||||
r.total += int64(n) |
||||
|
||||
var ( |
||||
length = binary.BigEndian.Uint16(hdr[1:]) |
||||
crc = binary.BigEndian.Uint32(hdr[3:]) |
||||
) |
||||
|
||||
if length > pageSize-recordHeaderSize { |
||||
return errors.Errorf("invalid record size %d", length) |
||||
} |
||||
n, err = io.ReadFull(r.rdr, buf[:length]) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
r.total += int64(n) |
||||
|
||||
if n != int(length) { |
||||
return errors.Errorf("invalid size: expected %d, got %d", length, n) |
||||
} |
||||
if c := crc32.Checksum(buf[:length], castagnoliTable); c != crc { |
||||
return errors.Errorf("unexpected checksum %x, expected %x", c, crc) |
||||
} |
||||
r.rec = append(r.rec, buf[:length]...) |
||||
|
||||
if err := validateRecord(r.curRecTyp, i); err != nil { |
||||
return err |
||||
} |
||||
if r.curRecTyp == recLast || r.curRecTyp == recFull { |
||||
return nil |
||||
} |
||||
|
||||
// Only increment i for non-zero records since we use it
|
||||
// to determine valid content record sequences.
|
||||
i++ |
||||
} |
||||
} |
||||
|
||||
// Err returns the last encountered error wrapped in a corruption error.
|
||||
// If the reader does not allow to infer a segment index and offset, a total
|
||||
// offset in the reader stream will be provided.
|
||||
func (r *Reader) Err() error { |
||||
if r.err == nil { |
||||
return nil |
||||
} |
||||
if b, ok := r.rdr.(*segmentBufReader); ok { |
||||
return &CorruptionErr{ |
||||
Err: r.err, |
||||
Dir: b.segs[b.cur].Dir(), |
||||
Segment: b.segs[b.cur].Index(), |
||||
Offset: int64(b.off), |
||||
} |
||||
} |
||||
return &CorruptionErr{ |
||||
Err: r.err, |
||||
Segment: -1, |
||||
Offset: r.total, |
||||
} |
||||
} |
||||
|
||||
// Record returns the current record. The returned byte slice is only
|
||||
// valid until the next call to Next.
|
||||
func (r *Reader) Record() []byte { |
||||
return r.rec |
||||
} |
||||
|
||||
// Segment returns the current segment being read.
|
||||
func (r *Reader) Segment() int { |
||||
if b, ok := r.rdr.(*segmentBufReader); ok { |
||||
return b.segs[b.cur].Index() |
||||
} |
||||
return -1 |
||||
} |
||||
|
||||
// Offset returns the current position of the segment being read.
|
||||
func (r *Reader) Offset() int64 { |
||||
if b, ok := r.rdr.(*segmentBufReader); ok { |
||||
return int64(b.off) |
||||
} |
||||
return r.total |
||||
} |
||||
Loading…
Reference in new issue