[WIP] Streaming POC #6291

buger · 2024-05-18T17:52:52Z

User description

Based on #6288

The main changes that Streaming configuration is now part of OAS API definition "x-tyk-streaming", and can look like this:

{
  "info": {
    "title": "Petstore",
    "version": "1.0.0"
  },
  "openapi": "3.0.3",
  "components": {},
  "paths": {},
  "x-tyk-streaming": {
      "streams": {
          "test": {
              "input": {
                   "generate": {
                        "count": 3,
                        "interval": "",
                        "mapping": "root = \"hello world\""
                    }
               },
               "output": {
                    "discard": {}
                }
           }
       }
   },
  "x-tyk-api-gateway": {
    "info": {
      "name": "Petstore",
      "state": {
        "active": true
      }
    },
    "upstream": {
      "url": "https://petstore.swagger.io/v2"
    },
    "server": {
      "listenPath": {
        "value": "/petstore/",
        "strip": true
      }
    }
  }
}

PR Type

Enhancement, Tests, Documentation

Description

Introduced StreamingMiddleware for handling streaming functionality with WebSocket and SSE support.
Implemented StreamManager for managing stream configurations and Redis integration.
Added PortalClient and portalOutput for interacting with the developer portal and posting messages to webhooks.
Integrated StreamingMiddleware into the API loading process and added unload hooks to APISpec.
Added comprehensive tests for new streaming and portal functionalities.
Updated dependencies and configuration to support new features.

Changes walkthrough 📝

Relevant files

Enhancement

8 files

mw_streaming.go `Introduce Streaming Middleware with WebSocket and SSE Support` gateway/mw_streaming.go Added new `StreamingMiddleware` for handling streaming functionality. Implemented WebSocket and SSE handling within the middleware. Integrated Redis for stream management.	+488/-0
manager.go `Implement StreamManager for Stream Configuration and Management` internal/streaming/manager.go Created `StreamManager` for managing stream configurations. Added methods for adding, removing, and subscribing to streams. Integrated Redis for stream message handling.	+464/-0
portal_client.go `Implement PortalClient for Developer Portal Interaction` internal/portal/portal_client.go Created `PortalClient` for interacting with the developer portal. Added methods for listing webhook credentials.	+152/-0
portal_output.go `Add PortalOutput for Posting Messages to Webhooks` internal/portal/portal_output.go Added `portalOutput` for posting messages to webhooks. Integrated with `PortalClient` to fetch webhook credentials.	+128/-0
api_loader.go `Integrate StreamingMiddleware into API Loading Process` gateway/api_loader.go Integrated `StreamingMiddleware` into the API loading process. Added logic to unload specs when they change.	+9/-5
middleware.go `Enhance TykMiddleware Interface with Spec Retrieval and Unloading` gateway/middleware.go Added methods to `TykMiddleware` interface for spec retrieval and unloading. Updated middleware initialization to include unload hooks.	+15/-1
api_definition.go `Add Unload Hooks to APISpec` gateway/api_definition.go Added unload hooks to `APISpec`. Updated `Release` method to call unload hooks.	+12/-1
mw_context_vars.go `Modify EnabledForSpec Method in MiddlewareContextVars` gateway/mw_context_vars.go Modified `EnabledForSpec` method to always return true.	+1/-1

Tests

5 files

mw_streaming_test.go `Add Tests for StreamingMiddleware WebSocket and SSE Handling` gateway/mw_streaming_test.go Added tests for `StreamingMiddleware` including WebSocket and SSE handling. Implemented test cases for stream generation and message handling.	+327/-0
manager_test.go `Add Unit Tests for StreamManager Functionality` internal/streaming/manager_test.go Added unit tests for `StreamManager`. Tested stream addition, removal, and HTTP path retrieval.	+162/-0
portal_test.go `Add Tests for PortalClient and Webhook Credential Listing` internal/portal/portal_test.go Added tests for `PortalClient` and webhook credential listing. Mocked server responses for testing.	+76/-0
portal_output_test.go `Add Unit Tests for PortalOutput Plugin` internal/portal/portal_output_test.go Added unit tests for `portalOutput` plugin. Mocked server to test webhook calls.	+74/-0
testutil.go `Update Test Utility for API Spec Unloading` gateway/testutil.go Updated test utility to unload API specs during cleanup.	+6/-0

Configuration changes

1 files

config.go `Add StreamingConfig to Global Configuration` config/config.go Added `StreamingConfig` to global configuration. Enabled streaming feature flag.	+7/-0

Dependencies

1 files

go.mod `Update Dependencies for Streaming and Portal Integration` go.mod Updated dependencies including `benthos` and `go-redis`. Added new dependencies for streaming and portal integration.	+289/-35

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

Based on #6288

github-actions · 2024-05-18T17:53:46Z

PR Description updated to latest commit (d2ca309)

github-actions · 2024-05-18T17:54:06Z

API Changes

--- prev.txt	2024-06-30 18:43:35.438528688 +0000
+++ current.txt	2024-06-30 18:43:32.546505401 +0000
@@ -5591,6 +5591,8 @@
 
 	// OAS holds the configuration for various OpenAPI-specific functionalities
 	OAS OASConfig `json:"oas_config"`
+
+	Labs labsConfig `json:"labs"`
 }
     Config is the configuration object used by Tyk to set up various parameters.
 
@@ -7425,6 +7427,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 MaxBackoffRetries = 4
@@ -7692,6 +7697,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
 
@@ -7707,9 +7715,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
@@ -7723,6 +7728,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.
@@ -7815,6 +7823,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)
@@ -7829,6 +7839,8 @@
 
 func (t *BaseMiddleware) SetRequestLogger(r *http.Request)
 
+func (t *BaseMiddleware) Unload()
+
 func (t *BaseMiddleware) UpdateRequestSession(r *http.Request) bool
 
 type BaseTykResponseHandler struct {
@@ -10057,6 +10069,25 @@
 
 type StatsDSinkSanitizationFunc func(*bytes.Buffer, string)
 
+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
 }
@@ -10287,6 +10318,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 · 2024-05-18T17:54:09Z

PR Review 🔍

⏱️ Estimated effort to review [1-5]	4, because the PR introduces a significant amount of new functionality related to streaming, including integration with Benthos, handling of streaming configurations, and extensive modifications across multiple files. The complexity and breadth of the changes require careful review to ensure correctness, performance, and maintainability.
🧪 Relevant tests	Yes
⚡ Possible issues	Possible Bug: In `gateway/streaming_test.go`, the function `ConvertYAMLToJSON` uses a recursive approach to convert keys to strings, which might lead to a stack overflow with deeply nested YAML structures.
⚡ Possible issues	Error Handling: In `internal/streaming/benthos_server.go`, the function `Start` uses `panic` for error handling in multiple places. This approach can cause the application to terminate unexpectedly. It would be better to handle these errors gracefully and return them to the caller.
🔒 Security concerns	No

Code feedback:

relevant file	internal/streaming/benthos_server.go
suggestion	Replace `panic` with proper error handling in the `Start` method to prevent the application from terminating unexpectedly. This change enhances the robustness and reliability of the server initialization process. [important]
relevant line	panic(err) // Handle error appropriately for your use case

relevant file	gateway/streaming_test.go
suggestion	Consider adding a check for deep recursion or refactoring the method to handle deeply nested structures more efficiently to prevent potential stack overflow issues in `ConvertYAMLToJSON`. [important]
relevant line	func ConvertYAMLToJSON(yamlData []byte) ([]byte, error) {

relevant file	internal/streaming/benthos_server.go
suggestion	Use a logging mechanism instead of `fmt.Printf` for stream creation and update messages to provide better control over log management and to integrate with the system's centralized logging strategy. [medium]
relevant line	fmt.Printf("Stream created: %+v\n", result)

relevant file	internal/streaming/benthos_server.go
suggestion	Consider implementing a more secure method for handling passwords, such as using environment variables or secure vault solutions, instead of generating them within the application. This change would enhance the security posture by reducing potential exposure of sensitive data. [important]
relevant line	password := generateRandomPassword()

github-actions · 2024-05-18T17:54:55Z

PR Code Suggestions ✨

Category	Suggestion	Score
Best practice	Add a descriptive comment for the new constant to improve code readability Consider adding a comment above the new constant `ExtensionTykStreaming` to describe its purpose and usage within the system. This will maintain consistency with other constants in the file and improve code readability. apidef/oas/oas.go [20] +// ExtensionTykStreaming is the OAS schema key for the Tyk streaming extension. ExtensionTykStreaming = "x-tyk-streaming" Suggestion importance[1-10]: 10 Why: The suggestion correctly identifies the lack of a descriptive comment for the new constant `ExtensionTykStreaming` and proposes an appropriate comment, improving consistency and readability.	10
Best practice	Replace fmt.Printf with log.Printf for structured logging Use structured logging instead of `fmt.Printf` for better log management and to enable structured log querying capabilities. internal/streaming/benthos_server.go [134] -fmt.Printf("Stream created: %+v\n", result) +log.Printf("Stream created: %+v", result) Suggestion importance[1-10]: 6 Why: Using structured logging like `log.Printf` instead of `fmt.Printf` improves logging consistency and is beneficial for log analysis, but it's a moderate improvement in the context of the entire application.	6
Maintainability	Refactor nested conditional logic into separate methods to improve readability and maintainability Refactor the nested conditional blocks inside the loop for checking and adding streams to reduce complexity and improve readability. Consider extracting this logic into a separate method. gateway/api_loader.go [1005-1021] if gw.GetConfig().Streaming.Enabled && spec.IsOAS { + gw.processStreamingAPI(spec) +} + +// New method in Gateway struct +func (gw Gateway) processStreamingAPI(spec APISpec) { if ext, ok := spec.OAS.T.Extensions[oas.ExtensionTykStreaming]; ok { - if streamsMap, ok := ext.(map[string]interface{}); ok { - if streams, ok := streamsMap["streams"].(map[string]interface{}); ok { - for streamID, stream := range streams { - if streamMap, ok := stream.(map[string]interface{}); ok { - if err := gw.StreamingServer.AddStream(spec.APIID+"_"+streamID, streamMap); err != nil { - mainLog.Errorf("Error adding stream to streaming server: %v", err) - } else { - activeStreams[spec.APIID+"_"+streamID] = struct{}{} - mainLog.Infof("Added stream %s to streaming server", spec.APIID+"_"+streamID) - } - } - } + gw.handleStreamExtensions(ext, spec) + } +} + +func (gw Gateway) handleStreamExtensions(ext interface{}, spec APISpec) { + if streamsMap, ok := ext.(map[string]interface{}); ok { + if streams, ok := streamsMap["streams"].(map[string]interface{}); ok { + gw.addStreamsToServer(streams, spec) + } + } +} + +func (gw Gateway) addStreamsToServer(streams map[string]interface{}, spec APISpec) { + for streamID, stream := range streams { + if streamMap, ok := stream.(map[string]interface{}); ok { + streamFullName := spec.APIID + "_" + streamID + if err := gw.StreamingServer.AddStream(streamFullName, streamMap); err != nil { + mainLog.Errorf("Error adding stream to streaming server: %v", err) + } else { + activeStreams[streamFullName] = struct{}{} + mainLog.Infof("Added stream %s to streaming server", streamFullName) } } } } Suggestion importance[1-10]: 9 Why: The suggestion effectively addresses the complexity of nested conditionals by proposing a refactor into separate methods, significantly enhancing code readability and maintainability.	9
Maintainability	Use constants for error messages in tests to reduce duplication and ease maintenance Replace the hardcoded error message checks in tests with constants or variables to reduce duplication and facilitate easier updates to test conditions. gateway/streaming_test.go [69-172] -t.Fatalf("Failed to create temporary file: %v", err) -t.Fatalf("Failed to convert YAML to JSON: %v", err) -t.Fatalf("Failed to parse JSON: %v", err) -t.Fatalf("Failed to get streams: %v", err) -t.Fatalf("Failed to read file: %v", err) -t.Fatalf("Expected 'hello world', got '%s'", line) +const ( + errCreateTempFile = "Failed to create temporary file: %v" + errConvertYAMLToJSON = "Failed to convert YAML to JSON: %v" + errParseJSON = "Failed to parse JSON: %v" + errGetStreams = "Failed to get streams: %v" + errReadFile = "Failed to read file: %v" + errExpectedOutput = "Expected 'hello world', got '%s'" +) +t.Fatalf(errCreateTempFile, err) +t.Fatalf(errConvertYAMLToJSON, err) +t.Fatalf(errParseJSON, err) +t.Fatalf(errGetStreams, err) +t.Fatalf(errReadFile, err) +t.Fatalf(errExpectedOutput, line) + Suggestion importance[1-10]: 7 Why: Using constants for repeated error messages in tests is a good practice for maintainability and ease of updates, correctly identified by the suggestion.	7
Possible issue	Replace panic with error handling in the server start method Replace the panic handling in the `Start` method with proper error handling to avoid abrupt termination of the program. This will improve the robustness of the server operation by allowing it to handle errors gracefully. internal/streaming/benthos_server.go [55] -panic(err) // Handle error appropriately for your use case +return fmt.Errorf("failed to listen on TCP port: %w", err) Suggestion importance[1-10]: 9 Why: Replacing panic with proper error handling in critical server operations is crucial for robustness and prevents the server from crashing unexpectedly, which is a significant improvement.	9
Possible bug	Add error handling for uninitialized streaming server to prevent nil pointer dereference Add error handling for the scenario where `gw.StreamingServer` is nil to prevent potential nil pointer dereference when the streaming feature is enabled but the server fails to initialize. gateway/server.go [1906-1911] if gw.GetConfig().Streaming.Enabled { gw.StreamingServer = streaming.New() + if gw.StreamingServer == nil { + mainLog.Error("Failed to initialize streaming server") + return + } err := gw.StreamingServer.Start() if err != nil { mainLog.WithError(err).Error("Error starting streaming server") } } Suggestion importance[1-10]: 8 Why: This suggestion correctly identifies a potential bug where `gw.StreamingServer` could be nil, leading to a nil pointer dereference. The proposed error handling is crucial for robustness.	8
Enhancement	Implement exponential backoff for retries in the `WaitForReady` method Consider implementing a retry mechanism with exponential backoff for the `WaitForReady` method to handle transient errors more effectively. internal/streaming/benthos_server.go [104] -time.Sleep(500 * time.Millisecond) +time.Sleep(time.Duration(math.Pow(2, float64(t))) * time.Second) Suggestion importance[1-10]: 7 Why: Implementing exponential backoff is a good practice for handling retries, especially in network operations where transient errors can occur. This would enhance the reliability of the `WaitForReady` method.	7
Enhancement	Provide meaningful defaults for configuration parameters Avoid using empty strings for configuration defaults in the server configuration template. Provide meaningful defaults or document the need for these fields to be populated. internal/streaming/benthos_server.go [74] -salt: "" +salt: "default_salt_value" # Replace with a meaningful default or ensure it's set by the user Suggestion importance[1-10]: 5 Why: Providing meaningful defaults or documentation for configuration parameters can prevent potential misconfigurations, but this is a relatively minor enhancement compared to other possible improvements.	5

github-actions · 2024-05-18T17:55:10Z

💥 CI tests failed 🙈