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

[streaming] - Made namespace consistent in logging & put a null check to stop paincs on shutdown #42315

Merged
merged 10 commits into from
Jan 17, 2025
Merged

[streaming] - Made namespace consistent in logging & put a null check to stop paincs on shutdown #42315

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
0176f47
added blanket & infinite retry options and impoved logging
ShourieG Jan 7, 2025
a04b20f
removed an unnecessary namespace variable for a debug log causing a b…
ShourieG Jan 15, 2025
864925a
merged conflict resolved
ShourieG Jan 15, 2025
339327b
updated changelog
ShourieG Jan 15, 2025
9ab94bf
fixed a situation when the input would panic during shutdown
ShourieG Jan 15, 2025
fcb19b4
addressed Andrew's suggestions
ShourieG Jan 16, 2025
31503e8
updated changelog
ShourieG Jan 16, 2025
71d0403
updated linter
ShourieG Jan 16, 2025
8d1c570
Merge remote-tracking branch 'upstream/main' into streaming/dubug_fix
ShourieG Jan 16, 2025
68b89bd
updated changelog
ShourieG Jan 17, 2025
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 @@ -206,6 +206,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
- The `_id` generation process for S3 events has been updated to incorporate the LastModified field. This enhancement ensures that the `_id` is unique. {pull}42078[42078]
- Fix Netflow Template Sharing configuration handling. {pull}42080[42080]
- Updated websocket retry error code list to allow more scenarios to be retried which could have been missed previously. {pull}42218[42218]
- Removed an unnecessary namespace variable & put a null check to stop paincs on shutdown. {pull}42315[42315]

*Heartbeat*

Expand Down
10 changes: 6 additions & 4 deletions x-pack/filebeat/input/streaming/websocket.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -154,9 +154,11 @@ func (s *websocketStream) FollowStream(ctx context.Context) error {

// ensures this is the last connection closed when the function returns
defer func() {
if err := c.Close(); err != nil {
s.metrics.errorsTotal.Inc()
s.log.Errorw("encountered an error while closing the websocket connection", "error", err)
if c != nil {
if err := c.Close(); err != nil {
s.metrics.errorsTotal.Inc()
s.log.Errorw("encountered an error while closing the websocket connection", "error", err)
}
}
}()

Expand Down Expand Up @@ -217,7 +219,7 @@ func (s *websocketStream) FollowStream(ctx context.Context) error {
}
s.metrics.receivedBytesTotal.Add(uint64(len(message)))
state["response"] = message
s.log.Debugw("received websocket message", logp.Namespace("websocket"), "msg", string(message))
s.log.Debugw("received websocket message", "msg", string(message))
Copy link
Member

Choose a reason for hiding this comment

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

There was nothing wrong with this version of the line. Checking the history, the problem was already addressed via 2ea4a2b which has become part of v8.16.2.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Ok, but is the namespace call required though? We are already mentioning websocket in the error description, and none of the other local debug logs are using namespace. Should we still keep it around at this place ?

Copy link
Member

Choose a reason for hiding this comment

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

Some context data is namespaced (and some isn't).

https://github.com/search?q=repo%3Aelastic%2Fbeats+path%3A%2F%5Ex-pack%5C%2Ffilebeat%5C%2Finput%5C%2Fstreaming%5C%2F%2F+lang%3Ago+%2Flogp%5C.Namespace%5C%28%2F&type=code

If anything I would ensure that it uses the namespace consistently, rather than removing it. The other calls depend on ns variable. And in one case, rather than using a custom namespaced field there is a perfectly good ECS field for the job -- http.response.body.content.

diff --git a/x-pack/filebeat/input/streaming/crowdstrike.go b/x-pack/filebeat/input/streaming/crowdstrike.go
index eb1797d2f6..7d062ce827 100644
--- a/x-pack/filebeat/input/streaming/crowdstrike.go
+++ b/x-pack/filebeat/input/streaming/crowdstrike.go
@@ -156,7 +156,7 @@ func (s *falconHoseStream) followSession(ctx context.Context, cli *http.Client,
 	if err != nil {
 		return state, Warning{fmt.Errorf("failed to decode discover body: %w", err)}
 	}
-	s.log.Debugw("stream discover metadata", "meta", mapstr.M(body.Meta))
+	s.log.Debugw("stream discover metadata", logp.Namespace(s.ns), "meta", mapstr.M(body.Meta))
 
 	var offset int
 	if cursor, ok := state["cursor"].(map[string]any); ok {
@@ -241,7 +241,7 @@ func (s *falconHoseStream) followSession(ctx context.Context, cli *http.Client,
 			}
 			s.metrics.receivedBytesTotal.Add(uint64(len(msg)))
 			state["response"] = []byte(msg)
-			s.log.Debugw("received firehose message", logp.Namespace("falcon_hose"), "msg", debugMsg(msg))
+			s.log.Debugw("received firehose message", logp.Namespace(s.ns), "msg", debugMsg(msg))
 			err = s.process(ctx, state, s.cursor, s.now().In(time.UTC))
 			if err != nil {
 				s.log.Errorw("failed to process and publish data", "error", err)
diff --git a/x-pack/filebeat/input/streaming/websocket.go b/x-pack/filebeat/input/streaming/websocket.go
index eeb89ad5c9..8377c29e37 100644
--- a/x-pack/filebeat/input/streaming/websocket.go
+++ b/x-pack/filebeat/input/streaming/websocket.go
@@ -217,7 +217,7 @@ func (s *websocketStream) FollowStream(ctx context.Context) error {
 			}
 			s.metrics.receivedBytesTotal.Add(uint64(len(message)))
 			state["response"] = message
-			s.log.Debugw("received websocket message", logp.Namespace("websocket"), "msg", string(message))
+			s.log.Debugw("received websocket message", logp.Namespace(s.ns), "msg", string(message))
 			err = s.process(ctx, state, s.cursor, s.now().In(time.UTC))
 			if err != nil {
 				s.metrics.errorsTotal.Inc()
@@ -294,7 +294,7 @@ func handleConnectionResponse(resp *http.Response, metrics *inputMetrics, log *l
 			buf.WriteString("... truncated")
 		}
 
-		log.Debugw("websocket connection response", "body", &buf)
+		log.Debugw("websocket connection response", "http.response.body.content", &buf)
 	}
 }

err = s.process(ctx, state, s.cursor, s.now().In(time.UTC))
if err != nil {
s.metrics.errorsTotal.Inc()
Expand Down
Loading