1
2
3
4
5
6
7
8
9

// Handle counter resets:
resultFloat = samples.Floats[numSamplesMinusOne].F - samples.Floats[0].F
prevValue := samples.Floats[0].F
for _, currPoint := range samples.Floats[1:] {
        if currPoint.F < prevValue {
        resultFloat += prevValue
        }
        prevValue = currPoint.F
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

// === rate(node parser.ValueTypeMatrix) Vector ===
func funcRate(vals []parser.Value, args parser.Expressions, enh *EvalNodeHelper) Vector {
    return extrapolatedRate(vals, args, enh, true, true)
}
 
// === increase(node parser.ValueTypeMatrix) Vector ===
func funcIncrease(vals []parser.Value, args parser.Expressions, enh *EvalNodeHelper) Vector {
    return extrapolatedRate(vals, args, enh, true, false)
}
 
// extrapolatedRate is a utility function for rate/increase/delta.
// It calculates the rate (allowing for counter resets if isCounter is true),
// extrapolates if the first/last sample is close to the boundary, and returns
// the result as either per-second (if isRate is true) or overall.
func extrapolatedRate(vals []parser.Value, args parser.Expressions, enh *EvalNodeHelper, isCounter, isRate bool) Vector {
    ms := args[0].(*parser.MatrixSelector)
    vs := ms.VectorSelector.(*parser.VectorSelector)
    var (
        samples            = vals[0].(Matrix)[0]
        rangeStart         = enh.Ts - durationMilliseconds(ms.Range+vs.Offset)
        rangeEnd           = enh.Ts - durationMilliseconds(vs.Offset)
        resultFloat        float64
        resultHistogram    *histogram.FloatHistogram
        firstT, lastT      int64
        numSamplesMinusOne int
    )
 
    // We need either at least two Histograms and no Floats, or at least two
    // Floats and no Histograms to calculate a rate. Otherwise, drop this
    // Vector element.
    if len(samples.Histograms) > 0 && len(samples.Floats) > 0 {
        // Mix of histograms and floats. TODO(beorn7): Communicate this failure reason.
        return enh.Out
    }
 
    switch {
    case len(samples.Histograms) > 1:
        numSamplesMinusOne = len(samples.Histograms) - 1
        firstT = samples.Histograms[0].T
        lastT = samples.Histograms[numSamplesMinusOne].T
        resultHistogram = histogramRate(samples.Histograms, isCounter)
        if resultHistogram == nil {
            // The histograms are not compatible with each other.
            // TODO(beorn7): Communicate this failure reason.
            return enh.Out
        }
    case len(samples.Floats) > 1:
        numSamplesMinusOne = len(samples.Floats) - 1
        firstT = samples.Floats[0].T
        lastT = samples.Floats[numSamplesMinusOne].T
        resultFloat = samples.Floats[numSamplesMinusOne].F - samples.Floats[0].F
        if !isCounter {
            break
        }
        // Handle counter resets:
        prevValue := samples.Floats[0].F
        for _, currPoint := range samples.Floats[1:] {
            if currPoint.F < prevValue {
                resultFloat += prevValue
            }
            prevValue = currPoint.F
        }
    default:
        // Not enough samples. TODO(beorn7): Communicate this failure reason.
        return enh.Out
    }
 
    // Duration between first/last samples and boundary of range.
    durationToStart := float64(firstT-rangeStart) / 1000
    durationToEnd := float64(rangeEnd-lastT) / 1000
 
    sampledInterval := float64(lastT-firstT) / 1000
    averageDurationBetweenSamples := sampledInterval / float64(numSamplesMinusOne)
 
    // TODO(beorn7): Do this for histograms, too.
    if isCounter && resultFloat > 0 && len(samples.Floats) > 0 && samples.Floats[0].F >= 0 {
        // Counters cannot be negative. If we have any slope at all
        // (i.e. resultFloat went up), we can extrapolate the zero point
        // of the counter. If the duration to the zero point is shorter
        // than the durationToStart, we take the zero point as the start
        // of the series, thereby avoiding extrapolation to negative
        // counter values.
        durationToZero := sampledInterval * (samples.Floats[0].F / resultFloat)
        if durationToZero < durationToStart {
            durationToStart = durationToZero
        }
    }
 
    // If the first/last samples are close to the boundaries of the range,
    // extrapolate the result. This is as we expect that another sample
    // will exist given the spacing between samples we've seen thus far,
    // with an allowance for noise.
    extrapolationThreshold := averageDurationBetweenSamples * 1.1
    extrapolateToInterval := sampledInterval
 
    if durationToStart < extrapolationThreshold {
        extrapolateToInterval += durationToStart
    } else {
        extrapolateToInterval += averageDurationBetweenSamples / 2
    }
    if durationToEnd < extrapolationThreshold {
        extrapolateToInterval += durationToEnd
    } else {
        extrapolateToInterval += averageDurationBetweenSamples / 2
    }
    factor := extrapolateToInterval / sampledInterval
    if isRate {
        factor /= ms.Range.Seconds()
    }
    if resultHistogram == nil {
        resultFloat *= factor
    } else {
        resultHistogram.Mul(factor)
    }
 
    return append(enh.Out, Sample{F: resultFloat, H: resultHistogram})
}

1

TBD

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151

// === histogram_quantile(k parser.ValueTypeScalar, Vector parser.ValueTypeVector) Vector ===
func funcHistogramQuantile(vals []parser.Value, args parser.Expressions, enh *EvalNodeHelper) Vector {
    q := vals[0].(Vector)[0].F
    inVec := vals[1].(Vector)
 
    if enh.signatureToMetricWithBuckets == nil {
        enh.signatureToMetricWithBuckets = map[string]*metricWithBuckets{}
    } else {
        for _, v := range enh.signatureToMetricWithBuckets {
            v.buckets = v.buckets[:0]
        }
    }
 
    var histogramSamples []Sample
 
    for _, sample := range inVec {
        // We are only looking for conventional buckets here. Remember
        // the histograms for later treatment.
        if sample.H != nil {
            histogramSamples = append(histogramSamples, sample)
            continue
        }
 
        upperBound, err := strconv.ParseFloat(
            sample.Metric.Get(model.BucketLabel), 64,
        )
        if err != nil {
            // Oops, no bucket label or malformed label value. Skip.
            // TODO(beorn7): Issue a warning somehow.
            continue
        }
        enh.lblBuf = sample.Metric.BytesWithoutLabels(enh.lblBuf, labels.BucketLabel)
        mb, ok := enh.signatureToMetricWithBuckets[string(enh.lblBuf)]
        if !ok {
            sample.Metric = labels.NewBuilder(sample.Metric).
                Del(excludedLabels...).
                Labels()
 
            mb = &metricWithBuckets{sample.Metric, nil}
            enh.signatureToMetricWithBuckets[string(enh.lblBuf)] = mb
        }
        mb.buckets = append(mb.buckets, bucket{upperBound, sample.F})
 
    }
 
    // Now deal with the histograms.
    for _, sample := range histogramSamples {
        // We have to reconstruct the exact same signature as above for
        // a conventional histogram, just ignoring any le label.
        enh.lblBuf = sample.Metric.Bytes(enh.lblBuf)
        if mb, ok := enh.signatureToMetricWithBuckets[string(enh.lblBuf)]; ok && len(mb.buckets) > 0 {
            // At this data point, we have conventional histogram
            // buckets and a native histogram with the same name and
            // labels. Do not evaluate anything.
            // TODO(beorn7): Issue a warning somehow.
            delete(enh.signatureToMetricWithBuckets, string(enh.lblBuf))
            continue
        }
 
        enh.Out = append(enh.Out, Sample{
            Metric: enh.DropMetricName(sample.Metric),
            F:      histogramQuantile(q, sample.H),
        })
    }
 
    for _, mb := range enh.signatureToMetricWithBuckets {
        if len(mb.buckets) > 0 {
            enh.Out = append(enh.Out, Sample{
                Metric: mb.metric,
                F:      bucketQuantile(q, mb.buckets),
            })
        }
    }
 
    return enh.Out
}
 
// bucketQuantile calculates the quantile 'q' based on the given buckets. The
// buckets will be sorted by upperBound by this function (i.e. no sorting
// needed before calling this function). The quantile value is interpolated
// assuming a linear distribution within a bucket. However, if the quantile
// falls into the highest bucket, the upper bound of the 2nd highest bucket is
// returned. A natural lower bound of 0 is assumed if the upper bound of the
// lowest bucket is greater 0. In that case, interpolation in the lowest bucket
// happens linearly between 0 and the upper bound of the lowest bucket.
// However, if the lowest bucket has an upper bound less or equal 0, this upper
// bound is returned if the quantile falls into the lowest bucket.
//
// There are a number of special cases (once we have a way to report errors
// happening during evaluations of AST functions, we should report those
// explicitly):
//
// If 'buckets' has 0 observations, NaN is returned.
//
// If 'buckets' has fewer than 2 elements, NaN is returned.
//
// If the highest bucket is not +Inf, NaN is returned.
//
// If q==NaN, NaN is returned.
//
// If q<0, -Inf is returned.
//
// If q>1, +Inf is returned.
func bucketQuantile(q float64, buckets buckets) float64 {
    if math.IsNaN(q) {
        return math.NaN()
    }
    if q < 0 {
        return math.Inf(-1)
    }
    if q > 1 {
        return math.Inf(+1)
    }
    slices.SortFunc(buckets, func(a, b bucket) bool {
        return a.upperBound < b.upperBound
    })
    if !math.IsInf(buckets[len(buckets)-1].upperBound, +1) {
        return math.NaN()
    }
 
    buckets = coalesceBuckets(buckets)
    ensureMonotonic(buckets)
 
    if len(buckets) < 2 {
        return math.NaN()
    }
    observations := buckets[len(buckets)-1].count
    if observations == 0 {
        return math.NaN()
    }
    rank := q * observations
    b := sort.Search(len(buckets)-1, func(i int) bool { return buckets[i].count >= rank })
 
    if b == len(buckets)-1 {
        return buckets[len(buckets)-2].upperBound
    }
    if b == 0 && buckets[0].upperBound <= 0 {
        return buckets[0].upperBound
    }
    var (
        bucketStart float64
        bucketEnd   = buckets[b].upperBound
        count       = buckets[b].count
    )
    if b > 0 {
        bucketStart = buckets[b-1].upperBound
        count -= buckets[b-1].count
        rank -= buckets[b-1].count
    }
    return bucketStart + (bucketEnd-bucketStart)*(rank/count)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

func TestBucketQuantile__rate5m(t *testing.T) {
    q := 0.4
 
    myBuckets := buckets{
        {10, 5},
        {20, 8},
        {30, 12},
        {40, 15},
        {math.Inf(1), 15},
    }
 
    f := bucketQuantile(q, myBuckets)
    println("==== before rate =====")
    fmt.Println(myBuckets)
    fmt.Printf("%.2f\n", f)
 
    for i, myBucket := range myBuckets {
        myBuckets[i].count = myBucket.count / float64(5*60)
    }
    println("==== after rate =====")
    fmt.Println(myBuckets)
 
    f1 := bucketQuantile(q, myBuckets)
    fmt.Printf("%.2f\n", f1)
}