Openapi import (#1849)

* Initial import from internal repo (code mainly by https://github.com/krzpiesiewicz)

* Dispatch => HttpClient

* SharedService, remove obsolete code

* Parser package private

* Refactor HttpConfig + separare Flink and Standalone

* Tests, example

* Cleanup, docs, tests

* Review

build.sbt

@@ -212,6 +212,10 @@ lazy val commonSettings =
         //Our main kafka dependencies are Confluent (for avro) and Flink (Kafka connector)
         "org.apache.kafka" % "kafka-clients" % kafkaV,
         "org.apache.kafka" %% "kafka" % kafkaV,
+
+        "io.netty" % "netty-handler" % nettyV,
+        "io.netty" % "netty-codec" % nettyV,
+        "io.netty" % "netty-transport-native-epoll" % nettyV,
       )
     )
 
@@ -246,6 +250,7 @@ val commonsIOV = "2.4"
 val dropWizardV = "5.0.0-rc3"
 val scalaCollectionsCompatV = "2.3.2"
 val testcontainersScalaV = "0.39.3"
+val nettyV = "4.1.48.Final"
 
 val akkaHttpV = "10.1.8"
 val akkaHttpCirceV = "1.27.0"
@@ -378,6 +383,9 @@ lazy val dist = {
 
 def engine(name: String) = file(s"engine/$name")
 
+def component(name: String) = file(s"engine/components/$name")
+
+
 lazy val engineStandalone = (project in engine("standalone/engine")).
   configs(IntegrationTest).
   settings(commonSettings).
@@ -545,6 +553,7 @@ lazy val generic = (project in engine("flink/generic")).
       )
     })
   .dependsOn(process % "runtime,test", avroFlinkUtil, flinkModelUtil, flinkTestUtil % "test", kafkaTestUtil % "test",
+    openapi,
     //for local development
     ui % "test")
 
@@ -900,6 +909,35 @@ lazy val queryableState = (project in engine("queryableState")).
   ).dependsOn(api)
 
 
+val swaggerParserV = "2.0.20"
+val swaggerIntegrationV = "2.1.3"
+
+lazy val openapi = (project in component("openapi")).
+    configs(IntegrationTest).
+    settings(commonSettings).
+    settings(Defaults.itSettings).
+    settings(
+      name := "nussknacker-openapi",
+      libraryDependencies ++= Seq(
+        "io.swagger.parser.v3" % "swagger-parser" % swaggerParserV excludeAll(
+          ExclusionRule(organization = "javax.mail"),
+          ExclusionRule(organization = "javax.validation"),
+          ExclusionRule(organization = "jakarta.activation"),
+          ExclusionRule(organization = "jakarta.validation")
+        ),
+        "io.swagger.core.v3" % "swagger-integration" % swaggerIntegrationV  excludeAll(
+          ExclusionRule(organization = "jakarta.activation"),
+          ExclusionRule(organization = "jakarta.validation")
+        ),
+        "com.softwaremill.sttp.client" %% "circe" % sttpV excludeAll ExclusionRule(organization = "io.circe"),
+        "com.softwaremill.sttp.client" %% "async-http-client-backend-future" % sttpV  excludeAll(
+          ExclusionRule(organization = "com.sun.activation", name = "javax.activation"),
+        ),
+        "io.netty" % "netty-transport-native-epoll" % nettyV,
+        "org.apache.flink" %% "flink-streaming-scala" % flinkV % "provided,optional",
+        "org.scalatest" %% "scalatest" % scalaTestV %  "it,test"
+      ),
+    ).dependsOn(api, process % "provided,optional", engineStandalone % "provided,optional", standaloneUtil % "provided,optional", httpUtils % Provided, flinkTestUtil % "it,test", kafkaTestUtil % "it,test")
 
 lazy val buildUi = taskKey[Unit]("builds ui")
 
@@ -1042,7 +1080,7 @@ lazy val bom = (project in file("bom"))
 
 lazy val modules = List[ProjectReference](
   engineStandalone, standaloneApp, flinkProcessManager, flinkPeriodicProcessManager, standaloneSample, flinkManagementSample, managementJavaSample, generic,
-  process, interpreter, benchmarks, kafka, avroFlinkUtil, kafkaFlinkUtil, kafkaTestUtil, util, testUtil, flinkUtil, flinkModelUtil,
+  openapi, process, interpreter, benchmarks, kafka, avroFlinkUtil, kafkaFlinkUtil, kafkaTestUtil, util, testUtil, flinkUtil, flinkModelUtil,
   flinkTestUtil, standaloneUtil, standaloneApi, api, security, flinkApi, processReports, httpUtils, queryableState,
   restmodel, listenerApi, ui
 )

demo/docker/customerservice/Dockerfile

@@ -0,0 +1,7 @@
+FROM python:3
+RUN pip install flask-smorest
+EXPOSE 5000
+ADD app.py /
+RUN mkdir /static && flask openapi print > /static/swagger.json
+CMD [ "python3", "-m" , "flask", "run", "--host=0.0.0.0"]
+

demo/docker/customerservice/Readme.md

@@ -0,0 +1,4 @@
+This is sample service with OpenAPI description. It's implemented in python, as Nussknacker 
+can integrate with various technologies via enrichers. 
+
+Of course, it's just a stub, not intended for any kind of production usage.

demo/docker/customerservice/app.py

@@ -0,0 +1,47 @@
+import marshmallow as ma
+from flask import Flask
+from flask.views import MethodView
+from flask_smorest import Api, Blueprint
+
+class Customer:
+    def __init__(self, id, name, category):
+        self.name = name
+        self.id = id
+        self.category = category
+
+app = Flask(__name__, static_folder = '/static')
+
+@app.route('/swagger')
+def root():
+    return app.send_static_file('swagger.json')
+
+app.config['API_TITLE'] = 'Customers'
+app.config['API_VERSION'] = 'v1'
+app.config['OPENAPI_VERSION'] = '3.0.2'
+api = Api(app)
+
+class CustomerSchema(ma.Schema):
+    id = ma.fields.Int(dump_only=True)
+    name = ma.fields.String()
+    category = ma.fields.String()
+
+class CustomerQueryArgsSchema(ma.Schema):
+    name = ma.fields.String()
+
+blp = Blueprint(
+    'customers', 'customers', url_prefix='/customers',
+    description='Operations on customers'
+)
+@blp.route('/<customer_id>')
+class CustomerById(MethodView):
+
+    @blp.response(200, CustomerSchema)
+    @blp.doc(operationId='getCustomer')
+    def get(self, customer_id):
+        if customer_id == 10:
+            return Customer(customer_id, "John Doe", "SILVER")
+        else:
+            return Customer(customer_id, "Robert Wright", "GOLD")
+
+api.register_blueprint(blp)
+

demo/docker/docker-compose-env.yml

@@ -151,7 +151,9 @@ services:
       - jobmanager
       - taskmanager
       - grafana
-
+  customerservice:
+    container_name: nussknacker_customerservice
+    build: customerservice
 
 volumes:
   nussknacker_storage_zookeeper_datalog:

demo/docker/docker-compose.yml

@@ -7,14 +7,17 @@ services:
     ports:
       - "3081:8080"
     environment:
-      NUSSKNACKER_CONFIG_FILE: ${NUSSKNACKER_CONFIG_FILE:-/opt/nussknacker/conf/application.conf}
+      #multiple, comma separated, config files can be used. They will be merged in order, via HOCON fallback mechanism
+      #https://github.com/lightbend/config/blob/master/HOCON.md#config-object-merging-and-file-merging
+      NUSSKNACKER_CONFIG_FILE: ${NUSSKNACKER_CONFIG_FILE-/opt/nussknacker/conf/application.conf,/opt/nussknacker/conf/nussknacker.conf}
       COUNTS_URL: http://influxdb:8086/query
       JDK_JAVA_OPTIONS: -Xmx256M
       FLINK_ROCKSDB_CHECKPOINT_DATA_URI: file:///opt/flink/data/rocksdb-checkpoints
     volumes:
       - nussknacker_storage_app:/opt/nussknacker/storage
       #this is needed to be able to verify savepoints during deployments
       - nussknacker_storage_flink:/opt/flink/data
+      - ./nussknacker/nussknacker.conf:/opt/nussknacker/conf/nussknacker.conf
 
 
 volumes:

demo/docker/nussknacker/nussknacker.conf

@@ -0,0 +1,9 @@
+#This configuration auguments and overrides configuration in docker image
+#Here we configure OpenAPI based enricher, which is implemented by python service in customerservice
+{
+  processTypes.streaming.modelConfig.components.openAPI {
+    url: "http://customerservice:5000/swagger"
+    rootUrl: "http://customerservice:5000"
+    categories: ["Default"]
+  }
+}

docs/components/OpenAPI.md

@@ -0,0 +1,64 @@
+Overview
+========
+                              
+Nussknacker can use services documented with OpenAPI specification.
+We use Swagger to parse OpenAPI, versions 2.x and 3.x are supported 
+(version 2.x should be considered as deprecated).
+
+Nussknacker applies following rules when mapping OpenAPI services to enrichers:
+- Each operation is mapped to one enricher
+- If operationId is configured it's used as enricher name, otherwise we create it as concatenation
+of HTTP method, path and parameters (to provide some level of uniqueness)
+- Parameters (path, query of header) are used to define enricher parameters
+- If specification declares body parameter as object, we expand to parameter list  
+- We expect operation to define 200/201 response, returned object is the one that is the result of enricher
+- We map 404 HTTP code to null value of enricher
+- We use `externalDocs.url` to extract documentation link
+
+Table below describes data types that OpenAPI integration handles:
+| OpenAPI Type  | OpenAPI Format | Type in Nussknacker |
+| ------------- | -------------- | ------------------- |
+| boolean       |                | Boolean             |
+| string        |                | String              |
+| string        | date-time      | LocalDateTime       |
+| integer       |                | Long                |
+| number        |                | BigDecimal          |
+| number        | double         | Double              |
+| number        | float          | Double              |
+| array         |                | array               |
+| map/object    |                | record              |
+
+OpenAPI integration can handle schema references. 
+For objects and maps we use `properties` to define structure.
+For arrays we use `items` to define type of elements.                    
+
+
+Configuration
+=============
+
+Sample configuration:
+```
+components {
+  service1: {
+      type: openAPI  
+      url = "http://myservice.com/swagger"
+      rootUrl = "http://myservice.com/endpoint"
+      security {
+          apikey {
+              type = "apiKey"
+              apiKeyValue = "34534asfdasf"
+          }
+      }
+      namePattern: "customer.*"
+      allowedMethods: ["GET", "POST"]
+  }
+}
+```
+
+| Parameter      | Required | Default | Description                                                                                                                   |
+| ----------     | -------- | ------- | -----------                                                                                                                   |
+| url            | true     |         | URL with OpenAPI resource                                                                                                     |
+| rootUrl        | false    |         | Base URL of service, can be used to override value from OpenAPI in NAT settings                                               |
+| allowedMethods | false    | ["GET"] | Usually only GET services should be used as enrichers are meant to be idempotent and not change data                          |
+| namePattern    | false    | .*      | Regexp for filtering operations by operationId (i.e. enricher name)                                                           |
+| security       | false    |         | Configuration for [authentication](https://swagger.io/docs/specification/authentication/). Currently only apiKey is supported |

engine/api/src/main/scala/pl/touk/nussknacker/engine/api/component/ComponentProvider.scala

@@ -73,7 +73,7 @@ object NussknackerVersion {
 
 case class NussknackerVersion(value: Semver)
 
-case class ComponentDefinition(name: String, component: Component)
+case class ComponentDefinition(name: String, component: Component, icon: Option[String] = None, docsUrl: Option[String] = None)
 
 
 

engine/components/openapi/src/it/resources/application.conf

@@ -0,0 +1,12 @@
+components {
+  openAPI: {
+      url = "https://rdb-simpledb.restdb.io/rest/_swagger.json"
+      rootUrl = "https://rdb-simpledb.restdb.io/rest/"
+      security {
+          apikey {
+              type = "apiKey"
+              apiKeyValue = "TODO"
+          }
+      }
+  }
+}

engine/components/openapi/src/it/resources/customer-swagger.json

@@ -0,0 +1,66 @@
+{
+  "paths": {
+    "/customers/{customer_id}": {
+      "get": {
+        "responses": {
+          "200": {
+            "description": "OK",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/Customer"
+                }
+              }
+            }
+          }
+        },
+        "tags": [
+          "customers"
+        ],
+        "operationId": "getCustomer"
+      },
+      "parameters": [
+        {
+          "in": "path",
+          "name": "customer_id",
+          "required": true,
+          "schema": {
+            "type": "string",
+            "minLength": 1
+          }
+        }
+      ]
+    }
+  },
+  "info": {
+    "title": "Customers",
+    "version": "v1"
+  },
+  "tags": [
+    {
+      "name": "customers",
+      "description": "Operations on customers"
+    }
+  ],
+  "openapi": "3.0.2",
+  "components": {
+    "schemas": {
+      "Customer": {
+        "type": "object",
+        "properties": {
+          "name": {
+            "type": "string"
+          },
+          "category": {
+            "type": "string"
+          },
+          "id": {
+            "type": "integer",
+            "readOnly": true
+          }
+        }
+      }
+    },
+    "responses": {}
+  }
+}