public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

OpenTSDB: Support all query options in the backend (#114822)

Browse Source 
* update backend to support all query options

* update backend tests

* move formatDownsampleInterval to utils
This commit is contained in:
Gareth
2025-12-09 17:18:06 +09:00
committed by GitHub
parent 16f5eab786
commit 8eb25a0164
5 changed files with 287 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/tsdb/opentsdb/opentsdb.go
+28 -3
View File
@@ -62,6 +62,7 @@ type QueryModel struct {
IsCounter bool `json:"isCounter"`
CounterMax string `json:"counterMax"`
CounterResetValue string `json:"counterResetValue"`
ExplicitTags bool `json:"explicitTags"`
}
func newInstanceSettings(httpClientProvider *httpclient.Provider) datasource.InstanceFactoryFunc {
@@ -236,8 +237,19 @@ func createInitialFrame(val OpenTsdbCommon, length int, refID string) *data.Fram
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}}
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
@@ -355,10 +367,19 @@ func (s *Service) buildMetric(query backend.DataQuery) map[string]any {
if !model.DisableDownsampling {
downsampleInterval := model.DownsampleInterval
if downsampleInterval == "" {
downsampleInterval = "1m" // default value for blank
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 != "none" {
if model.DownsampleFillPolicy != "" && model.DownsampleFillPolicy != "none" {
metric["downsample"] = downsample + "-" + model.DownsampleFillPolicy
} else {
metric["downsample"] = downsample
@@ -408,6 +429,10 @@ func (s *Service) buildMetric(query backend.DataQuery) map[string]any {
metric["filters"] = model.Filters
}
if model.ExplicitTags {
metric["explicitTags"] = true
}
return metric
}
pkg/tsdb/opentsdb/opentsdb_test.go
+201
View File
@@ -70,6 +70,164 @@ func TestCheckHealth(t *testing.T) {
}
}
func TestBuildMetric(t *testing.T) {
service := &Service{}
t.Run("Metric with no downsampleInterval should use query interval", func(t *testing.T) {
query := backend.DataQuery{
JSON: []byte(`
{
"metric": "cpu.average.percent",
"aggregator": "avg",
"disableDownsampling": false,
"downsampleInterval": "",
"downsampleAggregator": "avg",
"downsampleFillPolicy": "none"
}`,
),
Interval: 30 * time.Second,
}
metric := service.buildMetric(query)
require.Equal(t, "30s-avg", metric["downsample"], "should use query interval formatted as seconds")
})
t.Run("Metric with downsampleInterval converts decimal seconds to milliseconds", func(t *testing.T) {
query := backend.DataQuery{
JSON: []byte(`
{
"metric": "cpu.average.percent",
"aggregator": "avg",
"disableDownsampling": false,
"downsampleInterval": "0.5s",
"downsampleAggregator": "avg",
"downsampleFillPolicy": "none"
}`,
),
}
metric := service.buildMetric(query)
require.Equal(t, "500ms-avg", metric["downsample"], "should convert 0.5s to 500ms")
})
t.Run("Metric with no downsampleInterval uses milliseconds for sub-second query interval", func(t *testing.T) {
query := backend.DataQuery{
JSON: []byte(`
{
"metric": "cpu.average.percent",
"aggregator": "avg",
"disableDownsampling": false,
"downsampleInterval": "",
"downsampleAggregator": "avg",
"downsampleFillPolicy": "none"
}`,
),
Interval: 500 * time.Millisecond,
}
metric := service.buildMetric(query)
require.Equal(t, "500ms-avg", metric["downsample"], "should use query interval formatted as milliseconds")
})
t.Run("Metric with no downsampleInterval uses minutes for longer intervals", func(t *testing.T) {
query := backend.DataQuery{
JSON: []byte(`
{
"metric": "cpu.average.percent",
"aggregator": "avg",
"disableDownsampling": false,
"downsampleInterval": "",
"downsampleAggregator": "sum",
"downsampleFillPolicy": "none"
}`,
),
Interval: 5 * time.Minute,
}
metric := service.buildMetric(query)
require.Equal(t, "5m-sum", metric["downsample"], "should use query interval formatted as minutes")
})
t.Run("Metric with no downsampleInterval uses hours for multi-hour intervals", func(t *testing.T) {
query := backend.DataQuery{
JSON: []byte(`
{
"metric": "cpu.average.percent",
"aggregator": "avg",
"disableDownsampling": false,
"downsampleInterval": "",
"downsampleAggregator": "max",
"downsampleFillPolicy": "none"
}`,
),
Interval: 2 * time.Hour,
}
metric := service.buildMetric(query)
require.Equal(t, "2h-max", metric["downsample"], "should use query interval formatted as hours")
})
t.Run("Metric with no downsampleInterval uses days for multi-day intervals", func(t *testing.T) {
query := backend.DataQuery{
JSON: []byte(`
{
"metric": "cpu.average.percent",
"aggregator": "avg",
"disableDownsampling": false,
"downsampleInterval": "",
"downsampleAggregator": "min",
"downsampleFillPolicy": "none"
}`,
),
Interval: 48 * time.Hour,
}
metric := service.buildMetric(query)
require.Equal(t, "2d-min", metric["downsample"], "should use query interval formatted as days")
})
t.Run("Build metric with explicitTags enabled", func(t *testing.T) {
query := backend.DataQuery{
JSON: []byte(`
{
"metric": "cpu.average.percent",
"aggregator": "avg",
"disableDownsampling": true,
"explicitTags": true,
"tags": {
"host": "server01"
}
}`,
),
}
metric := service.buildMetric(query)
require.True(t, metric["explicitTags"].(bool), "explicitTags should be true")
metricTags := metric["tags"].(map[string]any)
require.Equal(t, "server01", metricTags["host"])
})
t.Run("Build metric with explicitTags disabled does not include explicitTags", func(t *testing.T) {
query := backend.DataQuery{
JSON: []byte(`
{
"metric": "cpu.average.percent",
"aggregator": "avg",
"disableDownsampling": true,
"explicitTags": false,
"tags": {
"host": "server01"
}
}`,
),
}
metric := service.buildMetric(query)
require.Nil(t, metric["explicitTags"], "explicitTags should not be present when false")
})
}
func TestOpenTsdbExecutor(t *testing.T) {
service := &Service{}
@@ -119,6 +277,7 @@ func TestOpenTsdbExecutor(t *testing.T) {
testFrame.Meta = &data.FrameMeta{
Type: data.FrameTypeTimeSeriesMulti,
TypeVersion: data.FrameTypeVersion{0, 1},
Custom: map[string]any{"tagKeys": []string{"app", "env"}},
}
testFrame.RefID = "A"
tsdbVersion := float32(4)
@@ -160,6 +319,7 @@ func TestOpenTsdbExecutor(t *testing.T) {
testFrame.Meta = &data.FrameMeta{
Type: data.FrameTypeTimeSeriesMulti,
TypeVersion: data.FrameTypeVersion{0, 1},
Custom: map[string]any{"tagKeys": []string{"app", "env"}},
}
testFrame.RefID = "A"
tsdbVersion := float32(3)
@@ -232,6 +392,7 @@ func TestOpenTsdbExecutor(t *testing.T) {
testFrame.Meta = &data.FrameMeta{
Type: data.FrameTypeTimeSeriesMulti,
TypeVersion: data.FrameTypeVersion{0, 1},
Custom: map[string]any{"tagKeys": []string{"app", "env"}},
}
testFrame.RefID = "A"
tsdbVersion := float32(3)
@@ -275,6 +436,7 @@ func TestOpenTsdbExecutor(t *testing.T) {
testFrame.Meta = &data.FrameMeta{
Type: data.FrameTypeTimeSeriesMulti,
TypeVersion: data.FrameTypeVersion{0, 1},
Custom: map[string]any{"tagKeys": []string{"app", "env"}},
}
testFrame.RefID = myRefid
@@ -290,6 +452,45 @@ func TestOpenTsdbExecutor(t *testing.T) {
}
})
t.Run("tagKeys are returned sorted alphabetically in frame metadata", func(t *testing.T) {
response := `
[
{
"metric": "cpu.usage",
"dps": [
[1405544146, 75.5]
],
"tags" : {
"zone": "us-east-1",
"host": "server01",
"app": "api",
"env": "production"
}
}
]`
tsdbVersion := float32(4)
resp := http.Response{Body: io.NopCloser(strings.NewReader(response))}
resp.StatusCode = 200
result, err := service.parseResponse(logger, &resp, "A", tsdbVersion)
require.NoError(t, err)
frame := result.Responses["A"].Frames[0]
require.NotNil(t, frame.Meta, "frame metadata should not be nil")
require.NotNil(t, frame.Meta.Custom, "frame custom metadata should not be nil")
customMeta, ok := frame.Meta.Custom.(map[string]any)
require.True(t, ok, "custom metadata should be a map")
tagKeys, ok := customMeta["tagKeys"].([]string)
require.True(t, ok, "tagKeys should be present and be a string slice")
require.Len(t, tagKeys, 4, "should have 4 tag keys")
expectedTagKeys := []string{"app", "env", "host", "zone"}
require.Equal(t, expectedTagKeys, tagKeys, "tagKeys should be sorted alphabetically")
})
t.Run("Build metric with downsampling enabled", func(t *testing.T) {
query := backend.DataQuery{
JSON: []byte(`
pkg/tsdb/opentsdb/types.go
+3 -2
View File
@@ -7,8 +7,9 @@ type OpenTsdbQuery struct {
}
type OpenTsdbCommon struct {
Metric string `json:"metric"`
Tags map[string]string `json:"tags"`
Metric string `json:"metric"`
Tags map[string]string `json:"tags"`
AggregateTags []string `json:"aggregateTags"`
}
type OpenTsdbResponse struct {
pkg/tsdb/opentsdb/utils.go
+31
View File
@@ -0,0 +1,31 @@
package opentsdb
import (
"strconv"
"time"
)
func FormatDownsampleInterval(ms int64) string {
duration := time.Duration(ms) * time.Millisecond
seconds := int64(duration / time.Second)
if seconds < 60 {
if seconds < 1 {
return strconv.FormatInt(ms, 10) + "ms"
}
return strconv.FormatInt(seconds, 10) + "s"
}
minutes := int64(duration / time.Minute)
if minutes < 60 {
return strconv.FormatInt(minutes, 10) + "m"
}
hours := int64(duration / time.Hour)
if hours < 24 {
return strconv.FormatInt(hours, 10) + "h"
}
days := int64(duration / (24 * time.Hour))
return strconv.FormatInt(days, 10) + "d"
}
public/app/plugins/datasource/opentsdb/datasource.ts
+24
View File
@@ -18,6 +18,7 @@ import { catchError, map } from 'rxjs/operators';
import {
AnnotationEvent,
DataFrame,
DataQueryRequest,
DataQueryResponse,
dateMath,
@@ -78,6 +79,20 @@ export default class OpenTsDatasource extends DataSourceWithBackend<OpenTsdbQuer
// Called once per panel (graph)
query(options: DataQueryRequest<OpenTsdbQuery>): Observable<DataQueryResponse> {
if (config.featureToggles.opentsdbBackendMigration) {
const hasValidTargets = options.targets.some((target) => target.metric && !target.hide);
if (!hasValidTargets) {
return of({ data: [] });
}
return super.query(options).pipe(
map((response) => {
this._saveTagKeysFromFrames(response.data);
return response;
})
);
}
// migrate annotations
if (options.targets.some((target: OpenTsdbQuery) => target.fromAnnotations)) {
const streams: Array<Observable<DataQueryResponse>> = [];
@@ -265,6 +280,15 @@ export default class OpenTsDatasource extends DataSourceWithBackend<OpenTsdbQuer
this.tagKeys[metricData.metric] = tagKeys;
}
_saveTagKeysFromFrames(frames: DataFrame[]) {
for (const frame of frames) {
const tagKeys = frame.meta?.custom?.tagKeys;
if (frame.name && tagKeys) {
this.tagKeys[frame.name] = tagKeys;
}
}
}
_performSuggestQuery(query: string, type: string) {
return this._get('/api/suggest', { type, q: query, max: this.lookupLimit }).pipe(
map((result) => {