chore: fix defer usage for stopped status reporting

[Kavindu-Dodan]

Proposed commit message

Improve how stopped status get reported in S3 & CloudWatch inputs.

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.

[github-actions]

🤖 GitHub comments

Expand to view the GitHub comments

Just comment with:

• run docs-build : Re-trigger the docs validation. (use unformatted text in the comment!)

[mergify]

This pull request does not have a backport label.
If this is a bug or security fix, could you label this PR @Kavindu-Dodan? 🙏.
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-8./d is the label to automatically backport to the 8./d branch. /d is the digit
• backport-active-all is the label that automatically backports to all active branches.
• backport-active-8 is the label that automatically backports to all active minor branches for the 8 major.
• backport-active-9 is the label that automatically backports to all active minor branches for the 9 major.

[elasticmachine]

Pinging @elastic/obs-ds-hosted-services (Team:obs-ds-hosted-services)

[elasticmachine]

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

[cmacknz]

Thanks, the reason the Stopped state shows as healthy is because the block below doesn't consider the Stopped state when aggregating the status of streams up into the parent input (the health protocol only understands inputs and not streams right now).

beats/x-pack/libbeat/management/unit.go

Lines 219 to 233 in eadc27e

	// inputLevelState state is marked as running, check the stream states
	reportedStatus := status.Running
	reportedMsg := "Healthy"
	for _, streamState := range u.streamStates {
	switch streamState.state {
	case status.Degraded:
	if reportedStatus != status.Degraded {
	reportedStatus = status.Degraded
	reportedMsg = streamState.msg
	}
	case status.Failed:
	// return the first failed stream
	return streamState.state, streamState.msg
	}
	}

[joecompute]

LGTM

[cmacknz]

I think the data plane team is only requested here because we are default code owners of the new changelog/fragments directory. #46931 should fix that.

So approving that part. We aren't code owners of the s3 or cloudwatch inputs, but if we were I'd ask for a unit test of this change ;)

[Kavindu-Dodan]

@joecompute @cmacknz thank you for the review and explanation on why state doesn't get updated.

Regarding unit tests, I checked into this. But currently we do not have any unit tests targeting Input.run. Given this PR focus on a fix for incorrect reporting and the time investment to add all mocks needed to test the method, I will skip the unit test on this PR.

[Kavindu-Dodan]

Update : I wil merge this on 06.10.2025