add support for Kinesis targets in EventBridge (localstack#3925)

Author: whummer

.dockerignore

@@ -1,6 +1,5 @@
 .venv*
 localstack/dashboard/web/node_modules/
-localstack/ext/java/target
 localstack/infra/
 !localstack/infra/stepfunctions
 !localstack/infra/localstack-*.jar

.env-exporter

@@ -0,0 +1,12 @@
+# ~/.env-exporter
+# about zlib
+export CFLAGS="-I$(xcrun --show-sdk-path)/usr/include"
+# about readline
+export CFLAGS="-I$(brew --prefix readline)/include $CFLAGS"
+export LDFLAGS="-L$(brew --prefix readline)/lib $LDFLAGS"
+# about openssl
+export CFLAGS="-I$(brew --prefix openssl)/include $CFLAGS"
+export LDFLAGS="-L$(brew --prefix openssl)/lib $LDFLAGS"
+# about SQLite (maybe not necessary)
+export CFLAGS="-I$(brew --prefix sqlite)/include $CFLAGS"
+export LDFLAGS="-L$(brew --prefix sqlite)/lib $LDFLAGS"

.gitignore

@@ -6,7 +6,6 @@
 
 /infra/
 localstack/infra/
-localstack/ext/java/.factorypath
 
 /node_modules/
 package-lock.json

Dockerfile

@@ -34,6 +34,7 @@ ADD localstack/utils/common.py localstack/utils/bootstrap.py localstack/utils/
 ADD localstack/utils/aws/ localstack/utils/aws/
 ADD localstack/utils/kinesis/ localstack/utils/kinesis/
 ADD localstack/utils/analytics/ localstack/utils/analytics/
+ADD localstack/utils/generic/ localstack/utils/generic/
 ADD localstack/package.json localstack/package.json
 ADD localstack/services/__init__.py localstack/services/install.py localstack/services/
 

README.md

@@ -94,7 +94,7 @@ In addition to the above, the [**Pro version** of LocalStack](https://localstack
 
 ## Requirements
 
-* `python` (both Python 2.x and 3.x supported)
+* `python` (Python 2.x up to 3.8 supported)
 * `pip` (python package manager)
 * `Docker`
 

bin/mvn_release.sh

@@ -66,18 +66,19 @@ def filter_event(event_pattern, event):
         return True
 
     rule_information = self.events_backend.describe_rule(rule)
-    if rule_information.event_pattern:
-        event_pattern = rule_information.event_pattern._filter
-        if event_pattern and not filter_event(event_pattern, event):
-            return False
+    rule_event_pattern = json.loads(str(rule_information.event_pattern))
+
+    if rule_event_pattern and not filter_event(rule_event_pattern, event):
+        return False
+
     return True
 
 
 def process_events(event, targets):
     for target in targets:
         arn = target['Arn']
         changed_event = filter_event_with_target_input_path(target, event)
-        aws_stack.send_event_to_target(arn, changed_event)
+        aws_stack.send_event_to_target(arn, changed_event, aws_stack.get_events_target_attributes(target))
 
 
 def apply_patches():

localstack/services/events/events_starter.py

@@ -12,6 +12,7 @@
     APPLICATION_AMZ_JSON_1_0, APPLICATION_X_WWW_FORM_URLENCODED, TEST_AWS_ACCOUNT_ID,
     MAX_POOL_CONNECTIONS, TEST_AWS_ACCESS_KEY_ID, TEST_AWS_SECRET_ACCESS_KEY, S3_VIRTUAL_HOSTNAME)
 from localstack.utils.aws import templating
+from localstack.utils.generic import dict_utils
 from localstack.utils.common import (
     run_safe, to_str, is_string, is_string_or_bytes, make_http_request,
     is_port_open, get_service_protocol, retry, to_bytes)
@@ -47,6 +48,9 @@
 # maps SQS queue ARNs to queue URLs
 SQS_ARN_TO_URL_CACHE = {}
 
+# List of parameters with additional event target parameters
+EVENT_TARGET_PARAMETERS = ['$.SqsParameters', '$.KinesisParameters']
+
 
 class Environment(object):
     def __init__(self, region=None, prefix=None):
@@ -573,7 +577,7 @@ def send_event_to_target(arn, event, target_attributes=None, asynchronous=True):
     elif ':sqs:' in arn:
         sqs_client = connect_to_service('sqs', region_name=region)
         queue_url = get_sqs_queue_url(arn)
-        msg_group_id = (target_attributes or {}).get('MessageGroupId')
+        msg_group_id = dict_utils.get_safe(target_attributes, '$.SqsParameters.MessageGroupId')
         kwargs = {'MessageGroupId': msg_group_id} if msg_group_id else {}
         sqs_client.send_message(QueueUrl=queue_url, MessageBody=json.dumps(event), **kwargs)
 
@@ -600,13 +604,29 @@ def send_event_to_target(arn, event, target_attributes=None, asynchronous=True):
             }]
         )
 
+    elif ':kinesis:' in arn:
+        partition_key_path = dict_utils.get_safe(
+            target_attributes,
+            '$.KinesisParameters.PartitionKeyPath',
+            default_value='$.id'
+        )
+
+        stream_name = arn.split('/')[-1]
+        partition_key = dict_utils.get_safe(event, partition_key_path, event['id'])
+        kinesis_client = connect_to_service('kinesis', region_name=region)
+
+        kinesis_client.put_record(
+            StreamName=stream_name,
+            Data=to_bytes(json.dumps(event)),
+            PartitionKey=partition_key
+        )
+
     else:
         LOG.warning('Unsupported Events rule target ARN: "%s"' % arn)
 
 
 def get_events_target_attributes(target):
-    # TODO: add support for other target types
-    return target.get('SqsParameters')
+    return dict_utils.pick_attributes(target, EVENT_TARGET_PARAMETERS)
 
 
 def get_or_create_bucket(bucket_name):

localstack/utils/aws/aws_stack.py

@@ -0,0 +1,107 @@
+def get_safe(dictionary, path, default_value=None):
+    """
+    Performs a safe navigation on a Dictionary object and
+    returns the result or default value (if specified).
+    The function follows a common AWS path resolution pattern "$.a.b.c".
+
+    :type dictionary: dict
+    :param dictionary: Dict to perform safe navigation.
+
+    :type path: list|str
+    :param path: List or dot-separated string containing the path of an attribute,
+                 starting from the root node "$".
+
+    :type default_value: any
+    :param default_value: Default value to return in case resolved value is None.
+
+    :rtype: any
+    :return: Resolved value or default_value.
+    """
+    if not isinstance(dictionary, dict) or len(dictionary) == 0:
+        return default_value
+
+    attribute_path = path if isinstance(path, list) else path.split('.')
+    if len(attribute_path) == 0 or attribute_path[0] != '$':
+        raise AttributeError('Safe navigation must begin with a root node "$"')
+
+    current_value = dictionary
+    for path_node in attribute_path:
+        if path_node == '$':
+            continue
+        elif isinstance(current_value, dict) and path_node in current_value:
+            current_value = current_value[path_node]
+        else:
+            current_value = None
+
+    return current_value or default_value
+
+
+def set_safe_mutable(dictionary, path, value):
+    """
+    Mutates original dict and sets the specified value under provided path.
+
+    :type dictionary: dict
+    :param dictionary: Dict to mutate.
+
+    :type path: list|str
+    :param path: List or dot-separated string containing the path of an attribute,
+                 starting from the root node "$".
+
+    :type value: any
+    :param value: Value to set under specified path.
+
+    :rtype: dict
+    :return: Returns mutated dictionary.
+    """
+    if not isinstance(dictionary, dict):
+        raise AttributeError('"dictionary" must be of type "dict"')
+
+    attribute_path = path if isinstance(path, list) else path.split('.')
+    attribute_path_len = len(attribute_path)
+
+    if attribute_path_len == 0 or attribute_path[0] != '$':
+        raise AttributeError('Dict navigation must begin with a root node "$"')
+
+    current_pointer = dictionary
+    for i in range(attribute_path_len):
+        path_node = attribute_path[i]
+
+        if path_node == '$':
+            continue
+
+        if i < attribute_path_len - 1:
+            if path_node not in current_pointer:
+                current_pointer[path_node] = {}
+            if not isinstance(current_pointer, dict):
+                raise RuntimeError('Error while deeply setting a dict value. Supplied path is not of type "dict"')
+        else:
+            current_pointer[path_node] = value
+
+        current_pointer = current_pointer[path_node]
+
+    return dictionary
+
+
+def pick_attributes(dictionary, paths):
+    """
+    Picks selected attributes a returns them as a new dictionary.
+    This function works as a whitelist of attributes to keep in a new dictionary.
+
+    :type dictionary: dict
+    :param dictionary: Dict to pick attributes from.
+
+    :type paths: list of (list or str)
+    :param paths: List of lists or strings with dot-separated paths, starting from the root node "$".
+
+    :rtype: dict
+    :return: Returns whitelisted dictionary.
+    """
+    new_dictionary = {}
+
+    for path in paths:
+        value = get_safe(dictionary, path)
+
+        if value is not None:
+            set_safe_mutable(new_dictionary, path, value)
+
+    return new_dictionary