func()

in exporter/collector/internal/normalization/standard_normalizer.go [50:112]

• 41 lines of code
• 10 McCabe index (conditional complexity)

func (s *standardNormalizer) NormalizeExponentialHistogramDataPoint(point pmetric.ExponentialHistogramDataPoint, identifier uint64) bool {
	start, hasStart := s.startCache.GetExponentialHistogramDataPoint(identifier)
	if !hasStart {
		if point.StartTimestamp() == 0 || !point.StartTimestamp().AsTime().Before(point.Timestamp().AsTime()) {
			// This is the first time we've seen this metric, or we received
			// an explicit reset point as described in
			// https://github.com/open-telemetry/opentelemetry-specification/blob/9555f9594c7ffe5dc333b53da5e0f880026cead1/specification/metrics/datamodel.md#resets-and-gaps
			// Record it in history and drop the point.
			s.startCache.SetExponentialHistogramDataPoint(identifier, point)
			s.previousCache.SetExponentialHistogramDataPoint(identifier, point)
			return false
		}
		// No normalization required, since we haven't cached anything, and the start TS is non-zero.
		return true
	}

	// TODO(#366): It is possible, but difficult to compare exponential
	// histograms with different scales. For now, treat a change in
	// scale as a reset.  Drop this point, and normalize against it for
	// subsequent points.
	if point.Scale() != start.Scale() {
		s.startCache.SetExponentialHistogramDataPoint(identifier, point)
		s.previousCache.SetExponentialHistogramDataPoint(identifier, point)
		return false
	}

	previous, hasPrevious := s.previousCache.GetExponentialHistogramDataPoint(identifier)
	if !hasPrevious {
		// This should never happen, but fall-back to the start point if we
		// don't find a previous point
		previous = start
	}
	if !point.StartTimestamp().AsTime().Before(point.Timestamp().AsTime()) ||
		(point.StartTimestamp() == 0 && lessThanExponentialHistogramDataPoint(point, previous)) {
		// This is a reset point, but we have seen this timeseries before, so we know the reset happened in the time period since the last point.
		// Assume the reset occurred at T - 1 ms, and leave the value untouched.
		point.SetStartTimestamp(pcommon.Timestamp(uint64(point.Timestamp()) - uint64(time.Millisecond)))
		s.previousCache.SetExponentialHistogramDataPoint(identifier, point)
		// For subsequent points, we don't want to modify the value, but we do
		// want to make the start timestamp match the point we write here.
		// Store a point with the same timestamps, but zero value to achieve
		// that behavior.
		zeroPoint := pmetric.NewExponentialHistogramDataPoint()
		zeroPoint.SetTimestamp(point.StartTimestamp())
		zeroPoint.SetScale(point.Scale())
		s.startCache.SetExponentialHistogramDataPoint(identifier, zeroPoint)
		return true
	}
	if !start.Timestamp().AsTime().Before(point.Timestamp().AsTime()) {
		// We found a cached start timestamp that wouldn't produce a valid point.
		// Drop it and log.
		s.log.Info(
			"data point being processed older than last recorded reset, will not be emitted",
			zap.String("lastRecordedReset", start.Timestamp().String()),
			zap.String("dataPoint", point.Timestamp().String()),
		)
		return false
	}
	// There was no reset, so normalize the point against the start point
	subtractExponentialHistogramDataPoint(point, start)
	s.previousCache.SetExponentialHistogramDataPoint(identifier, point)
	return true
}