added functional tests

Author: AndrewChubatiuk

test/framework/functional/framework.go

@@ -427,8 +427,14 @@ func (f *CollectorFunctionalFramework) addOutputContainers(b *runtime.PodBuilder
 				return err
 			}
 		case obs.OutputTypeElasticsearch:
-			if err := f.AddESOutput(ElasticsearchVersion(output.Elasticsearch.Version), b, output, nil); err != nil {
-				return err
+			if len(output.Elasticsearch.Headers) == 0 {
+				if err := f.AddESOutput(ElasticsearchVersion(output.Elasticsearch.Version), b, output, nil); err != nil {
+					return err
+				}
+			} else {
+				if err := f.AddVLOutput(b, output, nil); err != nil {
+					return err
+				}
 			}
 		case obs.OutputTypeHTTP:
 			if err := f.AddVectorHttpOutput(b, output); err != nil {

test/framework/functional/output_victorialogs.go

@@ -0,0 +1,80 @@
+package functional
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"strings"
+
+	obs "github.com/openshift/cluster-logging-operator/api/observability/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+
+	"github.com/openshift/cluster-logging-operator/internal/generator/url"
+	"github.com/openshift/cluster-logging-operator/internal/runtime"
+
+	log "github.com/ViaQ/logerr/v2/log/static"
+)
+
+func (f *CollectorFunctionalFramework) AddVLOutput(b *runtime.PodBuilder, output obs.OutputSpec, args []string) error {
+	log.V(2).Info("Adding output for victorialogs", "name", output.Name)
+	name := strings.ToLower(output.Name)
+
+	esURL, err := url.Parse(output.Elasticsearch.URL)
+	if err != nil {
+		return err
+	}
+	transportPort, portError := determineTransportPort(esURL.Port())
+	if portError != nil {
+		return portError
+	}
+
+	log.V(2).Info("Adding container", "name", name)
+	log.V(2).Info("Adding VictoriaLogs output container", "name", obs.OutputTypeElasticsearch)
+
+	cmdArgs := []string{"-httpListenAddr=:" + transportPort}
+	if len(args) > 0 {
+		cmdArgs = append(cmdArgs, args...)
+	}
+
+	b.AddContainer(name, "victoriametrics/victoria-logs:1.34.0").
+		AddRunAsUser(2000).
+		WithCmdArgs(cmdArgs).
+		End().
+		AddContainer(fmt.Sprintf("%s-sidecar", name), "curlimages/curl:8.16.0").
+		AddRunAsUser(2000).
+		End()
+	return nil
+}
+
+func (f *CollectorFunctionalFramework) GetLogsFromVL(outputName string, headers map[string]string, options ...Option) (results []string, err error) {
+	port := "9428"
+	if found, o := OptionsInclude("port", options); found {
+		port = o.Value
+	}
+
+	pod := runtime.NewPod(f.Test.NS.Name, outputName)
+	if err = f.Test.Get(pod); err != nil {
+		pod = f.Pod
+	}
+
+	err = wait.PollUntilContextTimeout(context.TODO(), defaultRetryInterval, maxDuration, true, func(cxt context.Context) (done bool, err error) {
+		curlArgs := `-X GET -H "Content-Type: application/json"`
+		for k, v := range headers {
+			curlArgs += fmt.Sprintf(` -H "%s: %s"`, k, v)
+		}
+		cmd := fmt.Sprintf(`"localhost:%s/select/logsql/query?query='*'" %s`, port, curlArgs)
+		var result string
+		result, err = f.RunCommandInPod(pod, fmt.Sprintf("%s-sidecar", outputName), cmd)
+		if result != "" && err == nil {
+			scanner := bufio.NewScanner(strings.NewReader(result))
+			log.V(2).Info("results", "response", results)
+			for scanner.Scan() {
+				results = append(results, scanner.Text())
+			}
+		}
+		log.V(4).Info("Polling from VictoriaLogs", "err", err, "result", result)
+		return false, nil
+	})
+	log.V(4).Info("Returning", "logs", results)
+	return results, err
+}

test/functional/outputs/elasticsearch/forward_to_victorialogs_test.go

@@ -0,0 +1,80 @@
+package elasticsearch
+
+import (
+	"sort"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	obs "github.com/openshift/cluster-logging-operator/api/observability/v1"
+	"github.com/openshift/cluster-logging-operator/test/framework/functional"
+
+	"github.com/openshift/cluster-logging-operator/test/helpers/types"
+	obstestruntime "github.com/openshift/cluster-logging-operator/test/runtime/observability"
+)
+
+var _ = Describe("[Functional][Outputs][ElasticSearch] Logforwarding to VictoriaLogs", func() {
+
+	var (
+		framework *functional.CollectorFunctionalFramework
+
+		// Template expected as output Log
+		outputLogTemplate = functional.NewApplicationLogTemplate()
+	)
+
+	Context("should write to victorialogs", func() {
+		DescribeTable("with custom headers", func(headers map[string]string) {
+			outputLogTemplate.ViaqIndexName = "app-write"
+			framework = functional.NewCollectorFunctionalFramework()
+			obstestruntime.NewClusterLogForwarderBuilder(framework.Forwarder).
+				FromInput(obs.InputTypeApplication).
+				ToElasticSearchOutput(func(output *obs.OutputSpec) {
+					output.Elasticsearch.Headers = headers
+					output.Elasticsearch.URL = "http://0.0.0.0:9428/insert/elasticsearch/"
+				})
+			defer framework.Cleanup()
+			Expect(framework.Deploy()).To(BeNil())
+			timestamp := functional.CRIOTime(time.Now())
+			ukr := "привіт "
+			jp := "こんにちは "
+			ch := "你好"
+			msg := functional.NewCRIOLogMessage(timestamp, ukr+jp+ch, false)
+			Expect(framework.WriteMessagesToApplicationLog(msg, 1)).To(BeNil())
+			Expect(framework.WriteMessagesWithNotUTF8SymbolsToLog()).To(BeNil())
+			requestHeaders := map[string]string{
+				"AccountID": "0",
+				"ProjectID": "0",
+			}
+			for headerName := range requestHeaders {
+				if v, ok := headers[headerName]; ok {
+					requestHeaders[headerName] = v
+				}
+			}
+			raw, err := framework.GetLogsFromVL(string(obs.OutputTypeElasticsearch), requestHeaders)
+			Expect(err).To(BeNil(), "Expected no errors reading the logs")
+			Expect(raw).To(Not(BeEmpty()))
+			// Parse log line
+			var logs []types.ApplicationLog
+			err = types.StrictlyParseLogsFromSlice(raw, &logs)
+			Expect(err).To(BeNil(), "Expected no errors parsing the logs")
+			Expect(len(logs)).To(Equal(2))
+			//sort log by time before matching
+			sort.Slice(logs, func(i, j int) bool {
+				return logs[i].TimestampLegacy.Before(logs[j].TimestampLegacy)
+			})
+
+			Expect(logs[0].Message).To(Equal(ukr + jp + ch))
+			Expect(logs[1].Message).To(Equal("������������"))
+		},
+			Entry("Non-default tenant ID", map[string]string{
+				"AccountID":    "10",
+				"ProjectID":    "10",
+				"VL-Msg-Field": "message",
+			}),
+			Entry("Add extra fields", map[string]string{
+				"VL-Extra-Fields": "field_name=field_value",
+			}),
+		)
+	})
+})