You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@metron.apache.org by ni...@apache.org on 2016/11/16 22:05:59 UTC
incubator-metron git commit: METRON-557 Create Stellar Functions for
Kafka (nickwallen) closes apache/incubator-metron#354
Repository: incubator-metron
Updated Branches:
refs/heads/master d03a363f8 -> dbb56cb5e
METRON-557 Create Stellar Functions for Kafka (nickwallen) closes apache/incubator-metron#354
Project: http://git-wip-us.apache.org/repos/asf/incubator-metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-metron/commit/dbb56cb5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-metron/tree/dbb56cb5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-metron/diff/dbb56cb5
Branch: refs/heads/master
Commit: dbb56cb5e85425e75970cffb62593e5ef7f90578
Parents: d03a363
Author: nickwallen <ni...@nickallen.org>
Authored: Wed Nov 16 17:05:31 2016 -0500
Committer: Nick Allen <ni...@nickallen.org>
Committed: Wed Nov 16 17:05:31 2016 -0500
----------------------------------------------------------------------
metron-platform/metron-common/README.md | 344 ++++++++++----
.../metron/integration/BaseIntegrationTest.java | 2 +-
metron-platform/metron-management/pom.xml | 35 ++
.../metron/management/KafkaFunctions.java | 452 +++++++++++++++++++
.../KafkaFunctionsIntegrationTest.java | 185 ++++++++
5 files changed, 938 insertions(+), 80 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/dbb56cb5/metron-platform/metron-common/README.md
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/README.md b/metron-platform/metron-common/README.md
index c9b2506..1444345 100644
--- a/metron-platform/metron-common/README.md
+++ b/metron-platform/metron-common/README.md
@@ -1,4 +1,15 @@
-#Stellar Language
+
+
+# Contents
+
+* [Stellar Language](#stellar-language)
+ * [Stellar Language Keywords](#stellar-language-keywords)
+ * [Stellar Core Functions](#stellar-core-functions)
+ * [Stellar Shell](#stellar-shell)
+* [Global Configuration](#global-configuration)
+* [Management Utility](#management-utility)
+
+# Stellar Language
For a variety of components (threat intelligence triage and field
transformations) we have the need to do simple computation and
@@ -16,7 +27,7 @@ The query language supports the following:
* The ability to have parenthesis to make order of operations explicit
* User defined functions
-##Stellar Language Keywords
+## Stellar Language Keywords
The following keywords need to be single quote escaped in order to be used in Stellar expressions:
| | | |
@@ -28,70 +39,157 @@ The following keywords need to be single quote escaped in order to be used in St
| ? | \+ | \- |
| , | \* | / |
-such as: "foo" : "\<ok\>" requires escaping such as "foo": "\'\<ok\>\'"
-
-The following functions are supported:
-
-* `BLOOM_ADD`
+Using parens such as: "foo" : "\<ok\>" requires escaping; "foo": "\'\<ok\>\'"
+
+## Stellar Core Functions
+
+| |
+| ---------- |
+| [ `BLOOM_ADD`](#bloom_add)|
+| [ `BLOOM_EXISTS`](#bloom_exists)|
+| [ `BLOOM_INIT`](#bloom_init)|
+| [ `BLOOM_MERGE`](#bloom_merge)|
+| [ `DAY_OF_MONTH`](#day_of_month)|
+| [ `DAY_OF_WEEK`](#day_of_week)|
+| [ `DAY_OF_YEAR`](#day_of_year)|
+| [ `DOMAIN_REMOVE_SUBDOMAINS`](#domain_remove_subdomains)|
+| [ `DOMAIN_REMOVE_TLD`](#domain_remove_tld)|
+| [ `DOMAIN_TO_TLD`](#domain_to_tld)|
+| [ `ENDS_WITH`](#ends_with)|
+| [ `ENRICHMENT_EXISTS`](#enrichment_exists)|
+| [ `ENRICHMENT_GET`](#enrichment_get)|
+| [ `GET`](#get)|
+| [ `GET_FIRST`](#get_first)|
+| [ `GET_LAST`](#get_last)|
+| [ `IN_SUBNET`](#in_subnet)|
+| [ `IS_DATE`](#is_date)|
+| [ `IS_DOMAIN`](#is_domain)|
+| [ `IS_EMAIL`](#is_email)|
+| [ `IS_EMPTY`](#is_empty)|
+| [ `IS_INTEGER`](#is_integer)|
+| [ `IS_IP`](#is_ip)|
+| [ `IS_URL`](#is_url)|
+| [ `JOIN`](#join)|
+| [ `KAFKA_GET`](#kafka_get)|
+| [ `KAFKA_PROPS`](#kafka_props)|
+| [ `KAFKA_PUT`](#kafka_put)|
+| [ `KAFKA_TAIL`](#kafka_tail)|
+| [ `LENGTH`](#length)|
+| [ `MAAS_GET_ENDPOINT`](#maas_get_endpoint)|
+| [ `MAAS_MODEL_APPLY`](#maas_model_apply)|
+| [ `MAP_EXISTS`](#map_exists)|
+| [ `MONTH`](#month)|
+| [ `PROFILE_GET`](#profile_get)|
+| [ `PROTOCOL_TO_NAME`](#protocol_to_name)|
+| [ `REGEXP_MATCH`](#regexp_match)|
+| [ `SPLIT`](#split)|
+| [ `STARTS_WITH`](#starts_with)|
+| [ `STATS_ADD`](#stats_add)|
+| [ `STATS_COUNT`](#stats_count)|
+| [ `STATS_GEOMETRIC_MEAN`](#stats_geometric_mean)|
+| [ `STATS_INIT`](#stats_init)|
+| [ `STATS_KURTOSIS`](#stats_kurtosis)|
+| [ `STATS_MAX`](#stats_max)|
+| [ `STATS_MEAN`](#stats_mean)|
+| [ `STATS_MERGE`](#stats_merge)|
+| [ `STATS_MIN`](#stats_min)|
+| [ `STATS_PERCENTILE`](#stats_percentile)|
+| [ `STATS_POPULATION_VARIANCE`](#stats_population_variance)|
+| [ `STATS_QUADRATIC_MEAN`](#stats_quadriatic_mean)|
+| [ `STATS_SD`](#stats_sd)|
+| [ `STATS_SKEWNESS`](#stats_skewness)|
+| [ `STATS_SUM`](#stats_sum)|
+| [ `STATS_SUM_LOGS`](#stats_sum_logs)|
+| [ `STATS_SUM_SQUARES`](#stats_sum_squares)|
+| [ `STATS_VARIANCE`](#stats_variance)|
+| [ `SYSTEM_ENV_GET`](#stats_env_get)|
+| [ `SYSTEM_PROPERTY_GET`](#system_property_get)|
+| [ `TO_DOUBLE`](#to_double)|
+| [ `TO_EPOCH_TIMESTAMP`](#to_epoch_timestamp)|
+| [ `TO_INTEGER`](#to_integer)|
+| [ `TO_LOWER`](#to_lower)|
+| [ `TO_STRING`](#to_string)|
+| [ `TO_UPPER`](#to_upper)|
+| [ `TRIM`](#trim)|
+| [ `URL_TO_HOST`](#url_to_host)|
+| [ `URL_TO_PATH`](#url_to_path)|
+| [ `URL_TO_PORT`](#url_to_port)|
+| [ `URL_TO_PROTOCOL`](#url_to_protocol)|
+| [ `WEEK_OF_MONTH`](#week_of_month)|
+| [ `WEEK_OF_YEAR`](#week_of_year)|
+| [ `YEAR`](#year)|
+
+### `BLOOM_ADD`
* Description: Adds an element to the bloom filter passed in
* Input:
* bloom - The bloom filter
* value* - The values to add
* Returns: Bloom Filter
-* `BLOOM_EXISTS`
+
+### `BLOOM_EXISTS`
* Description: If the bloom filter contains the value
* Input:
* bloom - The bloom filter
* value - The value to check
* Returns: True if the filter might contain the value and false otherwise
-* `BLOOM_INIT`
+
+### `BLOOM_INIT`
* Description: Returns an empty bloom filter
* Input:
* expectedInsertions - The expected insertions
* falsePositiveRate - The false positive rate you are willing to tolerate
* Returns: Bloom Filter
-* `BLOOM_MERGE`
+
+### `BLOOM_MERGE`
* Description: Returns a merged bloom filter
* Input:
* bloomfilters - A list of bloom filters to merge
* Returns: Bloom Filter or null if the list is empty
-* `DAY_OF_MONTH`
+
+### `DAY_OF_MONTH`
* Description: The numbered day within the month. The first day within the month has a value of 1.
* Input:
* dateTime - The datetime as a long representing the milliseconds since unix epoch
* Returns: The numbered day within the month.
-* `DAY_OF_WEEK`
+
+### `DAY_OF_WEEK`
* Description: The numbered day within the week. The first day of the week, Sunday, has a value of 1.
* Input:
* dateTime - The datetime as a long representing the milliseconds since unix epoch
* Returns: The numbered day within the week.
-* `DAY_OF_YEAR`
+
+### `DAY_OF_YEAR`
* Description: The day number within the year. The first day of the year has value of 1.
* Input:
* dateTime - The datetime as a long representing the milliseconds since unix epoch
* Returns: The day number within the year.
-* `DOMAIN_REMOVE_SUBDOMAINS`
+
+### `DOMAIN_REMOVE_SUBDOMAINS`
* Description: Removes the subdomains from a domain.
* Input:
* domain - Fully qualified domain name
* Returns: The domain without the subdomains. (for example, DOMAIN_REMOVE_SUBDOMAINS('mail.yahoo.com') yields 'yahoo.com')
-* `DOMAIN_REMOVE_TLD`
+
+### `DOMAIN_REMOVE_TLD`
* Description: Removes the top level domain (TLD) suffix from a domain.
* Input:
* domain - Fully qualified domain name
* Returns: The domain without the TLD. (for example, DOMAIN_REMOVE_TLD('mail.yahoo.co.uk') yields 'mail.yahoo')
-* `DOMAIN_TO_TLD`
+
+### `DOMAIN_TO_TLD`
* Description: Extracts the top level domain from a domain
* Input:
* domain - Fully qualified domain name
* Returns: The TLD of the domain. (for example, DOMAIN_TO_TLD('mail.yahoo.co.uk') yields 'co.uk')
-* `ENDS_WITH`
+
+### `ENDS_WITH`
* Description: Determines whether a string ends with a specified suffix
* Input:
* string - The string to test
* suffix - The proposed suffix
* Returns: True if the string ends with the specified suffix and false if otherwise
-* `ENRICHMENT_EXISTS`
+
+### `ENRICHMENT_EXISTS`
* Description: Interrogates the HBase table holding the simple hbase enrichment data and returns whether the enrichment type and indicator are in the table.
* Input:
* enrichment_type - The enrichment type
@@ -99,7 +197,8 @@ The following functions are supported:
* nosql_table - The NoSQL Table to use
* column_family - The Column Family to use
* Returns: True if the enrichment indicator exists and false otherwise
-* `ENRICHMENT_GET`
+
+### `ENRICHMENT_GET`
* Description: Interrogates the HBase table holding the simple hbase enrichment data and retrieves the tabular value associated with the enrichment type and indicator.
* Input:
* enrichment_type - The enrichment type
@@ -107,108 +206,156 @@ The following functions are supported:
* nosql_table - The NoSQL Table to use
* column_family - The Column Family to use
* Returns: A Map associated with the indicator and enrichment type. Empty otherwise.
-* `GET`
+
+### `GET`
* Description: Returns the i'th element of the list
* Input:
* input - List
* i - The index (0-based)
* Returns: First element of the list
-* `GET_FIRST`
+
+### `GET_FIRST`
* Description: Returns the first element of the list
* Input:
* input - List
* Returns: First element of the list
-* `GET_LAST`
+
+### `GET_LAST`
* Description: Returns the last element of the list
* Input:
* input - List
* Returns: Last element of the list
-* `IN_SUBNET`
+
+### `IN_SUBNET`
* Description: Returns true if an IP is within a subnet range.
* Input:
* ip - The IP address in string form
* cidr+ - One or more IP ranges specified in CIDR notation (for example 192.168.0.0/24)
* Returns: True if the IP address is within at least one of the network ranges and false if otherwise
-* `IS_DATE`
+
+### `IS_DATE`
* Description: Determines if the date contained in the string conforms to the specified format.
* Input:
* date - The date in string form
* format - The format of the date
* Returns: True if the date is in the specified format and false if otherwise.
-* `IS_DOMAIN`
+
+### `IS_DOMAIN`
* Description: Tests if a string refers to a valid domain name. Domain names are evaluated according to the standards RFC1034 section 3, and RFC1123 section 2.1.
* Input:
* address - The string to test
* Returns: True if the string refers to a valid domain name and false if otherwise
-* `IS_EMAIL`
+
+### `IS_EMAIL`
* Description: Tests if a string is a valid email address
* Input:
* address - The string to test
* Returns: True if the string is a valid email address and false if otherwise.
-* `IS_EMPTY`
+
+### `IS_EMPTY`
* Description: Returns true if string or collection is empty or null and false if otherwise.
* Input:
* input - Object of string or collection type (for example, list)
* Returns: True if the string or collection is empty or null and false if otherwise.
-* `IS_INTEGER`
+
+### `IS_INTEGER`
* Description: Determines whether or not an object is an integer.
* Input:
* x - The object to test
* Returns: True if the object can be converted to an integer and false if otherwise.
-* `IS_IP`
+
+### `IS_IP`
* Description: Determine if an string is an IP or not.
* Input:
* ip - An object which we wish to test is an ip
* type (optional) - Object of string or collection type (e.g. list) one of IPV4 or IPV6 or both. The default is IPV4.
* Returns: True if the string is an IP and false otherwise.
-* `IS_URL`
+
+### `IS_URL`
* Description: Tests if a string is a valid URL
* Input:
* url - The string to test
* Returns: True if the string is a valid URL and false if otherwise.
-* `JOIN`
+
+### `JOIN`
* Description: Joins the components in the list of strings with the specified delimiter.
* Input:
* list - List of strings
* delim - String delimiter
* Returns: String
-* `LENGTH`
+
+### `KAFKA_GET`
+ * Description: Retrieves messages from a Kafka topic. Subsequent calls will continue retrieving messages sequentially from the original offset.
+ * Input:
+ * topic - The name of the Kafka topic.
+ * count - The number of Kafka messages to retrieve.
+ * config - Optional map of key/values that override any global properties.
+ * Returns: List of String
+
+### `KAFKA_PROPS`
+ * Description: Retrieves the Kafka properties that are used by other KAFKA_* functions like KAFKA_GET and KAFKA_PUT. The Kafka properties are compiled from a set of default properties, the global properties, and any overrides.
+ * Input:
+ * config - An optional map of key/values that override any global properties.
+ * Returns: Map of key/value pairs
+
+### `KAFKA_PUT`
+ * Description: Sends messages to a Kafka topic.
+ * Input:
+ * topic - The name of the Kafka topic.
+ * messages - A list of messages to write.
+ * config - Optional map of key/values that override any global properties.
+ * Returns: n/a
+
+### `KAFKA_TAIL`
+ * Description: etrieves messages from a Kafka topic always starting with the most recent message first.
+ * Input:
+ * topic - The name of the Kafka topic.
+ * count - The number of Kafka messages to retrieve.
+ * config - Optional map of key/values that override any global properties.
+ * Returns: List of String
+
+### `LENGTH`
* Description: Returns the length of a string or size of a collection. Returns 0 for empty or null Strings
* Input:
* input - Object of string or collection type (e.g. list)
* Returns: Integer
-* `MAAS_GET_ENDPOINT`
+
+### `MAAS_GET_ENDPOINT`
* Description: Inspects ZooKeeper and returns a map containing the name, version and url for the model referred to by the input parameters.
* Input:
* model_name - The name of the model
* model_version - The optional version of the model. If the model version is not specified, the most current version is used.
* Returns: A map containing the name, version, and url for the REST endpoint (fields named name, version and url). Note that the output of this function is suitable for input into the first argument of MAAS_MODEL_APPLY.
-* `MAAS_MODEL_APPLY`
+
+### `MAAS_MODEL_APPLY`
* Description: Returns the output of a model deployed via Model as a Service. NOTE: Results are cached locally for 10 minutes.
* Input:
* endpoint - A map containing the name, version, and url for the REST endpoint
* function - The optional endpoint path; default is 'apply'
* model_args - A Dictionary of arguments for the model (these become request params)
* Returns: The output of the model deployed as a REST endpoint in Map form. Assumes REST endpoint returns a JSON Map.
-* `MAP_EXISTS`
+
+### `MAP_EXISTS`
* Description: Checks for existence of a key in a map.
* Input:
* key - The key to check for existence
* map - The map to check for existence of the key
* Returns: True if the key is found in the map and false if otherwise.
-* `MAP_GET`
+MAP_GET`
* Description: Gets the value associated with a key from a map
* Input:
* key - The key
* map - The map
* default - Optionally the default value to return if the key is not in the map.
* Returns: The object associated with the key in the map. If no value is associated with the key and default is specified, then default is returned. If no value is associated with the key or default, then null is returned.
-* `MONTH`
+
+### `MONTH`
* Description: The number representing the month. The first month, January, has a value of 0.
* Input:
* dateTime - The datetime as a long representing the milliseconds since unix epoch
* Returns: The current month (0-based).
-* `PROFILE_GET`
+
+### `PROFILE_GET`
* Description: Retrieves a series of values from a stored profile.
* Input:
* profile - The name of the profile.
@@ -217,199 +364,237 @@ The following functions are supported:
* units - The units of 'durationAgo'.
* groups - Optional - The groups used to sort the profile.
* Returns: The profile measurements.
-* `PROTOCOL_TO_NAME`
+
+### `PROTOCOL_TO_NAME`
* Description: Converts the IANA protocol number to the protocol name
* Input:
* IANA Number
* Returns: The protocol name associated with the IANA number.
-* `REGEXP_MATCH`
+
+### `REGEXP_MATCH`
* Description: Determines whether a regex matches a string
* Input:
* string - The string to test
* pattern - The proposed regex pattern
* Returns: True if the regex pattern matches the string and false if otherwise.
-* `SPLIT`
+
+### `SPLIT`
* Description: Splits the string by the delimiter.
* Input:
* input - String to split
* delim - String delimiter
* Returns: List of strings
-* `STARTS_WITH`
+
+### `STARTS_WITH`
* Description: Determines whether a string starts with a prefix
* Input:
* string - The string to test
* prefix - The proposed prefix
* Returns: True if the string starts with the specified prefix and false if otherwise
-* `STATS_ADD`
+
+### `STATS_ADD`
* Description: Adds one or more input values to those that are used to calculate the summary statistics.
* Input:
* stats - The Stellar statistics object. If null, then a new one is initialized.
* value+ - One or more numbers to add
* Returns: A Stellar statistics object
-* `STATS_COUNT`
+
+### `STATS_COUNT`
* Description: Calculates the count of the values accumulated (or in the window if a window is used).
* Input:
* stats - The Stellar statistics object
* Returns: The count of the values in the window or NaN if the statistics object is null.
-* `STATS_GEOMETRIC_MEAN`
+
+### `STATS_GEOMETRIC_MEAN`
* Description: Calculates the geometric mean of the accumulated values (or in the window if a window is used). See http://commons.apache.org/proper/commons-math/userguide/stat.html#a1.2_Descriptive_statistics
* Input:
* stats - The Stellar statistics object
* Returns: The geometric mean of the values in the window or NaN if the statistics object is null.
-* `STATS_INIT`
+
+### `STATS_INIT`
* Description: Initializes a statistics object
* Input:
* window_size - The number of input data values to maintain in a rolling window in memory. If window_size is equal to 0, then no rolling window is maintained. Using no rolling window is less memory intensive, but cannot calculate certain statistics like percentiles and kurtosis.
* Returns: A Stellar statistics object
-* `STATS_KURTOSIS`
+
+### `STATS_KURTOSIS`
* Description: Calculates the kurtosis of the accumulated values (or in the window if a window is used). See http://commons.apache.org/proper/commons-math/userguide/stat.html#a1.2_Descriptive_statistics
* Input:
* stats - The Stellar statistics object
* Returns: The kurtosis of the values in the window or NaN if the statistics object is null.
-* `STATS_MAX`
+
+### `STATS_MAX`
* Description: Calculates the maximum of the accumulated values (or in the window if a window is used).
* Input:
* stats - The Stellar statistics object
* Returns: The maximum of the accumulated values in the window or NaN if the statistics object is null.
-* `STATS_MEAN`
+
+### `STATS_MEAN`
* Description: Calculates the mean of the accumulated values (or in the window if a window is used).
* Input:
* stats - The Stellar statistics object
* Returns: The mean of the values in the window or NaN if the statistics object is null.
-* `STATS_MERGE`
+
+### `STATS_MERGE`
* Description: Merges statistics objects.
* Input:
* statistics - A list of statistics objects
* Returns: A Stellar statistics object
-* `STATS_MIN`
+
+### `STATS_MIN`
* Description: Calculates the minimum of the accumulated values (or in the window if a window is used).
* Input:
* stats - The Stellar statistics object
* Returns: The minimum of the accumulated values in the window or NaN if the statistics object is null.
-* `STATS_PERCENTILE`
+
+### `STATS_PERCENTILE`
* Description: Computes the p'th percentile of the accumulated values (or in the window if a window is used).
* Input:
* stats - The Stellar statistics object
* p - a double where 0 <= p < 1 representing the percentile
* Returns: The p'th percentile of the data or NaN if the statistics object is null
-* `STATS_POPULATION_VARIANCE`
+
+### `STATS_POPULATION_VARIANCE`
* Description: Calculates the population variance of the accumulated values (or in the window if a window is used). See http://commons.apache.org/proper/commons-math/userguide/stat.html#a1.2_Descriptive_statistics
* Input:
* stats - The Stellar statistics object
* Returns: The population variance of the values in the window or NaN if the statistics object is null.
-* `STATS_QUADRATIC_MEAN`
+
+### `STATS_QUADRATIC_MEAN`
* Description: Calculates the quadratic mean of the accumulated values (or in the window if a window is used). See http://commons.apache.org/proper/commons-math/userguide/stat.html#a1.2_Descriptive_statistics
* Input:
* stats - The Stellar statistics object
* Returns: The quadratic mean of the values in the window or NaN if the statistics object is null.
-* `STATS_SD`
+
+### `STATS_SD`
* Description: Calculates the standard deviation of the accumulated values (or in the window if a window is used). See http://commons.apache.org/proper/commons-math/userguide/stat.html#a1.2_Descriptive_statistics
* Input:
* stats - The Stellar statistics object
* Returns: The standard deviation of the values in the window or NaN if the statistics object is null.
-* `STATS_SKEWNESS`
+
+### `STATS_SKEWNESS`
* Description: Calculates the skewness of the accumulated values (or in the window if a window is used). See http://commons.apache.org/proper/commons-math/userguide/stat.html#a1.2_Descriptive_statistics
* Input:
* stats - The Stellar statistics object
* Returns: The skewness of the values in the window or NaN if the statistics object is null.
-* `STATS_SUM`
+
+### `STATS_SUM`
* Description: Calculates the sum of the accumulated values (or in the window if a window is used).
* Input:
* stats - The Stellar statistics object
* Returns: The sum of the values in the window or NaN if the statistics object is null.
-* `STATS_SUM_LOGS`
+
+### `STATS_SUM_LOGS`
* Description: Calculates the sum of the (natural) log of the accumulated values (or in the window if a window is used). See http://commons.apache.org/proper/commons-math/userguide/stat.html#a1.2_Descriptive_statistics
* Input:
* stats - The Stellar statistics object
* Returns: The sum of the (natural) log of the values in the window or NaN if the statistics object is null.
-* `STATS_SUM_SQUARES`
+
+### `STATS_SUM_SQUARES`
* Description: Calculates the sum of the squares of the accumulated values (or in the window if a window is used).
* Input:
* stats - The Stellar statistics object
* Returns: The sum of the squares of the values in the window or NaN if the statistics object is null.
-* `STATS_VARIANCE`
+
+### `STATS_VARIANCE`
* Description: Calculates the variance of the accumulated values (or in the window if a window is used). See http://commons.apache.org/proper/commons-math/userguide/stat.html#a1.2_Descriptive_statistics
* Input:
* stats - The Stellar statistics object
* Returns: The variance of the values in the window or NaN if the statistics object is null.
-* `SYSTEM_ENV_GET`
+
+### `SYSTEM_ENV_GET`
* Description: Returns the value associated with an environment variable
* Input:
* env_var - Environment variable name to get the value for
* Returns: String
-* `SYSTEM_PROPERTY_GET`
+
+### `SYSTEM_PROPERTY_GET`
* Description: Returns the value associated with a Java system property
* Input:
* key - Property to get the value for
* Returns: String
-* `TO_DOUBLE`
+
+### `TO_DOUBLE`
* Description: Transforms the first argument to a double precision number
* Input:
* input - Object of string or numeric type
* Returns: Double version of the first argument
-* `TO_EPOCH_TIMESTAMP`
+
+### `TO_EPOCH_TIMESTAMP`
* Description: Returns the epoch timestamp of the dateTime in the specified format. If the format does not have a timestamp and you wish to assume a given timestamp, you may specify the timezone optionally.
* Input:
* dateTime - DateTime in String format
* format - DateTime format as a String
* timezone - Optional timezone in String format
* Returns: Epoch timestamp
-* `TO_INTEGER`
+
+### `TO_INTEGER`
* Description: Transforms the first argument to an integer
* Input:
* input - Object of string or numeric type
* Returns: Integer version of the first argument
-* `TO_LOWER`
+
+### `TO_LOWER`
* Description: Transforms the first argument to a lowercase string
* Input:
* input - String
* Returns: Lowercase string
-* `TO_STRING`
+
+### `TO_STRING`
* Description: Transforms the first argument to a string
* Input:
* input - Object
* Returns: String
-* `TO_UPPER`
+
+### `TO_UPPER`
* Description: Transforms the first argument to an uppercase string
* Input:
* input - String
* Returns: Uppercase string
-* `TRIM`
+
+### `TRIM`
* Description: Trims whitespace from both sides of a string.
* Input:
* input - String
* Returns: String
-* `URL_TO_HOST`
+
+### `URL_TO_HOST`
* Description: Extract the hostname from a URL.
* Input:
* url - URL in String form
* Returns: The hostname from the URL as a String. e.g. URL_TO_HOST('http://www.yahoo.com/foo') would yield 'www.yahoo.com'
-* `URL_TO_PATH`
+
+### `URL_TO_PATH`
* Description: Extract the path from a URL.
* Input:
* url - URL in String form
* Returns: The path from the URL as a String. e.g. URL_TO_PATH('http://www.yahoo.com/foo') would yield 'foo'
-* `URL_TO_PORT`
+
+### `URL_TO_PORT`
* Description: Extract the port from a URL. If the port is not explicitly stated in the URL, then an implicit port is inferred based on the protocol.
* Input:
* url - URL in string form
* Returns: The port used in the URL as an integer (for example, URL_TO_PORT('http://www.yahoo.com/foo') would yield 80)
-* `URL_TO_PROTOCOL`
+
+### `URL_TO_PROTOCOL`
* Description: Extract the protocol from a URL.
* Input:
* url - URL in String form
* Returns: The protocol from the URL as a String. e.g. URL_TO_PROTOCOL('http://www.yahoo.com/foo') would yield 'http'
-* `WEEK_OF_MONTH`
+
+### `WEEK_OF_MONTH`
* Description: The numbered week within the month. The first week within the month has a value of 1.
* Input:
* dateTime - The datetime as a long representing the milliseconds since unix epoch
* Returns: The numbered week within the month.
-* `WEEK_OF_YEAR`
+
+### `WEEK_OF_YEAR`
* Description: The numbered week within the year. The first week in the year has a value of 1.
* Input:
* dateTime - The datetime as a long representing the milliseconds since unix epoch
* Returns: The numbered week within the year.
-* `YEAR`
+
+### `YEAR`
* Description: The number representing the year.
* Input:
* dateTime - The datetime as a long representing the milliseconds since unix epoch
@@ -581,7 +766,8 @@ IS_EMAIL
[Stellar]>>>
```
-##Global Configuration
+# Global Configuration
+
The format of the global enrichment is a JSON String to Object map. This is intended for
configuration which is non sensor specific configuration.
@@ -605,7 +791,7 @@ This configuration is stored in zookeeper, but looks something like
}
```
-###Validation Framework
+## Validation Framework
Inside of the global configuration, there is a validation framework in
place that enables the validation that messages coming from all parsers
@@ -629,7 +815,7 @@ structured like so:
* `NOT_EMPTY` : Validates that the fields exist and are not empty (after trimming.)
-##Management Utility
+# Management Utility
Configurations should be stored on disk in the following structure starting at `$BASE_DIR`:
* global.json : The global config
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/dbb56cb5/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/BaseIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/BaseIntegrationTest.java b/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/BaseIntegrationTest.java
index 5fdae49..ddce400 100644
--- a/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/BaseIntegrationTest.java
+++ b/metron-platform/metron-integration-test/src/main/java/org/apache/metron/integration/BaseIntegrationTest.java
@@ -27,7 +27,7 @@ import java.util.Properties;
public abstract class BaseIntegrationTest {
- protected KafkaWithZKComponent getKafkaComponent(final Properties topologyProperties, List<KafkaWithZKComponent.Topic> topics) {
+ protected static KafkaWithZKComponent getKafkaComponent(final Properties topologyProperties, List<KafkaWithZKComponent.Topic> topics) {
return new KafkaWithZKComponent().withTopics(topics)
.withPostStartCallback(new Function<KafkaWithZKComponent, Void>() {
@Nullable
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/dbb56cb5/metron-platform/metron-management/pom.xml
----------------------------------------------------------------------
diff --git a/metron-platform/metron-management/pom.xml b/metron-platform/metron-management/pom.xml
index 8767499..18651fb 100644
--- a/metron-platform/metron-management/pom.xml
+++ b/metron-platform/metron-management/pom.xml
@@ -116,6 +116,12 @@
</dependency>
<dependency>
<groupId>org.apache.metron</groupId>
+ <artifactId>metron-integration-test</artifactId>
+ <version>${parent.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.metron</groupId>
<artifactId>metron-common</artifactId>
<version>${parent.version}</version>
<type>test-jar</type>
@@ -144,6 +150,35 @@
<classifier>tests</classifier>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_2.10</artifactId>
+ <version>${global_kafka_version}</version>
+ <classifier>test</classifier>
+ <exclusions>
+ <exclusion>
+ <artifactId>slf4j-log4j12</artifactId>
+ <groupId>org.slf4j</groupId>
+ </exclusion>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${global_kafka_version}</version>
+ <classifier>test</classifier>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
</dependencies>
<reporting>
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/dbb56cb5/metron-platform/metron-management/src/main/java/org/apache/metron/management/KafkaFunctions.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-management/src/main/java/org/apache/metron/management/KafkaFunctions.java b/metron-platform/metron-management/src/main/java/org/apache/metron/management/KafkaFunctions.java
new file mode 100644
index 0000000..b0858bf
--- /dev/null
+++ b/metron-platform/metron-management/src/main/java/org/apache/metron/management/KafkaFunctions.java
@@ -0,0 +1,452 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.metron.management;
+
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.metron.common.dsl.Context;
+import org.apache.metron.common.dsl.ParseException;
+import org.apache.metron.common.dsl.Stellar;
+import org.apache.metron.common.dsl.StellarFunction;
+import org.apache.metron.common.utils.ConversionUtils;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+
+import static org.apache.metron.common.dsl.Context.Capabilities.GLOBAL_CONFIG;
+
+/**
+ * Kafka functions available in Stellar.
+ *
+ * KAFKA_GET
+ * KAFKA_TAIL
+ * KAFKA_PUT
+ * KAFKA_PROPS
+ */
+public class KafkaFunctions {
+
+ /**
+ * How long to wait on each poll request in milliseconds. There will be multiple
+ * poll requests, each waiting this period of time. The maximum amount of time
+ * that the user is willing to wait for a message will be some multiple of this value.
+ */
+ private static final int POLL_TIMEOUT = 1000;
+
+ /**
+ * The key for the property that defines the maximum amount of time
+ * to wait to receive messages.
+ */
+ private static final String MAX_WAIT_PROPERTY = "stellar.kafka.max.wait";
+
+ /**
+ * Maintaining the default Kafka properties as a static class member is
+ * critical to consumer offset management.
+ *
+ * A unique 'group.id' is generated when creating these default properties. This
+ * value is used when storing Kafka consumer offsets. Multiple executions of any
+ * KAFKA_* functions, within the same Stellar REPL session, must maintain the same
+ * 'group.id'. At the same time, different Stellar REPL sessions running
+ * simultaneously should each have their own 'group.id'.
+ */
+ private static Properties defaultProperties = defaultKafkaProperties();
+
+ /**
+ * KAFKA_GET
+ *
+ * Retrieves messages from a Kafka topic. Subsequent calls will continue retrieving messages
+ * sequentially from the original offset.
+ *
+ * Example: Retrieve one message from a topic.
+ * KAFKA_GET('topic')
+ *
+ * Example: Retrieve 10 messages from a topic.
+ * KAFKA_GET('topic', 10)
+ *
+ * Example: Retrieve the first message from a topic. This must be the first retrieval
+ * from the topic, otherwise the messages will be retrieved starting from the
+ * previously stored consumer offset.
+ * KAFKA_GET('topic', 1, { "auto.offset.reset": "earliest" })
+ */
+ @Stellar(
+ namespace = "KAFKA",
+ name = "GET",
+ description = "Retrieves messages from a Kafka topic. Subsequent calls will" +
+ "continue retrieving messages sequentially from the original offset.",
+ params = {
+ "topic - The name of the Kafka topic",
+ "count - The number of Kafka messages to retrieve",
+ "config - Optional map of key/values that override any global properties."
+ },
+ returns = "List of strings"
+ )
+ public static class KafkaGet implements StellarFunction {
+
+ @Override
+ public Object apply(List<Object> args, Context context) throws ParseException {
+ List<String> messages = new ArrayList<>();
+
+ // required - name of the topic to retrieve messages from
+ String topic = ConversionUtils.convert(args.get(0), String.class);
+
+ // optional - how many messages should be retrieved?
+ int count = 1;
+ if(args.size() > 1) {
+ count = ConversionUtils.convert(args.get(1), Integer.class);
+ }
+
+ // optional - property overrides provided by the user
+ Map<String, String> overrides = new HashMap<>();
+ if(args.size() > 2) {
+ overrides = ConversionUtils.convert(args.get(2), Map.class);
+ }
+
+ // build the properties for kafka
+ Properties properties = buildKafkaProperties(overrides, context);
+
+ // read some messages
+ try (KafkaConsumer<String, String> consumer = new KafkaConsumer(properties)) {
+ consumer.subscribe(Arrays.asList(topic));
+
+ int maxAttempts = getMaxAttempts(properties);
+ int i = 0;
+ while(messages.size() < count && i++ < maxAttempts) {
+ consumer.poll(POLL_TIMEOUT).forEach(record -> messages.add(record.value()));
+ consumer.commitSync();
+ }
+ }
+
+ return messages;
+ }
+
+ @Override
+ public void initialize(Context context) {
+ // no initialization required
+ }
+
+ @Override
+ public boolean isInitialized() {
+ // no initialization required
+ return true;
+ }
+ }
+
+ /**
+ * KAFKA_TAIL
+ *
+ * Retrieves messages from a Kafka topic always starting with
+ * the most recent message first.
+ *
+ * Example: Retrieve the latest message from a topic.
+ * KAFKA_TAIL('topic')
+ *
+ * Example: Retrieve 10 messages from a topic starting with the latest.
+ * KAFKA_TAIL('topic', 10)
+ */
+ @Stellar(
+ namespace = "KAFKA",
+ name = "TAIL",
+ description = "Retrieves messages from a Kafka topic always starting with the most recent message first.",
+ params = {
+ "topic - The name of the Kafka topic",
+ "count - The number of Kafka messages to retrieve",
+ "config - Optional map of key/values that override any global properties."
+ },
+ returns = "Messages retrieved from the Kafka topic"
+ )
+ public static class KafkaTail implements StellarFunction {
+
+ @Override
+ public Object apply(List<Object> args, Context context) throws ParseException {
+ List<String> messages = new ArrayList<>();
+
+ // required - name of the topic to retrieve messages from
+ String topic = ConversionUtils.convert(args.get(0), String.class);
+
+ // optional - how many messages should be retrieved?
+ int count = 1;
+ if(args.size() > 1) {
+ count = ConversionUtils.convert(args.get(1), Integer.class);
+ }
+
+ // optional - property overrides provided by the user
+ Map<String, String> overrides = new HashMap<>();
+ if(args.size() > 2) {
+ overrides = ConversionUtils.convert(args.get(2), Map.class);
+ }
+
+ // build the properties for kafka
+ Properties properties = buildKafkaProperties(overrides, context);
+
+ // ensures messages pulled from latest offset, versus a previously stored consumer offset
+ properties.put("group.id", generateGroupId());
+ properties.put("auto.offset.reset", "latest");
+
+ try (KafkaConsumer<String, String> consumer = new KafkaConsumer(properties)) {
+ consumer.subscribe(Arrays.asList(topic));
+
+ int maxAttempts = getMaxAttempts(properties);
+ int i = 0;
+ while(messages.size() < count && i++ < maxAttempts) {
+ consumer.poll(POLL_TIMEOUT).forEach(record -> messages.add(record.value()));
+ consumer.commitSync();
+ }
+ }
+
+ return messages;
+ }
+
+ @Override
+ public void initialize(Context context) {
+ // no initialization required
+ }
+
+ @Override
+ public boolean isInitialized() {
+ // no initialization required
+ return true;
+ }
+ }
+
+ /**
+ * KAFKA_PUT
+ *
+ * Sends messages to a Kafka topic.
+ *
+ * Example: Put two messages on the topic 'topic'.
+ * KAFKA_PUT('topic', ["message1", "message2"])
+ *
+ * Example: Put a message on a topic and also define an alternative Kafka broker.
+ * KAFKA_PUT('topic', ["message1"], { "bootstrap.servers": "kafka-broker-1:6667" })
+ */
+ @Stellar(
+ namespace = "KAFKA",
+ name = "PUT",
+ description = "Sends messages to a Kafka topic.",
+ params = {
+ "topic - The name of the Kafka topic.",
+ "messages - A list of messages to write.",
+ "config - An optional map of key/values that override any global properties."
+ },
+ returns = " "
+ )
+ public static class KafkaPut implements StellarFunction {
+
+ @Override
+ public Object apply(List<Object> args, Context context) throws ParseException {
+
+ String topic = ConversionUtils.convert(args.get(0), String.class);
+ List<String> messages = ConversionUtils.convert(args.get(1), List.class);
+
+ // build the properties for kafka
+ Map<String, String> overrides = new HashMap<>();
+ if(args.size() > 2) {
+ overrides = ConversionUtils.convert(args.get(2), Map.class);
+ }
+ Properties properties = buildKafkaProperties(overrides, context);
+
+ // send the messages
+ try {
+ send(topic, messages, properties);
+
+ } catch(InterruptedException | ExecutionException e) {
+ throw new ParseException(e.getMessage(), e);
+ }
+
+ return null;
+ }
+
+ /**
+ * Send each message synchronously.
+ * @param topic The topic to send messages to.
+ * @param messages The messages to send.
+ * @param properties The properties to use with Kafka.
+ */
+ private void send(String topic, List<String> messages, Properties properties) throws InterruptedException, ExecutionException {
+ try (KafkaProducer<String, String> producer = new KafkaProducer(properties)) {
+
+ // send each message synchronously, hence the get()
+ for(String msg : messages) {
+ producer.send(new ProducerRecord<>(topic, msg)).get();
+ }
+ producer.flush();
+ }
+ }
+
+ @Override
+ public void initialize(Context context) {
+ // no initialization required
+ }
+
+ @Override
+ public boolean isInitialized() {
+ // no initialization required
+ return true;
+ }
+ }
+
+ /**
+ * KAFKA_PROPS
+ *
+ * Retrieves the Kafka properties that are used by other KAFKA_* functions
+ * like KAFKA_GET and KAFKA_PUT. The Kafka properties are compiled from a
+ * set of default properties, the global properties, and any overrides.
+ *
+ * Example: Retrieve the current Kafka properties.
+ * KAFKA_PROPS()
+ *
+ * Example: Retrieve the current Kafka properties taking into account a set of overrides.
+ * KAFKA_PROPS({ "max.poll.records": 1 })
+ */
+ @Stellar(
+ namespace = "KAFKA",
+ name = "PROPS",
+ description = "Retrieves the Kafka properties that are used by other KAFKA_* functions " +
+ "like KAFKA_GET and KAFKA_PUT. The Kafka properties are compiled from a " +
+ "set of default properties, the global properties, and any overrides.",
+ params = { "config - An optional map of key/values that override any global properties." },
+ returns = " "
+ )
+ public static class KafkaProps implements StellarFunction {
+
+ @Override
+ public Object apply(List<Object> args, Context context) throws ParseException {
+
+ // optional - did the user provide any overrides?
+ Map<String, String> overrides = new HashMap<>();
+ if(args.size() > 0) {
+ overrides = ConversionUtils.convert(args.get(0), Map.class);
+ }
+
+ return buildKafkaProperties(overrides, context);
+ }
+
+ @Override
+ public void initialize(Context context) {
+ // no initialization required
+ }
+
+ @Override
+ public boolean isInitialized() {
+ // no initialization required
+ return true;
+ }
+ }
+
+ /**
+ * Assembles the set of Properties required by the Kafka client.
+ *
+ * A set of default properties has been defined to provide minimum functionality.
+ * Any properties defined in the global configuration override these defaults.
+ * Any user-defined overrides then override all others.
+ *
+ * @param overrides Property overrides provided by the user.
+ * @param context The Stellar context.
+ */
+ private static Properties buildKafkaProperties(Map<String, String> overrides, Context context) {
+
+ // start with minimal set of default properties
+ Properties properties = new Properties();
+ properties.putAll(defaultProperties);
+
+ // override the default properties with those in the global configuration
+ Optional<Object> globalCapability = context.getCapability(GLOBAL_CONFIG, false);
+ if(globalCapability.isPresent()) {
+ Map<String, Object> global = (Map<String, Object>) globalCapability.get();
+ properties.putAll(global);
+ }
+
+ // any user-defined properties will override both the defaults and globals
+ properties.putAll(overrides);
+
+ return properties;
+ }
+
+ /**
+ * Determine how many poll attempts should be made based on the user's patience.
+ * @param properties
+ * @return The maximum number of poll attempts to make.
+ */
+ private static int getMaxAttempts(Properties properties) {
+ int maxWait = ConversionUtils.convert(properties.get(MAX_WAIT_PROPERTY), Integer.class);
+ return maxWait / POLL_TIMEOUT;
+ }
+
+ /**
+ * Defines a minimal set of default parameters that can be overridden
+ * via the global properties.
+ */
+ private static Properties defaultKafkaProperties() {
+
+ Properties properties = new Properties();
+ properties.put("bootstrap.servers", "localhost:9092");
+
+ /*
+ * A unique 'group.id' is generated when creating these default properties. This
+ * value is used when storing Kafka consumer offsets. Multiple executions of any
+ * KAFKA_* functions, within the same Stellar REPL session, must maintain the same
+ * 'group.id'. At the same time, different Stellar REPL sessions running
+ * simultaneously should each have their own 'group.id'.
+ */
+ properties.put("group.id", generateGroupId());
+
+ /*
+ * What to do when there is no initial offset in Kafka or if the current
+ * offset does not exist any more on the server (e.g. because that data has been deleted):
+ *
+ * "earliest": automatically reset the offset to the earliest offset
+ * "latest": automatically reset the offset to the latest offset
+ * "none": throw exception to the consumer if no previous offset is found or the consumer's group
+ * anything else: throw exception to the consumer.
+ */
+ properties.put("auto.offset.reset", "latest");
+
+ // limits the number of messages read in a single poll request
+ properties.put("max.poll.records", 1);
+
+ // consumer deserialization
+ properties.put("key.deserializer", StringDeserializer.class.getName());
+ properties.put("value.deserializer", StringDeserializer.class.getName());
+
+ // producer serialization
+ properties.put("key.serializer", StringSerializer.class.getName());
+ properties.put("value.serializer", StringSerializer.class.getName());
+
+ // the maximum time to wait for messages
+ properties.put(MAX_WAIT_PROPERTY, 5000);
+
+ return properties;
+ }
+
+ /**
+ * Generates a unique 'group.id' for a session.
+ */
+ private static String generateGroupId() {
+ return String.format("stellar-shell-%s", UUID.randomUUID().toString());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/dbb56cb5/metron-platform/metron-management/src/test/java/org/apache/metron/management/KafkaFunctionsIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-management/src/test/java/org/apache/metron/management/KafkaFunctionsIntegrationTest.java b/metron-platform/metron-management/src/test/java/org/apache/metron/management/KafkaFunctionsIntegrationTest.java
new file mode 100644
index 0000000..0f12f0b
--- /dev/null
+++ b/metron-platform/metron-management/src/test/java/org/apache/metron/management/KafkaFunctionsIntegrationTest.java
@@ -0,0 +1,185 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.metron.management;
+
+import org.apache.metron.common.dsl.Context;
+import org.apache.metron.common.dsl.StellarFunctions;
+import org.apache.metron.common.stellar.StellarProcessor;
+import org.apache.metron.integration.BaseIntegrationTest;
+import org.apache.metron.integration.ComponentRunner;
+import org.apache.metron.integration.components.KafkaWithZKComponent;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests the KafkaFunctions class.
+ *
+ * Labelled as an integration test as the tests stand-up a Kafka Broker for the
+ * Stellar Kafka functions to interact with.
+ */
+public class KafkaFunctionsIntegrationTest extends BaseIntegrationTest {
+
+ private static final String message1 = "{ \"ip_src_addr\": \"10.0.0.1\", \"value\": 14687 }";
+ private static final String message2 = "{ \"ip_src_addr\": \"10.0.0.1\", \"value\": 23 }";
+ private static final String message3 = "{ \"ip_src_addr\": \"10.0.0.1\", \"value\": 29011 }";
+
+ private static Map<String, Object> variables = new HashMap<>();
+ private static KafkaWithZKComponent kafkaComponent;
+ private static ComponentRunner runner;
+ private static Properties global;
+
+ @BeforeClass
+ public static void setupKafka() throws Exception {
+
+ Properties properties = new Properties();
+ kafkaComponent = getKafkaComponent(properties, new ArrayList<>());
+
+ runner = new ComponentRunner.Builder()
+ .withComponent("kafka", kafkaComponent)
+ .withMillisecondsBetweenAttempts(5000)
+ .withNumRetries(5)
+ .build();
+ runner.start();
+ }
+
+ @Before
+ public void setup() {
+
+ // messages that will be read/written during the tests
+ variables.put("message1", message1);
+ variables.put("message2", message2);
+ variables.put("message3", message3);
+
+ // global properties
+ global = new Properties();
+ global.put("bootstrap.servers", kafkaComponent.getBrokerList());
+
+ // start reading from the earliest offset, which is necessary for these tests
+ global.put("auto.offset.reset", "earliest");
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ runner.stop();
+ }
+
+ /**
+ * Write one message, read one message.
+ */
+ @Test
+ public void testOneMessage() {
+ run("KAFKA_PUT('topic1', [message1])");
+ Object actual = run("KAFKA_GET('topic1')");
+ assertEquals(Collections.singletonList(message1), actual);
+ }
+
+ /**
+ * Multiple messages can be read with a single call.
+ */
+ @Test
+ public void testMultipleMessages() {
+ run("KAFKA_PUT('topic2', [message1, message2, message3])");
+ Object actual = run("KAFKA_GET('topic2', 3)");
+
+ List<String> expected = new ArrayList<String>() {{
+ add(message1);
+ add(message2);
+ add(message3);
+ }};
+ assertEquals(expected, actual);
+ }
+
+ /**
+ * Does the client maintain the consumer offset correctly?
+ */
+ @Test
+ public void testConsumerOffsets() {
+ run("KAFKA_PUT('topic3', [message1, message2, message3])");
+
+ // the offsets must be maintained correctly for us to read each message, in order,
+ // sequentially across separate calls to KAFKA_GET
+ assertEquals(Collections.singletonList(message1), run("KAFKA_GET('topic3', 1)"));
+ assertEquals(Collections.singletonList(message2), run("KAFKA_GET('topic3', 1)"));
+ assertEquals(Collections.singletonList(message3), run("KAFKA_GET('topic3', 1)"));
+ }
+
+ /**
+ * The properties used for the KAFKA_* functions are calculated by compiling the default, global and user
+ * properties into a single set of properties. The global properties should override any default properties.
+ */
+ @Test
+ public void testKafkaPropsWithGlobalOverride() {
+
+ // setup - override a key in the global properties
+ final String overriddenKey = "bootstrap.servers";
+ final String expected = "foo.global.override.com:9092";
+ global.setProperty(overriddenKey, expected);
+
+ // validate - ensure the global overrides the default property value
+ Map<String, String> properties = (Map<String, String>) run("KAFKA_PROPS()");
+ assertEquals(expected, properties.get(overriddenKey));
+ }
+
+ /**
+ * The properties used for the KAFKA_* functions are calculated by compiling the default, global and user
+ * properties into a single set of properties. The user properties should override any default or global properties.
+ */
+ @Test
+ public void testKafkaPropsWithUserOverride() {
+
+ // setup - override a key in the global properties
+ final String overriddenKey = "bootstrap.servers";
+ global.setProperty(overriddenKey, "foo.global.override.com:9092");
+
+ // setup - override the same key in the user properties
+ final String expected = "foo.user.override.com:9092";
+ String expression = String.format("KAFKA_PROPS({ '%s' : '%s' })", overriddenKey, expected);
+
+ // validate - ensure the user properties override the global and defaults
+ Map<String, String> properties = (Map<String, String>) run(expression);
+ assertEquals(expected, properties.get(overriddenKey));
+ }
+
+ /**
+ * Runs a Stellar expression.
+ * @param expr The expression to run.
+ */
+ private Object run(String expr) {
+
+ // make the global properties available to the function
+ Context context = new Context.Builder()
+ .with(Context.Capabilities.GLOBAL_CONFIG, () -> global)
+ .build();
+
+ StellarProcessor processor = new StellarProcessor();
+ return processor.parse(expr, x -> variables.get(x), StellarFunctions.FUNCTION_RESOLVER(), context);
+ }
+
+}