Remote write queue config (#1517)

RemoteWriteSpec Expose QueueConfig Parameters (#1488)

Documentation/api.md

@@ -26,6 +26,7 @@ This Document documents the types introduced by the Prometheus Operator to be co
 * [PrometheusRuleSpec](#prometheusrulespec)
 * [PrometheusSpec](#prometheusspec)
 * [PrometheusStatus](#prometheusstatus)
+* [QueueConfig](#queueconfig)
 * [RelabelConfig](#relabelconfig)
 * [RemoteReadSpec](#remotereadspec)
 * [RemoteWriteSpec](#remotewritespec)
@@ -286,6 +287,22 @@ Most recent observed status of the Prometheus cluster. Read-only. Not included w
 
 [Back to TOC](#table-of-contents)
 
+## QueueConfig
+
+QueueConfig allows the tuning of remote_write queue_config parameters. This object is referenced in the RemoteWriteSpec object.
+
+| Field | Description | Scheme | Required |
+| ----- | ----------- | ------ | -------- |
+| capacity | Capacity is the number of samples to buffer per shard before we start dropping them. | int | false |
+| maxShards | MaxShards is the maximum number of shards, i.e. amount of concurrency. | int | false |
+| maxSamplesPerSend | MaxSamplesPerSend is the maximum number of samples per send. | int | false |
+| batchSendDeadline | BatchSendDeadline is the maximum time a sample will wait in buffer. | string | false |
+| maxRetries | MaxRetries is the maximum number of times to retry a batch on recoverable errors. | int | false |
+| minBackoff | MinBackoff is the initial retry delay. Gets doubled for every retry. | string | false |
+| maxBackoff | MaxBackoff is the maximum retry delay. | string | false |
+
+[Back to TOC](#table-of-contents)
+
 ## RelabelConfig
 
 RelabelConfig allows dynamic rewriting of the label set, being applied to samples before ingestion. It defines `<metric_relabel_configs>`-section of Prometheus configuration. More info: https://prometheus.io/docs/prometheus/latest/configuration/configuration/#metric_relabel_configs
@@ -334,6 +351,7 @@ RemoteWriteSpec defines the remote_write configuration for prometheus.
 | bearerTokenFile | File to read bearer token for remote write. | string | false |
 | tlsConfig | TLS Config to use for remote write. | *[TLSConfig](#tlsconfig) | false |
 | proxyUrl | Optional ProxyURL | string | false |
+| queueConfig | QueueConfig allows tuning of the remote write queue parameters. | *[QueueConfig](#queueconfig) | false |
 
 [Back to TOC](#table-of-contents)
 

contrib/kube-prometheus/Makefile

@@ -29,7 +29,7 @@ generate: manifests/** **.md
 	echo '>>> inside **.md'
 	$(EMBEDMD_BINARY) -w `find . -name "*.md" | grep -v vendor`
 
-manifests/**: vendor/** example.jsonnet
+manifests/**: vendor/** $(wildcard jsonnet/**/*) example.jsonnet
 	./build.sh
 
 vendor/**: $(JB_BINARY) jsonnetfile.json

example/prometheus-operator-crd/prometheus.crd.yaml

@@ -1802,6 +1802,42 @@ spec:
                   proxyUrl:
                     description: Optional ProxyURL
                     type: string
+                  queueConfig:
+                    description: QueueConfig allows the tuning of remote_write queue_config
+                      parameters. This object is referenced in the RemoteWriteSpec
+                      object.
+                    properties:
+                      batchSendDeadline:
+                        description: BatchSendDeadline is the maximum time a sample
+                          will wait in buffer.
+                        type: string
+                      capacity:
+                        description: Capacity is the number of samples to buffer per
+                          shard before we start dropping them.
+                        format: int32
+                        type: integer
+                      maxBackoff:
+                        description: MaxBackoff is the maximum retry delay.
+                        type: string
+                      maxRetries:
+                        description: MaxRetries is the maximum number of times to
+                          retry a batch on recoverable errors.
+                        format: int32
+                        type: integer
+                      maxSamplesPerSend:
+                        description: MaxSamplesPerSend is the maximum number of samples
+                          per send.
+                        format: int32
+                        type: integer
+                      maxShards:
+                        description: MaxShards is the maximum number of shards, i.e.
+                          amount of concurrency.
+                        format: int32
+                        type: integer
+                      minBackoff:
+                        description: MinBackoff is the initial retry delay. Gets doubled
+                          for every retry.
+                        type: string
                   remoteTimeout:
                     description: Timeout for requests to the remote write endpoint.
                     type: string

pkg/client/monitoring/v1/openapi_generated.go

@@ -1101,6 +1101,65 @@ func GetOpenAPIDefinitions(ref common.ReferenceCallback) map[string]common.OpenA
 			},
 			Dependencies: []string{},
 		},
+		"github.com/coreos/prometheus-operator/pkg/client/monitoring/v1.QueueConfig": {
+			Schema: spec.Schema{
+				SchemaProps: spec.SchemaProps{
+					Description: "QueueConfig allows the tuning of remote_write queue_config parameters. This object is referenced in the RemoteWriteSpec object.",
+					Properties: map[string]spec.Schema{
+						"capacity": {
+							SchemaProps: spec.SchemaProps{
+								Description: "Capacity is the number of samples to buffer per shard before we start dropping them.",
+								Type:        []string{"integer"},
+								Format:      "int32",
+							},
+						},
+						"maxShards": {
+							SchemaProps: spec.SchemaProps{
+								Description: "MaxShards is the maximum number of shards, i.e. amount of concurrency.",
+								Type:        []string{"integer"},
+								Format:      "int32",
+							},
+						},
+						"maxSamplesPerSend": {
+							SchemaProps: spec.SchemaProps{
+								Description: "MaxSamplesPerSend is the maximum number of samples per send.",
+								Type:        []string{"integer"},
+								Format:      "int32",
+							},
+						},
+						"batchSendDeadline": {
+							SchemaProps: spec.SchemaProps{
+								Description: "BatchSendDeadline is the maximum time a sample will wait in buffer.",
+								Type:        []string{"string"},
+								Format:      "",
+							},
+						},
+						"maxRetries": {
+							SchemaProps: spec.SchemaProps{
+								Description: "MaxRetries is the maximum number of times to retry a batch on recoverable errors.",
+								Type:        []string{"integer"},
+								Format:      "int32",
+							},
+						},
+						"minBackoff": {
+							SchemaProps: spec.SchemaProps{
+								Description: "MinBackoff is the initial retry delay. Gets doubled for every retry.",
+								Type:        []string{"string"},
+								Format:      "",
+							},
+						},
+						"maxBackoff": {
+							SchemaProps: spec.SchemaProps{
+								Description: "MaxBackoff is the maximum retry delay.",
+								Type:        []string{"string"},
+								Format:      "",
+							},
+						},
+					},
+				},
+			},
+			Dependencies: []string{},
+		},
 		"github.com/coreos/prometheus-operator/pkg/client/monitoring/v1.RelabelConfig": {
 			Schema: spec.Schema{
 				SchemaProps: spec.SchemaProps{
@@ -1312,12 +1371,18 @@ func GetOpenAPIDefinitions(ref common.ReferenceCallback) map[string]common.OpenA
 								Format:      "",
 							},
 						},
+						"queueConfig": {
+							SchemaProps: spec.SchemaProps{
+								Description: "QueueConfig allows tuning of the remote write queue parameters.",
+								Ref:         ref("github.com/coreos/prometheus-operator/pkg/client/monitoring/v1.QueueConfig"),
+							},
+						},
 					},
 					Required: []string{"url"},
 				},
 			},
 			Dependencies: []string{
-				"github.com/coreos/prometheus-operator/pkg/client/monitoring/v1.BasicAuth", "github.com/coreos/prometheus-operator/pkg/client/monitoring/v1.RelabelConfig", "github.com/coreos/prometheus-operator/pkg/client/monitoring/v1.TLSConfig"},
+				"github.com/coreos/prometheus-operator/pkg/client/monitoring/v1.BasicAuth", "github.com/coreos/prometheus-operator/pkg/client/monitoring/v1.QueueConfig", "github.com/coreos/prometheus-operator/pkg/client/monitoring/v1.RelabelConfig", "github.com/coreos/prometheus-operator/pkg/client/monitoring/v1.TLSConfig"},
 		},
 		"github.com/coreos/prometheus-operator/pkg/client/monitoring/v1.Rule": {
 			Schema: spec.Schema{

pkg/client/monitoring/v1/types.go

@@ -285,6 +285,28 @@ type RemoteWriteSpec struct {
 	TLSConfig *TLSConfig `json:"tlsConfig,omitempty"`
 	//Optional ProxyURL
 	ProxyURL string `json:"proxyUrl,omitempty"`
+	// QueueConfig allows tuning of the remote write queue parameters.
+	QueueConfig *QueueConfig `json:"queueConfig,omitempty"`
+}
+
+// QueueConfig allows the tuning of remote_write queue_config parameters. This object
+// is referenced in the RemoteWriteSpec object.
+// +k8s:openapi-gen=true
+type QueueConfig struct {
+	// Capacity is the number of samples to buffer per shard before we start dropping them.
+	Capacity int `json:"capacity,omitempty"`
+	// MaxShards is the maximum number of shards, i.e. amount of concurrency.
+	MaxShards int `json:"maxShards,omitempty"`
+	// MaxSamplesPerSend is the maximum number of samples per send.
+	MaxSamplesPerSend int `json:"maxSamplesPerSend,omitempty"`
+	// BatchSendDeadline is the maximum time a sample will wait in buffer.
+	BatchSendDeadline string `json:"batchSendDeadline,omitempty"`
+	// MaxRetries is the maximum number of times to retry a batch on recoverable errors.
+	MaxRetries int `json:"maxRetries,omitempty"`
+	// MinBackoff is the initial retry delay. Gets doubled for every retry.
+	MinBackoff string `json:"minBackoff,omitempty"`
+	// MaxBackoff is the maximum retry delay.
+	MaxBackoff string `json:"maxBackoff,omitempty"`
 }
 
 // RemoteReadSpec defines the remote_read configuration for prometheus.

pkg/client/monitoring/v1/zz_generated.deepcopy.go

@@ -709,6 +709,22 @@ func (in *PrometheusStatus) DeepCopy() *PrometheusStatus {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *QueueConfig) DeepCopyInto(out *QueueConfig) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new QueueConfig.
+func (in *QueueConfig) DeepCopy() *QueueConfig {
+	if in == nil {
+		return nil
+	}
+	out := new(QueueConfig)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *RelabelConfig) DeepCopyInto(out *RelabelConfig) {
 	*out = *in
@@ -799,6 +815,15 @@ func (in *RemoteWriteSpec) DeepCopyInto(out *RemoteWriteSpec) {
 			**out = **in
 		}
 	}
+	if in.QueueConfig != nil {
+		in, out := &in.QueueConfig, &out.QueueConfig
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(QueueConfig)
+			**out = **in
+		}
+	}
 	return
 }
 

pkg/prometheus/promcfg.go

@@ -710,6 +710,40 @@ func generateRemoteWriteConfig(version semver.Version, specs []v1.RemoteWriteSpe
 			cfg = append(cfg, yaml.MapItem{Key: "proxy_url", Value: spec.ProxyURL})
 		}
 
+		if spec.QueueConfig != nil {
+			queueConfig := yaml.MapSlice{}
+
+			if spec.QueueConfig.Capacity != int(0) {
+				queueConfig = append(queueConfig, yaml.MapItem{Key: "capacity", Value: spec.QueueConfig.Capacity})
+			}
+
+			if spec.QueueConfig.MaxShards != int(0) {
+				queueConfig = append(queueConfig, yaml.MapItem{Key: "max_shards", Value: spec.QueueConfig.MaxShards})
+			}
+
+			if spec.QueueConfig.MaxSamplesPerSend != int(0) {
+				queueConfig = append(queueConfig, yaml.MapItem{Key: "max_samples_per_send", Value: spec.QueueConfig.MaxSamplesPerSend})
+			}
+
+			if spec.QueueConfig.BatchSendDeadline != "" {
+				queueConfig = append(queueConfig, yaml.MapItem{Key: "batch_send_deadline", Value: spec.QueueConfig.BatchSendDeadline})
+			}
+
+			if spec.QueueConfig.MaxRetries != int(0) {
+				queueConfig = append(queueConfig, yaml.MapItem{Key: "max_retries", Value: spec.QueueConfig.MaxRetries})
+			}
+
+			if spec.QueueConfig.MinBackoff != "" {
+				queueConfig = append(queueConfig, yaml.MapItem{Key: "min_backoff", Value: spec.QueueConfig.MinBackoff})
+			}
+
+			if spec.QueueConfig.MaxBackoff != "" {
+				queueConfig = append(queueConfig, yaml.MapItem{Key: "max_backoff", Value: spec.QueueConfig.MaxBackoff})
+			}
+
+			cfg = append(cfg, yaml.MapItem{Key: "queue_config", Value: queueConfig})
+		}
+
 		cfgs = append(cfgs, cfg)
 	}