prometheus

1
// Copyright 2020 The Prometheus Authors
2
// Licensed under the Apache License, Version 2.0 (the "License");
3
// you may not use this file except in compliance with the License.
4
// You may obtain a copy of the License at
5
//
6
// http://www.apache.org/licenses/LICENSE-2.0
7
//
8
// Unless required by applicable law or agreed to in writing, software
9
// distributed under the License is distributed on an "AS IS" BASIS,
10
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11
// See the License for the specific language governing permissions and
12
// limitations under the License.
13

14
package storage
15

16
import (
17
	"fmt"
18
	"math"
19
	"sort"
20

21
	"github.com/prometheus/prometheus/model/histogram"
22
	"github.com/prometheus/prometheus/model/labels"
23
	"github.com/prometheus/prometheus/tsdb/chunkenc"
24
	"github.com/prometheus/prometheus/tsdb/chunks"
25
)
26

27
type SeriesEntry struct {
28
	Lset             labels.Labels
29
	SampleIteratorFn func(chunkenc.Iterator) chunkenc.Iterator
30
}
31

32
func (s *SeriesEntry) Labels() labels.Labels                           { return s.Lset }
33
func (s *SeriesEntry) Iterator(it chunkenc.Iterator) chunkenc.Iterator { return s.SampleIteratorFn(it) }
34

35
type ChunkSeriesEntry struct {
36
	Lset            labels.Labels
37
	ChunkIteratorFn func(chunks.Iterator) chunks.Iterator
38
}
39

40
func (s *ChunkSeriesEntry) Labels() labels.Labels                       { return s.Lset }
41
func (s *ChunkSeriesEntry) Iterator(it chunks.Iterator) chunks.Iterator { return s.ChunkIteratorFn(it) }
42

43
// NewListSeries returns series entry with iterator that allows to iterate over provided samples.
44
func NewListSeries(lset labels.Labels, s []chunks.Sample) *SeriesEntry {
45
	samplesS := Samples(samples(s))
46
	return &SeriesEntry{
47
		Lset: lset,
48
		SampleIteratorFn: func(it chunkenc.Iterator) chunkenc.Iterator {
49
			if lsi, ok := it.(*listSeriesIterator); ok {
50
				lsi.Reset(samplesS)
51
				return lsi
52
			}
53
			return NewListSeriesIterator(samplesS)
54
		},
55
	}
56
}
57

58
// NewListChunkSeriesFromSamples returns a chunk series entry that allows to iterate over provided samples.
59
// NOTE: It uses an inefficient chunks encoding implementation, not caring about chunk size.
60
// Use only for testing.
61
func NewListChunkSeriesFromSamples(lset labels.Labels, samples ...[]chunks.Sample) *ChunkSeriesEntry {
62
	chksFromSamples := make([]chunks.Meta, 0, len(samples))
63
	for _, s := range samples {
64
		cfs, err := chunks.ChunkFromSamples(s)
65
		if err != nil {
66
			return &ChunkSeriesEntry{
67
				Lset: lset,
68
				ChunkIteratorFn: func(it chunks.Iterator) chunks.Iterator {
69
					return errChunksIterator{err: err}
70
				},
71
			}
72
		}
73
		chksFromSamples = append(chksFromSamples, cfs)
74
	}
75
	return &ChunkSeriesEntry{
76
		Lset: lset,
77
		ChunkIteratorFn: func(it chunks.Iterator) chunks.Iterator {
78
			lcsi, existing := it.(*listChunkSeriesIterator)
79
			var chks []chunks.Meta
80
			if existing {
81
				chks = lcsi.chks[:0]
82
			} else {
83
				chks = make([]chunks.Meta, 0, len(samples))
84
			}
85
			chks = append(chks, chksFromSamples...)
86
			if existing {
87
				lcsi.Reset(chks...)
88
				return lcsi
89
			}
90
			return NewListChunkSeriesIterator(chks...)
91
		},
92
	}
93
}
94

95
type listSeriesIterator struct {
96
	samples Samples
97
	idx     int
98
}
99

100
type samples []chunks.Sample
101

102
func (s samples) Get(i int) chunks.Sample { return s[i] }
103
func (s samples) Len() int                { return len(s) }
104

105
// Samples interface allows to work on arrays of types that are compatible with chunks.Sample.
106
type Samples interface {
107
	Get(i int) chunks.Sample
108
	Len() int
109
}
110

111
// NewListSeriesIterator returns listSeriesIterator that allows to iterate over provided samples.
112
func NewListSeriesIterator(samples Samples) chunkenc.Iterator {
113
	return &listSeriesIterator{samples: samples, idx: -1}
114
}
115

116
func (it *listSeriesIterator) Reset(samples Samples) {
117
	it.samples = samples
118
	it.idx = -1
119
}
120

121
func (it *listSeriesIterator) At() (int64, float64) {
122
	s := it.samples.Get(it.idx)
123
	return s.T(), s.F()
124
}
125

126
func (it *listSeriesIterator) AtHistogram(*histogram.Histogram) (int64, *histogram.Histogram) {
127
	s := it.samples.Get(it.idx)
128
	return s.T(), s.H()
129
}
130

131
func (it *listSeriesIterator) AtFloatHistogram(*histogram.FloatHistogram) (int64, *histogram.FloatHistogram) {
132
	s := it.samples.Get(it.idx)
133
	return s.T(), s.FH()
134
}
135

136
func (it *listSeriesIterator) AtT() int64 {
137
	s := it.samples.Get(it.idx)
138
	return s.T()
139
}
140

141
func (it *listSeriesIterator) Next() chunkenc.ValueType {
142
	it.idx++
143
	if it.idx >= it.samples.Len() {
144
		return chunkenc.ValNone
145
	}
146
	return it.samples.Get(it.idx).Type()
147
}
148

149
func (it *listSeriesIterator) Seek(t int64) chunkenc.ValueType {
150
	if it.idx == -1 {
151
		it.idx = 0
152
	}
153
	if it.idx >= it.samples.Len() {
154
		return chunkenc.ValNone
155
	}
156
	// No-op check.
157
	if s := it.samples.Get(it.idx); s.T() >= t {
158
		return s.Type()
159
	}
160
	// Do binary search between current position and end.
161
	it.idx += sort.Search(it.samples.Len()-it.idx, func(i int) bool {
162
		s := it.samples.Get(i + it.idx)
163
		return s.T() >= t
164
	})
165

166
	if it.idx >= it.samples.Len() {
167
		return chunkenc.ValNone
168
	}
169
	return it.samples.Get(it.idx).Type()
170
}
171

172
func (it *listSeriesIterator) Err() error { return nil }
173

174
type listChunkSeriesIterator struct {
175
	chks []chunks.Meta
176
	idx  int
177
}
178

179
// NewListChunkSeriesIterator returns listChunkSeriesIterator that allows to iterate over provided chunks.
180
func NewListChunkSeriesIterator(chks ...chunks.Meta) chunks.Iterator {
181
	return &listChunkSeriesIterator{chks: chks, idx: -1}
182
}
183

184
func (it *listChunkSeriesIterator) Reset(chks ...chunks.Meta) {
185
	it.chks = chks
186
	it.idx = -1
187
}
188

189
func (it *listChunkSeriesIterator) At() chunks.Meta {
190
	return it.chks[it.idx]
191
}
192

193
func (it *listChunkSeriesIterator) Next() bool {
194
	it.idx++
195
	return it.idx < len(it.chks)
196
}
197

198
func (it *listChunkSeriesIterator) Err() error { return nil }
199

200
type chunkSetToSeriesSet struct {
201
	ChunkSeriesSet
202

203
	iter             chunks.Iterator
204
	chkIterErr       error
205
	sameSeriesChunks []Series
206
}
207

208
// NewSeriesSetFromChunkSeriesSet converts ChunkSeriesSet to SeriesSet by decoding chunks one by one.
209
func NewSeriesSetFromChunkSeriesSet(chk ChunkSeriesSet) SeriesSet {
210
	return &chunkSetToSeriesSet{ChunkSeriesSet: chk}
211
}
212

213
func (c *chunkSetToSeriesSet) Next() bool {
214
	if c.Err() != nil || !c.ChunkSeriesSet.Next() {
215
		return false
216
	}
217

218
	c.iter = c.ChunkSeriesSet.At().Iterator(c.iter)
219
	c.sameSeriesChunks = nil
220

221
	for c.iter.Next() {
222
		c.sameSeriesChunks = append(
223
			c.sameSeriesChunks,
224
			newChunkToSeriesDecoder(c.ChunkSeriesSet.At().Labels(), c.iter.At()),
225
		)
226
	}
227

228
	if c.iter.Err() != nil {
229
		c.chkIterErr = c.iter.Err()
230
		return false
231
	}
232
	return true
233
}
234

235
func (c *chunkSetToSeriesSet) At() Series {
236
	// Series composed of same chunks for the same series.
237
	return ChainedSeriesMerge(c.sameSeriesChunks...)
238
}
239

240
func (c *chunkSetToSeriesSet) Err() error {
241
	if c.chkIterErr != nil {
242
		return c.chkIterErr
243
	}
244
	return c.ChunkSeriesSet.Err()
245
}
246

247
func newChunkToSeriesDecoder(labels labels.Labels, chk chunks.Meta) Series {
248
	return &SeriesEntry{
249
		Lset: labels,
250
		SampleIteratorFn: func(it chunkenc.Iterator) chunkenc.Iterator {
251
			// TODO(bwplotka): Can we provide any chunkenc buffer?
252
			return chk.Chunk.Iterator(it)
253
		},
254
	}
255
}
256

257
type seriesSetToChunkSet struct {
258
	SeriesSet
259
}
260

261
// NewSeriesSetToChunkSet converts SeriesSet to ChunkSeriesSet by encoding chunks from samples.
262
func NewSeriesSetToChunkSet(chk SeriesSet) ChunkSeriesSet {
263
	return &seriesSetToChunkSet{SeriesSet: chk}
264
}
265

266
func (c *seriesSetToChunkSet) Next() bool {
267
	if c.Err() != nil || !c.SeriesSet.Next() {
268
		return false
269
	}
270
	return true
271
}
272

273
func (c *seriesSetToChunkSet) At() ChunkSeries {
274
	return NewSeriesToChunkEncoder(c.SeriesSet.At())
275
}
276

277
func (c *seriesSetToChunkSet) Err() error {
278
	return c.SeriesSet.Err()
279
}
280

281
type seriesToChunkEncoder struct {
282
	Series
283
}
284

285
const seriesToChunkEncoderSplit = 120
286

287
// NewSeriesToChunkEncoder encodes samples to chunks with 120 samples limit.
288
func NewSeriesToChunkEncoder(series Series) ChunkSeries {
289
	return &seriesToChunkEncoder{series}
290
}
291

292
func (s *seriesToChunkEncoder) Iterator(it chunks.Iterator) chunks.Iterator {
293
	var (
294
		chk, newChk chunkenc.Chunk
295
		app         chunkenc.Appender
296
		err         error
297
		recoded     bool
298
	)
299
	mint := int64(math.MaxInt64)
300
	maxt := int64(math.MinInt64)
301

302
	var chks []chunks.Meta
303
	lcsi, existing := it.(*listChunkSeriesIterator)
304
	if existing {
305
		chks = lcsi.chks[:0]
306
	}
307

308
	i := 0
309
	seriesIter := s.Series.Iterator(nil)
310
	lastType := chunkenc.ValNone
311
	for typ := seriesIter.Next(); typ != chunkenc.ValNone; typ = seriesIter.Next() {
312
		if typ != lastType || i >= seriesToChunkEncoderSplit {
313
			// Create a new chunk if the sample type changed or too many samples in the current one.
314
			chks = appendChunk(chks, mint, maxt, chk)
315
			chk, err = chunkenc.NewEmptyChunk(typ.ChunkEncoding())
316
			if err != nil {
317
				return errChunksIterator{err: err}
318
			}
319
			app, err = chk.Appender()
320
			if err != nil {
321
				return errChunksIterator{err: err}
322
			}
323
			mint = int64(math.MaxInt64)
324
			// maxt is immediately overwritten below which is why setting it here won't make a difference.
325
			i = 0
326
		}
327
		lastType = typ
328

329
		var (
330
			t  int64
331
			v  float64
332
			h  *histogram.Histogram
333
			fh *histogram.FloatHistogram
334
		)
335
		switch typ {
336
		case chunkenc.ValFloat:
337
			t, v = seriesIter.At()
338
			app.Append(t, v)
339
		case chunkenc.ValHistogram:
340
			t, h = seriesIter.AtHistogram(nil)
341
			newChk, recoded, app, err = app.AppendHistogram(nil, t, h, false)
342
			if err != nil {
343
				return errChunksIterator{err: err}
344
			}
345
			if newChk != nil {
346
				if !recoded {
347
					chks = appendChunk(chks, mint, maxt, chk)
348
					mint = int64(math.MaxInt64)
349
					// maxt is immediately overwritten below which is why setting it here won't make a difference.
350
					i = 0
351
				}
352
				chk = newChk
353
			}
354
		case chunkenc.ValFloatHistogram:
355
			t, fh = seriesIter.AtFloatHistogram(nil)
356
			newChk, recoded, app, err = app.AppendFloatHistogram(nil, t, fh, false)
357
			if err != nil {
358
				return errChunksIterator{err: err}
359
			}
360
			if newChk != nil {
361
				if !recoded {
362
					chks = appendChunk(chks, mint, maxt, chk)
363
					mint = int64(math.MaxInt64)
364
					// maxt is immediately overwritten below which is why setting it here won't make a difference.
365
					i = 0
366
				}
367
				chk = newChk
368
			}
369
		default:
370
			return errChunksIterator{err: fmt.Errorf("unknown sample type %s", typ.String())}
371
		}
372

373
		maxt = t
374
		if mint == math.MaxInt64 {
375
			mint = t
376
		}
377
		i++
378
	}
379
	if err := seriesIter.Err(); err != nil {
380
		return errChunksIterator{err: err}
381
	}
382

383
	chks = appendChunk(chks, mint, maxt, chk)
384

385
	if existing {
386
		lcsi.Reset(chks...)
387
		return lcsi
388
	}
389
	return NewListChunkSeriesIterator(chks...)
390
}
391

392
func appendChunk(chks []chunks.Meta, mint, maxt int64, chk chunkenc.Chunk) []chunks.Meta {
393
	if chk != nil {
394
		chks = append(chks, chunks.Meta{
395
			MinTime: mint,
396
			MaxTime: maxt,
397
			Chunk:   chk,
398
		})
399
	}
400
	return chks
401
}
402

403
type errChunksIterator struct {
404
	err error
405
}
406

407
func (e errChunksIterator) At() chunks.Meta { return chunks.Meta{} }
408
func (e errChunksIterator) Next() bool      { return false }
409
func (e errChunksIterator) Err() error      { return e.err }
410

411
// ExpandSamples iterates over all samples in the iterator, buffering all in slice.
412
// Optionally it takes samples constructor, useful when you want to compare sample slices with different
413
// sample implementations. if nil, sample type from this package will be used.
414
func ExpandSamples(iter chunkenc.Iterator, newSampleFn func(t int64, f float64, h *histogram.Histogram, fh *histogram.FloatHistogram) chunks.Sample) ([]chunks.Sample, error) {
415
	if newSampleFn == nil {
416
		newSampleFn = func(t int64, f float64, h *histogram.Histogram, fh *histogram.FloatHistogram) chunks.Sample {
417
			switch {
418
			case h != nil:
419
				return hSample{t, h}
420
			case fh != nil:
421
				return fhSample{t, fh}
422
			default:
423
				return fSample{t, f}
424
			}
425
		}
426
	}
427

428
	var result []chunks.Sample
429
	for {
430
		switch iter.Next() {
431
		case chunkenc.ValNone:
432
			return result, iter.Err()
433
		case chunkenc.ValFloat:
434
			t, f := iter.At()
435
			// NaNs can't be compared normally, so substitute for another value.
436
			if math.IsNaN(f) {
437
				f = -42
438
			}
439
			result = append(result, newSampleFn(t, f, nil, nil))
440
		case chunkenc.ValHistogram:
441
			t, h := iter.AtHistogram(nil)
442
			result = append(result, newSampleFn(t, 0, h, nil))
443
		case chunkenc.ValFloatHistogram:
444
			t, fh := iter.AtFloatHistogram(nil)
445
			result = append(result, newSampleFn(t, 0, nil, fh))
446
		}
447
	}
448
}
449

450
// ExpandChunks iterates over all chunks in the iterator, buffering all in slice.
451
func ExpandChunks(iter chunks.Iterator) ([]chunks.Meta, error) {
452
	var result []chunks.Meta
453
	for iter.Next() {
454
		result = append(result, iter.At())
455
	}
456
	return result, iter.Err()
457
}
458