[TT-13139] Request times out in some cases when sending input via http inputs

[buraksezer]

User description

PR for https://tyktech.atlassian.net/browse/TT-13139

• defaultStreamManager has removed, I used Gorilla's muxer directly instead of creating a default stream manager at the beginning
• I managed to remove some code because defaultStreamManager is gone.
• ProcessRequest method has simplified and split into different methods for readability.
• Tyk Streams not creating a new Bento stream if you publish messages via the HTTP endpoint. It iterates over the existing streams and hands over the request to Bento.
• Added two new integration tests

---

PR Type

Enhancement, Tests

---

Description

• Removed the defaultStreamManager and refactored the streaming middleware to use Gorilla's muxer directly, simplifying the codebase.
• Enhanced the ProcessRequest method by splitting it into smaller, more manageable functions for better readability and maintainability.
• Introduced new methods such as registerHandlers, inputHttpServerPublishHandler, and subscriptionHandler to improve HTTP request handling.
• Added integration tests to verify the functionality of HTTP server input handling, ensuring that messages are correctly received by WebSocket clients.

---

Changes walkthrough 📝

	Relevant files
Enhancement	mw_streaming.go Refactor streaming middleware and enhance HTTP handling    gateway/mw_streaming.go Removed the defaultStreamManager and used Gorilla's muxer directly. Simplified and refactored ProcessRequest method. Added new methods registerHandlers, inputHttpServerPublishHandler, and subscriptionHandler. Introduced dummyResponseWriter for handling HTTP responses. +191/-82
Tests	mw_streaming_test.go Add integration tests for HTTP server input handling          gateway/mw_streaming_test.go Added integration tests for HTTP server input handling. Introduced new test templates for HTTP server input. Verified WebSocket message reception for single and multiple clients. +119/-1

---

💡 PR-Agent usage: Comment /help "your question" on any pull request to receive relevant information

[github-actions]

PR Reviewer Guide 🔍

⏱️ Estimated effort to review: 4 🔵🔵🔵🔵⚪

🧪 PR contains tests

🔒 No security concerns identified

⚡ Key issues to review Error Handling The error handling in inputHttpServerPublishHandler does not properly handle the case where copyBody fails. It should return an HTTP error response immediately instead of continuing to iterate over the streams. Concurrency Concerns The streamManagers.Range function in inputHttpServerPublishHandler iterates over stream managers and modifies requests concurrently. This could lead to race conditions or inconsistent states if not handled properly. Resource Leak In inputHttpServerPublishHandler, the body of the request is cloned but not closed in some paths, potentially leading to resource leaks. Missing Error Handling The function getRouteMatch in StreamingMiddleware does not handle the case where muxer.Match might return an error. This could lead to unhandled errors during route matching.

[github-actions]

API Changes

--- prev.txt	2024-10-01 14:03:30.211762176 +0000
+++ current.txt	2024-10-01 14:03:24.067722220 +0000
@@ -7563,6 +7563,11 @@
 	ErrOAuthClientDeleted               = "oauth.client_deleted"
 )
 const (
+
+	// ExtensionTykStreaming is the oas extension for tyk streaming
+	ExtensionTykStreaming = "x-tyk-streaming"
+)
+const (
 	ResetQuota              string = "resetQuota"
 	CertificateRemoved      string = "CertificateRemoved"
 	CertificateAdded        string = "CertificateAdded"
@@ -7614,10 +7619,6 @@
     The name for event handlers as defined in the API Definition JSON/BSON
     format
 
-const (
-	// ExtensionTykStreaming is the oas extension for tyk streaming
-	ExtensionTykStreaming = "x-tyk-streaming"
-)
 const ListDetailed = "detailed"
 const LoopScheme = "tyk"
 const OIDPREFIX = "openid"

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Score
Possible bug	Handle missing keys in streamConfig to prevent runtime errors Consider handling the case where streamConfig does not contain an 'input' or 'http_server' key to avoid runtime panics. gateway/mw_streaming.go [210-218] if componentMap, ok := streamConfig["input"].(map[string]interface{}); ok { if httpServerConfig, ok := componentMap["http_server"].(map[string]interface{}); ok { if val, ok := httpServerConfig["path"].(string); ok { return val } } } +return defaultHTTPServerInputPath // Handle missing keys gracefully Suggestion importance[1-10]: 8 Why: Handling cases where streamConfig lacks expected keys prevents runtime panics, which is crucial for maintaining application stability. This suggestion addresses a potential bug and improves code robustness.	8
	Add error handling for route matching failures to improve robustness Implement error handling for the router.Match method to manage cases where the route matching fails. gateway/mw_streaming.go [339-340] if !s.router.Match(strippedPathRequest, routeMatch) { - return nil, http.StatusOK + return fmt.Errorf("failed to match route"), http.StatusInternalServerError } Suggestion importance[1-10]: 8 Why: Implementing error handling for route matching failures enhances the robustness of the code by ensuring that unexpected conditions are managed gracefully, preventing potential silent failures.	8
	Enhance the reliability of request body copying by handling errors and nil bodies Use a more robust method for copying the request body to handle potential errors and ensure the body is not empty. gateway/mw_streaming.go [361-363] body, err = copyBody(r.Body, true) -if err != nil { - return false // break +if err != nil || body == nil { + return false // break, handle nil body or error } Suggestion importance[1-10]: 7 Why: Improving the method for copying the request body to handle errors and nil bodies increases the reliability of the code, preventing potential issues during request processing. This suggestion addresses a possible bug and enhances code robustness.	7
Security	Add validation for the httpServerInputPath to enhance security and reliability Ensure that the httpServerInputPath is properly validated before usage to prevent potential security risks or runtime errors. gateway/mw_streaming.go [308] -httpServerInputPath: findHTTPServerInputPath(config), +httpServerInputPath: validateHTTPServerInputPath(findHTTPServerInputPath(config)), Suggestion importance[1-10]: 7 Why: Adding validation for httpServerInputPath can prevent potential security risks or runtime errors, enhancing the robustness and security of the code. This suggestion is relevant and improves the reliability of the system.	7

💡 Need additional feedback ? start a PR chat

[buger]

@buraksezer if there is no defaultStreamManager how it would support background jobs? You have to create default streamManager if theare no http_server entries in any case. E.g. you have a input as kafka, and output as webhook.

Also you can't depend on findHTTPServerInputPath, you should use GetHTTPPaths, because findHTTPServerInputPath does not consider broker. Also it check only for input but it should work with output too.

[buraksezer]

@buraksezer if there is no defaultStreamManager how it would support background jobs? You have to create default streamManager if theare no http_server entries in any case. E.g. you have a input as kafka, and output as webhook.

I tried to fix this by calling s.createStreamManager(nil) in Init method. Could you please review it again?

Also you can't depend on findHTTPServerInputPath, you should use GetHTTPPaths, because findHTTPServerInputPath does not consider broker. Also it check only for input but it should work with output too.

I understood the issue but I'm not sure how to extract path from the return value of GetHTTPPaths. I thought I only need the configuration variable named path of the input section.

Here it is: https://warpstreamlabs.github.io/bento/docs/components/inputs/http_server/#path

[buraksezer]

@buger

Also you can't depend on findHTTPServerInputPath, you should use GetHTTPPaths, because findHTTPServerInputPath does not consider broker. Also it check only for input but it should work with output too.

I would like to extend the previous comment.inputHttpServerPublishHandler only intended in the case of HTTP Server is configured as input. So findHTTPServerInputPath only tries to find this configuration element: https://warpstreamlabs.github.io/bento/docs/components/inputs/http_server/#path

[sonarcloud]

Quality Gate failed

Failed conditions
C Reliability Rating on New Code (required ≥ A)

See analysis details on SonarCloud

Catch issues before they fail your Quality Gate with our IDE extension SonarLint