2019-05-22 23:16:55 +02:00
|
|
|
package netstorage
|
|
|
|
|
|
|
|
import (
|
|
|
|
"container/heap"
|
|
|
|
"flag"
|
|
|
|
"fmt"
|
|
|
|
"runtime"
|
|
|
|
"sort"
|
|
|
|
"sync"
|
|
|
|
"time"
|
|
|
|
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
|
2020-06-24 18:36:55 +02:00
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
|
2019-05-22 23:16:55 +02:00
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
|
|
|
|
"github.com/VictoriaMetrics/metrics"
|
|
|
|
)
|
|
|
|
|
|
|
|
var (
|
2019-08-27 13:25:18 +02:00
|
|
|
maxTagKeysPerSearch = flag.Int("search.maxTagKeys", 100e3, "The maximum number of tag keys returned per search")
|
|
|
|
maxTagValuesPerSearch = flag.Int("search.maxTagValues", 100e3, "The maximum number of tag values returned per search")
|
|
|
|
maxMetricsPerSearch = flag.Int("search.maxUniqueTimeseries", 300e3, "The maximum number of unique time series each search can scan")
|
2019-05-22 23:16:55 +02:00
|
|
|
)
|
|
|
|
|
|
|
|
// Result is a single timeseries result.
|
|
|
|
//
|
|
|
|
// ProcessSearchQuery returns Result slice.
|
|
|
|
type Result struct {
|
|
|
|
// The name of the metric.
|
|
|
|
MetricName storage.MetricName
|
|
|
|
|
|
|
|
// Values are sorted by Timestamps.
|
|
|
|
Values []float64
|
|
|
|
Timestamps []int64
|
|
|
|
|
|
|
|
// Marshaled MetricName. Used only for results sorting
|
|
|
|
// in app/vmselect/promql
|
|
|
|
MetricNameMarshaled []byte
|
|
|
|
}
|
|
|
|
|
|
|
|
func (r *Result) reset() {
|
|
|
|
r.MetricName.Reset()
|
|
|
|
r.Values = r.Values[:0]
|
|
|
|
r.Timestamps = r.Timestamps[:0]
|
|
|
|
r.MetricNameMarshaled = r.MetricNameMarshaled[:0]
|
|
|
|
}
|
|
|
|
|
|
|
|
// Results holds results returned from ProcessSearchQuery.
|
|
|
|
type Results struct {
|
2019-08-04 21:15:33 +02:00
|
|
|
tr storage.TimeRange
|
|
|
|
fetchData bool
|
|
|
|
deadline Deadline
|
2019-05-22 23:16:55 +02:00
|
|
|
|
|
|
|
packedTimeseries []packedTimeseries
|
2020-04-27 07:13:41 +02:00
|
|
|
sr *storage.Search
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// Len returns the number of results in rss.
|
|
|
|
func (rss *Results) Len() int {
|
|
|
|
return len(rss.packedTimeseries)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Cancel cancels rss work.
|
|
|
|
func (rss *Results) Cancel() {
|
2020-04-27 07:13:41 +02:00
|
|
|
rss.mustClose()
|
|
|
|
}
|
|
|
|
|
|
|
|
func (rss *Results) mustClose() {
|
|
|
|
putStorageSearch(rss.sr)
|
|
|
|
rss.sr = nil
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
|
2020-06-23 19:29:19 +02:00
|
|
|
var timeseriesWorkCh = make(chan *timeseriesWork, gomaxprocs)
|
|
|
|
|
|
|
|
type timeseriesWork struct {
|
|
|
|
rss *Results
|
|
|
|
pts *packedTimeseries
|
|
|
|
f func(rs *Result, workerID uint)
|
|
|
|
doneCh chan error
|
|
|
|
|
|
|
|
rowsProcessed int
|
|
|
|
}
|
|
|
|
|
|
|
|
func init() {
|
|
|
|
for i := 0; i < gomaxprocs; i++ {
|
|
|
|
go timeseriesWorker(uint(i))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func timeseriesWorker(workerID uint) {
|
|
|
|
var rs Result
|
2020-06-24 18:36:55 +02:00
|
|
|
var rsLastResetTime uint64
|
2020-06-23 19:29:19 +02:00
|
|
|
for tsw := range timeseriesWorkCh {
|
|
|
|
rss := tsw.rss
|
|
|
|
if time.Until(rss.deadline.Deadline) < 0 {
|
|
|
|
tsw.doneCh <- fmt.Errorf("timeout exceeded during query execution: %s", rss.deadline.String())
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
if err := tsw.pts.Unpack(&rs, rss.tr, rss.fetchData); err != nil {
|
|
|
|
tsw.doneCh <- fmt.Errorf("error during time series unpacking: %s", err)
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
if len(rs.Timestamps) > 0 || !rss.fetchData {
|
|
|
|
tsw.f(&rs, workerID)
|
|
|
|
}
|
|
|
|
tsw.rowsProcessed = len(rs.Values)
|
|
|
|
tsw.doneCh <- nil
|
2020-06-24 18:36:55 +02:00
|
|
|
currentTime := fasttime.UnixTimestamp()
|
|
|
|
if cap(rs.Values) > 1024*1024 && 4*len(rs.Values) < cap(rs.Values) && currentTime-rsLastResetTime > 10 {
|
2020-06-23 19:29:19 +02:00
|
|
|
// Reset rs in order to preseve memory usage after processing big time series with millions of rows.
|
|
|
|
rs = Result{}
|
2020-06-24 18:36:55 +02:00
|
|
|
rsLastResetTime = currentTime
|
2020-06-23 19:29:19 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-05-22 23:16:55 +02:00
|
|
|
// RunParallel runs in parallel f for all the results from rss.
|
|
|
|
//
|
|
|
|
// f shouldn't hold references to rs after returning.
|
2019-07-12 14:51:02 +02:00
|
|
|
// workerID is the id of the worker goroutine that calls f.
|
2019-05-22 23:16:55 +02:00
|
|
|
//
|
|
|
|
// rss becomes unusable after the call to RunParallel.
|
2019-07-12 14:51:02 +02:00
|
|
|
func (rss *Results) RunParallel(f func(rs *Result, workerID uint)) error {
|
2020-04-27 07:13:41 +02:00
|
|
|
defer rss.mustClose()
|
2019-05-22 23:16:55 +02:00
|
|
|
|
|
|
|
// Feed workers with work.
|
2020-06-23 19:29:19 +02:00
|
|
|
tsws := make([]*timeseriesWork, len(rss.packedTimeseries))
|
2019-05-22 23:16:55 +02:00
|
|
|
for i := range rss.packedTimeseries {
|
2020-06-23 19:29:19 +02:00
|
|
|
tsw := ×eriesWork{
|
|
|
|
rss: rss,
|
|
|
|
pts: &rss.packedTimeseries[i],
|
|
|
|
f: f,
|
|
|
|
doneCh: make(chan error, 1),
|
|
|
|
}
|
|
|
|
timeseriesWorkCh <- tsw
|
|
|
|
tsws[i] = tsw
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
2019-11-23 12:22:55 +01:00
|
|
|
seriesProcessedTotal := len(rss.packedTimeseries)
|
2019-05-22 23:16:55 +02:00
|
|
|
rss.packedTimeseries = rss.packedTimeseries[:0]
|
|
|
|
|
2020-06-23 19:29:19 +02:00
|
|
|
// Wait until work is complete.
|
|
|
|
var firstErr error
|
|
|
|
rowsProcessedTotal := 0
|
|
|
|
for _, tsw := range tsws {
|
|
|
|
if err := <-tsw.doneCh; err != nil && firstErr == nil {
|
|
|
|
// Return just the first error, since other errors
|
|
|
|
// are likely duplicate the first error.
|
|
|
|
firstErr = err
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
2020-06-23 19:29:19 +02:00
|
|
|
rowsProcessedTotal += tsw.rowsProcessed
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
2020-06-23 19:29:19 +02:00
|
|
|
|
2019-11-23 12:22:55 +01:00
|
|
|
perQueryRowsProcessed.Update(float64(rowsProcessedTotal))
|
|
|
|
perQuerySeriesProcessed.Update(float64(seriesProcessedTotal))
|
2020-06-23 19:29:19 +02:00
|
|
|
return firstErr
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
|
2019-11-23 12:22:55 +01:00
|
|
|
var perQueryRowsProcessed = metrics.NewHistogram(`vm_per_query_rows_processed_count`)
|
|
|
|
var perQuerySeriesProcessed = metrics.NewHistogram(`vm_per_query_series_processed_count`)
|
|
|
|
|
2019-05-22 23:16:55 +02:00
|
|
|
var gomaxprocs = runtime.GOMAXPROCS(-1)
|
|
|
|
|
|
|
|
type packedTimeseries struct {
|
|
|
|
metricName string
|
2020-04-27 07:13:41 +02:00
|
|
|
brs []storage.BlockRef
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
|
2020-06-23 19:29:19 +02:00
|
|
|
var unpackWorkCh = make(chan *unpackWork, gomaxprocs)
|
2019-05-22 23:16:55 +02:00
|
|
|
|
2020-06-23 19:29:19 +02:00
|
|
|
type unpackWork struct {
|
|
|
|
br storage.BlockRef
|
|
|
|
tr storage.TimeRange
|
|
|
|
fetchData bool
|
|
|
|
doneCh chan error
|
|
|
|
sb *sortBlock
|
|
|
|
}
|
2019-05-22 23:16:55 +02:00
|
|
|
|
2020-06-23 19:29:19 +02:00
|
|
|
func init() {
|
|
|
|
for i := 0; i < gomaxprocs; i++ {
|
|
|
|
go unpackWorker()
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
2020-06-23 19:29:19 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
func unpackWorker() {
|
|
|
|
for upw := range unpackWorkCh {
|
|
|
|
sb := getSortBlock()
|
|
|
|
if err := sb.unpackFrom(upw.br, upw.tr, upw.fetchData); err != nil {
|
|
|
|
putSortBlock(sb)
|
|
|
|
upw.doneCh <- fmt.Errorf("cannot unpack block: %s", err)
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
upw.sb = sb
|
|
|
|
upw.doneCh <- nil
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
2020-06-23 19:29:19 +02:00
|
|
|
}
|
2019-05-22 23:16:55 +02:00
|
|
|
|
2020-06-23 19:29:19 +02:00
|
|
|
// Unpack unpacks pts to dst.
|
|
|
|
func (pts *packedTimeseries) Unpack(dst *Result, tr storage.TimeRange, fetchData bool) error {
|
|
|
|
dst.reset()
|
2019-05-22 23:16:55 +02:00
|
|
|
|
2020-06-23 19:29:19 +02:00
|
|
|
if err := dst.MetricName.Unmarshal(bytesutil.ToUnsafeBytes(pts.metricName)); err != nil {
|
|
|
|
return fmt.Errorf("cannot unmarshal metricName %q: %s", pts.metricName, err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// Feed workers with work
|
2020-06-23 19:29:19 +02:00
|
|
|
upws := make([]*unpackWork, len(pts.brs))
|
|
|
|
for i, br := range pts.brs {
|
|
|
|
upw := &unpackWork{
|
|
|
|
br: br,
|
|
|
|
tr: tr,
|
|
|
|
fetchData: fetchData,
|
|
|
|
doneCh: make(chan error, 1),
|
|
|
|
}
|
|
|
|
unpackWorkCh <- upw
|
|
|
|
upws[i] = upw
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
2020-04-27 07:13:41 +02:00
|
|
|
pts.brs = pts.brs[:0]
|
2019-05-22 23:16:55 +02:00
|
|
|
|
2020-06-23 19:29:19 +02:00
|
|
|
// Wait until work is complete
|
|
|
|
sbs := make([]*sortBlock, 0, len(pts.brs))
|
|
|
|
var firstErr error
|
|
|
|
for _, upw := range upws {
|
|
|
|
if err := <-upw.doneCh; err != nil && firstErr == nil {
|
|
|
|
// Return the first error only, since other errors are likely the same.
|
|
|
|
firstErr = err
|
|
|
|
}
|
|
|
|
if firstErr == nil {
|
|
|
|
sbs = append(sbs, upw.sb)
|
|
|
|
} else {
|
|
|
|
putSortBlock(upw.sb)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
}
|
2020-06-23 19:29:19 +02:00
|
|
|
if firstErr != nil {
|
|
|
|
return firstErr
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
mergeSortBlocks(dst, sbs)
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func getSortBlock() *sortBlock {
|
|
|
|
v := sbPool.Get()
|
|
|
|
if v == nil {
|
|
|
|
return &sortBlock{}
|
|
|
|
}
|
|
|
|
return v.(*sortBlock)
|
|
|
|
}
|
|
|
|
|
|
|
|
func putSortBlock(sb *sortBlock) {
|
|
|
|
sb.reset()
|
|
|
|
sbPool.Put(sb)
|
|
|
|
}
|
|
|
|
|
|
|
|
var sbPool sync.Pool
|
|
|
|
|
|
|
|
var metricRowsSkipped = metrics.NewCounter(`vm_metric_rows_skipped_total{name="vmselect"}`)
|
|
|
|
|
|
|
|
func mergeSortBlocks(dst *Result, sbh sortBlocksHeap) {
|
|
|
|
// Skip empty sort blocks, since they cannot be passed to heap.Init.
|
|
|
|
src := sbh
|
|
|
|
sbh = sbh[:0]
|
|
|
|
for _, sb := range src {
|
|
|
|
if len(sb.Timestamps) == 0 {
|
|
|
|
putSortBlock(sb)
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
sbh = append(sbh, sb)
|
|
|
|
}
|
|
|
|
if len(sbh) == 0 {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
heap.Init(&sbh)
|
|
|
|
for {
|
|
|
|
top := sbh[0]
|
|
|
|
heap.Pop(&sbh)
|
|
|
|
if len(sbh) == 0 {
|
|
|
|
dst.Timestamps = append(dst.Timestamps, top.Timestamps[top.NextIdx:]...)
|
|
|
|
dst.Values = append(dst.Values, top.Values[top.NextIdx:]...)
|
|
|
|
putSortBlock(top)
|
2020-01-31 00:09:44 +01:00
|
|
|
break
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
sbNext := sbh[0]
|
|
|
|
tsNext := sbNext.Timestamps[sbNext.NextIdx]
|
|
|
|
idxNext := len(top.Timestamps)
|
|
|
|
if top.Timestamps[idxNext-1] > tsNext {
|
|
|
|
idxNext = top.NextIdx
|
|
|
|
for top.Timestamps[idxNext] <= tsNext {
|
|
|
|
idxNext++
|
|
|
|
}
|
|
|
|
}
|
|
|
|
dst.Timestamps = append(dst.Timestamps, top.Timestamps[top.NextIdx:idxNext]...)
|
|
|
|
dst.Values = append(dst.Values, top.Values[top.NextIdx:idxNext]...)
|
|
|
|
if idxNext < len(top.Timestamps) {
|
|
|
|
top.NextIdx = idxNext
|
|
|
|
heap.Push(&sbh, top)
|
|
|
|
} else {
|
|
|
|
// Return top to the pool.
|
|
|
|
putSortBlock(top)
|
|
|
|
}
|
|
|
|
}
|
2020-01-31 00:09:44 +01:00
|
|
|
|
2020-02-27 22:47:05 +01:00
|
|
|
timestamps, values := storage.DeduplicateSamples(dst.Timestamps, dst.Values)
|
|
|
|
dedups := len(dst.Timestamps) - len(timestamps)
|
|
|
|
dedupsDuringSelect.Add(dedups)
|
|
|
|
dst.Timestamps = timestamps
|
|
|
|
dst.Values = values
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
|
2020-02-27 22:47:05 +01:00
|
|
|
var dedupsDuringSelect = metrics.NewCounter(`vm_deduplicated_samples_total{type="select"}`)
|
|
|
|
|
2019-05-22 23:16:55 +02:00
|
|
|
type sortBlock struct {
|
|
|
|
// b is used as a temporary storage for unpacked rows before they
|
|
|
|
// go to Timestamps and Values.
|
|
|
|
b storage.Block
|
|
|
|
|
|
|
|
Timestamps []int64
|
|
|
|
Values []float64
|
|
|
|
NextIdx int
|
|
|
|
}
|
|
|
|
|
|
|
|
func (sb *sortBlock) reset() {
|
|
|
|
sb.b.Reset()
|
|
|
|
sb.Timestamps = sb.Timestamps[:0]
|
|
|
|
sb.Values = sb.Values[:0]
|
|
|
|
sb.NextIdx = 0
|
|
|
|
}
|
|
|
|
|
2020-04-27 07:13:41 +02:00
|
|
|
func (sb *sortBlock) unpackFrom(br storage.BlockRef, tr storage.TimeRange, fetchData bool) error {
|
|
|
|
br.MustReadBlock(&sb.b, fetchData)
|
2019-08-04 21:15:33 +02:00
|
|
|
if fetchData {
|
|
|
|
if err := sb.b.UnmarshalData(); err != nil {
|
|
|
|
return fmt.Errorf("cannot unmarshal block: %s", err)
|
|
|
|
}
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
timestamps := sb.b.Timestamps()
|
|
|
|
|
|
|
|
// Skip timestamps smaller than tr.MinTimestamp.
|
|
|
|
i := 0
|
|
|
|
for i < len(timestamps) && timestamps[i] < tr.MinTimestamp {
|
|
|
|
i++
|
|
|
|
}
|
|
|
|
|
|
|
|
// Skip timestamps bigger than tr.MaxTimestamp.
|
|
|
|
j := len(timestamps)
|
|
|
|
for j > i && timestamps[j-1] > tr.MaxTimestamp {
|
|
|
|
j--
|
|
|
|
}
|
|
|
|
skippedRows := sb.b.RowsCount() - (j - i)
|
|
|
|
metricRowsSkipped.Add(skippedRows)
|
|
|
|
|
|
|
|
// Copy the remaining values.
|
|
|
|
if i == j {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
values := sb.b.Values()
|
|
|
|
sb.Timestamps = append(sb.Timestamps, timestamps[i:j]...)
|
|
|
|
sb.Values = decimal.AppendDecimalToFloat(sb.Values, values[i:j], sb.b.Scale())
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
type sortBlocksHeap []*sortBlock
|
|
|
|
|
|
|
|
func (sbh sortBlocksHeap) Len() int {
|
|
|
|
return len(sbh)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (sbh sortBlocksHeap) Less(i, j int) bool {
|
|
|
|
a := sbh[i]
|
|
|
|
b := sbh[j]
|
|
|
|
return a.Timestamps[a.NextIdx] < b.Timestamps[b.NextIdx]
|
|
|
|
}
|
|
|
|
|
|
|
|
func (sbh sortBlocksHeap) Swap(i, j int) {
|
|
|
|
sbh[i], sbh[j] = sbh[j], sbh[i]
|
|
|
|
}
|
|
|
|
|
|
|
|
func (sbh *sortBlocksHeap) Push(x interface{}) {
|
|
|
|
*sbh = append(*sbh, x.(*sortBlock))
|
|
|
|
}
|
|
|
|
|
|
|
|
func (sbh *sortBlocksHeap) Pop() interface{} {
|
|
|
|
a := *sbh
|
|
|
|
v := a[len(a)-1]
|
|
|
|
*sbh = a[:len(a)-1]
|
|
|
|
return v
|
|
|
|
}
|
|
|
|
|
|
|
|
// DeleteSeries deletes time series matching the given tagFilterss.
|
|
|
|
func DeleteSeries(sq *storage.SearchQuery) (int, error) {
|
|
|
|
tfss, err := setupTfss(sq.TagFilterss)
|
|
|
|
if err != nil {
|
|
|
|
return 0, err
|
|
|
|
}
|
|
|
|
return vmstorage.DeleteMetrics(tfss)
|
|
|
|
}
|
|
|
|
|
|
|
|
// GetLabels returns labels until the given deadline.
|
|
|
|
func GetLabels(deadline Deadline) ([]string, error) {
|
|
|
|
labels, err := vmstorage.SearchTagKeys(*maxTagKeysPerSearch)
|
|
|
|
if err != nil {
|
|
|
|
return nil, fmt.Errorf("error during labels search: %s", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Substitute "" with "__name__"
|
|
|
|
for i := range labels {
|
|
|
|
if labels[i] == "" {
|
|
|
|
labels[i] = "__name__"
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Sort labels like Prometheus does
|
|
|
|
sort.Strings(labels)
|
|
|
|
|
|
|
|
return labels, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// GetLabelValues returns label values for the given labelName
|
|
|
|
// until the given deadline.
|
|
|
|
func GetLabelValues(labelName string, deadline Deadline) ([]string, error) {
|
|
|
|
if labelName == "__name__" {
|
|
|
|
labelName = ""
|
|
|
|
}
|
|
|
|
|
|
|
|
// Search for tag values
|
|
|
|
labelValues, err := vmstorage.SearchTagValues([]byte(labelName), *maxTagValuesPerSearch)
|
|
|
|
if err != nil {
|
|
|
|
return nil, fmt.Errorf("error during label values search for labelName=%q: %s", labelName, err)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Sort labelValues like Prometheus does
|
|
|
|
sort.Strings(labelValues)
|
|
|
|
|
|
|
|
return labelValues, nil
|
|
|
|
}
|
|
|
|
|
2019-06-10 17:55:20 +02:00
|
|
|
// GetLabelEntries returns all the label entries until the given deadline.
|
|
|
|
func GetLabelEntries(deadline Deadline) ([]storage.TagEntry, error) {
|
|
|
|
labelEntries, err := vmstorage.SearchTagEntries(*maxTagKeysPerSearch, *maxTagValuesPerSearch)
|
|
|
|
if err != nil {
|
|
|
|
return nil, fmt.Errorf("error during label entries request: %s", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Substitute "" with "__name__"
|
|
|
|
for i := range labelEntries {
|
|
|
|
e := &labelEntries[i]
|
|
|
|
if e.Key == "" {
|
|
|
|
e.Key = "__name__"
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Sort labelEntries by the number of label values in each entry.
|
|
|
|
sort.Slice(labelEntries, func(i, j int) bool {
|
|
|
|
a, b := labelEntries[i].Values, labelEntries[j].Values
|
2019-12-14 23:07:09 +01:00
|
|
|
if len(a) != len(b) {
|
|
|
|
return len(a) > len(b)
|
2019-06-10 17:55:20 +02:00
|
|
|
}
|
2019-12-14 23:07:09 +01:00
|
|
|
return labelEntries[i].Key > labelEntries[j].Key
|
2019-06-10 17:55:20 +02:00
|
|
|
})
|
|
|
|
|
|
|
|
return labelEntries, nil
|
|
|
|
}
|
|
|
|
|
2020-04-22 18:57:36 +02:00
|
|
|
// GetTSDBStatusForDate returns tsdb status according to https://prometheus.io/docs/prometheus/latest/querying/api/#tsdb-stats
|
|
|
|
func GetTSDBStatusForDate(deadline Deadline, date uint64, topN int) (*storage.TSDBStatus, error) {
|
|
|
|
status, err := vmstorage.GetTSDBStatusForDate(date, topN)
|
|
|
|
if err != nil {
|
|
|
|
return nil, fmt.Errorf("error during tsdb status request: %s", err)
|
|
|
|
}
|
|
|
|
return status, nil
|
|
|
|
}
|
|
|
|
|
2019-05-22 23:16:55 +02:00
|
|
|
// GetSeriesCount returns the number of unique series.
|
|
|
|
func GetSeriesCount(deadline Deadline) (uint64, error) {
|
|
|
|
n, err := vmstorage.GetSeriesCount()
|
|
|
|
if err != nil {
|
|
|
|
return 0, fmt.Errorf("error during series count request: %s", err)
|
|
|
|
}
|
|
|
|
return n, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func getStorageSearch() *storage.Search {
|
|
|
|
v := ssPool.Get()
|
|
|
|
if v == nil {
|
|
|
|
return &storage.Search{}
|
|
|
|
}
|
|
|
|
return v.(*storage.Search)
|
|
|
|
}
|
|
|
|
|
|
|
|
func putStorageSearch(sr *storage.Search) {
|
|
|
|
sr.MustClose()
|
|
|
|
ssPool.Put(sr)
|
|
|
|
}
|
|
|
|
|
|
|
|
var ssPool sync.Pool
|
|
|
|
|
|
|
|
// ProcessSearchQuery performs sq on storage nodes until the given deadline.
|
2020-04-27 07:13:41 +02:00
|
|
|
//
|
|
|
|
// Results.RunParallel or Results.Cancel must be called on the returned Results.
|
2019-08-04 21:15:33 +02:00
|
|
|
func ProcessSearchQuery(sq *storage.SearchQuery, fetchData bool, deadline Deadline) (*Results, error) {
|
2019-05-22 23:16:55 +02:00
|
|
|
// Setup search.
|
|
|
|
tfss, err := setupTfss(sq.TagFilterss)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
tr := storage.TimeRange{
|
|
|
|
MinTimestamp: sq.MinTimestamp,
|
|
|
|
MaxTimestamp: sq.MaxTimestamp,
|
|
|
|
}
|
|
|
|
|
|
|
|
vmstorage.WG.Add(1)
|
|
|
|
defer vmstorage.WG.Done()
|
|
|
|
|
|
|
|
sr := getStorageSearch()
|
2020-04-27 07:13:41 +02:00
|
|
|
sr.Init(vmstorage.Storage, tfss, tr, *maxMetricsPerSearch)
|
2019-05-22 23:16:55 +02:00
|
|
|
|
2020-04-27 07:13:41 +02:00
|
|
|
m := make(map[string][]storage.BlockRef)
|
2020-04-26 15:25:35 +02:00
|
|
|
var orderedMetricNames []string
|
2019-07-28 11:12:30 +02:00
|
|
|
blocksRead := 0
|
2019-05-22 23:16:55 +02:00
|
|
|
for sr.NextMetricBlock() {
|
2019-07-28 11:12:30 +02:00
|
|
|
blocksRead++
|
2019-05-22 23:16:55 +02:00
|
|
|
if time.Until(deadline.Deadline) < 0 {
|
2020-01-22 14:50:34 +01:00
|
|
|
return nil, fmt.Errorf("timeout exceeded while fetching data block #%d from storage: %s", blocksRead, deadline.String())
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
2020-04-27 07:13:41 +02:00
|
|
|
metricName := sr.MetricBlockRef.MetricName
|
|
|
|
brs := m[string(metricName)]
|
|
|
|
if len(brs) == 0 {
|
2020-04-26 15:25:35 +02:00
|
|
|
orderedMetricNames = append(orderedMetricNames, string(metricName))
|
|
|
|
}
|
2020-04-27 07:13:41 +02:00
|
|
|
m[string(metricName)] = append(brs, *sr.MetricBlockRef.BlockRef)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
if err := sr.Error(); err != nil {
|
2019-07-28 11:12:30 +02:00
|
|
|
return nil, fmt.Errorf("search error after reading %d data blocks: %s", blocksRead, err)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
var rss Results
|
|
|
|
rss.tr = tr
|
2019-08-04 21:15:33 +02:00
|
|
|
rss.fetchData = fetchData
|
2019-05-22 23:16:55 +02:00
|
|
|
rss.deadline = deadline
|
2020-04-26 15:25:35 +02:00
|
|
|
pts := make([]packedTimeseries, len(orderedMetricNames))
|
|
|
|
for i, metricName := range orderedMetricNames {
|
|
|
|
pts[i] = packedTimeseries{
|
|
|
|
metricName: metricName,
|
2020-04-27 07:13:41 +02:00
|
|
|
brs: m[metricName],
|
2020-04-26 15:25:35 +02:00
|
|
|
}
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
2020-04-26 15:25:35 +02:00
|
|
|
rss.packedTimeseries = pts
|
2020-04-27 07:13:41 +02:00
|
|
|
rss.sr = sr
|
2019-05-22 23:16:55 +02:00
|
|
|
return &rss, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func setupTfss(tagFilterss [][]storage.TagFilter) ([]*storage.TagFilters, error) {
|
|
|
|
tfss := make([]*storage.TagFilters, 0, len(tagFilterss))
|
|
|
|
for _, tagFilters := range tagFilterss {
|
|
|
|
tfs := storage.NewTagFilters()
|
|
|
|
for i := range tagFilters {
|
|
|
|
tf := &tagFilters[i]
|
|
|
|
if err := tfs.Add(tf.Key, tf.Value, tf.IsNegative, tf.IsRegexp); err != nil {
|
|
|
|
return nil, fmt.Errorf("cannot parse tag filter %s: %s", tf, err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
tfss = append(tfss, tfs)
|
2020-03-30 17:34:51 +02:00
|
|
|
tfss = append(tfss, tfs.Finalize()...)
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
return tfss, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// Deadline contains deadline with the corresponding timeout for pretty error messages.
|
|
|
|
type Deadline struct {
|
|
|
|
Deadline time.Time
|
2020-01-22 14:50:34 +01:00
|
|
|
|
|
|
|
timeout time.Duration
|
|
|
|
flagHint string
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// NewDeadline returns deadline for the given timeout.
|
2020-01-22 14:50:34 +01:00
|
|
|
//
|
|
|
|
// flagHint must contain a hit for command-line flag, which could be used
|
|
|
|
// in order to increase timeout.
|
|
|
|
func NewDeadline(timeout time.Duration, flagHint string) Deadline {
|
2019-05-22 23:16:55 +02:00
|
|
|
return Deadline{
|
|
|
|
Deadline: time.Now().Add(timeout),
|
2020-01-22 14:50:34 +01:00
|
|
|
timeout: timeout,
|
|
|
|
flagHint: flagHint,
|
2019-05-22 23:16:55 +02:00
|
|
|
}
|
|
|
|
}
|
2020-01-22 14:50:34 +01:00
|
|
|
|
|
|
|
// String returns human-readable string representation for d.
|
|
|
|
func (d *Deadline) String() string {
|
|
|
|
return fmt.Sprintf("%.3f seconds; the timeout can be adjusted with `%s` command-line flag", d.timeout.Seconds(), d.flagHint)
|
|
|
|
}
|