grafana/pkg/tsdb/opentsdb/opentsdb.go

package opentsdb

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"path"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/grafana/grafana-plugin-sdk-go/backend"
	"github.com/grafana/grafana-plugin-sdk-go/backend/datasource"
	"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"
)

var logger = backend.NewLoggerWith("tsdb.opentsdb")

type Service struct {
	im instancemgmt.InstanceManager
}

func ProvideService(httpClientProvider *httpclient.Provider) *Service {
	return &Service{
		im: datasource.NewInstanceManager(newInstanceSettings(httpClientProvider)),
	}
}

type datasourceInfo struct {
	HTTPClient     *http.Client
	URL            string
	TSDBVersion    float32
	TSDBResolution int32
	LookupLimit    int32
}

type DsAccess string

type JSONData struct {
	TSDBVersion    float32 `json:"tsdbVersion"`
	TSDBResolution int32   `json:"tsdbResolution"`
	LookupLimit    int32   `json:"lookupLimit"`
}

type QueryModel struct {
	Metric               string                 `json:"metric"`
	Aggregator           string                 `json:"aggregator"`
	DownsampleInterval   string                 `json:"downsampleInterval"`
	DownsampleAggregator string                 `json:"downsampleAggregator"`
	DownsampleFillPolicy string                 `json:"downsampleFillPolicy"`
	DisableDownsampling  bool                   `json:"disableDownsampling"`
	Filters              []any                  `json:"filters"`
	Tags                 map[string]interface{} `json:"tags"`
	ShouldComputeRate    bool                   `json:"shouldComputeRate"`
	IsCounter            bool                   `json:"isCounter"`
	CounterMax           string                 `json:"counterMax"`
	CounterResetValue    string                 `json:"counterResetValue"`
	ExplicitTags         bool                   `json:"explicitTags"`
}

func newInstanceSettings(httpClientProvider *httpclient.Provider) datasource.InstanceFactoryFunc {
	return func(ctx context.Context, settings backend.DataSourceInstanceSettings) (instancemgmt.Instance, error) {
		opts, err := settings.HTTPClientOptions(ctx)
		if err != nil {
			return nil, err
		}

		client, err := httpClientProvider.New(opts)
		if err != nil {
			return nil, err
		}

		jsonData := JSONData{}
		err = json.Unmarshal(settings.JSONData, &jsonData)
		if err != nil {
			return nil, fmt.Errorf("error reading settings: %w", err)
		}

		model := &datasourceInfo{
			HTTPClient:     client,
			URL:            settings.URL,
			TSDBVersion:    jsonData.TSDBVersion,
			TSDBResolution: jsonData.TSDBResolution,
			LookupLimit:    jsonData.LookupLimit,
		}

		return model, nil
	}
}

func (s *Service) CheckHealth(ctx context.Context, req *backend.CheckHealthRequest) (*backend.CheckHealthResult, error) {
	logger := logger.FromContext(ctx)

	dsInfo, err := s.getDSInfo(ctx, req.PluginContext)
	if err != nil {
		return &backend.CheckHealthResult{
			Status:  backend.HealthStatusError,
			Message: err.Error(),
		}, nil
	}

	u, err := url.Parse(dsInfo.URL)
	if err != nil {
		return &backend.CheckHealthResult{
			Status:  backend.HealthStatusError,
			Message: err.Error(),
		}, nil
	}

	u.Path = path.Join(u.Path, "api/suggest")
	query := u.Query()
	query.Set("q", "cpu")
	query.Set("type", "metrics")
	u.RawQuery = query.Encode()

	httpReq, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
	if err != nil {
		return &backend.CheckHealthResult{
			Status:  backend.HealthStatusError,
			Message: err.Error(),
		}, nil
	}

	res, err := dsInfo.HTTPClient.Do(httpReq)
	if err != nil {
		return &backend.CheckHealthResult{
			Status:  backend.HealthStatusError,
			Message: err.Error(),
		}, nil
	}

	defer func() {
		if err := res.Body.Close(); err != nil {
			logger.Error("Failed to close response body", "error", err)
		}
	}()

	if res.StatusCode != 200 {
		return &backend.CheckHealthResult{
			Status:  backend.HealthStatusError,
			Message: fmt.Sprintf("OpenTSDB suggest endpoint returned status %d", res.StatusCode),
		}, nil
	}

	return &backend.CheckHealthResult{
		Status:  backend.HealthStatusOk,
		Message: "Data source is working",
	}, nil
}

func (s *Service) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
	logger := logger.FromContext(ctx)

	dsInfo, err := s.getDSInfo(ctx, req.PluginContext)
	if err != nil {
		return nil, err
	}

	result := backend.NewQueryDataResponse()

	for _, query := range req.Queries {
		// Build OpenTsdbQuery with per-query time range
		tsdbQuery := OpenTsdbQuery{
			Start: query.TimeRange.From.Unix(),
			End:   query.TimeRange.To.Unix(),
			Queries: []map[string]any{
				s.buildMetric(query),
			},
		}

		httpReq, err := s.createRequest(ctx, logger, dsInfo, tsdbQuery)
		if err != nil {
			return nil, err
		}

		httpRes, err := dsInfo.HTTPClient.Do(httpReq)
		if err != nil {
			return nil, err
		}

		defer func() {
			if cerr := httpRes.Body.Close(); cerr != nil {
				logger.Warn("failed to close response body", "error", cerr)
			}
		}()

		queryRes, err := s.parseResponse(logger, httpRes, query.RefID, dsInfo.TSDBVersion)
		if err != nil {
			return nil, err
		}

		// Attach parsed result for this query's RefID
		result.Responses[query.RefID] = queryRes.Responses[query.RefID]
	}

	return result, nil
}

func (s *Service) createRequest(ctx context.Context, logger log.Logger, dsInfo *datasourceInfo, data OpenTsdbQuery) (*http.Request, error) {
	u, err := url.Parse(dsInfo.URL)
	if err != nil {
		return nil, err
	}
	u.Path = path.Join(u.Path, "api/query")
	if dsInfo.TSDBVersion == 4 {
		queryParams := u.Query()
		queryParams.Set("arrays", "true")
		u.RawQuery = queryParams.Encode()
	}

	postData, err := json.Marshal(data)
	if err != nil {
		logger.Info("Failed marshaling data", "error", err)
		return nil, fmt.Errorf("failed to create request: %w", err)
	}

	req, err := http.NewRequestWithContext(ctx, http.MethodPost, u.String(), strings.NewReader(string(postData)))
	if err != nil {
		logger.Info("Failed to create request", "error", err)
		return nil, fmt.Errorf("failed to create request: %w", err)
	}

	req.Header.Set("Content-Type", "application/json")
	return req, nil
}

func createInitialFrame(val OpenTsdbCommon, length int, refID string) *data.Frame {
	labels := data.Labels{}
	for label, value := range val.Tags {
		labels[label] = value
	}

	tagKeys := make([]string, 0, len(val.Tags)+len(val.AggregateTags))
	for tagKey := range val.Tags {
		tagKeys = append(tagKeys, tagKey)
	}
	sort.Strings(tagKeys)
	tagKeys = append(tagKeys, val.AggregateTags...)

	frame := data.NewFrameOfFieldTypes(val.Metric, length, data.FieldTypeTime, data.FieldTypeFloat64)
	frame.Meta = &data.FrameMeta{
		Type:        data.FrameTypeTimeSeriesMulti,
		TypeVersion: data.FrameTypeVersion{0, 1},
		Custom:      map[string]any{"tagKeys": tagKeys},
	}
	frame.RefID = refID
	timeField := frame.Fields[0]
	timeField.Name = data.TimeSeriesTimeFieldName
	dataField := frame.Fields[1]
	dataField.Name = val.Metric
	dataField.Labels = labels

	return frame
}

// Parse response function for OpenTSDB version 2.4
func parseResponse24(responseData []OpenTsdbResponse24, refID string, frames data.Frames) data.Frames {
	for _, val := range responseData {
		frame := createInitialFrame(val.OpenTsdbCommon, len(val.DataPoints), refID)

		for i, point := range val.DataPoints {
			frame.SetRow(i, time.Unix(int64(point[0]), 0).UTC(), point[1])
		}

		frames = append(frames, frame)
	}

	return frames
}

// Parse response function for OpenTSDB versions < 2.4
func parseResponseLT24(responseData []OpenTsdbResponse, refID string, frames data.Frames) (data.Frames, error) {
	for _, val := range responseData {
		frame := createInitialFrame(val.OpenTsdbCommon, len(val.DataPoints), refID)

		// Order the timestamps in ascending order to avoid issues like https://github.com/grafana/grafana/issues/38729
		timestamps := make([]string, 0, len(val.DataPoints))
		for timestamp := range val.DataPoints {
			timestamps = append(timestamps, timestamp)
		}
		sort.Strings(timestamps)

		for i, timeString := range timestamps {
			timestamp, err := strconv.ParseInt(timeString, 10, 64)
			if err != nil {
				logger.Info("Failed to unmarshal opentsdb timestamp", "timestamp", timeString)
				return frames, err
			}
			frame.SetRow(i, time.Unix(timestamp, 0).UTC(), val.DataPoints[timeString])
		}

		frames = append(frames, frame)
	}

	return frames, nil
}

func (s *Service) parseResponse(logger log.Logger, res *http.Response, refID string, tsdbVersion float32) (*backend.QueryDataResponse, error) {
	resp := backend.NewQueryDataResponse()

	body, err := io.ReadAll(res.Body)
	if err != nil {
		return nil, err
	}
	defer func() {
		if err := res.Body.Close(); err != nil {
			logger.Warn("Failed to close response body", "err", err)
		}
	}()

	if res.StatusCode/100 != 2 {
		logger.Info("Request failed", "status", res.Status, "body", string(body))
		return nil, fmt.Errorf("request failed, status: %s", res.Status)
	}

	frames := data.Frames{}

	var responseData []OpenTsdbResponse
	var responseData24 []OpenTsdbResponse24
	if tsdbVersion == 4 {
		err = json.Unmarshal(body, &responseData24)
		if err != nil {
			logger.Info("Failed to unmarshal opentsdb response", "error", err, "status", res.Status, "body", string(body))
			return nil, err
		}

		frames = parseResponse24(responseData24, refID, frames)
	} else {
		err = json.Unmarshal(body, &responseData)
		if err != nil {
			logger.Info("Failed to unmarshal opentsdb response", "error", err, "status", res.Status, "body", string(body))
			return nil, err
		}

		frames, err = parseResponseLT24(responseData, refID, frames)
		if err != nil {
			return nil, err
		}
	}

	result := resp.Responses[refID]
	result.Frames = frames
	resp.Responses[refID] = result
	return resp, nil
}

func (s *Service) buildMetric(query backend.DataQuery) map[string]any {
	metric := make(map[string]any)

	var model QueryModel
	if err := json.Unmarshal(query.JSON, &model); err != nil {
		return nil
	}

	// Setting metric and aggregator
	metric["metric"] = model.Metric
	metric["aggregator"] = model.Aggregator

	// Setting downsampling options
	if !model.DisableDownsampling {
		downsampleInterval := model.DownsampleInterval
		if downsampleInterval == "" {
			if ms := query.Interval.Milliseconds(); ms > 0 {
				downsampleInterval = FormatDownsampleInterval(ms)
			} else {
				downsampleInterval = "1m"
			}
		} else if strings.Contains(downsampleInterval, ".") && strings.HasSuffix(downsampleInterval, "s") {
			if val, err := strconv.ParseFloat(strings.TrimSuffix(downsampleInterval, "s"), 64); err == nil {
				downsampleInterval = strconv.FormatInt(int64(val*1000), 10) + "ms"
			}
		}

		downsample := downsampleInterval + "-" + model.DownsampleAggregator
		if model.DownsampleFillPolicy != "" && model.DownsampleFillPolicy != "none" {
			metric["downsample"] = downsample + "-" + model.DownsampleFillPolicy
		} else {
			metric["downsample"] = downsample
		}
	}

	// Setting rate options
	if model.ShouldComputeRate {
		metric["rate"] = true
		rateOptions := make(map[string]any)
		rateOptions["counter"] = model.IsCounter

		var counterMax *float64
		if model.CounterMax != "" {
			if val, err := strconv.ParseFloat(model.CounterMax, 64); err == nil {
				counterMax = &val
			}
		}
		if counterMax != nil {
			rateOptions["counterMax"] = *counterMax
		}

		var counterResetValue *float64
		if model.CounterResetValue != "" {
			if val, err := strconv.ParseFloat(model.CounterResetValue, 64); err == nil {
				counterResetValue = &val
			}
		}
		if counterResetValue != nil {
			rateOptions["resetValue"] = *counterResetValue
		}

		if counterMax == nil && (counterResetValue == nil || *counterResetValue == 0) {
			rateOptions["dropResets"] = true
		}

		metric["rateOptions"] = rateOptions
	}

	// Setting tags
	if len(model.Tags) > 0 {
		metric["tags"] = model.Tags
	}

	// Setting filters
	if len(model.Filters) > 0 {
		metric["filters"] = model.Filters
	}

	if model.ExplicitTags {
		metric["explicitTags"] = true
	}

	return metric
}

func (s *Service) getDSInfo(ctx context.Context, pluginCtx backend.PluginContext) (*datasourceInfo, error) {
	i, err := s.im.Get(ctx, pluginCtx)
	if err != nil {
		return nil, err
	}

	instance, ok := i.(*datasourceInfo)
	if !ok {
		return nil, fmt.Errorf("failed to cast datasource info")
	}

	return instance, nil
}