[8.17](backport #44342) [filebeat][streaming] - Fixed a websocket panic scenario which would occur after exhausting max retries (#44389)

* [filebeat][streaming] - Fixed a websocket panic scenario which would occur after exhausting max retries (#44342)

In some scenarios the websocket response object can become nil and
cause  certain logging statements to cause an unexpected panic. This PR
fixes once  such identified scenario.

(cherry picked from commit f6994c0)

* Update CHANGELOG.next.asciidoc

---------

Co-authored-by: Shourie Ganguly <shourie.ganguly@elastic.co>

Author: mergify[bot]

CHANGELOG.next.asciidoc

@@ -114,6 +114,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Fix publication of group data from the Okta entity analytics provider. {pull}40681[40681]
 - Ensure netflow custom field configuration is applied. {issue}40735[40735] {pull}40730[40730]
 - Fix a bug in Salesforce input to only handle responses with 200 status code {pull}41015[41015]
+- Fixed a websocket panic scenario which would occur after exhausting max retries. {pull}44342[44342]
 
 *Heartbeat*
 

x-pack/filebeat/input/streaming/input_test.go

@@ -127,7 +127,8 @@ var inputTests = []struct {
 					"events": [inner_body],
 				})`,
 		},
-		response: []string{`
+		response: []string{
+			`
 			{
 				"pps": {
 					"agent": "example.proofpoint.com",
@@ -177,7 +178,8 @@ var inputTests = []struct {
 					"pri": 35342
 				},
 				"id": "ZeYGULpZmL5N0151HN1OyX"
-	   }`},
+	   }`,
+		},
 		want: []map[string]interface{}{
 			{
 				"pps": map[string]interface{}{
@@ -278,7 +280,8 @@ var inputTests = []struct {
 				},
 			},
 		},
-		response: []string{`
+		response: []string{
+			`
 	       {
 	          "pps": {
 	              "agent": "example.proofpoint.com",
@@ -317,7 +320,8 @@ var inputTests = []struct {
 				"basic_token": basicToken,
 			},
 		},
-		response: []string{`
+		response: []string{
+			`
 	       {
 	          "pps": {
 	              "agent": "example.proofpoint.com",
@@ -349,7 +353,8 @@ var inputTests = []struct {
 				"bearer_token": bearerToken,
 			},
 		},
-		response: []string{`
+		response: []string{
+			`
 	       {
 	          "pps": {
 	              "agent": "example.proofpoint.com",
@@ -384,7 +389,8 @@ var inputTests = []struct {
 				},
 			},
 		},
-		response: []string{`
+		response: []string{
+			`
 	       {
 	          "pps": {
 	              "agent": "example.proofpoint.com",
@@ -418,7 +424,8 @@ var inputTests = []struct {
 				"wait_max":     "2s",
 			},
 		},
-		response: []string{`
+		response: []string{
+			`
 	       {
 	          "pps": {
 	              "agent": "example.proofpoint.com",
@@ -815,7 +822,6 @@ func TestURLEval(t *testing.T) {
 	logp.TestingSetup()
 	for _, test := range urlEvalTests {
 		t.Run(test.name, func(t *testing.T) {
-
 			cfg := conf.MustNewConfigFrom(test.config)
 
 			conf := config{}

x-pack/filebeat/input/streaming/websocket.go

@@ -332,6 +332,9 @@ func connectWebSocket(ctx context.Context, cfg config, url string, log *logp.Log
 				waitTime := calculateWaitTime(retryConfig.WaitMin, retryConfig.WaitMax, attempt)
 				time.Sleep(waitTime)
 			}
+			if response == nil {
+				return nil, nil, fmt.Errorf("failed to establish WebSocket connection after %d attempts with error %w", retryConfig.MaxAttempts, err)
+			}
 			return nil, nil, fmt.Errorf("failed to establish WebSocket connection after %d attempts with error %w and (status %d)", retryConfig.MaxAttempts, err, response.StatusCode)
 		} else {
 			for attempt := 1; ; attempt++ {