Skip to content

Commit

Permalink
All trace list requests go through backend (#310)
Browse files Browse the repository at this point in the history
  • Loading branch information
@sarahzinger
sarahzinger authored Nov 21, 2023
1 parent 0b33f6d commit dcff2a3
Show file tree
Hide file tree
Showing 6 changed files with 1,059 additions and 13 deletions.
68 changes: 62 additions & 6 deletions pkg/opensearch/response_parser.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -96,21 +96,26 @@ func (rp *responseParser) parseResponse() (*backend.QueryDataResponse, error) {
Frames: data.Frames{},
}

// trace span condition
// trace queries are sent from the FE with a metrics field, returning early so the switch doesn't overwrite the response from the traces query
var queryType string
if target.luceneQueryType == luceneQueryTypeTraces {
queryRes = processTraceSpansResponse(res, queryRes)
result.Responses[target.RefID] = queryRes
continue
queryType = luceneQueryTypeTraces
} else {
queryType = target.Metrics[0].Type
}

switch target.Metrics[0].Type {
switch queryType {
case rawDataType:
queryRes = processRawDataResponse(res, rp.ConfiguredFields, queryRes)
case rawDocumentType:
queryRes = processRawDocumentResponse(res, target.RefID, queryRes)
case logsType:
queryRes = processLogsResponse(res, rp.ConfiguredFields, queryRes)
case luceneQueryTypeTraces:
if strings.HasPrefix(target.RawQuery, "traceId:") {
queryRes = processTraceSpansResponse(res, queryRes)
} else {
queryRes = processTraceListResponse(res, rp.DSSettings.UID, rp.DSSettings.Name, queryRes)
}
default:
props := make(map[string]string)
err := rp.processBuckets(res.Aggregations, target, &queryRes, props, 0)
Expand All @@ -123,6 +128,7 @@ func (rp *responseParser) parseResponse() (*backend.QueryDataResponse, error) {

result.Responses[target.RefID] = queryRes
}

return result, nil
}

Expand Down Expand Up @@ -264,6 +270,56 @@ func processTraceSpansResponse(res *es.SearchResponse, queryRes backend.DataResp
queryRes.Frames = data.Frames{frame}
return queryRes
}

func processTraceListResponse(res *es.SearchResponse, dsUID string, dsName string, queryRes backend.DataResponse) backend.DataResponse {
// trace list queries are hardcoded with a fairly hardcoded response format
// but es.SearchResponse is deliberately not typed as in other query cases it can be much more open ended
rawTraces := res.Aggregations["traces"].(map[string]interface{})["buckets"].([]interface{})

// get values from raw traces response
traceIds := []string{}
traceGroups := []string{}
traceLatencies := []float64{}
traceErrorCounts := []float64{}
traceLastUpdated := []time.Time{}
for _, t := range rawTraces {
trace := t.(map[string]interface{})

traceIds = append(traceIds, trace["key"].(string))
traceGroups = append(traceGroups, trace["trace_group"].(map[string]interface{})["buckets"].([]interface{})[0].(map[string]interface{})["key"].(string))
traceLatencies = append(traceLatencies, trace["latency"].(map[string]interface{})["value"].(float64))
traceErrorCounts = append(traceErrorCounts, trace["error_count"].(map[string]interface{})["doc_count"].(float64))
lastUpdated := trace["last_updated"].(map[string]interface{})["value"].(float64)
traceLastUpdated = append(traceLastUpdated, time.Unix(0, int64(lastUpdated)*int64(time.Millisecond)))
}

allFields := make([]*data.Field, 0, 5)
traceIdColumn := data.NewField("Trace Id", nil, traceIds)
traceIdColumn.Config = &data.FieldConfig{
Links: []data.DataLink{
{
Internal: &data.InternalDataLink{
Query: map[string]interface{}{
"query": "traceId: ${__value.raw}",
"luceneQueryType": "Traces",
},
DatasourceUID: dsUID,
DatasourceName: dsName,
},
},
},
}

allFields = append(allFields, traceIdColumn)
allFields = append(allFields, data.NewField("Trace Group", nil, traceGroups))
allFields = append(allFields, data.NewField("Latency (ms)", nil, traceLatencies))
allFields = append(allFields, data.NewField("Error Count", nil, traceErrorCounts))
allFields = append(allFields, data.NewField("Last Updated", nil, traceLastUpdated))

queryRes.Frames = data.Frames{data.NewFrame("Trace List", allFields...)}
return queryRes
}

func processLogsResponse(res *es.SearchResponse, configuredFields es.ConfiguredFields, queryRes backend.DataResponse) backend.DataResponse {
propNames := make(map[string]bool)
docs := make([]map[string]interface{}, len(res.Hits.Hits))
Expand Down
86 changes: 86 additions & 0 deletions pkg/opensearch/response_parser_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2573,3 +2573,89 @@ func sortLogsByTimestamp(rawObject *data.Field, t *testing.T) []Log {
})
return sortedArray
}

func TestProcessTraceListResponse(t *testing.T) {
targets := map[string]string{
"A": `{
"timeField": "@timestamp",
"metrics": [{ "type": "count", "id": "1" }],
"luceneQueryType": "Traces"
}`,
}

response := `
{
"responses": [{
"aggregations": {
"traces": {
"buckets": [{
"doc_count": 50,
"key": "000000000000000001c01e08995dd2e2",
"last_updated": {
"value": 1700074430928,
"value_as_string": "2023-11-15T18:53:50.928Z"
},
"latency": {
"value": 656.43
},
"trace_group": {
"buckets":[{
"doc_count":50,
"key": "HTTP GET /dispatch"
}]
},
"error_count": {
"doc_count":0
}
}]
}
}
}]
}
`

rp, err := newResponseParserForTest(targets, response, nil, client.ConfiguredFields{TimeField: "@timestamp"}, &backend.DataSourceInstanceSettings{UID: "123", Name: "DatasourceInstanceName"})
assert.Nil(t, err)

result, err := rp.parseResponse()
require.NoError(t, err)
require.Len(t, result.Responses, 1)

queryRes := result.Responses["A"]
require.NotNil(t, queryRes)

dataframes := queryRes.Frames
require.Len(t, dataframes, 1)

frame := dataframes[0]

traceId := frame.Fields[0]
assert.Equal(t, "000000000000000001c01e08995dd2e2", traceId.At(0))
assert.Equal(t, "Trace Id", traceId.Name)
assert.Equal(t, "string", traceId.Type().ItemTypeString())
//deep link config to make it possible to click through to individual trace view
assert.Equal(t, "traceId: ${__value.raw}", traceId.Config.Links[0].Internal.Query.(map[string]interface{})["query"])
assert.Equal(t, "Traces", traceId.Config.Links[0].Internal.Query.(map[string]interface{})["luceneQueryType"])
assert.Equal(t, "123", traceId.Config.Links[0].Internal.DatasourceUID)
assert.Equal(t, "DatasourceInstanceName", traceId.Config.Links[0].Internal.DatasourceName)

traceGroup := frame.Fields[1]
assert.Equal(t, "HTTP GET /dispatch", traceGroup.At(0))
assert.Equal(t, "Trace Group", traceGroup.Name)
assert.Equal(t, "string", traceGroup.Type().ItemTypeString())

latency := frame.Fields[2]
assert.Equal(t, 656.43, latency.At(0))
assert.Equal(t, "Latency (ms)", latency.Name)
assert.Equal(t, "float64", latency.Type().ItemTypeString())

errorCount := frame.Fields[3]
assert.Equal(t, float64(0), errorCount.At(0))
assert.Equal(t, "Error Count", errorCount.Name)
assert.Equal(t, "float64", errorCount.Type().ItemTypeString())

lastUpdated := frame.Fields[4]
assert.Equal(t, time.Unix(0, int64(1700074430928)*int64(time.Millisecond)), lastUpdated.At(0))
assert.Equal(t, "Last Updated", lastUpdated.Name)
assert.Equal(t, "time.Time", lastUpdated.Type().ItemTypeString())
}
25 changes: 25 additions & 0 deletions pkg/opensearch/snapshot_tests/lucene_trace_list_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,11 @@ import (
"context"
"io"
"net/http"
"os"
"testing"

"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana-plugin-sdk-go/experimental"
"github.com/grafana/opensearch-datasource/pkg/opensearch"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
Expand Down Expand Up @@ -43,3 +45,26 @@ func Test_trace_list_request(t *testing.T) {
`
assert.Equal(t, expectedRequest, string(interceptedRequest))
}

func Test_trace_list_response(t *testing.T) {
responseFromOpenSearch, err := os.ReadFile("testdata/lucene_trace_list.response_from_opensearch.json")
require.NoError(t, err)
queries, err := setUpDataQueriesFromFileWithFixedTimeRange(t, "testdata/lucene_trace_list.query_input.json")
require.NoError(t, err)
openSearchDatasource := opensearch.OpenSearchDatasource{
HttpClient: &http.Client{
Transport: &queryDataTestRoundTripper{body: responseFromOpenSearch, statusCode: 200, requestCallback: func(req *http.Request) error { return nil }},
},
}

result, err := openSearchDatasource.QueryData(context.Background(), &backend.QueryDataRequest{
PluginContext: backend.PluginContext{DataSourceInstanceSettings: newTestDsSettings()},
Headers: nil,
Queries: queries,
})
require.NoError(t, err)

responseForRefIdA, ok := result.Responses["A"]
assert.True(t, ok)
experimental.CheckGoldenJSONResponse(t, "testdata", "lucene_trace_list.expected_result_generated_snapshot.golden", &responseForRefIdA, false)
}
167 changes: 167 additions & 0 deletions ...snapshot_tests/testdata/lucene_trace_list.expected_result_generated_snapshot.golden.jsonc
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,167 @@
// 🌟 This was machine generated. Do not edit. 🌟
//
// Frame[0]
// Name: Trace List
// Dimensions: 5 Fields by 14 Rows
// +----------------------------------+--------------------+--------------------+-------------------+-----------------------------------+
// | Name: Trace Id | Name: Trace Group | Name: Latency (ms) | Name: Error Count | Name: Last Updated |
// | Labels: | Labels: | Labels: | Labels: | Labels: |
// | Type: []string | Type: []string | Type: []float64 | Type: []float64 | Type: []time.Time |
// +----------------------------------+--------------------+--------------------+-------------------+-----------------------------------+
// | 00000000000000001c826277770e267d | HTTP GET /dispatch | 671.91 | 0 | 2023-11-21 14:39:46.811 -0500 EST |
// | 0000000000000000252c7c74849b6fe7 | HTTP GET /dispatch | 760.23 | 0 | 2023-11-21 14:39:48.782 -0500 EST |
// | 0000000000000000260d9137e9aea627 | HTTP GET /dispatch | 735.64 | 0 | 2023-11-21 14:48:19.622 -0500 EST |
// | 00000000000000003a2735bf3ecf9fbe | HTTP GET / | 2.92 | 0 | 2023-11-21 14:39:41.248 -0500 EST |
// | 000000000000000046f0ef81931b97f9 | HTTP GET / | 0.63 | 0 | 2023-11-21 14:48:17.544 -0500 EST |
// | 000000000000000057603fd19d265431 | HTTP GET / | 1.02 | 0 | 2023-11-21 14:39:41.225 -0500 EST |
// | 000000000000000057a6dd673748973d | HTTP GET /config | 0.02 | 0 | 2023-11-21 14:39:50.157 -0500 EST |
// | 00000000000000005bb75b8cd50e57ca | HTTP GET / | 0.81 | 0 | 2023-11-21 14:48:17.482 -0500 EST |
// | 000000000000000063a3f64f32254597 | HTTP GET /config | 0.13 | 0 | 2023-11-21 14:39:43.828 -0500 EST |
// | ... | ... | ... | ... | ... |
// +----------------------------------+--------------------+--------------------+-------------------+-----------------------------------+
//
//
// 🌟 This was machine generated. Do not edit. 🌟
{
"status": 200,
"frames": [
{
"schema": {
"name": "Trace List",
"fields": [
{
"name": "Trace Id",
"type": "string",
"typeInfo": {
"frame": "string"
},
"config": {
"links": [
{
"internal": {
"query": {
"luceneQueryType": "Traces",
"query": "traceId: ${__value.raw}"
}
}
}
]
}
},
{
"name": "Trace Group",
"type": "string",
"typeInfo": {
"frame": "string"
}
},
{
"name": "Latency (ms)",
"type": "number",
"typeInfo": {
"frame": "float64"
}
},
{
"name": "Error Count",
"type": "number",
"typeInfo": {
"frame": "float64"
}
},
{
"name": "Last Updated",
"type": "time",
"typeInfo": {
"frame": "time.Time"
}
}
]
},
"data": {
"values": [
[
"00000000000000001c826277770e267d",
"0000000000000000252c7c74849b6fe7",
"0000000000000000260d9137e9aea627",
"00000000000000003a2735bf3ecf9fbe",
"000000000000000046f0ef81931b97f9",
"000000000000000057603fd19d265431",
"000000000000000057a6dd673748973d",
"00000000000000005bb75b8cd50e57ca",
"000000000000000063a3f64f32254597",
"00000000000000006486522b83a8a0b1",
"000000000000000071beb0b4e7e64589",
"000000000000000074921202d333f490",
"00000000000000007df1103ab1f31768",
"00000000000000007ec612d7ab1c325f"
],
[
"HTTP GET /dispatch",
"HTTP GET /dispatch",
"HTTP GET /dispatch",
"HTTP GET /",
"HTTP GET /",
"HTTP GET /",
"HTTP GET /config",
"HTTP GET /",
"HTTP GET /config",
"HTTP GET /config",
"HTTP GET /config",
"HTTP GET /dispatch",
"HTTP GET /config",
"HTTP GET /dispatch"
],
[
671.91,
760.23,
735.64,
2.92,
0.63,
1.02,
0.02,
0.81,
0.13,
0.04,
0.03,
679,
0.16,
727.8
],
[
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
0
],
[
1700595586811,
1700595588782,
1700596099622,
1700595581248,
1700596097544,
1700595581225,
1700595590157,
1700596097482,
1700595583828,
1700595586139,
1700595588022,
1700595590836,
1700596098886,
1700595584556
]
]
}
}
]
}
Loading

0 comments on commit dcff2a3

Please sign in to comment.