Merge pull request #308 from Accenture/307-use-ce-lib

307 use ce lib

Author: mmacai

CHANGELOG.md

@@ -13,7 +13,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Added validation for reverse proxy configuration. Now it crashes RIG on start when configuration is not valid or returns `400` when using REST API to update configuration. [#277](https://github.com/Accenture/reactive-interaction-gateway/issues/277)
 - Added basic distributed tracing support in [W3C Trace Context specification](https://www.w3.org/TR/trace-context/) with Jaeger and Openzipkin exporters. RIG opens a span at the API Gateway and emits trace context in Cloud Events following the [distributed tracing spec](https://github.com/cloudevents/spec/blob/v1.0/extensions/distributed-tracing.md). [#281](https://github.com/Accenture/reactive-interaction-gateway/issues/281)
 
-<!-- ### Changed -->
+### Changed
+
+- Incorporated [cloudevents-ex](https://github.com/kevinbader/cloudevents-ex) to handle binary and structured modes for [Kafka protocol binding](https://github.com/cloudevents/spec/blob/v1.0/kafka-protocol-binding.md) in a proper way. This introduces some **breaking changes**:
+  - Binary mode is now using `ce_` prefix for CloudEvents context attribute headers, before it was `ce-` - done according to the [Kafka protocol binding](https://github.com/cloudevents/spec/blob/v1.0/kafka-protocol-binding.md)
+- Change above affects also `"response_from": "kafka"` proxy functionality. RIG will forward to clients only Kafka body, no headers. This means, when using binary mode, clients receive only the data part, no CloudEvents context attributes.
 
 ### Fixed
 

config/config.exs

@@ -102,7 +102,9 @@ config :logger, :console,
   format: "\n$time [$level] $levelpad$message\n$metadata\n",
   metadata: metadata |> Enum.uniq()
 
-config :rig, Rig.Application, log_level: {:system, :atom, "LOG_LEVEL", :debug}
+config :rig, Rig.Application,
+  log_level: {:system, :atom, "LOG_LEVEL", :debug},
+  schema_registry_host: {:system, "KAFKA_SCHEMA_REGISTRY_HOST", nil}
 
 # --------------------------------------
 # Session and Authorization

config/rig_tests/test.exs

@@ -23,7 +23,7 @@ config :rig, RigTests.Proxy.ResponseFrom.KafkaTest,
   ssl_keyfile_pass: {:system, "KAFKA_SSL_KEYFILE_PASS", ""},
   # Credentials for SASL/Plain authentication. Example: "plain:myusername:mypassword"
   sasl: {:system, "KAFKA_SASL", nil},
-  response_topic: "rig-proxy-response"
+  response_topic: {:system, "PROXY_KAFKA_RESPONSE_TOPICS", "rig-proxy-response"}
 
 config :rig, RigTests.Proxy.PublishToEventStream.KafkaTest,
   server_id: :rig_proxy_publish_kafkatest_genserver,

docs/api-gateway.md

@@ -46,11 +46,11 @@ This defines a single service called "my-service". The URL is read from an given
 As a demo service, we use a small Node.js script:
 
 ```js
-const http = require('http');
+const http = require("http");
 const port = 3000;
 const handler = (_req, res) => res.end("Hi, I'm a demo service!\n");
 const server = http.createServer(handler);
-server.listen(port, err => {
+server.listen(port, (err) => {
   if (err) {
     return console.error(err);
   }
@@ -182,6 +182,8 @@ The endpoint expects the following request format:
 Sometimes it makes sense to provide a simple request-response API to something that runs asynchronously on the backend. For example, let's say there's a ticket reservation process that takes 10 seconds in total and involves three different services that communicate via message passing. For an external client, it may be simpler to wait 10 seconds for the response instead of polling for a response every other second.
 A behavior like this can be configured using an endpoints' `response_from` property. When set to `kafka`, the response to the request is not taken from the `target` (e.g., for `target` = `http` this means the backend's HTTP response is ignored), but instead it's read from a Kafka topic. In order to enable RIG to correlate the response from the topic with the original request, RIG adds a correlation ID to the request (using a query parameter in case of `target` = `http`, or backed into the produced CloudEvent otherwise). Backend services that work with the request need to include that correlation ID in their response; otherwise, RIG won't be able to forward it to the client (and times out).
 
+> In case you want to use _binary_ transport mode, make sure that `rig` extension (containing correlation ID) is prefixed with `ce_` as well.
+
 Configuration of such API endpoint might look like this:
 
 ```json

docs/avro.md

@@ -24,37 +24,37 @@ Adopting Avro for event (de)serialization is fairly straightforward. First you n
 ## RIG as a Kafka producer
 
 - producer evaluates if serialization is turned on by checking `KAFKA_SERIALIZER` environment variable and if it's value is `avro`
-- If it is, creates headers for Kafka event by appending `ce-` prefix for every field, besides `data` field
+- If it is, creates headers for Kafka event by appending `ce_` prefix for every field, except `data` field - **binary mode**
   - **nested context attributes are stringified**, since Kafka headers don't support nested values (this is common when using Cloud events extensions)
 - after that, the `data` field is serialized using the schema name (function for getting schemas from registry is cached in-memory)
 - producer sends event with created headers and data (in binary format `<<0, 0, 0, 0, 1, 5, 3, 8, ...>>`) to Kafka
 
-> If `KAFKA_SERIALIZER` is not set to `avro`, producer sets **only** `ce-contenttype` or `ce-contentType` for kafka event
+> If `KAFKA_SERIALIZER` is not set to `avro`, producer sets **only** `ce_contenttype` or `ce_contentType` for kafka event
 
 ## RIG as a Kafka consumer
 
-- when consuming Kafka event, RIG checks headers of such event and removes `ce-` prefix
-- based on headers decides cloud events version and content type
-- In case content type is `avro/binary`, schema ID is taken from event value and deserialized
-- If content type is **not** present, RIG checks for Avro format (`<<0, 0, 0, 0, 1, 5, 3, 8, ...>>`) and attempts for deserialization, otherwise event is sent to client as it is without any deserialization
+Event parsing is based on the [Kafka Transport Binding for CloudEvents v1.0](https://github.com/cloudevents/spec/blob/v1.0/kafka-protocol-binding.md) implemented via [cloudevents-ex](https://github.com/kevinbader/cloudevents-ex). Check the [Event format](./event-format.md#kafka-transport-binding) section.
 
 ## Example 1: producing to and consuming from the same topic
 
 In this example we'll have RIG send a message to itself to see whether RIG producing and consuming parts work correctly. The idea is that RIG produces a serialized event as a result to an HTTP request and, a few moments later, consumes that same event (and deserializes it correctly).
 
 ```bash
+
 ## 1. Start Kafka with Zookeeper and Kafka Schema Registry
+
 KAFKA_PORT_PLAIN=17092 KAFKA_PORT_SSL=17093 HOST=localhost docker-compose -f integration_tests/kafka_tests/docker-compose.yml up -d
 
 ## 2. Start Rig
+
 # Here we say to use Avro, consume on topic "rigRequest" and use "rigRequest-value" schema from Kafka Schema Registry
 # Proxy is turned on to be able to produce Kafka event with headers (needed for cloud events)
 docker run --name rig \
 -e KAFKA_BROKERS=kafka:9292 \
 -e KAFKA_SERIALIZER=avro \
 -e KAFKA_SCHEMA_REGISTRY_HOST=kafka-schema-registry:8081 \
 -e KAFKA_SOURCE_TOPICS=rigRequest \
--e PROXY_CONFIG_FILE=proxy/proxy.test.json \
+-e PROXY_CONFIG_FILE='[{"id":"my-api","name":"my-api","versioned":false,"version_data":{"default":{"endpoints":[{"id":"post-myapi-publish-async","path":"/myapi/publish-async","method":"POST","target":"kafka"}]}},"proxy":{"use_env":true,"target_url":"KAFKA_HOST","port":9092}}]' \
 -e PROXY_KAFKA_REQUEST_TOPIC=rigRequest \
 -e PROXY_KAFKA_REQUEST_AVRO=rigRequest-value \
 -e LOG_LEVEL=debug \
@@ -63,25 +63,25 @@ docker run --name rig \
 accenture/reactive-interaction-gateway
 
 ## 3. Register Avro schema in Kafka Schema Registry
+
 curl -d '{"schema":"{\"name\":\"rigproducer\",\"type\":\"record\",\"fields\":[{\"name\":\"example\",\"type\":\"string\"}]}"}' -H "Content-Type: application/vnd.schemaregistry.v1+json" -X POST http://localhost:8081/subjects/rigRequest-value/versions
 
 ## 4. Send HTTP request to RIG proxy
+
 # Request will produce serialized Kafka event to Kafka
 curl -d '{"event":{"id":"069711bf-3946-4661-984f-c667657b8d85","type":"com.example","time":"2018-04-05T17:31:00Z","specversion":"0.2","source":"\/cli","contenttype":"avro\/binary","data":{"example":"test"}},"partition":"test_key"}' -H "Content-Type: application/json" -X POST http://localhost:4000/myapi/publish-async
 
 ## 5. In terminal you should see something like below -- in nutshell it means event was successfully consumed, deserialized and forwarded to UI client
-16:46:31.549 [debug] Decoded Avro message="{\"example\":\"test\"}"
-application=rig_kafka module=RigKafka.Avro function=decode/1 file=lib/rig_kafka/avro.ex line=28 pid=<0.419.0>
 
-16:46:31.550 [debug] [:start_object, {:string, "contenttype"}, :colon, {:string, "avro/binary"}, :comma, {:string, "data"}, :colon, :start_object, {:string, "example"}, :colon, {:string, "test"}, :end_object, :comma, {:string, "id"}, :colon, {:string, "069711bf-3946-4661-984f-c667657b8d85"}, :comma, {:string, "rig"}, :colon, :start_object, {:string, "correlation"}, :colon, {:string, "g2dkAA1ub25vZGVAbm9ob3N0AAADzAAAAAAA"}, :comma, {:string, "headers"}, :colon, :start_array, :start_array, {:string, "accept"}, :end_array, :comma, :start_array, {:string, "*/*"}, :end_array, :comma, :start_array, {:string, "content-length"}, :end_array, :comma, :start_array, {:string, "221"}, :end_array, :comma, :start_array, {:string, "content-type"}, :end_array, :comma, :start_array, {:string, ...}, :end_array, ...]
-application=rig module=Rig.EventStream.KafkaToFilter function=kafka_handler/1 file=lib/rig/event_stream/kafka_to_filter.ex line=20 pid=<0.419.0>
+21:54:52.869 module=Avrora.Storage.Registry [debug] obtaining schema with global id `1`
+21:54:52.870 module=Rig.EventStream.KafkaToFilter [debug] %Cloudevents.Format.V_0_2.Event{contenttype: "avro/binary", data: %{"example" => "test"}, extensions: %{"rig" => %{"correlation" => "Ve1d-XF0Qi46lwh47X5IqI7m_FCIqCLsqyV0KTCxg28Hnd7ytczBe1cASZYPxA7GNFCZ4AzDC0QX1w0=", "headers" => [["accept", "*/*"], ["content-length", "221"], ["content-type", "application/json"], ["host", "localhost:4000"], ["user-agent", "curl/7.54.0"]], "host" => "localhost", "method" => "POST", "path" => "/myapi/publish-async", "port" => 4000, "query" => "", "remoteip" => "172.28.0.1", "scheme" => "http"}}, id: "069711bf-3946-4661-984f-c667657b8d85", schemaurl: nil, source: "/cli", specversion: "0.2", time: "2018-04-05T17:31:00Z", type: "com.example"}
 ```
 
 ## Example 2: Kafka schema Registry CLI
 
 To check if it works also with native serializer we can leverage the CLI shipped with the Kafka Schema Registry image.
 
-```bash
+``` bash
 # 1. Get inside Kafka Schema Registry container
 docker exec -it kafka-schema-registry bash
 

docs/event-format.md

@@ -14,17 +14,17 @@ We aim at following the [CloudEvents specification](https://github.com/cloudeven
 
 ```json
 {
-    "specversion" : "0.2",
-    "type" : "com.github.pull.create",
-    "source" : "https://github.com/cloudevents/spec/pull/123",
-    "id" : "A234-1234-1234",
-    "time" : "2018-04-05T17:31:00Z",
-    "comexampleextension1" : "value",
-    "comexampleextension2" : {
-        "othervalue": 5
-    },
-    "contenttype" : "text/xml",
-    "data" : "<much wow=\"xml\"/>"
+  "specversion": "0.2",
+  "type": "com.github.pull.create",
+  "source": "https://github.com/cloudevents/spec/pull/123",
+  "id": "A234-1234-1234",
+  "time": "2018-04-05T17:31:00Z",
+  "comexampleextension1": "value",
+  "comexampleextension2": {
+    "othervalue": 5
+  },
+  "contenttype": "text/xml",
+  "data": "<much wow=\"xml\"/>"
 }
 ```
 
@@ -41,8 +41,6 @@ For further details and examples, checkout the dedicated [Section on Avro](avro)
 
 ## Transport Bindings
 
-While the HTTP transport binding seems solid already, the Kafka transport mode is not yet available. To support Kafka anyway, we have adapted the HTTP transport modes for Kafka. Consequently, the Kafka transport mode implementation may change in the future, depending on the outcome of the [standardization process](https://github.com/cloudevents/spec/pull/337).
-
 When sending or receiving an event, you have a couple of options:
 
 - Send the event as-is, or send only the "data" (= payload) in the "body" and the rest (the so-called "context attributes") in headers. The former is called _structured_ transport mode, while the latter is known as _binary_ transport mode.
@@ -90,7 +88,7 @@ Request body:
 
 ### Binary
 
-In binary mode the request body only contains the `data` value of the corresponding CloudEvent. The _context attributes_ - i.e., all other fields - are moved into the HTTP header. The data/body encoding is determined by the `content-type` header. At the time of writing there are two content types supported: `application/json` and `avro/binary`.
+In binary mode the request body only contains the `data` value of the corresponding CloudEvent. The _context attributes_ - i.e., all other fields - are moved into the HTTP header (**this means also [extensions](https://github.com/cloudevents/spec/blob/v1.0/spec.md#extension-context-attributes)**). The data/body encoding is determined by the `content-type` header. At the time of writing there are two content types supported: `application/json` and `avro/binary`.
 
 <details>
 <summary>Same example event, sent using HTTP request in binary mode</summary>
@@ -119,7 +117,7 @@ Request body:
 
 ## Kafka Transport Binding
 
-As mentioned above, the corresponding CloudEvents transport binding specification is still under [active development](https://github.com/cloudevents/spec/pull/337/files). We follow an approach very similar to the HTTP transport binding outlined above. We utilize Kafka headers that have been introduced in Kafka version `0.11`. In order to support older Kafka versions as well, RIG defaults to structured mode and does not require any headers at all (see below).
+Implemented using [Kafka Transport Binding for CloudEvents v1.0](https://github.com/cloudevents/spec/blob/v1.0/kafka-protocol-binding.md). We utilize Kafka headers that have been introduced in Kafka version `0.11`. In order to support older Kafka versions as well, RIG defaults to structured mode and does not require any headers at all (see below).
 
 Like with the HTTP transport binding, we define two modes of operation: structured and binary.
 
@@ -158,7 +156,7 @@ Message body:
 
 ### Binary
 
-In binary mode the message body only contains the `data` value of the corresponding CloudEvent. The _context attributes_ - i.e., all other fields - are moved into the message header. The data/body encoding is determined by the `content-type` header. In this mode there is no default for `content-type` and RIG rejects messages that come without it. At the time of writing there are two content types supported: `application/json` and `avro/binary`.
+In binary mode the message body only contains the `data` value of the corresponding CloudEvent. The _context attributes_ - i.e., all other fields - are moved into the message header (**this means also [extensions](https://github.com/cloudevents/spec/blob/v1.0/spec.md#extension-context-attributes)**). The data/body encoding is determined by the `content-type` header. In this mode there is no default for `content-type` and RIG rejects messages that come without it. At the time of writing there are two content types supported: `application/json` and `avro/binary`.
 
 <details>
 <summary>Same example in binary mode</summary>
@@ -167,10 +165,10 @@ In binary mode the message body only contains the `data` value of the correspond
 In binary mode the message header contains all context attributes. It also announces the body encoding:
 
 ```plaintext
-ce-specversion: 0.2
-ce-type: com.example.someevent
-ce-source: example
-ce-id: 80dc037c-fb24-43e9-9759-94f91f310a4b1
+ce_specversion: 0.2
+ce_type: com.example.someevent
+ce_source: example
+ce_id: 80dc037c-fb24-43e9-9759-94f91f310a4b1
 Content-Type: application/json; charset=UTF-8
 ```
 
@@ -185,6 +183,6 @@ Message body:
 </p>
 </details>
 
-[CloudEvents]: https://cloudevents.io/
-[CNCF]: https://www.cncf.io/
-[Apache Avro]: https://avro.apache.org/
+[cloudevents]: https://cloudevents.io/
+[cncf]: https://www.cncf.io/
+[apache avro]: https://avro.apache.org/