Backend Plugins: Convert test data source to use SDK contracts (#29916)

Converts the core testdata data source to use the SDK contracts and by that 
implementing a backend plugin in core Grafana in similar manner as an external one.

Co-authored-by: Will Browne <will.browne@grafana.com>
Co-authored-by: Marcus Efraimsson <marefr@users.noreply.github.com>
Co-authored-by: Ryan McKinley <ryantxu@gmail.com>

pkg/tsdb/testdatasource/resource_handler.go

@@ -0,0 +1,157 @@
+package testdatasource
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"sort"
+	"strconv"
+	"time"
+
+	"github.com/grafana/grafana/pkg/infra/log"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend/resource/httpadapter"
+)
+
+func (p *testDataPlugin) registerRoutes(mux *http.ServeMux) {
+	mux.HandleFunc("/", p.testGetHandler)
+	mux.HandleFunc("/scenarios", p.getScenariosHandler)
+	mux.HandleFunc("/stream", p.testStreamHandler)
+	mux.Handle("/test", createJSONHandler(p.logger))
+	mux.Handle("/test/json", createJSONHandler(p.logger))
+	mux.HandleFunc("/boom", p.testPanicHandler)
+}
+
+func (p *testDataPlugin) testGetHandler(rw http.ResponseWriter, req *http.Request) {
+	p.logger.Debug("Received resource call", "url", req.URL.String(), "method", req.Method)
+
+	if req.Method != http.MethodGet {
+		return
+	}
+
+	if _, err := rw.Write([]byte("Hello world from test datasource!")); err != nil {
+		p.logger.Error("Failed to write response", "error", err)
+		return
+	}
+	rw.WriteHeader(http.StatusOK)
+}
+
+func (p *testDataPlugin) getScenariosHandler(rw http.ResponseWriter, req *http.Request) {
+	result := make([]interface{}, 0)
+
+	scenarioIds := make([]string, 0)
+	for id := range p.scenarios {
+		scenarioIds = append(scenarioIds, id)
+	}
+	sort.Strings(scenarioIds)
+
+	for _, scenarioID := range scenarioIds {
+		scenario := p.scenarios[scenarioID]
+		result = append(result, map[string]interface{}{
+			"id":          scenario.ID,
+			"name":        scenario.Name,
+			"description": scenario.Description,
+			"stringInput": scenario.StringInput,
+		})
+	}
+
+	bytes, err := json.Marshal(&result)
+	if err != nil {
+		p.logger.Error("Failed to marshal response body to JSON", "error", err)
+	}
+
+	rw.Header().Set("Content-Type", "application/json")
+	rw.WriteHeader(http.StatusOK)
+	if _, err := rw.Write(bytes); err != nil {
+		p.logger.Error("Failed to write response", "error", err)
+	}
+}
+
+func (p *testDataPlugin) testStreamHandler(rw http.ResponseWriter, req *http.Request) {
+	p.logger.Debug("Received resource call", "url", req.URL.String(), "method", req.Method)
+
+	if req.Method != http.MethodGet {
+		return
+	}
+
+	count := 10
+	countstr := req.URL.Query().Get("count")
+	if countstr != "" {
+		if i, err := strconv.Atoi(countstr); err == nil {
+			count = i
+		}
+	}
+
+	sleep := req.URL.Query().Get("sleep")
+	sleepDuration, err := time.ParseDuration(sleep)
+	if err != nil {
+		sleepDuration = time.Millisecond
+	}
+
+	rw.Header().Set("Content-Type", "text/plain")
+	rw.WriteHeader(http.StatusOK)
+
+	for i := 1; i <= count; i++ {
+		if _, err := io.WriteString(rw, fmt.Sprintf("Message #%d", i)); err != nil {
+			p.logger.Error("Failed to write response", "error", err)
+			return
+		}
+		rw.(http.Flusher).Flush()
+		time.Sleep(sleepDuration)
+	}
+}
+
+func createJSONHandler(logger log.Logger) http.Handler {
+	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		logger.Debug("Received resource call", "url", req.URL.String(), "method", req.Method)
+
+		var reqData map[string]interface{}
+		if req.Body != nil {
+			defer func() {
+				if err := req.Body.Close(); err != nil {
+					logger.Warn("Failed to close response body", "err", err)
+				}
+			}()
+			b, err := ioutil.ReadAll(req.Body)
+			if err != nil {
+				logger.Error("Failed to read request body to bytes", "error", err)
+			} else {
+				err := json.Unmarshal(b, &reqData)
+				if err != nil {
+					logger.Error("Failed to unmarshal request body to JSON", "error", err)
+				}
+
+				logger.Debug("Received resource call body", "body", reqData)
+			}
+		}
+
+		config := httpadapter.PluginConfigFromContext(req.Context())
+
+		data := map[string]interface{}{
+			"message": "Hello world from test datasource!",
+			"request": map[string]interface{}{
+				"method":  req.Method,
+				"url":     req.URL,
+				"headers": req.Header,
+				"body":    reqData,
+				"config":  config,
+			},
+		}
+		bytes, err := json.Marshal(&data)
+		if err != nil {
+			logger.Error("Failed to marshal response body to JSON", "error", err)
+		}
+
+		rw.Header().Set("Content-Type", "application/json")
+		rw.WriteHeader(http.StatusOK)
+		if _, err := rw.Write(bytes); err != nil {
+			logger.Error("Failed to write response", "error", err)
+		}
+	})
+}
+
+func (p *testDataPlugin) testPanicHandler(rw http.ResponseWriter, req *http.Request) {
+	panic("BOOM")
+}

pkg/tsdb/testdatasource/scenarios_test.go

@@ -1,55 +1,115 @@
 package testdatasource
 
 import (
+	"context"
+	"fmt"
 	"testing"
 	"time"
 
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana-plugin-sdk-go/data"
 	"github.com/grafana/grafana/pkg/components/simplejson"
 	"github.com/grafana/grafana/pkg/tsdb"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 func TestTestdataScenarios(t *testing.T) {
+	p := &testDataPlugin{}
+
 	t.Run("random walk ", func(t *testing.T) {
-		scenario := ScenarioRegistry["random_walk"]
-
 		t.Run("Should start at the requested value", func(t *testing.T) {
-			req := &tsdb.TsdbQuery{
-				TimeRange: tsdb.NewFakeTimeRange("5m", "now", time.Now()),
-				Queries: []*tsdb.Query{
-					{RefId: "A", IntervalMs: 100, MaxDataPoints: 100, Model: simplejson.New()},
+			timeRange := tsdb.NewFakeTimeRange("5m", "now", time.Now())
+
+			model := simplejson.New()
+			model.Set("startValue", 1.234)
+			modelBytes, err := model.MarshalJSON()
+			require.NoError(t, err)
+
+			query := backend.DataQuery{
+				RefID: "A",
+				TimeRange: backend.TimeRange{
+					From: timeRange.MustGetFrom(),
+					To:   timeRange.MustGetTo(),
 				},
+				Interval:      100 * time.Millisecond,
+				MaxDataPoints: 100,
+				JSON:          modelBytes,
 			}
-			query := req.Queries[0]
-			query.Model.Set("startValue", 1.234)
 
-			result := scenario.Handler(req.Queries[0], req)
-			require.NotNil(t, result.Series)
+			req := &backend.QueryDataRequest{
+				PluginContext: backend.PluginContext{},
+				Queries:       []backend.DataQuery{query},
+			}
 
-			points := result.Series[0].Points
-			require.Equal(t, 1.234, points[0][0].Float64)
+			resp, err := p.handleRandomWalkScenario(context.Background(), req)
+			require.NoError(t, err)
+			require.NotNil(t, resp)
+
+			dResp, exists := resp.Responses[query.RefID]
+			require.True(t, exists)
+			require.NoError(t, dResp.Error)
+
+			require.Len(t, dResp.Frames, 1)
+			frame := dResp.Frames[0]
+			require.Len(t, frame.Fields, 2)
+			require.Equal(t, "time", frame.Fields[0].Name)
+			require.Equal(t, "A-series", frame.Fields[1].Name)
+			val, ok := frame.Fields[1].ConcreteAt(0)
+			require.True(t, ok)
+			require.Equal(t, 1.234, val)
 		})
 	})
 
 	t.Run("random walk table", func(t *testing.T) {
-		scenario := ScenarioRegistry["random_walk_table"]
-
 		t.Run("Should return a table that looks like value/min/max", func(t *testing.T) {
-			req := &tsdb.TsdbQuery{
-				TimeRange: tsdb.NewFakeTimeRange("5m", "now", time.Now()),
-				Queries: []*tsdb.Query{
-					{RefId: "A", IntervalMs: 100, MaxDataPoints: 100, Model: simplejson.New()},
+			timeRange := tsdb.NewFakeTimeRange("5m", "now", time.Now())
+
+			model := simplejson.New()
+			modelBytes, err := model.MarshalJSON()
+			require.NoError(t, err)
+
+			query := backend.DataQuery{
+				RefID: "A",
+				TimeRange: backend.TimeRange{
+					From: timeRange.MustGetFrom(),
+					To:   timeRange.MustGetTo(),
 				},
+				Interval:      100 * time.Millisecond,
+				MaxDataPoints: 100,
+				JSON:          modelBytes,
 			}
 
-			result := scenario.Handler(req.Queries[0], req)
-			table := result.Tables[0]
+			req := &backend.QueryDataRequest{
+				PluginContext: backend.PluginContext{},
+				Queries:       []backend.DataQuery{query},
+			}
 
-			require.Greater(t, len(table.Rows), 50)
-			for _, row := range table.Rows {
-				value := row[1]
-				min := row[2]
-				max := row[3]
+			resp, err := p.handleRandomWalkTableScenario(context.Background(), req)
+			require.NoError(t, err)
+			require.NotNil(t, resp)
+
+			dResp, exists := resp.Responses[query.RefID]
+			require.True(t, exists)
+			require.NoError(t, dResp.Error)
+
+			require.Len(t, dResp.Frames, 1)
+			frame := dResp.Frames[0]
+			require.Greater(t, frame.Rows(), 50)
+			require.Len(t, frame.Fields, 5)
+			require.Equal(t, "Time", frame.Fields[0].Name)
+			require.Equal(t, "Value", frame.Fields[1].Name)
+			require.Equal(t, "Min", frame.Fields[2].Name)
+			require.Equal(t, "Max", frame.Fields[3].Name)
+			require.Equal(t, "Info", frame.Fields[4].Name)
+
+			for i := 0; i < frame.Rows(); i++ {
+				value, ok := frame.ConcreteAt(1, i)
+				require.True(t, ok)
+				min, ok := frame.ConcreteAt(2, i)
+				require.True(t, ok)
+				max, ok := frame.ConcreteAt(3, i)
+				require.True(t, ok)
 
 				require.Less(t, min, value)
 				require.Greater(t, max, value)
@@ -57,66 +117,98 @@ func TestTestdataScenarios(t *testing.T) {
 		})
 
 		t.Run("Should return a table with some nil values", func(t *testing.T) {
-			req := &tsdb.TsdbQuery{
-				TimeRange: tsdb.NewFakeTimeRange("5m", "now", time.Now()),
-				Queries: []*tsdb.Query{
-					{RefId: "A", IntervalMs: 100, MaxDataPoints: 100, Model: simplejson.New()},
+			timeRange := tsdb.NewFakeTimeRange("5m", "now", time.Now())
+
+			model := simplejson.New()
+			model.Set("withNil", true)
+
+			modelBytes, err := model.MarshalJSON()
+			require.NoError(t, err)
+
+			query := backend.DataQuery{
+				RefID: "A",
+				TimeRange: backend.TimeRange{
+					From: timeRange.MustGetFrom(),
+					To:   timeRange.MustGetTo(),
 				},
+				Interval:      100 * time.Millisecond,
+				MaxDataPoints: 100,
+				JSON:          modelBytes,
 			}
-			query := req.Queries[0]
-			query.Model.Set("withNil", true)
 
-			result := scenario.Handler(req.Queries[0], req)
-			table := result.Tables[0]
+			req := &backend.QueryDataRequest{
+				PluginContext: backend.PluginContext{},
+				Queries:       []backend.DataQuery{query},
+			}
 
-			nil1 := false
-			nil2 := false
-			nil3 := false
+			resp, err := p.handleRandomWalkTableScenario(context.Background(), req)
+			require.NoError(t, err)
+			require.NotNil(t, resp)
 
-			require.Greater(t, len(table.Rows), 50)
-			for _, row := range table.Rows {
-				if row[1] == nil {
-					nil1 = true
+			dResp, exists := resp.Responses[query.RefID]
+			require.True(t, exists)
+			require.NoError(t, dResp.Error)
+
+			require.Len(t, dResp.Frames, 1)
+			frame := dResp.Frames[0]
+			require.Greater(t, frame.Rows(), 50)
+			require.Len(t, frame.Fields, 5)
+			require.Equal(t, "Time", frame.Fields[0].Name)
+			require.Equal(t, "Value", frame.Fields[1].Name)
+			require.Equal(t, "Min", frame.Fields[2].Name)
+			require.Equal(t, "Max", frame.Fields[3].Name)
+			require.Equal(t, "Info", frame.Fields[4].Name)
+
+			valNil := false
+			minNil := false
+			maxNil := false
+
+			for i := 0; i < frame.Rows(); i++ {
+				_, ok := frame.ConcreteAt(1, i)
+				if !ok {
+					valNil = true
 				}
-				if row[2] == nil {
-					nil2 = true
+
+				_, ok = frame.ConcreteAt(2, i)
+				if !ok {
+					minNil = true
 				}
-				if row[3] == nil {
-					nil3 = true
+
+				_, ok = frame.ConcreteAt(3, i)
+				if !ok {
+					maxNil = true
 				}
 			}
 
-			require.True(t, nil1)
-			require.True(t, nil2)
-			require.True(t, nil3)
+			require.True(t, valNil)
+			require.True(t, minNil)
+			require.True(t, maxNil)
 		})
 	})
 }
 
 func TestParseLabels(t *testing.T) {
-	expectedTags := map[string]string{
+	expectedTags := data.Labels{
 		"job":      "foo",
 		"instance": "bar",
 	}
 
-	query1 := tsdb.Query{
-		Model: simplejson.NewFromAny(map[string]interface{}{
+	tcs := []struct {
+		model map[string]interface{}
+	}{
+		{model: map[string]interface{}{
 			"labels": `{job="foo", instance="bar"}`,
-		}),
-	}
-	require.Equal(t, expectedTags, parseLabels(&query1))
-
-	query2 := tsdb.Query{
-		Model: simplejson.NewFromAny(map[string]interface{}{
+		}},
+		{model: map[string]interface{}{
 			"labels": `job=foo, instance=bar`,
-		}),
-	}
-	require.Equal(t, expectedTags, parseLabels(&query2))
-
-	query3 := tsdb.Query{
-		Model: simplejson.NewFromAny(map[string]interface{}{
+		}},
+		{model: map[string]interface{}{
 			"labels": `job = foo,instance = bar`,
-		}),
+		}},
+	}
+
+	for i, tc := range tcs {
+		model := simplejson.NewFromAny(tc.model)
+		assert.Equal(t, expectedTags, parseLabels(model), fmt.Sprintf("Actual tags in test case %d doesn't match expected tags", i+1))
 	}
-	require.Equal(t, expectedTags, parseLabels(&query3))
 }

pkg/tsdb/testdatasource/testdata.go

@@ -1,42 +1,42 @@
 package testdatasource
 
 import (
-	"context"
+	"net/http"
 
+	"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/resource/httpadapter"
 	"github.com/grafana/grafana/pkg/infra/log"
-	"github.com/grafana/grafana/pkg/models"
-	"github.com/grafana/grafana/pkg/tsdb"
+	"github.com/grafana/grafana/pkg/plugins/backendplugin"
+	"github.com/grafana/grafana/pkg/plugins/backendplugin/coreplugin"
+	"github.com/grafana/grafana/pkg/registry"
 )
 
-type TestDataExecutor struct {
-	*models.DataSource
-	log log.Logger
-}
-
-func NewTestDataExecutor(dsInfo *models.DataSource) (tsdb.TsdbQueryEndpoint, error) {
-	return &TestDataExecutor{
-		DataSource: dsInfo,
-		log:        log.New("tsdb.testdata"),
-	}, nil
-}
-
 func init() {
-	tsdb.RegisterTsdbQueryEndpoint("testdata", NewTestDataExecutor)
+	registry.RegisterService(&testDataPlugin{})
 }
 
-func (e *TestDataExecutor) Query(ctx context.Context, dsInfo *models.DataSource, tsdbQuery *tsdb.TsdbQuery) (*tsdb.Response, error) {
-	result := &tsdb.Response{}
-	result.Results = make(map[string]*tsdb.QueryResult)
+type testDataPlugin struct {
+	BackendPluginManager backendplugin.Manager `inject:""`
+	logger               log.Logger
+	scenarios            map[string]*Scenario
+	queryMux             *datasource.QueryTypeMux
+}
 
-	for _, query := range tsdbQuery.Queries {
-		scenarioId := query.Model.Get("scenarioId").MustString("random_walk")
-		if scenario, exist := ScenarioRegistry[scenarioId]; exist {
-			result.Results[query.RefId] = scenario.Handler(query, tsdbQuery)
-			result.Results[query.RefId].RefId = query.RefId
-		} else {
-			e.log.Error("Scenario not found", "scenarioId", scenarioId)
-		}
+func (p *testDataPlugin) Init() error {
+	p.logger = log.New("tsdb.testdata")
+	p.scenarios = map[string]*Scenario{}
+	p.queryMux = datasource.NewQueryTypeMux()
+	p.registerScenarios()
+	resourceMux := http.NewServeMux()
+	p.registerRoutes(resourceMux)
+	factory := coreplugin.New(backend.ServeOpts{
+		QueryDataHandler:    p.queryMux,
+		CallResourceHandler: httpadapter.New(resourceMux),
+	})
+	err := p.BackendPluginManager.Register("testdata", factory)
+	if err != nil {
+		p.logger.Error("Failed to register plugin", "error", err)
 	}
-
-	return result, nil
+	return nil
 }