2020-08-03 10:32:56 +00:00
// Copyright 2020 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 storage
import (
"bytes"
"container/heap"
2023-09-12 10:37:38 +00:00
"context"
2022-07-01 16:59:50 +00:00
"fmt"
2020-09-08 14:23:01 +00:00
"math"
2020-08-03 10:32:56 +00:00
"sync"
2022-10-11 16:35:35 +00:00
"golang.org/x/exp/slices"
Style cleanup of all the changes in sparsehistogram so far
A lot of this code was hacked together, literally during a
hackathon. This commit intends not to change the code substantially,
but just make the code obey the usual style practices.
A (possibly incomplete) list of areas:
* Generally address linter warnings.
* The `pgk` directory is deprecated as per dev-summit. No new packages should
be added to it. I moved the new `pkg/histogram` package to `model`
anticipating what's proposed in #9478.
* Make the naming of the Sparse Histogram more consistent. Including
abbreviations, there were just too many names for it: SparseHistogram,
Histogram, Histo, hist, his, shs, h. The idea is to call it "Histogram" in
general. Only add "Sparse" if it is needed to avoid confusion with
conventional Histograms (which is rare because the TSDB really has no notion
of conventional Histograms). Use abbreviations only in local scope, and then
really abbreviate (not just removing three out of seven letters like in
"Histo"). This is in the spirit of
https://github.com/golang/go/wiki/CodeReviewComments#variable-names
* Several other minor name changes.
* A lot of formatting of doc comments. For one, following
https://github.com/golang/go/wiki/CodeReviewComments#comment-sentences
, but also layout question, anticipating how things will look like
when rendered by `godoc` (even where `godoc` doesn't render them
right now because they are for unexported types or not a doc comment
at all but just a normal code comment - consistency is queen!).
* Re-enabled `TestQueryLog` and `TestEndopints` (they pass now,
leaving them disabled was presumably an oversight).
* Bucket iterator for histogram.Histogram is now created with a
method.
* HistogramChunk.iterator now allows iterator recycling. (I think
@dieterbe only commented it out because he was confused by the
question in the comment.)
* HistogramAppender.Append panics now because we decided to treat
staleness marker differently.
Signed-off-by: beorn7 <beorn@grafana.com>
2021-10-09 13:57:07 +00:00
"github.com/prometheus/prometheus/model/histogram"
2021-11-08 14:23:17 +00:00
"github.com/prometheus/prometheus/model/labels"
2020-08-03 10:32:56 +00:00
"github.com/prometheus/prometheus/tsdb/chunkenc"
"github.com/prometheus/prometheus/tsdb/chunks"
tsdb_errors "github.com/prometheus/prometheus/tsdb/errors"
2023-09-14 16:57:31 +00:00
"github.com/prometheus/prometheus/util/annotations"
2020-08-03 10:32:56 +00:00
)
type mergeGenericQuerier struct {
queriers [ ] genericQuerier
// mergeFn is used when we see series from different queriers Selects with the same labels.
mergeFn genericSeriesMergeFunc
// TODO(bwplotka): Remove once remote queries are asynchronous. False by default.
concurrentSelect bool
}
// NewMergeQuerier returns a new Querier that merges results of given primary and secondary queriers.
// See NewFanout commentary to learn more about primary vs secondary differences.
//
// In case of overlaps between the data given by primaries' and secondaries' Selects, merge function will be used.
2021-10-22 08:06:44 +00:00
func NewMergeQuerier ( primaries , secondaries [ ] Querier , mergeFn VerticalSeriesMergeFunc ) Querier {
2020-08-03 10:32:56 +00:00
queriers := make ( [ ] genericQuerier , 0 , len ( primaries ) + len ( secondaries ) )
for _ , q := range primaries {
if _ , ok := q . ( noopQuerier ) ; ! ok && q != nil {
queriers = append ( queriers , newGenericQuerierFrom ( q ) )
}
}
for _ , q := range secondaries {
if _ , ok := q . ( noopQuerier ) ; ! ok && q != nil {
queriers = append ( queriers , newSecondaryQuerierFrom ( q ) )
}
}
concurrentSelect := false
if len ( secondaries ) > 0 {
concurrentSelect = true
}
return & querierAdapter { & mergeGenericQuerier {
mergeFn : ( & seriesMergerAdapter { VerticalSeriesMergeFunc : mergeFn } ) . Merge ,
queriers : queriers ,
concurrentSelect : concurrentSelect ,
} }
}
// NewMergeChunkQuerier returns a new Chunk Querier that merges results of given primary and secondary chunk queriers.
// See NewFanout commentary to learn more about primary vs secondary differences.
//
// In case of overlaps between the data given by primaries' and secondaries' Selects, merge function will be used.
// TODO(bwplotka): Currently merge will compact overlapping chunks with bigger chunk, without limit. Split it: https://github.com/prometheus/tsdb/issues/670
2021-10-22 08:06:44 +00:00
func NewMergeChunkQuerier ( primaries , secondaries [ ] ChunkQuerier , mergeFn VerticalChunkSeriesMergeFunc ) ChunkQuerier {
2020-08-03 10:32:56 +00:00
queriers := make ( [ ] genericQuerier , 0 , len ( primaries ) + len ( secondaries ) )
for _ , q := range primaries {
if _ , ok := q . ( noopChunkQuerier ) ; ! ok && q != nil {
queriers = append ( queriers , newGenericQuerierFromChunk ( q ) )
}
}
for _ , querier := range secondaries {
if _ , ok := querier . ( noopChunkQuerier ) ; ! ok && querier != nil {
queriers = append ( queriers , newSecondaryQuerierFromChunk ( querier ) )
}
}
concurrentSelect := false
if len ( secondaries ) > 0 {
concurrentSelect = true
}
return & chunkQuerierAdapter { & mergeGenericQuerier {
mergeFn : ( & chunkSeriesMergerAdapter { VerticalChunkSeriesMergeFunc : mergeFn } ) . Merge ,
queriers : queriers ,
concurrentSelect : concurrentSelect ,
} }
}
// Select returns a set of series that matches the given label matchers.
2023-09-12 10:37:38 +00:00
func ( q * mergeGenericQuerier ) Select ( ctx context . Context , sortSeries bool , hints * SelectHints , matchers ... * labels . Matcher ) genericSeriesSet {
2020-08-03 10:32:56 +00:00
if len ( q . queriers ) == 0 {
return noopGenericSeriesSet { }
}
if len ( q . queriers ) == 1 {
2023-09-12 10:37:38 +00:00
return q . queriers [ 0 ] . Select ( ctx , sortSeries , hints , matchers ... )
2020-08-03 10:32:56 +00:00
}
2021-10-22 08:06:44 +00:00
seriesSets := make ( [ ] genericSeriesSet , 0 , len ( q . queriers ) )
2020-08-03 10:32:56 +00:00
if ! q . concurrentSelect {
for _ , querier := range q . queriers {
// We need to sort for merge to work.
2023-09-12 10:37:38 +00:00
seriesSets = append ( seriesSets , querier . Select ( ctx , true , hints , matchers ... ) )
2020-08-03 10:32:56 +00:00
}
return & lazyGenericSeriesSet { init : func ( ) ( genericSeriesSet , bool ) {
s := newGenericMergeSeriesSet ( seriesSets , q . mergeFn )
return s , s . Next ( )
} }
}
var (
wg sync . WaitGroup
seriesSetChan = make ( chan genericSeriesSet )
)
// Schedule all Selects for all queriers we know about.
for _ , querier := range q . queriers {
wg . Add ( 1 )
go func ( qr genericQuerier ) {
defer wg . Done ( )
// We need to sort for NewMergeSeriesSet to work.
2023-09-12 10:37:38 +00:00
seriesSetChan <- qr . Select ( ctx , true , hints , matchers ... )
2020-08-03 10:32:56 +00:00
} ( querier )
}
go func ( ) {
wg . Wait ( )
close ( seriesSetChan )
} ( )
for r := range seriesSetChan {
seriesSets = append ( seriesSets , r )
}
return & lazyGenericSeriesSet { init : func ( ) ( genericSeriesSet , bool ) {
s := newGenericMergeSeriesSet ( seriesSets , q . mergeFn )
return s , s . Next ( )
} }
}
type labelGenericQueriers [ ] genericQuerier
func ( l labelGenericQueriers ) Len ( ) int { return len ( l ) }
func ( l labelGenericQueriers ) Get ( i int ) LabelQuerier { return l [ i ] }
func ( l labelGenericQueriers ) SplitByHalf ( ) ( labelGenericQueriers , labelGenericQueriers ) {
i := len ( l ) / 2
return l [ : i ] , l [ i : ]
}
// LabelValues returns all potential values for a label name.
2021-02-09 17:38:35 +00:00
// If matchers are specified the returned result set is reduced
// to label values of metrics matching the matchers.
2023-09-14 16:57:31 +00:00
func ( q * mergeGenericQuerier ) LabelValues ( ctx context . Context , name string , matchers ... * labels . Matcher ) ( [ ] string , annotations . Annotations , error ) {
2023-09-14 14:02:04 +00:00
res , ws , err := q . lvals ( ctx , q . queriers , name , matchers ... )
2020-08-03 10:32:56 +00:00
if err != nil {
2022-07-01 16:59:50 +00:00
return nil , nil , fmt . Errorf ( "LabelValues() from merge generic querier for label %s: %w" , name , err )
2020-08-03 10:32:56 +00:00
}
return res , ws , nil
}
// lvals performs merge sort for LabelValues from multiple queriers.
2023-09-14 16:57:31 +00:00
func ( q * mergeGenericQuerier ) lvals ( ctx context . Context , lq labelGenericQueriers , n string , matchers ... * labels . Matcher ) ( [ ] string , annotations . Annotations , error ) {
2020-08-03 10:32:56 +00:00
if lq . Len ( ) == 0 {
return nil , nil , nil
}
if lq . Len ( ) == 1 {
2023-09-14 14:02:04 +00:00
return lq . Get ( 0 ) . LabelValues ( ctx , n , matchers ... )
2020-08-03 10:32:56 +00:00
}
a , b := lq . SplitByHalf ( )
2023-09-14 16:57:31 +00:00
var ws annotations . Annotations
2023-09-14 14:02:04 +00:00
s1 , w , err := q . lvals ( ctx , a , n , matchers ... )
2023-09-14 16:57:31 +00:00
ws . Merge ( w )
2020-08-03 10:32:56 +00:00
if err != nil {
return nil , ws , err
}
2023-09-14 14:02:04 +00:00
s2 , ws , err := q . lvals ( ctx , b , n , matchers ... )
2023-09-14 16:57:31 +00:00
ws . Merge ( w )
2020-08-03 10:32:56 +00:00
if err != nil {
return nil , ws , err
}
return mergeStrings ( s1 , s2 ) , ws , nil
}
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 {
style: Replace `else if` cascades with `switch`
Wiser coders than myself have come to the conclusion that a `switch`
statement is almost always superior to a statement that includes any
`else if`.
The exceptions that I have found in our codebase are just these two:
* The `if else` is followed by an additional statement before the next
condition (separated by a `;`).
* The whole thing is within a `for` loop and `break` statements are
used. In this case, using `switch` would require tagging the `for`
loop, which probably tips the balance.
Why are `switch` statements more readable?
For one, fewer curly braces. But more importantly, the conditions all
have the same alignment, so the whole thing follows the natural flow
of going down a list of conditions. With `else if`, in contrast, all
conditions but the first are "hidden" behind `} else if `, harder to
spot and (for no good reason) presented differently from the first
condition.
I'm sure the aforemention wise coders can list even more reasons.
In any case, I like it so much that I have found myself recommending
it in code reviews. I would like to make it a habit in our code base,
without making it a hard requirement that we would test on the CI. But
for that, there has to be a role model, so this commit eliminates all
`if else` occurrences, unless it is autogenerated code or fits one of
the exceptions above.
Signed-off-by: beorn7 <beorn@grafana.com>
2023-04-12 14:14:31 +00:00
switch {
case a [ 0 ] == b [ 0 ] :
2020-08-03 10:32:56 +00:00
res = append ( res , a [ 0 ] )
a , b = a [ 1 : ] , b [ 1 : ]
style: Replace `else if` cascades with `switch`
Wiser coders than myself have come to the conclusion that a `switch`
statement is almost always superior to a statement that includes any
`else if`.
The exceptions that I have found in our codebase are just these two:
* The `if else` is followed by an additional statement before the next
condition (separated by a `;`).
* The whole thing is within a `for` loop and `break` statements are
used. In this case, using `switch` would require tagging the `for`
loop, which probably tips the balance.
Why are `switch` statements more readable?
For one, fewer curly braces. But more importantly, the conditions all
have the same alignment, so the whole thing follows the natural flow
of going down a list of conditions. With `else if`, in contrast, all
conditions but the first are "hidden" behind `} else if `, harder to
spot and (for no good reason) presented differently from the first
condition.
I'm sure the aforemention wise coders can list even more reasons.
In any case, I like it so much that I have found myself recommending
it in code reviews. I would like to make it a habit in our code base,
without making it a hard requirement that we would test on the CI. But
for that, there has to be a role model, so this commit eliminates all
`if else` occurrences, unless it is autogenerated code or fits one of
the exceptions above.
Signed-off-by: beorn7 <beorn@grafana.com>
2023-04-12 14:14:31 +00:00
case a [ 0 ] < b [ 0 ] :
2020-08-03 10:32:56 +00:00
res = append ( res , a [ 0 ] )
a = a [ 1 : ]
style: Replace `else if` cascades with `switch`
Wiser coders than myself have come to the conclusion that a `switch`
statement is almost always superior to a statement that includes any
`else if`.
The exceptions that I have found in our codebase are just these two:
* The `if else` is followed by an additional statement before the next
condition (separated by a `;`).
* The whole thing is within a `for` loop and `break` statements are
used. In this case, using `switch` would require tagging the `for`
loop, which probably tips the balance.
Why are `switch` statements more readable?
For one, fewer curly braces. But more importantly, the conditions all
have the same alignment, so the whole thing follows the natural flow
of going down a list of conditions. With `else if`, in contrast, all
conditions but the first are "hidden" behind `} else if `, harder to
spot and (for no good reason) presented differently from the first
condition.
I'm sure the aforemention wise coders can list even more reasons.
In any case, I like it so much that I have found myself recommending
it in code reviews. I would like to make it a habit in our code base,
without making it a hard requirement that we would test on the CI. But
for that, there has to be a role model, so this commit eliminates all
`if else` occurrences, unless it is autogenerated code or fits one of
the exceptions above.
Signed-off-by: beorn7 <beorn@grafana.com>
2023-04-12 14:14:31 +00:00
default :
2020-08-03 10:32:56 +00:00
res = append ( res , b [ 0 ] )
b = b [ 1 : ]
}
}
// Append all remaining elements.
res = append ( res , a ... )
res = append ( res , b ... )
return res
}
// LabelNames returns all the unique label names present in all queriers in sorted order.
2023-09-14 16:57:31 +00:00
func ( q * mergeGenericQuerier ) LabelNames ( ctx context . Context , matchers ... * labels . Matcher ) ( [ ] string , annotations . Annotations , error ) {
2020-08-03 10:32:56 +00:00
var (
labelNamesMap = make ( map [ string ] struct { } )
2023-09-14 16:57:31 +00:00
warnings annotations . Annotations
2020-08-03 10:32:56 +00:00
)
for _ , querier := range q . queriers {
2023-09-14 08:39:51 +00:00
names , wrn , err := querier . LabelNames ( ctx , matchers ... )
2020-08-03 10:32:56 +00:00
if wrn != nil {
// TODO(bwplotka): We could potentially wrap warnings.
2023-09-14 16:57:31 +00:00
warnings . Merge ( wrn )
2020-08-03 10:32:56 +00:00
}
if err != nil {
2022-07-01 16:59:50 +00:00
return nil , nil , fmt . Errorf ( "LabelNames() from merge generic querier: %w" , err )
2020-08-03 10:32:56 +00:00
}
for _ , name := range names {
labelNamesMap [ name ] = struct { } { }
}
}
if len ( labelNamesMap ) == 0 {
return nil , warnings , nil
}
labelNames := make ( [ ] string , 0 , len ( labelNamesMap ) )
for name := range labelNamesMap {
labelNames = append ( labelNames , name )
}
2022-09-30 14:33:56 +00:00
slices . Sort ( labelNames )
2020-08-03 10:32:56 +00:00
return labelNames , warnings , nil
}
// Close releases the resources of the generic querier.
func ( q * mergeGenericQuerier ) Close ( ) error {
2020-10-28 15:24:58 +00:00
errs := tsdb_errors . NewMulti ( )
2020-08-03 10:32:56 +00:00
for _ , querier := range q . queriers {
if err := querier . Close ( ) ; err != nil {
errs . Add ( err )
}
}
return errs . Err ( )
}
// VerticalSeriesMergeFunc returns merged series implementation that merges series with same labels together.
// It has to handle time-overlapped series as well.
type VerticalSeriesMergeFunc func ( ... Series ) Series
// NewMergeSeriesSet returns a new SeriesSet that merges many SeriesSets together.
func NewMergeSeriesSet ( sets [ ] SeriesSet , mergeFunc VerticalSeriesMergeFunc ) SeriesSet {
genericSets := make ( [ ] genericSeriesSet , 0 , len ( sets ) )
for _ , s := range sets {
genericSets = append ( genericSets , & genericSeriesSetAdapter { s } )
}
return & seriesSetAdapter { newGenericMergeSeriesSet ( genericSets , ( & seriesMergerAdapter { VerticalSeriesMergeFunc : mergeFunc } ) . Merge ) }
}
// VerticalChunkSeriesMergeFunc returns merged chunk series implementation that merges potentially time-overlapping
// chunk series with the same labels into single ChunkSeries.
//
// NOTE: It's up to implementation how series are vertically merged (if chunks are sorted, re-encoded etc).
type VerticalChunkSeriesMergeFunc func ( ... ChunkSeries ) ChunkSeries
// NewMergeChunkSeriesSet returns a new ChunkSeriesSet that merges many SeriesSet together.
func NewMergeChunkSeriesSet ( sets [ ] ChunkSeriesSet , mergeFunc VerticalChunkSeriesMergeFunc ) ChunkSeriesSet {
genericSets := make ( [ ] genericSeriesSet , 0 , len ( sets ) )
for _ , s := range sets {
genericSets = append ( genericSets , & genericChunkSeriesSetAdapter { s } )
}
return & chunkSeriesSetAdapter { newGenericMergeSeriesSet ( genericSets , ( & chunkSeriesMergerAdapter { VerticalChunkSeriesMergeFunc : mergeFunc } ) . Merge ) }
}
// genericMergeSeriesSet implements genericSeriesSet.
type genericMergeSeriesSet struct {
currentLabels labels . Labels
mergeFunc genericSeriesMergeFunc
heap genericSeriesSetHeap
sets [ ] genericSeriesSet
currentSets [ ] genericSeriesSet
}
// newGenericMergeSeriesSet returns a new genericSeriesSet that merges (and deduplicates)
// series returned by the series sets when iterating.
// Each series set must return its series in labels order, otherwise
// merged series set will be incorrect.
// Overlapped situations are merged using provided mergeFunc.
func newGenericMergeSeriesSet ( sets [ ] genericSeriesSet , mergeFunc genericSeriesMergeFunc ) genericSeriesSet {
if len ( sets ) == 1 {
return sets [ 0 ]
}
// We are pre-advancing sets, so we can introspect the label of the
// series under the cursor.
var h genericSeriesSetHeap
for _ , set := range sets {
if set == nil {
continue
}
if set . Next ( ) {
heap . Push ( & h , set )
}
if err := set . Err ( ) ; err != nil {
return errorOnlySeriesSet { err }
}
}
return & genericMergeSeriesSet {
mergeFunc : mergeFunc ,
sets : sets ,
heap : h ,
}
}
func ( c * genericMergeSeriesSet ) Next ( ) bool {
// Run in a loop because the "next" series sets may not be valid anymore.
// If, for the current label set, all the next series sets come from
// failed remote storage sources, we want to keep trying with the next label set.
for {
2020-08-28 23:21:39 +00:00
// Firstly advance all the current series sets. If any of them have run out,
2020-08-03 10:32:56 +00:00
// we can drop them, otherwise they should be inserted back into the heap.
for _ , set := range c . currentSets {
if set . Next ( ) {
heap . Push ( & c . heap , set )
}
}
if len ( c . heap ) == 0 {
return false
}
// Now, pop items of the heap that have equal label sets.
2023-10-06 11:28:07 +00:00
c . currentSets = c . currentSets [ : 0 ]
2020-08-03 10:32:56 +00:00
c . currentLabels = c . heap [ 0 ] . At ( ) . Labels ( )
for len ( c . heap ) > 0 && labels . Equal ( c . currentLabels , c . heap [ 0 ] . At ( ) . Labels ( ) ) {
set := heap . Pop ( & c . heap ) . ( genericSeriesSet )
c . currentSets = append ( c . currentSets , set )
}
// As long as the current set contains at least 1 set,
// then it should return true.
if len ( c . currentSets ) != 0 {
break
}
}
return true
}
func ( c * genericMergeSeriesSet ) At ( ) Labels {
if len ( c . currentSets ) == 1 {
return c . currentSets [ 0 ] . At ( )
}
series := make ( [ ] Labels , 0 , len ( c . currentSets ) )
for _ , seriesSet := range c . currentSets {
series = append ( series , seriesSet . At ( ) )
}
return c . mergeFunc ( series ... )
}
func ( c * genericMergeSeriesSet ) Err ( ) error {
for _ , set := range c . sets {
if err := set . Err ( ) ; err != nil {
return err
}
}
return nil
}
2023-09-14 16:57:31 +00:00
func ( c * genericMergeSeriesSet ) Warnings ( ) annotations . Annotations {
var ws annotations . Annotations
2020-08-03 10:32:56 +00:00
for _ , set := range c . sets {
2023-09-14 16:57:31 +00:00
ws . Merge ( set . Warnings ( ) )
2020-08-03 10:32:56 +00:00
}
return ws
}
type genericSeriesSetHeap [ ] genericSeriesSet
func ( h genericSeriesSetHeap ) Len ( ) int { return len ( h ) }
func ( h genericSeriesSetHeap ) Swap ( i , j int ) { h [ i ] , h [ j ] = h [ j ] , h [ i ] }
func ( h genericSeriesSetHeap ) Less ( i , j int ) bool {
a , b := h [ i ] . At ( ) . Labels ( ) , h [ j ] . At ( ) . Labels ( )
return labels . Compare ( a , b ) < 0
}
func ( h * genericSeriesSetHeap ) Push ( x interface { } ) {
* h = append ( * h , x . ( genericSeriesSet ) )
}
func ( h * genericSeriesSetHeap ) Pop ( ) interface { } {
old := * h
n := len ( old )
x := old [ n - 1 ]
* h = old [ 0 : n - 1 ]
return x
}
// ChainedSeriesMerge returns single series from many same, potentially overlapping series by chaining samples together.
// If one or more samples overlap, one sample from random overlapped ones is kept and all others with the same
// timestamp are dropped.
//
// This works the best with replicated series, where data from two series are exactly the same. This does not work well
// with "almost" the same data, e.g. from 2 Prometheus HA replicas. This is fine, since from the Prometheus perspective
// this never happens.
//
2020-09-08 14:23:01 +00:00
// It's optimized for non-overlap cases as well.
2020-08-03 10:32:56 +00:00
func ChainedSeriesMerge ( series ... Series ) Series {
if len ( series ) == 0 {
return nil
}
return & SeriesEntry {
Lset : series [ 0 ] . Labels ( ) ,
2022-09-20 18:31:28 +00:00
SampleIteratorFn : func ( it chunkenc . Iterator ) chunkenc . Iterator {
return ChainSampleIteratorFromSeries ( it , series )
2020-08-03 10:32:56 +00:00
} ,
}
}
// chainSampleIterator is responsible to iterate over samples from different iterators of the same time series in timestamps
// order. If one or more samples overlap, one sample from random overlapped ones is kept and all others with the same
2020-09-08 14:23:01 +00:00
// timestamp are dropped. It's optimized for non-overlap cases as well.
2020-08-03 10:32:56 +00:00
type chainSampleIterator struct {
iterators [ ] chunkenc . Iterator
h samplesIteratorHeap
2020-09-08 14:23:01 +00:00
curr chunkenc . Iterator
2021-11-29 07:54:23 +00:00
lastT int64
2023-01-18 16:59:29 +00:00
// Whether the previous and the current sample are direct neighbors
// within the same base iterator.
consecutive bool
2020-08-03 10:32:56 +00:00
}
2022-09-20 18:31:28 +00:00
// Return a chainSampleIterator initialized for length entries, re-using the memory from it if possible.
func getChainSampleIterator ( it chunkenc . Iterator , length int ) * chainSampleIterator {
csi , ok := it . ( * chainSampleIterator )
if ! ok {
csi = & chainSampleIterator { }
}
if cap ( csi . iterators ) < length {
csi . iterators = make ( [ ] chunkenc . Iterator , length )
} else {
csi . iterators = csi . iterators [ : length ]
}
csi . h = nil
csi . lastT = math . MinInt64
return csi
}
func ChainSampleIteratorFromSeries ( it chunkenc . Iterator , series [ ] Series ) chunkenc . Iterator {
csi := getChainSampleIterator ( it , len ( series ) )
for i , s := range series {
csi . iterators [ i ] = s . Iterator ( csi . iterators [ i ] )
2020-08-03 10:32:56 +00:00
}
2022-09-20 18:31:28 +00:00
return csi
}
2023-11-28 10:14:29 +00:00
func ChainSampleIteratorFromIterables ( it chunkenc . Iterator , iterables [ ] chunkenc . Iterable ) chunkenc . Iterator {
csi := getChainSampleIterator ( it , len ( iterables ) )
for i , c := range iterables {
csi . iterators [ i ] = c . Iterator ( csi . iterators [ i ] )
2022-09-20 18:31:28 +00:00
}
return csi
}
func ChainSampleIteratorFromIterators ( it chunkenc . Iterator , iterators [ ] chunkenc . Iterator ) chunkenc . Iterator {
csi := getChainSampleIterator ( it , 0 )
csi . iterators = iterators
return csi
2020-08-03 10:32:56 +00:00
}
2021-11-29 07:54:23 +00:00
func ( c * chainSampleIterator ) Seek ( t int64 ) chunkenc . ValueType {
// No-op check.
if c . curr != nil && c . lastT >= t {
return c . curr . Seek ( c . lastT )
}
2023-01-18 16:59:29 +00:00
// Don't bother to find out if the next sample is consecutive. Callers
// of Seek usually aren't interested anyway.
c . consecutive = false
2020-08-03 10:32:56 +00:00
c . h = samplesIteratorHeap { }
for _ , iter := range c . iterators {
2023-10-31 22:35:17 +00:00
if iter . Seek ( t ) == chunkenc . ValNone {
if iter . Err ( ) != nil {
// If any iterator is reporting an error, abort.
return chunkenc . ValNone
}
continue
2020-08-03 10:32:56 +00:00
}
2023-10-31 22:35:17 +00:00
heap . Push ( & c . h , iter )
2020-08-03 10:32:56 +00:00
}
2020-09-08 14:23:01 +00:00
if len ( c . h ) > 0 {
c . curr = heap . Pop ( & c . h ) . ( chunkenc . Iterator )
2021-11-29 07:54:23 +00:00
c . lastT = c . curr . AtT ( )
return c . curr . Seek ( c . lastT )
2020-09-08 14:23:01 +00:00
}
c . curr = nil
2021-11-29 07:54:23 +00:00
return chunkenc . ValNone
2020-08-03 10:32:56 +00:00
}
func ( c * chainSampleIterator ) At ( ) ( t int64 , v float64 ) {
2020-09-08 14:23:01 +00:00
if c . curr == nil {
2021-11-29 07:54:23 +00:00
panic ( "chainSampleIterator.At called before first .Next or after .Next returned false." )
2020-08-03 10:32:56 +00:00
}
2020-09-08 14:23:01 +00:00
return c . curr . At ( )
2020-08-03 10:32:56 +00:00
}
2021-11-12 18:07:41 +00:00
func ( c * chainSampleIterator ) AtHistogram ( ) ( int64 , * histogram . Histogram ) {
2021-06-29 14:38:46 +00:00
if c . curr == nil {
2021-11-29 07:54:23 +00:00
panic ( "chainSampleIterator.AtHistogram called before first .Next or after .Next returned false." )
2021-06-29 14:38:46 +00:00
}
2023-01-18 16:59:29 +00:00
t , h := c . curr . AtHistogram ( )
// If the current sample is not consecutive with the previous one, we
2023-09-19 15:06:46 +00:00
// cannot be sure anymore about counter resets for counter histograms.
// TODO(beorn7): If a `NotCounterReset` sample is followed by a
// non-consecutive `CounterReset` sample, we could keep the hint as
// `CounterReset`. But then we needed to track the previous sample
// in more detail, which might not be worth it.
if ! c . consecutive && h . CounterResetHint != histogram . GaugeType {
2023-01-18 16:59:29 +00:00
h . CounterResetHint = histogram . UnknownCounterReset
}
return t , h
2021-06-29 14:38:46 +00:00
}
2021-11-29 07:54:23 +00:00
func ( c * chainSampleIterator ) AtFloatHistogram ( ) ( int64 , * histogram . FloatHistogram ) {
if c . curr == nil {
panic ( "chainSampleIterator.AtFloatHistogram called before first .Next or after .Next returned false." )
}
2023-01-18 16:59:29 +00:00
t , fh := c . curr . AtFloatHistogram ( )
// If the current sample is not consecutive with the previous one, we
2023-01-25 17:23:10 +00:00
// cannot be sure anymore about counter resets for counter histograms.
// TODO(beorn7): If a `NotCounterReset` sample is followed by a
// non-consecutive `CounterReset` sample, we could keep the hint as
// `CounterReset`. But then we needed to track the previous sample
// in more detail, which might not be worth it.
if ! c . consecutive && fh . CounterResetHint != histogram . GaugeType {
2023-01-18 16:59:29 +00:00
fh . CounterResetHint = histogram . UnknownCounterReset
}
return t , fh
2021-11-29 07:54:23 +00:00
}
func ( c * chainSampleIterator ) AtT ( ) int64 {
2021-06-30 14:48:13 +00:00
if c . curr == nil {
2021-11-29 07:54:23 +00:00
panic ( "chainSampleIterator.AtT called before first .Next or after .Next returned false." )
2021-06-30 14:48:13 +00:00
}
2021-11-29 07:54:23 +00:00
return c . curr . AtT ( )
2021-06-30 14:48:13 +00:00
}
2021-11-29 07:54:23 +00:00
func ( c * chainSampleIterator ) Next ( ) chunkenc . ValueType {
2023-01-18 16:59:29 +00:00
var (
currT int64
currValueType chunkenc . ValueType
iteratorChanged bool
)
2020-08-03 10:32:56 +00:00
if c . h == nil {
2023-01-18 16:59:29 +00:00
iteratorChanged = true
2020-09-08 14:23:01 +00:00
c . h = samplesIteratorHeap { }
// We call c.curr.Next() as the first thing below.
// So, we don't call Next() on it here.
c . curr = c . iterators [ 0 ]
for _ , iter := range c . iterators [ 1 : ] {
2023-10-31 22:35:17 +00:00
if iter . Next ( ) == chunkenc . ValNone {
if iter . Err ( ) != nil {
// If any iterator is reporting an error, abort.
// If c.iterators[0] is reporting an error, we'll handle that below.
return chunkenc . ValNone
}
} else {
2020-08-03 10:32:56 +00:00
heap . Push ( & c . h , iter )
}
}
}
2020-09-08 14:23:01 +00:00
if c . curr == nil {
2021-11-29 07:54:23 +00:00
return chunkenc . ValNone
2020-09-04 16:10:42 +00:00
}
2020-09-08 14:23:01 +00:00
for {
2021-11-29 07:54:23 +00:00
currValueType = c . curr . Next ( )
2023-10-31 22:35:17 +00:00
if currValueType == chunkenc . ValNone {
if c . curr . Err ( ) != nil {
// Abort if we've hit an error.
return chunkenc . ValNone
}
if len ( c . h ) == 0 {
// No iterator left to iterate.
c . curr = nil
return chunkenc . ValNone
}
} else {
2021-11-29 07:54:23 +00:00
currT = c . curr . AtT ( )
if currT == c . lastT {
2020-09-08 14:23:01 +00:00
// Ignoring sample for the same timestamp.
continue
}
if len ( c . h ) == 0 {
// curr is the only iterator remaining,
// no need to check with the heap.
break
}
// Check current iterator with the top of the heap.
2021-11-29 07:54:23 +00:00
nextT := c . h [ 0 ] . AtT ( )
if currT < nextT {
2020-09-08 14:23:01 +00:00
// Current iterator has smaller timestamp than the heap.
break
}
// Current iterator does not hold the smallest timestamp.
heap . Push ( & c . h , c . curr )
2020-08-03 10:32:56 +00:00
}
2020-09-08 14:23:01 +00:00
c . curr = heap . Pop ( & c . h ) . ( chunkenc . Iterator )
2023-01-18 16:59:29 +00:00
iteratorChanged = true
2021-11-29 07:54:23 +00:00
currT = c . curr . AtT ( )
currValueType = c . curr . Seek ( currT )
if currT != c . lastT {
2020-09-08 14:23:01 +00:00
break
2020-09-04 16:10:42 +00:00
}
2020-08-03 10:32:56 +00:00
}
2020-09-04 16:10:42 +00:00
2023-01-18 16:59:29 +00:00
c . consecutive = ! iteratorChanged
2021-11-29 07:54:23 +00:00
c . lastT = currT
return currValueType
2020-08-03 10:32:56 +00:00
}
func ( c * chainSampleIterator ) Err ( ) error {
2020-10-28 15:24:58 +00:00
errs := tsdb_errors . NewMulti ( )
2020-08-03 10:32:56 +00:00
for _ , iter := range c . iterators {
2020-10-28 15:24:58 +00:00
errs . Add ( iter . Err ( ) )
2020-08-03 10:32:56 +00:00
}
return errs . Err ( )
}
type samplesIteratorHeap [ ] chunkenc . Iterator
func ( h samplesIteratorHeap ) Len ( ) int { return len ( h ) }
func ( h samplesIteratorHeap ) Swap ( i , j int ) { h [ i ] , h [ j ] = h [ j ] , h [ i ] }
func ( h samplesIteratorHeap ) Less ( i , j int ) bool {
2021-11-29 07:54:23 +00:00
return h [ i ] . AtT ( ) < h [ j ] . AtT ( )
2020-08-03 10:32:56 +00:00
}
func ( h * samplesIteratorHeap ) Push ( x interface { } ) {
* h = append ( * h , x . ( chunkenc . Iterator ) )
}
func ( h * samplesIteratorHeap ) Pop ( ) interface { } {
old := * h
n := len ( old )
x := old [ n - 1 ]
* h = old [ 0 : n - 1 ]
return x
}
// NewCompactingChunkSeriesMerger returns VerticalChunkSeriesMergeFunc that merges the same chunk series into single chunk series.
// In case of the chunk overlaps, it compacts those into one or more time-ordered non-overlapping chunks with merged data.
// Samples from overlapped chunks are merged using series vertical merge func.
// It expects the same labels for each given series.
//
// NOTE: Use the returned merge function only when you see potentially overlapping series, as this introduces small a overhead
// to handle overlaps between series.
func NewCompactingChunkSeriesMerger ( mergeFunc VerticalSeriesMergeFunc ) VerticalChunkSeriesMergeFunc {
return func ( series ... ChunkSeries ) ChunkSeries {
if len ( series ) == 0 {
return nil
}
return & ChunkSeriesEntry {
Lset : series [ 0 ] . Labels ( ) ,
2022-09-20 17:16:45 +00:00
ChunkIteratorFn : func ( chunks . Iterator ) chunks . Iterator {
2020-08-03 10:32:56 +00:00
iterators := make ( [ ] chunks . Iterator , 0 , len ( series ) )
for _ , s := range series {
2022-09-20 17:16:45 +00:00
iterators = append ( iterators , s . Iterator ( nil ) )
2020-08-03 10:32:56 +00:00
}
return & compactChunkIterator {
mergeFunc : mergeFunc ,
iterators : iterators ,
}
} ,
}
}
}
// compactChunkIterator is responsible to compact chunks from different iterators of the same time series into single chainSeries.
// If time-overlapping chunks are found, they are encoded and passed to series merge and encoded again into one bigger chunk.
// TODO(bwplotka): Currently merge will compact overlapping chunks with bigger chunk, without limit. Split it: https://github.com/prometheus/tsdb/issues/670
type compactChunkIterator struct {
mergeFunc VerticalSeriesMergeFunc
iterators [ ] chunks . Iterator
h chunkIteratorHeap
err error
curr chunks . Meta
}
func ( c * compactChunkIterator ) At ( ) chunks . Meta {
return c . curr
}
func ( c * compactChunkIterator ) Next ( ) bool {
if c . h == nil {
for _ , iter := range c . iterators {
if iter . Next ( ) {
heap . Push ( & c . h , iter )
}
}
}
if len ( c . h ) == 0 {
return false
}
iter := heap . Pop ( & c . h ) . ( chunks . Iterator )
c . curr = iter . At ( )
if iter . Next ( ) {
heap . Push ( & c . h , iter )
}
var (
overlapping [ ] Series
oMaxTime = c . curr . MaxTime
prev = c . curr
)
// Detect overlaps to compact. Be smart about it and deduplicate on the fly if chunks are identical.
for len ( c . h ) > 0 {
// Get the next oldest chunk by min, then max time.
next := c . h [ 0 ] . At ( )
if next . MinTime > oMaxTime {
// No overlap with current one.
break
}
2023-04-12 11:05:41 +00:00
// Only do something if it is not a perfect duplicate.
if next . MinTime != prev . MinTime ||
next . MaxTime != prev . MaxTime ||
! bytes . Equal ( next . Chunk . Bytes ( ) , prev . Chunk . Bytes ( ) ) {
// We operate on same series, so labels do not matter here.
2022-03-09 22:21:57 +00:00
overlapping = append ( overlapping , newChunkToSeriesDecoder ( labels . EmptyLabels ( ) , next ) )
2020-08-03 10:32:56 +00:00
if next . MaxTime > oMaxTime {
oMaxTime = next . MaxTime
}
prev = next
}
iter := heap . Pop ( & c . h ) . ( chunks . Iterator )
if iter . Next ( ) {
heap . Push ( & c . h , iter )
}
}
if len ( overlapping ) == 0 {
return true
}
// Add last as it's not yet included in overlap. We operate on same series, so labels does not matter here.
2022-03-09 22:21:57 +00:00
iter = NewSeriesToChunkEncoder ( c . mergeFunc ( append ( overlapping , newChunkToSeriesDecoder ( labels . EmptyLabels ( ) , c . curr ) ) ... ) ) . Iterator ( nil )
2020-08-03 10:32:56 +00:00
if ! iter . Next ( ) {
if c . err = iter . Err ( ) ; c . err != nil {
return false
}
panic ( "unexpected seriesToChunkEncoder lack of iterations" )
}
c . curr = iter . At ( )
if iter . Next ( ) {
heap . Push ( & c . h , iter )
}
return true
}
func ( c * compactChunkIterator ) Err ( ) error {
2020-10-28 15:24:58 +00:00
errs := tsdb_errors . NewMulti ( )
2020-08-03 10:32:56 +00:00
for _ , iter := range c . iterators {
2020-10-28 15:24:58 +00:00
errs . Add ( iter . Err ( ) )
2020-08-03 10:32:56 +00:00
}
errs . Add ( c . err )
return errs . Err ( )
}
type chunkIteratorHeap [ ] chunks . Iterator
func ( h chunkIteratorHeap ) Len ( ) int { return len ( h ) }
func ( h chunkIteratorHeap ) Swap ( i , j int ) { h [ i ] , h [ j ] = h [ j ] , h [ i ] }
func ( h chunkIteratorHeap ) Less ( i , j int ) bool {
at := h [ i ] . At ( )
bt := h [ j ] . At ( )
if at . MinTime == bt . MinTime {
return at . MaxTime < bt . MaxTime
}
return at . MinTime < bt . MinTime
}
func ( h * chunkIteratorHeap ) Push ( x interface { } ) {
* h = append ( * h , x . ( chunks . Iterator ) )
}
func ( h * chunkIteratorHeap ) Pop ( ) interface { } {
old := * h
n := len ( old )
x := old [ n - 1 ]
* h = old [ 0 : n - 1 ]
return x
}
2022-09-20 17:05:50 +00:00
// NewConcatenatingChunkSeriesMerger returns a VerticalChunkSeriesMergeFunc that simply concatenates the
// chunks from the series. The resultant stream of chunks for a series might be overlapping and unsorted.
func NewConcatenatingChunkSeriesMerger ( ) VerticalChunkSeriesMergeFunc {
return func ( series ... ChunkSeries ) ChunkSeries {
if len ( series ) == 0 {
return nil
}
return & ChunkSeriesEntry {
Lset : series [ 0 ] . Labels ( ) ,
2022-09-20 17:16:45 +00:00
ChunkIteratorFn : func ( chunks . Iterator ) chunks . Iterator {
2022-09-20 17:05:50 +00:00
iterators := make ( [ ] chunks . Iterator , 0 , len ( series ) )
for _ , s := range series {
2022-09-20 17:16:45 +00:00
iterators = append ( iterators , s . Iterator ( nil ) )
2022-09-20 17:05:50 +00:00
}
return & concatenatingChunkIterator {
iterators : iterators ,
}
} ,
}
}
}
type concatenatingChunkIterator struct {
iterators [ ] chunks . Iterator
idx int
curr chunks . Meta
}
func ( c * concatenatingChunkIterator ) At ( ) chunks . Meta {
return c . curr
}
func ( c * concatenatingChunkIterator ) Next ( ) bool {
if c . idx >= len ( c . iterators ) {
return false
}
if c . iterators [ c . idx ] . Next ( ) {
c . curr = c . iterators [ c . idx ] . At ( )
return true
}
2023-11-02 04:29:09 +00:00
if c . iterators [ c . idx ] . Err ( ) != nil {
return false
}
2022-09-20 17:05:50 +00:00
c . idx ++
return c . Next ( )
}
func ( c * concatenatingChunkIterator ) Err ( ) error {
errs := tsdb_errors . NewMulti ( )
for _ , iter := range c . iterators {
errs . Add ( iter . Err ( ) )
}
return errs . Err ( )
}