Skip to content

Commit

Permalink
Merge branch 'main' into feat_highlight
Browse files Browse the repository at this point in the history
  • Loading branch information
idrissneumann committed Dec 28, 2023
2 parents 54c0d0f + 0be5213 commit 121dbb4
Show file tree
Hide file tree
Showing 19 changed files with 1,125 additions and 112 deletions.
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,11 @@
# Changelog

## 0.3.0-beta.1

This version works only with quickwit main version (or docker edge).

- Add support for data links (possibility to create links to other datasources).

## 0.3.0-beta.0

This version works only with quickwit main version (or docker edge).
Expand Down
18 changes: 18 additions & 0 deletions docker-compose.yaml
Original file line number Diff line number Diff line change
@@ -1,6 +1,18 @@
version: '3.0'

services:
jaeger:
image: jaegertracing/jaeger-query:1.51
container_name: 'jaeger-quickwit'
environment:
- GRPC_STORAGE_SERVER=host.docker.internal:7281
- SPAN_STORAGE_TYPE=grpc-plugin
ports:
- 16686:16686
extra_hosts:
- "host.docker.internal:host-gateway"
networks:
- quickwit
grafana:
container_name: 'grafana-quickwit-datasource'
build:
Expand All @@ -15,6 +27,12 @@ services:
- gquickwit:/var/lib/grafana
extra_hosts:
- "host.docker.internal:host-gateway"
networks:
- quickwit

networks:
quickwit:
driver: bridge

volumes:
gquickwit:
4 changes: 2 additions & 2 deletions package-lock.json

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion package.json
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"name": "quickwit-datasource",
"version": "0.3.0-beta.1",
"version": "0.3.0-beta.2",
"description": "Quickwit datasource",
"scripts": {
"build": "webpack -c ./.config/webpack/webpack.config.ts --env production",
Expand Down
34 changes: 19 additions & 15 deletions pkg/quickwit/quickwit.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@ import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
Expand All @@ -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")
Expand All @@ -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 {
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -143,7 +145,9 @@ func (ds *QuickwitDatasource) CallResource(ctx context.Context, req *backend.Cal
// - empty string for fetching db version
// - ?/_mapping for fetching index mapping
// - _msearch for executing getTerms queries
if req.Path != "" && !strings.Contains(req.Path, "indexes/") && req.Path != "_elastic/_msearch" {
// - _field_caps for getting all the aggregeables fields
var isFieldCaps = req.Path != "" && strings.Contains(req.Path, "_elastic") && strings.Contains(req.Path, "/_field_caps")
if req.Path != "" && !strings.Contains(req.Path, "indexes/") && req.Path != "_elastic/_msearch" && !isFieldCaps {
return fmt.Errorf("invalid resource URL: %s", req.Path)
}

Expand Down
111 changes: 111 additions & 0 deletions pkg/quickwit/timestamp_infos.go
Original file line number Diff line number Diff line change
@@ -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)
}
Loading

0 comments on commit 121dbb4

Please sign in to comment.