Logging should be implemented to not only help in debugging any issues that may arise, but also to help in performance analysis. Interesting events to log include:

• If an active consumer is programmatically shut down due to an error (e.g. 404 firing a trigger)
• When a consumer must retry firing a trigger
• Start and end times when successfully consuming a message
• Size of consumed messages (in aggregate per each run of poll() is probably fine)
• When consumers are successfully created or deleted by user request
• Authentication failures
• Existing consumers are recreated during start up

Allow the feed subscriber to pass common consumer configuration options.
https://github.com/openwhisk/openwhisk-package-kafka/blob/master/provider/consumer.py#L223-L236

Modifying the default consumer options enables lots of different usecases for this service. For example, I might want to turn off auto-commit or enable larger batch sizes for message polling.

This design continues the previous description of the message consumer service #19.

The provider service has two major responsibilities: monitoring the changes of the event source and sending the message into the messaging service.

Integration with the event sources:
The provider service provides different templates to monitor different event sources. For example, object storage has one template, the database has another template, etc. Users can launch the provider instance based on the template to monitor the changes of the event source. Since each provider instance serves one user for one particular event source, the user can define which trigger will be fire in each provider instance. When an event happens, the provider generates the message and send it to the Kafka service.

In order to generate the message consumed by the consumer service, the provider needs to be compliant with the following format:
ID: used to save the message ID
event_source: used to save the event source, which generates the change.
event_source: which type of the event should be specified.
trigger_id: trigger id or name
trigger_uri: uri of the trigger
namespace: namepsace of the trigger
apikey: authentication to fire the trigger
payload: a json format string as the input of the trigger or action.

When the user wants to listens a one event source like object storage for changes, he needs to choose the template of object storage provider, specify the trigger information and launch the instance.

The Kafka/Message Hub package should include an action that allows the caller to post a message to some topic in some Kafka or Message Hub instance.

The README must be updated to demonstrate to users how to build and deploy the service.

While it is not really possible to use the CLI to create duplicate triggers, the triggers PUT endpoint should guard against direct invocations that might try to create duplicate consumers. A simple check in the consumers dictionary should be sufficient.

Rather than have one feed action that can service both Message Hub and Kafka instances, let's split it into two separate actions, one for each type of instance.

This will make not only the naming of the actions clear (MessageHubFeed and KafkaFeed) but it will also make describing the parameters much simpler as well.

From what I understand, there is some black magic that needs to happen when the Message Hub feed action is installed. This will tell package refresh which bmx service to use as the basis for new package bindings.

Python errors occur when trying to post a trigger that already exists. We should ensure that the trigger does not exist before attempting to post it.

The feed package_endpoint parameter must be specified when deleting a trigger. Otherwise, the consumer will not be removed from the service.

Currently it is not possible to use the __OW_NAMESPACE to get the user namespace. When changes are made to the backend to allow this, the feeds should be updated to use it.

Currently, the feed actions have a hardcoded URL for the feed service provider. This should be parameterized so that this URL can be set during deployment.

Currently, authentication for the generic Kafka (not Message Hub) scenario is not supported. As such, there is no need to initialize this consumer with a username parameter.

Looks like we need to have the log file in a certain format in order to be picked up by the log forwarder and eventually viewable in Kibana. Here is the format for the Kafka logs:

kafkaTriggers_log.log

here is how I think they are currently being stored as:

kafkatriggers.log

You also need to print the lines in a format that our current logstash with forwarder can parsed them and send them to ELK

When initializing the KafkaConsumer for Message Hub, the auto commit flag is set to True even though we definitely want to manually commit offsets only after successfully firing the trigger.

I happened to be running the feed service on my laptop when the network completely went out (I pulled the plug). As you would expect, the one consumer I had started freaking out at not being able to connect to Message Hub. However, it also did not recover once the network connection was restored.

The following messages were repeated indefinitely until I restarted the entire service:

WARNING:kafka.client:Node 1 connection failed -- refreshing metadata
ERROR:kafka.coordinator:Error sending GroupCoordinatorRequest_v0 to node 1 [NodeNotReadyError: 1]
WARNING:kafka.coordinator:Coordinator unknown during heartbeat -- will retry
WARNING:kafka.coordinator:Heartbeat failed ([Error 15] GroupCoordinatorNotAvailableError: ); retrying

Positing the same triggerURL using different HTTP routes should not be allowed. This results in the trigger for one consumer firing when multiple of the same triggerURLs exist.

If there is an error in a feed action, a string is printed as the response body. The information should, however; be displayed as JSON instead of a string.

Actual output:

"error": {
     "response": "{\n  \"error\": \"not authorized\", \n  \"success\": false\n}\n",
     "statusCode": 401
}

Expected output:

"error": {
     "response": {
          "error":  "not authorized",
          "success": false
     },
     "statusCode": 401
}

Implement a build system based on gradle. This system should be capable of:

• building the Docker image
• starting a Docker container with test-only Cloudant credentials
• running automated tests against that running container

Often when developing/testing locally, OpenWhisk uses a self-signed certificate. This causes problems for this feed provider which (by default) insists on checking the validity of the SSL cert. With checking enabled during local development, GET and POST calls to OpenWhisk fail.

Fix will be to optionally disable SSL cert checking, controlled by the use of an environment variable.

To ease deployment, the Cloudant DB/table should be created if it does not already exist.

It has been observed that a Consumer will get into a state where it is indefinitely repeating "socket disconnected" messages to the log. In the meantime, it is not consuming new messages.

This problem seems to come from the kafka-python client itself, where (I suppose) it is getting into an error-recovery loop.

When a trigger exists and the corresponding Message Hub instance is deleted, I believe an exception is thrown (likely from poll())and the consumer thread crashes. While the rest of the service continues, this is a pretty clumsy way to handle the situation.

When this happens, the exception should be caught and message logged, and the consumer should be shut down and cleaned up gracefully.

In the future, there may be a way to indicate this back to the OpenWhisk backend, but for now I think this should suffice.

When firing a trigger, there are some clear error cases where we want to simply retry firing the trigger. For example, when the user is throttled by firing triggers too quickly, the return code is 429. In this case, it is sufficient to pause and retry firing the trigger.

However, other error conditions warrant completely stopping/deleting the consumer entirely. For example if a 404 is hit when firing the trigger, it is clear that the trigger has been deleted and the consumer should be stopped entirely.

There are certainly other possible errors when firing triggers, which should each be handled appropriately with either a retry or complete abort.

Currently, the /health endpoint shows a list of all triggers by fully qualified name. The trouble is that the namespace value can often include personal information as they tend to be based on the user's email address.

One solution could be to generate a UUID for each consumer and store that in the DB. Then, the health endpoint could replace the namespace in each trigger's FQN with the UUID.

Exceptions occur if all the required fields needed to post a trigger are not provided by the client. These exception must be caught, and an appropriate error message must be sent back to the client.

api_key == username + password which means that asking for all three as parameters on the Message Hub feed action is just redundant.

In cases where the consumer can no longer continue (message hub instance removed, authentication changed, etc.) the last thing the consumer should do before shutting down is to post to the trigger URL. The body of this post should include useful information to inform the OW user that their trigger is being disabled and, if possible, provide a reason why.

We need an automated test that will be used to monitor the health of the system. This test should GET the /health endpoint and parse the output for any signs of sickness.

I am experiencing a situation where the time to post a trigger to OW is taking more than 2 seconds. Since this time contributes to the entire poll() time, simply posting a trigger results in a poll timeout and consumer restart.

Increase the timeout to something more reasonable.

Consumers that have invalidated Message Hub credential, have a trigger that is deleted, or fail to invoke a trigger more than ten times must be removed from the main dictionary of consumers. Otherwise, these consumers will exist until the Docker container is restarted.

This default route can/will be used by deployment/monitoring scripts to ensure that the service is alive (in the most basic way).

I have confirmed an instance where a single consumer stopped polling due to issues communicating with Kafka and the last successful poll time was significantly greater than the regular poll interval. This proves that a single consumer crash can be detected with this measurement.

Knowing this, we now need to gracefully handle this scenario and restart the consumer to recover from the error.

As the code currently exists, there is potentially no limit to the size of the payload we will attempt to post to a trigger. This is bad.

I believe there are limits we may be able to set with consumer parameters that may do this automatically for us (see max_partition_fetch_bytes though this may involve some partition math). However, failing that, we can examine the size of each consumed message and use that to actively limit the amount of data posted to the trigger.

http://kafka-python.readthedocs.io/en/master/apidoc/KafkaConsumer.html

Although kafka message content is, strictly speaking, always a byte array, it seems that JSON formatted strings will be so common that it may warrant special handling.

It has been proposed that if the trigger creator specifies that the message format is a JSON formatted string (using a new isJSONData parameter to the feed action), then the feed service should make a reasonable attempt at parsing the message content and passing it appropriately as the trigger payload.

If isJSONData is set to true and the message cannot be parsed as valid JSON, the result is the same as if isJSONData were set to false - which is the current behavior. In this case, the feed service should log a warning message.

@csantanapr

It appears there was either a network glitch or the kafka instance was unavailable during the poll. That being said, our code should handle this situation more gracefully.

Exception in thread Thread-1:
Traceback (most recent call last):
  File "/usr/local/lib/python2.7/threading.py", line 801, in __bootstrap_inner
    self.run()
  File "/KafkaFeedProvider/consumer.py", line 89, in run
    partition = self.consumer.poll(1000)
  File "/usr/local/lib/python2.7/site-packages/kafka/consumer/group.py", line 514, in poll
    records = self._poll_once(remaining)
  File "/usr/local/lib/python2.7/site-packages/kafka/consumer/group.py", line 541, in _poll_once
    self._coordinator.ensure_coordinator_known()
  File "/usr/local/lib/python2.7/site-packages/kafka/coordinator/base.py", line 221, in ensure_coordinator_known
    self._client.poll(future=metadata_update)
  File "/usr/local/lib/python2.7/site-packages/kafka/client_async.py", line 483, in poll
    self._maybe_connect(node_id)
  File "/usr/local/lib/python2.7/site-packages/kafka/client_async.py", line 322, in _maybe_connect
    conn.connect()
  File "/usr/local/lib/python2.7/site-packages/kafka/conn.py", line 304, in connect
    if self._try_handshake():
  File "/usr/local/lib/python2.7/site-packages/kafka/conn.py", line 366, in _try_handshake
    self._sock.do_handshake()
  File "/usr/local/lib/python2.7/ssl.py", line 830, in do_handshake
    self._sslobj.do_handshake()
error: [Errno 110] Operation timed out

Implement an endpoint that returns data regarding the health of the feed service provider. Information should include things like:

• average CPU load (past hour? array of averages for each of the past 24 hours?)
• memory usage
• number of active consumers

The caller of this endpoint is interested in determining how much work is being done by the feed service, and how much more work can it do before it is overloaded.

In order to avoid creating consumers for unauthorized trigger URLs, before creating a new consumer we should somehow authenticate with the OpenWhisk system using the credentials given for the trigger.

Have the feed action go get the value for namespace if the one pass as parameter is _

We are instructing users to unset the namespace here:
https://console.ng.bluemix.net/openwhisk/learn/cli

wsk property unset --namespace

The feed action will need to handle when user uses the new auth key method and no namespace is configured in the CLI.

Options:

1. Have the feed action do a a OW API call to get the namespace for the authkey
2. Have the runtime provide the namespace string associated with the authkey being used to invoke the feedaction. Today there is a namespace environment variable but it doesn't have the value that we need it has the value of the owner of the action not the namespace associated with the authkey.

We will use this issue to track which tests need to be create.

• Ensure proper error messages and status codes are returned when creating Kafka, and Message Hub tests with missing or invalid parameters. This will need to be done through creating feeds and posting triggers directly to the container.

• Do not allow duplicate triggers in the database (Whisk won't allow duplicate trigger creation)

• Do not allow a trigger that does not exist to be delete

• Allow for deletion of a trigger that exists

• Ensure a user has access to a trigger before creating it

• Ensure a user has access to a trigger before deleting it (Whisk doesn't allow trigger access to unauthenticated triggers)

• Have a consumer encounter an HTTP status code other than 503, 500, 408, 429, or 200 to ensure retries occur in the polling process and that the consumer recovers on purpose

• Have a consumer encounter an HTTP status code other than 503, 500, 408, 429, or 200 to ensure retries occur in the polling process and that the consumer gets shutdown on purpose

• Have the consumer encounter 503, 500, 408, and 429 error codes to make sure the consumer is shutdown without retries

• Post triggers to the service, shutdown the Docker container then restart it to make sure everything recovers properly

• Ensure doctor restarts a consumer that polls for too long (ypKafkaHeartBeat, ys1KafkaHeartBeat)

the actions in this package use the now deprecated whisk object - migration to promises (and npm openwhisk) should be in order.

OpenWhisk manages different components in different docker containers, each of whom is dedicated to host one service taking care of the workload for all the users. The current in-progress package messaging(kafka) service, designed in the same model, launches one docker container to manage all the consumers and triggers created by all the users, etc. The major pitfall of this design is that it restrains all the resources in one container and directly leads to single point of failure and lack of scalability.

Digging into the code a little bit, we can see a internal hashmap or dict is created to save all the consumers and mapping relation with the triggers, and each new consumer/new trigger has to create a new record in the database for consistency and recovery if there is a demand for the service restart.

The following issues are what we can predict with the current design:

• The container becomes a single point of failure. If this service container crashes, all the services are shut down. No service HA can be insured.
• If we have to restart the service container, we have to load everything available saved in the database, to make the service back online, e.g. each time the service provider or the package messaging(kafka) service, all the triggers for every user have to be reloaded and reinitialized. What if the database information is lost? We lose all the information of the consumers and triggers. No service HA can be insured.
• No matter how much resource we can allocate to this container, it is limited for sure, so that the container will be drained if too many triggers have been created. We have already run into this issue badly, since openwhisk developers have been told to check the maximum number of triggers that can be created with certain resources. We are running a service with limit, which is not a model able to run in cloud.

If the triggers exceeds this number, what can we do to this situation without changing the implementation? Some are possible, but no ideal solution at this moment.

• We can add a new consumer service to listen to the messagehub. The drawback is that we only have a limited topic to listen to for the kafka message. Even if we add one more consumer, we cannot make sure that the triggers over 1200 created in the second consumer pick up the message, because both the two consumers listen to the the same topic. Also, each consumer service maintains a separate and different internal map, and they are still single points of failure. If we lose one of them, we lose all the information in the internal map. Third, how can we control the message is consumed by the consumer service, which has the internal map we need? If the message needs to be consumed by trigger A, how do we know it can go to the consumer service with the mapping information for trigger A? Very difficult. We can send the message to all the consumer services, but in that case it is a waste of resources.
• We can add new messagehub service. In this case, we will end up managing multiple messaging services. It will be very complex to manage which trigger/consumer maps to which message hub.

We can see that the internal map of the registry is really an issue, which limits the number triggers we are able to handle, and make the service rather difficult to scale.

In order to resolve the above issues, we have to reconsider the design model for the message consumer service. I would like to propose a more lightweight model, in which, the consumer service only takes care of listening to the messaging service, picking up the message from the specific topic and consuming it. We are able to launch more consumer services in HA mode, since they have the same responsibilities.

Where do we get the information of the trigger to be fired?
From openwhisk's perspective, we are already able to create the actions and triggers, and associate a rule between them, so the only thing that the consumer service is to get the information for the trigger. Instead of maintaining the relationship between the consumer and the trigger in the internal map/dict and the database. We can offload the information of the trigger to provider service. When the provider discovers a change from the event source, it sends a message with the trigger information, e.g. the trigger ID, trigger uri, etc, and credentials and the payload of the final action into the messaging service. When the consumer service receives it, simply fire the trigger by the uri with the payload.

In this model, the consumer service has been tailored into a lightweight module, which listens to the topic, receives the message and fires the trigger with the payload and credentials, since all the information is prepared by the provider service and available in the message. There is no database involved and no specific information to be saved. The service is able to scale, since we can launch the consumer service as many as we need. As long as there is one consumer service running, our service can guarantee the availability.

The only thing we need to define is the message format to be consumed by the consumer service. I will describe the message format in details for the design of the provider service: #20.

In order to make the feed action deployment more flexible, the feed action should tolerate it when the apihost already contains the protocol (e.g. starts with "https://")

The two feed actions contain a lot of duplicate code. These should be refactored so that common code can go into a common file that is shared between the two actions. This will require deploying the feed actions as zip files.

I ran into a situation where there was a broken pipe posting a trigger to OW. However, it took an extremely long time (maybe a minute or more?) before the post failed. We should set an explicit timeout on the post to keep this to a reasonable time.

The namespace needs to be a unique string, it can be the openwhisk _ default

This is required to create a unique DB id for the trigger when store in the provider DB

    var namespace = encodeURIComponent(triggerComponents[1]);
``
Need to reject/error the action feed if namespace is not a string, and can't be `_`
For both actions:
https://github.com/openwhisk/openwhisk-package-kafka/blob/master/action/kafkaFeed.js#L17
https://github.com/openwhisk/openwhisk-package-kafka/blob/master/action/kafkaFeed.js#L17

Right now, logged messages are completely interleaved and it is hard to follow individual events through the system (creating a trigger, responding to a message, etc.).

If the logged messages used a transaction ID the way the OpenWhisk core system does, it would be a lot easier to follow events through the log.

An error will occur while attempting to authenticate a trigger if the host does not response. This error should be caught, and a proper response sent back to the user.