filebeat benchmark input and discard output

CHANGELOG.next.asciidoc

@@ -203,6 +203,8 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Update CEL mito extensions to v1.10.0 to add keys/values helper. {pull}38504[38504]
 - Add support for Active Directory an entity analytics provider. {pull}37919[37919]
 - Add debugging breadcrumb to logs when writing request trace log. {pull}38636[38636]
+- added benchmark input {pull}37437[37437]
+- added benchmark input and discard output {pull}37437[37437]
 
 *Auditbeat*
 

filebeat/docs/filebeat-options.asciidoc

@@ -70,6 +70,7 @@ You can configure {beatname_uc} to use the following inputs:
 * <<{beatname_lc}-input-aws-s3>>
 * <<{beatname_lc}-input-azure-eventhub>>
 * <<{beatname_lc}-input-azure-blob-storage>>
+* <<{beatname_lc}-input-benchmark>>
 * <<{beatname_lc}-input-cel>>
 * <<{beatname_lc}-input-cloudfoundry>>
 * <<{beatname_lc}-input-cometd>>
@@ -104,6 +105,8 @@ include::../../x-pack/filebeat/docs/inputs/input-azure-eventhub.asciidoc[]
 
 include::../../x-pack/filebeat/docs/inputs/input-azure-blob-storage.asciidoc[]
 
+include::../../x-pack/filebeat/docs/inputs/input-benchmark.asciidoc[]
+
 include::../../x-pack/filebeat/docs/inputs/input-cel.asciidoc[]
 
 include::../../x-pack/filebeat/docs/inputs/input-cloudfoundry.asciidoc[]

libbeat/docs/outputs-list.asciidoc

@@ -24,6 +24,9 @@ endif::[]
 ifndef::no_console_output[]
 * <<console-output>>
 endif::[]
+ifndef::no_discard_output[]
+* <<discard-output>>
+endif::[]
 
 //# end::outputs-list[]
 
@@ -77,6 +80,13 @@ endif::[]
 include::{libbeat-outputs-dir}/console/docs/console.asciidoc[]
 endif::[]
 
+ifndef::no_discard_output[]
+ifdef::requires_xpack[]
+[role="xpack"]
+endif::[]
+include::{libbeat-outputs-dir}/discard/docs/discard.asciidoc[]
+endif::[]
+
 ifndef::no_codec[]
 ifdef::requires_xpack[]
 [role="xpack"]

libbeat/outputs/discard/config.go

@@ -0,0 +1,30 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package discard
+
+import (
+	"github.com/elastic/elastic-agent-libs/config"
+)
+
+type discardOutConfig struct {
+	Queue config.Namespace `config:"queue"`
+}
+
+func defaultConfig() discardOutConfig {
+	return discardOutConfig{}
+}

libbeat/outputs/discard/discard.go

@@ -0,0 +1,79 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package discard
+
+import (
+	"context"
+
+	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/outputs"
+	"github.com/elastic/beats/v7/libbeat/publisher"
+	"github.com/elastic/elastic-agent-libs/config"
+	"github.com/elastic/elastic-agent-libs/logp"
+)
+
+func init() {
+	outputs.RegisterType("discard", makeDiscard)
+}
+
+type discardOutput struct {
+	log      *logp.Logger
+	beat     beat.Info
+	observer outputs.Observer
+}
+
+func makeDiscard(
+	_ outputs.IndexManager,
+	beat beat.Info,
+	observer outputs.Observer,
+	cfg *config.C,
+) (outputs.Group, error) {
+	out := &discardOutput{
+		log:      logp.NewLogger("discard"),
+		beat:     beat,
+		observer: observer,
+	}
+	doConfig := defaultConfig()
+	if err := cfg.Unpack(&doConfig); err != nil {
+		return outputs.Fail(err)
+	}
+
+	// disable bulk support in publisher pipeline
+	_ = cfg.SetInt("bulk_max_size", -1, -1)
+	out.log.Infof("Initialized discard output")
+	return outputs.Success(doConfig.Queue, -1, 0, nil, out)
+}
+
+// Implement Outputer
+func (out *discardOutput) Close() error {
+	return nil
+}
+
+func (out *discardOutput) Publish(_ context.Context, batch publisher.Batch) error {
+	defer batch.ACK()
+
+	st := out.observer
+	events := batch.Events()
+	st.NewBatch(len(events))
+	st.Acked(len(events))
+	return nil
+}
+
+func (out *discardOutput) String() string {
+	return "discard"
+}

libbeat/outputs/discard/docs/discard.asciidoc

@@ -0,0 +1,34 @@
+[[discard-output]]
+=== Configure the Discard output
+
+++++
+<titleabbrev>Discard</titleabbrev>
+++++
+
+The Discard output throws away data.
+
+WARNING: The Discard output should be used only for development or
+debugging issues. Data is lost.
+
+This can be useful if you want to work on your input configuration
+without needing to configure an output. It can also be useful to test
+how changes in input and processor configuration affect performance.
+
+Example configuration:
+
+["source","yaml",subs="attributes"]
+------------------------------------------------------------------------------
+output.discard:
+  enabled: true
+------------------------------------------------------------------------------
+
+==== Configuration options
+
+You can specify the following `output.discard` options in the +{beatname_lc}.yml+ config file:
+
+===== `enabled`
+
+The enabled config is a boolean setting to enable or disable the output. If set
+to false, the output is disabled.
+
+The default value is `true`.

libbeat/publisher/includes/includes.go

@@ -22,6 +22,7 @@ import (
 	_ "github.com/elastic/beats/v7/libbeat/outputs/codec/format"
 	_ "github.com/elastic/beats/v7/libbeat/outputs/codec/json"
 	_ "github.com/elastic/beats/v7/libbeat/outputs/console"
+	_ "github.com/elastic/beats/v7/libbeat/outputs/discard"
 	_ "github.com/elastic/beats/v7/libbeat/outputs/elasticsearch"
 	_ "github.com/elastic/beats/v7/libbeat/outputs/fileout"
 	_ "github.com/elastic/beats/v7/libbeat/outputs/kafka"

x-pack/filebeat/docs/inputs/input-benchmark.asciidoc

@@ -0,0 +1,93 @@
+[role="xpack"]
+
+:type: benchmark
+
+[id="{beatname_lc}-input-{type}"]
+=== Benchmark input
+
+++++
+<titleabbrev>Benchmark</titleabbrev>
+++++
+
+beta[]
+
+The Benchmark input generates generic events and sends them to the output. This can be useful when you want to benchmark the difference between outputs or output settings.
+
+Example configurations:
+
+Basic example, infinite events as quickly as possible:
+["source","yaml",subs="attributes"]
+----
+{beatname_lc}.inputs:
+- type: benchmark
+  enabled: true
+  message: "test message"
+  threads: 1
+----
+
+Send 1024 events and stop example:
+["source","yaml",subs="attributes"]
+----
+{beatname_lc}.inputs:
+- type: benchmark
+  enabled: true
+  message: "test message"
+  threads: 1
+  count: 1024
+----
+
+Send 5 events per second example:
+["source","yaml",subs="attributes"]
+----
+{beatname_lc}.inputs:
+- type: benchmark
+  enabled: true
+  message: "test message"
+  threads: 1
+  eps: 5
+----
+
+==== Configuration options
+
+The Benchmark input supports the following configuration options plus the
+<<{beatname_lc}-input-{type}-common-options>> described later.
+
+[float]
+==== `message`
+
+This is the value that will be in the `message` field of the json document.
+
+[float]
+==== `threads`
+
+This is the number of goroutines that will be started generating messages. Normally 1 thread can saturate an output but if necessary this can be increased.
+
+[float]
+==== `count`
+
+This is the number of messages to send. 0 represents sending infinite messages. This is mutually exclusive with the `eps` option.
+
+[float]
+==== `eps`
+
+This is the number of events per second to send. 0 represents sending as quickly as possible. This is mutually exclusive with the `count` option.
+
+
+[float]
+=== Metrics
+
+This input exposes metrics under the <<http-endpoint, HTTP monitoring endpoint>>.
+These metrics are exposed under the `/inputs` path. They can be used to
+observe the activity of the input.
+
+[options="header"]
+|=======
+| Metric                    | Description
+| `events_published_total`  | Number of events published.
+| `publishing_time`         | Histogram of the elapsed in nanoseconds (time of publisher.Publish).
+|=======
+
+[id="{beatname_lc}-input-{type}-common-options"]
+include::../../../../filebeat/docs/inputs/input-common-options.asciidoc[]
+
+:type!:

x-pack/filebeat/input/benchmark/config.go

@@ -0,0 +1,31 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package benchmark
+
+import "fmt"
+
+type benchmarkConfig struct {
+	Message string `config:"message"`
+	Count   uint64 `config:"count"`
+	Threads uint8  `config:"threads"`
+	Eps     uint64 `config:"eps"`
+}
+
+var (
+	defaultConfig = benchmarkConfig{
+		Message: "generic benchmark message",
+		Threads: 1,
+	}
+)
+
+func (c *benchmarkConfig) Validate() error {
+	if c.Count > 0 && c.Eps > 0 {
+		return fmt.Errorf("only one of count or eps may be specified, not both")
+	}
+	if c.Message == "" {
+		return fmt.Errorf("message must be specified")
+	}
+	return nil
+}

x-pack/filebeat/input/benchmark/config_test.go

@@ -0,0 +1,37 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package benchmark
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestValidate(t *testing.T) {
+	tests := map[string]struct {
+		cfg         benchmarkConfig
+		expectError bool
+		errorString string
+	}{
+		"default":     {cfg: defaultConfig},
+		"countAndEps": {cfg: benchmarkConfig{Message: "a", Count: 1, Eps: 1}, expectError: true, errorString: "only one of count or eps may be specified"},
+		"empty":       {cfg: benchmarkConfig{}, expectError: true, errorString: "message must be specified"},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			err := tc.cfg.Validate()
+			if err == nil && tc.expectError == true {
+				t.Fatalf("expected validation error, didn't get it")
+			}
+			if err != nil && tc.expectError == false {
+				t.Fatalf("unexpected validation error: %s", err)
+			}
+			if err != nil && !strings.Contains(err.Error(), tc.errorString) {
+				t.Fatalf("error: '%s' didn't contain expected string: '%s'", err, tc.errorString)
+			}
+		})
+	}
+}