Merge branch 'main' of github.com:elastic/beats into 47164-add-failure-store-count

Author: belimawr

changelog/fragments/1765204121-Correctly-report-V2-inputs-as-failed-to-the-Manager.yaml

@@ -0,0 +1,32 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: feature
+
+# Change summary; a 80ish characters long description of the change.
+summary: Improving input error reporting to Elastic Agent, specially when pipeline configurations are incorrect.
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+#description:
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: filebeat
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+pr: https://github.com/elastic/beats/pull/47905
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+issue: https://github.com/elastic/beats/issues/45649

changelog/fragments/1765389915-beat-receiver-subcomponent-status.yaml

@@ -0,0 +1,32 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: feature
+
+# Change summary; a 80ish characters long description of the change.
+summary: Make beats receivers emit status for their subcomponents
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+#description:
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: all
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+#pr: https://github.com/owner/repo/1234
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+#issue: https://github.com/owner/repo/1234

changelog/fragments/1765460540-upgrade-osquery-to-5.19.0.yaml

@@ -0,0 +1,45 @@
+# REQUIRED
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: feature
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: Upgrade osquery to 5.19.0
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# description:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: osquerybeat
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+# pr: https://github.com/owner/repo/1234
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+# issue: https://github.com/owner/repo/1234

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

@@ -28,6 +28,7 @@ import (
 	"github.com/elastic/beats/v7/filebeat/input/filestream/internal/task"
 	inputv2 "github.com/elastic/beats/v7/filebeat/input/v2"
 	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/management/status"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/go-concert/ctxtool"
 )
@@ -129,6 +130,7 @@ type defaultHarvesterGroup struct {
 	tg           *task.Group
 	metrics      *Metrics
 	notifyChan   chan HarvesterStatus
+	inputID      string
 }
 
 // HarvesterStatus is used to notify an observer that the harvester for the ID
@@ -160,7 +162,7 @@ func (hg *defaultHarvesterGroup) Start(ctx inputv2.Context, src Source) {
 	sourceName := hg.identifier.ID(src)
 	ctx.Logger = ctx.Logger.With("source_file", sourceName)
 
-	if err := hg.tg.Go(startHarvester(ctx, hg, src, false, hg.metrics)); err != nil {
+	if err := hg.tg.Go(startHarvester(ctx, hg, src, false, hg.metrics, hg.inputID)); err != nil {
 		ctx.Logger.Warnf(
 			"tried to start harvester for %s with task group already closed",
 			ctx.ID)
@@ -177,7 +179,7 @@ func (hg *defaultHarvesterGroup) Restart(ctx inputv2.Context, src Source) {
 	ctx.Logger = ctx.Logger.With("source_file", sourceName)
 	ctx.Logger.Debug("Restarting harvester for file")
 
-	if err := hg.tg.Go(startHarvester(ctx, hg, src, true, hg.metrics)); err != nil {
+	if err := hg.tg.Go(startHarvester(ctx, hg, src, true, hg.metrics, hg.inputID)); err != nil {
 		ctx.Logger.Warnf(
 			"input %s tried to restart harvester with task group already closed",
 			ctx.ID)
@@ -194,6 +196,7 @@ func startHarvester(
 	src Source,
 	restart bool,
 	metrics *Metrics,
+	inputID string,
 ) func(context.Context) error {
 	srcID := hg.identifier.ID(src)
 
@@ -204,6 +207,14 @@ func startHarvester(
 				ctx.Logger.Errorf("Harvester crashed with: %+v", err)
 				hg.readers.remove(srcID)
 			}
+
+			// Report any harvester error as a degraded state for the input
+			if err != nil {
+				ctx.StatusReporter.UpdateStatus(
+					status.Degraded,
+					fmt.Sprintf("Harvester for Filestream input %q failed: %s", inputID, err),
+				)
+			}
 		}()
 
 		if restart {

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

@@ -34,6 +34,7 @@ import (
 	"github.com/elastic/beats/v7/filebeat/input/filestream/internal/task"
 	input "github.com/elastic/beats/v7/filebeat/input/v2"
 	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/management/status"
 	"github.com/elastic/beats/v7/libbeat/tests/resources"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/logp/logptest"
@@ -115,7 +116,7 @@ func TestDefaultHarvesterGroup(t *testing.T) {
 		defer goroutinesChecker.WaitUntilOriginalCount()
 
 		wg.Add(1)
-		hg.Start(input.Context{Logger: logp.L(), Cancelation: context.Background()}, source)
+		hg.Start(input.Context{Logger: logp.L(), Cancelation: t.Context(), StatusReporter: mockStatusReporter{}}, source)
 
 		// wait until harvester.Run is done
 		wg.Wait()
@@ -167,10 +168,10 @@ func TestDefaultHarvesterGroup(t *testing.T) {
 		source2 := &testSource{name: "/path/to/test/2"}
 		wg.Add(2)
 		hg.Start(
-			input.Context{Logger: logp.L(), Cancelation: context.Background()},
+			input.Context{Logger: logp.L(), Cancelation: t.Context(), StatusReporter: mockStatusReporter{}},
 			source1)
 		hg.Start(
-			input.Context{Logger: logp.L(), Cancelation: context.Background()},
+			input.Context{Logger: logp.L(), Cancelation: t.Context(), StatusReporter: mockStatusReporter{}},
 			source2)
 
 		assert.Eventually(t,
@@ -221,7 +222,7 @@ func TestDefaultHarvesterGroup(t *testing.T) {
 
 		goroutinesChecker := resources.NewGoroutinesChecker()
 
-		hg.Start(input.Context{Logger: logp.L(), Cancelation: context.Background()}, source)
+		hg.Start(input.Context{Logger: logp.L(), Cancelation: t.Context(), StatusReporter: mockStatusReporter{}}, source)
 
 		goroutinesChecker.WaitUntilIncreased(1)
 		// wait until harvester is started
@@ -241,7 +242,7 @@ func TestDefaultHarvesterGroup(t *testing.T) {
 	t.Run("assert a harvester for same source cannot be started", func(t *testing.T) {
 		mockHarvester := &mockHarvester{onRun: blockUntilCancelOnRun}
 		hg := testDefaultHarvesterGroup(t, mockHarvester)
-		inputCtx := input.Context{Logger: logp.L(), Cancelation: context.Background()}
+		inputCtx := input.Context{Logger: logp.L(), Cancelation: t.Context(), StatusReporter: mockStatusReporter{}}
 
 		goroutinesChecker := resources.NewGoroutinesChecker()
 		defer goroutinesChecker.WaitUntilOriginalCount()
@@ -274,7 +275,7 @@ func TestDefaultHarvesterGroup(t *testing.T) {
 
 		goroutinesChecker := resources.NewGoroutinesChecker()
 
-		hg.Start(input.Context{Logger: logp.L(), Cancelation: context.Background()}, source)
+		hg.Start(input.Context{Logger: logp.L(), Cancelation: t.Context(), StatusReporter: mockStatusReporter{}}, source)
 
 		// wait until harvester is stopped
 		goroutinesChecker.WaitUntilOriginalCount()
@@ -295,7 +296,7 @@ func TestDefaultHarvesterGroup(t *testing.T) {
 		goroutinesChecker := resources.NewGoroutinesChecker()
 		defer goroutinesChecker.WaitUntilOriginalCount()
 
-		hg.Start(input.Context{Logger: logp.L(), Cancelation: context.Background()}, source)
+		hg.Start(input.Context{Logger: logp.L(), Cancelation: t.Context(), StatusReporter: mockStatusReporter{}}, source)
 
 		goroutinesChecker.WaitUntilOriginalCount()
 
@@ -311,7 +312,7 @@ func TestDefaultHarvesterGroup(t *testing.T) {
 		var wg sync.WaitGroup
 		mockHarvester := &mockHarvester{onRun: correctOnRun, wg: &wg}
 		hg := testDefaultHarvesterGroup(t, mockHarvester)
-		inputCtx := input.Context{Logger: logp.L(), Cancelation: context.Background()}
+		inputCtx := input.Context{Logger: logp.L(), Cancelation: t.Context(), StatusReporter: mockStatusReporter{}}
 
 		r, err := lock(inputCtx, hg.store, hg.identifier.ID(source))
 		if err != nil {
@@ -346,7 +347,7 @@ func TestDefaultHarvesterGroup(t *testing.T) {
 		mockHarvester := &mockHarvester{onRun: correctOnRun}
 		hg := testDefaultHarvesterGroup(t, mockHarvester)
 		hg.tg = task.NewGroup(0, 50*time.Millisecond, testLog, "")
-		inputCtx := input.Context{Logger: logp.L(), Cancelation: context.Background()}
+		inputCtx := input.Context{Logger: logp.L(), Cancelation: t.Context(), StatusReporter: mockStatusReporter{}}
 
 		goroutinesChecker := resources.NewGoroutinesChecker()
 		defer goroutinesChecker.WaitUntilOriginalCount()
@@ -369,7 +370,7 @@ func TestDefaultHarvesterGroup(t *testing.T) {
 		var wg sync.WaitGroup
 		mockHarvester := &mockHarvester{onRun: blockUntilCancelOnRun, wg: &wg}
 		hg := testDefaultHarvesterGroup(t, mockHarvester)
-		inputCtx := input.Context{Logger: logp.L(), Cancelation: context.Background()}
+		inputCtx := input.Context{Logger: logp.L(), Cancelation: t.Context(), StatusReporter: mockStatusReporter{}}
 
 		goroutinesChecker := resources.NewGoroutinesChecker()
 		defer goroutinesChecker.WaitUntilOriginalCount()
@@ -588,10 +589,10 @@ func panicOnRun(_ input.Context, _ Source, _ Cursor, _ Publisher) error {
 
 type testLogger strings.Builder
 
-func (tl *testLogger) Errorf(format string, args ...interface{}) {
+func (tl *testLogger) Errorf(format string, args ...any) {
 	sb := (*strings.Builder)(tl)
-	sb.WriteString(fmt.Sprintf(format, args...))
-	sb.WriteString("\n")
+	fmt.Fprintf(sb, format, args...)
+	fmt.Fprint(sb, "\n")
 }
 
 func (tl *testLogger) String() string {
@@ -681,3 +682,9 @@ func (mp *MockPipeline) ConnectWith(config beat.ClientConfig) (beat.Client, erro
 func (mp *MockPipeline) Connect() (beat.Client, error) {
 	return mp.ConnectWith(beat.ClientConfig{})
 }
+
+type mockStatusReporter struct{}
+
+// UpdateStatus is a no-op
+func (m mockStatusReporter) UpdateStatus(status status.Status, msg string) {
+}

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

@@ -81,6 +81,7 @@ func (inp *managedInput) Run(
 			ctx.Logger,
 			"harvester:"),
 		metrics: metrics,
+		inputID: inp.id,
 	}
 
 	prospectorStore := inp.manager.getRetainedStore()

filebeat/input/v2/compat/compat.go

@@ -161,7 +161,9 @@ func (r *runner) Start() {
 
 		err := r.input.Run(ctx, pc)
 		if err != nil && !errors.Is(err, context.Canceled) {
-			log.Errorf("Input '%s' failed with: %+v", name, err)
+			errMsg := fmt.Sprintf("Input '%s' failed with: %+v", name, err)
+			log.Error(errMsg)
+			ctx.StatusReporter.UpdateStatus(status.Failed, errMsg)
 		} else {
 			log.Infof("Input '%s' stopped (goroutine)", name)
 		}

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

@@ -224,6 +224,7 @@ func newInputACKHandler(log *logp.Logger) beat.EventListener {
 		if n == 0 {
 			return
 		}
+		//nolint:errcheck // We know it will always work
 		private[last].(*updateOp).Execute(n)
 	})
 }

libbeat/otelbeat/oteltest/oteltest.go

@@ -87,7 +87,7 @@ type CheckReceiversParams struct {
 	// The function is called periodically until the assertions are met or the timeout is reached.
 	AssertFunc func(t *assert.CollectT, logs map[string][]mapstr.M, zapLogs *observer.ObservedLogs)
 
-	Status      ExpectedStatus
+	Status      *componentstatus.Event
 	NumRestarts int
 }
 
@@ -200,12 +200,11 @@ func CheckReceivers(params CheckReceiversParams) {
 			}
 			require.NotNil(ct, host.getEvent(), "expected not nil, got nil")
 
-			if params.Status.Error == "" {
-				require.Equalf(ct, host.Evt.Status(), componentstatus.StatusOK, "expected %v, got %v", params.Status.Status, host.Evt.Status())
-				require.Nilf(ct, host.Evt.Err(), "expected nil, got %v", host.Evt.Err())
-			} else {
-				require.Equalf(ct, host.Evt.Status(), params.Status.Status, "expected %v, got %v", params.Status.Status, host.Evt.Status())
-				require.ErrorContainsf(ct, host.Evt.Err(), params.Status.Error, "expected error to contain '%v': %v", params.Status.Error, host.Evt.Err())
+			if params.Status != nil {
+				assert.Equal(t, params.Status.Status(), params.Status.Status(), host.Evt.Status(),
+					"expected status to be %v, got %v", params.Status.Status(), host.Evt.Status())
+				assert.Equal(t, params.Status.Err(), host.Evt.Err())
+				assert.Equal(t, params.Status.Attributes().AsRaw(), host.Evt.Attributes().AsRaw())
 			}
 
 			if params.AssertFunc != nil {

testing/environments/snapshot.yml

@@ -2,7 +2,7 @@
 
 services:
   elasticsearch:
-    image: docker.elastic.co/elasticsearch/elasticsearch:9.3.0-a866d8e8-SNAPSHOT
+    image: docker.elastic.co/elasticsearch/elasticsearch:9.3.0-70731819-SNAPSHOT
     # When extend is used it merges healthcheck.tests, see:
     # https://github.com/docker/compose/issues/8962
     # healthcheck:
@@ -37,7 +37,7 @@ services:
     - "./docker/elasticsearch/users_roles:/usr/share/elasticsearch/config/users_roles"
 
   logstash:
-    image: docker.elastic.co/logstash/logstash:9.3.0-a866d8e8-SNAPSHOT
+    image: docker.elastic.co/logstash/logstash:9.3.0-70731819-SNAPSHOT
     healthcheck:
       test: ["CMD", "curl", "-f", "http://localhost:9600/_node/stats"]
       retries: 600
@@ -50,7 +50,7 @@ services:
       - 5055:5055
 
   kibana:
-    image: docker.elastic.co/kibana/kibana:9.3.0-a866d8e8-SNAPSHOT
+    image: docker.elastic.co/kibana/kibana:9.3.0-70731819-SNAPSHOT
     environment:
     - "ELASTICSEARCH_USERNAME=kibana_system_user"
     - "ELASTICSEARCH_PASSWORD=testing"