x-pack/filebeat/input/cel: log unpublished event count and exit publish loop on context cancellation

x-pack/filebeat/input/cel/input.go

@@ -100,7 +100,7 @@
 func (input) Name() string { return inputName }
 
 func (input) Test(src inputcursor.Source, _ v2.TestContext) error {
 	cfg := src.(*source).cfg
 	if !wantClient(cfg) {
 		return nil
 	}
@@ -110,7 +110,7 @@
 // Run starts the input and blocks until it ends completes. It will return on
 // context cancellation or type invalidity errors, any other error will be retried.
 func (input) Run(env v2.Context, src inputcursor.Source, crsr inputcursor.Cursor, pub inputcursor.Publisher) error {
 	dataStreamName := src.(*source).cfg.DataStream // May be empty.
 
 	var cursor map[string]interface{}
 	env.UpdateStatus(status.Starting, dataStreamName)
@@ -128,7 +128,7 @@
 			parent: &env,
 		}
 	}
 	err := input{}.run(env, src.(*source), cursor, pub, health)
 	if err != nil {
 		msg := "failed to run: " + err.Error()
 		if dataStreamName != "" {
@@ -541,6 +541,25 @@
 						pubCursor = cursor
 					}
 				}
+				// This is checked prior to the publish attempt since the
+				// cursor.Publisher interface does not document the behaviour
+				// related to context cancellation and the context is not
+				// explicitly passed in, so favour this explicit clarity.
+				switch err := ctx.Err(); {
+				case err == nil:
+				case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
+					log.Infow("context cancelled with unpublished events", "unpublished", len(events)-i)
+					// Don't update status, since we are about to pass
+					// through the Running state and then fall through
+					// to the input exit with a change to Stopped.
+					break
+				default:
+					// This should never happen.
+					log.Warnw("failed with unpublished events", "error", err, "unpublished", len(events)-i)
+					health.UpdateStatus(status.Degraded, "error publishing events: "+err.Error())
+					isDegraded = true
+					break
+				}
 				err = pub.Publish(beat.Event{
 					Timestamp: time.Now(),
 					Fields:    event,