merge with master

Author: Knappek

CHANGELOG.md

@@ -9,13 +9,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Support publishing events consumed from [NATS](https://nats.io) topics. See the [documentation](https://accenture.github.io/reactive-interaction-gateway/docs/event-streams.html#nats) for how to get started. [#297](https://github.com/Accenture/reactive-interaction-gateway/issues/297)
+- 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)
+- Added possibility to set response code for `response_from` messages in reverse proxy (`kafka` and `http_async`). [#321](https://github.com/Accenture/reactive-interaction-gateway/pull/321)
+- Added new version - `v3` - for internal endpoints to support response code in the `/responses` endpoint
 - Added Helm v3 template to the `deployment` folder [#288](https://github.com/Accenture/reactive-interaction-gateway/issues/288)
 
 ### Changed
 
-- Support publishing events consumed from [NATS](https://nats.io) topics. See the [documentation](https://accenture.github.io/reactive-interaction-gateway/docs/event-streams.html#nats) for how to get started. [#297](https://github.com/Accenture/reactive-interaction-gateway/issues/297)
-- 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)
+- 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.
+- Changed `response_from` handler to expect a message in binary format, **NOT** a cloud event (`kafka` and `http_async`). [#321](https://github.com/Accenture/reactive-interaction-gateway/pull/321)
 - Updated Helm v2 template, kubectl yaml file and instructions in the `deployment` folder [#288](https://github.com/Accenture/reactive-interaction-gateway/issues/288)
 
 ### Fixed

Dockerfile

@@ -40,8 +40,11 @@ RUN apk add --no-cache bash
 ENV LANG C.UTF-8
 ENV LC_ALL C.UTF-8
 
+RUN addgroup -S rig -g 1000 && adduser -S rig -G rig --uid 1000
 WORKDIR /opt/sites/rig
 COPY --from=build /opt/sites/rig/_build/prod/rel/rig /opt/sites/rig/
+RUN chown -R rig:rig /opt/sites/rig
+USER rig
 
 # Proxy
 EXPOSE 4000

aws.dockerfile

@@ -49,9 +49,12 @@ ENV KINESIS_OTP_JAR=/opt/sites/rig/kinesis-client/local-maven-repo/org/erlang/ot
 # Install Java
 RUN apk add --no-cache openjdk8-jre
 
+RUN addgroup -S rig -g 1000 && adduser -S rig -G rig --uid 1000
 WORKDIR /opt/sites/rig
 COPY --from=elixir-build /opt/sites/rig/_build/prod/rel/rig /opt/sites/rig/
 COPY --from=java-build opt/sites/rig/kinesis-client /opt/sites/rig/kinesis-client
+RUN chown -R rig:rig /opt/sites/rig
+USER rig
 
 # Proxy
 EXPOSE 4000

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_api/config.exs

@@ -45,6 +45,7 @@ config :phoenix, :serve_endpoints, true
 
 config :rig, RigApi.V1.APIs, rig_proxy: RigInboundGateway.Proxy
 config :rig, RigApi.V2.APIs, rig_proxy: RigInboundGateway.Proxy
+config :rig, RigApi.V3.APIs, rig_proxy: RigInboundGateway.Proxy
 
 config :rig, :event_filter, Rig.EventFilter
 

config/rig_api/test.exs

@@ -10,5 +10,6 @@ config :rig, RigApi.Endpoint,
 
 config :rig, RigApi.V1.APIs, rig_proxy: RigInboundGateway.ProxyMock
 config :rig, RigApi.V2.APIs, rig_proxy: RigInboundGateway.ProxyMock
+config :rig, RigApi.V3.APIs, rig_proxy: RigInboundGateway.ProxyMock
 
 config :rig, :event_filter, Rig.EventFilterMock

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);
   }
@@ -180,7 +180,7 @@ The endpoint expects the following request format:
 ### Wait for response
 
 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).
+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).**
 
 Configuration of such API endpoint might look like this:
 
@@ -205,26 +205,37 @@ Configuration of such API endpoint might look like this:
 
 > Note the presence of `response_from` field. This tells RIG to wait for different event with the same correlation ID.
 
-As an alternative you can set `response_from` to `http_async`. This means that correlated response has to be sent to internal `:4010/v2/responses` `POST` endpoint with a body like this:
+#### Supported combinations (`target` -> `response_from`)
+
+- HTTP -> `kafka`/`http_async`/`kinesis`
+- Kafka -> `kafka`
+- Kinesis -> **not supported**
+- Nats -> `nats`
+
+`http_async` means that correlated response has to be sent to internal `:4010/v3/responses` `POST` endpoint.
+
+#### Supported formats
+
+All `response_from` options support only binary mode.
+
+Message headers:
+
+```plaintext
+rig-correlation: "correlation_id_sent_by_rig"
+rig-response-code: "201"
+content-type: "application/json"
+```
+
+> All headers are required.
+
+Message body:
 
 ```json
 {
-  "id": "1",
-  "specversion": "0.2",
-  "source": "my-service",
-  "type": "com.example",
-  "rig": {
-    "correlation": "_id_"
-  },
-  "data": {
-    ...
-  }
-  ...
+  "foo": "bar"
 }
 ```
 
-> **NOTE:** Kinesis doesn't support `response_from` field yet.
-
 ## Auth
 
 RIG can do simple auth check for endpoints. Currently supports JWT.

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