mirror of
https://github.com/prometheus/prometheus.git
synced 2025-01-12 14:27:27 -08:00
48d25e6fe7
Rather than keeping the offset of each postings list, instead keep the nth offset of the offset of the posting list. As postings list offsets have always been sorted, we can then get to the closest entry before the one we want an iterate forwards. I haven't done much tuning on the 32 number, it was chosen to try not to read through more than a 4k page of data. Switch to a bulk interface for fetching postings. Use it to avoid having to re-read parts of the posting offset table when querying lots of it. For a index with what BenchmarkHeadPostingForMatchers uses RAM for r.postings drops from 3.79MB to 80.19kB or about 48x. Bytes allocated go down by 30%, and suprisingly CPU usage drops by 4-6% for typical queries too. benchmark old ns/op new ns/op delta BenchmarkPostingsForMatchers/Block/n="1"-4 35231 36673 +4.09% BenchmarkPostingsForMatchers/Block/n="1",j="foo"-4 563380 540627 -4.04% BenchmarkPostingsForMatchers/Block/j="foo",n="1"-4 536782 534186 -0.48% BenchmarkPostingsForMatchers/Block/n="1",j!="foo"-4 533990 541550 +1.42% BenchmarkPostingsForMatchers/Block/i=~".*"-4 113374598 117969608 +4.05% BenchmarkPostingsForMatchers/Block/i=~".+"-4 146329884 139651442 -4.56% BenchmarkPostingsForMatchers/Block/i=~""-4 50346510 44961127 -10.70% BenchmarkPostingsForMatchers/Block/i!=""-4 41261550 35356165 -14.31% BenchmarkPostingsForMatchers/Block/n="1",i=~".*",j="foo"-4 112544418 116904010 +3.87% BenchmarkPostingsForMatchers/Block/n="1",i=~".*",i!="2",j="foo"-4 112487086 116864918 +3.89% BenchmarkPostingsForMatchers/Block/n="1",i!=""-4 41094758 35457904 -13.72% BenchmarkPostingsForMatchers/Block/n="1",i!="",j="foo"-4 41906372 36151473 -13.73% BenchmarkPostingsForMatchers/Block/n="1",i=~".+",j="foo"-4 147262414 140424800 -4.64% BenchmarkPostingsForMatchers/Block/n="1",i=~"1.+",j="foo"-4 28615629 27872072 -2.60% BenchmarkPostingsForMatchers/Block/n="1",i=~".+",i!="2",j="foo"-4 147117177 140462403 -4.52% BenchmarkPostingsForMatchers/Block/n="1",i=~".+",i!~"2.*",j="foo"-4 175096826 167902298 -4.11% benchmark old allocs new allocs delta BenchmarkPostingsForMatchers/Block/n="1"-4 4 6 +50.00% BenchmarkPostingsForMatchers/Block/n="1",j="foo"-4 7 11 +57.14% BenchmarkPostingsForMatchers/Block/j="foo",n="1"-4 7 11 +57.14% BenchmarkPostingsForMatchers/Block/n="1",j!="foo"-4 15 17 +13.33% BenchmarkPostingsForMatchers/Block/i=~".*"-4 100010 100012 +0.00% BenchmarkPostingsForMatchers/Block/i=~".+"-4 200069 200040 -0.01% BenchmarkPostingsForMatchers/Block/i=~""-4 200072 200045 -0.01% BenchmarkPostingsForMatchers/Block/i!=""-4 200070 200041 -0.01% BenchmarkPostingsForMatchers/Block/n="1",i=~".*",j="foo"-4 100013 100017 +0.00% BenchmarkPostingsForMatchers/Block/n="1",i=~".*",i!="2",j="foo"-4 100017 100023 +0.01% BenchmarkPostingsForMatchers/Block/n="1",i!=""-4 200073 200046 -0.01% BenchmarkPostingsForMatchers/Block/n="1",i!="",j="foo"-4 200075 200050 -0.01% BenchmarkPostingsForMatchers/Block/n="1",i=~".+",j="foo"-4 200074 200049 -0.01% BenchmarkPostingsForMatchers/Block/n="1",i=~"1.+",j="foo"-4 111165 111150 -0.01% BenchmarkPostingsForMatchers/Block/n="1",i=~".+",i!="2",j="foo"-4 200078 200055 -0.01% BenchmarkPostingsForMatchers/Block/n="1",i=~".+",i!~"2.*",j="foo"-4 311282 311238 -0.01% benchmark old bytes new bytes delta BenchmarkPostingsForMatchers/Block/n="1"-4 264 296 +12.12% BenchmarkPostingsForMatchers/Block/n="1",j="foo"-4 360 424 +17.78% BenchmarkPostingsForMatchers/Block/j="foo",n="1"-4 360 424 +17.78% BenchmarkPostingsForMatchers/Block/n="1",j!="foo"-4 520 552 +6.15% BenchmarkPostingsForMatchers/Block/i=~".*"-4 1600461 1600482 +0.00% BenchmarkPostingsForMatchers/Block/i=~".+"-4 24900801 17259077 -30.69% BenchmarkPostingsForMatchers/Block/i=~""-4 24900836 17259151 -30.69% BenchmarkPostingsForMatchers/Block/i!=""-4 24900760 17259048 -30.69% BenchmarkPostingsForMatchers/Block/n="1",i=~".*",j="foo"-4 1600557 1600621 +0.00% BenchmarkPostingsForMatchers/Block/n="1",i=~".*",i!="2",j="foo"-4 1600717 1600813 +0.01% BenchmarkPostingsForMatchers/Block/n="1",i!=""-4 24900856 17259176 -30.69% BenchmarkPostingsForMatchers/Block/n="1",i!="",j="foo"-4 24900952 17259304 -30.69% BenchmarkPostingsForMatchers/Block/n="1",i=~".+",j="foo"-4 24900993 17259333 -30.69% BenchmarkPostingsForMatchers/Block/n="1",i=~"1.+",j="foo"-4 3788311 3142630 -17.04% BenchmarkPostingsForMatchers/Block/n="1",i=~".+",i!="2",j="foo"-4 24901137 17259509 -30.69% BenchmarkPostingsForMatchers/Block/n="1",i=~".+",i!~"2.*",j="foo"-4 28693086 20405680 -28.88% Signed-off-by: Brian Brazil <brian.brazil@robustperception.io>
1241 lines
27 KiB
Go
1241 lines
27 KiB
Go
// Copyright 2017 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 tsdb
|
|
|
|
import (
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"unicode/utf8"
|
|
|
|
"github.com/pkg/errors"
|
|
"github.com/prometheus/prometheus/pkg/labels"
|
|
"github.com/prometheus/prometheus/tsdb/chunkenc"
|
|
"github.com/prometheus/prometheus/tsdb/chunks"
|
|
tsdb_errors "github.com/prometheus/prometheus/tsdb/errors"
|
|
"github.com/prometheus/prometheus/tsdb/index"
|
|
"github.com/prometheus/prometheus/tsdb/tombstones"
|
|
)
|
|
|
|
// Querier provides querying access over time series data of a fixed
|
|
// time range.
|
|
type Querier interface {
|
|
// Select returns a set of series that matches the given label matchers.
|
|
Select(...*labels.Matcher) (SeriesSet, error)
|
|
|
|
// LabelValues returns all potential values for a label name.
|
|
LabelValues(string) ([]string, error)
|
|
|
|
// LabelValuesFor returns all potential values for a label name.
|
|
// under the constraint of another label.
|
|
LabelValuesFor(string, labels.Label) ([]string, error)
|
|
|
|
// LabelNames returns all the unique label names present in the block in sorted order.
|
|
LabelNames() ([]string, error)
|
|
|
|
// Close releases the resources of the Querier.
|
|
Close() error
|
|
}
|
|
|
|
// Series exposes a single time series.
|
|
type Series interface {
|
|
// Labels returns the complete set of labels identifying the series.
|
|
Labels() labels.Labels
|
|
|
|
// Iterator returns a new iterator of the data of the series.
|
|
Iterator() SeriesIterator
|
|
}
|
|
|
|
// querier aggregates querying results from time blocks within
|
|
// a single partition.
|
|
type querier struct {
|
|
blocks []Querier
|
|
}
|
|
|
|
func (q *querier) LabelValues(n string) ([]string, error) {
|
|
return q.lvals(q.blocks, n)
|
|
}
|
|
|
|
// LabelNames returns all the unique label names present querier blocks.
|
|
func (q *querier) LabelNames() ([]string, error) {
|
|
labelNamesMap := make(map[string]struct{})
|
|
for _, b := range q.blocks {
|
|
names, err := b.LabelNames()
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "LabelNames() from Querier")
|
|
}
|
|
for _, name := range names {
|
|
labelNamesMap[name] = struct{}{}
|
|
}
|
|
}
|
|
|
|
labelNames := make([]string, 0, len(labelNamesMap))
|
|
for name := range labelNamesMap {
|
|
labelNames = append(labelNames, name)
|
|
}
|
|
sort.Strings(labelNames)
|
|
|
|
return labelNames, nil
|
|
}
|
|
|
|
func (q *querier) lvals(qs []Querier, n string) ([]string, error) {
|
|
if len(qs) == 0 {
|
|
return nil, nil
|
|
}
|
|
if len(qs) == 1 {
|
|
return qs[0].LabelValues(n)
|
|
}
|
|
l := len(qs) / 2
|
|
s1, err := q.lvals(qs[:l], n)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
s2, err := q.lvals(qs[l:], n)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return mergeStrings(s1, s2), nil
|
|
}
|
|
|
|
func (q *querier) LabelValuesFor(string, labels.Label) ([]string, error) {
|
|
return nil, fmt.Errorf("not implemented")
|
|
}
|
|
|
|
func (q *querier) Select(ms ...*labels.Matcher) (SeriesSet, error) {
|
|
if len(q.blocks) == 0 {
|
|
return EmptySeriesSet(), nil
|
|
}
|
|
ss := make([]SeriesSet, len(q.blocks))
|
|
var s SeriesSet
|
|
var err error
|
|
for i, b := range q.blocks {
|
|
s, err = b.Select(ms...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
ss[i] = s
|
|
}
|
|
|
|
return NewMergedSeriesSet(ss), nil
|
|
}
|
|
|
|
func (q *querier) Close() error {
|
|
var merr tsdb_errors.MultiError
|
|
|
|
for _, bq := range q.blocks {
|
|
merr.Add(bq.Close())
|
|
}
|
|
return merr.Err()
|
|
}
|
|
|
|
// verticalQuerier aggregates querying results from time blocks within
|
|
// a single partition. The block time ranges can be overlapping.
|
|
type verticalQuerier struct {
|
|
querier
|
|
}
|
|
|
|
func (q *verticalQuerier) Select(ms ...*labels.Matcher) (SeriesSet, error) {
|
|
return q.sel(q.blocks, ms)
|
|
}
|
|
|
|
func (q *verticalQuerier) sel(qs []Querier, ms []*labels.Matcher) (SeriesSet, error) {
|
|
if len(qs) == 0 {
|
|
return EmptySeriesSet(), nil
|
|
}
|
|
if len(qs) == 1 {
|
|
return qs[0].Select(ms...)
|
|
}
|
|
l := len(qs) / 2
|
|
|
|
a, err := q.sel(qs[:l], ms)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
b, err := q.sel(qs[l:], ms)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return newMergedVerticalSeriesSet(a, b), nil
|
|
}
|
|
|
|
// NewBlockQuerier returns a querier against the reader.
|
|
func NewBlockQuerier(b BlockReader, mint, maxt int64) (Querier, error) {
|
|
indexr, err := b.Index()
|
|
if err != nil {
|
|
return nil, errors.Wrapf(err, "open index reader")
|
|
}
|
|
chunkr, err := b.Chunks()
|
|
if err != nil {
|
|
indexr.Close()
|
|
return nil, errors.Wrapf(err, "open chunk reader")
|
|
}
|
|
tombsr, err := b.Tombstones()
|
|
if err != nil {
|
|
indexr.Close()
|
|
chunkr.Close()
|
|
return nil, errors.Wrapf(err, "open tombstone reader")
|
|
}
|
|
return &blockQuerier{
|
|
mint: mint,
|
|
maxt: maxt,
|
|
index: indexr,
|
|
chunks: chunkr,
|
|
tombstones: tombsr,
|
|
}, nil
|
|
}
|
|
|
|
// blockQuerier provides querying access to a single block database.
|
|
type blockQuerier struct {
|
|
index IndexReader
|
|
chunks ChunkReader
|
|
tombstones tombstones.Reader
|
|
|
|
closed bool
|
|
|
|
mint, maxt int64
|
|
}
|
|
|
|
func (q *blockQuerier) Select(ms ...*labels.Matcher) (SeriesSet, error) {
|
|
base, err := LookupChunkSeries(q.index, q.tombstones, ms...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &blockSeriesSet{
|
|
set: &populatedChunkSeries{
|
|
set: base,
|
|
chunks: q.chunks,
|
|
mint: q.mint,
|
|
maxt: q.maxt,
|
|
},
|
|
|
|
mint: q.mint,
|
|
maxt: q.maxt,
|
|
}, nil
|
|
}
|
|
|
|
func (q *blockQuerier) LabelValues(name string) ([]string, error) {
|
|
tpls, err := q.index.LabelValues(name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
res := make([]string, 0, tpls.Len())
|
|
|
|
for i := 0; i < tpls.Len(); i++ {
|
|
vals, err := tpls.At(i)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
res = append(res, vals[0])
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
func (q *blockQuerier) LabelNames() ([]string, error) {
|
|
return q.index.LabelNames()
|
|
}
|
|
|
|
func (q *blockQuerier) LabelValuesFor(string, labels.Label) ([]string, error) {
|
|
return nil, fmt.Errorf("not implemented")
|
|
}
|
|
|
|
func (q *blockQuerier) Close() error {
|
|
if q.closed {
|
|
return errors.New("block querier already closed")
|
|
}
|
|
|
|
var merr tsdb_errors.MultiError
|
|
merr.Add(q.index.Close())
|
|
merr.Add(q.chunks.Close())
|
|
merr.Add(q.tombstones.Close())
|
|
q.closed = true
|
|
return merr.Err()
|
|
}
|
|
|
|
// Bitmap used by func isRegexMetaCharacter to check whether a character needs to be escaped.
|
|
var regexMetaCharacterBytes [16]byte
|
|
|
|
// isRegexMetaCharacter reports whether byte b needs to be escaped.
|
|
func isRegexMetaCharacter(b byte) bool {
|
|
return b < utf8.RuneSelf && regexMetaCharacterBytes[b%16]&(1<<(b/16)) != 0
|
|
}
|
|
|
|
func init() {
|
|
for _, b := range []byte(`.+*?()|[]{}^$`) {
|
|
regexMetaCharacterBytes[b%16] |= 1 << (b / 16)
|
|
}
|
|
}
|
|
|
|
func findSetMatches(pattern string) []string {
|
|
// Return empty matches if the wrapper from Prometheus is missing.
|
|
if len(pattern) < 6 || pattern[:4] != "^(?:" || pattern[len(pattern)-2:] != ")$" {
|
|
return nil
|
|
}
|
|
escaped := false
|
|
sets := []*strings.Builder{{}}
|
|
for i := 4; i < len(pattern)-2; i++ {
|
|
if escaped {
|
|
switch {
|
|
case isRegexMetaCharacter(pattern[i]):
|
|
sets[len(sets)-1].WriteByte(pattern[i])
|
|
case pattern[i] == '\\':
|
|
sets[len(sets)-1].WriteByte('\\')
|
|
default:
|
|
return nil
|
|
}
|
|
escaped = false
|
|
} else {
|
|
switch {
|
|
case isRegexMetaCharacter(pattern[i]):
|
|
if pattern[i] == '|' {
|
|
sets = append(sets, &strings.Builder{})
|
|
} else {
|
|
return nil
|
|
}
|
|
case pattern[i] == '\\':
|
|
escaped = true
|
|
default:
|
|
sets[len(sets)-1].WriteByte(pattern[i])
|
|
}
|
|
}
|
|
}
|
|
matches := make([]string, 0, len(sets))
|
|
for _, s := range sets {
|
|
if s.Len() > 0 {
|
|
matches = append(matches, s.String())
|
|
}
|
|
}
|
|
return matches
|
|
}
|
|
|
|
// PostingsForMatchers assembles a single postings iterator against the index reader
|
|
// based on the given matchers.
|
|
func PostingsForMatchers(ix IndexReader, ms ...*labels.Matcher) (index.Postings, error) {
|
|
var its, notIts []index.Postings
|
|
// See which label must be non-empty.
|
|
// Optimization for case like {l=~".", l!="1"}.
|
|
labelMustBeSet := make(map[string]bool, len(ms))
|
|
for _, m := range ms {
|
|
if !m.Matches("") {
|
|
labelMustBeSet[m.Name] = true
|
|
}
|
|
}
|
|
|
|
for _, m := range ms {
|
|
if labelMustBeSet[m.Name] {
|
|
// If this matcher must be non-empty, we can be smarter.
|
|
matchesEmpty := m.Matches("")
|
|
isNot := m.Type == labels.MatchNotEqual || m.Type == labels.MatchNotRegexp
|
|
if isNot && matchesEmpty { // l!="foo"
|
|
// If the label can't be empty and is a Not and the inner matcher
|
|
// doesn't match empty, then subtract it out at the end.
|
|
inverse, err := m.Inverse()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
it, err := postingsForMatcher(ix, inverse)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
notIts = append(notIts, it)
|
|
} else if isNot && !matchesEmpty { // l!=""
|
|
// If the label can't be empty and is a Not, but the inner matcher can
|
|
// be empty we need to use inversePostingsForMatcher.
|
|
inverse, err := m.Inverse()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
it, err := inversePostingsForMatcher(ix, inverse)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
its = append(its, it)
|
|
} else { // l="a"
|
|
// Non-Not matcher, use normal postingsForMatcher.
|
|
it, err := postingsForMatcher(ix, m)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
its = append(its, it)
|
|
}
|
|
} else { // l=""
|
|
// If the matchers for a labelname selects an empty value, it selects all
|
|
// the series which don't have the label name set too. See:
|
|
// https://github.com/prometheus/prometheus/issues/3575 and
|
|
// https://github.com/prometheus/prometheus/pull/3578#issuecomment-351653555
|
|
it, err := inversePostingsForMatcher(ix, m)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
notIts = append(notIts, it)
|
|
}
|
|
}
|
|
|
|
// If there's nothing to subtract from, add in everything and remove the notIts later.
|
|
if len(its) == 0 && len(notIts) != 0 {
|
|
k, v := index.AllPostingsKey()
|
|
allPostings, err := ix.Postings(k, v)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
its = append(its, allPostings)
|
|
}
|
|
|
|
it := index.Intersect(its...)
|
|
|
|
for _, n := range notIts {
|
|
it = index.Without(it, n)
|
|
}
|
|
|
|
return ix.SortedPostings(it), nil
|
|
}
|
|
|
|
func postingsForMatcher(ix IndexReader, m *labels.Matcher) (index.Postings, error) {
|
|
// This method will not return postings for missing labels.
|
|
|
|
// Fast-path for equal matching.
|
|
if m.Type == labels.MatchEqual {
|
|
return ix.Postings(m.Name, m.Value)
|
|
}
|
|
|
|
// Fast-path for set matching.
|
|
if m.Type == labels.MatchRegexp {
|
|
setMatches := findSetMatches(m.Value)
|
|
if len(setMatches) > 0 {
|
|
sort.Strings(setMatches)
|
|
return ix.Postings(m.Name, setMatches...)
|
|
}
|
|
}
|
|
|
|
tpls, err := ix.LabelValues(m.Name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
var res []string
|
|
for i := 0; i < tpls.Len(); i++ {
|
|
vals, err := tpls.At(i)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if m.Matches(vals[0]) {
|
|
res = append(res, vals[0])
|
|
}
|
|
}
|
|
|
|
if len(res) == 0 {
|
|
return index.EmptyPostings(), nil
|
|
}
|
|
|
|
return ix.Postings(m.Name, res...)
|
|
}
|
|
|
|
// inversePostingsForMatcher returns the postings for the series with the label name set but not matching the matcher.
|
|
func inversePostingsForMatcher(ix IndexReader, m *labels.Matcher) (index.Postings, error) {
|
|
tpls, err := ix.LabelValues(m.Name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
var res []string
|
|
for i := 0; i < tpls.Len(); i++ {
|
|
vals, err := tpls.At(i)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if !m.Matches(vals[0]) {
|
|
res = append(res, vals[0])
|
|
}
|
|
}
|
|
|
|
return ix.Postings(m.Name, res...)
|
|
}
|
|
|
|
func mergeStrings(a, b []string) []string {
|
|
maxl := len(a)
|
|
if len(b) > len(a) {
|
|
maxl = len(b)
|
|
}
|
|
res := make([]string, 0, maxl*10/9)
|
|
|
|
for len(a) > 0 && len(b) > 0 {
|
|
d := strings.Compare(a[0], b[0])
|
|
|
|
if d == 0 {
|
|
res = append(res, a[0])
|
|
a, b = a[1:], b[1:]
|
|
} else if d < 0 {
|
|
res = append(res, a[0])
|
|
a = a[1:]
|
|
} else if d > 0 {
|
|
res = append(res, b[0])
|
|
b = b[1:]
|
|
}
|
|
}
|
|
|
|
// Append all remaining elements.
|
|
res = append(res, a...)
|
|
res = append(res, b...)
|
|
return res
|
|
}
|
|
|
|
// SeriesSet contains a set of series.
|
|
type SeriesSet interface {
|
|
Next() bool
|
|
At() Series
|
|
Err() error
|
|
}
|
|
|
|
var emptySeriesSet = errSeriesSet{}
|
|
|
|
// EmptySeriesSet returns a series set that's always empty.
|
|
func EmptySeriesSet() SeriesSet {
|
|
return emptySeriesSet
|
|
}
|
|
|
|
// mergedSeriesSet returns a series sets slice as a single series set. The input series sets
|
|
// must be sorted and sequential in time.
|
|
type mergedSeriesSet struct {
|
|
all []SeriesSet
|
|
buf []SeriesSet // A buffer for keeping the order of SeriesSet slice during forwarding the SeriesSet.
|
|
ids []int // The indices of chosen SeriesSet for the current run.
|
|
done bool
|
|
err error
|
|
cur Series
|
|
}
|
|
|
|
func NewMergedSeriesSet(all []SeriesSet) SeriesSet {
|
|
if len(all) == 1 {
|
|
return all[0]
|
|
}
|
|
s := &mergedSeriesSet{all: all}
|
|
// Initialize first elements of all sets as Next() needs
|
|
// one element look-ahead.
|
|
s.nextAll()
|
|
if len(s.all) == 0 {
|
|
s.done = true
|
|
}
|
|
|
|
return s
|
|
}
|
|
|
|
func (s *mergedSeriesSet) At() Series {
|
|
return s.cur
|
|
}
|
|
|
|
func (s *mergedSeriesSet) Err() error {
|
|
return s.err
|
|
}
|
|
|
|
// nextAll is to call Next() for all SeriesSet.
|
|
// Because the order of the SeriesSet slice will affect the results,
|
|
// we need to use an buffer slice to hold the order.
|
|
func (s *mergedSeriesSet) nextAll() {
|
|
s.buf = s.buf[:0]
|
|
for _, ss := range s.all {
|
|
if ss.Next() {
|
|
s.buf = append(s.buf, ss)
|
|
} else if ss.Err() != nil {
|
|
s.done = true
|
|
s.err = ss.Err()
|
|
break
|
|
}
|
|
}
|
|
s.all, s.buf = s.buf, s.all
|
|
}
|
|
|
|
// nextWithID is to call Next() for the SeriesSet with the indices of s.ids.
|
|
// Because the order of the SeriesSet slice will affect the results,
|
|
// we need to use an buffer slice to hold the order.
|
|
func (s *mergedSeriesSet) nextWithID() {
|
|
if len(s.ids) == 0 {
|
|
return
|
|
}
|
|
|
|
s.buf = s.buf[:0]
|
|
i1 := 0
|
|
i2 := 0
|
|
for i1 < len(s.all) {
|
|
if i2 < len(s.ids) && i1 == s.ids[i2] {
|
|
if !s.all[s.ids[i2]].Next() {
|
|
if s.all[s.ids[i2]].Err() != nil {
|
|
s.done = true
|
|
s.err = s.all[s.ids[i2]].Err()
|
|
break
|
|
}
|
|
i2++
|
|
i1++
|
|
continue
|
|
}
|
|
i2++
|
|
}
|
|
s.buf = append(s.buf, s.all[i1])
|
|
i1++
|
|
}
|
|
s.all, s.buf = s.buf, s.all
|
|
}
|
|
|
|
func (s *mergedSeriesSet) Next() bool {
|
|
if s.done {
|
|
return false
|
|
}
|
|
|
|
s.nextWithID()
|
|
if s.done {
|
|
return false
|
|
}
|
|
s.ids = s.ids[:0]
|
|
if len(s.all) == 0 {
|
|
s.done = true
|
|
return false
|
|
}
|
|
|
|
// Here we are looking for a set of series sets with the lowest labels,
|
|
// and we will cache their indexes in s.ids.
|
|
s.ids = append(s.ids, 0)
|
|
for i := 1; i < len(s.all); i++ {
|
|
cmp := labels.Compare(s.all[s.ids[0]].At().Labels(), s.all[i].At().Labels())
|
|
if cmp > 0 {
|
|
s.ids = s.ids[:1]
|
|
s.ids[0] = i
|
|
} else if cmp == 0 {
|
|
s.ids = append(s.ids, i)
|
|
}
|
|
}
|
|
|
|
if len(s.ids) > 1 {
|
|
series := make([]Series, len(s.ids))
|
|
for i, idx := range s.ids {
|
|
series[i] = s.all[idx].At()
|
|
}
|
|
s.cur = &chainedSeries{series: series}
|
|
} else {
|
|
s.cur = s.all[s.ids[0]].At()
|
|
}
|
|
return true
|
|
}
|
|
|
|
type mergedVerticalSeriesSet struct {
|
|
a, b SeriesSet
|
|
cur Series
|
|
adone, bdone bool
|
|
}
|
|
|
|
// NewMergedVerticalSeriesSet takes two series sets as a single series set.
|
|
// The input series sets must be sorted and
|
|
// the time ranges of the series can be overlapping.
|
|
func NewMergedVerticalSeriesSet(a, b SeriesSet) SeriesSet {
|
|
return newMergedVerticalSeriesSet(a, b)
|
|
}
|
|
|
|
func newMergedVerticalSeriesSet(a, b SeriesSet) *mergedVerticalSeriesSet {
|
|
s := &mergedVerticalSeriesSet{a: a, b: b}
|
|
// Initialize first elements of both sets as Next() needs
|
|
// one element look-ahead.
|
|
s.adone = !s.a.Next()
|
|
s.bdone = !s.b.Next()
|
|
|
|
return s
|
|
}
|
|
|
|
func (s *mergedVerticalSeriesSet) At() Series {
|
|
return s.cur
|
|
}
|
|
|
|
func (s *mergedVerticalSeriesSet) Err() error {
|
|
if s.a.Err() != nil {
|
|
return s.a.Err()
|
|
}
|
|
return s.b.Err()
|
|
}
|
|
|
|
func (s *mergedVerticalSeriesSet) compare() int {
|
|
if s.adone {
|
|
return 1
|
|
}
|
|
if s.bdone {
|
|
return -1
|
|
}
|
|
return labels.Compare(s.a.At().Labels(), s.b.At().Labels())
|
|
}
|
|
|
|
func (s *mergedVerticalSeriesSet) Next() bool {
|
|
if s.adone && s.bdone || s.Err() != nil {
|
|
return false
|
|
}
|
|
|
|
d := s.compare()
|
|
|
|
// Both sets contain the current series. Chain them into a single one.
|
|
if d > 0 {
|
|
s.cur = s.b.At()
|
|
s.bdone = !s.b.Next()
|
|
} else if d < 0 {
|
|
s.cur = s.a.At()
|
|
s.adone = !s.a.Next()
|
|
} else {
|
|
s.cur = &verticalChainedSeries{series: []Series{s.a.At(), s.b.At()}}
|
|
s.adone = !s.a.Next()
|
|
s.bdone = !s.b.Next()
|
|
}
|
|
return true
|
|
}
|
|
|
|
// ChunkSeriesSet exposes the chunks and intervals of a series instead of the
|
|
// actual series itself.
|
|
type ChunkSeriesSet interface {
|
|
Next() bool
|
|
At() (labels.Labels, []chunks.Meta, tombstones.Intervals)
|
|
Err() error
|
|
}
|
|
|
|
// baseChunkSeries loads the label set and chunk references for a postings
|
|
// list from an index. It filters out series that have labels set that should be unset.
|
|
type baseChunkSeries struct {
|
|
p index.Postings
|
|
index IndexReader
|
|
tombstones tombstones.Reader
|
|
|
|
lset labels.Labels
|
|
chks []chunks.Meta
|
|
intervals tombstones.Intervals
|
|
err error
|
|
}
|
|
|
|
// LookupChunkSeries retrieves all series for the given matchers and returns a ChunkSeriesSet
|
|
// over them. It drops chunks based on tombstones in the given reader.
|
|
func LookupChunkSeries(ir IndexReader, tr tombstones.Reader, ms ...*labels.Matcher) (ChunkSeriesSet, error) {
|
|
if tr == nil {
|
|
tr = tombstones.NewMemTombstones()
|
|
}
|
|
p, err := PostingsForMatchers(ir, ms...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &baseChunkSeries{
|
|
p: p,
|
|
index: ir,
|
|
tombstones: tr,
|
|
}, nil
|
|
}
|
|
|
|
func (s *baseChunkSeries) At() (labels.Labels, []chunks.Meta, tombstones.Intervals) {
|
|
return s.lset, s.chks, s.intervals
|
|
}
|
|
|
|
func (s *baseChunkSeries) Err() error { return s.err }
|
|
|
|
func (s *baseChunkSeries) Next() bool {
|
|
var (
|
|
lset = make(labels.Labels, len(s.lset))
|
|
chkMetas = make([]chunks.Meta, len(s.chks))
|
|
err error
|
|
)
|
|
|
|
for s.p.Next() {
|
|
ref := s.p.At()
|
|
if err := s.index.Series(ref, &lset, &chkMetas); err != nil {
|
|
// Postings may be stale. Skip if no underlying series exists.
|
|
if errors.Cause(err) == ErrNotFound {
|
|
continue
|
|
}
|
|
s.err = err
|
|
return false
|
|
}
|
|
|
|
s.lset = lset
|
|
s.chks = chkMetas
|
|
s.intervals, err = s.tombstones.Get(s.p.At())
|
|
if err != nil {
|
|
s.err = errors.Wrap(err, "get tombstones")
|
|
return false
|
|
}
|
|
|
|
if len(s.intervals) > 0 {
|
|
// Only those chunks that are not entirely deleted.
|
|
chks := make([]chunks.Meta, 0, len(s.chks))
|
|
for _, chk := range s.chks {
|
|
if !(tombstones.Interval{Mint: chk.MinTime, Maxt: chk.MaxTime}.IsSubrange(s.intervals)) {
|
|
chks = append(chks, chk)
|
|
}
|
|
}
|
|
|
|
s.chks = chks
|
|
}
|
|
|
|
return true
|
|
}
|
|
if err := s.p.Err(); err != nil {
|
|
s.err = err
|
|
}
|
|
return false
|
|
}
|
|
|
|
// populatedChunkSeries loads chunk data from a store for a set of series
|
|
// with known chunk references. It filters out chunks that do not fit the
|
|
// given time range.
|
|
type populatedChunkSeries struct {
|
|
set ChunkSeriesSet
|
|
chunks ChunkReader
|
|
mint, maxt int64
|
|
|
|
err error
|
|
chks []chunks.Meta
|
|
lset labels.Labels
|
|
intervals tombstones.Intervals
|
|
}
|
|
|
|
func (s *populatedChunkSeries) At() (labels.Labels, []chunks.Meta, tombstones.Intervals) {
|
|
return s.lset, s.chks, s.intervals
|
|
}
|
|
|
|
func (s *populatedChunkSeries) Err() error { return s.err }
|
|
|
|
func (s *populatedChunkSeries) Next() bool {
|
|
for s.set.Next() {
|
|
lset, chks, dranges := s.set.At()
|
|
|
|
for len(chks) > 0 {
|
|
if chks[0].MaxTime >= s.mint {
|
|
break
|
|
}
|
|
chks = chks[1:]
|
|
}
|
|
|
|
// This is to delete in place while iterating.
|
|
for i, rlen := 0, len(chks); i < rlen; i++ {
|
|
j := i - (rlen - len(chks))
|
|
c := &chks[j]
|
|
|
|
// Break out at the first chunk that has no overlap with mint, maxt.
|
|
if c.MinTime > s.maxt {
|
|
chks = chks[:j]
|
|
break
|
|
}
|
|
|
|
c.Chunk, s.err = s.chunks.Chunk(c.Ref)
|
|
if s.err != nil {
|
|
// This means that the chunk has be garbage collected. Remove it from the list.
|
|
if s.err == ErrNotFound {
|
|
s.err = nil
|
|
// Delete in-place.
|
|
s.chks = append(chks[:j], chks[j+1:]...)
|
|
}
|
|
return false
|
|
}
|
|
}
|
|
|
|
if len(chks) == 0 {
|
|
continue
|
|
}
|
|
|
|
s.lset = lset
|
|
s.chks = chks
|
|
s.intervals = dranges
|
|
|
|
return true
|
|
}
|
|
if err := s.set.Err(); err != nil {
|
|
s.err = err
|
|
}
|
|
return false
|
|
}
|
|
|
|
// blockSeriesSet is a set of series from an inverted index query.
|
|
type blockSeriesSet struct {
|
|
set ChunkSeriesSet
|
|
err error
|
|
cur Series
|
|
|
|
mint, maxt int64
|
|
}
|
|
|
|
func (s *blockSeriesSet) Next() bool {
|
|
for s.set.Next() {
|
|
lset, chunks, dranges := s.set.At()
|
|
s.cur = &chunkSeries{
|
|
labels: lset,
|
|
chunks: chunks,
|
|
mint: s.mint,
|
|
maxt: s.maxt,
|
|
|
|
intervals: dranges,
|
|
}
|
|
return true
|
|
}
|
|
if s.set.Err() != nil {
|
|
s.err = s.set.Err()
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (s *blockSeriesSet) At() Series { return s.cur }
|
|
func (s *blockSeriesSet) Err() error { return s.err }
|
|
|
|
// chunkSeries is a series that is backed by a sequence of chunks holding
|
|
// time series data.
|
|
type chunkSeries struct {
|
|
labels labels.Labels
|
|
chunks []chunks.Meta // in-order chunk refs
|
|
|
|
mint, maxt int64
|
|
|
|
intervals tombstones.Intervals
|
|
}
|
|
|
|
func (s *chunkSeries) Labels() labels.Labels {
|
|
return s.labels
|
|
}
|
|
|
|
func (s *chunkSeries) Iterator() SeriesIterator {
|
|
return newChunkSeriesIterator(s.chunks, s.intervals, s.mint, s.maxt)
|
|
}
|
|
|
|
// SeriesIterator iterates over the data of a time series.
|
|
type SeriesIterator interface {
|
|
// Seek advances the iterator forward to the given timestamp.
|
|
// If there's no value exactly at t, it advances to the first value
|
|
// after t.
|
|
Seek(t int64) bool
|
|
// At returns the current timestamp/value pair.
|
|
At() (t int64, v float64)
|
|
// Next advances the iterator by one.
|
|
Next() bool
|
|
// Err returns the current error.
|
|
Err() error
|
|
}
|
|
|
|
// chainedSeries implements a series for a list of time-sorted series.
|
|
// They all must have the same labels.
|
|
type chainedSeries struct {
|
|
series []Series
|
|
}
|
|
|
|
func (s *chainedSeries) Labels() labels.Labels {
|
|
return s.series[0].Labels()
|
|
}
|
|
|
|
func (s *chainedSeries) Iterator() SeriesIterator {
|
|
return newChainedSeriesIterator(s.series...)
|
|
}
|
|
|
|
// chainedSeriesIterator implements a series iterator over a list
|
|
// of time-sorted, non-overlapping iterators.
|
|
type chainedSeriesIterator struct {
|
|
series []Series // series in time order
|
|
|
|
i int
|
|
cur SeriesIterator
|
|
}
|
|
|
|
func newChainedSeriesIterator(s ...Series) *chainedSeriesIterator {
|
|
return &chainedSeriesIterator{
|
|
series: s,
|
|
i: 0,
|
|
cur: s[0].Iterator(),
|
|
}
|
|
}
|
|
|
|
func (it *chainedSeriesIterator) Seek(t int64) bool {
|
|
// We just scan the chained series sequentially as they are already
|
|
// pre-selected by relevant time and should be accessed sequentially anyway.
|
|
for i, s := range it.series[it.i:] {
|
|
cur := s.Iterator()
|
|
if !cur.Seek(t) {
|
|
continue
|
|
}
|
|
it.cur = cur
|
|
it.i += i
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (it *chainedSeriesIterator) Next() bool {
|
|
if it.cur.Next() {
|
|
return true
|
|
}
|
|
if err := it.cur.Err(); err != nil {
|
|
return false
|
|
}
|
|
if it.i == len(it.series)-1 {
|
|
return false
|
|
}
|
|
|
|
it.i++
|
|
it.cur = it.series[it.i].Iterator()
|
|
|
|
return it.Next()
|
|
}
|
|
|
|
func (it *chainedSeriesIterator) At() (t int64, v float64) {
|
|
return it.cur.At()
|
|
}
|
|
|
|
func (it *chainedSeriesIterator) Err() error {
|
|
return it.cur.Err()
|
|
}
|
|
|
|
// verticalChainedSeries implements a series for a list of time-sorted, time-overlapping series.
|
|
// They all must have the same labels.
|
|
type verticalChainedSeries struct {
|
|
series []Series
|
|
}
|
|
|
|
func (s *verticalChainedSeries) Labels() labels.Labels {
|
|
return s.series[0].Labels()
|
|
}
|
|
|
|
func (s *verticalChainedSeries) Iterator() SeriesIterator {
|
|
return newVerticalMergeSeriesIterator(s.series...)
|
|
}
|
|
|
|
// verticalMergeSeriesIterator implements a series iterator over a list
|
|
// of time-sorted, time-overlapping iterators.
|
|
type verticalMergeSeriesIterator struct {
|
|
a, b SeriesIterator
|
|
aok, bok, initialized bool
|
|
|
|
curT int64
|
|
curV float64
|
|
}
|
|
|
|
func newVerticalMergeSeriesIterator(s ...Series) SeriesIterator {
|
|
if len(s) == 1 {
|
|
return s[0].Iterator()
|
|
} else if len(s) == 2 {
|
|
return &verticalMergeSeriesIterator{
|
|
a: s[0].Iterator(),
|
|
b: s[1].Iterator(),
|
|
}
|
|
}
|
|
return &verticalMergeSeriesIterator{
|
|
a: s[0].Iterator(),
|
|
b: newVerticalMergeSeriesIterator(s[1:]...),
|
|
}
|
|
}
|
|
|
|
func (it *verticalMergeSeriesIterator) Seek(t int64) bool {
|
|
it.aok, it.bok = it.a.Seek(t), it.b.Seek(t)
|
|
it.initialized = true
|
|
return it.Next()
|
|
}
|
|
|
|
func (it *verticalMergeSeriesIterator) Next() bool {
|
|
if !it.initialized {
|
|
it.aok = it.a.Next()
|
|
it.bok = it.b.Next()
|
|
it.initialized = true
|
|
}
|
|
|
|
if !it.aok && !it.bok {
|
|
return false
|
|
}
|
|
|
|
if !it.aok {
|
|
it.curT, it.curV = it.b.At()
|
|
it.bok = it.b.Next()
|
|
return true
|
|
}
|
|
if !it.bok {
|
|
it.curT, it.curV = it.a.At()
|
|
it.aok = it.a.Next()
|
|
return true
|
|
}
|
|
|
|
acurT, acurV := it.a.At()
|
|
bcurT, bcurV := it.b.At()
|
|
if acurT < bcurT {
|
|
it.curT, it.curV = acurT, acurV
|
|
it.aok = it.a.Next()
|
|
} else if acurT > bcurT {
|
|
it.curT, it.curV = bcurT, bcurV
|
|
it.bok = it.b.Next()
|
|
} else {
|
|
it.curT, it.curV = bcurT, bcurV
|
|
it.aok = it.a.Next()
|
|
it.bok = it.b.Next()
|
|
}
|
|
return true
|
|
}
|
|
|
|
func (it *verticalMergeSeriesIterator) At() (t int64, v float64) {
|
|
return it.curT, it.curV
|
|
}
|
|
|
|
func (it *verticalMergeSeriesIterator) Err() error {
|
|
if it.a.Err() != nil {
|
|
return it.a.Err()
|
|
}
|
|
return it.b.Err()
|
|
}
|
|
|
|
// chunkSeriesIterator implements a series iterator on top
|
|
// of a list of time-sorted, non-overlapping chunks.
|
|
type chunkSeriesIterator struct {
|
|
chunks []chunks.Meta
|
|
|
|
i int
|
|
cur chunkenc.Iterator
|
|
bufDelIter *deletedIterator
|
|
|
|
maxt, mint int64
|
|
|
|
intervals tombstones.Intervals
|
|
}
|
|
|
|
func newChunkSeriesIterator(cs []chunks.Meta, dranges tombstones.Intervals, mint, maxt int64) *chunkSeriesIterator {
|
|
csi := &chunkSeriesIterator{
|
|
chunks: cs,
|
|
i: 0,
|
|
|
|
mint: mint,
|
|
maxt: maxt,
|
|
|
|
intervals: dranges,
|
|
}
|
|
csi.resetCurIterator()
|
|
|
|
return csi
|
|
}
|
|
|
|
func (it *chunkSeriesIterator) resetCurIterator() {
|
|
if len(it.intervals) == 0 {
|
|
it.cur = it.chunks[it.i].Chunk.Iterator(it.cur)
|
|
return
|
|
}
|
|
if it.bufDelIter == nil {
|
|
it.bufDelIter = &deletedIterator{
|
|
intervals: it.intervals,
|
|
}
|
|
}
|
|
it.bufDelIter.it = it.chunks[it.i].Chunk.Iterator(it.bufDelIter.it)
|
|
it.cur = it.bufDelIter
|
|
}
|
|
|
|
func (it *chunkSeriesIterator) Seek(t int64) (ok bool) {
|
|
if t > it.maxt {
|
|
return false
|
|
}
|
|
|
|
// Seek to the first valid value after t.
|
|
if t < it.mint {
|
|
t = it.mint
|
|
}
|
|
|
|
for ; it.chunks[it.i].MaxTime < t; it.i++ {
|
|
if it.i == len(it.chunks)-1 {
|
|
return false
|
|
}
|
|
}
|
|
|
|
it.resetCurIterator()
|
|
|
|
for it.cur.Next() {
|
|
t0, _ := it.cur.At()
|
|
if t0 >= t {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (it *chunkSeriesIterator) At() (t int64, v float64) {
|
|
return it.cur.At()
|
|
}
|
|
|
|
func (it *chunkSeriesIterator) Next() bool {
|
|
if it.cur.Next() {
|
|
t, _ := it.cur.At()
|
|
|
|
if t < it.mint {
|
|
if !it.Seek(it.mint) {
|
|
return false
|
|
}
|
|
t, _ = it.At()
|
|
|
|
return t <= it.maxt
|
|
}
|
|
if t > it.maxt {
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
if err := it.cur.Err(); err != nil {
|
|
return false
|
|
}
|
|
if it.i == len(it.chunks)-1 {
|
|
return false
|
|
}
|
|
|
|
it.i++
|
|
it.resetCurIterator()
|
|
|
|
return it.Next()
|
|
}
|
|
|
|
func (it *chunkSeriesIterator) Err() error {
|
|
return it.cur.Err()
|
|
}
|
|
|
|
// deletedIterator wraps an Iterator and makes sure any deleted metrics are not
|
|
// returned.
|
|
type deletedIterator struct {
|
|
it chunkenc.Iterator
|
|
|
|
intervals tombstones.Intervals
|
|
}
|
|
|
|
func (it *deletedIterator) At() (int64, float64) {
|
|
return it.it.At()
|
|
}
|
|
|
|
func (it *deletedIterator) Next() bool {
|
|
Outer:
|
|
for it.it.Next() {
|
|
ts, _ := it.it.At()
|
|
|
|
for _, tr := range it.intervals {
|
|
if tr.InBounds(ts) {
|
|
continue Outer
|
|
}
|
|
|
|
if ts > tr.Maxt {
|
|
it.intervals = it.intervals[1:]
|
|
continue
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
func (it *deletedIterator) Err() error {
|
|
return it.it.Err()
|
|
}
|
|
|
|
type errSeriesSet struct {
|
|
err error
|
|
}
|
|
|
|
func (s errSeriesSet) Next() bool { return false }
|
|
func (s errSeriesSet) At() Series { return nil }
|
|
func (s errSeriesSet) Err() error { return s.err }
|