added stream aggregation support for vmagent and vmsingle operator resources (#605)

* added stream aggregation support for vmagent and vmsingle operator resources

* added stream aggregation support for vmagent and vmsingle operator resources

* fix review comments

Author: Amper

api/v1beta1/additional.go

@@ -372,3 +372,81 @@ type ConfigMapKeyReference struct {
 	// The ConfigMap key to refer to.
 	Key string `json:"key"`
 }
+
+// StreamAggrConfig defines the stream aggregation config
+// +k8s:openapi-gen=true
+type StreamAggrConfig struct {
+	// Stream aggregation rules
+	Rules []StreamAggrRule `json:"rules"`
+	// Allows writing both raw and aggregate data
+	// +optional
+	KeepInput bool `json:"keepInput,omitempty"`
+	// Allows setting different de-duplication intervals per each configured remote storage
+	// +optional
+	DedupInterval string `json:"dedupInterval,omitempty"`
+}
+
+// StreamAggrRule defines the rule in stream aggregation config
+// +k8s:openapi-gen=true
+type StreamAggrRule struct {
+	// Match is a label selector for filtering time series for the given selector.
+	//
+	// If the match isn't set, then all the input time series are processed.
+	// +optional
+	Match string `json:"match,omitempty" yaml:"match,omitempty"`
+
+	// Interval is the interval between aggregations.
+	Interval string `json:"interval" yaml:"interval"`
+
+	// Outputs is a list of output aggregate functions to produce.
+	//
+	// The following names are allowed:
+	//
+	// - total - aggregates input counters
+	// - increase - counts the increase over input counters
+	// - count_series - counts the input series
+	// - count_samples - counts the input samples
+	// - sum_samples - sums the input samples
+	// - last - the last biggest sample value
+	// - min - the minimum sample value
+	// - max - the maximum sample value
+	// - avg - the average value across all the samples
+	// - stddev - standard deviation across all the samples
+	// - stdvar - standard variance across all the samples
+	// - histogram_bucket - creates VictoriaMetrics histogram for input samples
+	// - quantiles(phi1, ..., phiN) - quantiles' estimation for phi in the range [0..1]
+	//
+	// The output time series will have the following names:
+	//
+	//   input_name:aggr_<interval>_<output>
+	//
+	Outputs []string `json:"outputs"`
+
+	// By is an optional list of labels for grouping input series.
+	//
+	// See also Without.
+	//
+	// If neither By nor Without are set, then the Outputs are calculated
+	// individually per each input time series.
+	// +optional
+	By []string `json:"by,omitempty" yaml:"by,omitempty""`
+
+	// Without is an optional list of labels, which must be excluded when grouping input series.
+	//
+	// See also By.
+	//
+	// If neither By nor Without are set, then the Outputs are calculated
+	// individually per each input time series.
+	// +optional
+	Without []string `json:"without,omitempty" yaml:"without,omitempty"`
+
+	// InputRelabelConfigs is an optional relabeling rules, which are applied on the input
+	// before aggregation.
+	// +optional
+	InputRelabelConfigs []RelabelConfig `json:"input_relabel_configs,omitempty" yaml:"input_relabel_configs,omitempty"`
+
+	// OutputRelabelConfigs is an optional relabeling rules, which are applied
+	// on the aggregated output before being sent to remote storage.
+	// +optional
+	OutputRelabelConfigs []RelabelConfig `json:"output_relabel_configs,omitempty" yaml:"output_relabel_configs,omitempty"`
+}

api/v1beta1/vmagent_types.go

@@ -458,6 +458,9 @@ type VMAgentRemoteWriteSpec struct {
 	// vmagent supports since 1.79.0 version
 	// +optional
 	Headers []string `json:"headers,omitempty"`
+	// StreamAggrConfig defines stream aggregation configuration for VMAgent for -remoteWrite.url
+	// +optional
+	StreamAggrConfig *StreamAggrConfig `json:"streamAggrConfig,omitempty"`
 }
 
 // AsMapKey key for internal cache map
@@ -470,6 +473,16 @@ func (rw *VMAgentRemoteWriteSpec) AsSecretKey(idx int, suffix string) string {
 	return fmt.Sprintf("RWS_%d-SECRET-%s", idx, strings.ToUpper(suffix))
 }
 
+// AsConfigMapKey key for kubernetes configmap
+func (rw *VMAgentRemoteWriteSpec) AsConfigMapKey(idx int, suffix string) string {
+	return fmt.Sprintf("RWS_%d-CM-%s", idx, strings.ToUpper(suffix))
+}
+
+// HasStreamAggr returns true if stream aggregation is enabled for this remoteWrite
+func (rw *VMAgentRemoteWriteSpec) HasStreamAggr() bool {
+	return rw.StreamAggrConfig != nil && len(rw.StreamAggrConfig.Rules) > 0
+}
+
 // VmAgentStatus defines the observed state of VmAgent
 // +k8s:openapi-gen=true
 type VMAgentStatus struct {
@@ -588,6 +601,10 @@ func (cr VMAgent) RelabelingAssetName() string {
 	return fmt.Sprintf("relabelings-assets-vmagent-%s", cr.Name)
 }
 
+func (cr VMAgent) StreamAggrConfigName() string {
+	return fmt.Sprintf("stream-aggr-vmagent-%s", cr.Name)
+}
+
 func (cr VMAgent) HealthPath() string {
 	return buildPathWithPrefixFlag(cr.Spec.ExtraArgs, healthPath)
 }

api/v1beta1/vmsingle_types.go

@@ -187,6 +187,8 @@ type VMSingleSpec struct {
 	TerminationGracePeriodSeconds *int64 `json:"terminationGracePeriodSeconds,omitempty"`
 	// ReadinessGates defines pod readiness gates
 	ReadinessGates []v1.PodReadinessGate `json:"readinessGates,omitempty"`
+	// StreamAggrConfig defines stream aggregation configuration for VMSingle
+	StreamAggrConfig *StreamAggrConfig `json:"streamAggrConfig,omitempty"`
 }
 
 // UnmarshalJSON implements json.Unmarshaler interface
@@ -319,6 +321,10 @@ func (cr VMSingle) PrefixedName() string {
 	return fmt.Sprintf("vmsingle-%s", cr.Name)
 }
 
+func (cr VMSingle) StreamAggrConfigName() string {
+	return fmt.Sprintf("stream-aggr-vmsingle-%s", cr.Name)
+}
+
 func (cr VMSingle) MetricPath() string {
 	return buildPathWithPrefixFlag(cr.Spec.ExtraArgs, metricPath)
 }

api/v1beta1/zz_generated.deepcopy.go

@@ -372,3 +372,81 @@ type ConfigMapKeyReference struct {
 	// The ConfigMap key to refer to.
 	Key string `json:"key"`
 }
+
+// StreamAggrConfig defines the stream aggregation config
+// +k8s:openapi-gen=true
+type StreamAggrConfig struct {
+	// Stream aggregation rules
+	Rules []StreamAggrRule `json:"rules"`
+	// Allows writing both raw and aggregate data
+	// +optional
+	KeepInput bool `json:"keepInput,omitempty"`
+	// Allows setting different de-duplication intervals per each configured remote storage
+	// +optional
+	DedupInterval string `json:"dedupInterval,omitempty"`
+}
+
+// StreamAggrRule defines the rule in stream aggregation config
+// +k8s:openapi-gen=true
+type StreamAggrRule struct {
+	// Match is a label selector for filtering time series for the given selector.
+	//
+	// If the match isn't set, then all the input time series are processed.
+	// +optional
+	Match string `json:"match,omitempty" yaml:"match,omitempty"`
+
+	// Interval is the interval between aggregations.
+	Interval string `json:"interval" yaml:"interval"`
+
+	// Outputs is a list of output aggregate functions to produce.
+	//
+	// The following names are allowed:
+	//
+	// - total - aggregates input counters
+	// - increase - counts the increase over input counters
+	// - count_series - counts the input series
+	// - count_samples - counts the input samples
+	// - sum_samples - sums the input samples
+	// - last - the last biggest sample value
+	// - min - the minimum sample value
+	// - max - the maximum sample value
+	// - avg - the average value across all the samples
+	// - stddev - standard deviation across all the samples
+	// - stdvar - standard variance across all the samples
+	// - histogram_bucket - creates VictoriaMetrics histogram for input samples
+	// - quantiles(phi1, ..., phiN) - quantiles' estimation for phi in the range [0..1]
+	//
+	// The output time series will have the following names:
+	//
+	//   input_name:aggr_<interval>_<output>
+	//
+	Outputs []string `json:"outputs"`
+
+	// By is an optional list of labels for grouping input series.
+	//
+	// See also Without.
+	//
+	// If neither By nor Without are set, then the Outputs are calculated
+	// individually per each input time series.
+	// +optional
+	By []string `json:"by,omitempty" yaml:"by,omitempty""`
+
+	// Without is an optional list of labels, which must be excluded when grouping input series.
+	//
+	// See also By.
+	//
+	// If neither By nor Without are set, then the Outputs are calculated
+	// individually per each input time series.
+	// +optional
+	Without []string `json:"without,omitempty" yaml:"without,omitempty"`
+
+	// InputRelabelConfigs is an optional relabeling rules, which are applied on the input
+	// before aggregation.
+	// +optional
+	InputRelabelConfigs []RelabelConfig `json:"input_relabel_configs,omitempty" yaml:"input_relabel_configs,omitempty"`
+
+	// OutputRelabelConfigs is an optional relabeling rules, which are applied
+	// on the aggregated output before being sent to remote storage.
+	// +optional
+	OutputRelabelConfigs []RelabelConfig `json:"output_relabel_configs,omitempty" yaml:"output_relabel_configs,omitempty"`
+}

api/victoriametrics/v1beta1/additional.go

@@ -458,6 +458,9 @@ type VMAgentRemoteWriteSpec struct {
 	// vmagent supports since 1.79.0 version
 	// +optional
 	Headers []string `json:"headers,omitempty"`
+	// StreamAggrConfig defines stream aggregation configuration for VMAgent for -remoteWrite.url
+	// +optional
+	StreamAggrConfig *StreamAggrConfig `json:"streamAggrConfig,omitempty"`
 }
 
 // AsMapKey key for internal cache map
@@ -470,6 +473,16 @@ func (rw *VMAgentRemoteWriteSpec) AsSecretKey(idx int, suffix string) string {
 	return fmt.Sprintf("RWS_%d-SECRET-%s", idx, strings.ToUpper(suffix))
 }
 
+// AsConfigMapKey key for kubernetes configmap
+func (rw *VMAgentRemoteWriteSpec) AsConfigMapKey(idx int, suffix string) string {
+	return fmt.Sprintf("RWS_%d-CM-%s", idx, strings.ToUpper(suffix))
+}
+
+// HasStreamAggr returns true if stream aggregation is enabled for this remoteWrite
+func (rw *VMAgentRemoteWriteSpec) HasStreamAggr() bool {
+	return rw.StreamAggrConfig != nil && len(rw.StreamAggrConfig.Rules) > 0
+}
+
 // VmAgentStatus defines the observed state of VmAgent
 // +k8s:openapi-gen=true
 type VMAgentStatus struct {
@@ -588,6 +601,10 @@ func (cr VMAgent) RelabelingAssetName() string {
 	return fmt.Sprintf("relabelings-assets-vmagent-%s", cr.Name)
 }
 
+func (cr VMAgent) StreamAggrConfigName() string {
+	return fmt.Sprintf("stream-aggr-vmagent-%s", cr.Name)
+}
+
 func (cr VMAgent) HealthPath() string {
 	return buildPathWithPrefixFlag(cr.Spec.ExtraArgs, healthPath)
 }

api/victoriametrics/v1beta1/vmagent_types.go

@@ -187,6 +187,8 @@ type VMSingleSpec struct {
 	TerminationGracePeriodSeconds *int64 `json:"terminationGracePeriodSeconds,omitempty"`
 	// ReadinessGates defines pod readiness gates
 	ReadinessGates []v1.PodReadinessGate `json:"readinessGates,omitempty"`
+	// StreamAggrConfig defines stream aggregation configuration for VMSingle
+	StreamAggrConfig *StreamAggrConfig `json:"streamAggrConfig,omitempty"`
 }
 
 // UnmarshalJSON implements json.Unmarshaler interface
@@ -319,6 +321,10 @@ func (cr VMSingle) PrefixedName() string {
 	return fmt.Sprintf("vmsingle-%s", cr.Name)
 }
 
+func (cr VMSingle) StreamAggrConfigName() string {
+	return fmt.Sprintf("stream-aggr-vmsingle-%s", cr.Name)
+}
+
 func (cr VMSingle) MetricPath() string {
 	return buildPathWithPrefixFlag(cr.Spec.ExtraArgs, metricPath)
 }