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

[ShourieG]

Type of change

• Bug

Proposed commit message

1. Made namespace consistent in suggested debug logs.
2. Put in a null check during input shutdown to avoid a panic.

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.

Disruptive User Impact

Author's Checklist

• [ ]

How to test this PR locally

Related issues

Use cases

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 @ShourieG? 🙏.
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

[mergify]

backport-8.x has been added to help with the transition to the new branch 8.x.
If you don't need it please use backport-skip label and remove the backport-8.x label.

[elasticmachine]

Pinging @elastic/security-service-integrations (Team:Security-Service Integrations)

[andrewkroh]

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.

[ShourieG]

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 ?

[andrewkroh]

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)
 	}
 }

[andrewkroh]

The logging call doesn't need a change.

[andrewkroh]

Context for reviewers

This is the panic being addressed. It occurred under v8.16.1. (via OCR so it contains some wonky text)

Input 'streaming' failed with: input websocket-proofpoint_on_demand.audit-f16e440c-5bc4-4710-9836-067c1b2b3398 failed input panic with runtime error invalid memory address or nil pointer dereference goroutine 138 [running]:

runtime/debug.Stack()
    runtime/debug/stack.go:24 + 0x64
github.com/elastic/beats/v7/filebeat/input/v2/input-cursor.jmanagedinput).rumSource.func10
    github.com/elastic/beats/v7/filebeat/input/v2/input-cursor/input.gos144 +0x48
panic((Oxaaaad0c16f80?, Oxaaaad5d234107M
    runtime/panic.go:770 +0x124
github.com/gorilla/websocket.(*Conn).Close(...)
    github.com/gorilla/[email protected]/conn.go:345
github.com/elastic/beats/v7/x-pack/filebeat/input/streaming. (*websocketStream)FollowStream.fung10
    github.com/elastic/beats/v7/x-pack/filebeat/input/streaming/websocket.go:102 +0x24
github.com/elastic/beats/v7/x-pack/filebeat/input/streaming-(*websocketStream).FollowStream(Qx400250ca80, [Oxaaaad1c6dcd&0«4001136550})
    github.com/elastic/beats/vrx-pack/filebeat/input/streaming/websocket.go:130 +0x8ec
github.com/elastic/beats/v7/x-pack/filebeat/input/streaming.input.run(fOx0, 0x0}, 0x0, (0x0, 0x0}, (0x0, 0x0}, (0x0,0x0), 0x0,32)
    github.com/elastic/beats/v7/x-pack/filebeat/input/streaming/input.go:108 +0x2cc
github.com/elastic/beats/v7/x-pack/filebeat/input/streaming.inputRun(ROx, 0x0}, 0x0, (axa, 0x0}. (0x0,0x0)(0x0,0x0)0x0,47,)
    github.com/elastic/beats/v7/x-pack/filebeat/input/streaming/input.go:83 +0xdc
github.com/elastic/beats/V7/filebeat/input/v2/input-cursor.(managedinput).runSourcel 0x4001be75c0, (Qx40024c2dc0,0x92}, (0*40001a7540, 0*49), (Oxaaaacf2 1bee7, 0x8), (Oxaaaacf21bee7, 0×8},}, -}._)
    github.com/elastic/beats/v7/filebeat/input/v2/input-cursor/input.ga:168 +0x388
github.com/elastic/beats/v7/filebeat/input/v2/input-cursor.l*managedinput).Run.func10
    github.com/elastic/beats/v7/filebeat/input/v2/input-cursor/input.go:123 +0x168
github.com/elastic/go-concert/unison.(*MultiErrGroup).Ga.func10
    github.com/elastic/[email protected]/unison/multierrgroup.go:42 +0x6C
created by github.com/elastic/go-concert/unison.(*MultiErrGroup).Go in goroutine 356
    github.com/elastic/[email protected]/unison/multierrgroup.go:40 +0x84

[ShourieG]

@andrewkroh, I've made the suggested changes

[andrewkroh]

Suggested change

	- In the streaming input made namespace consistent in logs & put a null check to stop paincs on shutdown. {pull}42315[42315]
	- In the `streaming` input, prevent panics on shutdown with a null check and apply a consistent namespace to contextual data in debug logs. {pull}42315[42315]