Merge pull request #90 from ndipebot/save-histogram

Add support for saving local samples as Exponential Histograms

pkg/api/collectorcontrollerv1alpha1/types.go

@@ -896,6 +896,31 @@ type Level struct {
 	// written locally to a file for retention.
 	// The will be set in the "name" field of the SampleList.
 	RetentionName string `json:"retentionName,omitempty" yaml:"retentionName,omitempty"`
+
+	// RetentionExponentialBuckets if RetentionName is specified and the current Operation is a histogram operation
+	// then this defines an exponential histogram for each resource's raw samples .
+	RetentionExponentialBuckets map[string]ExponentialBuckets `json:"retentionExponentialBuckets,omitempty" yaml:"retentionExponentialBuckets,omitempty"`
+}
+
+// ExponentialBuckets histogram buckets for saving raw samples
+type ExponentialBuckets struct {
+	// MinimumBase is the smallest exponent base that can be used for any histogram
+	MinimumBase float64 `json:"minimumBase" yaml:"minimumBase"`
+
+	// Compression power-of-two level of compression at which buckets from the MinBase are merged together to form larger buckets
+	// Compression = 0, base=2. This is the basic case.
+	// Compression < 0, each base2 bucket is subdivided into 2^Compression log scale sub-buckets.
+	// Compression > 0, every 2^Compression base2 buckets are merged.
+	Compression int64 `json:"compression" yaml:"compression"`
+
+	// ExponentOffset is offset to start bounds for exponential histogram
+	// base = MinimumBase ^ (2 ^ Compression)
+	// start bound = base ^ (index + ExponentOffset)
+	// end bound = base ^ (index + 1 + ExponentOffset)
+	ExponentOffset int64 `json:"exponentOffset" yaml:"exponentOffset"`
+
+	// SaveMaxOnly indicates that only max value should be saved for resource instead of full histogram
+	SaveMaxOnly bool `json:"saveMaxOnly,omitempty" yaml:"saveMaxOnly,omitempty"`
 }
 
 // LabelsMask defines which labels to keep at a level.

pkg/collector/collector.go

@@ -1235,14 +1235,25 @@ func (c *Collector) AggregateAndCollect(
 						slb.SampleList.MetricName = aggregatedName.String()
 						slb.SampleList.Name = l.RetentionName
 						slb.Mask = l.Mask
+
 						for mk, mv := range metric.Values {
 							s := slb.NewSample(mk)
 							s.Operation = aggregatedName.Operation.String()
 							s.Level = aggregatedName.Level.String()
-							slb.AddQuantityValues(s,
-								aggregatedName.Resource,
-								aggregatedName.Source, mv...)
+							// save samples in histogram format
+							if aggregatedName.Operation == collectorcontrollerv1alpha1.HistogramOperation && l.RetentionExponentialBuckets != nil {
+								slb.AddHistogramValues(s,
+									aggregatedName.Resource,
+									aggregatedName.Source,
+									l.RetentionExponentialBuckets[aggregatedName.Resource.String()],
+									mv...)
+							} else {
+								slb.AddQuantityValues(s,
+									aggregatedName.Resource,
+									aggregatedName.Source, mv...)
+							}
 						}
+
 						sCh <- slb.SampleList
 						c.publishTimer("metric_aggregation_per_aggregated_metric", ch, start, "local_save", name.String(), a.Name, l.Name, "aggregated_metric", aggregatedName.String())
 					}