Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[azure-eventhub] Update input v1 status on start, failure, and stop #41469

Merged
merged 3 commits into from
Nov 7, 2024
Merged

[azure-eventhub] Update input v1 status on start, failure, and stop #41469

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5212494
Update input v1 status on start, failure, and stop
zmoog Oct 28, 2024
c631779
Update status in all input lifecycle phases
zmoog Nov 6, 2024
3fa45d2
Update CHANGELOG
zmoog Nov 6, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.next.asciidoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -173,6 +173,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
- Journald input now can read events from all boots {issue}41083[41083] {pull}41244[41244]
- Fix double encoding of client_secret in the Entity Analytics input's Azure Active Directory provider {pull}41393[41393]
- Fix errors in SQS host resolution in the `aws-s3` input when using custom (non-AWS) endpoints. {pull}41504[41504]
- The azure-eventhub input now correctly reports its status to the Elastic Agent on fatal errors {pull}41469[41469]

*Heartbeat*

Expand Down
10 changes: 10 additions & 0 deletions x-pack/filebeat/input/azureeventhub/v1_input.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
v2 "github.com/elastic/beats/v7/filebeat/input/v2"
"github.com/elastic/beats/v7/libbeat/beat"
"github.com/elastic/beats/v7/libbeat/common/acker"
"github.com/elastic/beats/v7/libbeat/management/status"
"github.com/elastic/elastic-agent-libs/logp"
"github.com/elastic/elastic-agent-libs/mapstr"
)
Expand Down Expand Up @@ -68,9 +69,13 @@ func (in *eventHubInputV1) Run(
) error {
var err error

// Update the status to starting
inputContext.UpdateStatus(status.Starting, "")

// Create pipelineClient for publishing events.
in.pipelineClient, err = createPipelineClient(pipeline)
if err != nil {
inputContext.UpdateStatus(status.Failed, err.Error())
return fmt.Errorf("failed to create pipeline pipelineClient: %w", err)
}
defer in.pipelineClient.Close()
Expand All @@ -82,6 +87,7 @@ func (in *eventHubInputV1) Run(
// Set up new and legacy sanitizers, if any.
sanitizers, err := newSanitizers(in.config.Sanitizers, in.config.LegacySanitizeOptions)
if err != nil {
inputContext.UpdateStatus(status.Failed, err.Error())
return fmt.Errorf("failed to create sanitizers: %w", err)
}

Expand All @@ -98,16 +104,20 @@ func (in *eventHubInputV1) Run(
// in preparation for the main run loop.
err = in.setup(ctx)
if err != nil {
in.log.Errorw("error setting up input", "error", err)
inputContext.UpdateStatus(status.Failed, err.Error())
return err
}

// Start the main run loop
err = in.run(ctx)
if err != nil {
in.log.Errorw("error running input", "error", err)
inputContext.UpdateStatus(status.Failed, err.Error())
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hey @faec, I missed the opportunity to report the input status to the Elastic Agent when I updated the input to the Input API v2.

I am trying to address scenarios (like elastic/integrations#9659) where the input starts, and after a while, it encounters a fatal error, and the SDK worker shuts down.

What are your recommendations for reporting input status to the agent? Which input should I use as a reference?

🙇

return err
}

inputContext.UpdateStatus(status.Stopping, "")
return nil
}

Expand Down
Loading