transform

Documentation

Overview

Package transform contains all the logic to parse and execute queries against the underlying metric system.

Index

• Variables
• func MapMaker(name string, fun func(float64) float64) function.Function

Variables

var Bound = function.MakeFunction(
	"transform.bound",
	func(list api.SeriesList, lowerBound float64, upperBound float64) (api.SeriesList, error) {
		if lowerBound > upperBound {
			return api.SeriesList{}, boundError{lowerBound, upperBound}
		}
		return mapper(list, func(value float64) float64 {
			if value < lowerBound {
				return lowerBound
			}
			if value > upperBound {
				return upperBound
			}
			return value
		}), nil
	},
)

Bound replaces values which fall outside the given limits with the limits themselves. If the lowest bound exceeds the upper bound, an error is returned.

var Cumulative = function.MakeFunction(
	"transform.cumulative",
	func(list api.SeriesList, timerange api.Timerange) api.SeriesList {
		return transformEach(list, func(values []float64) []float64 {
			result := make([]float64, len(values))
			sum := 0.0
			for i := range values {

				if i == 0 {
					continue
				}

				if !math.IsNaN(values[i]) {
					sum += values[i]
				}
				result[i] = sum
			}
			return result
		})
	},
)

Cumulative computes the cumulative sum of the given values.

var Derivative = function.MakeFunction(
	"transform.derivative",
	func(listExpression function.Expression, context function.EvaluationContext) (api.SeriesList, error) {
		newContext := context.WithTimerange(context.Timerange().ExtendBefore(context.Timerange().Resolution()))
		list, err := function.EvaluateToSeriesList(listExpression, newContext)
		if err != nil {
			return api.SeriesList{}, err
		}
		resultList := api.SeriesList{
			Series: make([]api.Timeseries, len(list.Series)),
		}
		for seriesIndex, series := range list.Series {
			newValues := make([]float64, len(series.Values)-1)
			for i := range series.Values {
				if i == 0 {
					continue
				}

				newValues[i-1] = (series.Values[i] - series.Values[i-1]) / context.Timerange().Resolution().Seconds()
			}
			resultList.Series[seriesIndex] = api.Timeseries{
				Values: newValues,
				TagSet: series.TagSet,
			}
		}
		return resultList, nil
	},
	function.Option{Name: function.WidenBy, Value: function.Slot(1)},
)

Derivative is special because it needs to get one extra data point to the left This transform estimates the "change per second" between the two samples (scaled consecutive difference)

var ExponentialMovingAverage = function.MakeFunction(
	"transform.exponential_moving_average",
	func(context function.EvaluationContext, listExpression function.Expression, size time.Duration) (api.SeriesList, error) {
		if size < 0 {
			return api.SeriesList{}, fmt.Errorf("transform.exponential_moving_average must be given a non-negative duration")
		}

		scale := float64(size) / float64(context.Timerange().Resolution())
		extraPoints := int(scale + 0.5)
		timerange := context.Timerange()
		newTimerange := timerange.ExtendBefore(time.Duration(extraPoints) * timerange.Resolution())
		newContext := context.WithTimerange(newTimerange)

		list, err := function.EvaluateToSeriesList(listExpression, newContext)
		if err != nil {
			return api.SeriesList{}, err
		}

		alpha := math.Exp(math.Log(0.5) * context.Timerange().Resolution().Seconds() / size.Seconds())

		resultList := api.SeriesList{
			Series: make([]api.Timeseries, len(list.Series)),
		}
		for i := range resultList.Series {
			weight := 0.0
			sum := 0.0
			values := make([]float64, newTimerange.Slots())
			for t, y := range list.Series[i].Values {
				sum *= alpha
				weight *= alpha
				if !math.IsNaN(y) {
					sum += y
					weight++
				}
				values[t] = sum / weight
			}
			resultList.Series[i] = api.Timeseries{
				Values: values[newTimerange.Slots()-timerange.Slots():],
				TagSet: list.Series[i].TagSet,
			}
		}
		return resultList, nil
	},
	function.Option{Name: function.WidenBy, Value: function.Argument(1)},
)

var Integral = function.MakeFunction(
	"transform.integral",
	func(list api.SeriesList, timerange api.Timerange) api.SeriesList {
		return transformEach(list, func(values []float64) []float64 {
			result := make([]float64, len(values))
			integral := 0.0
			for i := range values {

				if i == 0 {
					continue
				}

				if !math.IsNaN(values[i]) {
					integral += values[i]
				}
				result[i] = integral * timerange.Resolution().Seconds()
			}
			return result
		})
	},
)

Integral integrates a series whose values are "X per millisecond" to estimate "total X so far" if the series represents "X in this sampling interval" instead, then you should use transformCumulative.

var LowerBound = function.MakeFunction(
	"transform.lower_bound",
	func(list api.SeriesList, lowerBound float64) (api.SeriesList, error) {
		return mapper(list, func(value float64) float64 {
			if value < lowerBound {
				return lowerBound
			}
			return value
		}), nil
	},
)

LowerBound replaces values that fall below the given bound with the lower bound.

var MovingAverage = function.MakeFunction(
	"transform.moving_average",
	func(context function.EvaluationContext, listExpression function.Expression, size time.Duration) (api.SeriesList, error) {
		if size < 0 {
			return api.SeriesList{}, fmt.Errorf("transform.moving_average must be given a non-negative duration")
		}

		limit := int(float64(size)/float64(context.Timerange().Resolution()) + 0.5)
		if limit < 1 {

			limit = 1
		}

		timerange := context.Timerange()
		newTimerange := timerange.ExtendBefore(time.Duration(limit-1) * timerange.Resolution())
		newContext := context.WithTimerange(newTimerange)

		list, err := function.EvaluateToSeriesList(listExpression, newContext)
		if err != nil {
			return api.SeriesList{}, err
		}

		for index, series := range list.Series {

			results := make([]float64, context.Timerange().Slots())
			count := 0
			sum := 0.0
			for i := range series.Values {

				if !math.IsNaN(series.Values[i]) {
					sum += series.Values[i]
					count++
				}

				if i >= limit && !math.IsNaN(series.Values[i-limit]) {
					sum -= series.Values[i-limit]
					count--
				}

				if i-limit+1 >= 0 {
					if count == 0 {
						results[i-limit+1] = math.NaN()
					} else {
						results[i-limit+1] = sum / float64(count)
					}
				}
			}
			list.Series[index].Values = results
		}
		return list, nil
	},
	function.Option{Name: function.WidenBy, Value: function.Argument(1)},
)

var NaNFill = function.MakeFunction(
	"transform.nan_fill",
	func(list api.SeriesList, defaultValue float64) api.SeriesList {
		return mapper(list, func(value float64) float64 {
			if math.IsNaN(value) {
				return defaultValue
			}
			return value
		})
	},
)

NaNFill will replacing missing data (NaN) with the `default` value supplied as a parameter.

var NaNKeepLast = function.MakeFunction(
	"transform.nan_keep_last",
	func(list api.SeriesList) api.SeriesList {
		return transformEach(list, func(values []float64) []float64 {
			result := make([]float64, len(values))
			for i := range result {
				result[i] = values[i]
				if math.IsNaN(values[i]) && i > 0 {
					result[i] = result[i-1]
				}
			}
			return result
		})
	},
)

NaNKeepLast will replace missing NaN data with the data before it

var Rate = function.MakeFunction(
	"transform.rate",
	func(listExpression function.Expression, context function.EvaluationContext) (api.SeriesList, error) {
		newContext := context.WithTimerange(context.Timerange().ExtendBefore(context.Timerange().Resolution()))
		list, err := function.EvaluateToSeriesList(listExpression, newContext)
		if err != nil {
			return api.SeriesList{}, err
		}
		resultList := api.SeriesList{
			Series: make([]api.Timeseries, len(list.Series)),
		}
		for seriesIndex, series := range list.Series {
			newValues := make([]float64, len(series.Values)-1)
			for i := range series.Values {
				if i == 0 {
					continue
				}

				newValues[i-1] = (series.Values[i] - series.Values[i-1]) / context.Timerange().Resolution().Seconds()
				if newValues[i-1] < 0 {
					newValues[i-1] = 0
				}
				if i+1 < len(series.Values) && series.Values[i-1] > series.Values[i] && series.Values[i] <= series.Values[i+1] {

					context.AddNote(fmt.Sprintf("Rate(%v): The underlying counter reset between %f, %f\n", series.TagSet, series.Values[i-1], series.Values[i]))

					newValues[i-1] = math.Max(series.Values[i], 0) / context.Timerange().Resolution().Seconds()
				}
			}
			resultList.Series[seriesIndex] = api.Timeseries{
				Values: newValues,
				TagSet: series.TagSet,
			}
		}
		return resultList, nil
	},
	function.Option{Name: function.WidenBy, Value: function.Slot(1)},
)

Rate is special because it needs to get one extra data point to the left. This transform functions mostly like Derivative but bounds the result to be positive. Specifically this function is designed for strictly increasing counters that only decrease when reset to zero. That is, thie function returns consecutive differences which are at least 0, or math.Max of the newly reported value and 0

var Timeshift = function.MakeFunction(
	"transform.timeshift",
	func(expression function.Expression, duration time.Duration, context function.EvaluationContext) (function.Value, error) {
		newContext := context.WithTimerange(context.Timerange().Shift(duration))
		return expression.Evaluate(newContext)
	},
	function.Option{Name: function.ShiftBy, Value: function.Argument(1)},
)

var UpperBound = function.MakeFunction(
	"transform.upper_bound",
	func(list api.SeriesList, upperBound float64) (api.SeriesList, error) {
		return mapper(list, func(value float64) float64 {
			if value > upperBound {
				return upperBound
			}
			return value
		}), nil
	},
)

UpperBound replaces values that fall below the given bound with the lower bound.

Functions

func MapMaker

func MapMaker(name string, fun func(float64) float64) function.Function

MapMaker can be used to use a function as a transform, such as 'math.Abs' (or similar):

`MapMaker(math.Abs)` is a transform function which can be used, e.g. with ApplyTransform

The name is used for error-checking purposes.