Revert to using timestamp output format

ddelemeny · ddelemeny · commit 8a3c88b0301b · 2024-03-18T15:52:15.000-05:00
diff --git a/pkg/quickwit/client/client.go b/pkg/quickwit/client/client.go
@@ -26,9 +26,10 @@ type DatasourceInfo struct {
 }
 
 type ConfiguredFields struct {
-	TimeField       string
-	LogMessageField string
-	LogLevelField   string
+	TimeField        string
+	TimeOutputFormat string
+	LogMessageField  string
+	LogLevelField    string
 }
 
 // Client represents a client which can interact with elasticsearch api
diff --git a/pkg/quickwit/error_handling_test.go b/pkg/quickwit/error_handling_test.go
@@ -39,9 +39,10 @@ func TestErrorAvgMissingField(t *testing.T) {
 	`)
 
 	configuredFields := es.ConfiguredFields{
-		TimeField:       "testtime",
-		LogMessageField: "line",
-		LogLevelField:   "lvl",
+		TimeOutputFormat: Rfc3339,
+		TimeField:        "testtime",
+		LogMessageField:  "line",
+		LogLevelField:    "lvl",
 	}
 
 	result, err := queryDataTestWithResponseCode(query, 400, response, configuredFields)
@@ -71,9 +72,10 @@ func TestErrorAvgMissingFieldNoDetailedErrors(t *testing.T) {
 	`)
 
 	configuredFields := es.ConfiguredFields{
-		TimeField:       "testtime",
-		LogMessageField: "line",
-		LogLevelField:   "lvl",
+		TimeOutputFormat: Rfc3339,
+		TimeField:        "testtime",
+		LogMessageField:  "line",
+		LogLevelField:    "lvl",
 	}
 
 	result, err := queryDataTestWithResponseCode(query, 400, response, configuredFields)
@@ -117,9 +119,10 @@ func TestErrorTooManyDateHistogramBuckets(t *testing.T) {
 	`)
 
 	configuredFields := es.ConfiguredFields{
-		TimeField:       "testtime",
-		LogMessageField: "line",
-		LogLevelField:   "lvl",
+		TimeOutputFormat: Rfc3339,
+		TimeField:        "testtime",
+		LogMessageField:  "line",
+		LogLevelField:    "lvl",
 	}
 	result, err := queryDataTestWithResponseCode(query, 200, response, configuredFields)
 	require.NoError(t, err)
@@ -154,9 +157,10 @@ func TestNonElasticError(t *testing.T) {
 	response := []byte(`Access to the database is forbidden`)
 
 	configuredFields := es.ConfiguredFields{
-		TimeField:       "testtime",
-		LogMessageField: "line",
-		LogLevelField:   "lvl",
+		TimeOutputFormat: Rfc3339,
+		TimeField:        "testtime",
+		LogMessageField:  "line",
+		LogLevelField:    "lvl",
 	}
 
 	result, err := queryDataTestWithResponseCode(query, 403, response, configuredFields)
diff --git a/pkg/quickwit/models.go b/pkg/quickwit/models.go
@@ -122,3 +122,13 @@ func describeMetric(metricType, field string) string {
 	}
 	return text + " " + field
 }
+
+const (
+	Iso8601         string = "iso8601"
+	Rfc2822         string = "rfc2822"
+	Rfc3339         string = "rfc3339"
+	TimestampSecs   string = "unix_timestamp_secs"
+	TimestampMillis string = "unix_timestamp_millis"
+	TimestampMicros string = "unix_timestamp_micros"
+	TimestampNanos  string = "unix_timestamp_nanos"
+)
diff --git a/pkg/quickwit/querydata_test.go b/pkg/quickwit/querydata_test.go
@@ -144,9 +144,10 @@ func queryDataTestWithResponseCode(queriesBytes []byte, responseStatusCode int,
 
 func queryDataTest(queriesBytes []byte, responseBytes []byte) (queryDataTestResult, error) {
 	configuredFields := es.ConfiguredFields{
-		TimeField:       "testtime",
-		LogMessageField: "line",
-		LogLevelField:   "lvl",
+		TimeOutputFormat: Rfc3339,
+		TimeField:        "testtime",
+		LogMessageField:  "line",
+		LogLevelField:    "lvl",
 	}
 	return queryDataTestWithResponseCode(queriesBytes, 200, responseBytes, configuredFields)
 }
diff --git a/pkg/quickwit/quickwit.go b/pkg/quickwit/quickwit.go
@@ -58,6 +58,7 @@ func NewQuickwitDatasource(settings backend.DataSourceInstanceSettings) (instanc
 	}
 
 	timeField, toOk := jsonData["timeField"].(string)
+	timeOutputFormat, tofOk := jsonData["timeOutputFormat"].(string)
 
 	logLevelField, ok := jsonData["logLevelField"].(string)
 	if !ok {
@@ -91,17 +92,18 @@ func NewQuickwitDatasource(settings backend.DataSourceInstanceSettings) (instanc
 		maxConcurrentShardRequests = 256
 	}
 
-	if !toOk {
-		timeField, err = GetTimestampField(index, settings.URL, httpCli)
+	if !toOk || !tofOk {
+		timeField, timeOutputFormat, err = GetTimestampFieldInfos(index, settings.URL, httpCli)
 		if nil != err {
 			return nil, err
 		}
 	}
 
 	configuredFields := es.ConfiguredFields{
-		TimeField:       timeField,
-		LogLevelField:   logLevelField,
-		LogMessageField: logMessageField,
+		TimeField:        timeField,
+		TimeOutputFormat: timeOutputFormat,
+		LogLevelField:    logLevelField,
+		LogMessageField:  logMessageField,
 	}
 
 	model := es.DatasourceInfo{
diff --git a/pkg/quickwit/response_parser.go b/pkg/quickwit/response_parser.go
@@ -12,6 +12,7 @@ import (
 
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 	"github.com/grafana/grafana-plugin-sdk-go/data"
+	"golang.org/x/exp/slices"
 
 	es "github.com/quickwit-oss/quickwit-datasource/pkg/quickwit/client"
 	"github.com/quickwit-oss/quickwit-datasource/pkg/quickwit/simplejson"
@@ -246,7 +247,7 @@ func processDocsToDataFrameFields(docs []map[string]interface{}, propNames []str
 		if propName == configuredFields.TimeField {
 			timeVector := make([]*time.Time, size)
 			for i, doc := range docs {
-				timeValue, err := ParseToTime(doc["sort"].([]any)[0])
+				timeValue, err := ParseToTime(doc[configuredFields.TimeField], configuredFields.TimeOutputFormat)
 				if err != nil {
 					continue
 				}
@@ -293,13 +294,39 @@ func processDocsToDataFrameFields(docs []map[string]interface{}, propNames []str
 // Parses a value into Time given a timeOutputFormat. The conversion
 // only works with float64 as this is what we get when parsing a response.
 // TODO: understand why we get a float64?
-func ParseToTime(value interface{}) (time.Time, error) {
-	typed_value, ok := value.(float64)
-	if !ok {
-		return time.Time{}, errors.New("parse time only accepts float64 with timestamp based format")
+func ParseToTime(value interface{}, timeOutputFormat string) (time.Time, error) {
+
+	if timeOutputFormat == Iso8601 || timeOutputFormat == Rfc3339 {
+		value_string := value.(string)
+		timeValue, err := time.Parse(time.RFC3339, value_string)
+		if err != nil {
+			return time.Time{}, err
+		}
+		return timeValue, nil
+	} else if timeOutputFormat == Rfc2822 {
+		value_string := value.(string)
+		timeValue, err := time.Parse(time.RFC822Z, value_string)
+		if err != nil {
+			return time.Time{}, err
+		}
+		return timeValue, nil
+	} else if slices.Contains([]string{TimestampSecs, TimestampMillis, TimestampMicros, TimestampNanos}, timeOutputFormat) {
+		typed_value, ok := value.(float64)
+		if !ok {
+			return time.Time{}, errors.New("parse time only accepts float64 with timestamp based format")
+		}
+		int64_value := int64(typed_value)
+		if timeOutputFormat == TimestampSecs {
+			return time.Unix(int64_value, 0), nil
+		} else if timeOutputFormat == TimestampMillis {
+			return time.Unix(0, int64_value*1_000_000), nil
+		} else if timeOutputFormat == TimestampMicros {
+			return time.Unix(0, int64_value*1_000), nil
+		} else if timeOutputFormat == TimestampNanos {
+			return time.Unix(0, int64_value), nil
+		}
 	}
-	int64_value := int64(typed_value)
-	return time.Unix(0, int64_value), nil
+	return time.Time{}, fmt.Errorf("timeOutputFormat not supported yet %s", timeOutputFormat)
 }
 
 func processBuckets(aggs map[string]interface{}, target *Query,
diff --git a/pkg/quickwit/response_parser_qw_test.go b/pkg/quickwit/response_parser_qw_test.go
@@ -59,9 +59,10 @@ func TestProcessLogsResponseWithDifferentTimeOutputFormat(t *testing.T) {
 			`)
 
 		configuredFields := es.ConfiguredFields{
-			TimeField:       "testtime",
-			LogMessageField: "line",
-			LogLevelField:   "lvl",
+			TimeOutputFormat: TimestampNanos,
+			TimeField:        "testtime",
+			LogMessageField:  "line",
+			LogLevelField:    "lvl",
 		}
 		result, _ := queryDataTestWithResponseCode(query, 200, response, configuredFields)
 		frames := result.response.Responses["A"].Frames
@@ -125,9 +126,10 @@ func TestProcessLogsResponseWithDifferentTimeOutputFormat(t *testing.T) {
 			`)
 
 		configuredFields := es.ConfiguredFields{
-			TimeField:       "testtime",
-			LogMessageField: "line",
-			LogLevelField:   "lvl",
+			TimeOutputFormat: TimestampMicros,
+			TimeField:        "testtime",
+			LogMessageField:  "line",
+			LogLevelField:    "lvl",
 		}
 		result, _ := queryDataTestWithResponseCode(query, 200, response, configuredFields)
 		frames := result.response.Responses["A"].Frames
@@ -191,9 +193,10 @@ func TestProcessLogsResponseWithDifferentTimeOutputFormat(t *testing.T) {
 			`)
 
 		configuredFields := es.ConfiguredFields{
-			TimeField:       "testtime",
-			LogMessageField: "line",
-			LogLevelField:   "lvl",
+			TimeOutputFormat: TimestampMillis,
+			TimeField:        "testtime",
+			LogMessageField:  "line",
+			LogLevelField:    "lvl",
 		}
 		result, _ := queryDataTestWithResponseCode(query, 200, response, configuredFields)
 		frames := result.response.Responses["A"].Frames
@@ -257,9 +260,10 @@ func TestProcessLogsResponseWithDifferentTimeOutputFormat(t *testing.T) {
 			`)
 
 		configuredFields := es.ConfiguredFields{
-			TimeField:       "testtime",
-			LogMessageField: "line",
-			LogLevelField:   "lvl",
+			TimeOutputFormat: TimestampSecs,
+			TimeField:        "testtime",
+			LogMessageField:  "line",
+			LogLevelField:    "lvl",
 		}
 		result, _ := queryDataTestWithResponseCode(query, 200, response, configuredFields)
 		frames := result.response.Responses["A"].Frames
@@ -278,7 +282,7 @@ func TestProcessLogsResponseWithDifferentTimeOutputFormat(t *testing.T) {
 func TestConvertToTime(t *testing.T) {
 	t.Run("Test parse unix timestamps nanosecs of float type", func(t *testing.T) {
 		inputValue := interface{}(1234567890000000000.0)
-		value, _ := ParseToTime(inputValue)
+		value, _ := ParseToTime(inputValue, "unix_timestamp_nanos")
 		require.Equal(t, time.Unix(1234567890, 0), value)
 	})
 }
diff --git a/pkg/quickwit/response_parser_test.go b/pkg/quickwit/response_parser_test.go
@@ -3228,9 +3228,10 @@ func parseTestResponse(tsdbQueries map[string]string, responseBody string) (*bac
 	from := time.Date(2018, 5, 15, 17, 50, 0, 0, time.UTC)
 	to := time.Date(2018, 5, 15, 17, 55, 0, 0, time.UTC)
 	configuredFields := es.ConfiguredFields{
-		TimeField:       "@timestamp",
-		LogMessageField: "line",
-		LogLevelField:   "lvl",
+		TimeOutputFormat: Rfc3339,
+		TimeField:        "@timestamp",
+		LogMessageField:  "line",
+		LogLevelField:    "lvl",
 	}
 	timeRange := backend.TimeRange{
 		From: from,
diff --git a/pkg/quickwit/timestamp_infos.go b/pkg/quickwit/timestamp_infos.go
diff --git a/pkg/quickwit/timestamp_infos_test.go b/pkg/quickwit/timestamp_infos_test.go

Original file line number	Diff line number	Diff line change
`@@ -26,9 +26,10 @@ type DatasourceInfo struct {`
`26`	`26`	`}`
`27`	`27`
`28`	`28`	`type ConfiguredFields struct {`
`29`		`- TimeField string`
`30`		`- LogMessageField string`
`31`		`- LogLevelField string`
	`29`	`+ TimeField string`
	`30`	`+ TimeOutputFormat string`
	`31`	`+ LogMessageField string`
	`32`	`+ LogLevelField string`
`32`	`33`	`}`
`33`	`34`
`34`	`35`	`// Client represents a client which can interact with elasticsearch api`
Original file line number	Diff line number	Diff line change
`@@ -122,3 +122,13 @@ func describeMetric(metricType, field string) string {`
`122`	`122`	`}`
`123`	`123`	`return text + " " + field`
`124`	`124`	`}`
	`125`	`+`
	`126`	`+const (`
	`127`	`+ Iso8601 string = "iso8601"`
	`128`	`+ Rfc2822 string = "rfc2822"`
	`129`	`+ Rfc3339 string = "rfc3339"`
	`130`	`+ TimestampSecs string = "unix_timestamp_secs"`
	`131`	`+ TimestampMillis string = "unix_timestamp_millis"`
	`132`	`+ TimestampMicros string = "unix_timestamp_micros"`
	`133`	`+ TimestampNanos string = "unix_timestamp_nanos"`
	`134`	`+)`
Original file line number	Diff line number	Diff line change
`@@ -144,9 +144,10 @@ func queryDataTestWithResponseCode(queriesBytes []byte, responseStatusCode int,`
`144`	`144`
`145`	`145`	`func queryDataTest(queriesBytes []byte, responseBytes []byte) (queryDataTestResult, error) {`
`146`	`146`	`configuredFields := es.ConfiguredFields{`
`147`		`- TimeField: "testtime",`
`148`		`- LogMessageField: "line",`
`149`		`- LogLevelField: "lvl",`
	`147`	`+ TimeOutputFormat: Rfc3339,`
	`148`	`+ TimeField: "testtime",`
	`149`	`+ LogMessageField: "line",`
	`150`	`+ LogLevelField: "lvl",`
`150`	`151`	`}`
`151`	`152`	`return queryDataTestWithResponseCode(queriesBytes, 200, responseBytes, configuredFields)`
`152`	`153`	`}`
Original file line number	Diff line number	Diff line change
`@@ -58,6 +58,7 @@ func NewQuickwitDatasource(settings backend.DataSourceInstanceSettings) (instanc`
`58`	`58`	`}`
`59`	`59`
`60`	`60`	`timeField, toOk := jsonData["timeField"].(string)`
	`61`	`+ timeOutputFormat, tofOk := jsonData["timeOutputFormat"].(string)`
`61`	`62`
`62`	`63`	`logLevelField, ok := jsonData["logLevelField"].(string)`
`63`	`64`	`if !ok {`
`@@ -91,17 +92,18 @@ func NewQuickwitDatasource(settings backend.DataSourceInstanceSettings) (instanc`
`91`	`92`	`maxConcurrentShardRequests = 256`
`92`	`93`	`}`
`93`	`94`
`94`		`- if !toOk {`
`95`		`- timeField, err = GetTimestampField(index, settings.URL, httpCli)`
	`95`	`+ if !toOk \|\| !tofOk {`
	`96`	`+ timeField, timeOutputFormat, err = GetTimestampFieldInfos(index, settings.URL, httpCli)`
`96`	`97`	`if nil != err {`
`97`	`98`	`return nil, err`
`98`	`99`	`}`
`99`	`100`	`}`
`100`	`101`
`101`	`102`	`configuredFields := es.ConfiguredFields{`
`102`		`- TimeField: timeField,`
`103`		`- LogLevelField: logLevelField,`
`104`		`- LogMessageField: logMessageField,`
	`103`	`+ TimeField: timeField,`
	`104`	`+ TimeOutputFormat: timeOutputFormat,`
	`105`	`+ LogLevelField: logLevelField,`
	`106`	`+ LogMessageField: logMessageField,`
`105`	`107`	`}`
`106`	`108`
`107`	`109`	`model := es.DatasourceInfo{`