github-mirrors/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2024-11-21 14:44:00 +00:00
Code Issues Projects Releases Packages Wiki Activity

app/vmselect: move Deadline from netstorage to searchutils

Browse source 
This removes dependency on netstorage from searchutils.
This commit is contained in:
Aliaksandr Valialkin 2020-09-11 13:18:57 +03:00
parent 2114179e19
commit f0005c3007
6 changed files with 65 additions and 58 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
app/vmselect/graphite/graphite.go
View file

@ -197,7 +197,7 @@ func MetricsIndexHandler(startTime time.Time, w http.ResponseWriter, r *http.Req
} }
// metricsFind searches for label values that match the given query. // metricsFind searches for label values that match the given query.
func metricsFind(tr storage.TimeRange, label, query string, delimiter byte, deadline netstorage.Deadline) ([]string, error) { func metricsFind(tr storage.TimeRange, label, query string, delimiter byte, deadline searchutils.Deadline) ([]string, error) {
expandTail := strings.HasSuffix(query, "*") expandTail := strings.HasSuffix(query, "*")
for strings.HasSuffix(query, "*") { for strings.HasSuffix(query, "*") {
query = query[:len(query)-1] query = query[:len(query)-1]

62
app/vmselect/netstorage/netstorage.go
View file

@ -8,8 +8,8 @@ import (
"runtime" "runtime"
"sort" "sort"
"sync" "sync"
"time"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/searchutils"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage" "github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil" "github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal" "github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
@ -53,7 +53,7 @@ func (r *Result) reset() {
type Results struct { type Results struct {
tr storage.TimeRange tr storage.TimeRange
fetchData bool fetchData bool
deadline Deadline deadline searchutils.Deadline
packedTimeseries []packedTimeseries packedTimeseries []packedTimeseries
sr *storage.Search sr *storage.Search
@ -458,11 +458,11 @@ func DeleteSeries(sq *storage.SearchQuery) (int, error) {
} }
// GetLabels returns labels until the given deadline. // GetLabels returns labels until the given deadline.
func GetLabels(deadline Deadline) ([]string, error) { func GetLabels(deadline searchutils.Deadline) ([]string, error) {
if deadline.Exceeded() { if deadline.Exceeded() {
return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String()) return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
} }
labels, err := vmstorage.SearchTagKeys(*maxTagKeysPerSearch, deadline.deadline) labels, err := vmstorage.SearchTagKeys(*maxTagKeysPerSearch, deadline.Deadline())
if err != nil { if err != nil {
return nil, fmt.Errorf("error during labels search: %w", err) return nil, fmt.Errorf("error during labels search: %w", err)
} }
@ -482,7 +482,7 @@ func GetLabels(deadline Deadline) ([]string, error) {
// GetLabelValues returns label values for the given labelName // GetLabelValues returns label values for the given labelName
// until the given deadline. // until the given deadline.
func GetLabelValues(labelName string, deadline Deadline) ([]string, error) { func GetLabelValues(labelName string, deadline searchutils.Deadline) ([]string, error) {
if deadline.Exceeded() { if deadline.Exceeded() {
return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String()) return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
} }
@ -491,7 +491,7 @@ func GetLabelValues(labelName string, deadline Deadline) ([]string, error) {
} }
// Search for tag values // Search for tag values
labelValues, err := vmstorage.SearchTagValues([]byte(labelName), *maxTagValuesPerSearch, deadline.deadline) labelValues, err := vmstorage.SearchTagValues([]byte(labelName), *maxTagValuesPerSearch, deadline.Deadline())
if err != nil { if err != nil {
return nil, fmt.Errorf("error during label values search for labelName=%q: %w", labelName, err) return nil, fmt.Errorf("error during label values search for labelName=%q: %w", labelName, err)
} }
@ -505,11 +505,11 @@ func GetLabelValues(labelName string, deadline Deadline) ([]string, error) {
// GetTagValueSuffixes returns tag value suffixes for the given tagKey and the given tagValuePrefix. // GetTagValueSuffixes returns tag value suffixes for the given tagKey and the given tagValuePrefix.
// //
// It can be used for implementing https://graphite-api.readthedocs.io/en/latest/api.html#metrics-find // It can be used for implementing https://graphite-api.readthedocs.io/en/latest/api.html#metrics-find
func GetTagValueSuffixes(tr storage.TimeRange, tagKey, tagValuePrefix string, delimiter byte, deadline Deadline) ([]string, error) { func GetTagValueSuffixes(tr storage.TimeRange, tagKey, tagValuePrefix string, delimiter byte, deadline searchutils.Deadline) ([]string, error) {
if deadline.Exceeded() { if deadline.Exceeded() {
return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String()) return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
} }
suffixes, err := vmstorage.SearchTagValueSuffixes(tr, []byte(tagKey), []byte(tagValuePrefix), delimiter, *maxTagValueSuffixesPerSearch, deadline.deadline) suffixes, err := vmstorage.SearchTagValueSuffixes(tr, []byte(tagKey), []byte(tagValuePrefix), delimiter, *maxTagValueSuffixesPerSearch, deadline.Deadline())
if err != nil { if err != nil {
return nil, fmt.Errorf("error during search for suffixes for tagKey=%q, tagValuePrefix=%q, delimiter=%c on time range %s: %w", return nil, fmt.Errorf("error during search for suffixes for tagKey=%q, tagValuePrefix=%q, delimiter=%c on time range %s: %w",
tagKey, tagValuePrefix, delimiter, tr.String(), err) tagKey, tagValuePrefix, delimiter, tr.String(), err)
@ -518,11 +518,11 @@ func GetTagValueSuffixes(tr storage.TimeRange, tagKey, tagValuePrefix string, de
} }
// GetLabelEntries returns all the label entries until the given deadline. // GetLabelEntries returns all the label entries until the given deadline.
func GetLabelEntries(deadline Deadline) ([]storage.TagEntry, error) { func GetLabelEntries(deadline searchutils.Deadline) ([]storage.TagEntry, error) {
if deadline.Exceeded() { if deadline.Exceeded() {
return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String()) return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
} }
labelEntries, err := vmstorage.SearchTagEntries(*maxTagKeysPerSearch, *maxTagValuesPerSearch, deadline.deadline) labelEntries, err := vmstorage.SearchTagEntries(*maxTagKeysPerSearch, *maxTagValuesPerSearch, deadline.Deadline())
if err != nil { if err != nil {
return nil, fmt.Errorf("error during label entries request: %w", err) return nil, fmt.Errorf("error during label entries request: %w", err)
} }
@ -548,11 +548,11 @@ func GetLabelEntries(deadline Deadline) ([]storage.TagEntry, error) {
} }
// GetTSDBStatusForDate returns tsdb status according to https://prometheus.io/docs/prometheus/latest/querying/api/#tsdb-stats // 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) { func GetTSDBStatusForDate(deadline searchutils.Deadline, date uint64, topN int) (*storage.TSDBStatus, error) {
if deadline.Exceeded() { if deadline.Exceeded() {
return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String()) return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
} }
status, err := vmstorage.GetTSDBStatusForDate(date, topN, deadline.deadline) status, err := vmstorage.GetTSDBStatusForDate(date, topN, deadline.Deadline())
if err != nil { if err != nil {
return nil, fmt.Errorf("error during tsdb status request: %w", err) return nil, fmt.Errorf("error during tsdb status request: %w", err)
} }
@ -560,11 +560,11 @@ func GetTSDBStatusForDate(deadline Deadline, date uint64, topN int) (*storage.TS
} }
// GetSeriesCount returns the number of unique series. // GetSeriesCount returns the number of unique series.
func GetSeriesCount(deadline Deadline) (uint64, error) { func GetSeriesCount(deadline searchutils.Deadline) (uint64, error) {
if deadline.Exceeded() { if deadline.Exceeded() {
return 0, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String()) return 0, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
} }
n, err := vmstorage.GetSeriesCount(deadline.deadline) n, err := vmstorage.GetSeriesCount(deadline.Deadline())
if err != nil { if err != nil {
return 0, fmt.Errorf("error during series count request: %w", err) return 0, fmt.Errorf("error during series count request: %w", err)
} }
@ -589,7 +589,7 @@ var ssPool sync.Pool
// ProcessSearchQuery performs sq on storage nodes until the given deadline. // ProcessSearchQuery performs sq on storage nodes until the given deadline.
// //
// Results.RunParallel or Results.Cancel must be called on the returned Results. // Results.RunParallel or Results.Cancel must be called on the returned Results.
func ProcessSearchQuery(sq *storage.SearchQuery, fetchData bool, deadline Deadline) (*Results, error) { func ProcessSearchQuery(sq *storage.SearchQuery, fetchData bool, deadline searchutils.Deadline) (*Results, error) {
if deadline.Exceeded() { if deadline.Exceeded() {
return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String()) return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
} }
@ -611,7 +611,7 @@ func ProcessSearchQuery(sq *storage.SearchQuery, fetchData bool, deadline Deadli
defer vmstorage.WG.Done() defer vmstorage.WG.Done()
sr := getStorageSearch() sr := getStorageSearch()
maxSeriesCount := sr.Init(vmstorage.Storage, tfss, tr, *maxMetricsPerSearch, deadline.deadline) maxSeriesCount := sr.Init(vmstorage.Storage, tfss, tr, *maxMetricsPerSearch, deadline.Deadline())
m := make(map[string][]storage.BlockRef, maxSeriesCount) m := make(map[string][]storage.BlockRef, maxSeriesCount)
orderedMetricNames := make([]string, 0, maxSeriesCount) orderedMetricNames := make([]string, 0, maxSeriesCount)
@ -672,33 +672,3 @@ func setupTfss(tagFilterss [][]storage.TagFilter) ([]*storage.TagFilters, error)
} }
return tfss, nil return tfss, nil
} }
// Deadline contains deadline with the corresponding timeout for pretty error messages.
type Deadline struct {
deadline uint64
timeout time.Duration
flagHint string
}
// NewDeadline returns deadline for the given timeout.
//
// flagHint must contain a hit for command-line flag, which could be used
// in order to increase timeout.
func NewDeadline(startTime time.Time, timeout time.Duration, flagHint string) Deadline {
return Deadline{
deadline: uint64(startTime.Add(timeout).Unix()),
timeout: timeout,
flagHint: flagHint,
}
}
// Exceeded returns true if deadline is exceeded.
func (d *Deadline) Exceeded() bool {
return fasttime.UnixTimestamp() > d.deadline
}
// 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)
}

6
app/vmselect/prometheus/prometheus.go
View file

@ -150,7 +150,7 @@ func ExportHandler(startTime time.Time, w http.ResponseWriter, r *http.Request)
var exportDuration = metrics.NewSummary(`vm_request_duration_seconds{path="/api/v1/export"}`) var exportDuration = metrics.NewSummary(`vm_request_duration_seconds{path="/api/v1/export"}`)
func exportHandler(w http.ResponseWriter, matches []string, start, end int64, format string, maxRowsPerLine int, deadline netstorage.Deadline) error { func exportHandler(w http.ResponseWriter, matches []string, start, end int64, format string, maxRowsPerLine int, deadline searchutils.Deadline) error {
writeResponseFunc := WriteExportStdResponse writeResponseFunc := WriteExportStdResponse
writeLineFunc := func(rs *netstorage.Result, resultsCh chan<- *quicktemplate.ByteBuffer) { writeLineFunc := func(rs *netstorage.Result, resultsCh chan<- *quicktemplate.ByteBuffer) {
bb := quicktemplate.AcquireByteBuffer() bb := quicktemplate.AcquireByteBuffer()
@ -322,7 +322,7 @@ func LabelValuesHandler(startTime time.Time, labelName string, w http.ResponseWr
return nil return nil
} }
func labelValuesWithMatches(labelName string, matches []string, start, end int64, deadline netstorage.Deadline) ([]string, error) { func labelValuesWithMatches(labelName string, matches []string, start, end int64, deadline searchutils.Deadline) ([]string, error) {
if len(matches) == 0 { if len(matches) == 0 {
logger.Panicf("BUG: matches must be non-empty") logger.Panicf("BUG: matches must be non-empty")
} }
@ -485,7 +485,7 @@ func LabelsHandler(startTime time.Time, w http.ResponseWriter, r *http.Request)
return nil return nil
} }
func labelsWithMatches(matches []string, start, end int64, deadline netstorage.Deadline) ([]string, error) { func labelsWithMatches(matches []string, start, end int64, deadline searchutils.Deadline) ([]string, error) {
if len(matches) == 0 { if len(matches) == 0 {
logger.Panicf("BUG: matches must be non-empty") logger.Panicf("BUG: matches must be non-empty")
} }

3
app/vmselect/promql/eval.go
View file

@ -8,6 +8,7 @@ import (
"sync" "sync"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/netstorage" "github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/netstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/searchutils"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil" "github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger" "github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/memory" "github.com/VictoriaMetrics/VictoriaMetrics/lib/memory"
@ -88,7 +89,7 @@ type EvalConfig struct {
// QuotedRemoteAddr contains quoted remote address. // QuotedRemoteAddr contains quoted remote address.
QuotedRemoteAddr string QuotedRemoteAddr string
Deadline netstorage.Deadline Deadline searchutils.Deadline
MayCache bool MayCache bool

5
app/vmselect/promql/exec_test.go
View file

@ -5,6 +5,7 @@ import (
"time" "time"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/netstorage" "github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/netstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/searchutils"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage" "github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
) )
@ -21,7 +22,7 @@ func TestExecSuccess(t *testing.T) {
Start: start, Start: start,
End: end, End: end,
Step: step, Step: step,
Deadline: netstorage.NewDeadline(time.Now(), time.Minute, ""), Deadline: searchutils.NewDeadline(time.Now(), time.Minute, ""),
} }
for i := 0; i < 5; i++ { for i := 0; i < 5; i++ {
result, err := Exec(ec, q, false) result, err := Exec(ec, q, false)
@ -5896,7 +5897,7 @@ func TestExecError(t *testing.T) {
Start: 1000, Start: 1000,
End: 2000, End: 2000,
Step: 100, Step: 100,
Deadline: netstorage.NewDeadline(time.Now(), time.Minute, ""), Deadline: searchutils.NewDeadline(time.Now(), time.Minute, ""),
} }
for i := 0; i < 4; i++ { for i := 0; i < 4; i++ {
rv, err := Exec(ec, q, false) rv, err := Exec(ec, q, false)

45
app/vmselect/searchutils/searchutils.go
View file

@ -9,7 +9,7 @@ import (
"strings" "strings"
"time" "time"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/netstorage" "github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
"github.com/VictoriaMetrics/metricsql" "github.com/VictoriaMetrics/metricsql"
) )
@ -97,18 +97,18 @@ func GetDuration(r *http.Request, argKey string, defaultValue int64) (int64, err
const maxDurationMsecs = 100 * 365 * 24 * 3600 * 1000 const maxDurationMsecs = 100 * 365 * 24 * 3600 * 1000
// GetDeadlineForQuery returns deadline for the given query r. // GetDeadlineForQuery returns deadline for the given query r.
func GetDeadlineForQuery(r *http.Request, startTime time.Time) netstorage.Deadline { func GetDeadlineForQuery(r *http.Request, startTime time.Time) Deadline {
dMax := maxQueryDuration.Milliseconds() dMax := maxQueryDuration.Milliseconds()
return getDeadlineWithMaxDuration(r, startTime, dMax, "-search.maxQueryDuration") return getDeadlineWithMaxDuration(r, startTime, dMax, "-search.maxQueryDuration")
} }
// GetDeadlineForExport returns deadline for the given request to /api/v1/export. // GetDeadlineForExport returns deadline for the given request to /api/v1/export.
func GetDeadlineForExport(r *http.Request, startTime time.Time) netstorage.Deadline { func GetDeadlineForExport(r *http.Request, startTime time.Time) Deadline {
dMax := maxExportDuration.Milliseconds() dMax := maxExportDuration.Milliseconds()
return getDeadlineWithMaxDuration(r, startTime, dMax, "-search.maxExportDuration") return getDeadlineWithMaxDuration(r, startTime, dMax, "-search.maxExportDuration")
} }
func getDeadlineWithMaxDuration(r *http.Request, startTime time.Time, dMax int64, flagHint string) netstorage.Deadline { func getDeadlineWithMaxDuration(r *http.Request, startTime time.Time, dMax int64, flagHint string) Deadline {
d, err := GetDuration(r, "timeout", 0) d, err := GetDuration(r, "timeout", 0)
if err != nil { if err != nil {
d = 0 d = 0
@ -117,7 +117,7 @@ func getDeadlineWithMaxDuration(r *http.Request, startTime time.Time, dMax int64
d = dMax d = dMax
} }
timeout := time.Duration(d) * time.Millisecond timeout := time.Duration(d) * time.Millisecond
return netstorage.NewDeadline(startTime, timeout, flagHint) return NewDeadline(startTime, timeout, flagHint)
} }
// GetBool returns boolean value from the given argKey query arg. // GetBool returns boolean value from the given argKey query arg.
@ -130,3 +130,38 @@ func GetBool(r *http.Request, argKey string) bool {
return true return true
} }
} }
// Deadline contains deadline with the corresponding timeout for pretty error messages.
type Deadline struct {
deadline uint64
timeout time.Duration
flagHint string
}
// NewDeadline returns deadline for the given timeout.
//
// flagHint must contain a hit for command-line flag, which could be used
// in order to increase timeout.
func NewDeadline(startTime time.Time, timeout time.Duration, flagHint string) Deadline {
return Deadline{
deadline: uint64(startTime.Add(timeout).Unix()),
timeout: timeout,
flagHint: flagHint,
}
}
// Exceeded returns true if deadline is exceeded.
func (d *Deadline) Exceeded() bool {
return fasttime.UnixTimestamp() > d.deadline
}
// Deadline returns deadline in unix timestamp seconds.
func (d *Deadline) Deadline() uint64 {
return d.deadline
}
// 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)
}