Loki: backend: use streaming JSON parser, try2 (#48752)
* converter: remove __name__ customization because Loki does not do that Loki does not handle __name__ in a special way. for Prometheus, the caller can implement the formatting by themselves * converter: change labels-formatting the labels.String() method does not handle strange values well * loki: backend: use streaming-json parser * more idiomatic code Co-authored-by: Todd Treece <360020+toddtreece@users.noreply.github.com> * simpler row-length check * simpler code * fixed converter/prom tests Co-authored-by: Todd Treece <360020+toddtreece@users.noreply.github.com>
This commit is contained in:
Gábor Farkas
co-authored by
Todd Treece
Todd Treece
parent
8ed3fb1f2d
commit
da74dba7c8
@@ -10,8 +10,9 @@ import (
|
||||
"net/url"
|
||||
"strconv"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/data"
|
||||
"github.com/grafana/grafana/pkg/infra/log"
|
||||
"github.com/grafana/loki/pkg/loghttp"
|
||||
"github.com/grafana/grafana/pkg/util/converter"
|
||||
jsoniter "github.com/json-iterator/go"
|
||||
)
|
||||
|
||||
@@ -137,7 +138,7 @@ func makeLokiError(body io.ReadCloser) error {
|
||||
return fmt.Errorf("%v", errorMessage)
|
||||
}
|
||||
|
||||
func (api *LokiAPI) DataQuery(ctx context.Context, query lokiQuery) (*loghttp.QueryResponse, error) {
|
||||
func (api *LokiAPI) DataQuery(ctx context.Context, query lokiQuery) (data.Frames, error) {
|
||||
req, err := makeDataRequest(ctx, api.url, query, api.oauthToken)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -158,13 +159,14 @@ func (api *LokiAPI) DataQuery(ctx context.Context, query lokiQuery) (*loghttp.Qu
|
||||
return nil, makeLokiError(resp.Body)
|
||||
}
|
||||
|
||||
var response loghttp.QueryResponse
|
||||
err = jsoniter.NewDecoder(resp.Body).Decode(&response)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
iter := jsoniter.Parse(jsoniter.ConfigDefault, resp.Body, 1024)
|
||||
res := converter.ReadPrometheusStyleResult(iter)
|
||||
|
||||
if res.Error != nil {
|
||||
return nil, res.Error
|
||||
}
|
||||
|
||||
return &response, nil
|
||||
return res.Frames, nil
|
||||
}
|
||||
|
||||
func makeRawRequest(ctx context.Context, lokiDsUrl string, resourceURL string, oauthToken string) (*http.Request, error) {
|
||||
|
||||
+94
-18
@@ -6,7 +6,6 @@ import (
|
||||
"hash/fnv"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/data"
|
||||
)
|
||||
@@ -57,6 +56,9 @@ func adjustMetricFrame(frame *data.Frame, query *lokiQuery) error {
|
||||
frame.Meta = &data.FrameMeta{}
|
||||
}
|
||||
|
||||
frame.Meta.Stats = parseStats(frame.Meta.Custom)
|
||||
frame.Meta.Custom = nil
|
||||
|
||||
if isMetricRange {
|
||||
frame.Meta.ExecutedQueryString = "Expr: " + query.Expr + "\n" + "Step: " + query.Step.String()
|
||||
} else {
|
||||
@@ -81,53 +83,55 @@ func adjustMetricFrame(frame *data.Frame, query *lokiQuery) error {
|
||||
func adjustLogsFrame(frame *data.Frame, query *lokiQuery) error {
|
||||
// we check if the fields are of correct type and length
|
||||
fields := frame.Fields
|
||||
if len(fields) != 3 {
|
||||
if len(fields) != 4 {
|
||||
return fmt.Errorf("invalid fields in logs frame")
|
||||
}
|
||||
|
||||
labelsField := fields[0]
|
||||
timeField := fields[1]
|
||||
lineField := fields[2]
|
||||
stringTimeField := fields[3]
|
||||
|
||||
if (timeField.Type() != data.FieldTypeTime) || (lineField.Type() != data.FieldTypeString) || (labelsField.Type() != data.FieldTypeJSON) {
|
||||
if (timeField.Type() != data.FieldTypeTime) || (lineField.Type() != data.FieldTypeString) || (labelsField.Type() != data.FieldTypeJSON) || (stringTimeField.Type() != data.FieldTypeString) {
|
||||
return fmt.Errorf("invalid fields in logs frame")
|
||||
}
|
||||
|
||||
if (timeField.Len() != lineField.Len()) || (timeField.Len() != labelsField.Len()) {
|
||||
if (timeField.Len() != lineField.Len()) || (timeField.Len() != labelsField.Len()) || (timeField.Len() != stringTimeField.Len()) {
|
||||
return fmt.Errorf("invalid fields in logs frame")
|
||||
}
|
||||
|
||||
// this returns an error when the length of fields do not match
|
||||
_, err := frame.RowLen()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
labelsField.Name = "labels"
|
||||
stringTimeField.Name = "tsNs"
|
||||
|
||||
if frame.Meta == nil {
|
||||
frame.Meta = &data.FrameMeta{}
|
||||
}
|
||||
|
||||
frame.Meta.Stats = parseStats(frame.Meta.Custom)
|
||||
frame.Meta.Custom = nil
|
||||
|
||||
frame.Meta.ExecutedQueryString = "Expr: " + query.Expr
|
||||
|
||||
// we need to send to the browser the nanosecond-precision timestamp too.
|
||||
// usually timestamps become javascript-date-objects in the browser automatically, which only
|
||||
// have millisecond-precision.
|
||||
// so we send a separate timestamp-as-string field too.
|
||||
stringTimeField := makeStringTimeField(timeField)
|
||||
// so we send a separate timestamp-as-string field too. it is provided by the
|
||||
// loki-json-parser-code
|
||||
|
||||
idField, err := makeIdField(stringTimeField, lineField, labelsField, frame.RefID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
frame.Fields = append(frame.Fields, stringTimeField, idField)
|
||||
frame.Fields = append(frame.Fields, idField)
|
||||
return nil
|
||||
}
|
||||
|
||||
func makeStringTimeField(timeField *data.Field) *data.Field {
|
||||
length := timeField.Len()
|
||||
stringTimestamps := make([]string, length)
|
||||
|
||||
for i := 0; i < length; i++ {
|
||||
nsNumber := timeField.At(i).(time.Time).UnixNano()
|
||||
stringTimestamps[i] = fmt.Sprintf("%d", nsNumber)
|
||||
}
|
||||
return data.NewField("tsNs", timeField.Labels.Copy(), stringTimestamps)
|
||||
}
|
||||
|
||||
func calculateCheckSum(time string, line string, labels []byte) (string, error) {
|
||||
input := []byte(line + "_")
|
||||
input = append(input, labels...)
|
||||
@@ -211,3 +215,75 @@ func getFrameLabels(frame *data.Frame) map[string]string {
|
||||
|
||||
return labels
|
||||
}
|
||||
|
||||
func parseStats(frameMetaCustom interface{}) []data.QueryStat {
|
||||
customMap, ok := frameMetaCustom.(map[string]interface{})
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
rawStats, ok := customMap["stats"].(map[string]interface{})
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
|
||||
var stats []data.QueryStat
|
||||
|
||||
summary, ok := rawStats["summary"].(map[string]interface{})
|
||||
if ok {
|
||||
stats = append(stats,
|
||||
makeStat("Summary: bytes processed per second", summary["bytesProcessedPerSecond"], "Bps"),
|
||||
makeStat("Summary: lines processed per second", summary["linesProcessedPerSecond"], ""),
|
||||
makeStat("Summary: total bytes processed", summary["totalBytesProcessed"], "decbytes"),
|
||||
makeStat("Summary: total lines processed", summary["totalLinesProcessed"], ""),
|
||||
makeStat("Summary: exec time", summary["execTime"], "s"))
|
||||
}
|
||||
|
||||
store, ok := rawStats["store"].(map[string]interface{})
|
||||
if ok {
|
||||
stats = append(stats,
|
||||
makeStat("Store: total chunks ref", store["totalChunksRef"], ""),
|
||||
makeStat("Store: total chunks downloaded", store["totalChunksDownloaded"], ""),
|
||||
makeStat("Store: chunks download time", store["chunksDownloadTime"], "s"),
|
||||
makeStat("Store: head chunk bytes", store["headChunkBytes"], "decbytes"),
|
||||
makeStat("Store: head chunk lines", store["headChunkLines"], ""),
|
||||
makeStat("Store: decompressed bytes", store["decompressedBytes"], "decbytes"),
|
||||
makeStat("Store: decompressed lines", store["decompressedLines"], ""),
|
||||
makeStat("Store: compressed bytes", store["compressedBytes"], "decbytes"),
|
||||
makeStat("Store: total duplicates", store["totalDuplicates"], ""))
|
||||
}
|
||||
|
||||
ingester, ok := rawStats["ingester"].(map[string]interface{})
|
||||
if ok {
|
||||
stats = append(stats,
|
||||
makeStat("Ingester: total reached", ingester["totalReached"], ""),
|
||||
makeStat("Ingester: total chunks matched", ingester["totalChunksMatched"], ""),
|
||||
makeStat("Ingester: total batches", ingester["totalBatches"], ""),
|
||||
makeStat("Ingester: total lines sent", ingester["totalLinesSent"], ""),
|
||||
makeStat("Ingester: head chunk bytes", ingester["headChunkBytes"], "decbytes"),
|
||||
makeStat("Ingester: head chunk lines", ingester["headChunkLines"], ""),
|
||||
makeStat("Ingester: decompressed bytes", ingester["decompressedBytes"], "decbytes"),
|
||||
makeStat("Ingester: decompressed lines", ingester["decompressedLines"], ""),
|
||||
makeStat("Ingester: compressed bytes", ingester["compressedBytes"], "decbytes"),
|
||||
makeStat("Ingester: total duplicates", ingester["totalDuplicates"], ""))
|
||||
}
|
||||
|
||||
return stats
|
||||
}
|
||||
|
||||
func makeStat(name string, interfaceValue interface{}, unit string) data.QueryStat {
|
||||
var value float64
|
||||
switch v := interfaceValue.(type) {
|
||||
case float64:
|
||||
value = v
|
||||
case int:
|
||||
value = float64(v)
|
||||
}
|
||||
|
||||
return data.QueryStat{
|
||||
FieldConfig: data.FieldConfig{
|
||||
DisplayName: name,
|
||||
Unit: unit,
|
||||
},
|
||||
Value: value,
|
||||
}
|
||||
}
|
||||
|
||||
+99
-15
@@ -2,6 +2,7 @@ package loki
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -39,20 +40,30 @@ func TestFormatName(t *testing.T) {
|
||||
|
||||
func TestAdjustFrame(t *testing.T) {
|
||||
t.Run("logs-frame metadata should be set correctly", func(t *testing.T) {
|
||||
time1 := time.Date(2022, 1, 2, 3, 4, 5, 6, time.UTC)
|
||||
time2 := time.Date(2022, 1, 2, 3, 5, 5, 6, time.UTC)
|
||||
time3 := time.Date(2022, 1, 2, 3, 5, 5, 6, time.UTC)
|
||||
time4 := time.Date(2022, 1, 2, 3, 6, 5, 6, time.UTC)
|
||||
|
||||
timeNs1 := strconv.FormatInt(time1.UnixNano(), 10)
|
||||
timeNs2 := strconv.FormatInt(time2.UnixNano(), 10)
|
||||
timeNs3 := strconv.FormatInt(time3.UnixNano(), 10)
|
||||
timeNs4 := strconv.FormatInt(time4.UnixNano(), 10)
|
||||
|
||||
frame := data.NewFrame("",
|
||||
data.NewField("labels", nil, []json.RawMessage{
|
||||
data.NewField("__labels", nil, []json.RawMessage{
|
||||
json.RawMessage(`{"level":"info"}`),
|
||||
json.RawMessage(`{"level":"error"}`),
|
||||
json.RawMessage(`{"level":"error"}`),
|
||||
json.RawMessage(`{"level":"info"}`),
|
||||
}),
|
||||
data.NewField("time", nil, []time.Time{
|
||||
time.Date(2022, 1, 2, 3, 4, 5, 6, time.UTC),
|
||||
time.Date(2022, 1, 2, 3, 5, 5, 6, time.UTC),
|
||||
time.Date(2022, 1, 2, 3, 5, 5, 6, time.UTC),
|
||||
time.Date(2022, 1, 2, 3, 6, 5, 6, time.UTC),
|
||||
data.NewField("Time", nil, []time.Time{
|
||||
time1, time2, time3, time4,
|
||||
}),
|
||||
data.NewField("Line", nil, []string{"line1", "line2", "line2", "line3"}),
|
||||
data.NewField("TS", nil, []string{
|
||||
timeNs1, timeNs2, timeNs3, timeNs4,
|
||||
}),
|
||||
data.NewField("line", nil, []string{"line1", "line2", "line2", "line3"}),
|
||||
)
|
||||
|
||||
frame.RefID = "A"
|
||||
@@ -68,14 +79,6 @@ func TestAdjustFrame(t *testing.T) {
|
||||
fields := frame.Fields
|
||||
|
||||
require.Equal(t, 5, len(fields))
|
||||
tsNsField := fields[3]
|
||||
require.Equal(t, "tsNs", tsNsField.Name)
|
||||
require.Equal(t, data.FieldTypeString, tsNsField.Type())
|
||||
require.Equal(t, 4, tsNsField.Len())
|
||||
require.Equal(t, "1641092645000000006", tsNsField.At(0))
|
||||
require.Equal(t, "1641092705000000006", tsNsField.At(1))
|
||||
require.Equal(t, "1641092705000000006", tsNsField.At(2))
|
||||
require.Equal(t, "1641092765000000006", tsNsField.At(3))
|
||||
|
||||
idField := fields[4]
|
||||
require.Equal(t, "id", idField.Name)
|
||||
@@ -136,4 +139,85 @@ func TestAdjustFrame(t *testing.T) {
|
||||
require.NotNil(t, timeFieldConfig)
|
||||
require.Equal(t, float64(42000), timeFieldConfig.Interval)
|
||||
})
|
||||
|
||||
t.Run("should parse response stats", func(t *testing.T) {
|
||||
stats := map[string]interface{}{
|
||||
"summary": map[string]interface{}{
|
||||
"bytesProcessedPerSecond": 1,
|
||||
"linesProcessedPerSecond": 2,
|
||||
"totalBytesProcessed": 3,
|
||||
"totalLinesProcessed": 4,
|
||||
"execTime": 5.5,
|
||||
},
|
||||
|
||||
"store": map[string]interface{}{
|
||||
"totalChunksRef": 6,
|
||||
"totalChunksDownloaded": 7,
|
||||
"chunksDownloadTime": 8.8,
|
||||
"headChunkBytes": 9,
|
||||
"headChunkLines": 10,
|
||||
"decompressedBytes": 11,
|
||||
"decompressedLines": 12,
|
||||
"compressedBytes": 13,
|
||||
"totalDuplicates": 14,
|
||||
},
|
||||
|
||||
"ingester": map[string]interface{}{
|
||||
"totalReached": 15,
|
||||
"totalChunksMatched": 16,
|
||||
"totalBatches": 17,
|
||||
"totalLinesSent": 18,
|
||||
"headChunkBytes": 19,
|
||||
"headChunkLines": 20,
|
||||
"decompressedBytes": 21,
|
||||
"decompressedLines": 22,
|
||||
"compressedBytes": 23,
|
||||
"totalDuplicates": 24,
|
||||
},
|
||||
}
|
||||
|
||||
meta := data.FrameMeta{
|
||||
Custom: map[string]interface{}{
|
||||
"stats": stats,
|
||||
},
|
||||
}
|
||||
|
||||
expected := []data.QueryStat{
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Summary: bytes processed per second", Unit: "Bps"}, Value: 1},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Summary: lines processed per second", Unit: ""}, Value: 2},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Summary: total bytes processed", Unit: "decbytes"}, Value: 3},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Summary: total lines processed", Unit: ""}, Value: 4},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Summary: exec time", Unit: "s"}, Value: 5.5},
|
||||
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: total chunks ref", Unit: ""}, Value: 6},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: total chunks downloaded", Unit: ""}, Value: 7},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: chunks download time", Unit: "s"}, Value: 8.8},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: head chunk bytes", Unit: "decbytes"}, Value: 9},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: head chunk lines", Unit: ""}, Value: 10},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: decompressed bytes", Unit: "decbytes"}, Value: 11},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: decompressed lines", Unit: ""}, Value: 12},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: compressed bytes", Unit: "decbytes"}, Value: 13},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: total duplicates", Unit: ""}, Value: 14},
|
||||
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: total reached", Unit: ""}, Value: 15},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: total chunks matched", Unit: ""}, Value: 16},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: total batches", Unit: ""}, Value: 17},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: total lines sent", Unit: ""}, Value: 18},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: head chunk bytes", Unit: "decbytes"}, Value: 19},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: head chunk lines", Unit: ""}, Value: 20},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: decompressed bytes", Unit: "decbytes"}, Value: 21},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: decompressed lines", Unit: ""}, Value: 22},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: compressed bytes", Unit: "decbytes"}, Value: 23},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: total duplicates", Unit: ""}, Value: 24},
|
||||
}
|
||||
|
||||
result := parseStats(meta.Custom)
|
||||
|
||||
// NOTE: i compare it item-by-item otherwise the test-fail-error-message is very hard to read
|
||||
require.Len(t, result, len(expected))
|
||||
|
||||
for i := 0; i < len(result); i++ {
|
||||
require.Equal(t, expected[i], result[i])
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
+11
-2
@@ -209,12 +209,21 @@ func queryData(ctx context.Context, req *backend.QueryDataRequest, dsInfo *datas
|
||||
|
||||
// we extracted this part of the functionality to make it easy to unit-test it
|
||||
func runQuery(ctx context.Context, api *LokiAPI, query *lokiQuery) (data.Frames, error) {
|
||||
value, err := api.DataQuery(ctx, *query)
|
||||
frames, err := api.DataQuery(ctx, *query)
|
||||
if err != nil {
|
||||
return data.Frames{}, err
|
||||
}
|
||||
|
||||
return parseResponse(value, query)
|
||||
for _, frame := range frames {
|
||||
if err = adjustFrame(frame, query); err != nil {
|
||||
return data.Frames{}, err
|
||||
}
|
||||
if err != nil {
|
||||
return data.Frames{}, err
|
||||
}
|
||||
}
|
||||
|
||||
return frames, nil
|
||||
}
|
||||
|
||||
func (s *Service) getDSInfo(pluginCtx backend.PluginContext) (*datasourceInfo, error) {
|
||||
|
||||
@@ -1,206 +0,0 @@
|
||||
package loki
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/data"
|
||||
"github.com/grafana/loki/pkg/loghttp"
|
||||
"github.com/grafana/loki/pkg/logqlmodel/stats"
|
||||
jsoniter "github.com/json-iterator/go"
|
||||
)
|
||||
|
||||
func parseResponse(value *loghttp.QueryResponse, query *lokiQuery) (data.Frames, error) {
|
||||
frames, err := lokiResponseToDataFrames(value, query)
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for _, frame := range frames {
|
||||
err = adjustFrame(frame, query)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
return frames, nil
|
||||
}
|
||||
|
||||
func lokiResponseToDataFrames(value *loghttp.QueryResponse, query *lokiQuery) (data.Frames, error) {
|
||||
stats := parseStats(value.Data.Statistics)
|
||||
switch res := value.Data.Result.(type) {
|
||||
case loghttp.Matrix:
|
||||
return lokiMatrixToDataFrames(res, query, stats), nil
|
||||
case loghttp.Vector:
|
||||
return lokiVectorToDataFrames(res, query, stats), nil
|
||||
case loghttp.Streams:
|
||||
return lokiStreamsToDataFrames(res, query, stats)
|
||||
default:
|
||||
return nil, fmt.Errorf("resultType %T not supported{", res)
|
||||
}
|
||||
}
|
||||
|
||||
func lokiMatrixToDataFrames(matrix loghttp.Matrix, query *lokiQuery, stats []data.QueryStat) data.Frames {
|
||||
frames := data.Frames{}
|
||||
|
||||
for i, v := range matrix {
|
||||
tags := make(map[string]string, len(v.Metric))
|
||||
timeVector := make([]time.Time, 0, len(v.Values))
|
||||
values := make([]float64, 0, len(v.Values))
|
||||
|
||||
for k, v := range v.Metric {
|
||||
tags[string(k)] = string(v)
|
||||
}
|
||||
|
||||
for _, k := range v.Values {
|
||||
timeVector = append(timeVector, k.Timestamp.Time().UTC())
|
||||
values = append(values, float64(k.Value))
|
||||
}
|
||||
|
||||
timeField := data.NewField(data.TimeSeriesTimeFieldName, nil, timeVector)
|
||||
valueField := data.NewField(data.TimeSeriesValueFieldName, tags, values)
|
||||
|
||||
frame := data.NewFrame("", timeField, valueField)
|
||||
frame.SetMeta(&data.FrameMeta{
|
||||
Type: data.FrameTypeTimeSeriesMany,
|
||||
})
|
||||
|
||||
// only add the stats to the first dataframe
|
||||
if i == 0 {
|
||||
frame.Meta.Stats = stats
|
||||
}
|
||||
|
||||
frames = append(frames, frame)
|
||||
}
|
||||
|
||||
return frames
|
||||
}
|
||||
|
||||
func lokiVectorToDataFrames(vector loghttp.Vector, query *lokiQuery, stats []data.QueryStat) data.Frames {
|
||||
frames := data.Frames{}
|
||||
|
||||
for i, v := range vector {
|
||||
tags := make(map[string]string, len(v.Metric))
|
||||
timeVector := []time.Time{v.Timestamp.Time().UTC()}
|
||||
values := []float64{float64(v.Value)}
|
||||
|
||||
for k, v := range v.Metric {
|
||||
tags[string(k)] = string(v)
|
||||
}
|
||||
timeField := data.NewField(data.TimeSeriesTimeFieldName, nil, timeVector)
|
||||
valueField := data.NewField(data.TimeSeriesValueFieldName, tags, values)
|
||||
|
||||
frame := data.NewFrame("", timeField, valueField)
|
||||
frame.SetMeta(&data.FrameMeta{
|
||||
Type: data.FrameTypeTimeSeriesMany,
|
||||
})
|
||||
|
||||
// only add the stats to the first dataframe
|
||||
if i == 0 {
|
||||
frame.Meta.Stats = stats
|
||||
}
|
||||
|
||||
frames = append(frames, frame)
|
||||
}
|
||||
|
||||
return frames
|
||||
}
|
||||
|
||||
// we serialize the labels as an ordered list of pairs
|
||||
func labelsToRawJson(labels data.Labels) (json.RawMessage, error) {
|
||||
// data.Labels when converted to JSON keep the fields sorted
|
||||
bytes, err := jsoniter.Marshal(labels)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return json.RawMessage(bytes), nil
|
||||
}
|
||||
|
||||
func lokiStreamsToDataFrames(streams loghttp.Streams, query *lokiQuery, stats []data.QueryStat) (data.Frames, error) {
|
||||
var timeVector []time.Time
|
||||
var values []string
|
||||
var labelsVector []json.RawMessage
|
||||
|
||||
for _, v := range streams {
|
||||
labelsJson, err := labelsToRawJson(v.Labels.Map())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for _, k := range v.Entries {
|
||||
timeVector = append(timeVector, k.Timestamp.UTC())
|
||||
values = append(values, k.Line)
|
||||
labelsVector = append(labelsVector, labelsJson)
|
||||
}
|
||||
}
|
||||
|
||||
timeField := data.NewField(data.TimeSeriesTimeFieldName, nil, timeVector)
|
||||
valueField := data.NewField("Line", nil, values)
|
||||
labelsField := data.NewField("labels", nil, labelsVector)
|
||||
|
||||
frame := data.NewFrame("", labelsField, timeField, valueField)
|
||||
frame.SetMeta(&data.FrameMeta{
|
||||
Stats: stats,
|
||||
})
|
||||
|
||||
return data.Frames{frame}, nil
|
||||
}
|
||||
|
||||
func parseStats(result stats.Result) []data.QueryStat {
|
||||
data := []data.QueryStat{
|
||||
makeStat("Summary: bytes processed per second", float64(result.Summary.BytesProcessedPerSecond), "Bps"),
|
||||
makeStat("Summary: lines processed per second", float64(result.Summary.LinesProcessedPerSecond), ""),
|
||||
makeStat("Summary: total bytes processed", float64(result.Summary.TotalBytesProcessed), "decbytes"),
|
||||
makeStat("Summary: total lines processed", float64(result.Summary.TotalLinesProcessed), ""),
|
||||
makeStat("Summary: exec time", result.Summary.ExecTime, "s"),
|
||||
makeStat("Store: total chunks ref", float64(result.Store.TotalChunksRef), ""),
|
||||
makeStat("Store: total chunks downloaded", float64(result.Store.TotalChunksDownloaded), ""),
|
||||
makeStat("Store: chunks download time", result.Store.ChunksDownloadTime, "s"),
|
||||
makeStat("Store: head chunk bytes", float64(result.Store.HeadChunkBytes), "decbytes"),
|
||||
makeStat("Store: head chunk lines", float64(result.Store.HeadChunkLines), ""),
|
||||
makeStat("Store: decompressed bytes", float64(result.Store.DecompressedBytes), "decbytes"),
|
||||
makeStat("Store: decompressed lines", float64(result.Store.DecompressedLines), ""),
|
||||
makeStat("Store: compressed bytes", float64(result.Store.CompressedBytes), "decbytes"),
|
||||
makeStat("Store: total duplicates", float64(result.Store.TotalDuplicates), ""),
|
||||
makeStat("Ingester: total reached", float64(result.Ingester.TotalReached), ""),
|
||||
makeStat("Ingester: total chunks matched", float64(result.Ingester.TotalChunksMatched), ""),
|
||||
makeStat("Ingester: total batches", float64(result.Ingester.TotalBatches), ""),
|
||||
makeStat("Ingester: total lines sent", float64(result.Ingester.TotalLinesSent), ""),
|
||||
makeStat("Ingester: head chunk bytes", float64(result.Ingester.HeadChunkBytes), "decbytes"),
|
||||
makeStat("Ingester: head chunk lines", float64(result.Ingester.HeadChunkLines), ""),
|
||||
makeStat("Ingester: decompressed bytes", float64(result.Ingester.DecompressedBytes), "decbytes"),
|
||||
makeStat("Ingester: decompressed lines", float64(result.Ingester.DecompressedLines), ""),
|
||||
makeStat("Ingester: compressed bytes", float64(result.Ingester.CompressedBytes), "decbytes"),
|
||||
makeStat("Ingester: total duplicates", float64(result.Ingester.TotalDuplicates), ""),
|
||||
}
|
||||
|
||||
// it is not possible to know whether the given statistics was missing, or
|
||||
// it's value was zero.
|
||||
// we do a heuristic here, if every stat-value is zero, we assume we got no stats-data
|
||||
allStatsZero := true
|
||||
for _, stat := range data {
|
||||
if stat.Value > 0 {
|
||||
allStatsZero = false
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if allStatsZero {
|
||||
return nil
|
||||
}
|
||||
|
||||
return data
|
||||
}
|
||||
|
||||
func makeStat(name string, value float64, unit string) data.QueryStat {
|
||||
return data.QueryStat{
|
||||
FieldConfig: data.FieldConfig{
|
||||
DisplayName: name,
|
||||
Unit: unit,
|
||||
},
|
||||
Value: value,
|
||||
}
|
||||
}
|
||||
@@ -1,183 +0,0 @@
|
||||
package loki
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/google/go-cmp/cmp"
|
||||
"github.com/grafana/grafana-plugin-sdk-go/data"
|
||||
"github.com/grafana/loki/pkg/loghttp"
|
||||
"github.com/grafana/loki/pkg/logqlmodel/stats"
|
||||
p "github.com/prometheus/common/model"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestParseResponse(t *testing.T) {
|
||||
t.Run("value is not of supported type", func(t *testing.T) {
|
||||
value := loghttp.QueryResponse{
|
||||
Data: loghttp.QueryResponseData{
|
||||
Result: loghttp.Scalar{},
|
||||
},
|
||||
}
|
||||
res, err := parseResponse(&value, nil)
|
||||
require.Equal(t, len(res), 0)
|
||||
require.Error(t, err)
|
||||
})
|
||||
|
||||
t.Run("response should be parsed normally", func(t *testing.T) {
|
||||
values := []p.SamplePair{
|
||||
{Value: 1, Timestamp: 1000},
|
||||
{Value: 2, Timestamp: 2000},
|
||||
{Value: 3, Timestamp: 3000},
|
||||
{Value: 4, Timestamp: 4000},
|
||||
{Value: 5, Timestamp: 5000},
|
||||
}
|
||||
value := loghttp.QueryResponse{
|
||||
Data: loghttp.QueryResponseData{
|
||||
Result: loghttp.Matrix{
|
||||
p.SampleStream{
|
||||
Metric: p.Metric{"app": "Application", "tag2": "tag2"},
|
||||
Values: values,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
query := &lokiQuery{
|
||||
Expr: "up(ALERTS)",
|
||||
QueryType: QueryTypeRange,
|
||||
LegendFormat: "legend {{app}}",
|
||||
Step: time.Second * 42,
|
||||
}
|
||||
frame, err := parseResponse(&value, query)
|
||||
require.NoError(t, err)
|
||||
|
||||
labels, err := data.LabelsFromString("app=Application, tag2=tag2")
|
||||
require.NoError(t, err)
|
||||
field1 := data.NewField("Time", nil, []time.Time{
|
||||
time.Date(1970, 1, 1, 0, 0, 1, 0, time.UTC),
|
||||
time.Date(1970, 1, 1, 0, 0, 2, 0, time.UTC),
|
||||
time.Date(1970, 1, 1, 0, 0, 3, 0, time.UTC),
|
||||
time.Date(1970, 1, 1, 0, 0, 4, 0, time.UTC),
|
||||
time.Date(1970, 1, 1, 0, 0, 5, 0, time.UTC),
|
||||
})
|
||||
field1.Config = &data.FieldConfig{Interval: float64(42000)}
|
||||
field2 := data.NewField("Value", labels, []float64{1, 2, 3, 4, 5})
|
||||
field2.SetConfig(&data.FieldConfig{DisplayNameFromDS: "legend Application"})
|
||||
testFrame := data.NewFrame("legend Application", field1, field2)
|
||||
testFrame.SetMeta(&data.FrameMeta{
|
||||
ExecutedQueryString: "Expr: up(ALERTS)\nStep: 42s",
|
||||
Type: data.FrameTypeTimeSeriesMany,
|
||||
})
|
||||
|
||||
if diff := cmp.Diff(testFrame, frame[0], data.FrameTestCompareOptions()...); diff != "" {
|
||||
t.Errorf("Result mismatch (-want +got):\n%s", diff)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("should set interval-attribute in response", func(t *testing.T) {
|
||||
values := []p.SamplePair{
|
||||
{Value: 1, Timestamp: 1000},
|
||||
}
|
||||
value := loghttp.QueryResponse{
|
||||
Data: loghttp.QueryResponseData{
|
||||
Result: loghttp.Matrix{
|
||||
p.SampleStream{
|
||||
Values: values,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
query := &lokiQuery{
|
||||
Step: time.Second * 42,
|
||||
QueryType: QueryTypeRange,
|
||||
}
|
||||
|
||||
frames, err := parseResponse(&value, query)
|
||||
require.NoError(t, err)
|
||||
|
||||
// to keep the test simple, we assume the
|
||||
// first field is the time-field
|
||||
timeField := frames[0].Fields[0]
|
||||
require.NotNil(t, timeField)
|
||||
require.Equal(t, data.FieldTypeTime, timeField.Type())
|
||||
|
||||
timeFieldConfig := timeField.Config
|
||||
require.NotNil(t, timeFieldConfig)
|
||||
require.Equal(t, float64(42000), timeFieldConfig.Interval)
|
||||
})
|
||||
|
||||
t.Run("should parse response stats", func(t *testing.T) {
|
||||
stats := stats.Result{
|
||||
Summary: stats.Summary{
|
||||
BytesProcessedPerSecond: 1,
|
||||
LinesProcessedPerSecond: 2,
|
||||
TotalBytesProcessed: 3,
|
||||
TotalLinesProcessed: 4,
|
||||
ExecTime: 5.5,
|
||||
},
|
||||
Store: stats.Store{
|
||||
TotalChunksRef: 6,
|
||||
TotalChunksDownloaded: 7,
|
||||
ChunksDownloadTime: 8.8,
|
||||
HeadChunkBytes: 9,
|
||||
HeadChunkLines: 10,
|
||||
DecompressedBytes: 11,
|
||||
DecompressedLines: 12,
|
||||
CompressedBytes: 13,
|
||||
TotalDuplicates: 14,
|
||||
},
|
||||
Ingester: stats.Ingester{
|
||||
TotalReached: 15,
|
||||
TotalChunksMatched: 16,
|
||||
TotalBatches: 17,
|
||||
TotalLinesSent: 18,
|
||||
HeadChunkBytes: 19,
|
||||
HeadChunkLines: 20,
|
||||
DecompressedBytes: 21,
|
||||
DecompressedLines: 22,
|
||||
CompressedBytes: 23,
|
||||
TotalDuplicates: 24,
|
||||
},
|
||||
}
|
||||
|
||||
expected := []data.QueryStat{
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Summary: bytes processed per second", Unit: "Bps"}, Value: 1},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Summary: lines processed per second", Unit: ""}, Value: 2},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Summary: total bytes processed", Unit: "decbytes"}, Value: 3},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Summary: total lines processed", Unit: ""}, Value: 4},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Summary: exec time", Unit: "s"}, Value: 5.5},
|
||||
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: total chunks ref", Unit: ""}, Value: 6},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: total chunks downloaded", Unit: ""}, Value: 7},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: chunks download time", Unit: "s"}, Value: 8.8},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: head chunk bytes", Unit: "decbytes"}, Value: 9},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: head chunk lines", Unit: ""}, Value: 10},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: decompressed bytes", Unit: "decbytes"}, Value: 11},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: decompressed lines", Unit: ""}, Value: 12},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: compressed bytes", Unit: "decbytes"}, Value: 13},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Store: total duplicates", Unit: ""}, Value: 14},
|
||||
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: total reached", Unit: ""}, Value: 15},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: total chunks matched", Unit: ""}, Value: 16},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: total batches", Unit: ""}, Value: 17},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: total lines sent", Unit: ""}, Value: 18},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: head chunk bytes", Unit: "decbytes"}, Value: 19},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: head chunk lines", Unit: ""}, Value: 20},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: decompressed bytes", Unit: "decbytes"}, Value: 21},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: decompressed lines", Unit: ""}, Value: 22},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: compressed bytes", Unit: "decbytes"}, Value: 23},
|
||||
{FieldConfig: data.FieldConfig{DisplayName: "Ingester: total duplicates", Unit: ""}, Value: 24},
|
||||
}
|
||||
|
||||
result := parseStats((stats))
|
||||
|
||||
// NOTE: i compare it item-by-item otherwise the test-fail-error-message is very hard to read
|
||||
require.Len(t, result, len(expected))
|
||||
|
||||
for i := 0; i < len(result); i++ {
|
||||
require.Equal(t, expected[i], result[i])
|
||||
}
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user