github-mirrors/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2024-12-31 15:06:26 +00:00
Code Issues Projects Releases Packages Wiki Activity

wip

Browse source
This commit is contained in:
Aliaksandr Valialkin 2024-04-29 03:30:25 +02:00
parent 5d791d617b
commit fa8c611361
No known key found for this signature in database
GPG key ID: 52C003EE2BCDB9EB
2 changed files with 57 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
lib/logstorage/parser.go
View file

@ -206,7 +206,7 @@ func (q *Query) getResultColumnNames() []string {
switch t := p.(type) { switch t := p.(type) {
case *pipeFields: case *pipeFields:
return t.fields return t.fields
case *statsPipe: case *pipeStats:
return t.neededFields() return t.neededFields()
} }
} }

112
lib/logstorage/pipes.go
View file

@ -83,11 +83,11 @@ func parsePipes(lex *lexer) ([]pipe, error) {
} }
pipes = append(pipes, pf) pipes = append(pipes, pf)
case lex.isKeyword("stats"): case lex.isKeyword("stats"):
sp, err := parseStatsPipe(lex) ps, err := parseStatsPipe(lex)
if err != nil { if err != nil {
return nil, fmt.Errorf("cannot parse 'stats' pipe: %w", err) return nil, fmt.Errorf("cannot parse 'stats' pipe: %w", err)
} }
pipes = append(pipes, sp) pipes = append(pipes, ps)
case lex.isKeyword("head"): case lex.isKeyword("head"):
hp, err := parseHeadPipe(lex) hp, err := parseHeadPipe(lex)
if err != nil { if err != nil {
@ -95,11 +95,11 @@ func parsePipes(lex *lexer) ([]pipe, error) {
} }
pipes = append(pipes, hp) pipes = append(pipes, hp)
case lex.isKeyword("skip"): case lex.isKeyword("skip"):
sp, err := parseSkipPipe(lex) ps, err := parseSkipPipe(lex)
if err != nil { if err != nil {
return nil, fmt.Errorf("cannot parse 'skip' pipe: %w", err) return nil, fmt.Errorf("cannot parse 'skip' pipe: %w", err)
} }
pipes = append(pipes, sp) pipes = append(pipes, ps)
default: default:
return nil, fmt.Errorf("unexpected pipe %q", lex.token) return nil, fmt.Errorf("unexpected pipe %q", lex.token)
} }
@ -188,7 +188,7 @@ func parsePipeFields(lex *lexer) (*pipeFields, error) {
} }
} }
type statsPipe struct { type pipeStats struct {
byFields []string byFields []string
funcs []statsFunc funcs []statsFunc
} }
@ -228,17 +228,17 @@ type statsProcessor interface {
finalizeStats() (name, value string) finalizeStats() (name, value string)
} }
func (sp *statsPipe) String() string { func (ps *pipeStats) String() string {
s := "stats " s := "stats "
if len(sp.byFields) > 0 { if len(ps.byFields) > 0 {
s += "by (" + fieldNamesString(sp.byFields) + ") " s += "by (" + fieldNamesString(ps.byFields) + ") "
} }
if len(sp.funcs) == 0 { if len(ps.funcs) == 0 {
logger.Panicf("BUG: statsPipe must contain at least a single statsFunc") logger.Panicf("BUG: pipeStats must contain at least a single statsFunc")
} }
a := make([]string, len(sp.funcs)) a := make([]string, len(ps.funcs))
for i, f := range sp.funcs { for i, f := range ps.funcs {
a[i] = f.String() a[i] = f.String()
} }
s += strings.Join(a, ", ") s += strings.Join(a, ", ")
@ -247,20 +247,20 @@ func (sp *statsPipe) String() string {
const stateSizeBudgetChunk = 1 << 20 const stateSizeBudgetChunk = 1 << 20
func (sp *statsPipe) newPipeProcessor(workersCount int, stopCh <-chan struct{}, cancel func(), ppBase pipeProcessor) pipeProcessor { func (ps *pipeStats) newPipeProcessor(workersCount int, stopCh <-chan struct{}, cancel func(), ppBase pipeProcessor) pipeProcessor {
maxStateSize := int64(float64(memory.Allowed()) * 0.3) maxStateSize := int64(float64(memory.Allowed()) * 0.3)
shards := make([]statsPipeProcessorShard, workersCount) shards := make([]pipeStatsProcessorShard, workersCount)
for i := range shards { for i := range shards {
shard := &shards[i] shard := &shards[i]
shard.sp = sp shard.ps = ps
shard.m = make(map[string]*statsPipeGroup) shard.m = make(map[string]*pipeStatsGroup)
shard.stateSizeBudget = stateSizeBudgetChunk shard.stateSizeBudget = stateSizeBudgetChunk
maxStateSize -= stateSizeBudgetChunk maxStateSize -= stateSizeBudgetChunk
} }
spp := &statsPipeProcessor{ spp := &pipeStatsProcessor{
sp: sp, ps: ps,
stopCh: stopCh, stopCh: stopCh,
cancel: cancel, cancel: cancel,
ppBase: ppBase, ppBase: ppBase,
@ -274,28 +274,28 @@ func (sp *statsPipe) newPipeProcessor(workersCount int, stopCh <-chan struct{},
return spp return spp
} }
type statsPipeProcessor struct { type pipeStatsProcessor struct {
sp *statsPipe ps *pipeStats
stopCh <-chan struct{} stopCh <-chan struct{}
cancel func() cancel func()
ppBase pipeProcessor ppBase pipeProcessor
shards []statsPipeProcessorShard shards []pipeStatsProcessorShard
maxStateSize int64 maxStateSize int64
stateSizeBudget atomic.Int64 stateSizeBudget atomic.Int64
} }
type statsPipeProcessorShard struct { type pipeStatsProcessorShard struct {
statsPipeProcessorShardNopad pipeStatsProcessorShardNopad
// The padding prevents false sharing on widespread platforms with 128 mod (cache line size) = 0 . // The padding prevents false sharing on widespread platforms with 128 mod (cache line size) = 0 .
_ [128 - unsafe.Sizeof(statsPipeProcessorShardNopad{})%128]byte _ [128 - unsafe.Sizeof(pipeStatsProcessorShardNopad{})%128]byte
} }
type statsPipeProcessorShardNopad struct { type pipeStatsProcessorShardNopad struct {
sp *statsPipe ps *pipeStats
m map[string]*statsPipeGroup m map[string]*pipeStatsGroup
columnValues [][]string columnValues [][]string
keyBuf []byte keyBuf []byte
@ -303,16 +303,16 @@ type statsPipeProcessorShardNopad struct {
stateSizeBudget int stateSizeBudget int
} }
func (shard *statsPipeProcessorShard) getStatsProcessors(key []byte) []statsProcessor { func (shard *pipeStatsProcessorShard) getStatsProcessors(key []byte) []statsProcessor {
spg := shard.m[string(key)] spg := shard.m[string(key)]
if spg == nil { if spg == nil {
sfps := make([]statsProcessor, len(shard.sp.funcs)) sfps := make([]statsProcessor, len(shard.ps.funcs))
for i, f := range shard.sp.funcs { for i, f := range shard.ps.funcs {
sfp, stateSize := f.newStatsProcessor() sfp, stateSize := f.newStatsProcessor()
sfps[i] = sfp sfps[i] = sfp
shard.stateSizeBudget -= stateSize shard.stateSizeBudget -= stateSize
} }
spg = &statsPipeGroup{ spg = &pipeStatsGroup{
sfps: sfps, sfps: sfps,
} }
shard.m[string(key)] = spg shard.m[string(key)] = spg
@ -321,11 +321,11 @@ func (shard *statsPipeProcessorShard) getStatsProcessors(key []byte) []statsProc
return spg.sfps return spg.sfps
} }
type statsPipeGroup struct { type pipeStatsGroup struct {
sfps []statsProcessor sfps []statsProcessor
} }
func (spp *statsPipeProcessor) writeBlock(workerID uint, timestamps []int64, columns []BlockColumn) { func (spp *pipeStatsProcessor) writeBlock(workerID uint, timestamps []int64, columns []BlockColumn) {
shard := &spp.shards[workerID] shard := &spp.shards[workerID]
for shard.stateSizeBudget < 0 { for shard.stateSizeBudget < 0 {
@ -342,7 +342,7 @@ func (spp *statsPipeProcessor) writeBlock(workerID uint, timestamps []int64, col
shard.stateSizeBudget += stateSizeBudgetChunk shard.stateSizeBudget += stateSizeBudgetChunk
} }
byFields := spp.sp.byFields byFields := spp.ps.byFields
if len(byFields) == 0 { if len(byFields) == 0 {
// Fast path - pass all the rows to a single group with empty key. // Fast path - pass all the rows to a single group with empty key.
for _, sfp := range shard.getStatsProcessors(nil) { for _, sfp := range shard.getStatsProcessors(nil) {
@ -379,7 +379,7 @@ func (spp *statsPipeProcessor) writeBlock(workerID uint, timestamps []int64, col
} }
// Pre-calculate column values for byFields in order to speed up building group key in the loop below. // Pre-calculate column values for byFields in order to speed up building group key in the loop below.
shard.columnValues = appendBlockColumnValues(shard.columnValues[:0], columns, spp.sp.byFields, len(timestamps)) shard.columnValues = appendBlockColumnValues(shard.columnValues[:0], columns, byFields, len(timestamps))
columnValues := shard.columnValues columnValues := shard.columnValues
if areConstValues(columnValues) { if areConstValues(columnValues) {
@ -445,9 +445,9 @@ func isConstValue(values []string) bool {
return true return true
} }
func (spp *statsPipeProcessor) flush() error { func (spp *pipeStatsProcessor) flush() error {
if n := spp.stateSizeBudget.Load(); n <= 0 { if n := spp.stateSizeBudget.Load(); n <= 0 {
return fmt.Errorf("cannot calculate [%s], since it requires more than %dMB of memory", spp.sp.String(), spp.maxStateSize/(1<<20)) return fmt.Errorf("cannot calculate [%s], since it requires more than %dMB of memory", spp.ps.String(), spp.maxStateSize/(1<<20))
} }
// Merge states across shards // Merge states across shards
@ -477,7 +477,7 @@ func (spp *statsPipeProcessor) flush() error {
} }
// Write per-group states to ppBase // Write per-group states to ppBase
byFields := spp.sp.byFields byFields := spp.ps.byFields
if len(byFields) == 0 && len(m) == 0 { if len(byFields) == 0 && len(m) == 0 {
// Special case - zero matching rows. // Special case - zero matching rows.
_ = shards[0].getStatsProcessors(nil) _ = shards[0].getStatsProcessors(nil)
@ -533,7 +533,7 @@ func (spp *statsPipeProcessor) flush() error {
return nil return nil
} }
func (sp *statsPipe) neededFields() []string { func (ps *pipeStats) neededFields() []string {
var neededFields []string var neededFields []string
m := make(map[string]struct{}) m := make(map[string]struct{})
updateNeededFields := func(fields []string) { updateNeededFields := func(fields []string) {
@ -545,9 +545,9 @@ func (sp *statsPipe) neededFields() []string {
} }
} }
updateNeededFields(sp.byFields) updateNeededFields(ps.byFields)
for _, f := range sp.funcs { for _, f := range ps.funcs {
fields := f.neededFields() fields := f.neededFields()
updateNeededFields(fields) updateNeededFields(fields)
} }
@ -555,19 +555,19 @@ func (sp *statsPipe) neededFields() []string {
return neededFields return neededFields
} }
func parseStatsPipe(lex *lexer) (*statsPipe, error) { func parseStatsPipe(lex *lexer) (*pipeStats, error) {
if !lex.mustNextToken() { if !lex.mustNextToken() {
return nil, fmt.Errorf("missing stats config") return nil, fmt.Errorf("missing stats config")
} }
var sp statsPipe var ps pipeStats
if lex.isKeyword("by") { if lex.isKeyword("by") {
lex.nextToken() lex.nextToken()
fields, err := parseFieldNamesInParens(lex) fields, err := parseFieldNamesInParens(lex)
if err != nil { if err != nil {
return nil, fmt.Errorf("cannot parse 'by': %w", err) return nil, fmt.Errorf("cannot parse 'by': %w", err)
} }
sp.byFields = fields ps.byFields = fields
} }
var funcs []statsFunc var funcs []statsFunc
@ -578,8 +578,8 @@ func parseStatsPipe(lex *lexer) (*statsPipe, error) {
} }
funcs = append(funcs, sf) funcs = append(funcs, sf)
if lex.isKeyword("|", ")", "") { if lex.isKeyword("|", ")", "") {
sp.funcs = funcs ps.funcs = funcs
return &sp, nil return &ps, nil
} }
if !lex.isKeyword(",") { if !lex.isKeyword(",") {
return nil, fmt.Errorf("unexpected token %q; want ',', '|' or ')'", lex.token) return nil, fmt.Errorf("unexpected token %q; want ',', '|' or ')'", lex.token)
@ -707,19 +707,19 @@ type skipPipe struct {
n uint64 n uint64
} }
func (sp *skipPipe) String() string { func (ps *skipPipe) String() string {
return fmt.Sprintf("skip %d", sp.n) return fmt.Sprintf("skip %d", ps.n)
} }
func (sp *skipPipe) newPipeProcessor(workersCount int, _ <-chan struct{}, _ func(), ppBase pipeProcessor) pipeProcessor { func (ps *skipPipe) newPipeProcessor(workersCount int, _ <-chan struct{}, _ func(), ppBase pipeProcessor) pipeProcessor {
return &skipPipeProcessor{ return &skipPipeProcessor{
sp: sp, ps: ps,
ppBase: ppBase, ppBase: ppBase,
} }
} }
type skipPipeProcessor struct { type skipPipeProcessor struct {
sp *skipPipe ps *skipPipe
ppBase pipeProcessor ppBase pipeProcessor
rowsProcessed atomic.Uint64 rowsProcessed atomic.Uint64
@ -727,17 +727,17 @@ type skipPipeProcessor struct {
func (spp *skipPipeProcessor) writeBlock(workerID uint, timestamps []int64, columns []BlockColumn) { func (spp *skipPipeProcessor) writeBlock(workerID uint, timestamps []int64, columns []BlockColumn) {
rowsProcessed := spp.rowsProcessed.Add(uint64(len(timestamps))) rowsProcessed := spp.rowsProcessed.Add(uint64(len(timestamps)))
if rowsProcessed <= spp.sp.n { if rowsProcessed <= spp.ps.n {
return return
} }
rowsProcessed -= uint64(len(timestamps)) rowsProcessed -= uint64(len(timestamps))
if rowsProcessed >= spp.sp.n { if rowsProcessed >= spp.ps.n {
spp.ppBase.writeBlock(workerID, timestamps, columns) spp.ppBase.writeBlock(workerID, timestamps, columns)
return return
} }
rowsRemaining := spp.sp.n - rowsProcessed rowsRemaining := spp.ps.n - rowsProcessed
cs := make([]BlockColumn, len(columns)) cs := make([]BlockColumn, len(columns))
for i, c := range columns { for i, c := range columns {
cDst := &cs[i] cDst := &cs[i]
@ -761,10 +761,10 @@ func parseSkipPipe(lex *lexer) (*skipPipe, error) {
return nil, fmt.Errorf("cannot parse the number of rows to skip %q: %w", lex.token, err) return nil, fmt.Errorf("cannot parse the number of rows to skip %q: %w", lex.token, err)
} }
lex.nextToken() lex.nextToken()
sp := &skipPipe{ ps := &skipPipe{
n: n, n: n,
} }
return sp, nil return ps, nil
} }
func parseFieldNamesInParens(lex *lexer) ([]string, error) { func parseFieldNamesInParens(lex *lexer) ([]string, error) {