diff --git a/.github/workflows/grafana-ci.yml b/.github/workflows/grafana-ci.yml index e774c0c..c86161a 100644 --- a/.github/workflows/grafana-ci.yml +++ b/.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 diff --git a/config/datasources/grafana-datasources.yml b/config/datasources/grafana-datasources.yml index 10063d2..864a058 100644 --- a/config/datasources/grafana-datasources.yml +++ b/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' diff --git a/grafana-datasource-plugin/docker-compose.yaml b/grafana-datasource-plugin/docker-compose.yaml index 86fa806..c021298 100644 --- a/grafana-datasource-plugin/docker-compose.yaml +++ b/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 diff --git a/grafana-datasource-plugin/pkg/models/settings.go b/grafana-datasource-plugin/pkg/models/settings.go index 1e3a765..6dc1755 100644 --- a/grafana-datasource-plugin/pkg/models/settings.go +++ b/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 { diff --git a/grafana-datasource-plugin/pkg/plugin/datasource.go b/grafana-datasource-plugin/pkg/plugin/datasource.go index 6a2afc4..7713098 100644 --- a/grafana-datasource-plugin/pkg/plugin/datasource.go +++ b/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 } diff --git a/grafana-datasource-plugin/pkg/plugin/datasource_test.go b/grafana-datasource-plugin/pkg/plugin/datasource_test.go index 44b9fe5..88840bc 100644 --- a/grafana-datasource-plugin/pkg/plugin/datasource_test.go +++ b/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() diff --git a/grafana-datasource-plugin/pkg/plugin/hermes_connection.go b/grafana-datasource-plugin/pkg/plugin/hermes_connection.go new file mode 100644 index 0000000..b509c1b --- /dev/null +++ b/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() +} diff --git a/grafana-datasource-plugin/pkg/plugin/resources.go b/grafana-datasource-plugin/pkg/plugin/resources.go index 7c3e321..69449f8 100644 --- a/grafana-datasource-plugin/pkg/plugin/resources.go +++ b/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) } diff --git a/grafana-datasource-plugin/provisioning/datasources/datasources.yml b/grafana-datasource-plugin/provisioning/datasources/datasources.yml index 1f19345..f2c2ce4 100644 --- a/grafana-datasource-plugin/provisioning/datasources/datasources.yml +++ b/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' diff --git a/grafana-datasource-plugin/src/components/ConfigEditor.tsx b/grafana-datasource-plugin/src/components/ConfigEditor.tsx index a8359e3..500e28e 100644 --- a/grafana-datasource-plugin/src/components/ConfigEditor.tsx +++ b/grafana-datasource-plugin/src/components/ConfigEditor.tsx @@ -45,6 +45,13 @@ export function ConfigEditor(props: Props) { }); }; + const onHermesChange = (event: ChangeEvent) => { + onOptionsChange({ + ...options, + jsonData: { ...jsonData, hermesUrl: event.target.value } + }) + } + return ( <> @@ -85,6 +92,15 @@ export function ConfigEditor(props: Props) { width={40} /> + + + ); } diff --git a/grafana-datasource-plugin/src/components/TelemetryFields.tsx b/grafana-datasource-plugin/src/components/TelemetryFields.tsx index 69d2a27..c1f572f 100644 --- a/grafana-datasource-plugin/src/components/TelemetryFields.tsx +++ b/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> { +function toChannelOptions(entries: ChannelRefWithMetadata[]): Array> { return entries.map((e) => ({ label: `${e.component}.${e.name}`, - description: e.component, + group: e.component, + description: e.metadata.description, value: channelToKey(e), })); } diff --git a/grafana-datasource-plugin/src/datasource.ts b/grafana-datasource-plugin/src/datasource.ts index 8dc1c9c..f9d8c06 100644 --- a/grafana-datasource-plugin/src/datasource.ts +++ b/grafana-datasource-plugin/src/datasource.ts @@ -1,7 +1,7 @@ import { DataQueryRequest, DataSourceInstanceSettings, CoreApp, ScopedVars } from '@grafana/data'; import { DataSourceWithBackend, getTemplateSrv } from '@grafana/runtime'; import { map } from 'rxjs/operators'; -import { MyQuery, MyDataSourceOptions, DEFAULT_QUERY, ChannelRef, KeyRef, withDefaults } from './types'; +import { MyQuery, MyDataSourceOptions, DEFAULT_QUERY, ChannelRef, ChannelRefResponse, ChannelRefWithMetadata, KeyRef, withDefaults } from './types'; import { buildQuery } from 'query'; export class DataSource extends DataSourceWithBackend { @@ -66,8 +66,9 @@ export class DataSource extends DataSourceWithBackend { - return this.getResource('telemetry/channels'); + async getChannels(): Promise { + return this.getResource('telemetry/channels') + .then((entries) => entries.map((e) => ({ ...e, metadata: JSON.parse(e.metadata) }))); } async getSources(): Promise { diff --git a/grafana-datasource-plugin/src/types.ts b/grafana-datasource-plugin/src/types.ts index 43c1ff2..2bf4119 100644 --- a/grafana-datasource-plugin/src/types.ts +++ b/grafana-datasource-plugin/src/types.ts @@ -10,6 +10,16 @@ export interface ChannelRef { name: string; } +export interface ChannelRefResponse extends ChannelRef { + metadata: string; +} + +export interface ChannelRefWithMetadata extends ChannelRef { + metadata: { + description?: string; + }; +} + export interface KeyRef { component: string; channel: string; @@ -46,6 +56,7 @@ export interface MyDataSourceOptions extends DataSourceJsonData { host?: string; user?: string; database?: string; + hermesUrl?: string; } /** diff --git a/grafana-datasource-plugin/tests/configEditor.spec.ts b/grafana-datasource-plugin/tests/configEditor.spec.ts index f8a03d7..8881051 100644 --- a/grafana-datasource-plugin/tests/configEditor.spec.ts +++ b/grafana-datasource-plugin/tests/configEditor.spec.ts @@ -1,25 +1,90 @@ +import { exec, spawn } from 'child_process'; +import { createConnection } from 'net'; import { test, expect } from '@grafana/plugin-e2e'; import { MyDataSourceOptions, MySecureJsonData } from '../src/types'; +function runCommand(dir: string, name: string, ...args: string[]) { + return new Promise((resolve, reject) => exec(`${name} ${args.join(' ')}`, { cwd: dir }, (error, stdout, stderr) => { + if (error) { + reject(error); + } else { + resolve({ stdout, stderr }); + } + })); +} + +function startCommand(dir: string, name: string, ...args: string[]): () => Promise { + const backend = spawn(name, args, { cwd: dir, stdio: 'inherit' }) + .on('error', (err) => console.error(err)); + return () => new Promise(resolve => { + backend.kill(); + backend.once('exit', resolve); + }); +} + +function waitPort(target: string, retries = 10, timeout = 500) { + const [host, portStr] = target.split(':'); + const port = Number(portStr); + + return new Promise((resolve, reject) => { + const tryPort = () => { + const socket = createConnection({ host, port }).setTimeout(timeout); + let done = false; + + const onFinish = (sucess: boolean) => { + if (done) return; + done = true; + socket.destroy(); + if (sucess) { + resolve(); + } else if (retries-- <= 0) { + reject(new Error(`Port ${target} did not open in time`)); + } else { + setTimeout(tryPort, timeout); + } + }; + socket.on('connect', () => onFinish(true)) + .on('error', () => onFinish(false)) + .on('timeout', () => onFinish(false)) + }; + tryPort(); + }); +} + test('smoke: should render config editor', async ({ createDataSourceConfigPage, readProvisionedDataSource, page }) => { const ds = await readProvisionedDataSource({ fileName: 'datasources.yml' }); await createDataSourceConfigPage({ type: ds.type }); await expect(page.getByRole('textbox', { name: 'Host' })).toBeVisible(); await expect(page.getByRole('textbox', { name: 'User' })).toBeVisible(); await expect(page.getByRole('textbox', { name: 'Database' })).toBeVisible(); + await expect(page.getByRole('textbox', { name: 'Hermes' })).toBeVisible(); }); + test('"Save & test" should be successful when configuration is valid', async ({ createDataSourceConfigPage, readProvisionedDataSource, page, }) => { const ds = await readProvisionedDataSource({ fileName: 'datasources.yml' }); - const configPage = await createDataSourceConfigPage({ type: ds.type }); - await page.getByRole('textbox', { name: 'Host' }).fill(ds.jsonData.host ?? ''); - await page.getByRole('textbox', { name: 'User' }).fill(ds.jsonData.user ?? ''); - await page.locator('#config-editor-password').fill(ds.secureJsonData?.password ?? ''); - await page.getByRole('textbox', { name: 'Database' }).fill(ds.jsonData.database ?? ''); - await expect(configPage.saveAndTest()).toBeOK(); + + // Bind to 0.0.0.0 in CI so docker can connect to it + const bindHost = process.env.CI ? '0.0.0.0' : 'localhost'; + await runCommand('..', 'make', 'out/backend').catch((err) => console.error(err)); + const backendKill = startCommand('..', './out/backend', '--bind-type', 'tcp', '--bind', `${bindHost}:6880`); + await waitPort('localhost:6880').catch((err) => console.error(err)); + + try { + const configPage = await createDataSourceConfigPage({ type: ds.type }); + await page.getByRole('textbox', { name: 'Host' }).fill(ds.jsonData.host ?? ''); + await page.getByRole('textbox', { name: 'User' }).fill(ds.jsonData.user ?? ''); + await page.locator('#config-editor-password').fill(ds.secureJsonData?.password ?? ''); + await page.getByRole('textbox', { name: 'Database' }).fill(ds.jsonData.database ?? ''); + await page.getByRole('textbox', { name: 'Hermes' }).fill(ds.jsonData.hermesUrl ?? ''); + await expect(configPage.saveAndTest()).not.toBeOK(); + await expect(configPage).toHaveAlert('error', { hasText: 'Status of connection to Hermes is unknown, no dictionaries are loaded or registered yet.' }); + } finally { + await backendKill(); + } }); test('"Save & test" should fail when configuration is invalid', async ({ @@ -29,6 +94,7 @@ test('"Save & test" should fail when configuration is invalid', async ({ }) => { const ds = await readProvisionedDataSource({ fileName: 'datasources.yml' }); const configPage = await createDataSourceConfigPage({ type: ds.type }); + await page.getByRole('textbox', { name: 'Hermes' }).fill(ds.jsonData.hermesUrl ?? ''); await expect(configPage.saveAndTest()).not.toBeOK(); - await expect(configPage).toHaveAlert('error', { hasText: 'Host configuration parameter is missing' }); + await expect(configPage).toHaveAlert('error', { hasText: 'unable to initialize hermes client' }); });