You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@metron.apache.org by mm...@apache.org on 2018/11/16 00:06:02 UTC
[03/24] metron git commit: METRON-1834: Migrate Elasticsearch from
TransportClient to new Java REST API (mmiklavc via mmiklavc) closes
apache/metron#1242
http://git-wip-us.apache.org/repos/asf/metron/blob/8bf3b6ec/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/RestFunctions.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/RestFunctions.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/RestFunctions.java
new file mode 100644
index 0000000..354322a
--- /dev/null
+++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/RestFunctions.java
@@ -0,0 +1,388 @@
+/**
+ * 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.stellar.dsl.functions;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.protocol.HttpClientContext;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+import org.apache.http.util.EntityUtils;
+import org.apache.metron.stellar.common.utils.ConversionUtils;
+import org.apache.metron.stellar.common.utils.JSONUtils;
+import org.apache.metron.stellar.dsl.Context;
+import org.apache.metron.stellar.dsl.ParseException;
+import org.apache.metron.stellar.dsl.Stellar;
+import org.apache.metron.stellar.dsl.StellarFunction;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+import static java.lang.String.format;
+import static org.apache.metron.stellar.dsl.Context.Capabilities.GLOBAL_CONFIG;
+import static org.apache.metron.stellar.dsl.functions.RestConfig.POOLING_DEFAULT_MAX_PER_RUOTE;
+import static org.apache.metron.stellar.dsl.functions.RestConfig.POOLING_MAX_TOTAL;
+import static org.apache.metron.stellar.dsl.functions.RestConfig.STELLAR_REST_SETTINGS;
+
+/**
+ * Defines functions that enable REST requests with proper result and error handling. Depends on an
+ * Apache HttpComponents client being supplied as a Stellar HTTP_CLIENT capability. Exposes various Http settings
+ * including authentication, proxy and timeouts through the global config with the option to override any settings
+ * through a config object supplied in the expression.
+ */
+public class RestFunctions {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ /**
+ * Get an argument from a list of arguments.
+ *
+ * @param index The index within the list of arguments.
+ * @param clazz The type expected.
+ * @param args All of the arguments.
+ * @param <T> The type of the argument expected.
+ */
+ public static <T> T getArg(int index, Class<T> clazz, List<Object> args) {
+
+ if(index >= args.size()) {
+ throw new IllegalArgumentException(format("Expected at least %d argument(s), found %d", index+1, args.size()));
+ }
+
+ return ConversionUtils.convert(args.get(index), clazz);
+ }
+
+ @Stellar(
+ namespace = "REST",
+ name = "GET",
+ description = "Performs a REST GET request and parses the JSON results into a map.",
+ params = {
+ "url - URI to the REST service",
+ "rest_config - Optional - Map (in curly braces) of name:value pairs, each overriding the global config parameter " +
+ "of the same name. Default is the empty Map, meaning no overrides."
+ },
+ returns = "JSON results as a Map")
+ public static class RestGet implements StellarFunction {
+
+ /**
+ * Whether the function has been initialized.
+ */
+ private boolean initialized = false;
+
+ /**
+ * The CloseableHttpClient.
+ */
+ private CloseableHttpClient httpClient;
+
+ /**
+ * Executor used to impose a hard request timeout.
+ */
+ private ScheduledExecutorService scheduledExecutorService;
+
+ /**
+ * Initialize the function by creating a ScheduledExecutorService and looking up the CloseableHttpClient from the
+ * Stellar context.
+ * @param context
+ */
+ @Override
+ public void initialize(Context context) {
+ scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
+ httpClient = getHttpClient(context);
+ initialized = true;
+ }
+
+ @Override
+ public boolean isInitialized() {
+ return initialized;
+ }
+
+ /**
+ * Apply the function.
+ * @param args The function arguments including uri and rest config.
+ * @param context Stellar context
+ */
+ @Override
+ public Object apply(List<Object> args, Context context) throws ParseException {
+ RestConfig restConfig = new RestConfig();
+ try {
+ URI uri = new URI(getArg(0, String.class, args));
+ restConfig = getRestConfig(args, getGlobalConfig(context));
+
+ HttpHost target = new HttpHost(uri.getHost(), uri.getPort(), uri.getScheme());
+ Optional<HttpHost> proxy = getProxy(restConfig);
+ HttpClientContext httpClientContext = getHttpClientContext(restConfig, target, proxy);
+
+ HttpGet httpGet = new HttpGet(uri);
+ httpGet.addHeader("Accept", "application/json");
+ httpGet.setConfig(getRequestConfig(restConfig, proxy));
+
+ return doGet(restConfig, httpGet, httpClientContext);
+ } catch (URISyntaxException e) {
+ throw new IllegalArgumentException(e.getMessage(), e);
+ } catch (IOException e) {
+ LOG.error(e.getMessage(), e);
+ return restConfig.getErrorValueOverride();
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (httpClient != null) {
+ httpClient.close();
+ }
+ if (scheduledExecutorService != null) {
+ scheduledExecutorService.shutdown();
+ }
+ }
+
+ /**
+ * Retrieves the ClosableHttpClient from a pooling connection manager.
+ *
+ * @param context The execution context.
+ * @return A ClosableHttpClient.
+ */
+ protected CloseableHttpClient getHttpClient(Context context) {
+ RestConfig restConfig = getRestConfig(Collections.emptyList(), getGlobalConfig(context));
+
+ PoolingHttpClientConnectionManager cm = getConnectionManager(restConfig);
+
+ return HttpClients.custom()
+ .setConnectionManager(cm)
+ .build();
+ }
+
+ protected PoolingHttpClientConnectionManager getConnectionManager(RestConfig restConfig) {
+ PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager();
+ if (restConfig.containsKey(POOLING_MAX_TOTAL)) {
+ cm.setMaxTotal(restConfig.getPoolingMaxTotal());
+ }
+ if (restConfig.containsKey(POOLING_DEFAULT_MAX_PER_RUOTE)) {
+ cm.setDefaultMaxPerRoute(restConfig.getPoolingDefaultMaxPerRoute());
+ }
+ return cm;
+ }
+
+ /**
+ * Only used for testing.
+ * @param httpClient
+ */
+ protected void setHttpClient(CloseableHttpClient httpClient) {
+ this.httpClient = httpClient;
+ }
+
+ /**
+ * Perform the HttpClient get and handle the results. A configurable list of status codes are accepted and the
+ * response content (expected to be json) is parsed into a Map. Values returned on errors and when response content
+ * is also configurable. The rest config "timeout" setting is imposed in this method and will abort the get request
+ * if exceeded.
+ *
+ * @param restConfig
+ * @param httpGet
+ * @param httpClientContext
+ * @return
+ * @throws IOException
+ */
+ protected Object doGet(RestConfig restConfig, HttpGet httpGet, HttpClientContext httpClientContext) throws IOException {
+
+ // Schedule a command to abort the httpGet request if the timeout is exceeded
+ ScheduledFuture scheduledFuture = scheduledExecutorService.schedule(httpGet::abort, restConfig.getTimeout(), TimeUnit.MILLISECONDS);
+ CloseableHttpResponse response;
+ try {
+ response = httpClient.execute(httpGet, httpClientContext);
+ } catch(Exception e) {
+ // Report a timeout if the httpGet request was aborted. Otherwise rethrow exception.
+ if (httpGet.isAborted()) {
+ throw new IOException(String.format("Total Stellar REST request time to %s exceeded the configured timeout of %d ms.", httpGet.getURI().toString(), restConfig.getTimeout()));
+ } else {
+ throw e;
+ }
+ }
+
+ // Cancel the future if the request finished within the timeout
+ if (!scheduledFuture.isDone()) {
+ scheduledFuture.cancel(true);
+ }
+ int statusCode = response.getStatusLine().getStatusCode();
+ if (restConfig.getResponseCodesAllowed().contains(statusCode)) {
+ HttpEntity httpEntity = response.getEntity();
+
+ // Parse the reponse if present, return the empty value override if not
+ if (httpEntity != null && httpEntity.getContentLength() > 0) {
+ String json = EntityUtils.toString(response.getEntity());
+ return JSONUtils.INSTANCE.load(json, JSONUtils.MAP_SUPPLIER);
+ }
+ return restConfig.getEmptyContentOverride();
+ } else {
+ throw new IOException(String.format("Stellar REST request to %s expected status code to be one of %s but " +
+ "failed with http status code %d: %s",
+ httpGet.getURI().toString(),
+ restConfig.getResponseCodesAllowed().toString(),
+ statusCode,
+ EntityUtils.toString(response.getEntity())));
+ }
+ }
+
+ private Map<String, Object> getGlobalConfig(Context context) {
+ Optional<Object> globalCapability = context.getCapability(GLOBAL_CONFIG, false);
+ return globalCapability.map(o -> (Map<String, Object>) o).orElseGet(HashMap::new);
+ }
+
+ /**
+ * Build the RestConfig object using the following order of precedence:
+ * <ul>
+ * <li>rest config supplied as an expression parameter</li>
+ * <li>rest config stored in the global config</li>
+ * <li>default rest config</li>
+ * </ul>
+ * Only settings specified in the rest config will override lower priority config settings.
+ * @param args
+ * @param globalConfig
+ * @return
+ * @throws IOException
+ */
+ protected RestConfig getRestConfig(List<Object> args, Map<String, Object> globalConfig) {
+ Map<String, Object> globalRestConfig = (Map<String, Object>) globalConfig.get(STELLAR_REST_SETTINGS);
+ Map<String, Object> functionRestConfig = null;
+ if (args.size() > 1) {
+ functionRestConfig = getArg(1, Map.class, args);
+ }
+
+ // Add settings in order of precedence
+ RestConfig restConfig = new RestConfig();
+ if (globalRestConfig != null) {
+ restConfig.putAll(globalRestConfig);
+ }
+ if (functionRestConfig != null) {
+ restConfig.putAll(functionRestConfig);
+ }
+ return restConfig;
+ }
+
+ /**
+ * Returns the proxy HttpHost object if the proxy rest config settings are set.
+ * @param restConfig
+ * @return
+ */
+ protected Optional<HttpHost> getProxy(RestConfig restConfig) {
+ Optional<HttpHost> proxy = Optional.empty();
+ if (restConfig.getProxyHost() != null && restConfig.getProxyPort() != null) {
+ proxy = Optional.of(new HttpHost(restConfig.getProxyHost(), restConfig.getProxyPort(), "http"));
+ }
+ return proxy;
+ }
+
+ /**
+ * Builds the RequestConfig object by setting HttpClient settings defined in the rest config.
+ * @param restConfig
+ * @param proxy
+ * @return
+ */
+ protected RequestConfig getRequestConfig(RestConfig restConfig, Optional<HttpHost> proxy) {
+ RequestConfig.Builder requestConfigBuilder = RequestConfig.custom();
+ if (restConfig.getConnectTimeout() != null) {
+ requestConfigBuilder.setConnectTimeout(restConfig.getConnectTimeout());
+ }
+ if (restConfig.getConnectionRequestTimeout() != null) {
+ requestConfigBuilder.setConnectionRequestTimeout(restConfig.getConnectionRequestTimeout());
+ }
+ if (restConfig.getSocketTimeout() != null) {
+ requestConfigBuilder.setSocketTimeout(restConfig.getSocketTimeout());
+ }
+
+ proxy.ifPresent(requestConfigBuilder::setProxy);
+ return requestConfigBuilder.build();
+ }
+
+ /**
+ * Builds the HttpClientContext object by setting the basic auth and/or proxy basic auth credentials when the
+ * necessary rest config settings are configured. Passwords are stored in HDFS.
+ * @param restConfig
+ * @param target
+ * @param proxy
+ * @return
+ * @throws IOException
+ */
+ protected HttpClientContext getHttpClientContext(RestConfig restConfig, HttpHost target, Optional<HttpHost> proxy) throws IOException {
+ HttpClientContext httpClientContext = HttpClientContext.create();
+ boolean credentialsAdded = false;
+ CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
+
+ // Add the basic auth credentials if the rest config settings are present
+ if (restConfig.getBasicAuthUser() != null && restConfig.getBasicAuthPasswordPath() != null) {
+ String password = new String(readBytes(new Path(restConfig.getBasicAuthPasswordPath())), StandardCharsets.UTF_8);
+ credentialsProvider.setCredentials(
+ new AuthScope(target),
+ new UsernamePasswordCredentials(restConfig.getBasicAuthUser(), password));
+ credentialsAdded = true;
+ }
+
+ // Add the proxy basic auth credentials if the rest config settings are present
+ if (proxy.isPresent() && restConfig.getProxyBasicAuthUser() != null &&
+ restConfig.getProxyBasicAuthPasswordPath() != null) {
+ String password = new String(readBytes(new Path(restConfig.getProxyBasicAuthPasswordPath())), StandardCharsets.UTF_8);
+ credentialsProvider.setCredentials(
+ new AuthScope(proxy.get()),
+ new UsernamePasswordCredentials(restConfig.getProxyBasicAuthUser(), password));
+ credentialsAdded = true;
+ }
+ if (credentialsAdded) {
+ httpClientContext.setCredentialsProvider(credentialsProvider);
+ }
+ return httpClientContext;
+ }
+
+ /**
+ * Read bytes from a HDFS path.
+ * @param inPath
+ * @return
+ * @throws IOException
+ */
+ private byte[] readBytes(Path inPath) throws IOException {
+ FileSystem fs = FileSystem.get(inPath.toUri(), new Configuration());
+ try (FSDataInputStream inputStream = fs.open(inPath)) {
+ return IOUtils.toByteArray(inputStream);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/8bf3b6ec/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/resolver/BaseFunctionResolver.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/resolver/BaseFunctionResolver.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/resolver/BaseFunctionResolver.java
index aeed9d9..38a32d1 100644
--- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/resolver/BaseFunctionResolver.java
+++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/resolver/BaseFunctionResolver.java
@@ -23,6 +23,7 @@ import static java.lang.String.format;
import com.google.common.base.Joiner;
import com.google.common.base.Supplier;
import com.google.common.base.Suppliers;
+import java.io.IOException;
import java.io.Serializable;
import java.lang.invoke.MethodHandles;
import java.util.Arrays;
@@ -58,9 +59,15 @@ public abstract class BaseFunctionResolver implements FunctionResolver, Serializ
*/
protected Context context;
+ /**
+ * Indicates if closed has been called on this resolver.
+ */
+ private boolean closed;
+
public BaseFunctionResolver() {
// memoize provides lazy initialization and thread-safety (the ugly cast is necessary for serialization)
functions = Suppliers.memoize((Supplier<Map<String, StellarFunctionInfo>> & Serializable) this::resolveFunctions);
+ closed = false;
}
/**
@@ -95,6 +102,43 @@ public abstract class BaseFunctionResolver implements FunctionResolver, Serializ
}
/**
+ * Makes an attempt to close all Stellar functions. Calling close multiple times has no effect.
+ * @throws IOException Catches all exceptions and summarizes them.
+ */
+ @Override
+ public void close() throws IOException {
+ if (!closed) {
+ LOG.info("Calling close() on Stellar functions.");
+ Map<String, Throwable> errors = new HashMap<>();
+ for (StellarFunctionInfo info : getFunctionInfo()) {
+ try {
+ info.getFunction().close();
+ } catch (Throwable t) {
+ errors.put(info.getName(), t);
+ }
+ }
+ if (!errors.isEmpty()) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("Unable to close Stellar functions:");
+ for (Map.Entry<String, Throwable> e : errors.entrySet()) {
+ Throwable throwable = e.getValue();
+ String eText = String
+ .format("Exception - Function: %s; Message: %s; Cause: %s", e.getKey(),
+ throwable.getMessage(),
+ throwable.getCause());
+ sb.append(System.lineSeparator());
+ sb.append(eText);
+ }
+ closed = true;
+ throw new IOException(sb.toString());
+ }
+ closed = true;
+ } else {
+ LOG.info("close() already called on Stellar functions - skipping.");
+ }
+ }
+
+ /**
* Resolves a function by name.
* @param functionName The name of the function to resolve.
* @return The executable StellarFunction.
http://git-wip-us.apache.org/repos/asf/metron/blob/8bf3b6ec/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/resolver/FunctionResolver.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/resolver/FunctionResolver.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/resolver/FunctionResolver.java
index 5acb42c..4047586 100644
--- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/resolver/FunctionResolver.java
+++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/dsl/functions/resolver/FunctionResolver.java
@@ -17,16 +17,17 @@
*/
package org.apache.metron.stellar.dsl.functions.resolver;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.function.Function;
import org.apache.metron.stellar.dsl.Context;
import org.apache.metron.stellar.dsl.StellarFunction;
import org.apache.metron.stellar.dsl.StellarFunctionInfo;
-import java.util.function.Function;
-
/**
* Responsible for function resolution in Stellar.
*/
-public interface FunctionResolver extends Function<String, StellarFunction> {
+public interface FunctionResolver extends Function<String, StellarFunction>, Closeable {
/**
* Provides metadata about each Stellar function that is resolvable.
@@ -43,4 +44,11 @@ public interface FunctionResolver extends Function<String, StellarFunction> {
* @param context Context used to initialize.
*/
void initialize(Context context);
+
+ /**
+ * Perform any cleanup necessary for the loaded Stellar functions.
+ */
+ @Override
+ default void close() throws IOException {}
+
}
http://git-wip-us.apache.org/repos/asf/metron/blob/8bf3b6ec/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/BasicStellarTest.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/BasicStellarTest.java b/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/BasicStellarTest.java
index dec05a8..79f97bc 100644
--- a/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/BasicStellarTest.java
+++ b/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/BasicStellarTest.java
@@ -18,9 +18,20 @@
package org.apache.metron.stellar.dsl.functions;
+import static org.apache.metron.stellar.common.utils.StellarProcessorUtils.run;
+import static org.apache.metron.stellar.common.utils.StellarProcessorUtils.runPredicate;
+import static org.apache.metron.stellar.common.utils.StellarProcessorUtils.validate;
+
import com.google.common.base.Joiner;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
import org.apache.commons.lang3.StringUtils;
import org.apache.metron.stellar.common.StellarProcessor;
import org.apache.metron.stellar.dsl.Context;
@@ -37,12 +48,6 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
-import java.util.*;
-
-import static org.apache.metron.stellar.common.utils.StellarProcessorUtils.run;
-import static org.apache.metron.stellar.common.utils.StellarProcessorUtils.validate;
-import static org.apache.metron.stellar.common.utils.StellarProcessorUtils.runPredicate;
-
@SuppressWarnings("ALL")
public class BasicStellarTest {
@@ -70,6 +75,7 @@ public class BasicStellarTest {
public boolean isInitialized() {
return true;
}
+
}
@Stellar(
@@ -96,6 +102,7 @@ public class BasicStellarTest {
public boolean isInitialized() {
return true;
}
+
}
@Test
@@ -1000,4 +1007,5 @@ public class BasicStellarTest {
checkFalsey("{}");
checkFalsey("LIST_ADD([])");
}
+
}
http://git-wip-us.apache.org/repos/asf/metron/blob/8bf3b6ec/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/DateFunctionsTest.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/DateFunctionsTest.java b/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/DateFunctionsTest.java
index 1f1f4f4..48b2995 100644
--- a/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/DateFunctionsTest.java
+++ b/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/DateFunctionsTest.java
@@ -28,9 +28,12 @@ import org.apache.metron.stellar.dsl.StellarFunctions;
import org.junit.Before;
import org.junit.Test;
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
import java.util.Calendar;
import java.util.HashMap;
import java.util.Map;
+import java.util.TimeZone;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -225,4 +228,47 @@ public class DateFunctionsTest {
public void testDayOfYearNull() {
Object result = run("DAY_OF_YEAR(nada)");
}
+
+ @Test
+ public void testDateFormat() {
+ Object result = run("DATE_FORMAT('EEE MMM dd yyyy hh:mm:ss zzz', epoch, 'EST')");
+ assertEquals("Thu Aug 25 2016 08:27:10 EST", result);
+ }
+
+ /**
+ * Test that the String returned is formatted as specified.
+ * LocalDate.parse will throw if it is not.
+ * @throws Exception
+ */
+ @Test
+ public void testDateFormatDefault() throws Exception {
+ Object result = run("DATE_FORMAT('EEE MMM dd yyyy hh:mm:ss zzzz')");
+ DateTimeFormatter formatter = DateTimeFormatter.ofPattern("EEE MMM dd yyyy hh:mm:ss zzzz");
+ LocalDate.parse(result.toString(), formatter);
+ }
+
+ @Test
+ public void testDateFormatNow() {
+ Object result = run("DATE_FORMAT('EEE MMM dd yyyy hh:mm:ss zzz', 'GMT')");
+ assertTrue(result.toString().endsWith("GMT"));
+ }
+
+ @Test
+ public void testDateFormatDefaultTimezone() {
+ Object result = run("DATE_FORMAT('EEE MMM dd yyyy hh:mm:ss zzzz', epoch)");
+ assertTrue(result.toString().endsWith(TimeZone.getDefault().getDisplayName(true, 1)));
+ }
+
+ /**
+ * If refer to variable that does not exist, expect ParseException.
+ */
+ @Test(expected = ParseException.class)
+ public void testDateFormatNull() {
+ Object result = run("DATE_FORMAT('EEE MMM dd yyyy hh:mm:ss zzz', nada, 'EST')");
+ }
+
+ @Test(expected = ParseException.class)
+ public void testDateFormatInvalid() {
+ Object result = run("DATE_FORMAT('INVALID DATE FORMAT', epoch, 'EST')");
+ }
}
http://git-wip-us.apache.org/repos/asf/metron/blob/8bf3b6ec/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/RestFunctionsTest.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/RestFunctionsTest.java b/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/RestFunctionsTest.java
new file mode 100644
index 0000000..4b912ef
--- /dev/null
+++ b/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/RestFunctionsTest.java
@@ -0,0 +1,601 @@
+/**
+ * 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.stellar.dsl.functions;
+
+import org.adrianwalker.multilinestring.Multiline;
+import org.apache.commons.io.FileUtils;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.protocol.HttpClientContext;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+import org.apache.metron.stellar.dsl.Context;
+import org.apache.metron.stellar.dsl.ParseException;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.mockserver.client.server.MockServerClient;
+import org.mockserver.junit.MockServerRule;
+import org.mockserver.junit.ProxyRule;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.apache.metron.stellar.common.utils.StellarProcessorUtils.run;
+import static org.apache.metron.stellar.dsl.functions.RestConfig.BASIC_AUTH_PASSWORD_PATH;
+import static org.apache.metron.stellar.dsl.functions.RestConfig.BASIC_AUTH_USER;
+import static org.apache.metron.stellar.dsl.functions.RestConfig.CONNECTION_REQUEST_TIMEOUT;
+import static org.apache.metron.stellar.dsl.functions.RestConfig.CONNECT_TIMEOUT;
+import static org.apache.metron.stellar.dsl.functions.RestConfig.POOLING_DEFAULT_MAX_PER_RUOTE;
+import static org.apache.metron.stellar.dsl.functions.RestConfig.POOLING_MAX_TOTAL;
+import static org.apache.metron.stellar.dsl.functions.RestConfig.PROXY_BASIC_AUTH_PASSWORD_PATH;
+import static org.apache.metron.stellar.dsl.functions.RestConfig.PROXY_BASIC_AUTH_USER;
+import static org.apache.metron.stellar.dsl.functions.RestConfig.PROXY_HOST;
+import static org.apache.metron.stellar.dsl.functions.RestConfig.PROXY_PORT;
+import static org.apache.metron.stellar.dsl.functions.RestConfig.SOCKET_TIMEOUT;
+import static org.apache.metron.stellar.dsl.functions.RestConfig.STELLAR_REST_SETTINGS;
+import static org.apache.metron.stellar.dsl.functions.RestConfig.TIMEOUT;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockserver.model.HttpRequest.request;
+import static org.mockserver.model.HttpResponse.response;
+
+/**
+ * Tests the RestFunctions class.
+ */
+public class RestFunctionsTest {
+
+ @Rule
+ public ExpectedException thrown = ExpectedException.none();
+
+ @Rule
+ public MockServerRule mockServerRule = new MockServerRule(this);
+
+ @Rule
+ public ProxyRule proxyRule = new ProxyRule(1080, this);
+
+ private MockServerClient mockServerClient;
+ private String getUri;
+ private Context context;
+
+ private String basicAuthPasswordPath = "./target/basicAuth.txt";
+ private String basicAuthPassword = "password";
+ private String proxyBasicAuthPasswordPath = "./target/proxyBasicAuth.txt";
+ private String proxyAuthPassword = "proxyPassword";
+
+ @Before
+ public void setup() throws Exception {
+ context = new Context.Builder()
+ .with(Context.Capabilities.GLOBAL_CONFIG, HashMap::new)
+ .build();
+
+ // Store the passwords in the local file system
+ FileUtils.writeStringToFile(new File(basicAuthPasswordPath), basicAuthPassword, StandardCharsets.UTF_8);
+ FileUtils.writeStringToFile(new File(proxyBasicAuthPasswordPath), proxyAuthPassword, StandardCharsets.UTF_8);
+
+ // By default, the mock server expects a GET request with the path set to /get
+ getUri = String.format("http://localhost:%d/get", mockServerRule.getPort());
+ mockServerClient.when(
+ request()
+ .withMethod("GET")
+ .withPath("/get"))
+ .respond(response()
+ .withBody("{\"get\":\"success\"}"));
+ }
+
+ /**
+ * The REST_GET function should perform a get request and parse the results.
+ */
+ @Test
+ public void restGetShouldSucceed() throws Exception {
+ Map<String, Object> actual = (Map<String, Object>) run(String.format("REST_GET('%s')", getUri), context);
+
+ assertEquals(1, actual.size());
+ assertEquals("success", actual.get("get"));
+ }
+
+ /**
+ * The REST_GET function should perform a get request using a proxy and parse the results.
+ */
+ @Test
+ public void restGetShouldSucceedWithProxy() {
+ mockServerClient.when(
+ request()
+ .withMethod("GET")
+ .withPath("/get"))
+ .respond(response()
+ .withBody("{\"proxyGet\":\"success\"}"));
+
+ context.addCapability(Context.Capabilities.GLOBAL_CONFIG, () -> new HashMap<String, Object>() {{
+ put(PROXY_HOST, "localhost");
+ put(PROXY_PORT, proxyRule.getHttpPort());
+ }});
+
+ Map<String, Object> actual = (Map<String, Object>) run(String.format("REST_GET('%s')", getUri), context);
+
+ assertEquals(1, actual.size());
+ assertEquals("success", actual.get("proxyGet"));
+ }
+
+ /**
+ * The REST_GET function should handle an error status code and return null by default.
+ */
+ @Test
+ public void restGetShouldHandleErrorStatusCode() {
+ mockServerClient.when(
+ request()
+ .withMethod("GET")
+ .withPath("/get"))
+ .respond(response()
+ .withStatusCode(403));
+
+ assertNull(run(String.format("REST_GET('%s')", getUri), context));
+ }
+
+ /**
+ * {
+ * "response.codes.allowed": [200,404],
+ * "empty.content.override": {}
+ * }
+ */
+ @Multiline
+ private String emptyContentOverride;
+
+ /**
+ * The REST_GET function should return the empty content override setting when status is allowed and content is empty.
+ */
+ @Test
+ public void restGetShouldReturnEmptyContentOverride() {
+ mockServerClient.when(
+ request()
+ .withMethod("GET")
+ .withPath("/get"))
+ .respond(response()
+ .withStatusCode(404));
+
+ assertEquals(new HashMap<>(), run(String.format("REST_GET('%s', %s)", getUri, emptyContentOverride), context));
+ }
+
+ /**
+ * {
+ * "error.value.override": "error message"
+ * }
+ */
+ @Multiline
+ private String errorValueOverride;
+
+ /**
+ * The REST_GET function should return the error value override setting on error.
+ */
+ @Test
+ public void restGetShouldReturnErrorValueOverride() {
+ mockServerClient.when(
+ request()
+ .withMethod("GET")
+ .withPath("/get"))
+ .respond(response()
+ .withStatusCode(500));
+
+ Object result = run(String.format("REST_GET('%s', %s)", getUri, errorValueOverride), context);
+ assertEquals("error message" , result);
+ }
+
+ /**
+ * The REST_GET function should return a proxy HttpHost if the correct settings are present.
+ */
+ @Test
+ public void restGetShouldGetProxy() {
+ RestFunctions.RestGet restGet = new RestFunctions.RestGet();
+
+ {
+ RestConfig restConfig = new RestConfig();
+ Optional<HttpHost> actual = restGet.getProxy(restConfig);
+
+ assertEquals(Optional.empty(), actual);
+ }
+
+ {
+ RestConfig restConfig = new RestConfig();
+ restConfig.put(PROXY_HOST, "localhost");
+ Optional<HttpHost> actual = restGet.getProxy(restConfig);
+
+ assertEquals(Optional.empty(), actual);
+ }
+
+ {
+ RestConfig restConfig = new RestConfig();
+ restConfig.put(PROXY_PORT, proxyRule.getHttpPort());
+ Optional<HttpHost> actual = restGet.getProxy(restConfig);
+
+ assertEquals(Optional.empty(), actual);
+ }
+
+ {
+ RestConfig restConfig = new RestConfig();
+ restConfig.put(PROXY_HOST, "localhost");
+ restConfig.put(PROXY_PORT, proxyRule.getHttpPort());
+ Optional<HttpHost> actual = restGet.getProxy(restConfig);
+
+ assertEquals(new HttpHost("localhost", proxyRule.getHttpPort()), actual.get());
+ }
+ }
+
+ /**
+ * The REST_GET function should return settings in the correct order of precedence.
+ * @throws Exception
+ */
+ @Test
+ public void restGetShouldGetRestConfig() throws Exception {
+ RestFunctions.RestGet restGet = new RestFunctions.RestGet();
+
+ {
+ // Test for default timeout
+ RestConfig restConfig = restGet.getRestConfig(Collections.singletonList("uri"), new HashMap<>());
+
+ assertEquals(2, restConfig.size());
+ assertEquals(1000, restConfig.getTimeout().intValue());
+ assertEquals(Collections.singletonList(200), restConfig.getResponseCodesAllowed());
+ assertNull(restConfig.getBasicAuthUser());
+ }
+
+ Map<String, Object> globalRestConfig = new HashMap<String, Object>() {{
+ put(STELLAR_REST_SETTINGS, new HashMap<String, Object>() {{
+ put(SOCKET_TIMEOUT, 2000);
+ put(BASIC_AUTH_USER, "globalUser");
+ put(PROXY_HOST, "globalHost");
+ }});
+ }};
+
+ // Global config settings should take effect
+ {
+ RestConfig restConfig = restGet.getRestConfig(Collections.singletonList("uri"), globalRestConfig);
+
+ assertEquals(5, restConfig.size());
+ assertEquals(1000, restConfig.getTimeout().intValue());
+ assertEquals(Collections.singletonList(200), restConfig.getResponseCodesAllowed());
+ assertEquals(2000, restConfig.getSocketTimeout().intValue());
+ assertEquals("globalUser", restConfig.getBasicAuthUser());
+ assertEquals("globalHost", restConfig.getProxyHost());
+ }
+
+ Map<String, Object> functionRestConfig = new HashMap<String, Object>() {{
+ put(SOCKET_TIMEOUT, 1);
+ put(BASIC_AUTH_USER, "functionUser");
+ put(TIMEOUT, 100);
+ }};
+
+
+ // Function call settings should override global settings
+ {
+ RestConfig restConfig = restGet.getRestConfig(Arrays.asList("uri", functionRestConfig), globalRestConfig);
+
+ assertEquals(5, restConfig.size());
+ assertEquals(Collections.singletonList(200), restConfig.getResponseCodesAllowed());
+ assertEquals(100, restConfig.getTimeout().intValue());
+ assertEquals(1, restConfig.getSocketTimeout().intValue());
+ assertEquals("functionUser", restConfig.getBasicAuthUser());
+ assertEquals("globalHost", restConfig.getProxyHost());
+ }
+
+ functionRestConfig = new HashMap<String, Object>() {{
+ put(BASIC_AUTH_USER, "functionUser");
+ put(TIMEOUT, 100);
+ }};
+
+ // New function call settings should take effect with global settings staying the same
+ {
+ RestConfig restConfig = restGet.getRestConfig(Arrays.asList("uri", functionRestConfig), globalRestConfig);
+
+ assertEquals(5, restConfig.size());
+ assertEquals(Collections.singletonList(200), restConfig.getResponseCodesAllowed());
+ assertEquals(100, restConfig.getTimeout().intValue());
+ assertEquals(2000, restConfig.getSocketTimeout().intValue());
+ assertEquals("functionUser", restConfig.getBasicAuthUser());
+ assertEquals("globalHost", restConfig.getProxyHost());
+ }
+
+ globalRestConfig = new HashMap<String, Object>() {{
+ put(STELLAR_REST_SETTINGS, new HashMap<String, Object>() {{
+ put(SOCKET_TIMEOUT, 2000);
+ put(BASIC_AUTH_USER, "globalUser");
+ }});
+ }};
+
+ // New global settings should take effect with function call settings staying the same
+ {
+ RestConfig restConfig = restGet.getRestConfig(Arrays.asList("uri", functionRestConfig), globalRestConfig);
+
+ assertEquals(4, restConfig.size());
+ assertEquals(Collections.singletonList(200), restConfig.getResponseCodesAllowed());
+ assertEquals(100, restConfig.getTimeout().intValue());
+ assertEquals(2000, restConfig.getSocketTimeout().intValue());
+ assertEquals("functionUser", restConfig.getBasicAuthUser());
+ }
+
+ // Should fall back to global settings on missing function call config
+ {
+ RestConfig restConfig = restGet.getRestConfig(Collections.singletonList("uri"), globalRestConfig);
+
+ assertEquals(4, restConfig.size());
+ assertEquals(Collections.singletonList(200), restConfig.getResponseCodesAllowed());
+ assertEquals(1000, restConfig.getTimeout().intValue());
+ assertEquals(2000, restConfig.getSocketTimeout().intValue());
+ assertEquals("globalUser", restConfig.getBasicAuthUser());
+ }
+
+ // Should fall back to default settings on missing global settings
+ {
+ RestConfig restConfig = restGet.getRestConfig(Collections.singletonList("uri"), new HashMap<>());
+
+ assertEquals(2, restConfig.size());
+ assertEquals(Collections.singletonList(200), restConfig.getResponseCodesAllowed());
+ assertEquals(1000, restConfig.getTimeout().intValue());
+ }
+
+ }
+
+ /**
+ * The REST_GET function should properly set the HttpClient timeout settings and proxy
+ */
+ @Test
+ public void restGetShouldGetRequestConfig() {
+ RestFunctions.RestGet restGet = new RestFunctions.RestGet();
+
+ {
+ RequestConfig actual = restGet.getRequestConfig(new RestConfig(), Optional.empty());
+ RequestConfig expected = RequestConfig.custom().build();
+
+ assertEquals(expected.getConnectTimeout(), actual.getConnectTimeout());
+ assertEquals(expected.getConnectionRequestTimeout(), actual.getConnectionRequestTimeout());
+ assertEquals(expected.getSocketTimeout(), actual.getSocketTimeout());
+ assertEquals(expected.getProxy(), actual.getProxy());
+ }
+
+ {
+ RestConfig restConfig = new RestConfig();
+ restConfig.put(CONNECT_TIMEOUT, 1);
+ restConfig.put(CONNECTION_REQUEST_TIMEOUT, 2);
+ restConfig.put(SOCKET_TIMEOUT, 3);
+ HttpHost proxy = new HttpHost("localhost", proxyRule.getHttpPort());
+ Optional<HttpHost> proxyOptional = Optional.of(proxy);
+
+ RequestConfig actual = restGet.getRequestConfig(restConfig, proxyOptional);
+ RequestConfig expected = RequestConfig.custom()
+ .setConnectTimeout(1)
+ .setConnectionRequestTimeout(2)
+ .setSocketTimeout(3)
+ .setProxy(proxy)
+ .build();
+
+ assertEquals(expected.getConnectTimeout(), actual.getConnectTimeout());
+ assertEquals(expected.getConnectionRequestTimeout(), actual.getConnectionRequestTimeout());
+ assertEquals(expected.getSocketTimeout(), actual.getSocketTimeout());
+ assertEquals(expected.getProxy(), actual.getProxy());
+ }
+
+ }
+
+ /**
+ * The REST_GET function should set the proper credentials in the HttpClientContext.
+ * @throws Exception
+ */
+ @Test
+ public void restGetShouldGetHttpClientContext() throws Exception {
+ RestFunctions.RestGet restGet = new RestFunctions.RestGet();
+ HttpHost target = new HttpHost("localhost", mockServerRule.getPort());
+ HttpHost proxy = new HttpHost("localhost", proxyRule.getHttpPort());
+
+ {
+ RestConfig restConfig = new RestConfig();
+ HttpClientContext actual = restGet.getHttpClientContext(restConfig, target, Optional.empty());
+
+ assertNull(actual.getCredentialsProvider());
+ }
+
+ {
+ RestConfig restConfig = new RestConfig();
+ restConfig.put(BASIC_AUTH_USER, "user");
+ restConfig.put(BASIC_AUTH_PASSWORD_PATH, basicAuthPasswordPath);
+
+ HttpClientContext actual = restGet.getHttpClientContext(restConfig, target, Optional.empty());
+ HttpClientContext expected = HttpClientContext.create();
+ CredentialsProvider expectedCredentialsProvider = new BasicCredentialsProvider();
+ expectedCredentialsProvider.setCredentials(
+ new AuthScope(target),
+ new UsernamePasswordCredentials(restConfig.getBasicAuthUser(), basicAuthPassword));
+ expected.setCredentialsProvider(expectedCredentialsProvider);
+
+ assertEquals(expected.getCredentialsProvider().getCredentials(new AuthScope(target)),
+ actual.getCredentialsProvider().getCredentials(new AuthScope(target)));
+ assertEquals(expected.getCredentialsProvider().getCredentials(new AuthScope(proxy)),
+ actual.getCredentialsProvider().getCredentials(new AuthScope(proxy)));
+ }
+
+ {
+ RestConfig restConfig = new RestConfig();
+ restConfig.put(PROXY_BASIC_AUTH_USER, "proxyUser");
+ restConfig.put(PROXY_BASIC_AUTH_PASSWORD_PATH, proxyBasicAuthPasswordPath);
+
+ HttpClientContext actual = restGet.getHttpClientContext(restConfig, target, Optional.of(proxy));
+ HttpClientContext expected = HttpClientContext.create();
+ CredentialsProvider expectedCredentialsProvider = new BasicCredentialsProvider();
+ expectedCredentialsProvider.setCredentials(
+ new AuthScope(proxy),
+ new UsernamePasswordCredentials(restConfig.getProxyBasicAuthUser(), proxyAuthPassword));
+ expected.setCredentialsProvider(expectedCredentialsProvider);
+
+ assertEquals(expected.getCredentialsProvider().getCredentials(new AuthScope(target)),
+ actual.getCredentialsProvider().getCredentials(new AuthScope(target)));
+ assertEquals(expected.getCredentialsProvider().getCredentials(new AuthScope(proxy)),
+ actual.getCredentialsProvider().getCredentials(new AuthScope(proxy)));
+ }
+
+ {
+ RestConfig restConfig = new RestConfig();
+ restConfig.put(BASIC_AUTH_USER, "user");
+ restConfig.put(BASIC_AUTH_PASSWORD_PATH, basicAuthPasswordPath);
+ restConfig.put(PROXY_BASIC_AUTH_USER, "proxyUser");
+ restConfig.put(PROXY_BASIC_AUTH_PASSWORD_PATH, proxyBasicAuthPasswordPath);
+
+ HttpClientContext actual = restGet.getHttpClientContext(restConfig, target, Optional.of(proxy));
+ HttpClientContext expected = HttpClientContext.create();
+ CredentialsProvider expectedCredentialsProvider = new BasicCredentialsProvider();
+ expectedCredentialsProvider.setCredentials(
+ new AuthScope(target),
+ new UsernamePasswordCredentials(restConfig.getBasicAuthUser(), basicAuthPassword));
+ expectedCredentialsProvider.setCredentials(
+ new AuthScope(proxy),
+ new UsernamePasswordCredentials(restConfig.getProxyBasicAuthUser(), proxyAuthPassword));
+ expected.setCredentialsProvider(expectedCredentialsProvider);
+
+ assertEquals(expected.getCredentialsProvider().getCredentials(new AuthScope(target)),
+ actual.getCredentialsProvider().getCredentials(new AuthScope(target)));
+ assertEquals(expected.getCredentialsProvider().getCredentials(new AuthScope(proxy)),
+ actual.getCredentialsProvider().getCredentials(new AuthScope(proxy)));
+ }
+ }
+
+ /**
+ * The REST_GET function should timeout and return null.
+ */
+ @Test
+ public void restGetShouldTimeout() {
+ String uri = String.format("http://localhost:%d/get", mockServerRule.getPort());
+
+ mockServerClient.when(
+ request()
+ .withMethod("GET")
+ .withPath("/get"))
+ .respond(response()
+ .withBody("{\"get\":\"success\"}"));
+
+ Map<String, Object> globalConfig = new HashMap<String, Object>() {{
+ put(STELLAR_REST_SETTINGS, new HashMap<String, Object>() {{
+ put(TIMEOUT, 1);
+ }});
+ }};
+
+ context.addCapability(Context.Capabilities.GLOBAL_CONFIG, () -> globalConfig);
+
+ Map<String, Object> actual = (Map<String, Object>) run(String.format("REST_GET('%s')", uri), context);
+ assertNull(actual);
+ }
+
+ /**
+ * {
+ * "timeout": 1
+ * }
+ */
+ @Multiline
+ private String timeoutConfig;
+
+ /**
+ * The REST_GET function should honor the function supplied timeout setting.
+ */
+ @Test
+ public void restGetShouldTimeoutWithSuppliedTimeout() {
+ String expression = String.format("REST_GET('%s', %s)", getUri, timeoutConfig);
+ Map<String, Object> actual = (Map<String, Object>) run(expression, context);
+ assertNull(actual);
+ }
+
+ /**
+ * The REST_GET function should throw an exception on a malformed uri.
+ * @throws IllegalArgumentException
+ * @throws IOException
+ */
+ @Test
+ public void restGetShouldHandleURISyntaxException() throws IllegalArgumentException, IOException {
+ thrown.expect(ParseException.class);
+ thrown.expectMessage("Unable to parse REST_GET('some invalid uri'): Unable to parse: REST_GET('some invalid uri') due to: Illegal character in path at index 4: some invalid uri");
+
+ run("REST_GET('some invalid uri')", context);
+ }
+
+ /**
+ * The REST_GET function should handle IOExceptions and return null.
+ * @throws IllegalArgumentException
+ * @throws IOException
+ */
+ @Test
+ public void restGetShouldHandleIOException() throws IllegalArgumentException, IOException {
+ RestFunctions.RestGet restGet = spy(RestFunctions.RestGet.class);
+ doThrow(new IOException("io exception")).when(restGet).doGet(any(RestConfig.class), any(HttpGet.class), any(HttpClientContext.class));
+
+ Object result = restGet.apply(Collections.singletonList(getUri), context);
+ Assert.assertNull(result);
+ }
+
+ /**
+ * The REST_GET function should throw an exception when the required uri parameter is missing.
+ */
+ @Test
+ public void restGetShouldThrownExceptionOnMissingParameter() {
+ thrown.expect(ParseException.class);
+ thrown.expectMessage("Unable to parse REST_GET(): Unable to parse: REST_GET() due to: Expected at least 1 argument(s), found 0");
+
+ run("REST_GET()", context);
+ }
+
+ @Test
+ public void restGetShouldGetPoolingConnectionManager() {
+ RestFunctions.RestGet restGet = new RestFunctions.RestGet();
+
+ RestConfig restConfig = new RestConfig();
+ restConfig.put(POOLING_MAX_TOTAL, 5);
+ restConfig.put(POOLING_DEFAULT_MAX_PER_RUOTE, 2);
+
+ PoolingHttpClientConnectionManager cm = restGet.getConnectionManager(restConfig);
+
+ assertEquals(5, cm.getMaxTotal());
+ assertEquals(2, cm.getDefaultMaxPerRoute());
+ }
+
+ @Test
+ public void restGetShouldCloseHttpClient() throws Exception {
+ RestFunctions.RestGet restGet = new RestFunctions.RestGet();
+ CloseableHttpClient httpClient = mock(CloseableHttpClient.class);
+
+ restGet.setHttpClient(httpClient);
+ restGet.close();
+
+ verify(httpClient, times(1)).close();
+ verifyNoMoreInteractions(httpClient);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/8bf3b6ec/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/resolver/BaseFunctionResolverTest.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/resolver/BaseFunctionResolverTest.java b/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/resolver/BaseFunctionResolverTest.java
new file mode 100644
index 0000000..47cbda3
--- /dev/null
+++ b/metron-stellar/stellar-common/src/test/java/org/apache/metron/stellar/dsl/functions/resolver/BaseFunctionResolverTest.java
@@ -0,0 +1,169 @@
+/**
+ * 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.stellar.dsl.functions.resolver;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.metron.stellar.dsl.BaseStellarFunction;
+import org.apache.metron.stellar.dsl.Stellar;
+import org.apache.metron.stellar.dsl.StellarFunction;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+public class BaseFunctionResolverTest {
+
+ public static class TestResolver extends BaseFunctionResolver {
+
+ Set<Class<? extends StellarFunction>> classesToResolve = new HashSet<>();
+
+ @Override
+ public Set<Class<? extends StellarFunction>> resolvables() {
+ return classesToResolve;
+ }
+
+ /**
+ * Will attempt to resolve any Stellar functions defined within the specified class.
+ * @param clazz The class which may contain a Stellar function.
+ */
+ public TestResolver withClass(Class<? extends StellarFunction> clazz) {
+ this.classesToResolve.add(clazz);
+ return this;
+ }
+ }
+
+ /**
+ * Often imitated, never duplicated.
+ */
+ @Stellar(namespace = "namespace", name = "afunction", description = "description", returns = "returns", params = {
+ "param1"})
+ private static class IAmAFunction extends BaseStellarFunction {
+
+ public static int closeCallCount;
+ public static boolean throwException = false; // init here bc of reflection in resolver.
+
+ public IAmAFunction() {
+ closeCallCount = 0;
+ }
+
+ @Override
+ public Object apply(List<Object> args) {
+ return null;
+ }
+
+ @Override
+ public void close() throws IOException {
+ closeCallCount++;
+ if (throwException) {
+ Throwable cause = new Throwable("Some nasty nasty cause.");
+ throw new IOException("Bad things happened", cause);
+ }
+ }
+ }
+
+ /**
+ * Scratch that. I was wrong.
+ */
+ @Stellar(namespace = "namespace", name = "anotherfunction", description = "description", returns = "returns", params = {
+ "param1"})
+ private static class IAmAnotherFunction extends BaseStellarFunction {
+
+ public static int closeCallCount;
+ public static boolean throwException = false; // init here bc of reflection in resolver.
+
+ public IAmAnotherFunction() {
+ closeCallCount = 0;
+ }
+
+ @Override
+ public Object apply(List<Object> args) {
+ return null;
+ }
+
+ @Override
+ public void close() throws IOException {
+ closeCallCount++;
+ if (throwException) {
+ throw new NullPointerException("A most annoying exception.");
+ }
+ }
+ }
+
+ private TestResolver resolver;
+
+ @Before
+ public void setup() {
+ resolver = new TestResolver();
+ IAmAFunction.throwException = false;
+ IAmAnotherFunction.throwException = false;
+ }
+
+ @Test
+ public void close_calls_all_loaded_function_close_methods() throws IOException {
+ resolver.withClass(IAmAFunction.class);
+ resolver.withClass(IAmAnotherFunction.class);
+ resolver.close();
+ assertThat(IAmAFunction.closeCallCount, equalTo(1));
+ assertThat(IAmAnotherFunction.closeCallCount, equalTo(1));
+ }
+
+ @Rule
+ public final ExpectedException exception = ExpectedException.none();
+
+ @Test
+ public void close_collects_all_exceptions_thrown_on_loaded_function_close_methods()
+ throws IOException {
+ IAmAFunction.throwException = true;
+ IAmAnotherFunction.throwException = true;
+ resolver.withClass(IAmAFunction.class);
+ resolver.withClass(IAmAnotherFunction.class);
+ exception.expect(IOException.class);
+ resolver.close();
+ }
+
+ @Test
+ public void close_only_throws_exceptions_on_first_invocation()
+ throws IOException {
+ IAmAFunction.throwException = true;
+ IAmAnotherFunction.throwException = true;
+ resolver.withClass(IAmAFunction.class);
+ resolver.withClass(IAmAnotherFunction.class);
+ try {
+ resolver.close();
+ Assert.fail("Should have thrown an exception.");
+ } catch (IOException e) {
+ // intentionally empty
+ }
+ assertThat(IAmAFunction.closeCallCount, equalTo(1));
+ assertThat(IAmAnotherFunction.closeCallCount, equalTo(1));
+ // should not throw exceptions or call any function's close again.
+ resolver.close();
+ resolver.close();
+ resolver.close();
+ assertThat(IAmAFunction.closeCallCount, equalTo(1));
+ assertThat(IAmAnotherFunction.closeCallCount, equalTo(1));
+ }
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/8bf3b6ec/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 8fe1786..9a8bd6a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -126,7 +126,7 @@
<global_solr_version>6.6.2</global_solr_version>
<global_mockito_version>1.10.19</global_mockito_version>
<global_powermock_version>1.7.0</global_powermock_version>
- <global_shade_version>3.1.1</global_shade_version>
+ <global_shade_version>3.2.0</global_shade_version>
<global_jackson_version>2.7.4</global_jackson_version>
<global_errorprone_core_version>2.0.14</global_errorprone_core_version>
<global_jar_version>3.0.2</global_jar_version>
@@ -137,8 +137,9 @@
<global_reflections_version>0.9.10</global_reflections_version>
<global_checkstyle_version>8.0</global_checkstyle_version>
<global_log4j_core_version>2.1</global_log4j_core_version>
- <global_simple_syslog_version>0.0.8</global_simple_syslog_version>
+ <global_simple_syslog_version>0.0.9</global_simple_syslog_version>
<global_spark_version>2.3.1</global_spark_version>
+ <global_httpclient_version>4.3.2</global_httpclient_version>
</properties>
<profiles>
http://git-wip-us.apache.org/repos/asf/metron/blob/8bf3b6ec/site/develop/bylaws.md
----------------------------------------------------------------------
diff --git a/site/develop/bylaws.md b/site/develop/bylaws.md
deleted file mode 100644
index a8fc5fb..0000000
--- a/site/develop/bylaws.md
+++ /dev/null
@@ -1,281 +0,0 @@
----
-layout: page
-title: Apache Metron Bylaws
----
-
-## Introduction
-
-This document defines the bylaws under which the Apache Metron project
-operates. It defines the roles and responsibilities of the project,
-who may vote, how voting works, how conflicts are resolved, etc.
-
-Metron is a project of the Apache Software Foundation (ASF) and the foundation
-holds the trademark on the name "Metron" and copyright on the combined
-code base. The [Apache Foundation
-FAQ](http://www.apache.org/foundation/faq.html) and
-[How-It-Works](http://www.apache.org/foundation/how-it-works.html)
-explain the operation and background of the foundation.
-
-Apache has a [code of
-conduct](http://www.apache.org/foundation/policies/conduct.html) that
-it expects its members to follow. In particular:
-
-* Be **open and welcoming**. It is important that we grow and
- encourage the community of users and developers for our project.
-
-* Be **collaborative**. Working together on the open mailing lists and
- bug database to make decisions helps the project grow.
-
-* Be **respectful** of others. Everyone is volunteering their time and
- efforts to work on this project. Please be respectful of everyone
- and their views.
-
-Metron is typical of Apache projects in that it operates under a set of
-principles, known collectively as the "Apache Way". If you are new to
-Apache development, please refer to
-[this](http://www.apache.org/foundation/how-it-works.html) for more
-information on how Apache projects operate.
-
-## Roles and Responsibilities
-
-Apache projects define a set of roles with associated rights and
-responsibilities. These roles govern what tasks an individual may
-perform within the project. The roles are defined in the following
-sections.
-
-### Users
-
-The most important participants in the project are people who use our
-software. The majority of our developers start out as users and guide
-their development efforts from the user's perspective. Users
-contribute to the Apache projects by providing feedback to developers
-in the form of bug reports and feature suggestions. As well, users
-participate in the Apache community by helping other users on mailing
-lists and user support forums.
-
-### Contributors
-
-Contributors include all of the volunteers who donate time, code,
-documentation,
-or resources to the Metron Project. A contributor that makes sustained,
-welcome contributions to the project may be invited to become a
-committer, though the exact timing of such invitations depends on many
-factors.
-
-### Committers
-
-The project's committers are responsible for the project's technical
-management. Committers have the right to commit to the project's git
-repository. Committers may cast binding votes on any technical
-discussion.
-
-Committer access is by invitation only and must be approved by
-consensus approval of the active Project Management Committee (PMC)
-members.
-
-If a committer wishes to leave the project or does not contribute to
-the project in any form for six months, the PMC may make them emeritus.
-Emeritus committers lose their ability to commit code or cast binding
-votes. An emeritus committer may
-request reinstatement of commit access from the PMC. Such
-reinstatement is subject to consensus approval of active PMC members.
-
-All Apache committers are required to have a signed [Individual
-Contributor License
-Agreement](https://www.apache.org/licenses/icla.txt) (ICLA) on file
-with the Apache Software Foundation. There is a [Committer
-FAQ](http://www.apache.org/dev/committers.html) which provides more
-details on the requirements for Committers.
-
-A committer who makes a
-sustained contribution to the project may be invited to become a
-member of the PMC. The form of contribution
-is not limited to code. It can also include code review, helping out
-users on the mailing lists, documentation, testing, etc.
-
-### Release Manager
-
-A Release Manager (RM) is a committer who volunteers to produce a
-Release Candidate. The RM shall publish a Release Plan on the
-dev mailing list stating the branch from which they intend to
-make a Release Candidate.
-
-### Project Management Committee
-
-The Project Management Committee (PMC) for Apache Metron was created by
-the Apache Board in December 2015 when Metron moved out of Cisco's OpenSOC
-project and became an incubated project at Apache.
-The PMC is responsible to the board and
-the ASF for the management and oversight of the Apache Metron
-codebase. The responsibilities of the PMC include
-
- * Deciding what is distributed as products of the Apache Metron
- project. In particular all releases must be approved by the PMC.
-
- * Maintaining the project's shared resources, including the codebase
- repository, mailing lists, and websites.
-
- * Speaking on behalf of the project.
-
- * Resolving license disputes regarding products of the project
-
- * Nominating new PMC members and committers
-
- * Maintaining these bylaws and other guidelines of the project
-
-Membership of the PMC is by invitation only and must be approved by a
-consensus approval of active PMC members.
-
-A PMC member is considered
-emeritus by their own declaration or by not contributing in any form
-to the project for over six months. An emeritus member may request
-reinstatement to the PMC. Such reinstatement is subject to consensus
-approval of the active PMC members.
-
-The chair of the PMC is appointed by the ASF board. The chair is an
-office holder of the Apache Software Foundation (Vice President,
-Apache Metron) and has primary responsibility to the board for the
-management of the project within the scope of the Metron PMC. The
-chair reports to the board quarterly on developments within the Metron
-project.
-
-When the project desires a new PMC chair, the PMC votes to recommend a
-new chair using [Single Transferable
-Vote](http://wiki.apache.org/general/BoardVoting) voting. The decision
-must be ratified by the Apache board.
-
-## Decision Making
-
-Within the Metron project, different types of decisions require
-different forms of approval. For example, the previous section
-describes several decisions which require "consensus approval."
-This section defines how voting is performed, the types of
-approvals, and which types of decision require which type of approval.
-
-### Voting
-
-Decisions regarding the project are made by votes on the primary
-project development mailing list (dev@Metron.apache.org). Where
-necessary, PMC voting may take place on the private Metron PMC mailing
-list. Votes are clearly indicated by subject line starting with
-[VOTE]. Votes may contain multiple items for approval and these should
-be clearly separated. Voting is carried out by replying to the vote
-mail. Voting may take five flavors:
-
-* **+1** -- "Yes," "Agree," or "the action should be performed." In general,
- this vote also indicates a willingness on the behalf of the voter in
- "making it happen."
-
-* **+0** -- This vote indicates a willingness for the action under
- consideration to go ahead. The voter, however, will not be able to
- help.
-
-* **0** -- The voter is neutral on the topic under discussion.
-
-* **-0** -- This vote indicates that the voter does not, in general, agree
- with the proposed action but is not concerned enough to prevent the
- action going ahead.
-
-* **-1** -- This is a negative vote. On issues where consensus is required,
- this vote counts as a veto. All vetoes must contain an explanation
- of why the veto is appropriate. Vetoes with no explanation are
- void. It may also be appropriate for a -1 vote to include an
- alternative course of action.
-
-All participants in the Metron project are encouraged to show their
-agreement for or against a particular action by voting, regardless of
-whether their vote is binding. Nonbinding votes are useful for
-encouraging discussion and understanding the scope of opinions within
-the project.
-
-### Approvals
-
-These are the types of approvals that can be sought. Different actions
-require different types of approvals.
-
-* **Consensus Approval** -- Consensus approval requires 3 binding +1
- votes and no binding vetoes.
-
-* **Lazy Consensus** -- Lazy consensus requires at least one +1 vote and
- no -1 votes ('silence gives assent').
-
-* **Lazy Majority** -- A lazy majority vote requires 3 binding +1 votes
- and more binding +1 votes than -1 votes.
-
-* **Lazy 2/3 Majority** -- Lazy 2/3 majority votes requires at least 3
- votes and twice as many +1 votes as -1 votes.
-
-### Vetoes
-
-A valid, binding veto cannot be overruled. If a veto is cast, it must
-be accompanied by a valid reason explaining the reasons for the
-veto. The validity of a veto, if challenged, can be confirmed by
-anyone who has a binding vote. This does not necessarily signify
-agreement with the veto - merely that the veto is valid. If you
-disagree with a valid veto, you must lobby the person casting the veto
-to withdraw their veto. If a veto is not withdrawn, any action that
-has already been taken must be reversed in a timely manner.
-
-### Actions
-
-This section describes the various actions which are undertaken within
-the project, the corresponding approval required for that action and
-those who have binding votes over the action.
-
-#### Code Change
-
-A change made to a codebase of the project requires *lazy consensus*
-of active committers other than the author of the patch. The code can
-be committed after the first +1.
-
-#### Product Release
-
-To make a release, the release manager creates a release candidate and
-a vote requiring a *lazy majority* of the active PMC members is
-required. Once the vote passes, the release candidate becomes an
-official release.
-
-#### Adoption of New Codebase
-
-When the codebase for an existing, released product is to be replaced
-with an alternative codebase, it requires a *lazy 2/3 majority* of PMC
-members. This also covers the creation of new sub-projects and
-submodules within the project.
-
-#### New Committer
-
-When a new committer is proposed for the project, *consensus approval*
-of the active PMC members is required.
-
-#### New PMC Member
-
-To promote a committer to a PMC member requires *consensus approval*
-of active PMC members.
-
-If the vote passes, the Apache Board must be notified to make the change
-official.
-
-#### Committer Removal
-
-Removal of commit privileges requires a *lazy 2/3 majority* of active
-PMC members.
-
-#### PMC Member Removal
-
-Removing a PMC member requires a *lazy 2/3 majority* of active PMC
-members, excluding the member in question.
-
-If the vote passes, the Apache Board must be notified to make the change
-official.
-
-#### Modifying Bylaws
-
-Modifying this document requires a *lazy majority* of active PMC members.
-
-### Voting Timeframes
-
-Votes are open for a minimum period of 72 hours to allow all active
-voters time to consider the vote. For holiday weekends or conferences,
-consider using a longer vote window. Votes relating to code changes are
-not subject to a strict timetable but should be made as timely as
-possible.
http://git-wip-us.apache.org/repos/asf/metron/blob/8bf3b6ec/site/develop/coding.md
----------------------------------------------------------------------
diff --git a/site/develop/coding.md b/site/develop/coding.md
deleted file mode 100644
index 78c7e27..0000000
--- a/site/develop/coding.md
+++ /dev/null
@@ -1,34 +0,0 @@
----
-layout: page
-title: Coding Guidelines
----
-
-## General rules
-
-* All files must have an Apache copyright header at the top of the file.
-* Code should be removed rather than commented out.
-* All public functions should have javadoc comments.
-* Always use braces to surround branches.
-* try-finally should be avoided.
-
-## Formatting
-
-* All files must have an 80 character maximum line length.
-* Indentation should be 2 spaces.
-* Files should use spaces instead of tabs.
-* Wrapping lines
- * Break after a comma.
- * Break before an operator.
- * Prefer higher-level breaks to lower-level breaks.
- * Align the new line with beginning of the expression at the same level
- on the previous line.
- * If the above rules lead to confusing code, just indent 8 spaces.
-* One variable declaration per a line.
-
-## Naming
-
-* Packages should be all lowercase.
- * Java code should be in `org.apache.,metron`, except for compatibility classes
-* Classes should be in mixed case.
-* Variables should be in camel case.
-* Constants should be in upper case.
http://git-wip-us.apache.org/repos/asf/metron/blob/8bf3b6ec/site/develop/index.md
----------------------------------------------------------------------
diff --git a/site/develop/index.md b/site/develop/index.md
deleted file mode 100644
index 7db04fa..0000000
--- a/site/develop/index.md
+++ /dev/null
@@ -1,61 +0,0 @@
----
-layout: page
-title: Developing
----
-
-Information about the Metron project that is most important for
-developers working on the project. The project has created
-[bylaws](bylaws.html) for itself.
-
-## Project Members
-
-Name | Apache Id | Role
-:---------------------- | :------------- | :---
-Owen O'Malley | omalley | PMC
-Jim Baker | jimbaker | PMC
-Mark Bittmann | mbittmann | PMC
-Sheetal Dolas | sheetal_dolas | PMC
-P. Taylor Goetz | ptgoetz | PMC
-Brad Kolarov | billie | PMC
-Dave Hirko | dbhirko | PMC
-Larry McCay | lmccay | PMC
-Charles Porter | cporter | PMC
-James Sirota | jsirota | PMC
-Casey Stella | cestella | PMC
-Vinod Kumar Vavilapalli | vinodkv | PMC
-George Vetticaden | gvetticaden | PMC
-
-## Mailing Lists
-
-There are several development mailing lists for Metron
-
-* [dev@metron.apache.org](mailto:dev@metron.apache.org) - Development discussions
- with archive [here](https://mail-archives.apache.org/mod_mbox/metron-dev/)
-* [issues@metron.apache.org](mailto:issues@metron.apache.org) - Bug tracking
- with archive [here](https://mail-archives.apache.org/mod_mbox/metron-issues/)
-* [commits@metron.apache.org](mailto:commits@metron.apache.org) - Git tracking
- with archive [here](https://mail-archives.apache.org/mod_mbox/metron-commits/)
-
-You can subscribe to the lists by sending email to
-*list*-subscribe@metron.apache.org and unsubscribe by sending email to
-*list*-unsubscribe@metron.apache.org.
-
-## Source code
-
-Metron uses git for version control. Get the source code:
-
-`% git clone https://git-wip-us.apache.org/repos/asf/metron.git`
-
-The important branches are:
-
-* [master](https://github.com/apache/metron/tree/master) -
- The trunk for all developement
-* [asf-site](https://github.com/apache/metron/tree/asf-site) -
- The pages that are deployed to https://metron.apache.org/
-
-Please check our [coding guidelines](/develop/coding.html).
-
-## Reviews
-
-All code must be +1'ed by a committer other than the author prior to its
-commit.