mirror of
https://github.com/VictoriaMetrics/VictoriaMetrics.git
synced 2024-12-18 22:52:11 +01:00
4d27933041
Use local timezone of the host server in this case. The timezone can be overridden with TZ environment variable if needed. While at it, allow using whitespace instead of T as a delimiter between data and time in the ingested _time field. For example, '2024-09-20 10:20:30' is now accepted during data ingestion. This is valid ISO8601 format, which is used by some log shippers, so it should be supported. This format is also known as SQL datetime format. Also assume local time zone when time without timezone information is passed to querying APIs. Previously such a time was parsed in UTC timezone. Add `Z` to the end of the time string if the old behaviour is preferred. Updates https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6721
104 lines
3.1 KiB
Go
104 lines
3.1 KiB
Go
package elasticsearch
|
|
|
|
import (
|
|
"bytes"
|
|
"compress/gzip"
|
|
"fmt"
|
|
"testing"
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
|
|
)
|
|
|
|
func TestReadBulkRequest_Failure(t *testing.T) {
|
|
f := func(data string) {
|
|
t.Helper()
|
|
|
|
tlp := &insertutils.TestLogMessageProcessor{}
|
|
r := bytes.NewBufferString(data)
|
|
rows, err := readBulkRequest(r, false, "_time", "_msg", tlp)
|
|
if err == nil {
|
|
t.Fatalf("expecting non-empty error")
|
|
}
|
|
if rows != 0 {
|
|
t.Fatalf("unexpected non-zero rows=%d", rows)
|
|
}
|
|
}
|
|
f("foobar")
|
|
f(`{}`)
|
|
f(`{"create":{}}`)
|
|
f(`{"creat":{}}
|
|
{}`)
|
|
f(`{"create":{}}
|
|
foobar`)
|
|
}
|
|
|
|
func TestReadBulkRequest_Success(t *testing.T) {
|
|
f := func(data, timeField, msgField string, rowsExpected int, timestampsExpected []int64, resultExpected string) {
|
|
t.Helper()
|
|
|
|
tlp := &insertutils.TestLogMessageProcessor{}
|
|
|
|
// Read the request without compression
|
|
r := bytes.NewBufferString(data)
|
|
rows, err := readBulkRequest(r, false, timeField, msgField, tlp)
|
|
if err != nil {
|
|
t.Fatalf("unexpected error: %s", err)
|
|
}
|
|
if rows != rowsExpected {
|
|
t.Fatalf("unexpected rows read; got %d; want %d", rows, rowsExpected)
|
|
}
|
|
if err := tlp.Verify(rowsExpected, timestampsExpected, resultExpected); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
// Read the request with compression
|
|
tlp = &insertutils.TestLogMessageProcessor{}
|
|
compressedData := compressData(data)
|
|
r = bytes.NewBufferString(compressedData)
|
|
rows, err = readBulkRequest(r, true, timeField, msgField, tlp)
|
|
if err != nil {
|
|
t.Fatalf("unexpected error: %s", err)
|
|
}
|
|
if rows != rowsExpected {
|
|
t.Fatalf("unexpected rows read; got %d; want %d", rows, rowsExpected)
|
|
}
|
|
if err := tlp.Verify(rowsExpected, timestampsExpected, resultExpected); err != nil {
|
|
t.Fatalf("verification failure after compression: %s", err)
|
|
}
|
|
}
|
|
|
|
// Verify an empty data
|
|
f("", "_time", "_msg", 0, nil, "")
|
|
f("\n", "_time", "_msg", 0, nil, "")
|
|
f("\n\n", "_time", "_msg", 0, nil, "")
|
|
|
|
// Verify non-empty data
|
|
data := `{"create":{"_index":"filebeat-8.8.0"}}
|
|
{"@timestamp":"2023-06-06T04:48:11.735Z","log":{"offset":71770,"file":{"path":"/var/log/auth.log"}},"message":"foobar"}
|
|
{"create":{"_index":"filebeat-8.8.0"}}
|
|
{"@timestamp":"2023-06-06 04:48:12.735+01:00","message":"baz"}
|
|
{"index":{"_index":"filebeat-8.8.0"}}
|
|
{"message":"xyz","@timestamp":"2023-06-06T04:48:13.735Z","x":"y"}
|
|
`
|
|
timeField := "@timestamp"
|
|
msgField := "message"
|
|
rowsExpected := 3
|
|
timestampsExpected := []int64{1686026891735000000, 1686023292735000000, 1686026893735000000}
|
|
resultExpected := `{"@timestamp":"","log.offset":"71770","log.file.path":"/var/log/auth.log","_msg":"foobar"}
|
|
{"@timestamp":"","_msg":"baz"}
|
|
{"_msg":"xyz","@timestamp":"","x":"y"}`
|
|
f(data, timeField, msgField, rowsExpected, timestampsExpected, resultExpected)
|
|
}
|
|
|
|
func compressData(s string) string {
|
|
var bb bytes.Buffer
|
|
zw := gzip.NewWriter(&bb)
|
|
if _, err := zw.Write([]byte(s)); err != nil {
|
|
panic(fmt.Errorf("unexpected error when compressing data: %w", err))
|
|
}
|
|
if err := zw.Close(); err != nil {
|
|
panic(fmt.Errorf("unexpected error when closing gzip writer: %w", err))
|
|
}
|
|
return bb.String()
|
|
}
|