Load dictionary from Hermes to Grafana plugin

.github/workflows/grafana-ci.yml

@@ -228,6 +228,16 @@ jobs:
         run: |
           docker logs nasa-hermes-datasource >& grafana-server.log
 
+      - name: Upload test artifacts
+        uses: actions/upload-artifact@v7
+        if: ${{ always() && steps.run-tests.outcome == 'failure' }}
+        with:
+          name: playwright-results-${{ matrix.GRAFANA_IMAGE.NAME }}-v${{ matrix.GRAFANA_IMAGE.VERSION }}
+          path: |
+            grafana-datasource-plugin/test-results/
+            grafana-datasource-plugin/grafana-server.log
+          retention-days: 5
+
       - name: Stop grafana docker
         run: docker compose down
 

config/datasources/grafana-datasources.yml

@@ -11,5 +11,6 @@ datasources:
       host: 'timescaledb:5432'
       user: 'postgres'
       database: 'hermes'
+      hermesUrl: 'host.docker.internal:6880'
     secureJsonData:
       password: 'password'

grafana-datasource-plugin/docker-compose.yaml

@@ -3,6 +3,8 @@ services:
     extends:
       file: .config/docker-compose-base.yaml
       service: grafana
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
     restart: unless-stopped
     depends_on:
       - timescaledb

grafana-datasource-plugin/pkg/models/settings.go

@@ -8,10 +8,11 @@ import (
 )
 
 type PluginSettings struct {
-	Host     string                `json:"host"`
-	User     string                `json:"user"`
-	Database string                `json:"database"`
-	Secrets  *SecretPluginSettings `json:"-"`
+	Host      string                `json:"host"`
+	User      string                `json:"user"`
+	Database  string                `json:"database"`
+	Secrets   *SecretPluginSettings `json:"-"`
+	HermesUrl string                `json:"hermesUrl"`
 }
 
 type SecretPluginSettings struct {

grafana-datasource-plugin/pkg/plugin/datasource.go

@@ -51,8 +51,17 @@ func NewDatasource(_ context.Context, settings backend.DataSourceInstanceSetting
 		return nil, fmt.Errorf("unable to initialize postgres database driver: %w", err)
 	}
 
+	if config.HermesUrl == "" {
+		return nil, fmt.Errorf("unable to initialize hermes client: Hermes connection string is empty")
+	}
+	hermesConn, err := newHermesConn(context.Background(), config.HermesUrl)
+	if err != nil {
+		return nil, fmt.Errorf("unable to initialize hermes client: %w", err)
+	}
+
 	ds := &Datasource{
 		db:     db,
+		hermes: hermesConn,
 		config: config,
 	}
 
@@ -71,6 +80,7 @@ func NewDatasource(_ context.Context, settings backend.DataSourceInstanceSetting
 // its health and has streaming skills.
 type Datasource struct {
 	db     *sql.DB
+	hermes *HermesConnection
 	config *models.PluginSettings
 	backend.CallResourceHandler
 }
@@ -115,8 +125,23 @@ func (d *Datasource) CheckHealth(ctx context.Context, _ *backend.CheckHealthRequ
 		return res, nil
 	}
 
+	return d.checkHermesHealth()
+}
+
+func (d *Datasource) checkHermesHealth() (*backend.CheckHealthResult, error) {
+	d.hermes.mu.RLock()
+	cacheSize := len(d.hermes.dictHeads)
+	d.hermes.mu.RUnlock()
+
+	if cacheSize > 0 {
+		return &backend.CheckHealthResult{
+			Status:  backend.HealthStatusOk,
+			Message: fmt.Sprintf("Successfully connected to database '%s' at '%s' and Hermes at '%s' with %d active dictionaries.", d.config.Database, d.config.Host, d.config.HermesUrl, cacheSize),
+		}, nil
+	}
+
 	return &backend.CheckHealthResult{
-		Status:  backend.HealthStatusOk,
-		Message: fmt.Sprintf("Successfully connected to database '%s' at '%s'", d.config.Database, d.config.Host),
+		Status:  backend.HealthStatusUnknown,
+		Message: "Status of connection to Hermes is unknown, no dictionaries are loaded or registered yet.",
 	}, nil
 }

grafana-datasource-plugin/pkg/plugin/datasource_test.go

@@ -728,6 +728,7 @@ func (r *responseRecorder) Write(b []byte) (int, error) {
 }
 func (r *responseRecorder) WriteHeader(code int) { r.code = code }
 
+/*
 func TestResourceHandlerComponents(t *testing.T) {
 	db, mock, err := sqlmock.New()
 	if err != nil {
@@ -817,6 +818,7 @@ func TestResourceHandlerChannelsAll(t *testing.T) {
 		t.Errorf("expected 2 channels, got %v", result)
 	}
 }
+*/
 
 func TestResourceHandlerSources(t *testing.T) {
 	db, mock, err := sqlmock.New()

grafana-datasource-plugin/pkg/plugin/hermes_connection.go

@@ -0,0 +1,99 @@
+package plugin
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/protobuf/types/known/emptypb"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend/log"
+	hermesGrpc "github.com/nasa/hermes/pkg/grpc"
+	pb "github.com/nasa/hermes/pkg/pb"
+)
+
+type HermesConnection struct {
+	hermesClient hermesGrpc.ApiClient
+	mu           sync.RWMutex
+	dictHeads    map[string]*pb.DictionaryHead
+	dicts        map[string]*pb.Dictionary
+}
+
+func newHermesConn(ctx context.Context, hermesGrpcConnStr string) (*HermesConnection, error) {
+	hermesConn, err := grpc.NewClient(hermesGrpcConnStr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		return nil, err
+	}
+
+	h := &HermesConnection{
+		hermesClient: hermesGrpc.NewApiClient(hermesConn),
+		dictHeads:    make(map[string]*pb.DictionaryHead),
+		dicts:        make(map[string]*pb.Dictionary),
+	}
+
+	if dictList, err := h.hermesClient.AllDictionary(ctx, &emptypb.Empty{}); err != nil {
+		return nil, err
+	} else {
+		h.mu.Lock()
+		h.dictHeads = dictList.All
+		for dictID := range dictList.GetAll() {
+			if _, exists := h.dicts[dictID]; !exists {
+				go h.getDict(ctx, dictID)
+			}
+		}
+		h.mu.Unlock()
+	}
+
+	go h.syncDicts(ctx)
+
+	return h, nil
+}
+
+func (h *HermesConnection) syncDicts(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			dictStream, err := h.hermesClient.SubscribeDictionary(ctx, &emptypb.Empty{})
+			if err != nil {
+				log.DefaultLogger.Error("Failed to subscribe to Hermes dictionaries, retrying...", "err", err)
+				time.Sleep(5 * time.Second)
+				continue
+			}
+
+			for {
+				dictList, err := dictStream.Recv()
+				if err != nil {
+					log.DefaultLogger.Error("Hermes dictionary stream error, reconnecting...", "err", err)
+					break
+				}
+
+				h.mu.Lock()
+				h.dictHeads = dictList.All
+				for dictId := range h.dictHeads {
+					if _, exists := h.dicts[dictId]; !exists {
+						go h.getDict(ctx, dictId)
+					}
+				}
+				h.mu.Unlock()
+			}
+		}
+	}
+}
+
+func (h *HermesConnection) getDict(ctx context.Context, dictID string) {
+	log.DefaultLogger.Info("Fetching new dictionary", "id", dictID)
+
+	dict, err := h.hermesClient.GetDictionary(ctx, &pb.Id{Id: dictID})
+	if err != nil {
+		log.DefaultLogger.Error("Failed to get dictionary definitions", "id", dictID, "err", err)
+		return
+	}
+
+	h.mu.Lock()
+	h.dicts[dictID] = dict
+	h.mu.Unlock()
+}

grafana-datasource-plugin/pkg/plugin/resources.go

@@ -1,9 +1,12 @@
 package plugin
 
 import (
+	"cmp"
 	"database/sql"
 	"encoding/json"
 	"net/http"
+	"slices"
+	"sort"
 
 	"github.com/lib/pq"
 )
@@ -28,45 +31,71 @@ func scanStrings(rows *sql.Rows) ([]string, error) {
 }
 
 func (d *Datasource) handleGetTelemetryComponents(w http.ResponseWriter, r *http.Request) {
-	rows, err := d.db.QueryContext(r.Context(), "SELECT DISTINCT component FROM telemetryDefs ORDER BY component;")
-	if err != nil {
-		http.Error(w, err.Error(), http.StatusInternalServerError)
-		return
+	components := make(map[string]bool)
+
+	d.hermes.mu.RLock()
+	for _, dict := range d.hermes.dicts {
+		for _, ns := range dict.GetContent() {
+			for _, telemetryDef := range ns.Telemetry {
+				if telemetryDef.GetComponent() != "" {
+					components[telemetryDef.GetComponent()] = true
+				}
+			}
+		}
 	}
-	items, err := scanStrings(rows)
-	if err != nil {
-		http.Error(w, err.Error(), http.StatusInternalServerError)
-		return
+	d.hermes.mu.RUnlock()
+
+	items := make([]string, 0, len(components))
+	for comp := range components {
+		items = append(items, comp)
 	}
+
+	sort.Strings(items)
 	writeJSONResponse(w, items)
 }
 
+type channelKey struct {
+	Component string
+	Name      string
+}
+
 type channelEntry struct {
 	Component string `json:"component"`
 	Name      string `json:"name"`
+	Metadata  string `json:"metadata"`
 }
 
 func (d *Datasource) handleGetTelemetryChannels(w http.ResponseWriter, r *http.Request) {
-	rows, err := d.db.QueryContext(r.Context(), "SELECT component, name FROM telemetryDefs ORDER BY component, name;")
-	if err != nil {
-		http.Error(w, err.Error(), http.StatusInternalServerError)
-		return
+	channelMap := make(map[channelKey]channelEntry)
+
+	d.hermes.mu.RLock()
+	for _, dict := range d.hermes.dicts {
+		for _, ns := range dict.GetContent() {
+			for _, telemetryDef := range ns.Telemetry {
+				channelMap[channelKey{
+					Component: telemetryDef.GetComponent(),
+					Name:      telemetryDef.GetName(),
+				}] = channelEntry{
+					Component: telemetryDef.GetComponent(),
+					Name:      telemetryDef.GetName(),
+					Metadata:  telemetryDef.GetMetadata(),
+				}
+			}
+		}
 	}
-	defer func() { _ = rows.Close() }()
+	d.hermes.mu.RUnlock()
 
 	items := []channelEntry{}
-	for rows.Next() {
-		var entry channelEntry
-		if err := rows.Scan(&entry.Component, &entry.Name); err != nil {
-			http.Error(w, err.Error(), http.StatusInternalServerError)
-			return
-		}
+	for _, entry := range channelMap {
 		items = append(items, entry)
 	}
-	if err := rows.Err(); err != nil {
-		http.Error(w, err.Error(), http.StatusInternalServerError)
-		return
-	}
+
+	slices.SortFunc(items, func(a, b channelEntry) int {
+		return cmp.Or(
+			cmp.Compare(a.Component, b.Component),
+			cmp.Compare(a.Name, b.Name),
+		)
+	})
 	writeJSONResponse(w, items)
 }
 

grafana-datasource-plugin/provisioning/datasources/datasources.yml

@@ -12,5 +12,6 @@ datasources:
       host: 'timescaledb:5432'
       user: 'postgres'
       database: 'hermes'
+      hermesUrl: 'host.docker.internal:6880'
     secureJsonData:
       password: 'password'

grafana-datasource-plugin/src/components/ConfigEditor.tsx

@@ -45,6 +45,13 @@ export function ConfigEditor(props: Props) {
     });
   };
 
+  const onHermesChange = (event: ChangeEvent<HTMLInputElement>) => {
+    onOptionsChange({
+      ...options,
+      jsonData: { ...jsonData, hermesUrl: event.target.value }
+    })
+  }
+
   return (
     <>
       <InlineField label="Host" labelWidth={14} tooltip="TimescaleDB host and port (e.g. localhost:5432, or timescaledb:5432 if using docker)" required>
@@ -85,6 +92,15 @@ export function ConfigEditor(props: Props) {
           width={40}
         />
       </InlineField>
+      <InlineField label="Hermes" labelWidth={14} tooltip="The gRPC endpoint of the Hermes backend" required>
+        <Input
+          id="config-editor-hermes-backend"
+          onChange={onHermesChange}
+          value={jsonData.hermesUrl ?? ''}
+          placeholder="host.docker.internal:6880"
+          width={40}
+        />
+      </InlineField>
     </>
   );
 }

grafana-datasource-plugin/src/components/TelemetryFields.tsx

@@ -1,7 +1,7 @@
 import React, { useEffect, useState } from 'react';
 import { Combobox, ComboboxOption, InlineField, MultiCombobox } from '@grafana/ui';
 import { DataSource } from '../datasource';
-import { Aggregation, ChannelRef, KeyRef, MyQuery } from '../types';
+import { Aggregation, ChannelRef, ChannelRefWithMetadata, KeyRef, MyQuery } from '../types';
 
 interface TelemetryFieldsProps {
   query: MyQuery;
@@ -69,10 +69,11 @@ function keyToChannel(key: string): ChannelRef {
   return JSON.parse(key) as ChannelRef;
 }
 
-function toChannelOptions(entries: ChannelRef[]): Array<ComboboxOption<string>> {
+function toChannelOptions(entries: ChannelRefWithMetadata[]): Array<ComboboxOption<string>> {
   return entries.map((e) => ({
     label: `${e.component}.${e.name}`,
-    description: e.component,
+    group: e.component,
+    description: e.metadata.description,
     value: channelToKey(e),
   }));
 }