Align time series data if possible

pkg/datasource/models.go

@@ -16,6 +16,7 @@ type ZabbixDatasourceSettingsDTO struct {
 	CacheTTL    string `json:"cacheTTL"`
 	Timeout     string `json:"timeout"`
 
+	DisableDataAlignment    bool `json:"disableDataAlignment"`
 	DisableReadOnlyUsersAck bool `json:"disableReadOnlyUsersAck"`
 }
 
@@ -27,6 +28,7 @@ type ZabbixDatasourceSettings struct {
 	CacheTTL    time.Duration
 	Timeout     time.Duration
 
+	DisableDataAlignment    bool `json:"disableDataAlignment"`
 	DisableReadOnlyUsersAck bool `json:"disableReadOnlyUsersAck"`
 }
 
@@ -62,6 +64,7 @@ type QueryFilter struct {
 // QueryOptions model
 type QueryOptions struct {
 	ShowDisabledItems    bool `json:"showDisabledItems"`
+	DisableDataAlignment bool `json:"disableDataAlignment"`
 }
 
 // QueryOptions model

pkg/datasource/response_handler.go

@@ -2,9 +2,11 @@ package datasource
 
 import (
 	"fmt"
+	"regexp"
 	"strconv"
 	"time"
 
+	"github.com/alexanderzobnin/grafana-zabbix/pkg/gtime"
 	"github.com/alexanderzobnin/grafana-zabbix/pkg/timeseries"
 	"github.com/alexanderzobnin/grafana-zabbix/pkg/zabbix"
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
@@ -33,6 +35,7 @@ func convertHistoryToTimeSeries(history zabbix.History, items []*zabbix.Item) []
 			if len(pointItem.Hosts) > 0 {
 				pointSeries.Meta.Name = fmt.Sprintf("%s: %s", pointItem.Hosts[0].Name, itemName)
 			}
+			pointSeries.Meta.Interval = parseItemUpdateInterval(pointItem.Delay)
 		}
 
 		value := point.Value
@@ -209,3 +212,18 @@ func getTrendPointValue(point zabbix.TrendPoint, valueType string) (float64, err
 
 	return 0, fmt.Errorf("failed to get trend value, unknown value type: %s", valueType)
 }
+
+var fixedUpdateIntervalPattern = regexp.MustCompile(`^(\d+)([shd]?)$`)
+
+func parseItemUpdateInterval(delay string) *time.Duration {
+	if valid := fixedUpdateIntervalPattern.MatchString(delay); !valid {
+		return nil
+	}
+
+	interval, err := gtime.ParseInterval(delay)
+	if err != nil {
+		return nil
+	}
+
+	return &interval
+}

pkg/datasource/zabbix.go

@@ -79,6 +79,16 @@ func (ds *ZabbixDatasourceInstance) queryNumericDataForItems(ctx context.Context
 	}
 
 	series := convertHistoryToTimeSeries(history, items)
+
+	// Align time series data if possible
+	if query.Options.DisableDataAlignment == false && ds.Settings.DisableDataAlignment == false {
+		for _, s := range series {
+			if s.Meta.Interval != nil {
+				s.TS = s.TS.Align(*s.Meta.Interval)
+			}
+		}
+	}
+
 	series, err = applyFunctions(series, query.Functions)
 	if err != nil {
 		return nil, err

pkg/timeseries/align.go

@@ -0,0 +1,52 @@
+package timeseries
+
+import (
+	"math"
+	"sort"
+	"time"
+)
+
+// Aligns point's time stamps according to provided interval.
+func (ts TimeSeries) Align(interval time.Duration) TimeSeries {
+	if interval <= 0 || ts.Len() < 2 {
+		return ts
+	}
+
+	alignedTs := NewTimeSeries()
+	var frameTs = ts[0].GetTimeFrame(interval)
+	var pointFrameTs time.Time
+	var point TimePoint
+
+	for i := 0; i < ts.Len(); i++ {
+		point = ts[i]
+		pointFrameTs = point.GetTimeFrame(interval)
+
+		if pointFrameTs.After(frameTs) {
+			for frameTs.Before(pointFrameTs) {
+				alignedTs = append(alignedTs, TimePoint{Time: frameTs, Value: nil})
+				frameTs = frameTs.Add(interval)
+			}
+		}
+
+		alignedTs = append(alignedTs, TimePoint{Time: pointFrameTs, Value: point.Value})
+		frameTs = frameTs.Add(interval)
+	}
+
+	return alignedTs
+}
+
+// Detects interval between data points in milliseconds based on median delta between points.
+func (ts TimeSeries) DetectInterval() time.Duration {
+	if ts.Len() < 2 {
+		return 0
+	}
+
+	deltas := make([]int, 0)
+	for i := 1; i < ts.Len(); i++ {
+		delta := ts[i].Time.Sub(ts[i-1].Time)
+		deltas = append(deltas, int(delta.Milliseconds()))
+	}
+	sort.Ints(deltas)
+	midIndex := int(math.Floor(float64(len(deltas)) * 0.5))
+	return time.Duration(deltas[midIndex]) * time.Millisecond
+}

pkg/timeseries/models.go

@@ -29,6 +29,9 @@ type TimeSeriesData struct {
 type TimeSeriesMeta struct {
 	Name string
 	Item *zabbix.Item
+
+	// Item update interval. nil means not supported intervals (flexible, schedule, etc)
+	Interval *time.Duration
 }
 
 type AggFunc = func(points []TimePoint) *float64

pkg/timeseries/timeseries.go

@@ -1,12 +1,9 @@
 package timeseries
 
 import (
-	"errors"
-	"math"
 	"sort"
 	"time"
 
-	"github.com/grafana/grafana-plugin-sdk-go/backend"
 	"github.com/grafana/grafana-plugin-sdk-go/data"
 )
 
@@ -286,94 +283,11 @@ func findNearestLeft(series TimeSeries, pointIndex int) *TimePoint {
 	return nil
 }
 
-// Aligns point's time stamps according to provided interval.
-func (ts TimeSeries) Align(interval time.Duration) TimeSeries {
-	if interval <= 0 || ts.Len() < 2 {
-		return ts
-	}
-
-	alignedTs := NewTimeSeries()
-	var frameTs = ts[0].GetTimeFrame(interval)
-	var pointFrameTs time.Time
-	var point TimePoint
-
-	for i := 1; i < ts.Len(); i++ {
-		point = ts[i]
-		pointFrameTs = point.GetTimeFrame(interval)
-
-		if pointFrameTs.After(frameTs) {
-			for frameTs.Before(pointFrameTs) {
-				alignedTs = append(alignedTs, TimePoint{Time: frameTs, Value: nil})
-				frameTs = frameTs.Add(interval)
-			}
-		}
-
-		alignedTs = append(alignedTs, TimePoint{Time: pointFrameTs, Value: point.Value})
-		frameTs = frameTs.Add(interval)
-	}
-
-	return alignedTs
-}
-
-// Detects interval between data points in milliseconds based on median delta between points.
-func (ts TimeSeries) DetectInterval() time.Duration {
-	if ts.Len() < 2 {
-		return 0
-	}
-
-	deltas := make([]int, 0)
-	for i := 1; i < ts.Len(); i++ {
-		delta := ts[i].Time.Sub(ts[i-1].Time)
-		deltas = append(deltas, int(delta.Milliseconds()))
-	}
-	sort.Ints(deltas)
-	midIndex := int(math.Floor(float64(len(deltas)) * 0.5))
-	return time.Duration(deltas[midIndex]) * time.Millisecond
-}
-
 // Gets point timestamp rounded according to provided interval.
 func (p *TimePoint) GetTimeFrame(interval time.Duration) time.Time {
 	return p.Time.Truncate(interval)
 }
 
-func alignDataPoints(frame *data.Frame, interval time.Duration) *data.Frame {
-	if interval <= 0 || frame.Rows() < 2 {
-		return frame
-	}
-
-	timeFieldIdx := getTimeFieldIndex(frame)
-	if timeFieldIdx < 0 {
-		return frame
-	}
-	var frameTs = getPointTimeFrame(getTimestampAt(frame, 0), interval)
-	var pointFrameTs *time.Time
-	var pointsInserted = 0
-
-	for i := 1; i < frame.Rows(); i++ {
-		pointFrameTs = getPointTimeFrame(getTimestampAt(frame, i), interval)
-		if pointFrameTs == nil || frameTs == nil {
-			continue
-		}
-
-		if pointFrameTs.After(*frameTs) {
-			for frameTs.Before(*pointFrameTs) {
-				insertAt := i + pointsInserted
-				err := insertNullPointAt(frame, *frameTs, insertAt)
-				if err != nil {
-					backend.Logger.Debug("Error inserting null point", "error", err)
-				}
-				*frameTs = frameTs.Add(interval)
-				pointsInserted++
-			}
-		}
-
-		setTimeAt(frame, *pointFrameTs, i+pointsInserted)
-		*frameTs = frameTs.Add(interval)
-	}
-
-	return frame
-}
-
 func getPointTimeFrame(ts *time.Time, interval time.Duration) *time.Time {
 	if ts == nil {
 		return nil
@@ -407,19 +321,6 @@ func getTimestampAt(frame *data.Frame, index int) *time.Time {
 	return &ts
 }
 
-func insertNullPointAt(frame *data.Frame, frameTs time.Time, index int) error {
-	for _, field := range frame.Fields {
-		if field.Type() == data.FieldTypeTime {
-			field.Insert(index, frameTs)
-		} else if field.Type().Nullable() {
-			field.Insert(index, nil)
-		} else {
-			return errors.New("field is not nullable")
-		}
-	}
-	return nil
-}
-
 func setTimeAt(frame *data.Frame, frameTs time.Time, index int) {
 	for _, field := range frame.Fields {
 		if field.Type() == data.FieldTypeTime {

pkg/zabbix/methods.go

@@ -243,7 +243,7 @@ func filterGroupsByQuery(items []Group, filter string) ([]Group, error) {
 
 func (ds *Zabbix) GetAllItems(ctx context.Context, hostids []string, appids []string, itemtype string) ([]*Item, error) {
 	params := ZabbixAPIParams{
-		"output":         []string{"itemid", "name", "key_", "value_type", "hostid", "status", "state"},
+		"output":         []string{"itemid", "name", "key_", "value_type", "hostid", "status", "state", "units", "valuemapid", "delay"},
 		"sortfield":      "name",
 		"webitems":       true,
 		"filter":         map[string]interface{}{},

pkg/zabbix/models.go

@@ -48,6 +48,9 @@ type Item struct {
 	Hosts      []ItemHost `json:"hosts,omitempty"`
 	Status     string     `json:"status,omitempty"`
 	State      string     `json:"state,omitempty"`
+	Delay      string     `json:"delay,omitempty"`
+	Units      string     `json:"units,omitempty"`
+	ValueMapID string     `json:"valuemapid,omitempty"`
 }
 
 type ItemHost struct {