[TT-13201] Streams Definition Validator (#6656)

### **User description** <details open> <summary><a href="https://tyktech.atlassian.net/browse/TT-13201" title="TT-13201" target="_blank">TT-13201</a></summary> <br /> <table> <tr> <th>Summary</th> <td>Streams definition validator</td> </tr> <tr> <th>Type</th> <td> <img alt="Story" src="https://tyktech.atlassian.net/rest/api/2/universal_avatar/view/type/issuetype/avatar/10315?size=medium" /> Story </td> </tr> <tr> <th>Status</th> <td>In Dev</td> </tr> <tr> <th>Points</th> <td>N/A</td> </tr> <tr> <th>Labels</th> <td>-</td> </tr> </table> </details>  --- PR for [TT-13201](https://tyktech.atlassian.net/browse/TT-13201) We decided to copy and modify the existing OAS validator. It shares the same API and adds extra functionality to validate Bento configurations. The Bento schema only includes what we support in our current Tyk Streams implementation. Additional properties are allowed. It's also possible to implement new validators for Bento that implements `bento.ConfigValidator` interface. Currently, we only have a default validator. This one works with `bento.DefaultValidator`: ```go var document = []byte(<OAS Document that includes Bento configuration in the streams section>) err := streams.ValidateOASObject(document, "3.0.3") ``` This is how we can inject a Bento validator: ```go var document = []byte(<OAS Document that includes Bento configuration in the streams section>) err := ValidateOASObjectWithBentoConfigValidator(document, "3.0.3", bento.AsyncAPIValidator) ``` Here is a sample for invalid configuration. `auto_replay_nacks` must be a boolean, not a string. ```json { "streams": { "test-kafka-stream": { "input": { "label": "", "kafka": { "addresses": [], "topics": [], "target_version": "2.1.0", "consumer_group": "", "checkpoint_limit": 1024, "auto_replay_nacks": "true" } } } } } ``` `streams.ValidateOASObject(document, "3.0.3")` call should return for this document: ``` test-kafka-stream: input.kafka.auto_replay_nacks: Invalid type. Expected: boolean, given: string ``` [TT-13201]: https://tyktech.atlassian.net/browse/TT-13201?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ ___ ### **PR Type** Enhancement, Tests ___ ### **Description** - Implemented a new Tyk Streams OAS validator with support for Bento configuration validation. - Added comprehensive tests for validating Tyk Streams OAS objects, including valid and invalid scenarios. - Defined and implemented a default Bento configuration validator and its associated schema. - Added test data for various OAS templates to support testing efforts. ___ ### **Changes walkthrough** 📝 <table><thead><tr><th></th><th align="left">Relevant files</th></tr></thead><tbody><tr><td><strong>Tests</strong></td><td><details><summary>4 files</summary><table> <tr> <td> <details> <summary><strong>validator_test.go</strong><dd><code>Add tests for Tyk Streams OAS validation</code>                                  </dd></summary> <hr> apidef/streams/validator_test.go <li>Added tests for validating Tyk Streams OAS objects.<br> <li> Included tests for valid and invalid OAS objects.<br> <li> Added tests for loading OAS schemas and default version handling.<br> </details> </td> <td><a href="https://github.com/TykTechnologies/tyk/pull/6656/files#diff-a2b6f11dd78fabd9764d82f19af456f7b2bb835951d49e5e79d3488240513d0a">+373/-0</a>  </td> </tr> <tr> <td> <details> <summary><strong>validator_test.go</strong><dd><code>Add tests for Bento configuration validation</code>                          </dd></summary> <hr> apidef/streams/bento/validator_test.go <li>Added tests for Bento configuration validation.<br> <li> Included tests for valid and invalid configurations.<br> <li> Tested additional properties handling in configurations.<br> </details> </td> <td><a href="https://github.com/TykTechnologies/tyk/pull/6656/files#diff-dd179da8d36aa9dd6be42e239dac7ca819e488b587b8387a62dbea2caec7802b">+88/-0</a>    </td> </tr> <tr> <td> <details> <summary><strong>non-empty-x-tyk-ext-oas-template.json</strong><dd><code>Add test data for non-empty x-tyk-streaming template</code>          </dd></summary> <hr> apidef/streams/testdata/non-empty-x-tyk-ext-oas-template.json <li>Added test data for non-empty x-tyk-streaming OAS template.<br> <li> Included basic structure for Tyk streaming configurations.<br> </details> </td> <td><a href="https://github.com/TykTechnologies/tyk/pull/6656/files#diff-fdb20b6618bae07cf883190b25bcc2baa58f91e23f03b651c039c24a152fbf9f">+19/-0</a>    </td> </tr> <tr> <td> <details> <summary><strong>empty-x-tyk-ext-oas-template.json</strong><dd><code>Add test data for empty x-tyk-streams template</code>                      </dd></summary> <hr> apidef/streams/testdata/empty-x-tyk-ext-oas-template.json <li>Added test data for empty x-tyk-streams OAS template.<br> <li> Provided minimal structure for testing purposes.<br> </details> </td> <td><a href="https://github.com/TykTechnologies/tyk/pull/6656/files#diff-da79102eb03d04785a7cd94e1099ec603e964324aae539221a4d381a6fcaa851">+9/-0</a>      </td> </tr> </table></details></td></tr><tr><td><strong>Enhancement</strong></td><td><details><summary>4 files</summary><table> <tr> <td> <details> <summary><strong>validator.go</strong><dd><code>Implement Tyk Streams OAS validation logic</code>                              </dd></summary> <hr> apidef/streams/validator.go <li>Implemented validation logic for Tyk Streams OAS objects.<br> <li> Integrated Bento configuration validation.<br> <li> Added functions to load and manage OAS schemas.<br> </details> </td> <td><a href="https://github.com/TykTechnologies/tyk/pull/6656/files#diff-351cfcacb241ec2c3a3172d8c17e1217d6e942b5962081e7f9fd1582e801ca7f">+273/-0</a>  </td> </tr> <tr> <td> <details> <summary><strong>validator.go</strong><dd><code>Implement Bento configuration validation</code>                                  </dd></summary> <hr> apidef/streams/bento/validator.go <li>Defined interface for Bento configuration validation.<br> <li> Implemented default Bento configuration validator.<br> <li> Added schema loading for Bento configurations.<br> </details> </td> <td><a href="https://github.com/TykTechnologies/tyk/pull/6656/files#diff-462103c9f2f33bbe3bd4a21e46b5614fa0a4bfc3c3774f6c5f2ef858ae3fbb3f">+108/-0</a>  </td> </tr> <tr> <td> <details> <summary><strong>root.go</strong><dd><code>Define structure for Tyk streaming configurations</code>                </dd></summary> <hr> apidef/streams/root.go <li>Defined structure for Tyk streaming configurations.<br> <li> Included fields for Info, Server, Streams, and Middleware.<br> </details> </td> <td><a href="https://github.com/TykTechnologies/tyk/pull/6656/files#diff-89a589a5376efd0b8efd515e10a8399f54f8c7adadc948439f6a322b67140d5c">+17/-0</a>    </td> </tr> <tr> <td> <details> <summary><strong>bento-v1.2.0-supported-schema.json</strong><dd><code>Add JSON schema for Bento v1.2.0 configurations</code>                    </dd></summary> <hr> apidef/streams/bento/schema/bento-v1.2.0-supported-schema.json <li>Added JSON schema for Bento v1.2.0.<br> <li> Defined input and output properties for Bento configurations.<br> <li> Included processor and scanner definitions.<br> </details> </td> <td><a href="https://github.com/TykTechnologies/tyk/pull/6656/files#diff-b5ac63f348b5940aa0841fc033958eea71acbd11c789f7ee86e50b5af1301789">+2607/-0</a></td> </tr> </table></details></td></tr><tr><td><strong>Additional files (token-limit)</strong></td><td><details><summary>2 files</summary><table> <tr> <td> <details> <summary><strong>x-tyk-streaming.json</strong><dd><code>...</code>                                                                                                            </dd></summary> <hr> apidef/streams/schema/x-tyk-streaming.json ... </details> </td> <td><a href="https://github.com/TykTechnologies/tyk/pull/6656/files#diff-2bfe87e941e1bf7da030056225cc6052606d7e4c13f0388d4cf28b6287886ed1">+2041/-0</a></td> </tr> <tr> <td> <details> <summary><strong>3.0.json</strong><dd><code>...</code>                                                                                                            </dd></summary> <hr> apidef/streams/schema/3.0.json ... </details> </td> <td><a href="https://github.com/TykTechnologies/tyk/pull/6656/files#diff-fba2e5d2c55ab85af743db26dbc0e51cbaa797c7e93913475eaec0dda758b95a">+1662/-0</a></td> </tr> </table></details></td></tr></tr></tbody></table> ___ > 💡 **PR-Agent usage**: Comment `/help "your question"` on any pull request to receive relevant information
TykTechnologies · Oct 31, 2024 · 4a14e3a · 4a14e3a
1 parent 3633678
commit 4a14e3a
Show file tree

Hide file tree

Showing 13 changed files with 7,231 additions and 8 deletions.
diff --git a/apidef/api_definitions.go b/apidef/api_definitions.go
@@ -128,14 +128,15 @@ const (
 
 var (
 	// Deprecated: Use ErrClassicAPIExpected instead.
-	ErrAPIMigrated                = errors.New("the supplied API definition is in Tyk classic format, please use OAS format for this API")
-	ErrClassicAPIExpected         = errors.New("this API endpoint only supports Tyk Classic APIs; please use the appropriate Tyk OAS API endpoint")
-	ErrAPINotMigrated             = errors.New("the supplied API definition is in OAS format, please use the Tyk classic format for this API")
-	ErrOASGetForOldAPI            = errors.New("the requested API definition is in Tyk classic format, please use old api endpoint")
-	ErrImportWithTykExtension     = errors.New("the import payload should not contain x-tyk-api-gateway")
-	ErrPayloadWithoutTykExtension = errors.New("the payload should contain x-tyk-api-gateway")
-	ErrAPINotFound                = errors.New("API not found")
-	ErrMissingAPIID               = errors.New("missing API ID")
+	ErrAPIMigrated                         = errors.New("the supplied API definition is in Tyk classic format, please use OAS format for this API")
+	ErrClassicAPIExpected                  = errors.New("this API endpoint only supports Tyk Classic APIs; please use the appropriate Tyk OAS API endpoint")
+	ErrAPINotMigrated                      = errors.New("the supplied API definition is in OAS format, please use the Tyk classic format for this API")
+	ErrOASGetForOldAPI                     = errors.New("the requested API definition is in Tyk classic format, please use old api endpoint")
+	ErrImportWithTykExtension              = errors.New("the import payload should not contain x-tyk-api-gateway")
+	ErrPayloadWithoutTykExtension          = errors.New("the payload should contain x-tyk-api-gateway")
+	ErrPayloadWithoutTykStreamingExtension = errors.New("the payload should contain x-tyk-streaming")
+	ErrAPINotFound                         = errors.New("API not found")
+	ErrMissingAPIID                        = errors.New("missing API ID")
 )
 
 type EndpointMethodMeta struct {

diff --git a/apidef/oas/oas.go b/apidef/oas/oas.go
@@ -17,6 +17,9 @@ const (
 	// ExtensionTykAPIGateway is the OAS schema key for the Tyk extension.
 	ExtensionTykAPIGateway = "x-tyk-api-gateway"
 
+	// ExtensionTykStreaming is the OAS schema key for the Tyk Streams extension.
+	ExtensionTykStreaming = "x-tyk-streaming"
+
 	// Main holds the default version value (empty).
 	Main = ""
 
@@ -107,6 +110,51 @@ func (s *OAS) ExtractTo(api *apidef.APIDefinition) {
 	api.VersionData.Versions[Main] = vInfo
 }
 
+func (s *OAS) SetTykStreamingExtension(xTykStreaming *XTykStreaming) {
+	if s.Extensions == nil {
+		s.Extensions = make(map[string]interface{})
+	}
+
+	s.Extensions[ExtensionTykStreaming] = xTykStreaming
+}
+
+func (s *OAS) GetTykStreamingExtension() *XTykStreaming {
+	if s.Extensions == nil {
+		return nil
+	}
+
+	if ext := s.Extensions[ExtensionTykStreaming]; ext != nil {
+		rawTykStreaming, ok := ext.(json.RawMessage)
+		if ok {
+			var xTykStreaming XTykStreaming
+			_ = json.Unmarshal(rawTykStreaming, &xTykStreaming)
+			s.Extensions[ExtensionTykStreaming] = &xTykStreaming
+			return &xTykStreaming
+		}
+
+		mapTykAPIGateway, ok := ext.(map[string]interface{})
+		if ok {
+			var xTykStreaming XTykStreaming
+			dbByte, _ := json.Marshal(mapTykAPIGateway)
+			_ = json.Unmarshal(dbByte, &xTykStreaming)
+			s.Extensions[ExtensionTykStreaming] = &xTykStreaming
+			return &xTykStreaming
+		}
+
+		return ext.(*XTykStreaming)
+	}
+
+	return nil
+}
+
+func (s *OAS) RemoveTykStreamingExtension() {
+	if s.Extensions == nil {
+		return
+	}
+
+	delete(s.Extensions, ExtensionTykStreaming)
+}
+
 // SetTykExtension populates our OAS schema extension inside *OAS.
 func (s *OAS) SetTykExtension(xTykAPIGateway *XTykAPIGateway) {
 	if s.Extensions == nil {

diff --git a/apidef/oas/tyk_streaming_extension.go b/apidef/oas/tyk_streaming_extension.go
@@ -0,0 +1,7 @@
+package oas
+
+// XTykStreaming represents the structure for Tyk streaming configurations.
+type XTykStreaming struct {
+	// Streams contains the configurations related to Tyk Streams.
+	Streams map[string]interface{} `bson:"streams" json:"streams"` // required
+}