connectors/grafana-plugin/pkg/plugin/plugin.go

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package plugin

import (
	"bytes"
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"math"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/grafana/grafana-plugin-sdk-go/backend"
	"github.com/grafana/grafana-plugin-sdk-go/backend/httpclient"
	"github.com/grafana/grafana-plugin-sdk-go/backend/instancemgmt"
	"github.com/grafana/grafana-plugin-sdk-go/backend/log"
	"github.com/grafana/grafana-plugin-sdk-go/data"
)

// Make sure IoTDBDatasource implements required interfaces. This is important to do
// since otherwise we will only get a not implemented error response from plugin in
// runtime. In this example datasource instance implements backend.QueryDataHandler,
// backend.CheckHealthHandler, backend.StreamHandler interfaces. Plugin should not
// implement all these interfaces - only those which are required for a particular task.
// For example if plugin does not need streaming functionality then you are free to remove
// methods that implement backend.StreamHandler. Implementing instancemgmt.InstanceDisposer
// is useful to clean up resources used by previous datasource instance when a new datasource
// instance created upon datasource settings changed.
var (
	_ backend.QueryDataHandler    = (*IoTDBDataSource)(nil)
	_ backend.CheckHealthHandler  = (*IoTDBDataSource)(nil)
	_ backend.CallResourceHandler = (*IoTDBDataSource)(nil)
)

// ApacheIoTDBDatasource creates a new datasource instance.
func ApacheIoTDBDatasource(ctx context.Context, d backend.DataSourceInstanceSettings) (instancemgmt.Instance, error) {
	var dm dataSourceModel
	if err := json.Unmarshal(d.JSONData, &dm); err != nil {
		return nil, err
	}
	ops, err := d.HTTPClientOptions(ctx)
	if err != nil {
		return nil, fmt.Errorf("http client options: %w", err)
	}
	httpClient, err := httpclient.New(ops)
	if err != nil {
		return nil, fmt.Errorf("new httpclient error: %w", err)
	}
	var authorization = ""
	if password, exists := d.DecryptedSecureJSONData["password"]; exists {
		authorization = "Basic " + base64.StdEncoding.EncodeToString([]byte(dm.Username+":"+password))
	}
	return &IoTDBDataSource{CallResourceHandler: iotdbResourceHandler(authorization, httpClient), Username: dm.Username, Ulr: dm.Url, httpClient: httpClient}, nil
}

// SampleDatasource is an example datasource which can respond to data queries, reports
// its health and has streaming skills.
type IoTDBDataSource struct {
	backend.CallResourceHandler
	Username   string
	Ulr        string
	httpClient *http.Client
}

// Dispose here tells plugin SDK that plugin wants to clean up resources when a new instance
// created. As soon as datasource settings change detected by SDK old datasource instance will
// be disposed and a new one will be created using ApacheIoTDBDatasource factory function.
func (d *IoTDBDataSource) Dispose() {
	// Clean up datasource instance resources.
	d.httpClient.CloseIdleConnections()
}

// QueryData handles multiple queries and returns multiple responses.
// req contains the queries []DataQuery (where each query contains RefID as a unique identifier).
// The QueryDataResponse contains a map of RefID to the response for each query, and each response
// contains Frames ([]*Frame).
func (d *IoTDBDataSource) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
	// create response struct
	response := backend.NewQueryDataResponse()

	// loop over queries and execute them individually.
	for _, q := range req.Queries {
		res := d.query(ctx, req.PluginContext, q)

		// save the response in a hashmap
		// based on with RefID as identifier
		response.Responses[q.RefID] = res
	}

	return response, nil
}

type dataSourceModel struct {
	Username string `json:"username"`
	Url      string `json:"url"`
}

type groupBy struct {
	GroupByLevel     string `json:"groupByLevel"`
	SamplingInterval string `json:"samplingInterval"`
	Step             string `json:"step"`
}

type queryParam struct {
	Expression   []string `json:"expression"`
	PrefixPath   []string `json:"prefixPath"`
	StartTime    int64    `json:"startTime"`
	EndTime      int64    `json:"endTime"`
	Condition    string   `json:"condition"`
	Control      string   `json:"control"`
	SqlType      string   `json:"sqlType"`
	Paths        []string `json:"paths"`
	AggregateFun string   `json:"aggregateFun"`
	FillClauses  string   `json:"fillClauses"`
	GroupBy      groupBy  `json:"groupBy"`
	Hide         bool     `json:"hide"`
}

type QueryDataReq struct {
	Expression []string `json:"expression"`
	PrefixPath []string `json:"prefixPath"`
	StartTime  int64    `json:"startTime"`
	EndTime    int64    `json:"endTime"`
	Condition  string   `json:"condition"`
	Control    string   `json:"control"`
}

type QueryDataResponse struct {
	Expressions []string        `json:"expressions"`
	Timestamps  []int64         `json:"timestamps"`
	Values      [][]interface{} `json:"values"`
	ColumnNames interface{}     `json:"columnNames"`
	Code        int32           `json:"code"`
	Message     string          `json:"message"`
}

type loginStatus struct {
	Code    int    `json:"code"`
	Message string `json:"message"`
}

func NewQueryDataReq(expression []string, prefixPath []string, startTime int64, endTime int64, condition string, control string) *QueryDataReq {
	return &QueryDataReq{Expression: expression, PrefixPath: prefixPath, StartTime: startTime, EndTime: endTime, Condition: condition, Control: control}
}

func verifyQuery(query backend.DataQuery) (qp *queryParam, errMsg string) {

	err := json.Unmarshal(query.JSON, &qp)
	if err != nil {
		return nil, ""
	}

	if qp.Hide {
		return nil, "none"
	}

	if qp.SqlType == "SQL: Drop-down List" {
		if len(qp.Paths) < 1 {
			return nil, "Input error, please select TIME-SERIES"
		}
		if qp.GroupBy.SamplingInterval != "" && qp.AggregateFun == "" {
			return nil, "Input error, please select FUNCTION when SAMPLING INTERVAL has a value"
		}
	} else if qp.SqlType == "SQL: Full Customized" {
		if len(qp.Expression) == 0 {
			return nil, "Input error, SELECT is required"
		}

		for i := 0; i < len(qp.Expression); i++ {
			if qp.Expression[i] == "" {
				return nil, "Input error, SELECT is required"
			}

		}
		if len(qp.PrefixPath) == 0 {
			return nil, "Input error, FROM is required"
		}
		for i := 0; i < len(qp.PrefixPath); i++ {
			if qp.PrefixPath[i] == "" {
				return nil, "Input error, FROM is required"
			}
		}
	} else {
		return nil, "none"
	}

	return qp, ""
}

func (d *IoTDBDataSource) query(cxt context.Context, pCtx backend.PluginContext, query backend.DataQuery) backend.DataResponse {
	response := backend.DataResponse{}

	instanceSettings := pCtx.DataSourceInstanceSettings
	var authorization = ""
	if password, exists := instanceSettings.DecryptedSecureJSONData["password"]; exists {
		// Use the decrypted API key.
		authorization = "Basic " + base64.StdEncoding.EncodeToString([]byte(d.Username+":"+password))
	}

	// Unmarshal the JSON into our queryModel.

	var qdReq QueryDataReq
	qp, msg := verifyQuery(query)
	if msg == "none" {
		return response
	} else if msg != "" {
		response.Error = errors.New(msg)
		return response
	}

	qp.StartTime = query.TimeRange.From.UnixNano() / 1000000
	qp.EndTime = query.TimeRange.To.UnixNano() / 1000000

	if qp.SqlType == "SQL: Drop-down List" {
		qp.Control = ""
		var expressions []string = qp.Paths[len(qp.Paths)-1:]
		var paths []string = qp.Paths[0 : len(qp.Paths)-1]
		path := "root." + strings.Join(paths, ".")
		var prefixPaths = []string{path}
		if qp.AggregateFun != "" {
			expressions[0] = qp.AggregateFun + "(" + expressions[0] + ")"
		}
		if qp.GroupBy.SamplingInterval != "" && qp.GroupBy.Step == "" {
			qp.Control += " group by([" + strconv.FormatInt(qp.StartTime, 10) + "," + strconv.FormatInt(qp.EndTime, 10) + ")," + qp.GroupBy.SamplingInterval + ")"
		}
		if qp.GroupBy.SamplingInterval != "" && qp.GroupBy.Step != "" {
			qp.Control += " group by([" + strconv.FormatInt(qp.StartTime, 10) + "," + strconv.FormatInt(qp.EndTime, 10) + ")," + qp.GroupBy.SamplingInterval + "," + qp.GroupBy.Step + ")"
		}
		if qp.GroupBy.GroupByLevel != "" {
			qp.Control += " " + qp.GroupBy.GroupByLevel
		}
		if qp.FillClauses != "" {
			qp.Control += " fill" + qp.FillClauses
		}
		qdReq = *NewQueryDataReq(expressions, prefixPaths, qp.StartTime, qp.EndTime, qp.Condition, qp.Control)
	} else if qp.SqlType == "SQL: Full Customized" {
		qdReq = *NewQueryDataReq(qp.Expression, qp.PrefixPath, qp.StartTime, qp.EndTime, qp.Condition, qp.Control)
	} else {
		return response
	}
	qpJson, _ := json.Marshal(qdReq)
	reader := bytes.NewReader(qpJson)

	var dataSourceUrl = DataSourceUrlHandler(d.Ulr)

	request, _ := http.NewRequest(http.MethodPost, dataSourceUrl+"/grafana/v1/query/expression", reader)
	request.Header.Set("Content-Type", "application/json")
	request.Header.Add("Authorization", authorization)

	rsp, _ := d.httpClient.Do(request)
	body, err := io.ReadAll(rsp.Body)
	if err != nil {
		response.Error = errors.New("Data source is not working properly")
		log.DefaultLogger.Error("Data source is not working properly", err)
		return response
	}

	var queryDataResp QueryDataResponse
	err = json.Unmarshal(body, &queryDataResp)
	if err != nil {
		response.Error = errors.New("Parsing JSON error")
		log.DefaultLogger.Error("Parsing JSON error", err)
		return response
	}

	defer rsp.Body.Close()
	if queryDataResp.Code > 0 {
		response.Error = errors.New(queryDataResp.Message)
		log.DefaultLogger.Error(queryDataResp.Message)
		return response

	}
	// create data frame response.
	frame := data.NewFrame("response")
	for i := 0; i < len(queryDataResp.Expressions); i++ {
		if queryDataResp.Timestamps != nil && len(queryDataResp.Timestamps) > 0 {
			times := make([]time.Time, len(queryDataResp.Timestamps))
			for c := 0; c < len(queryDataResp.Timestamps); c++ {
				times[c] = time.Unix(0, queryDataResp.Timestamps[c]*1000000)
			}
			if queryDataResp.Values != nil && len(queryDataResp.Values) > 0 {
				values := recoverType(queryDataResp.Values[i])
				frame.Fields = append(frame.Fields,
					data.NewField("time", nil, times),
					data.NewField(queryDataResp.Expressions[i], nil, values),
				)
			}
		} else {
			if queryDataResp.Values != nil && len(queryDataResp.Values) > 0 {
				values := recoverType(queryDataResp.Values[i])
				frame.Fields = append(frame.Fields,
					data.NewField(queryDataResp.Expressions[i], nil, values),
				)
			}
		}

	}

	response.Frames = append(response.Frames, frame)
	return response
}

func recoverType(m []interface{}) interface{} {
	if len(m) > 0 {
		switch m[0].(type) {
		case float64:
			tmp := make([]float64, len(m))
			for i := range m {
				if m[i] == nil {
					tmp[i] = math.NaN()
				} else {
					tmp[i] = m[i].(float64)
				}
			}
			return tmp
		case string:
			tmp := make([]string, len(m))
			for i := range m {
				tmp[i] = m[i].(string)
			}
			return tmp
		case bool:
			tmp := make([]float64, len(m))
			for i := range m {
				if m[i] == nil {
					tmp[i] = math.NaN()
				} else if m[i].(bool) {
					tmp[i] = 1
				} else {
					tmp[i] = 0
				}
			}
			return tmp
		default:
			tmp := make([]float64, len(m))
			for i := range m {
				if m[i] == nil {
					tmp[i] = math.NaN()
				} else {
					tmp[i] = m[i].(float64)
				}
			}
			return tmp
		}
	} else {
		return make([]float64, 0)
	}
}

// Whether the last character of the URL for processing datasource configuration is "/"
func DataSourceUrlHandler(url string) string {
	var lastCharacter = url[len(url)-1:]
	if lastCharacter == "/" {
		url = url[0 : len(url)-1]
	}
	return url
}

// CheckHealth handles health checks sent from Grafana to the plugin.
// The main use case for these health checks is the test button on the
// datasource configuration page which allows users to verify that
// a datasource is working as expected.
func (d *IoTDBDataSource) CheckHealth(_ context.Context, req *backend.CheckHealthRequest) (*backend.CheckHealthResult, error) {

	var status = backend.HealthStatusOk
	var message = "Data source is working"

	loginStatus, err := d.IoTDBLogin(req)

	if err != nil {
		log.DefaultLogger.Error("Parsing JSON error", err)
		status = backend.HealthStatusError
		message = fmt.Sprint("Failed to connect to iotdb service.", err.Error())
	} else if loginStatus.Code != 200 {
		status = backend.HealthStatusError
		message = loginStatus.Message
	}
	return &backend.CheckHealthResult{
		Status:  status,
		Message: message,
	}, nil
}

func (d *IoTDBDataSource) IoTDBLogin(req *backend.CheckHealthRequest) (*loginStatus, error) {
	instanceSettings := req.PluginContext.DataSourceInstanceSettings
	var authorization = ""
	if password, exists := instanceSettings.DecryptedSecureJSONData["password"]; exists {
		authorization = "Basic " + base64.StdEncoding.EncodeToString([]byte(d.Username+":"+password))
	}
	var dataSourceUrl = DataSourceUrlHandler(d.Ulr)
	request, err := http.NewRequest(http.MethodGet, dataSourceUrl+"/grafana/v1/login", nil)
	if err != nil {
		log.DefaultLogger.Error("Error creating NewRequest", err.Error())
		return nil, err
	}
	request.Header.Add("Authorization", authorization)
	response, err := d.httpClient.Do(request)
	if err != nil {
		log.DefaultLogger.Error("Failed to connect to iotdb service", err.Error())
		return nil, err
	}
	body, err := io.ReadAll(response.Body)
	if err != nil {
		log.DefaultLogger.Error("Failed to get iotdb service data", err.Error())
		return nil, err
	}
	var loginStatus loginStatus
	err = json.Unmarshal(body, &loginStatus)
	if err != nil {
		log.DefaultLogger.Error("Parsing JSON error", err)
		return nil, err
	}

	defer response.Body.Close()
	return &loginStatus, nil
}