Merge branch 'main' into fix-clean-inactive-v2

.buildkite/metricbeat/pipeline.yml

@@ -8,6 +8,7 @@ env:
   GCP_HI_PERF_MACHINE_TYPE: "c2d-highcpu-16"
   GCP_WIN_MACHINE_TYPE: "n2-standard-8"
 
+  IMAGE_MACOS_ARM: "generic-13-ventura-arm"
   IMAGE_MACOS_X86_64: "generic-13-ventura-x64"
   IMAGE_UBUNTU_ARM_64: "platform-ingest-beats-ubuntu-2204-aarch64"
   IMAGE_UBUNTU_X86_64: "family/platform-ingest-beats-ubuntu-2204"
@@ -214,8 +215,8 @@ steps:
     key: "metricbeat-extended-macos-tests"
     if: build.env("BUILDKITE_PULL_REQUEST") == "false" || build.env("GITHUB_PR_LABELS") =~ /.*macOS.*/
     steps:
-      - label: ":mac: MacOS Unit Tests"
-        key: "extended-macos-unit-tests"
+      - label: ":mac: MacOS x64_64 Unit Tests"
+        key: "extended-macos-x64-64-unit-tests"
         command: |
           set -euo pipefail
           source .buildkite/scripts/install_macos_tools.sh
@@ -228,7 +229,25 @@ steps:
           - "metricbeat/build/*.json"
         notify:
           - github_commit_status:
-              context: "metricbeat: Extended MacOS Unit Tests"
+              context: "metricbeat: Extended MacOS x86_64 Unit Tests"
+
+      - label: ":mac: MacOS arm64 Unit Tests"
+        key: "extended-macos-arm64-unit-tests"
+        skip: "due to https://github.com/elastic/beats/issues/33035"
+        command: |
+          set -euo pipefail
+          source .buildkite/scripts/install_macos_tools.sh
+          cd metricbeat && mage build unitTest
+        agents:
+          provider: "orka"
+          imagePrefix: "${IMAGE_MACOS_ARM}"
+        artifact_paths:
+          - "metricbeat/build/*.xml"
+          - "metricbeat/build/*.json"
+        notify:
+          - github_commit_status:
+              context: "metricbeat: Extended MacOS arm64 Unit Tests"
+
 
   - wait: ~
     # with PRs, we want to run packaging only if mandatory tests succeed

.buildkite/pipeline.yml

@@ -9,6 +9,7 @@ steps:
             - path:
                 - auditbeat/
                 - .buildkite/auditbeat/
+                - .buildkite/scripts
                 #OSS
                 - go.mod
                 - pytest.ini
@@ -33,6 +34,7 @@ steps:
             - path:
                 - heartbeat/
                 - .buildkite/heartbeat/
+                - .buildkite/scripts
                 #OSS
                 - go.mod
                 - pytest.ini
@@ -57,6 +59,7 @@ steps:
             - path:
                 - x-pack/dockerlogbeat/
                 - .buildkite/x-pack/pipeline.xpack.dockerlogbeat.yml
+                - .buildkite/scripts
                 #OSS
                 - go.mod
                 - pytest.ini
@@ -106,6 +109,7 @@ steps:
             - path:
                 - x-pack/winlogbeat/
                 - .buildkite/x-pack/pipeline.xpack.winlogbeat.yml
+                - .buildkite/scripts
                 #OSS
                 - go.mod
                 - pytest.ini
@@ -131,6 +135,7 @@ steps:
                 - .buildkite/deploy/kubernetes/**
                 - deploy/kubernetes/**
                 - libbeat/docs/version.asciidoc
+                - .buildkite/scripts
               config:
                 trigger: "deploy-k8s"
                 build:

.buildkite/pull-requests.json

@@ -5,7 +5,7 @@
             "pipelineSlug": "beats",
             "allow_org_users": true,
             "allowed_repo_permissions": ["admin", "write"],
-            "allowed_list": [ ],
+            "allowed_list": [ "mergify[bot]" ],
             "set_commit_status": true,
             "build_on_commit": true,
             "build_on_comment": true,

.github/CODEOWNERS

@@ -229,6 +229,6 @@ CHANGELOG*
 
 # Ownership of CI or related files by the Ingest Eng Prod team
 /.buildkite @elastic/ingest-eng-prod
-/catalog-info.yml @elastic/ingest-eng-prod
+/catalog-info.yaml @elastic/ingest-eng-prod
 /libbeat/scripts @elastic/ingest-eng-prod
 /metricbeat/tests @elastic/ingest-eng-prod

CHANGELOG-developer.next.asciidoc

@@ -66,6 +66,7 @@ The list below covers the major changes between 7.0.0-rc2 and main only.
 - Rename `queue.Batch.ACK()` to `queue.Batch.Done()`. {pull}31903[31903]
 - `queue.ACKListener` has been removed. Queue configurations now accept an explicit callback function for ACK handling. {pull}35078[35078]
 - Split split httpmon out of x-pack/filebeat/input/internal/httplog. {pull}36385[36385]
+- Beats publishing pipeline does not propagate the close signal to its clients any more. It's responsibility of the user to close the pipeline client. {issue}38197[38197] {pull}38556[38556]
 
 ==== Bugfixes
 
@@ -93,6 +94,7 @@ The list below covers the major changes between 7.0.0-rc2 and main only.
 - Make winlogbeat/sys/wineventlog follow the unsafe.Pointer rules. {pull}36650[36650]
 - Cleaned up documentation errors & fixed a minor bug in Filebeat Azure blob storage input. {pull}36714[36714]
 - Fix copy arguments for strict aligned architectures. {pull}36976[36976]
+- Fix panic when more than 32767 pipeline clients are active. {issue}38197[38197] {pull}38556[38556]
 
 ==== Added
 

CHANGELOG.next.asciidoc

@@ -101,9 +101,9 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Fix filestream's registry GC: registry entries are now removed from the in-memory and disk store when they're older than the set TTL {issue}36761[36761] {pull}38488[38488]
 - Fix indexing failures by re-enabling event normalisation in netflow input. {issue}38703[38703] {pull}38780[38780]
 - Fix handling of truncated files in Filestream {issue}38070[38070] {pull}38416[38416]
-- Fix `clean_removed` on Filestream input. {issue}36761[36761] {pull}38488[38488]
+- Fix panic when more than 32767 pipeline clients are active. {issue}38197[38197] {pull}38556[38556]
 - Fix filestream's registry GC: registry entries are now removed from the in-memory and disk store when they're older than the set TTL {issue}36761[36761] {pull}38488[38488]
-- Fix `clean_inactive` on Filestream input by correctly parsing the value from the configuration yaml {issue}36629[36629] {pull}38632[38632]
+
 
 *Heartbeat*
 
@@ -207,6 +207,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*
 

NOTICE.txt

@@ -24111,11 +24111,11 @@ Contents of probable licence file $GOMODCACHE/go.elastic.co/apm/module/apmhttp/v
 
 --------------------------------------------------------------------------------
 Dependency : go.elastic.co/apm/v2
-Version: v2.5.0
+Version: v2.6.0
 Licence type (autodetected): Apache-2.0
 --------------------------------------------------------------------------------
 
-Contents of probable licence file $GOMODCACHE/go.elastic.co/apm/v2@v2.5.0/LICENSE:
+Contents of probable licence file $GOMODCACHE/go.elastic.co/apm/v2@v2.6.0/LICENSE:
 
                                  Apache License
                            Version 2.0, January 2004

catalog-info.yaml

@@ -39,10 +39,11 @@ spec:
       name: beats
       description: "Beats Main pipeline"
     spec:
-      branch_configuration: "main 7.17 8.* mergify*"
+      branch_configuration: "main 7.17 8.*"
       pipeline_file: ".buildkite/pipeline.yml"
       provider_settings:
         build_pull_request_forks: false
+        build_pull_request_labels_changed: true # automatically re trigger build if GH labels change
         build_pull_requests: true # requires filter_enabled and filter_condition settings as below when used with buildkite-pr-bot
         build_tags: true
         filter_enabled: true

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[]

filebeat/input/filestream/internal/input-logfile/harvester.go

@@ -219,7 +219,6 @@ func startHarvester(
 		defer releaseResource(resource)
 
 		client, err := hg.pipeline.ConnectWith(beat.ClientConfig{
-			CloseRef:      ctx.Cancelation,
 			EventListener: newInputACKHandler(hg.ackCH),
 		})
 		if err != nil {

filebeat/input/kafka/input.go

@@ -120,12 +120,12 @@ func (input *kafkaInput) Run(ctx input.Context, pipeline beat.Pipeline) error {
 				}
 			}),
 		),
-		CloseRef:  ctx.Cancelation,
 		WaitClose: input.config.WaitClose,
 	})
 	if err != nil {
 		return err
 	}
+	defer client.Close()
 
 	log.Info("Starting Kafka input")
 	defer log.Info("Kafka input stopped")

filebeat/input/v2/input-cursor/input.go

@@ -146,7 +146,6 @@ func (inp *managedInput) runSource(
 	}()
 
 	client, err := pipeline.ConnectWith(beat.ClientConfig{
-		CloseRef:      ctx.Cancelation,
 		EventListener: newInputACKHandler(ctx.Logger),
 	})
 	if err != nil {

filebeat/input/v2/input-stateless/stateless.go

@@ -85,9 +85,6 @@ func (si configuredInput) Run(ctx v2.Context, pipeline beat.PipelineConnector) (
 
 	client, err := pipeline.ConnectWith(beat.ClientConfig{
 		PublishMode: beat.DefaultGuarantees,
-
-		// configure pipeline to disconnect input on stop signal.
-		CloseRef: ctx.Cancelation,
 	})
 	if err != nil {
 		return err

filebeat/input/v2/input-stateless/stateless_test.go

@@ -107,22 +107,23 @@ func TestStateless_Run(t *testing.T) {
 			},
 		}), nil)
 
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
 		// connector creates a client the blocks forever until the shutdown signal is received
 		var publishCalls atomic.Int
 		connector := pubtest.FakeConnector{
 			ConnectFunc: func(config beat.ClientConfig) (beat.Client, error) {
 				return &pubtest.FakeClient{
 					PublishFunc: func(event beat.Event) {
 						publishCalls.Inc()
-						<-config.CloseRef.Done()
+						// Unlock Publish once the input has been cancelled
+						<-ctx.Done()
 					},
 				}, nil
 			},
 		}
 
-		ctx, cancel := context.WithCancel(context.Background())
-		defer cancel()
-
 		var wg sync.WaitGroup
 		var err error
 		wg.Add(1)

go.mod

@@ -229,7 +229,7 @@ require (
 	github.com/tklauser/go-sysconf v0.3.10
 	go.elastic.co/apm/module/apmelasticsearch/v2 v2.4.8
 	go.elastic.co/apm/module/apmhttp/v2 v2.5.0
-	go.elastic.co/apm/v2 v2.5.0
+	go.elastic.co/apm/v2 v2.6.0
 	go.mongodb.org/mongo-driver v1.5.1
 	golang.org/x/exp v0.0.0-20231127185646-65229373498e
 	golang.org/x/tools/go/vcs v0.1.0-deprecated

go.sum

@@ -1954,8 +1954,8 @@ go.elastic.co/apm/module/apmelasticsearch/v2 v2.4.8 h1:4j3wI1e+WV6u+9ZR7lorkJI2r
 go.elastic.co/apm/module/apmelasticsearch/v2 v2.4.8/go.mod h1:C9ajbSjZ3akTrFOjBr+pMq8bPVOH9vhIG+knZAuPW3s=
 go.elastic.co/apm/module/apmhttp/v2 v2.5.0 h1:4AWlw8giL7hRYBQiwF1/Thm0GDsbQH/Ofe4eySAnURo=
 go.elastic.co/apm/module/apmhttp/v2 v2.5.0/go.mod h1:ZP7gLEzY/OAPTqNZjp8AzA06HF82zfwXEpKI2sSVTgk=
-go.elastic.co/apm/v2 v2.5.0 h1:UYqdu/bjcubcP9BIy5+os2ExRzw03yOQFG+sRGGhVlQ=
-go.elastic.co/apm/v2 v2.5.0/go.mod h1:+CiBUdrrAGnGCL9TNx7tQz3BrfYV23L8Ljvotoc87so=
+go.elastic.co/apm/v2 v2.6.0 h1:VieBMLQFtXua2YxpYxaSdYGnmmxhLT46gosI5yErJgY=
+go.elastic.co/apm/v2 v2.6.0/go.mod h1:33rOXgtHwbgZcDgi6I/GtCSMZQqgxkHC0IQT3gudKvo=
 go.elastic.co/ecszap v1.0.2 h1:iW5OGx8IiokiUzx/shD4AJCPFMC9uUtr7ycaiEIU++I=
 go.elastic.co/ecszap v1.0.2/go.mod h1:dJkSlK3BTiwG/qXhCwe50Mz/jwu854vSip8sIeQhNZg=
 go.elastic.co/fastjson v1.1.0 h1:3MrGBWWVIxe/xvsbpghtkFoPciPhOCmjsR/HfwEeQR4=

libbeat/beat/pipeline.go

@@ -49,8 +49,6 @@ type ClientConfig struct {
 
 	Processing ProcessingConfig
 
-	CloseRef CloseRef
-
 	// WaitClose sets the maximum duration to wait on ACK, if client still has events
 	// active non-acknowledged events in the publisher pipeline.
 	// WaitClose is only effective if one of ACKCount, ACKEvents and ACKLastEvents
@@ -91,13 +89,6 @@ type EventListener interface {
 	ClientClosed()
 }
 
-// CloseRef allows users to close the client asynchronously.
-// A CloseRef implements a subset of function required for context.Context.
-type CloseRef interface {
-	Done() <-chan struct{}
-	Err() error
-}
-
 // ProcessingConfig provides additional event processing settings a client can
 // pass to the publisher pipeline on Connect.
 type ProcessingConfig struct {

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{}
+}