You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@openwhisk.apache.org by st...@apache.org on 2022/05/30 04:13:47 UTC
[openwhisk-package-kafka] branch master updated: Support dedicated providers (#385)
This is an automated email from the ASF dual-hosted git repository.
style95 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/openwhisk-package-kafka.git
The following commit(s) were added to refs/heads/master by this push:
new 8209440 Support dedicated providers (#385)
8209440 is described below
commit 82094403081259f6ee924a46b0384dedeeb3a5cb
Author: jiangpengcheng <sc...@gmail.com>
AuthorDate: Mon May 30 12:13:44 2022 +0800
Support dedicated providers (#385)
* Support dedicated kafka providers
* Update doc
* Update doc
---
README.md | 16 ++++++++++++++++
action/lib/common.js | 16 ++++++++++------
docs/dev/README.md | 3 ++-
installKafka.sh | 17 +++++++++++------
4 files changed, 39 insertions(+), 13 deletions(-)
diff --git a/README.md b/README.md
index 6d42987..3b90cd1 100644
--- a/README.md
+++ b/README.md
@@ -109,6 +109,22 @@ Example:
$ wsk trigger create MyKafkaTrigger -f /whisk.system/messaging/kafkaFeed -p brokers "[\"mykafkahost:9092\", \"mykafkahost:9093\"]" -p topic mytopic -p isJSONData true
```
+### Using a separated kafka feed provider for each user
+Sometimes users may not want to expose their kafka to the shared feed providers which are provided by OpenWhisk cloud provider.
+They can run their own providers and use their own CouchDB/Cloudant by passing below additional parameters when **create/update/get/delete** triggers:
+
+|Name|Type|Description|
+|---|---|---|
+|dedicated|Boolean|`true` to use dedicated kafka feed providers and CouchDB/Cloudant, default is `false`|
+|DB_URL|URL|The base URL(including username:password) for persistent storage (either CouchDB or Cloudant)|
+|DB_NAME|String|The database name for triggers|
+|workers|An array of the IDs of the running instances with each ID of the form `workerX`. e.g. `["worker0", "worker1"]`|
+
+Example:
+```
+$ wsk trigger create MyKafkaTrigger -f /whisk.system/messaging/kafkaFeed -p brokers "[\"mykafkahost:9092\", \"mykafkahost:9093\"]" -p topic mytopic -p isJSONData true -p dedicated true -p DB_URL http://admin:admin@localhost:5984 -p DB_NAME dedicated_triggers -p workers "[\"worker0\"]"
+```
+
### Listening for messages
After creating a trigger, the system will monitor the specified topic in your messaging service. When new messages are posted, the trigger will be fired.
diff --git a/action/lib/common.js b/action/lib/common.js
index a09bc4e..9ec1481 100644
--- a/action/lib/common.js
+++ b/action/lib/common.js
@@ -104,16 +104,20 @@ function massageParamsForWeb(rawParams) {
return massagedParams;
}
-function getWebActionURL(endpoint, actionName) {
+function getWebActionURL(endpoint, dedicated, actionName) {
var apiHost = addHTTPS(endpoint);
+ var package = 'messagingWeb';
+ if (dedicated === true || dedicated === 'true') {
+ package = 'messagingWebDedicated'
+ }
- return `${apiHost}/api/v1/web/whisk.system/messagingWeb/${actionName}`;
+ return `${apiHost}/api/v1/web/whisk.system/${package}/${actionName}`;
}
function createTrigger(endpoint, params, actionName) {
var options = {
method: 'POST',
- url: getWebActionURL(endpoint, actionName),
+ url: getWebActionURL(endpoint, params.dedicated, actionName),
rejectUnauthorized: false,
json: true,
body: params,
@@ -140,7 +144,7 @@ function createTrigger(endpoint, params, actionName) {
function deleteTrigger(endpoint, params, actionName) {
var options = {
method: 'DELETE',
- url: getWebActionURL(endpoint, actionName),
+ url: getWebActionURL(endpoint, params.dedicated, actionName),
rejectUnauthorized: false,
json: true,
body: params,
@@ -164,7 +168,7 @@ function deleteTrigger(endpoint, params, actionName) {
function getTrigger(endpoint, params, actionName) {
var options = {
method: 'GET',
- url: getWebActionURL(endpoint, actionName),
+ url: getWebActionURL(endpoint, params.dedicated, actionName),
rejectUnauthorized: false,
json: true,
qs: params,
@@ -187,7 +191,7 @@ function getTrigger(endpoint, params, actionName) {
function updateTrigger(endpoint, params, actionName) {
var options = {
method: 'PUT',
- url: getWebActionURL(endpoint, actionName),
+ url: getWebActionURL(endpoint, params.dedicated, actionName),
rejectUnauthorized: false,
json: true,
body: params,
diff --git a/docs/dev/README.md b/docs/dev/README.md
index e81bca6..3ee8f88 100644
--- a/docs/dev/README.md
+++ b/docs/dev/README.md
@@ -42,11 +42,12 @@ Now we need to start the provider service. This is also a simple matter of runni
|LOCAL_DEV|Boolean|If you are using a locally-deployed OpenWhisk core system, it likely has a self-signed certificate. Set `LOCAL_DEV` to `true` to allow firing triggers without checking the certificate validity. *Do not use this for production systems!*|
|PAYLOAD_LIMIT|Integer (default=900000)|The maximum payload size, in bytes, allowed during message batching. This value should be less than your OpenWhisk deployment's payload limit.|
|WORKER|String|The ID of this running instances. Useful when running multiple instances. This should be of the form `workerX`. e.g. `worker0`.
+|DB_PREFIX|String|A prefix to be prepended to the default DB name|
With that in mind, starting the feed service might look something like:
```sh
-docker run -e DB_URL=https://myDbHost -e DB_USER=MyDbUser -e DB_PASS=MySuperSecret -p 80:5000 kafkafeedprovider
+docker run -e DB_URL=https://myDbHost -e DB_USER=MyDbUser -e DB_PASS=MySuperSecret -e DB_PREFIX=ow_ -p 80:5000 kafkafeedprovider
```
This example will start the provider service with the specified DB details. The container provides a number of RESTful endpoints which can be accessed on port 5000 _inside_ the container. To expose this port to the rest of the world `-p 80:5000` tells Docker to map port 80 of the host machine into port 5000 inside this new container.
diff --git a/installKafka.sh b/installKafka.sh
index 8f17dac..a180cfe 100755
--- a/installKafka.sh
+++ b/installKafka.sh
@@ -84,20 +84,25 @@ $WSK_CLI -i --apihost "$EDGEHOST" action update --kind nodejs:default messaging/
-a parameters '[ {"name":"brokers", "required":true, "updatable":false, "description": "Array of Kafka brokers"}, {"name":"topic", "required":true, "updatable":false, "description": "Topic to subscribe to"}, {"name":"isJSONData", "required":false, "updatable":true, "description": "Attempt to parse message value as JSON"}, {"name":"isBinaryKey", "required":false, "updatable":true, "description": "Encode key as Base64"}, {"name":"isBinaryValue", "required":false, "updatable":true, "des [...]
-a sampleInput '{"brokers":"[\"127.0.0.1:9093\"]", "topic":"mytopic", "isJSONData":"false", "endpoint": "openwhisk.ng.bluemix.net"}'
-# create messagingWeb package and web version of feed action
+# create messagingWebDedicated package and web version of feed action
+$WSK_CLI -i --apihost "$EDGEHOST" package update messagingWebDedicated \
+ --auth "$AUTH" \
+ --shared no \
+ -p endpoint "$APIHOST"
+
+# rebind package
+$WSK_CLI -i --apihost "$EDGEHOST" package delete messagingWeb --auth "$AUTH"
if [ -n "$WORKERS" ];
then
- $WSK_CLI -i --apihost "$EDGEHOST" package update messagingWeb \
+ $WSK_CLI -i --apihost "$EDGEHOST" package bind messagingWebDedicated messagingWeb \
--auth "$AUTH" \
- --shared no \
-p endpoint "$APIHOST" \
-p DB_URL "$DB_URL" \
-p DB_NAME "$DB_NAME" \
-p workers "$WORKERS"
else
- $WSK_CLI -i --apihost "$EDGEHOST" package update messagingWeb \
+ $WSK_CLI -i --apihost "$EDGEHOST" package bind messagingWebDedicated messagingWeb \
--auth "$AUTH" \
- --shared no \
-p endpoint "$APIHOST" \
-p DB_URL "$DB_URL" \
-p DB_NAME "$DB_NAME"
@@ -120,7 +125,7 @@ zip -r kafkaFeedWeb.zip lib package.json kafkaFeedWeb.js node_modules
cd $OLD_PATH
-$WSK_CLI -i --apihost "$EDGEHOST" action update --kind nodejs:default messagingWeb/kafkaFeedWeb "$PACKAGE_HOME/action/kafkaFeedWeb.zip" \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --kind nodejs:default messagingWebDedicated/kafkaFeedWeb "$PACKAGE_HOME/action/kafkaFeedWeb.zip" \
--auth "$AUTH" \
--web true \
-a description 'Write a new trigger to Kafka provider DB' \