Skip to content

Commit f15a5ea

Browse files
ryantxuryantxu
authored
Merge pull request #4 from grafana/use-managed-grpc
Use automatic instance manager
2 parents e055714 + 6a5ed87 commit f15a5ea

File tree

8 files changed

+172
-230
lines changed

8 files changed

+172
-230
lines changed

‎.vscode/launch.json‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
{
2+
"version": "0.2.0",
3+
"configurations": [
4+
{
5+
"name": "Run standalone plugin",
6+
"type": "go",
7+
"request": "launch",
8+
"mode": "auto",
9+
"program": "${workspaceFolder}/pkg/main.go",
10+
"env": {},
11+
"args": [
12+
"--standalone=true",
13+
]
14+
}
15+
]
16+
}

‎README.md‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -28,12 +28,10 @@ This plugin currently supports MQTT v3.1.x.
2828

2929
### Verify that the plugin is installed
3030

31-
1. In Grafana Enterprise from the left-hand menu, navigate to **Configuration** > **Data sources**.
31+
1. In Grafana from the left-hand menu, navigate to **Configuration** > **Data sources**.
3232
2. From the top-right corner, click the **Add data source** button.
33-
3. Search for `MQTT` in the search field, and hover over the AppDynamics search result.
33+
3. Search for `MQTT` in the search field, and hover over the MQTT search result.
3434
4. Click the **Select** button for MQTT.
35-
- If you can click the **Select** button, then it is installed.
36-
- If the button is missing or disabled, then the plugin is not installed. If you still need help, [contact Grafana Labs](https://grafana.com/contact).
3735

3836
## Configure the data source
3937

‎go.mod‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ module github.com/grafana/mqtt-datasource
33
go 1.16
44

55
require (
6-
github.com/eclipse/paho.mqtt.golang v1.3.3
7-
github.com/grafana/grafana-plugin-sdk-go v0.92.0
6+
github.com/eclipse/paho.mqtt.golang v1.3.4
7+
github.com/grafana/grafana-plugin-sdk-go v0.97.0
88
github.com/stretchr/testify v1.7.0
99
)

‎go.sum‎

Lines changed: 31 additions & 30 deletions
Large diffs are not rendered by default.

‎pkg/main.go‎

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -9,11 +9,7 @@ import (
99
)
1010

1111
func main() {
12-
13-
im := datasource.NewInstanceManager(plugin.NewServerInstance)
14-
err := datasource.Serve(plugin.GetDatasourceOpts(im))
15-
16-
if err != nil {
12+
if err := datasource.Manage("grafana-mqtt-datasource", plugin.NewMQTTInstance, datasource.ManageOpts{}); err != nil {
1713
log.DefaultLogger.Error(err.Error())
1814
os.Exit(1)
1915
}

‎pkg/plugin/datasource.go‎

Lines changed: 110 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,15 @@
11
package plugin
22

33
import (
4+
"context"
45
"encoding/json"
6+
"errors"
57
"fmt"
68
"strconv"
79
"time"
810

911
"github.com/grafana/grafana-plugin-sdk-go/backend"
12+
"github.com/grafana/grafana-plugin-sdk-go/backend/instancemgmt"
1013
"github.com/grafana/grafana-plugin-sdk-go/backend/log"
1114
"github.com/grafana/grafana-plugin-sdk-go/data"
1215
"github.com/grafana/mqtt-datasource/pkg/mqtt"
@@ -24,6 +27,7 @@ type MQTTClient interface {
2427
type MQTTDatasource struct {
2528
Client MQTTClient
2629
channelPrefix string
30+
closeCh chan struct{}
2731
}
2832

2933
func GetDatasourceSettings(s backend.DataSourceInstanceSettings) (*mqtt.Options, error) {
@@ -34,6 +38,112 @@ func GetDatasourceSettings(s backend.DataSourceInstanceSettings) (*mqtt.Options,
3438
return settings, nil
3539
}
3640

41+
// Make sure SampleDatasource implements required interfaces.
42+
// This is important to do since otherwise we will only get a
43+
// not implemented error response from plugin in runtime.
44+
var (
45+
_ backend.QueryDataHandler = (*MQTTDatasource)(nil)
46+
_ backend.CheckHealthHandler = (*MQTTDatasource)(nil)
47+
_ backend.StreamHandler = (*MQTTDatasource)(nil)
48+
_ instancemgmt.InstanceDisposer = (*MQTTDatasource)(nil)
49+
)
50+
51+
// NewMQTTDatasource creates a new datasource instance.
52+
func NewMQTTDatasource(client MQTTClient, id int64) *MQTTDatasource {
53+
return &MQTTDatasource{
54+
Client: client,
55+
channelPrefix: fmt.Sprintf("ds/%d/", id),
56+
closeCh: make(chan struct{}),
57+
}
58+
}
59+
60+
// NewMQTTDatasource creates a new datasource instance.
61+
func NewMQTTInstance(s backend.DataSourceInstanceSettings) (instancemgmt.Instance, error) {
62+
settings, err := GetDatasourceSettings(s)
63+
if err != nil {
64+
return nil, err
65+
}
66+
67+
client, err := mqtt.NewClient(*settings)
68+
if err != nil {
69+
return nil, err
70+
}
71+
72+
return NewMQTTDatasource(client, s.ID), nil
73+
}
74+
75+
// Dispose here tells plugin SDK that plugin wants to clean up resources
76+
// when a new instance created. As soon as datasource settings change detected
77+
// by SDK old datasource instance will be disposed and a new one will be created
78+
// using NewSampleDatasource factory function.
79+
func (ds *MQTTDatasource) Dispose() {
80+
close(ds.closeCh)
81+
}
82+
83+
func (ds *MQTTDatasource) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
84+
response := backend.NewQueryDataResponse()
85+
86+
for _, q := range req.Queries {
87+
res := ds.Query(q)
88+
response.Responses[q.RefID] = res
89+
}
90+
91+
return response, nil
92+
}
93+
94+
func (ds *MQTTDatasource) CheckHealth(ctx context.Context, req *backend.CheckHealthRequest) (*backend.CheckHealthResult, error) {
95+
if !ds.Client.IsConnected() {
96+
return &backend.CheckHealthResult{
97+
Status: backend.HealthStatusError,
98+
Message: "MQTT Disconnected",
99+
}, nil
100+
}
101+
102+
return &backend.CheckHealthResult{
103+
Status: backend.HealthStatusOk,
104+
Message: "MQTT Connected",
105+
}, nil
106+
}
107+
108+
func (ds *MQTTDatasource) SubscribeStream(ctx context.Context, req *backend.SubscribeStreamRequest) (*backend.SubscribeStreamResponse, error) {
109+
ds.Client.Subscribe(req.Path)
110+
111+
bytes, err := data.FrameToJSON(ToFrame([]mqtt.Message{}), true, false) // only schema
112+
if err != nil {
113+
return nil, err
114+
}
115+
return &backend.SubscribeStreamResponse{
116+
Status: backend.SubscribeStreamStatusOK,
117+
Data: bytes, // just the schema
118+
}, nil
119+
}
120+
121+
func (ds *MQTTDatasource) RunStream(ctx context.Context, req *backend.RunStreamRequest, sender backend.StreamPacketSender) error {
122+
defer ds.Client.Unsubscribe(req.Path)
123+
124+
for {
125+
select {
126+
case <-ds.closeCh:
127+
log.DefaultLogger.Info("Datasource restart")
128+
return errors.New("datasource closed")
129+
case <-ctx.Done():
130+
backend.Logger.Info("stop streaming (context canceled)")
131+
return nil
132+
case message := <-ds.Client.Stream():
133+
err := ds.SendMessage(message, req, sender)
134+
if err != nil {
135+
log.DefaultLogger.Error(fmt.Sprintf("unable to send message: %s", err.Error()))
136+
}
137+
}
138+
}
139+
}
140+
141+
func (ds *MQTTDatasource) PublishStream(ctx context.Context, req *backend.PublishStreamRequest) (*backend.PublishStreamResponse, error) {
142+
return &backend.PublishStreamResponse{
143+
Status: backend.PublishStreamStatusPermissionDenied, // ?? Unsupported
144+
}, nil
145+
}
146+
37147
type queryModel struct {
38148
Topic string `json:"queryText"`
39149
}
@@ -61,25 +171,6 @@ func ToFrame(messages []mqtt.Message) *data.Frame {
61171
return frame
62172
}
63173

64-
func NewMQTTDatasource(s backend.DataSourceInstanceSettings) (*MQTTDatasource, error) {
65-
settings, err := GetDatasourceSettings(s)
66-
if err != nil {
67-
return nil, err
68-
}
69-
70-
client, err := mqtt.NewClient(*settings)
71-
if err != nil {
72-
return nil, err
73-
}
74-
75-
ds := MQTTDatasource{
76-
Client: client,
77-
channelPrefix: fmt.Sprintf("ds/%d/", s.ID),
78-
}
79-
80-
return &ds, nil
81-
}
82-
83174
func (m *MQTTDatasource) Query(query backend.DataQuery) backend.DataResponse {
84175
var qm queryModel
85176

@@ -133,8 +224,3 @@ func (m *MQTTDatasource) SendMessage(msg mqtt.StreamMessage, req *backend.RunStr
133224
log.DefaultLogger.Debug(fmt.Sprintf("Sending message to client for topic %s", msg.Topic))
134225
return sender.Send(packet)
135226
}
136-
137-
func (m *MQTTDatasource) Dispose() {
138-
// Called before creating a a new instance to allow plugin authors
139-
// to cleanup.
140-
}

pkg/plugin/handler_test.go renamed to pkg/plugin/datasource_test.go

Lines changed: 10 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -5,23 +5,19 @@ import (
55
"testing"
66

77
"github.com/grafana/grafana-plugin-sdk-go/backend"
8-
"github.com/grafana/grafana-plugin-sdk-go/backend/instancemgmt"
98
"github.com/grafana/mqtt-datasource/pkg/mqtt"
109
"github.com/grafana/mqtt-datasource/pkg/plugin"
1110
"github.com/stretchr/testify/require"
1211
)
1312

1413
func TestCheckHealthHandler(t *testing.T) {
1514
t.Run("HealthStatusOK when can connect", func(t *testing.T) {
16-
im := fakeInstanceManager{
17-
client: &fakeMQTTClient{
18-
connected: true,
19-
subscribed: false,
20-
}, err: nil}
15+
ds := plugin.NewMQTTDatasource(&fakeMQTTClient{
16+
connected: true,
17+
subscribed: false,
18+
}, 5)
2119

22-
ds := plugin.GetDatasourceOpts(&im)
23-
24-
res, _ := ds.CheckHealthHandler.CheckHealth(
20+
res, _ := ds.CheckHealth(
2521
context.Background(),
2622
&backend.CheckHealthRequest{},
2723
)
@@ -31,15 +27,12 @@ func TestCheckHealthHandler(t *testing.T) {
3127
})
3228

3329
t.Run("HealthStatusError when disconnected", func(t *testing.T) {
34-
im := fakeInstanceManager{
35-
client: &fakeMQTTClient{
36-
connected: false,
37-
subscribed: false,
38-
}, err: nil}
39-
40-
ds := plugin.GetDatasourceOpts(&im)
30+
ds := plugin.NewMQTTDatasource(&fakeMQTTClient{
31+
connected: false,
32+
subscribed: false,
33+
}, 5)
4134

42-
res, _ := ds.CheckHealthHandler.CheckHealth(
35+
res, _ := ds.CheckHealth(
4336
context.Background(),
4437
&backend.CheckHealthRequest{},
4538
)
@@ -49,21 +42,6 @@ func TestCheckHealthHandler(t *testing.T) {
4942
})
5043
}
5144

52-
type fakeInstanceManager struct {
53-
client *fakeMQTTClient
54-
err error
55-
}
56-
57-
func (im *fakeInstanceManager) Get(pc backend.PluginContext) (instancemgmt.Instance, error) {
58-
return &plugin.MQTTDatasource{
59-
Client: im.client,
60-
}, im.err
61-
}
62-
63-
func (im *fakeInstanceManager) Do(pc backend.PluginContext, fn instancemgmt.InstanceCallbackFunc) error {
64-
return nil
65-
}
66-
6745
type fakeMQTTClient struct {
6846
connected bool
6947
subscribed bool

0 commit comments

Comments
 (0)