using Vector otlp sink

Signed-off-by: Vitalii Parfonov <[email protected]>

Author: vparfonov

api/observability/v1/output_types.go

@@ -1314,7 +1314,7 @@ type OTLPTuningSpec struct {
 	// Compression causes data to be compressed before sending over the network.
 	// It is an error if the compression type is not supported by the output.
 	//
-	// +kubebuilder:validation:Enum:=gzip;none
+	// +kubebuilder:validation:Enum:=gzip;snappy;zlib;zstd;none
 	// +operator-sdk:csv:customresourcedefinitions:type=spec,displayName="Compression",xDescriptors={"urn:alm:descriptor:com.tectonic.ui:text"}
 	Compression string `json:"compression,omitempty"`
 }

bundle/manifests/observability.openshift.io_clusterlogforwarders.yaml

@@ -3100,6 +3100,9 @@ spec:
                                 It is an error if the compression type is not supported by the output.
                               enum:
                               - gzip
+                              - snappy
+                              - zlib
+                              - zstd
                               - none
                               type: string
                             deliveryMode:

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

@@ -3100,6 +3100,9 @@ spec:
                                 It is an error if the compression type is not supported by the output.
                               enum:
                               - gzip
+                              - snappy
+                              - zlib
+                              - zstd
                               - none
                               type: string
                             deliveryMode:

internal/cmd/functional-benchmarker/config/options.go

@@ -17,7 +17,7 @@ import (
 
 const (
 	LogStressorImage = "quay.io/openshift-logging/cluster-logging-load-client:0.2"
-	imageVector      = "quay.io/openshift-logging/vector:6.0"
+	imageVector      = "quay.io/vparfono/vector:v0.46.1-rh"
 )
 
 type Options struct {

internal/generator/vector/output/otlp/otlp.go

@@ -28,7 +28,7 @@ type Otlp struct {
 	ComponentID string
 	Inputs      string
 	URI         string
-	common.RootMixin
+	Compression genhelper.OptionalPair
 }
 
 func (p Otlp) Name() string {
@@ -38,21 +38,20 @@ func (p Otlp) Name() string {
 func (p Otlp) Template() string {
 	return `{{define "` + p.Name() + `" -}}
 [sinks.{{.ComponentID}}]
-type = "http"
+type = "opentelemetry"
 inputs = {{.Inputs}}
-uri = "{{.URI}}"
-method = "post"
-payload_prefix = "{\"resourceLogs\":"
-payload_suffix = "}"
+protocol.uri = "{{.URI}}"
+protocol.type = "http"
+protocol.method = "post"
+protocol.encoding.codec = "json"
+protocol.encoding.except_fields = ["_internal"]
+protocol.payload_prefix = "{\"resourceLogs\":"
+protocol.payload_suffix = "}"
 {{.Compression}}
 {{end}}
 `
 }
 
-func (p *Otlp) SetCompression(algo string) {
-	p.Compression.Value = algo
-}
-
 const (
 	logSourceContainer    = string(obs.ApplicationSourceContainer)
 	logSourceNode         = string(obs.InfrastructureSourceNode)
@@ -145,20 +144,18 @@ func New(id string, o obs.OutputSpec, inputs []string, secrets observability.Sec
 	els = append(els, FormatResourceLog(formatResourceLogsID, reduceInputs))
 	// Create sink and wrap in `resourceLogs`
 	sink := Output(id, o, []string{formatResourceLogsID}, secrets, op)
-	if strategy != nil {
-		strategy.VisitSink(sink)
-	}
+
+	protocolId := id + ".protocol"
 	return MergeElements(
 		els,
 		[]Element{
 			sink,
-			common.NewEncoding(id, common.CodecJSON),
 			common.NewAcknowledgments(id, strategy),
-			common.NewBatch(id, strategy),
+			common.NewBatch(protocolId, strategy),
 			common.NewBuffer(id, strategy),
-			common.NewRequest(id, strategy),
+			common.NewRequest(protocolId, strategy),
 			tls.New(id, o.TLS, secrets, op),
-			auth.HTTPAuth(id, o.OTLP.Authentication, secrets, op),
+			auth.HTTPAuth(protocolId, o.OTLP.Authentication, secrets, op),
 		},
 	)
 }
@@ -180,10 +177,14 @@ func RouteBySource(id string, inputs []string, logSources []string) Element {
 }
 
 func Output(id string, o obs.OutputSpec, inputs []string, secrets observability.Secrets, op Options) *Otlp {
+	compression := genhelper.NewOptionalPair("protocol.compression", nil)
+	if o.OTLP.Tuning != nil && o.OTLP.Tuning.Compression != "" {
+		compression = genhelper.NewOptionalPair("protocol.compression", o.OTLP.Tuning.Compression)
+	}
 	return &Otlp{
 		ComponentID: id,
 		Inputs:      vectorhelpers.MakeInputs(inputs...),
 		URI:         o.OTLP.URL,
-		RootMixin:   common.NewRootMixin(nil),
+		Compression: compression,
 	}
 }

test/functional/outputs/otlp/container_test.go

@@ -125,6 +125,7 @@ var _ = Describe("[Functional][Outputs][OTLP] Functional tests", func() {
 		},
 			Entry("should pass with gzip", "gzip"),
 			Entry("should pass with zlib", "zlib"),
+			Entry("should pass with zstd", "zstd"),
 			Entry("should pass with no compression", "none"))
 	})
 })