Fix panic when more than 32767 pipeline clients are active

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,6 +101,7 @@ 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 panic when more than 32767 pipeline clients are active. {issue}38197[38197] {pull}38556[38556]
 
 *Heartbeat*
 

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)

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/publisher/pipeline/client.go

@@ -42,10 +42,8 @@ type client struct {
 	eventWaitGroup *sync.WaitGroup
 
 	// Open state, signaling, and sync primitives for coordinating client Close.
-	isOpen    atomic.Bool   // set to false during shutdown, such that no new events will be accepted anymore.
-	closeOnce sync.Once     // closeOnce ensure that the client shutdown sequence is only executed once
-	closeRef  beat.CloseRef // extern closeRef for sending a signal that the client should be closed.
-	done      chan struct{} // the done channel will be closed if the closeReg gets closed, or Close is run.
+	isOpen    atomic.Bool // set to false during shutdown, such that no new events will be accepted anymore.
+	closeOnce sync.Once   // closeOnce ensure that the client shutdown sequence is only executed once
 
 	observer       observer
 	eventListener  beat.EventListener
@@ -137,8 +135,6 @@ func (c *client) Close() error {
 	// first stop ack handling. ACK handler might block on wait (with timeout), waiting
 	// for pending events to be ACKed.
 	c.closeOnce.Do(func() {
-		close(c.done)
-
 		c.isOpen.Store(false)
 		c.onClosing()
 

libbeat/publisher/pipeline/client_test.go

@@ -18,7 +18,6 @@
 package pipeline
 
 import (
-	"context"
 	"errors"
 	"io"
 	"sync"
@@ -95,15 +94,7 @@ func TestClient(t *testing.T) {
 				pipeline := makePipeline(t, Settings{}, makeTestQueue())
 				defer pipeline.Close()
 
-				var ctx context.Context
-				var cancel func()
-				if test.context {
-					ctx, cancel = context.WithCancel(context.Background())
-				}
-
-				client, err := pipeline.ConnectWith(beat.ClientConfig{
-					CloseRef: ctx,
-				})
+				client, err := pipeline.ConnectWith(beat.ClientConfig{})
 				if err != nil {
 					t.Fatal(err)
 				}
@@ -116,7 +107,9 @@ func TestClient(t *testing.T) {
 					client.Publish(beat.Event{})
 				}()
 
-				test.close(client, cancel)
+				test.close(client, func() {
+					client.Close()
+				})
 				wg.Wait()
 			})
 		}

libbeat/publisher/pipeline/pipeline.go

@@ -22,7 +22,6 @@
 
 import (
 	"fmt"
-	"reflect"
 	"sync"
 	"time"
 
@@ -73,8 +72,8 @@
 	eventWaitGroup   *sync.WaitGroup
 
 	// closeRef signal propagation support
 	guardStartSigPropagation sync.Once
 	sigNewClient             chan *client
 
 	processors processing.Supporter
 }
@@ -197,9 +196,6 @@
 	p.outputController.Close()
 
 	p.observer.cleanup()
-	if p.sigNewClient != nil {
-		close(p.sigNewClient)
-	}
 	return nil
 }
 
@@ -212,6 +208,8 @@
 // The client behavior on close and ACK handling can be configured by setting
 // the appropriate fields in the passed ClientConfig.
 // If not set otherwise the defaut publish mode is OutputChooses.
+//
+// It is responsibility of the caller to close the client.
 func (p *Pipeline) ConnectWith(cfg beat.ClientConfig) (beat.Client, error) {
 	var (
 		canDrop    bool
@@ -239,8 +237,6 @@
 
 	client := &client{
 		logger:         p.monitors.Logger,
-		closeRef:       cfg.CloseRef,
-		done:           make(chan struct{}),
 		isOpen:         atomic.MakeBool(true),
 		clientListener: cfg.ClientListener,
 		processors:     processors,
@@ -295,93 +291,9 @@
 	}
 
 	p.observer.clientConnected()
-
-	if client.closeRef != nil {
-		p.registerSignalPropagation(client)
-	}
-
 	return client, nil
 }
 
-func (p *Pipeline) registerSignalPropagation(c *client) {
-	p.guardStartSigPropagation.Do(func() {
-		p.sigNewClient = make(chan *client, 1)
-		go p.runSignalPropagation()
-	})
-	p.sigNewClient <- c
-}
-
-func (p *Pipeline) runSignalPropagation() {
-	var channels []reflect.SelectCase
-	var clients []*client
-
-	channels = append(channels, reflect.SelectCase{
-		Dir:  reflect.SelectRecv,
-		Chan: reflect.ValueOf(p.sigNewClient),
-	})
-
-	for {
-		chosen, recv, recvOK := reflect.Select(channels)
-		if chosen == 0 {
-			if !recvOK {
-				// sigNewClient was closed
-				return
-			}
-
-			// new client -> register client for signal propagation.
-			if client := recv.Interface().(*client); client != nil {
-				channels = append(channels,
-					reflect.SelectCase{
-						Dir:  reflect.SelectRecv,
-						Chan: reflect.ValueOf(client.closeRef.Done()),
-					},
-					reflect.SelectCase{
-						Dir:  reflect.SelectRecv,
-						Chan: reflect.ValueOf(client.done),
-					},
-				)
-				clients = append(clients, client)
-			}
-			continue
-		}
-
-		// find client we received a signal for. If client.done was closed, then
-		// we have to remove the client only. But if closeRef did trigger the signal, then
-		// we have to propagate the async close to the client.
-		// In either case, the client will be removed
-
-		i := (chosen - 1) / 2
-		isSig := (chosen & 1) == 1
-		if isSig {
-			client := clients[i]
-			client.Close()
-		}
-
-		// remove:
-		last := len(clients) - 1
-		ch1 := i*2 + 1
-		ch2 := ch1 + 1
-		lastCh1 := last*2 + 1
-		lastCh2 := lastCh1 + 1
-
-		clients[i], clients[last] = clients[last], nil
-		channels[ch1], channels[lastCh1] = channels[lastCh1], reflect.SelectCase{}
-		channels[ch2], channels[lastCh2] = channels[lastCh2], reflect.SelectCase{}
-
-		clients = clients[:last]
-		channels = channels[:lastCh1]
-		if cap(clients) > 10 && len(clients) <= cap(clients)/2 {
-			clientsTmp := make([]*client, len(clients))
-			copy(clientsTmp, clients)
-			clients = clientsTmp
-
-			channelsTmp := make([]reflect.SelectCase, len(channels))
-			copy(channelsTmp, channels)
-			channels = channelsTmp
-		}
-	}
-}
-
 func (p *Pipeline) createEventProcessing(cfg beat.ProcessingConfig, noPublish bool) (beat.Processor, error) {
 	if p.processors == nil {
 		return nil, nil