Merge pull request #2203 from jcantrill/log4597

LOG-4597: modify vector buffer and request to have mostly parity with…

apis/logging/v1/output_types.go

@@ -286,7 +286,7 @@ type Http struct {
 
 	// Timeout specifies the Http request timeout in seconds. If not set, 10secs is used.
 	// +optional
-	Timeout string `json:"timeout,omitempty"`
+	Timeout int `json:"timeout,omitempty"`
 
 	// Method specifies the Http method to be used for sending logs. If not set, 'POST' is used.
 	// +kubebuilder:validation:Enum:=GET;HEAD;POST;PUT;DELETE;OPTIONS;TRACE;PATCH

bundle/manifests/clusterlogging.clusterserviceversion.yaml

@@ -118,7 +118,7 @@ metadata:
     certified: "false"
     console.openshift.io/plugins: '["logging-view-plugin"]'
     containerImage: quay.io/openshift-logging/cluster-logging-operator:latest
-    createdAt: "2023-10-11T17:54:55Z"
+    createdAt: "2023-10-12T13:12:40Z"
     description: The Red Hat OpenShift Logging Operator for OCP provides a means for
       configuring and managing your aggregated logging stack.
     olm.skipRange: '>=5.6.0-0 <5.8.0'

bundle/manifests/logging.openshift.io_clusterlogforwarders.yaml

@@ -495,7 +495,7 @@ spec:
                         timeout:
                           description: Timeout specifies the Http request timeout
                             in seconds. If not set, 10secs is used.
-                          type: string
+                          type: integer
                       type: object
                     kafka:
                       description: 'Kafka provides optional extra properties for `type:

config/crd/bases/logging.openshift.io_clusterlogforwarders.yaml

@@ -496,7 +496,7 @@ spec:
                         timeout:
                           description: Timeout specifies the Http request timeout
                             in seconds. If not set, 10secs is used.
-                          type: string
+                          type: integer
                       type: object
                     kafka:
                       description: 'Kafka provides optional extra properties for `type:

internal/generator/fluentd/output/http/http_test.go

@@ -98,7 +98,7 @@ var _ = Describe("Generate fluentd config", func() {
 						Name: "http-receiver",
 						URL:  "https://my-logstore.com/logs/app-logs",
 						OutputTypeSpec: v1.OutputTypeSpec{Http: &v1.Http{
-							Timeout: "50",
+							Timeout: 50,
 							Headers: map[string]string{
 								"k1": "v1",
 								"k2": "v2",
@@ -170,7 +170,7 @@ var _ = Describe("Generate fluentd config", func() {
 						Name: "http-receiver",
 						URL:  "https://my-logstore.com/logs/app-logs",
 						OutputTypeSpec: v1.OutputTypeSpec{Http: &v1.Http{
-							Timeout: "50",
+							Timeout: 50,
 							Headers: map[string]string{
 								"k1": "v1",
 								"k2": "v2",
@@ -253,7 +253,7 @@ var _ = Describe("Generate fluentd config", func() {
 						Name: "http-receiver",
 						URL:  "https://my-logstore.com/logs/app-logs",
 						OutputTypeSpec: v1.OutputTypeSpec{Http: &v1.Http{
-							Timeout: "50",
+							Timeout: 50,
 							Headers: map[string]string{
 								"Content-Type": "application/json",
 								"k1":           "v1",

internal/generator/helpers/optional_pair.go

@@ -0,0 +1,29 @@
+package helpers
+
+import (
+	"fmt"
+	"reflect"
+)
+
+type OptionalPair struct {
+	key   string
+	Value interface{}
+}
+
+func NewOptionalPair(key string, value interface{}) OptionalPair {
+	return OptionalPair{
+		key,
+		value,
+	}
+}
+
+func (op OptionalPair) String() string {
+	if op.Value == nil {
+		return ""
+	}
+	format := "%s = %v"
+	if reflect.TypeOf(op.Value).Kind() == reflect.String {
+		format = "%s = %q"
+	}
+	return fmt.Sprintf(format, op.key, op.Value)
+}

internal/generator/helpers/optional_pair_test.go

@@ -0,0 +1,34 @@
+package helpers
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("OptionalPair", func() {
+	Context("#Template", func() {
+		var (
+			pair OptionalPair
+		)
+		It("should return an empty template when the value is nil", func() {
+			pair = NewOptionalPair("abc", nil)
+			Expect(pair.String()).To(BeEmpty())
+		})
+		It("should return a formatted string config when value is a string", func() {
+			pair = NewOptionalPair("abc", "xyz")
+			Expect(pair.String()).To(Equal(`abc = "xyz"`))
+		})
+		It("should return a formatted numerical config when value is an int", func() {
+			pair = NewOptionalPair("abc", 123)
+			Expect(pair.String()).To(Equal(`abc = 123`))
+		})
+		It("should return a formatted bool config when value is bool", func() {
+			pair = NewOptionalPair("abc", true)
+			Expect(pair.String()).To(Equal(`abc = true`))
+		})
+		It("should return a formatted false bool config when value is bool", func() {
+			pair = NewOptionalPair("abc", false)
+			Expect(pair.String()).To(Equal(`abc = false`))
+		})
+	})
+})

internal/generator/helpers/suite_test.go

@@ -0,0 +1,13 @@
+package helpers
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestSuite(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "[internal][generator][helpers] Suite")
+}

internal/generator/utils/utils.go

@@ -7,6 +7,9 @@ import (
 )
 
 func ToHeaderStr(h map[string]string, formatStr string) string {
+	if len(h) == 0 {
+		return ""
+	}
 	sortedKeys := make([]string, len(h))
 	i := 0
 	for k := range h {

internal/generator/vector/conf_test.go

@@ -73,15 +73,6 @@ var _ = Describe("Testing Complete Config Generation", func() {
 			Options: generator.Options{
 				generator.ClusterTLSProfileSpec: tls.GetClusterTLSProfileSpec(nil),
 			},
-			CLSpec: logging.CollectionSpec{
-				Fluentd: &logging.FluentdForwarderSpec{
-					Buffer: &logging.FluentdBufferSpec{
-						ChunkLimitSize: "8m",
-						TotalLimitSize: "800000000",
-						OverflowAction: "throw_exception",
-					},
-				},
-			},
 			CLFSpec: logging.ClusterLogForwarderSpec{
 				Inputs: []logging.InputSpec{
 					{

internal/generator/vector/conf_test/complex.toml

@@ -381,6 +381,9 @@ topic = "topic"
 codec = "json"
 timestamp_format = "rfc3339"
 
+[sinks.kafka_receiver.buffer]
+when_full = "drop_newest"
+
 [sinks.kafka_receiver.tls]
 enabled = true
 min_tls_version = "VersionTLS12"

internal/generator/vector/conf_test/complex_custom_data_dir.toml

@@ -382,6 +382,9 @@ topic = "topic"
 codec = "json"
 timestamp_format = "rfc3339"
 
+[sinks.kafka_receiver.buffer]
+when_full = "drop_newest"
+
 [sinks.kafka_receiver.tls]
 enabled = true
 min_tls_version = "VersionTLS12"

internal/generator/vector/conf_test/complex_es_no_ver.toml

@@ -421,10 +421,16 @@ endpoints = ["https://es-1.svc.messaging.cluster.local:9200"]
 bulk.index = "{{ write_index }}"
 bulk.action = "create"
 encoding.except_fields = ["write_index"]
-request.timeout_secs = 2147483648
 id_key = "_id"
 api_version = "v6"
 
+[sinks.es_1.buffer]
+when_full = "drop_newest"
+
+[sinks.es_1.request]
+retry_attempts = 17
+timeout_secs = 2147483648
+
 [sinks.es_1.tls]
 min_tls_version = "VersionTLS12"
 ciphersuites = "TLS_AES_128_GCM_SHA256,TLS_AES_256_GCM_SHA384,TLS_CHACHA20_POLY1305_SHA256,ECDHE-ECDSA-AES128-GCM-SHA256,ECDHE-RSA-AES128-GCM-SHA256,ECDHE-ECDSA-AES256-GCM-SHA384,ECDHE-RSA-AES256-GCM-SHA384,ECDHE-ECDSA-CHACHA20-POLY1305,ECDHE-RSA-CHACHA20-POLY1305,DHE-RSA-AES128-GCM-SHA256,DHE-RSA-AES256-GCM-SHA384"
@@ -535,10 +541,17 @@ endpoints = ["https://es-2.svc.messaging.cluster.local:9200"]
 bulk.index = "{{ write_index }}"
 bulk.action = "create"
 encoding.except_fields = ["write_index"]
-request.timeout_secs = 2147483648
+
 id_key = "_id"
 api_version = "v6"
 
+[sinks.es_2.buffer]
+when_full = "drop_newest"
+
+[sinks.es_2.request]
+retry_attempts = 17
+timeout_secs = 2147483648
+
 [sinks.es_2.tls]
 min_tls_version = "VersionTLS12"
 ciphersuites = "TLS_AES_128_GCM_SHA256,TLS_AES_256_GCM_SHA384,TLS_CHACHA20_POLY1305_SHA256,ECDHE-ECDSA-AES128-GCM-SHA256,ECDHE-RSA-AES128-GCM-SHA256,ECDHE-ECDSA-AES256-GCM-SHA384,ECDHE-RSA-AES256-GCM-SHA384,ECDHE-ECDSA-CHACHA20-POLY1305,ECDHE-RSA-CHACHA20-POLY1305,DHE-RSA-AES128-GCM-SHA256,DHE-RSA-AES256-GCM-SHA384"

internal/generator/vector/conf_test/complex_es_v6.toml

@@ -420,10 +420,16 @@ endpoints = ["https://elasticsearch:9200"]
 bulk.index = "{{ write_index }}"
 bulk.action = "create"
 encoding.except_fields = ["write_index"]
-request.timeout_secs = 2147483648
 id_key = "_id"
 api_version = "v6"
 
+[sinks.default.buffer]
+when_full = "drop_newest"
+
+[sinks.default.request]
+retry_attempts = 17
+timeout_secs = 2147483648
+
 [sinks.default.tls]
 key_file = "/var/run/ocp-collector/secrets/collector/tls.key"
 crt_file = "/var/run/ocp-collector/secrets/collector/tls.crt"
@@ -537,10 +543,16 @@ endpoints = ["https://es-1.svc.messaging.cluster.local:9200"]
 bulk.index = "{{ write_index }}"
 bulk.action = "create"
 encoding.except_fields = ["write_index"]
-request.timeout_secs = 2147483648
 id_key = "_id"
 api_version = "v6"
 
+[sinks.es_1.buffer]
+when_full = "drop_newest"
+
+[sinks.es_1.request]
+retry_attempts = 17
+timeout_secs = 2147483648
+
 [sinks.es_1.tls]
 min_tls_version = "VersionTLS12"
 ciphersuites = "TLS_AES_128_GCM_SHA256,TLS_AES_256_GCM_SHA384,TLS_CHACHA20_POLY1305_SHA256,ECDHE-ECDSA-AES128-GCM-SHA256,ECDHE-RSA-AES128-GCM-SHA256,ECDHE-ECDSA-AES256-GCM-SHA384,ECDHE-RSA-AES256-GCM-SHA384,ECDHE-ECDSA-CHACHA20-POLY1305,ECDHE-RSA-CHACHA20-POLY1305,DHE-RSA-AES128-GCM-SHA256,DHE-RSA-AES256-GCM-SHA384"
@@ -655,10 +667,16 @@ endpoints = ["https://es-2.svc.messaging.cluster.local:9200"]
 bulk.index = "{{ write_index }}"
 bulk.action = "create"
 encoding.except_fields = ["write_index"]
-request.timeout_secs = 2147483648
 id_key = "_id"
 api_version = "v8"
 
+[sinks.es_2.buffer]
+when_full = "drop_newest"
+
+[sinks.es_2.request]
+retry_attempts = 17
+timeout_secs = 2147483648
+
 [sinks.es_2.tls]
 min_tls_version = "VersionTLS12"
 ciphersuites = "TLS_AES_128_GCM_SHA256,TLS_AES_256_GCM_SHA384,TLS_CHACHA20_POLY1305_SHA256,ECDHE-ECDSA-AES128-GCM-SHA256,ECDHE-RSA-AES128-GCM-SHA256,ECDHE-ECDSA-AES256-GCM-SHA384,ECDHE-RSA-AES256-GCM-SHA384,ECDHE-ECDSA-CHACHA20-POLY1305,ECDHE-RSA-CHACHA20-POLY1305,DHE-RSA-AES128-GCM-SHA256,DHE-RSA-AES256-GCM-SHA384"

internal/generator/vector/conf_test/complex_otel.toml

@@ -427,7 +427,11 @@ method = "post"
 [sinks.http_receiver.encoding]
 codec = "json"
 
+[sinks.http_receiver.buffer]
+when_full = "drop_newest"
+
 [sinks.http_receiver.request]
+retry_attempts = 17
 timeout_secs = 10
 headers = {"h1"="v1","h2"="v2"}
 

internal/generator/vector/conf_test/es_pipeline_w_spaces.toml

@@ -421,10 +421,16 @@ endpoints = ["https://es-1.svc.messaging.cluster.local:9200"]
 bulk.index = "{{ write_index }}"
 bulk.action = "create"
 encoding.except_fields = ["write_index"]
-request.timeout_secs = 2147483648
 id_key = "_id"
 api_version = "v6"
 
+[sinks.es_1.buffer]
+when_full = "drop_newest"
+
+[sinks.es_1.request]
+retry_attempts = 17
+timeout_secs = 2147483648
+
 [sinks.es_1.tls]
 min_tls_version = "VersionTLS12"
 ciphersuites = "TLS_AES_128_GCM_SHA256,TLS_AES_256_GCM_SHA384,TLS_CHACHA20_POLY1305_SHA256,ECDHE-ECDSA-AES128-GCM-SHA256,ECDHE-RSA-AES128-GCM-SHA256,ECDHE-ECDSA-AES256-GCM-SHA384,ECDHE-RSA-AES256-GCM-SHA384,ECDHE-ECDSA-CHACHA20-POLY1305,ECDHE-RSA-CHACHA20-POLY1305,DHE-RSA-AES128-GCM-SHA256,DHE-RSA-AES256-GCM-SHA384"

internal/generator/vector/elements/keyval.go

@@ -13,6 +13,9 @@ func (kv KeyVal) Name() string {
 }
 
 func (kv KeyVal) Template() string {
+	if kv.Val == "" {
+		return `{{define "` + kv.Name() + `" -}}{{end -}}`
+	}
 	return `{{define "` + kv.Name() + `" -}}
 {{.Key}} = {{.Val}}
 {{end -}}`

internal/generator/vector/output/buffer.go

@@ -0,0 +1,24 @@
+package output
+
+type Buffer struct {
+	ComponentID string
+	WhenFull    string
+}
+
+func NewBuffer(id string) Buffer {
+	return Buffer{
+		ComponentID: id,
+		WhenFull:    "drop_newest",
+	}
+}
+
+func (b Buffer) Name() string {
+	return "buffer"
+}
+
+func (b Buffer) Template() string {
+	return `{{define "` + b.Name() + `" -}}
+[sinks.{{.ComponentID}}.buffer]
+when_full = "{{.WhenFull}}"
+{{end}}`
+}