Merge pull request #38 from quickwit-oss/feat_timestamp_field

Issue #4: timestamp field

Author: idrissneumann

pkg/quickwit/quickwit.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -26,6 +25,13 @@ type QuickwitDatasource struct {
 	dsInfo es.DatasourceInfo
 }
 
+type FieldMappings struct {
+	Name          string          `json:"name"`
+	Type          string          `json:"type"`
+	OutputFormat  *string         `json:"output_format,omitempty"`
+	FieldMappings []FieldMappings `json:"field_mappings,omitempty"`
+}
+
 // Creates a Quickwit datasource.
 func NewQuickwitDatasource(settings backend.DataSourceInstanceSettings) (instancemgmt.Instance, error) {
 	qwlog.Debug("Initializing new data source instance")
@@ -50,19 +56,8 @@ func NewQuickwitDatasource(settings backend.DataSourceInstanceSettings) (instanc
 		return nil, err
 	}
 
-	timeField, ok := jsonData["timeField"].(string)
-	if !ok {
-		return nil, errors.New("timeField cannot be cast to string")
-	}
-
-	if timeField == "" {
-		return nil, errors.New("a time field name is required")
-	}
-
-	timeOutputFormat, ok := jsonData["timeOutputFormat"].(string)
-	if !ok {
-		return nil, errors.New("timeOutputFormat cannot be cast to string")
-	}
+	timeField, toOk := jsonData["timeField"].(string)
+	timeOutputFormat, tofOk := jsonData["timeOutputFormat"].(string)
 
 	logLevelField, ok := jsonData["logLevelField"].(string)
 	if !ok {
@@ -96,6 +91,13 @@ func NewQuickwitDatasource(settings backend.DataSourceInstanceSettings) (instanc
 		maxConcurrentShardRequests = 256
 	}
 
+	if !toOk || !tofOk {
+		timeField, timeOutputFormat, err = GetTimestampFieldInfos(index, settings.URL, httpCli)
+		if nil != err {
+			return nil, err
+		}
+	}
+
 	configuredFields := es.ConfiguredFields{
 		TimeField:        timeField,
 		TimeOutputFormat: timeOutputFormat,

pkg/quickwit/timestamp_infos.go

@@ -0,0 +1,111 @@
+package quickwit
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+)
+
+type QuickwitMapping struct {
+	IndexConfig struct {
+		DocMapping struct {
+			TimestampField string          `json:"timestamp_field"`
+			FieldMappings  []FieldMappings `json:"field_mappings"`
+		} `json:"doc_mapping"`
+	} `json:"index_config"`
+}
+
+type QuickwitCreationErrorPayload struct {
+	Message    string `json:"message"`
+	StatusCode int    `json:"status"`
+}
+
+func NewErrorCreationPayload(statusCode int, message string) error {
+	var payload QuickwitCreationErrorPayload
+	payload.Message = message
+	payload.StatusCode = statusCode
+	json, err := json.Marshal(payload)
+	if nil != err {
+		return err
+	}
+
+	return errors.New(string(json))
+}
+
+func FindTimeStampFormat(timestampFieldName string, parentName *string, fieldMappings []FieldMappings) *string {
+	if nil == fieldMappings {
+		return nil
+	}
+
+	for _, field := range fieldMappings {
+		fieldName := field.Name
+		if nil != parentName {
+			fieldName = fmt.Sprintf("%s.%s", *parentName, fieldName)
+		}
+
+		if field.Type == "datetime" && fieldName == timestampFieldName && nil != field.OutputFormat {
+			return field.OutputFormat
+		} else if field.Type == "object" && nil != field.FieldMappings {
+			format := FindTimeStampFormat(timestampFieldName, &field.Name, field.FieldMappings)
+			if nil != format {
+				return format
+			}
+		}
+	}
+
+	return nil
+}
+
+func DecodeTimestampFieldInfos(statusCode int, body []byte) (string, string, error) {
+	var payload QuickwitMapping
+	err := json.Unmarshal(body, &payload)
+
+	if err != nil {
+		errMsg := fmt.Sprintf("Unmarshalling body error: err = %s, body = %s", err.Error(), (body))
+		qwlog.Error(errMsg)
+		return "", "", NewErrorCreationPayload(statusCode, errMsg)
+	}
+
+	timestampFieldName := payload.IndexConfig.DocMapping.TimestampField
+	timestampFieldFormat := FindTimeStampFormat(timestampFieldName, nil, payload.IndexConfig.DocMapping.FieldMappings)
+
+	if nil == timestampFieldFormat {
+		errMsg := fmt.Sprintf("No format found for field: %s", string(timestampFieldName))
+		qwlog.Error(errMsg)
+		return timestampFieldName, "", NewErrorCreationPayload(statusCode, errMsg)
+	}
+
+	qwlog.Info(fmt.Sprintf("Found timestampFieldName = %s, timestampFieldFormat = %s", timestampFieldName, *timestampFieldFormat))
+	return timestampFieldName, *timestampFieldFormat, nil
+}
+
+func GetTimestampFieldInfos(index string, qwUrl string, cli *http.Client) (string, string, error) {
+	mappingEndpointUrl := qwUrl + "/indexes/" + index
+	qwlog.Info("Calling quickwit endpoint: " + mappingEndpointUrl)
+	r, err := cli.Get(mappingEndpointUrl)
+	if err != nil {
+		errMsg := fmt.Sprintf("Error when calling url = %s: err = %s", mappingEndpointUrl, err.Error())
+		qwlog.Error(errMsg)
+		return "", "", err
+	}
+
+	statusCode := r.StatusCode
+
+	if statusCode < 200 || statusCode >= 400 {
+		errMsg := fmt.Sprintf("Error when calling url = %s", mappingEndpointUrl)
+		qwlog.Error(errMsg)
+		return "", "", NewErrorCreationPayload(statusCode, errMsg)
+	}
+
+	defer r.Body.Close()
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		errMsg := fmt.Sprintf("Error when calling url = %s: err = %s", mappingEndpointUrl, err.Error())
+		qwlog.Error(errMsg)
+		return "", "", NewErrorCreationPayload(statusCode, errMsg)
+	}
+
+	return DecodeTimestampFieldInfos(statusCode, body)
+}