Merge pull request #177 from GridProtectionAlliance/feature/v5.2.0

Feature/v5.2.0

Author: elwills

CHANGELOG.md

@@ -61,3 +61,11 @@
 - Truncate time from grafana date time picker to seconds
 - Fixed warnings during deploy
 - Fixed LICENSE file
+
+### 5.2.0
+
+- Improved query performance to PiWebAPI by joing all queries in Panel into one batch request only
+- Change the Query Editor layout
+- Increased WebID cache from 1 hour to 12 hours and made it configurable
+
+- Added experimental feature to cache latest response in case of request failure to PiWebAPI

dist/module.js

@@ -39,8 +39,8 @@
       {"name": "Datasource Configuration", "path": "img/configuration.png"},
       {"name": "Annotations Editor", "path": "img/annotations.png"}
     ],
-    "version": "5.1.0",
-    "updated": "2024-11-01"
+    "version": "5.2.0",
+    "updated": "2025-03-21"
   },
   "dependencies": {
     "grafanaDependency": ">=10.1.0",

dist/module.js.map

@@ -1,6 +1,6 @@
 {
   "name": "grid-protection-alliance-osisoftpi-grafana",
-  "version": "5.1.0",
+  "version": "5.2.0",
   "description": "OSISoft PI Grafana Plugin",
   "scripts": {
     "build": "webpack -c ./.config/webpack/webpack.config.ts --env production",

dist/plugin.json

@@ -18,7 +18,7 @@ func main() {
 	// ID). When datasource configuration changed Dispose method will be called and
 	// new datasource instance created using NewSampleDatasource factory.
 	if err := datasource.Manage("gridprotectionalliance-osisoftpi-datasource", plugin.NewPIWebAPIDatasource, datasource.ManageOpts{}); err != nil {
-		log.DefaultLogger.Error("Manage", "Plugin", err.Error())
+		log.DefaultLogger.Error("PiWebAPI main", "Plugin", err.Error())
 		os.Exit(1)
 	}
 }

package.json

@@ -22,7 +22,7 @@ func (d *Datasource) processAnnotationQuery(ctx context.Context, query backend.D
 	// if there are errors we'll set the error and return the PiProcessedQuery with an error set.
 	tempJson, err := json.Marshal(query)
 	if err != nil {
-		log.DefaultLogger.Error("Error marshalling query", "error", err)
+		log.DefaultLogger.Error("Process annotation - Error marshalling", "error", err)
 
 		// create a processed query with the error set
 		ProcessedQuery = PiProcessedAnnotationQuery{
@@ -33,7 +33,7 @@ func (d *Datasource) processAnnotationQuery(ctx context.Context, query backend.D
 
 	err = json.Unmarshal(tempJson, &PiAnnotationQuery)
 	if err != nil {
-		log.DefaultLogger.Error("Error unmarshalling query", "error", err)
+		log.DefaultLogger.Error("Process annotation - Error unmarshalling", "error", err)
 
 		// create a processed query with the error set
 		ProcessedQuery = PiProcessedAnnotationQuery{
@@ -165,9 +165,6 @@ func convertAnnotationResponseToFrame(refID string, rawAnnotationResponse []byte
 	var annotationResponse map[string]AnnotationBatchResponse
 	var attributeDataItems []string
 
-	// log attributesEnabled
-	// log.DefaultLogger.Debug("Attributes Enabled", "attributesEnabled", attributesEnabled)
-
 	err := json.Unmarshal(rawAnnotationResponse, &annotationResponse)
 	if err != nil {
 		return nil, err

pkg/main.go

@@ -0,0 +1,59 @@
+package plugin
+
+import (
+	"sync"
+)
+
+// Cache is a basic in-memory key-value cache implementation.
+type Cache[K comparable, V any] struct {
+	items map[K]V    // The map storing key-value pairs.
+	mu    sync.Mutex // Mutex for controlling concurrent access to the cache.
+}
+
+// New creates a new Cache instance.
+func newCache[K comparable, V any]() *Cache[K, V] {
+	return &Cache[K, V]{
+		items: make(map[K]V),
+	}
+}
+
+// Set adds or updates a key-value pair in the cache.
+func (c *Cache[K, V]) Set(key K, value V) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.items[key] = value
+}
+
+// Get retrieves the value associated with the given key from the cache. The bool
+// return value will be false if no matching key is found, and true otherwise.
+func (c *Cache[K, V]) Get(key K) (V, bool) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	value, found := c.items[key]
+	return value, found
+}
+
+// Remove deletes the key-value pair with the specified key from the cache.
+func (c *Cache[K, V]) Remove(key K) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	delete(c.items, key)
+}
+
+// Pop removes and returns the value associated with the specified key from the cache.
+func (c *Cache[K, V]) Pop(key K) (V, bool) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	value, found := c.items[key]
+
+	// If the key is found, delete the key-value pair from the cache.
+	if found {
+		delete(c.items, key)
+	}
+
+	return value, found
+}

pkg/plugin/annotation_query.go

@@ -52,17 +52,25 @@ func NewPIWebAPIDatasource(ctx context.Context, settings backend.DataSourceInsta
 		return nil, fmt.Errorf("httpclient new: %w", err)
 	}
 
-	webIDCache := newWebIDCache()
+	var maxDuration int
+	if dataSourceOptions.MaxCacheTime != nil && *dataSourceOptions.MaxCacheTime > 0 {
+		maxDuration = *dataSourceOptions.MaxCacheTime
+	} else {
+		maxDuration = 12
+	}
+	webIDCache := newWebIDCache(maxDuration)
+	webCache := newCache[string, PiBatchData]()
 
-	// Create a new scheduler that will be used to clean the webIDCache every 60 minutes.
+	// Create a new scheduler that will be used to clean the webIDCache every MaxCacheTime hours.
 	scheduler := gocron.NewScheduler(time.UTC)
-	scheduler.Every(1).Hour().Do(cleanWebIDCache, webIDCache)
+	scheduler.Every(maxDuration).Hour().Do(cleanWebIDCache, webIDCache)
 	scheduler.StartAsync()
 
 	ds := &Datasource{
 		settings:                  settings,
 		httpClient:                httpClient,
 		webIDCache:                webIDCache,
+		webCache:                  webCache,
 		scheduler:                 scheduler,
 		websocketConnectionsMutex: &sync.Mutex{},
 		datasourceMutex:           &sync.Mutex{},
@@ -79,7 +87,7 @@ func NewPIWebAPIDatasource(ctx context.Context, settings backend.DataSourceInsta
 	// Create a new query mux and assign it to the datasource.
 	ds.queryMux = ds.newQueryMux()
 
-	backend.Logger.Info("NewPIWebAPIDatasource Created")
+	log.DefaultLogger.Info("PIWebAPI Datasource Created", "UID", settings.UID, "Name", settings.Name)
 
 	return ds, nil
 }
@@ -103,8 +111,8 @@ func (d *Datasource) updateRate() {
 		time.Sleep(time.Duration(d.callRate) * time.Millisecond)
 	}
 
-	// reset every 5 minutes (300 s)
-	if time.Since(d.initalTime).Seconds() > 300 {
+	// reset every 10 minutes (600 s)
+	if time.Since(d.initalTime).Seconds() > 600 {
 		d.initalTime = time.Now()
 		d.totalCalls = 1
 		d.callRate = float64(d.totalCalls) / float64(time.Now().Unix()-d.initalTime.Unix())
@@ -132,34 +140,40 @@ func (d *Datasource) newQueryMux() *datasource.QueryTypeMux {
 // The QueryDataResponse contains a map of RefID to the response for each query, and each response
 // contains Frames ([]*Frame).
 func (d *Datasource) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
-	// //TODO: Remove this debug information
-	// jsonReq, err := json.Marshal(req)
-	// if err != nil {
-	// 	return nil, fmt.Errorf("error marshaling QueryDataRequest: %v", err)
-	// }
-	// backend.Logger.Info("QueryDataRequest: ", "REQUEST", string(jsonReq))
-	// end remove this debug information
 	// Pass the query to the query muxer.
 	return d.queryMux.QueryData(ctx, req)
 }
 
 // TODO: Missing functionality: Add Replace Bad Values
 // QueryTSData is called by Grafana when a user executes a time series data query.
 func (d *Datasource) QueryTSData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
-	processedPIWebAPIQueries := make(map[string][]PiProcessedQuery)
 	datasourceUID := req.PluginContext.DataSourceInstanceSettings.UID
 
+	// tracer
+	ctx, span := tracing.DefaultTracer().Start(
+		ctx,
+		"New annotation query recieved",
+	)
+	defer span.End()
+
 	// Process queries generic query objects and turn them into a suitable format for the PI Web API
-	for _, q := range req.Queries {
-		processedPIWebAPIQueries[q.RefID] = d.processQuery(q, datasourceUID)
-	}
+	processedPIWebAPIQueries := d.processQuery(req.Queries, datasourceUID)
+
+	// span
+	span.AddEvent("Completed processing query request")
 
 	// Send the queries to the PI Web API
 	processedQueries_temp := d.batchRequest(ctx, processedPIWebAPIQueries)
 
+	// span
+	span.AddEvent("Completed processing batch request")
+
 	// Convert the PI Web API response into Grafana frames
 	response := d.processBatchtoFrames(processedQueries_temp)
 
+	// span
+	span.AddEvent("Completed processing batch to frames")
+
 	// Update rate and do backpressure
 	d.updateRate()
 
@@ -178,7 +192,6 @@ func (d *Datasource) QueryAnnotations(ctx context.Context, req *backend.QueryDat
 	defer span.End()
 
 	for _, q := range req.Queries {
-
 		span.AddEvent("Processing annotation query request",
 			trace.WithAttributes(
 				attribute.String("query.ref_id", q.RefID),
@@ -188,7 +201,6 @@ func (d *Datasource) QueryAnnotations(ctx context.Context, req *backend.QueryDat
 			),
 		)
 
-		// backend.Logger.Info("Processing Annotation Query", "RefID", q.RefID)
 		// Process the annotation query request, extracting only the useful information
 		ProcessedAnnotationQuery := d.processAnnotationQuery(ctx, q)
 		span.AddEvent("Completed processing annotation query request")
@@ -299,12 +311,13 @@ func (d *Datasource) CheckHealth(ctx context.Context, _ *backend.CheckHealthRequ
 	}
 	defer func() {
 		if err := resp.Body.Close(); err != nil {
-			log.DefaultLogger.Error("check health: failed to close response body", "err", err.Error())
+			log.DefaultLogger.Error("PiWebAPI Check health: failed to close response body", "err", err.Error())
 		}
 	}()
 	if resp.StatusCode != http.StatusOK {
 		return newHealthCheckErrorf("got response code %d", resp.StatusCode), nil
 	}
+	// return good status
 	return &backend.CheckHealthResult{
 		Status:  backend.HealthStatusOk,
 		Message: "Data source is working",
@@ -317,6 +330,25 @@ func newHealthCheckErrorf(format string, args ...interface{}) *backend.CheckHeal
 	return &backend.CheckHealthResult{Status: backend.HealthStatusError, Message: fmt.Sprintf(format, args...)}
 }
 
+// isUsingNewFormat checks whether the datasource is configured to use a new format.
+// This is determined by the NewFormat option in dataSourceOptions.
+// Returns true if NewFormat is set and enabled; otherwise, false.
 func (d *Datasource) isUsingNewFormat() bool {
 	return d.dataSourceOptions.NewFormat != nil && *d.dataSourceOptions.NewFormat
 }
+
+// isUsingStreaming checks whether the datasource has streaming enabled in experimental mode.
+// This requires both the UseExperimental and UseStreaming options to be set and enabled.
+// Returns true if both options are enabled; otherwise, false.
+func (d *Datasource) isUsingStreaming() bool {
+	return d.dataSourceOptions.UseExperimental != nil && *d.dataSourceOptions.UseExperimental &&
+		d.dataSourceOptions.UseStreaming != nil && *d.dataSourceOptions.UseStreaming
+}
+
+// isUsingResponseCache checks if response caching is enabled in experimental mode for the datasource.
+// This requires both the UseExperimental and UseResponseCache options to be set and enabled.
+// Returns true if both options are enabled; otherwise, false.
+func (d *Datasource) isUsingResponseCache() bool {
+	return d.dataSourceOptions.UseExperimental != nil && *d.dataSourceOptions.UseExperimental &&
+		d.dataSourceOptions.UseResponseCache != nil && *d.dataSourceOptions.UseResponseCache
+}

pkg/plugin/cache.go

@@ -17,6 +17,7 @@ type Datasource struct {
 	StreamHandler             backend.StreamHandler
 	httpClient                *http.Client
 	webIDCache                WebIDCache
+	webCache                  *Cache[string, PiBatchData]
 	channelConstruct          map[string]StreamChannelConstruct
 	datasourceMutex           *sync.Mutex
 	scheduler                 *gocron.Scheduler
@@ -31,14 +32,16 @@ type Datasource struct {
 }
 
 type PIWebAPIDataSourceJsonData struct {
-	URL             *string `json:"url,omitempty"`
-	Access          *string `json:"access,omitempty"`
-	PIServer        *string `json:"piserver,omitempty"`
-	AFServer        *string `json:"afserver,omitempty"`
-	AFDatabase      *string `json:"afdatabase,omitempty"`
-	PIPoint         *bool   `json:"pipoint,omitempty"`
-	NewFormat       *bool   `json:"newFormat,omitempty"`
-	UseUnit         *bool   `json:"useUnit,omitempty"`
-	UseExperimental *bool   `json:"useExperimental,omitempty"`
-	UseStreaming    *bool   `json:"useStreaming,omitempty"`
+	URL              *string `json:"url,omitempty"`
+	Access           *string `json:"access,omitempty"`
+	PIServer         *string `json:"piserver,omitempty"`
+	AFServer         *string `json:"afserver,omitempty"`
+	AFDatabase       *string `json:"afdatabase,omitempty"`
+	PIPoint          *bool   `json:"pipoint,omitempty"`
+	NewFormat        *bool   `json:"newFormat,omitempty"`
+	MaxCacheTime     *int    `json:"maxCacheTime,omitempty"`
+	UseUnit          *bool   `json:"useUnit,omitempty"`
+	UseExperimental  *bool   `json:"useExperimental,omitempty"`
+	UseStreaming     *bool   `json:"useStreaming,omitempty"`
+	UseResponseCache *bool   `json:"useResponseCache,omitempty"`
 }