mirror of
https://github.com/VictoriaMetrics/VictoriaMetrics.git
synced 2025-03-11 15:34:56 +00:00
lib/logstorage: properly search for the surrounding logs in stream_context
pipe
The set of log fields in the found logs may differ from the set of log fields present in the log stream. So compare only the log fields in the found logs when searching for the matching log entry in the log stream. While at it, return _stream field in the delimiter log entry, since this field is used by VictoriaLogs Web UI for grouping logs by log streams.
This commit is contained in:
parent
516c7b2ca0
commit
208a624d4d
4 changed files with 27 additions and 18 deletions
|
@ -19,6 +19,8 @@ according to [these docs](https://docs.victoriametrics.com/victorialogs/quicksta
|
||||||
|
|
||||||
## tip
|
## tip
|
||||||
|
|
||||||
|
* BUGFIX: return the proper surrounding logs for [`stream_context` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#stream_context-pipe) when additional [pipes](https://docs.victoriametrics.com/victorialogs/logsql/#pipes) are put after the `stream_context` pipe. This has been broken in [v0.26.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.26.0-victorialogs).
|
||||||
|
|
||||||
## [v0.26.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.26.0-victorialogs)
|
## [v0.26.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.26.0-victorialogs)
|
||||||
|
|
||||||
Released at 2024-07-01
|
Released at 2024-07-01
|
||||||
|
|
|
@ -364,6 +364,10 @@ func (wctx *pipeStreamContextWriteContext) writeStreamContextRows(streamID strin
|
||||||
Name: "_stream_id",
|
Name: "_stream_id",
|
||||||
Value: streamID,
|
Value: streamID,
|
||||||
},
|
},
|
||||||
|
{
|
||||||
|
Name: "_stream",
|
||||||
|
Value: getFieldValue(r.fields, "_stream"),
|
||||||
|
},
|
||||||
{
|
{
|
||||||
Name: "_msg",
|
Name: "_msg",
|
||||||
Value: "---",
|
Value: "---",
|
||||||
|
@ -402,20 +406,30 @@ func getStreamContextRowIdx(rows []streamContextRow, r *streamContextRow) int {
|
||||||
if n == len(rows) {
|
if n == len(rows) {
|
||||||
return -1
|
return -1
|
||||||
}
|
}
|
||||||
if rows[n].timestamp != r.timestamp {
|
|
||||||
return -1
|
equalFields := func(fields []Field) bool {
|
||||||
|
for _, f := range r.fields {
|
||||||
|
if f.Value != getFieldValue(fields, f.Name) {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return true
|
||||||
}
|
}
|
||||||
for rows[n].timestamp == r.timestamp && !equalFields(rows[n].fields, r.fields) {
|
|
||||||
|
for rows[n].timestamp == r.timestamp && !equalFields(rows[n].fields) {
|
||||||
n++
|
n++
|
||||||
if n >= len(rows) {
|
if n >= len(rows) {
|
||||||
return -1
|
return -1
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if rows[n].timestamp != r.timestamp {
|
||||||
|
return -1
|
||||||
|
}
|
||||||
return n
|
return n
|
||||||
}
|
}
|
||||||
|
|
||||||
func sortStreamContextRows(rows []streamContextRow) {
|
func sortStreamContextRows(rows []streamContextRow) {
|
||||||
sort.SliceStable(rows, func(i, j int) bool {
|
sort.Slice(rows, func(i, j int) bool {
|
||||||
return rows[i].timestamp < rows[j].timestamp
|
return rows[i].timestamp < rows[j].timestamp
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
|
@ -79,20 +79,13 @@ func (f *Field) marshalToLogfmt(dst []byte) []byte {
|
||||||
return dst
|
return dst
|
||||||
}
|
}
|
||||||
|
|
||||||
func equalFields(a, b []Field) bool {
|
func getFieldValue(fields []Field, name string) string {
|
||||||
if len(a) != len(b) {
|
for _, f := range fields {
|
||||||
return false
|
if f.Name == name {
|
||||||
}
|
return f.Value
|
||||||
for i, x := range a {
|
|
||||||
y := b[i]
|
|
||||||
if x.Name != y.Name {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
if x.Value != y.Value {
|
|
||||||
return false
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return true
|
return ""
|
||||||
}
|
}
|
||||||
|
|
||||||
func needLogfmtQuoting(s string) bool {
|
func needLogfmtQuoting(s string) bool {
|
||||||
|
|
|
@ -707,7 +707,7 @@ func TestStorageRunQuery(t *testing.T) {
|
||||||
| stream_context before 1000
|
| stream_context before 1000
|
||||||
| stats count() rows`, [][]Field{
|
| stats count() rows`, [][]Field{
|
||||||
{
|
{
|
||||||
{"rows", "858"},
|
{"rows", "825"},
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
|
@ -716,7 +716,7 @@ func TestStorageRunQuery(t *testing.T) {
|
||||||
| stream_context after 1000
|
| stream_context after 1000
|
||||||
| stats count() rows`, [][]Field{
|
| stats count() rows`, [][]Field{
|
||||||
{
|
{
|
||||||
{"rows", "462"},
|
{"rows", "495"},
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
|
|
Loading…
Reference in a new issue