TT-13269 - Refactor/streams #6593
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
API Changes --- prev.txt 2024-10-17 15:26:12.273044467 +0000
+++ current.txt 2024-10-17 15:26:06.309080445 +0000
@@ -897,11 +897,17 @@
},
"scopes":{
"type": ["array", "null"]
+ },
+ "endpoint_params": {
+ "type": ["object", "null"]
}
}
},
"header_name": {
"type": "string"
+ },
+ "distributed_token": {
+ "type": "boolean"
}
}
}
@@ -1207,24 +1213,17 @@
DisableHalfOpenState bool `bson:"disable_half_open_state" json:"disable_half_open_state"`
}
-type ClientAuthData struct {
+type ClientCredentials struct {
// ClientID is the application's ID.
ClientID string `bson:"client_id" json:"client_id"`
// ClientSecret is the application's secret.
ClientSecret string `bson:"client_secret" json:"client_secret"`
-}
- ClientAuthData holds the client ID and secret for upstream OAuth2
- authentication.
-
-type ClientCredentials struct {
- ClientAuthData
// TokenURL is the resource server's token endpoint
// URL. This is a constant specific to each server.
TokenURL string `bson:"token_url" json:"token_url"`
// Scopes specifies optional requested permissions.
Scopes []string `bson:"scopes" json:"scopes,omitempty"`
- // TokenProvider is the OAuth2 token provider for internal use.
TokenProvider oauth2.TokenSource `bson:"-" json:"-"`
}
ClientCredentials holds the client credentials for upstream OAuth2
@@ -7664,6 +7663,136 @@
func (ms *MockStorage) Set(key string, addrs []string)
+# Package: ./ee/middleware/streams
+
+package streams // import "github.com/TykTechnologies/tyk/ee/middleware/streams"
+
+
+CONSTANTS
+
+const (
+ // ExtensionTykStreaming is the OAS extension for Tyk streaming.
+ ExtensionTykStreaming = "x-tyk-streaming"
+ StreamGCInterval = 1 * time.Minute
+)
+
+VARIABLES
+
+var GlobalStreamCounter atomic.Int64
+ GlobalStreamCounter is used for testing.
+
+
+FUNCTIONS
+
+func GetHTTPPaths(streamConfig map[string]interface{}) []string
+ GetHTTPPaths is the main function to get HTTP paths from the stream
+ configuration.
+
+
+TYPES
+
+type APISpec struct {
+ APIID string
+ Name string
+ IsOAS bool
+ OAS oas.OAS
+
+ StripListenPath model.StripListenPathFunc
+}
+ APISpec is a subset of gateway.APISpec for the values the middleware
+ consumes.
+
+func NewAPISpec(id string, name string, isOasDef bool, oasDef oas.OAS, stripListenPath model.StripListenPathFunc) *APISpec
+ NewAPISpec creates a new APISpec object based on the required inputs.
+ The resulting object is a subset of `*gateway.APISpec`.
+
+type BaseMiddleware interface {
+ model.LoggerProvider
+}
+ BaseMiddleware is the subset of BaseMiddleware APIs that the middleware
+ uses.
+
+type Gateway interface {
+ model.ConfigProvider
+ model.ReplaceTykVariables
+}
+ Gateway is the subset of Gateway APIs that the middleware uses.
+
+type Manager struct {
+ // Has unexported fields.
+}
+ Manager is responsible for creating a single stream.
+
+type Middleware struct {
+ Spec *APISpec
+ Gw Gateway
+
+ StreamManagerCache sync.Map // Map of payload hash to Manager
+
+ // Has unexported fields.
+}
+ Middleware implements a streaming middleware.
+
+func NewMiddleware(gw Gateway, mw BaseMiddleware, spec *APISpec) *Middleware
+ NewMiddleware returns a new instance of Middleware.
+
+func (s *Middleware) CreateStreamManager(r *http.Request) *Manager
+ CreateStreamManager creates or retrieves a stream manager based on the
+ request.
+
+func (s *Middleware) EnabledForSpec() bool
+ EnabledForSpec checks if streaming is enabled on the config.
+
+func (s *Middleware) GC()
+ GC removes inactive stream managers.
+
+func (s *Middleware) Init()
+ Init initializes the middleware
+
+func (s *Middleware) Logger() *logrus.Entry
+ Logger returns a logger with middleware filled out.
+
+func (s *Middleware) Name() string
+ Name returns the name for the middleware.
+
+func (s *Middleware) ProcessRequest(w http.ResponseWriter, r *http.Request, _ interface{}) (error, int)
+ ProcessRequest will handle the streaming functionality.
+
+func (s *Middleware) Unload()
+ Unload closes and remove active streams.
+
+type Stream struct {
+ // Has unexported fields.
+}
+ Stream is a wrapper around stream
+
+func NewStream(allowUnsafe []string) *Stream
+ NewStream creates a new stream without initializing it
+
+func (s *Stream) GetConfig() string
+ GetConfig returns the configuration of the stream
+
+func (s *Stream) Reset() error
+ Reset stops the stream
+
+func (s *Stream) SetLogger(logger *logrus.Logger)
+ SetLogger to be used by the stream
+
+func (s *Stream) Start(config map[string]interface{}, mux service.HTTPMultiplexer) error
+ Start loads up the configuration and starts the stream. Non blocking
+
+func (s *Stream) Stop() error
+ Stop cleans up the stream
+
+type StreamsConfig struct {
+ Info struct {
+ Version string `json:"version"`
+ } `json:"info"`
+
+ Streams map[string]any `json:"streams"`
+}
+ StreamsConfig represents a stream configuration.
+
# Package: ./gateway
package gateway // import "github.com/TykTechnologies/tyk/gateway"
@@ -7693,6 +7822,8 @@
swagger:meta
+Provides getStreamingMiddleware
+
CONSTANTS
const (
@@ -7809,11 +7940,6 @@
ErrOAuthClientDeleted = "oauth.client_deleted"
)
const (
- // ExtensionTykStreaming is the oas extension for tyk streaming
- ExtensionTykStreaming = "x-tyk-streaming"
- StreamGCInterval = 1 * time.Minute
-)
-const (
ResetQuota string = "resetQuota"
CertificateRemoved string = "CertificateRemoved"
CertificateAdded string = "CertificateAdded"
@@ -7965,10 +8091,6 @@
func GenerateTestBinaryData() (buf *bytes.Buffer)
func GetAccessDefinitionByAPIIDOrSession(session *user.SessionState, api *APISpec) (accessDef *user.AccessDefinition, allowanceScope string, err error)
-func GetHTTPPaths(streamConfig map[string]interface{}) []string
- GetHTTPPaths is the ain function to get HTTP paths from the stream
- configuration
-
func GetTLSClient(cert *tls.Certificate, caCert []byte) *http.Client
func GetTLSConfig(cert *tls.Certificate, caCert []byte) *tls.Config
func InitTestMain(ctx context.Context, m *testing.M) int
@@ -8406,8 +8528,6 @@
Skip bool
}
-type ClientCredentialsOAuthProvider struct{}
-
type CoProcessEventHandler struct {
Spec *APISpec
SpecJSON json.RawMessage
@@ -8583,6 +8703,8 @@
resetTTLTo int64, hashed bool) error
UpdateSession updates the session state in the storage engine
+type DistributedCacheOAuthProvider struct{}
+
type DummyProxyHandler struct {
SH SuccessHandler
Gw *Gateway `json:"-"`
@@ -8780,7 +8902,7 @@
HostCheckerClient *http.Client
TracerProvider otel.TracerProvider
// UpstreamOAuthCache is used to cache upstream OAuth tokens
- UpstreamOAuthCache UpstreamOAuthCache
+ UpstreamOAuthCache *upstreamOAuthCache
SessionLimiter SessionLimiter
SessionMonitor Monitor
@@ -8904,6 +9026,12 @@
func (gw *Gateway) ProcessSingleOauthClientEvent(apiId, oauthClientId, orgID, event string)
+func (gw *Gateway) ReplaceTykVariables(r *http.Request, in string, escape bool) string
+ ReplaceTykVariables implements a variable replacement hook. It will replace
+ the template `in`. If `escape` is true, the values get escaped as a query
+ parameter for a HTTP request would. If no replacement has been made, `in` is
+ returned without modification.
+
func (gw *Gateway) RevokeAllTokensHandler(w http.ResponseWriter, r *http.Request)
func (gw *Gateway) RevokeTokenHandler(w http.ResponseWriter, r *http.Request)
@@ -9775,7 +9903,7 @@
func (k *OrganizationMonitor) SetOrgSentinel(orgChan chan bool, orgId string)
-type PerAPIClientCredentialsOAuthProvider struct{}
+type PerAPIOAuthProvider struct{}
type PersistGraphQLOperationMiddleware struct {
*BaseMiddleware
@@ -10204,11 +10332,6 @@
TickOk triggers a reload and ensures a queue happened and a reload cycle
happens. This will block until all the cases are met.
-type Repository interface {
- policy.Repository
-}
- Repository is a description of our Gateway API promises.
-
type RequestDefinition struct {
Method string `json:"method"`
Headers map[string]string `json:"headers"`
@@ -10571,41 +10694,6 @@
type StatsDSinkSanitizationFunc func(*bytes.Buffer, string)
-type StreamManager struct {
- // Has unexported fields.
-}
- StreamManager is responsible for creating a single stream
-
-type StreamingMiddleware struct {
- *BaseMiddleware
-
- // Has unexported fields.
-}
- StreamingMiddleware is a middleware that handles streaming functionality
-
-func (s *StreamingMiddleware) EnabledForSpec() bool
- EnabledForSpec checks if streaming is enabled on the config
-
-func (s *StreamingMiddleware) Init()
- Init initializes the middleware
-
-func (s *StreamingMiddleware) Name() string
- Name is StreamingMiddleware
-
-func (s *StreamingMiddleware) ProcessRequest(w http.ResponseWriter, r *http.Request, _ interface{}) (error, int)
- ProcessRequest will handle the streaming functionality
-
-func (s *StreamingMiddleware) Unload()
- Unload closes and remove active streams
-
-type StreamsConfig struct {
- Info struct {
- Version string `json:"version"`
- } `json:"info"`
- Streams map[string]any `json:"streams"`
-}
- StreamsConfig represents a stream configuration
-
type StripAuth struct {
*BaseMiddleware
}
@@ -10826,9 +10914,10 @@
}
type TykMiddleware interface {
- Init()
Base() *BaseMiddleware
+ GetSpec() *APISpec
+ Init()
SetName(string)
SetRequestLogger(*http.Request)
Logger() *logrus.Entry
@@ -10837,11 +10926,14 @@
EnabledForSpec() bool
Name() string
- GetSpec() *APISpec
-
Unload()
}
+func WrapMiddleware(base *BaseMiddleware, in model.Middleware) TykMiddleware
+ WrapMiddleware returns a new TykMiddleware with the provided base
+ middleware, and the smaller model.Middleware interface. It allows to
+ implement model.Middleware, and use it as a TykMiddleware.
+
type TykOsinServer struct {
osin.Server
Config *osin.ServerConfig
@@ -11001,10 +11093,6 @@
ProcessRequest will inject basic auth info into request context so that it
can be used during reverse proxy.
-type UpstreamOAuthCache interface {
- // Has unexported methods.
-}
-
type UpstreamOAuthProvider struct {
// HeaderName is the header name to be used to fill upstream auth with.
HeaderName string |
PR Reviewer Guide 🔍Here are some key observations to aid the review process:
|
PR Code Suggestions ✨No code suggestions found for the PR. |
added 9 commits
October 9, 2024 16:21
titpetric
force-pushed
the
refactor/streams
branch
from
6b1f5e8 to
20045ab
Compare
titpetric
force-pushed
the
refactor/streams
branch
from
0591aed to
9f5341b
Compare
|
@titpetric as discussed it should live in "ee" folder, e.g. "ee/internal/middleware" |
|
kofoworola
approved these changes
Oct 18, 2024
pvormste
approved these changes
Oct 18, 2024
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
User description
Extracts streaming middleware into internal/middleware/stream;
Implements a number of interfaces to cover gateway/base middleware
Implements
wrappedMiddlewareto allow middlewares with less coupling.I tried to update couplings along as I went, it's passing
go buildandgo test -c.PR Type
enhancement, tests
Description
Managerfor handling streams and updated theStreamingMiddlewareto use this new structure.apideftypes withmodeltypes across multiple files to improve modularity and reduce coupling.WrapMiddlewarefunctionality to allow middleware wrapping with less coupling.modelpackage.modelpackage to define common behaviors for Gateway and Middleware components.ReplaceTykVariablesfunction for consistency.Changes walkthrough 📝
22 files
api_loader.go
Integrate new streaming middleware and wrap functionalitygateway/api_loader.go
internal/middleware/stream.StreamingMiddlewarewith a new streaming middlewareimplementation.
WrapMiddlewarefor middleware wrapping.gateway.go
Update Gateway interface implementationgateway/gateway.go
model.Gateway.health_check.go
Use model package for health check itemsgateway/health_check.go
apidef.HealthCheckItemwithmodel.HealthCheckItem.middleware.go
Update TykMiddleware interface and constantsgateway/middleware.go
GetSpecmethod toTykMiddlewareinterface.mwStatusRespondto usemiddleware.StatusRespond.middleware_wrap.go
Introduce middleware wrapping functionalitygateway/middleware_wrap.go
wrapMiddlewarestruct for middleware wrapping.WrapMiddlewarefunction.mw_auth_key.go
Update variable replacement method callgateway/mw_auth_key.go
ReplaceTykVariables.mw_graphql.go
Update variable replacement method callgateway/mw_graphql.go
ReplaceTykVariables.mw_modify_headers.go
Update variable replacement method callgateway/mw_modify_headers.go
ReplaceTykVariables.mw_persist_graphql_operation.go
Update variable replacement method callgateway/mw_persist_graphql_operation.go
ReplaceTykVariables.mw_rate_limiting.go
Update variable replacement method callgateway/mw_rate_limiting.go
ReplaceTykVariables.mw_transform.go
Update variable replacement method callgateway/mw_transform.go
ReplaceTykVariables.mw_url_rewrite.go
Rename and document ReplaceTykVariables functiongateway/mw_url_rewrite.go
replaceTykVariablestoReplaceTykVariablesand addeddocumentation.
res_handler_header_injector.go
Update variable replacement method callgateway/res_handler_header_injector.go
ReplaceTykVariables.rpc_storage_handler.go
Use model package for RPC storage handler typesgateway/rpc_storage_handler.go
apideftypes withmodeltypes for RPC storage handling.const.go
Add StatusRespond constant for middlewareinternal/middleware/const.go
StatusRespondfor middleware processing.stream_manager.go
Introduce Manager for stream managementinternal/middleware/stream/stream_manager.go
Managerstruct for stream management.streaming_middleware.go
Refactor StreamingMiddleware with new Managerinternal/middleware/stream/streaming_middleware.go
StreamingMiddlewareto use newManager.health_check.go
Rename package to modelinternal/model/health_check.go
apideftomodel.interfaces.go
Add interfaces for Gateway and Middlewareinternal/model/interfaces.go
rpc.go
Move RPC types to model packageinternal/model/rpc.go
modelpackage.apply.go
Use model.PolicyProvider in Service structinternal/policy/apply.go
Servicestruct to usemodel.PolicyProvider.synchronization_forcer.go
Use model package for GroupLoginRequestrpc/synchronization_forcer.go
apidef.GroupLoginRequestwithmodel.GroupLoginRequest.3 files
mw_streaming_test.go
Update streaming test cases to use new packagegateway/mw_streaming_test.go
streampackage for streamingfunctionalities.
rpc_storage_handler_test.go
Update RPC storage handler tests to use model packagegateway/rpc_storage_handler_test.go
modelpackage for RPC storage handler.rpc_test.go
Update RPC tests to use model packagegateway/rpc_test.go
modelpackage for API definitions.