[Proposal] Controller Spec - Message Broker as Subscribable APIs over Web-friendly Protocols

[senthuran16]

Overview

The API Platform Event Gateway should support protocol mediation, where a message broker (such as Kafka, MQTT) is exposed as a subscribable API to clients over web-friendly protocols (such as WebSocket, SSE, WebSub). The broker is considered as a backend, and an external client consumes messages from, and/or produces messages to that broker, over web-friendly protocols.

So far, we support WebSub APIs, adhering to the WebSub Spec, where the Event Gateway functions as the WebSubHub, backed by a broker.

As the next step, we will be supporting exposing message brokers over web-friendly protocols like WebSocket and SSE, as subscribable APIs.

Spec (See the updated spec here)

We have come up with the following spec to represent such an API. This in an example which shows a WebSocket to Kafka API.

{
    "apiVersion": "gateway.api-platform.wso2.com/v1alpha1",
    "kind": "WebBrokerApi",
    "metadata": {
        "name": "websocket-kafka-api-v1-0"
    },
    "spec": {
        "displayName": "websocket-kafka-api",
        "version": "v1.0",
        "context": "/websocket-kafka",

        "receiver": {
            "name": "my-websocket-client",
            "type": "websocket",
            "properties": {}
        },

        "brokerDriver": {
            "name": "my-kafka-broker",
            "type": "kafka",
            "properties": {
                "topic": "repo-events",
                "bootstrap.servers": "localhost:9092",
                "security.protocol": "PLAINTEXT"
            }
        },

        "allChannelPolicies": {
            "on_connection_init": {
                "request": [
                    {
                        "name": "api-key-auth",
                        "version": "v1",
                        "params": {
                            "in": "header",
                            "name": "X-API-Key"
                        }
                    }
                ],
                "response": []
            },
            "on_produce": [
                {
                    "name": "map-topics",
                    "version": "v1",
                    "params": {
                        "extraction": {
                            "source": "header",
                            "key": "X-Client-Topic"
                        },
                        "mappings": {
                            "client-issues-topic": "kafka-repo-issues",
                            "client-commits-topic": "kafka-repo-commits",
                            "client-pr-topic": "kafka-repo-pull-requests"
                        },
                        "defaultTopic": "kafka-repo-events"
                    }
                }                
            ],
            "on_consume": []
        },

        "deploymentState": "deployed"
    }
}

• receiver is the Web-friendly protocol client (eg: WebSocket).
• brokerDriver points to the message broker (eg: Kafka).
• on_connection_init refers to the initial connection (eg: handshake in WebSocket).
• on_produce - 'Produce' refers to sending a message from the web-friendly protocol client, to the broker.
• on_consume - 'Consume' refers to the web-friendly protocol client, consuming a message from the broker.
• We are planning to utilize a policy to map the topic used by the web-friendly client, to the corresponding broker's topic.

[senthuran16]

Upon further development & testing, we have updated the above spec as follows.

Spec

{
    "apiVersion": "gateway.api-platform.wso2.com/v1alpha1",
    "kind": "WebBrokerApi",
    "metadata": { "name": "websocket-kafka-api-v1-0" },
    "spec": {
      "displayName": "websocket-kafka-api",
      "version": "v1.0",
      "context": "/ws-kafka",
      "receiver": {
        "name": "ws-receiver",
        "type": "websocket",
        "properties": {}
      },
      "brokerDriver": {
        "name": "kafka-driver",
        "type": "kafka",
        "properties": {
          "bootstrap.servers": "kafka:29092"
        }
      },
      "policies": {
        "onConnectionInit": {
          "request": [],
          "response": []
        },
        "onProduce": [],
        "onConsume": []
      },
      "channels": {
        "/issues": {
          "on_produce": [
            {
              "name": "map-topic",
              "version": "v1",
              "params": {
                "mode": "produceTo",
                "topic": "produce_issues"
              }
            }
          ],
          "on_consume": [
            {
              "name": "map-topic",
              "version": "v1",
              "params": {
                "mode": "consumeFrom",
                "topic": "consume_issues"
              }
            }
          ]
        },
        "/pull-requests": {
          "on_produce": [
            {
              "name": "map-topic",
              "version": "v1",
              "params": {
                "mode": "produceTo",
                "topic": "produce_pull-requests"
              }
            }
          ],
          "on_consume": [
            {
              "name": "map-topic",
              "version": "v1",
              "params": {
                "mode": "consumeFrom",
                "topic": "consume_pull-requests"
              }
            }
          ]
        }
      },
      "deploymentState": "deployed"
    }
  }

Topic Mapping

• Topic mapping will be handled via a channel policy (map-topic in the above example).
• In the above example, the /issues channel of the receiver (WebSocket) will produce to the produce_issues Kafka topic. Also, the /issues channel of the WebSocket client will receive messages from the consume_issues Kafka topic. Altogether, the WebSocket client which is connected to the /issues channel, will be able to:
  • Send messages to the produce_issues Kafka topic.
  • Receive messages from the consume_issues Kafka topic.
• When no topic mapping is present, the receiver (WebSocket) channel's name will be equal to the broker (Kafka) topic name. i.e, in the following:

    "channels": {
      "/issues": {
        "on_produce": [],
        "on_consume": []
      },
      "/pull-requests": {
        "on_produce": [],
        "on_consume": []
      }
    },

  • The WebSocket client which is connected to the /issues channel will be able to send and receive messages to the _2f_issues topic (topic name is normalized).

[nuwand]

I have a few questions/concerns.

1. What you have provided above is not a spec, its an example. We need the spec first.

2. I am confused by the policies {. . . . } section. You seem to have a special section dedicated for policies, but at the same time policies are also declared under events for each channel. It is confusing where I should put policies and when.

3. When you say policies, one would naturally expect an array of policy objects. But that doesn't seem to be the case on the policies section. policies is an object! However, at the channel level, it actually seems to be expecting an array of policy objects. So this is semantically inconsistent. These issues are stemming from not having a proper spec in the first place. If we do the spec correctly these kind of issues won't happen.

4. Where do I put policies that should be applied to all channels?

[senthuran16]

Hi @nuwand ,

Please find the spec below, alongwith descriptions that addresses the above questions:

---

Spec

{
    "apiVersion": "gateway.api-platform.wso2.com/v1alpha1",
    "kind": "WebBrokerApi",
    "metadata": { "name": "<api-name>" },
    "spec": {
        "displayName": "<api-display-name>",
        "version": "<api-version>",
        "context": "<api-context>",

        "receiver": {
            "name": "<receiver-name>",
            "type": "<receiver-type>",
            "properties": { /* Properties of the receiver */ }
        },

        "brokerDriver": {
            "name": "<brokerDriver-name>",
            "type": "<brokerDriver-type>",
            "properties": { /* Properties of the broker driver */ }
        },

        "policies": {
            "on_connection_init": [
                /* Policies applied to every initial connection */
            ],
            "on_produce": [
                /* Policies applied to every inbound message (web-friendly client to broker) */
            ],
            "on_consume": [ 
                /* Policies applied to every outbound message (broker to web-friendly client) */
            ]
        },

        "channels": {
            "<channel-name>": {
                "produceTo": { "topic": "<produce-messages-to-this-topic>" },
                "consumeFrom": { "topic": "<consume-messages-to-this-topic>" },
                "policies": {
                    "on_connection_init": [
                        /* channel-specific initial connection response policies */
                    ],
                    "on_produce": [ 
                        /* channel-specific policies for inbound messages   (web-friendly client to broker) */ 
                    ],
                    "on_consume": [ 
                        /* channel-specific policies for outbound messages  (broker to web-friendly client) */ 
                    ]
                }
            }
        },

        "deploymentState": "deployed"
    }
}

For an API which exposes a broker over a web-friendly protocol, there will be following events, on which we can apply policies:

1. on_connection_init - The initial connection of the web-friendly protocol client (eg: The handshake for a WebSocket client).
2. on_produce - Sending a message from the web-friendly protocol client, to the broker.
3. on_consume - The web-friendly protocol client consuming a message from the broker.

Policies can be applied for the above events, across all channels (policies{...}), or to a particular channel (channels.<channelName>.policies{...}).

Naming convention for all-channel policies and channel-specific policies follows the same convention of WebSub APIs.

Topic Mapping

Messages that are sent via a particular channel from the web-friendly client, will be mapped and sent to the topic denoted by produceTo. Likewise, the web-friendly client which is connected to the particular channel, will consume events from the topic denoted by consumeFrom.

        "channels": {
            "<channel-name>": {
                "produceTo": { "topic": "<produce-messages-to-this-topic>" },
                "consumeFrom": { "topic": "<consume-messages-to-this-topic>" },
                ...
            }
         }

An Example

{
  "apiVersion": "gateway.api-platform.wso2.com/v1alpha1",
  "kind": "WebBrokerApi",
  "metadata": {
    "name": "stock-trading-v1.0"
  },
  "spec": {
    "displayName": "Stock Trading WebBroker API",
    "version": "v1.0",
    "context": "/stock-trading/v1.0",
    "receiver": {
      "name": "websocket-receiver",
      "type": "websocket"
    },
    "brokerDriver": {
      "name": "kafka-driver",
      "type": "kafka",
      "properties": {
        "bootstrapServers": [
          "kafka-broker-1:9092",
          "kafka-broker-2:9092"
        ]
      }
    },
    "channels": {
      "prices": {
        "produceTo": {
          "topic": "stock.prices"
        },
        "consumeFrom": {
          "topic": "stock.prices"
        },
        "policies": {
          "on_connection_init": [],
          "on_produce": [],
          "on_consume": []
        }
      }
    }
  }
}

[senthuran16]

We had a discussion, and we agreed on the following terms & semantics. Will come up with a spec with this, and provide an update here.

{
  "apiVersion": "gateway.api-platform.wso2.com/v1alpha1",
  "kind": "WebBrokerApi",
  "metadata": {
    "name": "stock-trading-v1.0"
  },
  "spec": {
    "displayName": "Stock Trading WebBroker API",
    "version": "v1.0",
    "context": "/stock-trading/v1.0",
    "receiver": {
      "name": "websocket-receiver",
      "type": "websocket"
    },
    "broker": {
      "name": "kafka-driver",
      "type": "kafka",
      "properties": {
        "bootstrapServers": [
          "kafka-broker-1:9092",
          "kafka-broker-2:9092"
        ]
      }
    },
    "allChannels": {
      "on_connection_init": {
        "policies": []
      },
      "on_produce": {
        "policies": []
      },
      "on_consume": {
        "policies": []
      }
    },
    "channels": {
      "prices": {
        "produceTo": {
          "topic": "stock.prices"
        },
        "consumeFrom": {
          "topic": "stock.prices"
        },
        "on_connection_init": {
          "policies": [ policyObject[] ]
        },
        "on_produce": {
          "policies": [ policyObject[] ]
        },
        "on_consume": {
          "policies": [ policyObject[] ]
        }
      }
    }
  }
}

[senthuran16]

Please find the revamped spec below, based on our latest discussion.

Spec for WebBrokerApi

WebBrokerApi:
  apiVersion: "gateway.api-platform.wso2.com/v1alpha1"
  kind: "WebBrokerApi"

  metadata:
    name: string

  spec:
    displayName: string
    version: string
    context: string

    receiver: WebBrokerApiReceiver  # Web-friendly protocol client

    broker: WebBrokerApiBroker  # Message broker

    # Policies that are applied to all the channels
    allChannels:
      # During the initial connection (eg: WebSocket handshake)
      on_connection_init:
        policies: PolicyObject[]

      # When a message is sent from the Web-friendly protocol client, to the broker
      on_produce:
        policies: PolicyObject[]

      # When a message is sent from the broker, to the Web-friendly protocol client
      on_consume:
        policies: PolicyObject[]

    channels:
      <channel-name>:
        produceTo:
          topic: string   # The channel will produce messages to this topic

        consumeFrom:
          topic: string   # The channel will consume messages from this topic

        # The below policies are applied to this specific <channel-name> channel, upon events

        on_connection_init:
          policies: PolicyObject[]

        on_produce:
          policies: PolicyObject[]

        on_consume:
          policies: PolicyObject[]

---

WebBrokerApiReceiver:
  name: string
  type: string
  properties:
    <stringKey>: <objectValue>

---

WebBrokerApiBroker:
  name: string
  type: string
  properties:
    <stringKey>: <objectValue>

---

PolicyObject:
  name: string
  version: string
  parameters:
    <key>: <value>

---

WebBrokerApiProduceConfig:
  topic: string

---

WebBrokerApiConsumeConfig:
  topic: string

An Example based on the Above Spec

{
  "apiVersion": "gateway.api-platform.wso2.com/v1alpha1",
  "kind": "WebBrokerApi",
  "metadata": {
    "name": "stock-trading-v1.0"
  },
  "spec": {
    "displayName": "Stock Trading WebBroker API",
    "version": "v1.0",
    "context": "/stock-trading/v1.0",
    "receiver": {
      "name": "websocket-receiver",
      "type": "websocket"
    },
    "broker": {
      "name": "kafka-driver",
      "type": "kafka",
      "properties": {
        "brokers": [
          "kafka:29092"
        ]
      }
    },
    "allChannels": {
      "on_connection_init": {
        "policies": []
      },
      "on_produce": {
        "policies": []
      },
      "on_consume": {
        "policies": []
      }
    },
    "channels": {
      "prices": {
        "produceTo": {
          "topic": "stock.prices"
        },
        "consumeFrom": {
          "topic": "dummy.prices"
        },
        "on_connection_init": {
          "policies": []
        },
        "on_produce": {
          "policies": []
        },
        "on_consume": {
          "policies": []
        }
      }
    }
  }
}