VictoriaMetrics/lib/storage/search.go

526 lines
14 KiB
Go
Raw Normal View History

2019-05-22 21:16:55 +00:00
package storage
import (
"fmt"
"io"
"strings"
2019-05-22 21:16:55 +00:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
2019-05-22 21:16:55 +00:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/querytracer"
2019-05-22 21:16:55 +00:00
)
// BlockRef references a Block.
//
// BlockRef is valid only until the corresponding Search is valid,
// i.e. it becomes invalid after Search.MustClose is called.
type BlockRef struct {
p *part
bh blockHeader
}
func (br *BlockRef) reset() {
br.p = nil
br.bh = blockHeader{}
}
func (br *BlockRef) init(p *part, bh *blockHeader) {
br.p = p
br.bh = *bh
}
// MustReadBlock reads block from br to dst.
func (br *BlockRef) MustReadBlock(dst *Block) {
dst.Reset()
dst.bh = br.bh
dst.timestampsData = bytesutil.ResizeNoCopyMayOverallocate(dst.timestampsData, int(br.bh.TimestampsBlockSize))
br.p.timestampsFile.MustReadAt(dst.timestampsData, int64(br.bh.TimestampsBlockOffset))
dst.valuesData = bytesutil.ResizeNoCopyMayOverallocate(dst.valuesData, int(br.bh.ValuesBlockSize))
br.p.valuesFile.MustReadAt(dst.valuesData, int64(br.bh.ValuesBlockOffset))
}
// MetricBlockRef contains reference to time series block for a single metric.
type MetricBlockRef struct {
// The metric name
MetricName []byte
// The block reference. Call BlockRef.MustReadBlock in order to obtain the block.
BlockRef *BlockRef
}
2019-05-22 21:16:55 +00:00
// MetricBlock is a time series block for a single metric.
type MetricBlock struct {
// MetricName is metric name for the given Block.
2019-05-22 21:16:55 +00:00
MetricName []byte
// Block is a block for the given MetricName
Block Block
2019-05-22 21:16:55 +00:00
}
// Marshal marshals MetricBlock to dst
func (mb *MetricBlock) Marshal(dst []byte) []byte {
dst = encoding.MarshalBytes(dst, mb.MetricName)
return MarshalBlock(dst, &mb.Block)
2019-05-22 21:16:55 +00:00
}
// CopyFrom copies src to mb.
func (mb *MetricBlock) CopyFrom(src *MetricBlock) {
mb.MetricName = append(mb.MetricName[:0], src.MetricName...)
mb.Block.CopyFrom(&src.Block)
}
2019-05-22 21:16:55 +00:00
// MarshalBlock marshals b to dst.
//
// b.MarshalData must be called on b before calling MarshalBlock.
func MarshalBlock(dst []byte, b *Block) []byte {
dst = b.bh.Marshal(dst)
dst = encoding.MarshalBytes(dst, b.timestampsData)
dst = encoding.MarshalBytes(dst, b.valuesData)
return dst
}
// Unmarshal unmarshals MetricBlock from src
func (mb *MetricBlock) Unmarshal(src []byte) ([]byte, error) {
mb.Block.Reset()
2019-05-22 21:16:55 +00:00
tail, mn, err := encoding.UnmarshalBytes(src)
if err != nil {
return tail, fmt.Errorf("cannot unmarshal MetricName: %w", err)
2019-05-22 21:16:55 +00:00
}
mb.MetricName = append(mb.MetricName[:0], mn...)
src = tail
return UnmarshalBlock(&mb.Block, src)
2019-05-22 21:16:55 +00:00
}
// UnmarshalBlock unmarshal Block from src to dst.
//
// dst.UnmarshalData isn't called on the block.
func UnmarshalBlock(dst *Block, src []byte) ([]byte, error) {
tail, err := dst.bh.Unmarshal(src)
if err != nil {
return tail, fmt.Errorf("cannot unmarshal blockHeader: %w", err)
2019-05-22 21:16:55 +00:00
}
src = tail
tail, tds, err := encoding.UnmarshalBytes(src)
if err != nil {
return tail, fmt.Errorf("cannot unmarshal timestampsData: %w", err)
2019-05-22 21:16:55 +00:00
}
dst.timestampsData = append(dst.timestampsData[:0], tds...)
src = tail
tail, vd, err := encoding.UnmarshalBytes(src)
if err != nil {
return tail, fmt.Errorf("cannot unmarshal valuesData: %w", err)
2019-05-22 21:16:55 +00:00
}
dst.valuesData = append(dst.valuesData[:0], vd...)
src = tail
return src, nil
}
// Search is a search for time series.
type Search struct {
// MetricBlockRef is updated with each Search.NextMetricBlock call.
MetricBlockRef MetricBlockRef
2019-05-22 21:16:55 +00:00
// idb is used for MetricName lookup for the found data blocks.
idb *indexDB
2019-05-22 21:16:55 +00:00
// retentionDeadline is used for filtering out blocks outside the configured retention.
retentionDeadline int64
2019-05-22 21:16:55 +00:00
ts tableSearch
2023-02-13 12:27:13 +00:00
// tr contains time range used in the search.
tr TimeRange
// tfss contains tag filters used in the search.
tfss []*TagFilters
// deadline in unix timestamp seconds for the current search.
deadline uint64
2019-05-22 21:16:55 +00:00
err error
needClosing bool
loops int
prevMetricID uint64
2019-05-22 21:16:55 +00:00
}
func (s *Search) reset() {
s.MetricBlockRef.MetricName = s.MetricBlockRef.MetricName[:0]
s.MetricBlockRef.BlockRef = nil
2019-05-22 21:16:55 +00:00
s.idb = nil
s.retentionDeadline = 0
2019-05-22 21:16:55 +00:00
s.ts.reset()
s.tr = TimeRange{}
s.tfss = nil
s.deadline = 0
2019-05-22 21:16:55 +00:00
s.err = nil
s.needClosing = false
s.loops = 0
s.prevMetricID = 0
2019-05-22 21:16:55 +00:00
}
// Init initializes s from the given storage, tfss and tr.
//
// MustClose must be called when the search is done.
//
// Init returns the upper bound on the number of found time series.
func (s *Search) Init(qt *querytracer.Tracer, storage *Storage, tfss []*TagFilters, tr TimeRange, maxMetrics int, deadline uint64) int {
qt = qt.NewChild("init series search: filters=%s, timeRange=%s", tfss, &tr)
defer qt.Done()
2019-05-22 21:16:55 +00:00
if s.needClosing {
logger.Panicf("BUG: missing MustClose call before the next call to Init")
}
retentionDeadline := int64(fasttime.UnixTimestamp()*1e3) - storage.retentionMsecs
2019-05-22 21:16:55 +00:00
s.reset()
s.idb = storage.idb()
s.retentionDeadline = retentionDeadline
s.tr = tr
s.tfss = tfss
s.deadline = deadline
2019-05-22 21:16:55 +00:00
s.needClosing = true
var tsids []TSID
metricIDs, err := s.idb.searchMetricIDs(qt, tfss, tr, maxMetrics, deadline)
if err == nil && len(metricIDs) > 0 && len(tfss) > 0 {
accountID := tfss[0].accountID
projectID := tfss[0].projectID
tsids, err = s.idb.getTSIDsFromMetricIDs(qt, accountID, projectID, metricIDs, deadline)
if err == nil {
err = storage.prefetchMetricNames(qt, accountID, projectID, metricIDs, deadline)
}
}
// It is ok to call Init on non-nil err.
2019-05-22 21:16:55 +00:00
// Init must be called before returning because it will fail
2023-02-13 12:27:13 +00:00
// on Search.MustClose otherwise.
s.ts.Init(storage.tb, tsids, tr)
qt.Printf("search for parts with data for %d series", len(tsids))
2019-05-22 21:16:55 +00:00
if err != nil {
s.err = err
return 0
2019-05-22 21:16:55 +00:00
}
return len(tsids)
2019-05-22 21:16:55 +00:00
}
// MustClose closes the Search.
func (s *Search) MustClose() {
if !s.needClosing {
logger.Panicf("BUG: missing Init call before MustClose")
}
s.ts.MustClose()
s.reset()
}
// Error returns the last error from s.
func (s *Search) Error() error {
if s.err == io.EOF || s.err == nil {
2019-05-22 21:16:55 +00:00
return nil
}
return fmt.Errorf("error when searching for tagFilters=%s on the time range %s: %w", s.tfss, s.tr.String(), s.err)
2019-05-22 21:16:55 +00:00
}
// NextMetricBlock proceeds to the next MetricBlockRef.
2019-05-22 21:16:55 +00:00
func (s *Search) NextMetricBlock() bool {
if s.err != nil {
return false
}
for s.ts.NextBlock() {
if s.loops&paceLimiterSlowIterationsMask == 0 {
if err := checkSearchDeadlineAndPace(s.deadline); err != nil {
s.err = err
return false
}
}
s.loops++
tsid := &s.ts.BlockRef.bh.TSID
if tsid.MetricID != s.prevMetricID {
if s.ts.BlockRef.bh.MaxTimestamp < s.retentionDeadline {
// Skip the block, since it contains only data outside the configured retention.
continue
}
var err error
s.MetricBlockRef.MetricName, err = s.idb.searchMetricNameWithCache(s.MetricBlockRef.MetricName[:0], tsid.MetricID, tsid.AccountID, tsid.ProjectID)
if err != nil {
if err == io.EOF {
// Skip missing metricName for tsid.MetricID.
// It should be automatically fixed. See indexDB.searchMetricNameWithCache for details.
continue
}
s.err = err
return false
2019-05-22 21:16:55 +00:00
}
s.prevMetricID = tsid.MetricID
2019-05-22 21:16:55 +00:00
}
s.MetricBlockRef.BlockRef = s.ts.BlockRef
2019-05-22 21:16:55 +00:00
return true
}
if err := s.ts.Error(); err != nil {
s.err = err
return false
}
s.err = io.EOF
return false
}
// SearchQuery is used for sending search queries from vmselect to vmstorage.
type SearchQuery struct {
AccountID uint32
ProjectID uint32
// The time range for searching time series
2019-05-22 21:16:55 +00:00
MinTimestamp int64
MaxTimestamp int64
// Tag filters for the search query
TagFilterss [][]TagFilter
// The maximum number of time series the search query can return.
MaxMetrics int
2019-05-22 21:16:55 +00:00
}
// GetTimeRange returns time range for the given sq.
func (sq *SearchQuery) GetTimeRange() TimeRange {
return TimeRange{
MinTimestamp: sq.MinTimestamp,
MaxTimestamp: sq.MaxTimestamp,
}
}
// NewSearchQuery creates new search query for the given args.
func NewSearchQuery(accountID, projectID uint32, start, end int64, tagFilterss [][]TagFilter, maxMetrics int) *SearchQuery {
if start < 0 {
// This is needed for https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5553
start = 0
}
if maxMetrics <= 0 {
maxMetrics = 2e9
}
return &SearchQuery{
AccountID: accountID,
ProjectID: projectID,
MinTimestamp: start,
MaxTimestamp: end,
TagFilterss: tagFilterss,
MaxMetrics: maxMetrics,
}
}
2019-05-22 21:16:55 +00:00
// TagFilter represents a single tag filter from SearchQuery.
type TagFilter struct {
Key []byte
Value []byte
IsNegative bool
IsRegexp bool
}
// String returns string representation of tf.
func (tf *TagFilter) String() string {
op := tf.getOp()
value := bytesutil.LimitStringLen(string(tf.Value), 60)
if len(tf.Key) == 0 {
return fmt.Sprintf("__name__%s%q", op, value)
}
return fmt.Sprintf("%s%s%q", tf.Key, op, value)
}
func (tf *TagFilter) getOp() string {
if tf.IsNegative {
if tf.IsRegexp {
return "!~"
}
return "!="
}
if tf.IsRegexp {
return "=~"
}
return "="
2019-05-22 21:16:55 +00:00
}
// Marshal appends marshaled tf to dst and returns the result.
func (tf *TagFilter) Marshal(dst []byte) []byte {
dst = encoding.MarshalBytes(dst, tf.Key)
dst = encoding.MarshalBytes(dst, tf.Value)
x := 0
if tf.IsNegative {
x = 2
}
if tf.IsRegexp {
x |= 1
}
dst = append(dst, byte(x))
return dst
}
// Unmarshal unmarshals tf from src and returns the tail.
func (tf *TagFilter) Unmarshal(src []byte) ([]byte, error) {
tail, k, err := encoding.UnmarshalBytes(src)
if err != nil {
return tail, fmt.Errorf("cannot unmarshal Key: %w", err)
2019-05-22 21:16:55 +00:00
}
tf.Key = append(tf.Key[:0], k...)
src = tail
tail, v, err := encoding.UnmarshalBytes(src)
if err != nil {
return tail, fmt.Errorf("cannot unmarshal Value: %w", err)
2019-05-22 21:16:55 +00:00
}
tf.Value = append(tf.Value[:0], v...)
src = tail
if len(src) < 1 {
return src, fmt.Errorf("cannot unmarshal IsNegative+IsRegexp from empty src")
}
x := src[0]
switch x {
case 0:
tf.IsNegative = false
tf.IsRegexp = false
case 1:
tf.IsNegative = false
tf.IsRegexp = true
case 2:
tf.IsNegative = true
tf.IsRegexp = false
case 3:
tf.IsNegative = true
tf.IsRegexp = true
default:
return src, fmt.Errorf("unexpected value for IsNegative+IsRegexp: %d; must be in the range [0..3]", x)
}
src = src[1:]
return src, nil
}
// String returns string representation of the search query.
func (sq *SearchQuery) String() string {
a := make([]string, len(sq.TagFilterss))
for i, tfs := range sq.TagFilterss {
a[i] = tagFiltersToString(tfs)
}
start := TimestampToHumanReadableFormat(sq.MinTimestamp)
end := TimestampToHumanReadableFormat(sq.MaxTimestamp)
return fmt.Sprintf("accountID=%d, projectID=%d, filters=%s, timeRange=[%s..%s]", sq.AccountID, sq.ProjectID, a, start, end)
}
func tagFiltersToString(tfs []TagFilter) string {
a := make([]string, len(tfs))
for i, tf := range tfs {
a[i] = tf.String()
2019-05-22 21:16:55 +00:00
}
return "{" + strings.Join(a, ",") + "}"
2019-05-22 21:16:55 +00:00
}
// Marshal appends marshaled sq to dst and returns the result.
func (sq *SearchQuery) Marshal(dst []byte) []byte {
2019-05-22 21:23:23 +00:00
dst = encoding.MarshalUint32(dst, sq.AccountID)
dst = encoding.MarshalUint32(dst, sq.ProjectID)
2019-05-22 21:16:55 +00:00
dst = encoding.MarshalVarInt64(dst, sq.MinTimestamp)
dst = encoding.MarshalVarInt64(dst, sq.MaxTimestamp)
dst = encoding.MarshalVarUint64(dst, uint64(len(sq.TagFilterss)))
for _, tagFilters := range sq.TagFilterss {
dst = encoding.MarshalVarUint64(dst, uint64(len(tagFilters)))
for i := range tagFilters {
dst = tagFilters[i].Marshal(dst)
}
}
dst = encoding.MarshalUint32(dst, uint32(sq.MaxMetrics))
2019-05-22 21:16:55 +00:00
return dst
}
// Unmarshal unmarshals sq from src and returns the tail.
func (sq *SearchQuery) Unmarshal(src []byte) ([]byte, error) {
2019-05-22 21:23:23 +00:00
if len(src) < 4 {
return src, fmt.Errorf("cannot unmarshal AccountID: too short src len: %d; must be at least %d bytes", len(src), 4)
}
sq.AccountID = encoding.UnmarshalUint32(src)
src = src[4:]
if len(src) < 4 {
return src, fmt.Errorf("cannot unmarshal ProjectID: too short src len: %d; must be at least %d bytes", len(src), 4)
}
sq.ProjectID = encoding.UnmarshalUint32(src)
src = src[4:]
2019-05-22 21:16:55 +00:00
tail, minTs, err := encoding.UnmarshalVarInt64(src)
if err != nil {
return src, fmt.Errorf("cannot unmarshal MinTimestamp: %w", err)
2019-05-22 21:16:55 +00:00
}
sq.MinTimestamp = minTs
src = tail
tail, maxTs, err := encoding.UnmarshalVarInt64(src)
if err != nil {
return src, fmt.Errorf("cannot unmarshal MaxTimestamp: %w", err)
2019-05-22 21:16:55 +00:00
}
sq.MaxTimestamp = maxTs
src = tail
tail, tfssCount, err := encoding.UnmarshalVarUint64(src)
if err != nil {
return src, fmt.Errorf("cannot unmarshal the count of TagFilterss: %w", err)
2019-05-22 21:16:55 +00:00
}
if n := int(tfssCount) - cap(sq.TagFilterss); n > 0 {
sq.TagFilterss = append(sq.TagFilterss[:cap(sq.TagFilterss)], make([][]TagFilter, n)...)
}
sq.TagFilterss = sq.TagFilterss[:tfssCount]
src = tail
for i := 0; i < int(tfssCount); i++ {
tail, tfsCount, err := encoding.UnmarshalVarUint64(src)
if err != nil {
return src, fmt.Errorf("cannot unmarshal the count of TagFilters: %w", err)
2019-05-22 21:16:55 +00:00
}
src = tail
tagFilters := sq.TagFilterss[i]
if n := int(tfsCount) - cap(tagFilters); n > 0 {
tagFilters = append(tagFilters[:cap(tagFilters)], make([]TagFilter, n)...)
}
tagFilters = tagFilters[:tfsCount]
for j := 0; j < int(tfsCount); j++ {
tail, err := tagFilters[j].Unmarshal(src)
if err != nil {
return tail, fmt.Errorf("cannot unmarshal TagFilter #%d: %w", j, err)
2019-05-22 21:16:55 +00:00
}
src = tail
}
sq.TagFilterss[i] = tagFilters
}
if len(src) < 4 {
return src, fmt.Errorf("cannot unmarshal MaxMetrics: too short src len: %d; must be at least %d bytes", len(src), 4)
}
sq.MaxMetrics = int(encoding.UnmarshalUint32(src))
src = src[4:]
2019-05-22 21:16:55 +00:00
return src, nil
}
func checkSearchDeadlineAndPace(deadline uint64) error {
if fasttime.UnixTimestamp() > deadline {
return ErrDeadlineExceeded
}
return nil
}
const (
paceLimiterFastIterationsMask = 1<<16 - 1
paceLimiterMediumIterationsMask = 1<<14 - 1
paceLimiterSlowIterationsMask = 1<<12 - 1
)