[filebeat][filestream] Enable status reporter for filestream input

[VihasMakwana]

Proposed commit message

Enable StatusReporter for filestream input which was introduced in #39209

Checklist

• My code follows the style guidelines of this project
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation
• I have made corresponding change to the default configuration files
• I have added tests that prove my fix is effective or that my feature works
• I have added an entry in CHANGELOG.next.asciidoc or CHANGELOG-developer.next.asciidoc.

Related issues

• Closes 39734

Use cases

This PR allows the filestream to report its status to elastic-agent. This would keep the status of the unit up-to-date and would help with diagnostics.

Screenshots

Logs

[mergify]

This pull request does not have a backport label.
If this is a bug or security fix, could you label this PR @VihasMakwana? 🙏.
For such, you'll need to label your PR with:

• The upcoming major version of the Elastic Stack
• The upcoming minor version of the Elastic Stack (if you're not pushing a breaking change)

To fixup this pull request, you need to add the backport labels for the needed
branches, such as:

• backport-v8./d.0 is the label to automatically backport to the 8./d branch. /d is the digit

[elasticmachine]

Pinging @elastic/elastic-agent-data-plane (Team:Elastic-Agent-Data-Plane)

[cmacknz]

Sorry for the late comment, if this exits with a transient error, what is setting the input state back to healthy?

In general the thing I am watching out for in the status reporting PRs situations where we permanently mark inputs as degraded or failed after the error condition has cleared.

[VihasMakwana]

@cmacknz I understand your concern.
I did some digging and turns out that readFromSource only throws an error when it calls Publish()

beats/filebeat/input/filestream/input.go

Lines 394 to 397 in d87d15b

	if err := p.Publish(message.ToEvent(), s); err != nil {
	metrics.ProcessingErrors.Inc()
	return err
	}

The above error is returned when the context is cancelled.

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

Lines 91 to 97 in d87d15b

	func (c *cursorPublisher) forward(event beat.Event) error {
	c.client.Publish(event)
	if c.canceler == nil {
	return nil
	}
	return c.canceler.Err()
	}

For other cases (EOF, reader was closed etc.), we just log the error and return nil. So, we will not mark the input as degraded, as they are transient in nature.

[VihasMakwana]

@cmacknz maybe we don't need this particular ctx.UpdateStatus? I don't think we need to mark input as degraded when the context is cancelled (which should ideally happen during beat shutdown).

I overlooked this particular ctx.UpdateStatus.