[TT-12893]: Adding first implementation of streams API

[kofoworola]

User description

Description

Based off this POC here
TT-12893

Related Issue

Motivation and Context

How This Has Been Tested

Screenshots (if appropriate)

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)
• Refactoring or add test (improvements in base code or adds test coverage to functionality)

Checklist

• I ensured that the documentation is up to date
• I explained why this PR updates go.mod in detail with reasoning why it's required
• I would like a code coverage CI quality gate exception and have explained why

---

PR Type

Enhancement, Tests

---

Description

• Implemented StreamingMiddleware to handle streaming functionality, including stream management and integration with API specifications.
• Added comprehensive tests for streaming API, covering single and multiple client scenarios, asynchronous API, and WebSocket connections.
• Developed Stream struct for managing streaming configurations, with methods for starting, stopping, and resetting streams.
• Introduced PortalClient for interacting with the developer portal, including methods to list webhook credentials and fetch app details.
• Implemented portalOutput for sending messages to webhooks, with configuration and connection handling.
• Enhanced middleware interface to support unloading functionality, with hooks for unloading middleware when API specs change.
• Updated dependencies to support new streaming and portal functionalities, including Kafka, NATS, and Benthos integration.

---

Changes walkthrough 📝

	Relevant files
Tests	3 files mw_streaming_test.go Add comprehensive tests for streaming API functionality    gateway/mw_streaming_test.go Added tests for streaming API with single and multiple clients. Implemented setup functions for streaming API tests. Included test cases for asynchronous API and WebSocket connections. +670/-0  manager_test.go Implement tests for Stream management and safety checks    internal/streaming/manager_test.go Added tests for starting and stopping streams. Tested removal and whitelisting of unsafe components. Verified stream configuration handling. +151/-0  portal_test.go Add tests for PortalClient webhook credential listing        internal/portal/portal_test.go Added mock server for testing portal client interactions. Implemented tests for listing webhook credentials. Verified correct handling of multiple apps and webhooks. +76/-0
Enhancement	6 files mw_streaming.go Implement StreamingMiddleware for API streaming functionality gateway/mw_streaming.go Implemented StreamingMiddleware for handling streaming functionality. Added methods for initializing, creating, and removing streams. Integrated stream management with API specifications. +298/-0  manager.go Develop Stream struct for managing streaming configurations internal/streaming/manager.go Created Stream struct for managing streaming configurations. Added methods for starting, stopping, and resetting streams. Implemented safety checks for removing unsafe components. +236/-0  portal_client.go Add PortalClient for developer portal interactions              internal/portal/portal_client.go Introduced PortalClient for interacting with the developer portal. Added methods to list webhook credentials and fetch app details. Defined structures for app and webhook details. +152/-0  portal_output.go Implement portalOutput for webhook message delivery            internal/portal/portal_output.go Implemented portalOutput for sending messages to webhooks. Added configuration and connection handling for portal output. Registered output plugin with Benthos service. +128/-0  api_loader.go Integrate StreamingMiddleware into API loading process      gateway/api_loader.go Integrated StreamingMiddleware into the API loading process. Adjusted logic for unloading API specifications. +9/-5      middleware.go Enhance middleware interface with unload functionality      gateway/middleware.go Enhanced middleware interface with unload functionality. Added hooks for unloading middleware when API specs change. +15/-1
Configuration changes	1 files config.go Add StreamingConfig to global configuration settings          config/config.go Added StreamingConfig struct to configuration. Enabled streaming configuration in global settings. +26/-0
Dependencies	1 files go.mod Update dependencies for streaming and portal integration  go.mod Updated dependencies for streaming and portal functionalities. Added new modules for Kafka, NATS, and Benthos integration. +307/-23
Additional files (token-limit)	1 files go.sum ...                                                                                                            go.sum ... +1134/-65

---

💡 PR-Agent usage:
Comment /help on the PR to get a list of all available PR-Agent tools and their descriptions

[github-actions]

API Changes

--- prev.txt	2024-09-12 15:16:42.513965515 +0000
+++ current.txt	2024-09-12 15:16:39.493961087 +0000
@@ -5155,6 +5155,10 @@
 		LivenessCheck: LivenessCheckConfig{
 			CheckDuration: time.Second * 10,
 		},
+		Streaming: StreamingConfig{
+			Enabled:     false,
+			AllowUnsafe: []string{},
+		},
 	}
 )
 var Global func() Config
@@ -5765,6 +5769,10 @@
 
 	// OAS holds the configuration for various OpenAPI-specific functionalities
 	OAS OASConfig `json:"oas_config"`
+
+	Streaming StreamingConfig `json:"streaming"`
+
+	Labs labsConfig `json:"labs"`
 }
     Config is the configuration object used by Tyk to set up various parameters.
 
@@ -6363,6 +6371,12 @@
 func (config *StorageOptionsConf) HostAddrs() (addrs []string)
     HostAddrs returns a sanitized list of hosts to connect to.
 
+type StreamingConfig struct {
+	Enabled     bool     `json:"enabled"`
+	AllowUnsafe []string `json:"allow_unsafe"`
+}
+    Add this new struct definition
+
 type Tracer struct {
 	// The name of the tracer to initialize. For instance appdash, to use appdash tracer
 	Name string `json:"name"`
@@ -7641,6 +7655,9 @@
     The name for event handlers as defined in the API Definition JSON/BSON
     format
 
+const (
+	ExtensionTykStreaming = "x-tyk-streaming"
+)
 const ListDetailed = "detailed"
 const LoopScheme = "tyk"
 const OIDPREFIX = "openid"
@@ -7904,6 +7921,9 @@
 
 func CloneAPI(a *APISpec) *APISpec
 
+func (s *APISpec) AddUnloadHook(hook func())
+    AddUnloadHook adds a function to be called when the API spec is unloaded
+
 func (a *APISpec) CheckSpecMatchesStatus(r *http.Request, rxPaths []URLSpec, mode URLStatus) (bool, interface{})
     CheckSpecMatchesStatus checks if a URL spec has a specific status.
     Deprecated: The function doesn't follow go return conventions (T, ok);
@@ -7925,9 +7945,6 @@
 
 func (a *APISpec) Init(authStore, sessionStore, healthStore, orgStore storage.Handler)
 
-func (s *APISpec) Release()
-    Release releases all resources associated with API spec
-
 func (a *APISpec) RequestValid(r *http.Request) (bool, RequestStatus)
     RequestValid will check if an incoming request has valid version data and
     return a RequestStatus that describes the status of the request
@@ -7947,6 +7964,9 @@
 func (a *APISpec) URLAllowedAndIgnored(r *http.Request, rxPaths []URLSpec, whiteListStatus bool) (RequestStatus, interface{})
     URLAllowedAndIgnored checks if a url is allowed and ignored.
 
+func (s *APISpec) Unload()
+    Release releases all resources associated with API spec
+
 func (s *APISpec) Validate(oasConfig config.OASConfig) error
     Validate returns nil if s is a valid spec and an error stating why the spec
     is not valid.
@@ -8039,6 +8059,8 @@
     FireEvent is added to the BaseMiddleware object so it is available across
     the entire stack
 
+func (t *BaseMiddleware) GetSpec() *APISpec
+
 func (t *BaseMiddleware) Init()
 
 func (t *BaseMiddleware) Logger() (logger *logrus.Entry)
@@ -8053,6 +8075,8 @@
 
 func (t *BaseMiddleware) SetRequestLogger(r *http.Request)
 
+func (t *BaseMiddleware) Unload()
+
 func (t *BaseMiddleware) UpdateRequestSession(r *http.Request) bool
 
 type BaseTykResponseHandler struct {
@@ -10305,6 +10329,29 @@
 
 type StatsDSinkSanitizationFunc func(*bytes.Buffer, string)
 
+type StreamManager struct {
+	// Has unexported fields.
+}
+
+type StreamingMiddleware struct {
+	*BaseMiddleware
+
+	// Has unexported fields.
+}
+    StreamingMiddleware is a middleware that handles streaming functionality
+
+func (s *StreamingMiddleware) EnabledForSpec() bool
+
+func (s *StreamingMiddleware) Init()
+    Init initializes the middleware
+
+func (s *StreamingMiddleware) Name() string
+
+func (s *StreamingMiddleware) ProcessRequest(w http.ResponseWriter, r *http.Request, _ interface{}) (error, int)
+    ProcessRequest will handle the streaming functionality
+
+func (s *StreamingMiddleware) Unload()
+
 type StripAuth struct {
 	*BaseMiddleware
 }
@@ -10535,6 +10582,10 @@
 	ProcessRequest(w http.ResponseWriter, r *http.Request, conf interface{}) (error, int) // Handles request
 	EnabledForSpec() bool
 	Name() string
+
+	GetSpec() *APISpec
+
+	Unload()
 }
 
 type TykOsinServer struct {

[github-actions]

PR Reviewer Guide 🔍

⏱️ Estimated effort to review: 5 🔵🔵🔵🔵🔵

🧪 No relevant tests

🔒 No security concerns identified

⚡ Key issues to review Possible Bug The test TestStreamingAPISingleClient and TestStreamingAPIMultipleClients rely on external services via testcontainers. This approach can lead to flaky tests if the external service is not available or the environment is not correctly configured. Consider mocking these services or using a more controlled integration testing strategy. Performance Issue The StreamingMiddleware implementation could lead to performance bottlenecks. The middleware initializes a new StreamManager for each request, which could be very inefficient under high load. Consider reusing stream managers or using a pool of pre-initialized managers. Security Concern The method removeUnsafe in Stream class attempts to sanitize configurations by removing unsafe components. However, this method uses regular expressions to manipulate YAML strings directly, which can be error-prone and might not correctly handle all YAML complexities. Consider using a more robust YAML parsing strategy to manipulate the configurations safely.

[github-actions]

PR Code Suggestions ✨

Category	Suggestion	Score
Possible issue	Add locking around accesses to streamManagers to ensure thread safety To avoid potential race conditions, consider using a lock when accessing and modifying streamManagers within StreamingMiddleware. gateway/mw_streaming.go [122] +s.streamManagersMutex.Lock() s.streamManagers.Store(streamID, newStreamManager) +s.streamManagersMutex.Unlock() Suggestion importance[1-10]: 9 Why: The suggestion addresses a potential race condition by adding a lock around streamManagers access, which is important for ensuring thread safety in concurrent environments.	9
Security	Improve security by enabling TLS certificate verification Replace the insecure TLS configuration with a secure one by removing InsecureSkipVerify: true. This change will ensure that the client validates the server's certificate, which is crucial for preventing man-in-the-middle attacks. gateway/mw_streaming_test.go [118] -TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, +TLSClientConfig: &tls.Config{}, Suggestion importance[1-10]: 9 Why: This suggestion addresses a significant security concern by ensuring that the TLS client configuration verifies the server's certificate, preventing man-in-the-middle attacks. It is a crucial improvement for secure communication.	9
	Prevent potential JSON injection by properly escaping JSON keys Replace the use of fmt.Sprintf for constructing JSON keys with a more robust method that ensures keys are properly escaped to prevent JSON injection issues. gateway/mw_streaming_test.go [56] -m[fmt.Sprintf("%v", k)] = convertToStringKeyMap(v) +m[strconv.Quote(fmt.Sprintf("%v", k))] = convertToStringKeyMap(v) Suggestion importance[1-10]: 8 Why: The suggestion enhances security by ensuring JSON keys are properly escaped, preventing potential JSON injection vulnerabilities. This is an important improvement for handling untrusted input safely.	8
Best practice	Use a passed-in context instead of context.Background() for better control over goroutines Replace the direct usage of context.Background() with a context passed as a parameter to the Start and Stop methods to allow better control over context deadlines and cancellation propagation from calling functions. internal/streaming/manager.go [93-119] -errChan <- stream.Run(context.Background()) -stopCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) +errChan <- stream.Run(ctx) +stopCtx, cancel := context.WithTimeout(ctx, 5*time.Second) Suggestion importance[1-10]: 9 Why: This suggestion improves the flexibility and control over the context, which is a best practice for managing goroutines and their lifecycles. It addresses a significant aspect of the code's robustness and maintainability.	9
Possible bug	Add a nil check for Spec to prevent potential nil pointer dereference Ensure that the Unload method checks if the Spec is not nil before accessing it to prevent a nil pointer dereference. gateway/middleware.go [282] -return t.Spec +if t.Spec != nil { + return t.Spec +} +return nil Suggestion importance[1-10]: 8 Why: The suggestion correctly adds a nil check for Spec, which is crucial to prevent potential nil pointer dereference, enhancing the safety and stability of the code.	8
	Add a nil check for handler to prevent a nil pointer dereference Consider handling the case where handler is nil after type assertion to avoid a potential nil pointer dereference when calling ServeHTTP. gateway/mw_streaming.go [219] -handler.ServeHTTP(w, r) +if handler != nil { + handler.ServeHTTP(w, r) +} else { + s.Logger().Error("Handler is nil") + http.Error(w, "Internal server error", http.StatusInternalServerError) +} Suggestion importance[1-10]: 8 Why: This suggestion adds a necessary nil check for handler, preventing a potential nil pointer dereference, which is crucial for avoiding runtime errors and ensuring the code's robustness.	8
	Add error handling for nil mux to prevent runtime errors Handle the case where mux is nil before calling builder.SetHTTPMux(mux) to prevent potential nil dereference issues. internal/streaming/manager.go [75-77] -if mux != nil { - builder.SetHTTPMux(mux) +if mux == nil { + s.log.Error("HTTP multiplexer is nil") + return fmt.Errorf("HTTP multiplexer cannot be nil") } +builder.SetHTTPMux(mux) Suggestion importance[1-10]: 8 Why: The suggestion addresses a potential bug by adding error handling for a nil mux, which could prevent runtime errors and improve the robustness of the code.	8
	Add a nil check for the slice before iterating to prevent potential runtime errors Consider checking if unloadHooks is not nil before iterating over it to avoid potential nil pointer dereference. gateway/api_definition.go [279-281] -for _, hook := range s.unloadHooks { - hook() +if s.unloadHooks != nil { + for _, hook := range s.unloadHooks { + hook() + } } Suggestion importance[1-10]: 7 Why: Adding a nil check before iterating over unloadHooks is a good practice to prevent potential runtime errors, even though Go's range over nil slices is safe. It improves code robustness and readability.	7
	Add nil check for response body to prevent potential nil pointer dereference Consider handling the case where resp.Body might be nil in the error handling logic to avoid a potential nil pointer dereference. gateway/mw_streaming_test.go [110] -defer resp.Body.Close() +if resp.Body != nil { + defer resp.Body.Close() +} Suggestion importance[1-10]: 7 Why: This suggestion improves code robustness by adding a nil check for the response body, preventing potential runtime errors due to nil pointer dereference. It addresses a possible bug in the error handling logic.	7
Enhancement	Improve the reliability of stream start verification Instead of using a fixed delay to assume successful stream start, implement a more robust check or signal mechanism to confirm the stream has started without errors. internal/streaming/manager.go [102-104] case <-time.After(100 * time.Millisecond): - // If no error after a short delay, assume stream started successfully + s.log.Warn("Stream start not confirmed, proceeding with caution") Suggestion importance[1-10]: 7 Why: While the suggestion improves the reliability of stream start verification, it only adds a warning log without implementing a more robust mechanism, thus providing limited enhancement.	7
Performance	Use buffered channels to improve performance and prevent goroutine leaks Use buffered channels for better performance and to prevent potential goroutine leaks where the sender might block indefinitely if the receiver stops listening. gateway/mw_streaming_test.go [542] -done := make(chan bool) +done := make(chan bool, 1) Suggestion importance[1-10]: 6 Why: Using buffered channels can enhance performance and prevent goroutine leaks by ensuring that the sender does not block indefinitely if the receiver stops listening. This is a minor performance improvement.	6
Maintainability	Refactor error handling in the Start method to reduce code duplication Refactor the error handling in the Start method to avoid code duplication and improve readability by extracting the error handling into a separate method. internal/streaming/manager.go [56-82] -if err != nil { - s.log.Errorf("Failed to marshal config: %v", err) +if handleStartError(err, "marshal config") { return err } -if err != nil { - s.log.Errorf("Failed to remove consumer_group: %v", err) +if handleStartError(err, "remove consumer_group") { return err } -if err != nil { - s.log.Errorf("Failed to set YAML: %v", err) +if handleStartError(err, "set YAML") { return err } -if err != nil { - s.log.Errorf("Failed to build stream: %v", err) +if handleStartError(err, "build stream") { return err } +func (s *Stream) handleStartError(err error, action string) bool { + if err != nil { + s.log.Errorf("Failed to %s: %v", action, err) + return true + } + return false +} + Suggestion importance[1-10]: 6 Why: This suggestion enhances code maintainability by reducing duplication and improving readability, although it does not address any critical issues.	6

[pvormste]

This comment should be changed

[buraksezer]

Why does MiddlewareContextVars.EnabledForSpec return always true?

[pvormste]

The idea for streaming is to have context vars always enabled. There won't be an option to disable it in UI

[buraksezer]

Do we need to protect the fields of StreamingMiddleware struct with a lock? Members of the struct have been modified by different methods. How can be sure about thread-safety? It's hard and error prone to follow the execution flow with eye. We can use the race detector but it might be useful with an integration test that crafted for this purpose.

[kofoworola]

Yeah agree, i will be changing it in the next Pr that comes with a refactor.

[sonarcloud]

Quality Gate failed

Failed conditions
0.0% Coverage on New Code (required ≥ 80%)

See analysis details on SonarCloud

[titpetric]

This should be a public symbol as it's part of the Config data model