support ndjson for http sink

Author: AndrewChubatiuk

api/observability/v1/output_types.go

@@ -636,6 +636,12 @@ type HTTP struct {
 	// +kubebuilder:validation:Optional
 	// +kubebuilder:validation:XValidation:rule="self == '' ||  isURL(self)", message="invalid URL"
 	ProxyURL string `json:"proxyURL,omitempty"`
+
+	// LinePerEvent uses NDJSON instead of JSON to send data to remote destination.
+	//
+	// +kubebuilder:validation:Optional
+	// +operator-sdk:csv:customresourcedefinitions:type=spec,displayName="Line Per Event"
+	LinePerEvent bool `json:"line_per_event,omitempty"`
 }
 
 type KafkaTuningSpec struct {

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

@@ -2391,6 +2391,10 @@ spec:
                           description: Headers specify optional headers to be sent
                             with the request
                           type: object
+                        line_per_event:
+                          description: LinePerEvent uses NDJSON instead of JSON to
+                            send data to remote destination.
+                          type: boolean
                         method:
                           description: Method specifies the HTTP method to be used
                             for sending logs. If not set, 'POST' is used.

docs/reference/operator/api_observability_v1.adoc

@@ -1276,7 +1276,7 @@ NOTE2: If this filter is used in a pipeline with GoogleCloudLogging, `.hostname`
 
 === .spec.filters[].prune.in[]
 
-FieldPath represents a path to find a value for a given field.  The format must a value that can be converted to a
+FieldPath represents a path to find a value for a given field.  The format must be a value that can be converted to a
 valid collector configuration. It is a dot delimited path to a field in the log record. It must start with a `.`.
 The path can contain alphanumeric characters and underscores (a-zA-Z0-9_).
 If segments contain characters outside of this range, the segment must be quoted.
@@ -1286,7 +1286,7 @@ Type:: array
 
 === .spec.filters[].prune.notIn[]
 
-FieldPath represents a path to find a value for a given field.  The format must a value that can be converted to a
+FieldPath represents a path to find a value for a given field.  The format must be a value that can be converted to a
 valid collector configuration. It is a dot delimited path to a field in the log record. It must start with a `.`.
 The path can contain alphanumeric characters and underscores (a-zA-Z0-9_).
 If segments contain characters outside of this range, the segment must be quoted.
@@ -2260,6 +2260,8 @@ The 'username@password' part of `url` is ignored.
 
 |headers|object|  Headers specify optional headers to be sent with the request
 
+|line_per_event|bool|  LinePerEvent uses NDJSON instead of JSON to send data to remote destination.
+
 |method|string|  Method specifies the HTTP method to be used for sending logs. If not set, 'POST' is used.
 
 |proxyURL|string|  ProxyURL URL of a HTTP or HTTPS proxy to be used instead of direct connection.
@@ -3147,6 +3149,39 @@ Example:
 
 3. foo.{.bar.baz||.qux.quux.corge||.grault||"nil"}-waldo.fred{.plugh||"none"}
 
+|indexedFields|array|  IndexedFields are the list of fields to be indexed by Splunk, increase storage usage, they should be used sparingly and only for high-value fields that provide significant search benefits.
+Nested fields are flattened into top-level fields.
+Field paths are joined using dot notation, and unsupported characters are replaced with underscores (_).
+Non-string values are automatically converted to strings (e.g., 3 → "3", true → "true").
+Object values serialized as JSON strings (e.g., { status: 200 } → "{\"status\":200}").
+
+Examples: [`.kubernetes`, `.log_type`, '.kubernetes.labels.foobar', `.kubernetes.labels."foo-bar/baz"`]
+|payloadKey|string|  PayloadKey specifies record field to use as payload.
+The PayloadKey must be a single field path.
+
+Field paths must only contain alphanumeric and underscores. Any field with other characters must be quoted.
+
+By default, payloadKey is not set, which means the complete log record is forwarded as the payload.
+Use payloadKey carefully. Selecting a single field as the payload may cause other important information in the
+log to be dropped, potentially leading to inconsistent or incomplete log events.
+
+Examples: `.kubernetes`, `.log_type`, '.kubernetes.labels.foobar', `.kubernetes.labels."foo-bar/baz"`
+
+|source|string|  Source identifies the origin of a log event.
+The Source can be a combination of static and dynamic values consisting of field paths followed by `||` followed by another field path or a static value.
+A dynamic value is encased in single curly brackets `{}` and MUST end with a static fallback value separated with `||`.
+Static values can only contain alphanumeric characters along with dashes, underscores, dots and forward slashes.
+If not specified will be detected according to .log_source and .log_type value.
+Details see in: docs/features/logforwarding/outputs/splunk-forwarding.adoc
+
+Example:
+
+1. foo-{.bar||"none"}
+
+2. {.foo||.bar||"missing"}
+
+3. foo.{.bar.baz||.qux.quux.corge||.grault||"nil"}-waldo.fred{.plugh||"none"}
+
 |tuning|object|  Tuning specs tuning for the output
 
 |======================
@@ -3181,6 +3216,16 @@ Type:: object
 
 |======================
 
+=== .spec.outputs[].splunk.indexedFields[]
+
+FieldPath represents a path to find a value for a given field.  The format must be a value that can be converted to a
+valid collector configuration. It is a dot delimited path to a field in the log record. It must start with a `.`.
+The path can contain alphanumeric characters and underscores (a-zA-Z0-9_).
+If segments contain characters outside of this range, the segment must be quoted.
+Examples: `.kubernetes.namespace_name`, `.log_type`, '.kubernetes.labels.foobar', `.kubernetes.labels."foo-bar/baz"`
+
+Type:: array
+
 === .spec.outputs[].splunk.tuning
 
 Type:: object
@@ -3255,6 +3300,15 @@ uucp cron authpriv ftp ntp security console solaris-cron
 
 local0 local1 local2 local3 local4 local5 local6 local7
 
+The Facility can be a combination of static and dynamic values consisting of field paths followed by `||` followed by another field path or a static value.
+A dynamic value is encased in single curly brackets `{}` and MUST end with a static fallback value separated with `||`.
+
+Static values can only contain alphanumeric characters along with dashes, underscores, dots and forward slashes.
+
+Example:
+
+1. {.foo||"user"}
+
 |msgId|string|  MsgId is MSGID part of the syslog-msg header. This supports template syntax to allow dynamic per-event values.
 
 The MsgId can be a combination of static and dynamic values consisting of field paths followed by `||` followed by another field path or a static value.
@@ -3321,6 +3375,15 @@ The value can be a decimal integer or one of these case-insensitive keywords:
 
 Emergency Alert Critical Error Warning Notice Informational Debug
 
+The Severity can be a combination of static and dynamic values consisting of field paths followed by `||` followed by another field path or a static value.
+A dynamic value is encased in single curly brackets `{}` and MUST end with a static fallback value separated with `||`.
+
+Static values can only contain alphanumeric characters along with dashes, underscores, dots and forward slashes.
+
+Example:
+
+1. {.foo||"Error"}
+
 |tuning|object|  Tuning specs tuning for the output
 
 |url|string|  An absolute URL, with a scheme. Valid schemes are: `tcp`, `tls`, `udp`

internal/generator/vector/output/http/http.go

@@ -14,11 +14,12 @@ import (
 )
 
 type Http struct {
-	ComponentID string
-	Inputs      string
-	URI         string
-	Method      string
-	Proxy       string
+	ComponentID  string
+	Inputs       string
+	URI          string
+	Method       string
+	Proxy        string
+	LinePerEvent bool
 	common.RootMixin
 }
 
@@ -33,6 +34,9 @@ type = "http"
 inputs = {{.Inputs}}
 uri = "{{.URI}}"
 method = "{{.Method}}"
+{{with .LinePerEvent}}
+framing.method = "newline_delimited"
+{{end}}
 {{with .Proxy -}}
 proxy.enabled = true
 proxy.http = "{{.}}"
@@ -76,12 +80,13 @@ func New(id string, o obs.OutputSpec, inputs []string, secrets observability.Sec
 
 func Output(id string, o obs.OutputSpec, inputs []string, secrets observability.Secrets, op Options) *Http {
 	return &Http{
-		ComponentID: id,
-		Inputs:      vectorhelpers.MakeInputs(inputs...),
-		URI:         o.HTTP.URL,
-		Method:      Method(o.HTTP),
-		Proxy:       o.HTTP.ProxyURL,
-		RootMixin:   common.NewRootMixin(nil),
+		ComponentID:  id,
+		Inputs:       vectorhelpers.MakeInputs(inputs...),
+		URI:          o.HTTP.URL,
+		Method:       Method(o.HTTP),
+		Proxy:        o.HTTP.ProxyURL,
+		LinePerEvent: o.HTTP.LinePerEvent,
+		RootMixin:    common.NewRootMixin(nil),
 	}
 }
 

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

@@ -143,6 +143,9 @@ var _ = Describe("Generate vector config", func() {
 					BaseOutputTuningSpec: *baseTune,
 				}
 			}, secrets, true, framework.NoOptions, "http_with_tuning.toml"),
+			Entry("with ndjson", func(spec *obs.OutputSpec) {
+				spec.HTTP.LinePerEvent = true
+			}, secrets, true, framework.NoOptions, "http_with_ndjson.toml"),
 			Entry("with proxy", func(spec *obs.OutputSpec) {
 				spec.HTTP.ProxyURL = "http://somewhere.org/proxy"
 				spec.HTTP.Headers = nil

internal/generator/vector/output/http/http_with_ndjson.toml

@@ -0,0 +1,20 @@
+[sinks.http_receiver]
+type = "http"
+inputs = ["application"]
+uri = "https://my-logstore.com"
+method = "post"
+framing.method = "newline_delimited"
+
+[sinks.http_receiver.encoding]
+codec = "json"
+except_fields = ["_internal"]
+
+[sinks.http_receiver.request]
+headers = {"h1"="v1","h2"="v2"}
+
+[sinks.http_receiver.auth]
+strategy = "basic"
+user = "SECRET[kubernetes_secret.http-receiver/username]"
+password = "SECRET[kubernetes_secret.http-receiver/password]"
+
+