You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@metron.apache.org by ce...@apache.org on 2017/08/29 13:50:32 UTC
metron git commit: METRON-1128: MAAS_GET_ENDPOINT - Unable to resolve
function error seen for squid example this closes apache/incubator-metron#714
Repository: metron
Updated Branches:
refs/heads/master a9079f546 -> 7f9f86c00
METRON-1128: MAAS_GET_ENDPOINT - Unable to resolve function error seen for squid example this closes apache/incubator-metron#714
Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/7f9f86c0
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/7f9f86c0
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/7f9f86c0
Branch: refs/heads/master
Commit: 7f9f86c005eb5d9d966d33df95cd060e2f8784bb
Parents: a9079f5
Author: cstella <ce...@gmail.com>
Authored: Tue Aug 29 09:50:23 2017 -0400
Committer: cstella <ce...@gmail.com>
Committed: Tue Aug 29 09:50:23 2017 -0400
----------------------------------------------------------------------
.gitignore | 2 +-
metron-analytics/metron-maas-common/pom.xml | 11 +
.../metron/maas/functions/MaaSFunctions.java | 324 +++++++++++++++++++
metron-analytics/metron-maas-service/README.md | 21 +-
metron-analytics/metron-maas-service/pom.xml | 11 -
.../metron/maas/functions/MaaSFunctions.java | 324 -------------------
.../org/apache/metron/maas/service/Client.java | 15 +
.../apache/metron/maas/service/Constants.java | 5 +-
.../metron/maas/service/runner/Runner.java | 9 +-
.../metron/maas/submit/ModelSubmission.java | 5 +
10 files changed, 383 insertions(+), 344 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/metron/blob/7f9f86c0/.gitignore
----------------------------------------------------------------------
diff --git a/.gitignore b/.gitignore
index d505c9c..12fd7cd 100644
--- a/.gitignore
+++ b/.gitignore
@@ -21,5 +21,5 @@ tmp/**
tmp/**/*
temp/**
temp/**/*
-
+metron-interface/metron-alerts/node/
repodata/
http://git-wip-us.apache.org/repos/asf/metron/blob/7f9f86c0/metron-analytics/metron-maas-common/pom.xml
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-maas-common/pom.xml b/metron-analytics/metron-maas-common/pom.xml
index 13fb7b9..6921e51 100644
--- a/metron-analytics/metron-maas-common/pom.xml
+++ b/metron-analytics/metron-maas-common/pom.xml
@@ -34,6 +34,17 @@
</properties>
<dependencies>
<dependency>
+ <groupId>org.apache.metron</groupId>
+ <artifactId>stellar-common</artifactId>
+ <version>${project.parent.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-auth</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
<groupId>commons-cli</groupId>
<artifactId>commons-cli</artifactId>
<version>1.2</version>
http://git-wip-us.apache.org/repos/asf/metron/blob/7f9f86c0/metron-analytics/metron-maas-common/src/main/java/org/apache/metron/maas/functions/MaaSFunctions.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-maas-common/src/main/java/org/apache/metron/maas/functions/MaaSFunctions.java b/metron-analytics/metron-maas-common/src/main/java/org/apache/metron/maas/functions/MaaSFunctions.java
new file mode 100644
index 0000000..eacb64d
--- /dev/null
+++ b/metron-analytics/metron-maas-common/src/main/java/org/apache/metron/maas/functions/MaaSFunctions.java
@@ -0,0 +1,324 @@
+/**
+ * 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.maas.functions;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import java.lang.invoke.MethodHandles;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.metron.maas.config.Endpoint;
+import org.apache.metron.maas.config.MaaSConfig;
+import org.apache.metron.maas.config.ModelEndpoint;
+import org.apache.metron.maas.discovery.ServiceDiscoverer;
+import org.apache.metron.maas.util.ConfigUtil;
+import org.apache.metron.maas.util.RESTUtil;
+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;
+
+public class MaaSFunctions {
+ protected static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+ private static class ModelCacheKey {
+ String name;
+ String version;
+ String method;
+ Map<String, String> args;
+ public ModelCacheKey(String name, String version, String method, Map<String, String> args) {
+ this.name = name;
+ this.version = version;
+ this.method = method;
+ this.args = args;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ ModelCacheKey that = (ModelCacheKey) o;
+
+ if (name != null ? !name.equals(that.name) : that.name != null) return false;
+ if (version != null ? !version.equals(that.version) : that.version != null) return false;
+ if (method != null ? !method.equals(that.method) : that.method != null) return false;
+ return args != null ? args.equals(that.args) : that.args == null;
+
+ }
+
+ @Override
+ public int hashCode() {
+ int result = name != null ? name.hashCode() : 0;
+ result = 31 * result + (version != null ? version.hashCode() : 0);
+ result = 31 * result + (method != null ? method.hashCode() : 0);
+ result = 31 * result + (args != null ? args.hashCode() : 0);
+ return result;
+ }
+ }
+
+ @Stellar(name="MODEL_APPLY"
+ , namespace="MAAS"
+ , description = "Returns the output of a model deployed via Model as a Service. NOTE: Results are cached locally for 10 minutes."
+ , params = { "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."
+ )
+ public static class ModelApply implements StellarFunction {
+ private boolean isInitialized = false;
+ private ServiceDiscoverer discoverer;
+ private Cache<ModelCacheKey, Map<String, Object> > resultCache;
+ public ModelApply() {
+ resultCache = CacheBuilder.newBuilder()
+ .concurrencyLevel(4)
+ .weakKeys()
+ .maximumSize(100000)
+ .expireAfterWrite(10, TimeUnit.MINUTES)
+ .build();
+ }
+
+ @Override
+ public Object apply(List<Object> args, Context context) throws ParseException {
+ if(args.size() < 2) {
+ throw new ParseException("Unable to execute model_apply. " +
+ "Expected arguments: endpoint_map:map, " +
+ " [endpoint method:string], model_args:map"
+ );
+ }
+ if(!isInitialized) {
+ return null;
+ }
+ int i = 0;
+ if(args.size() == 0) {
+ return null;
+ }
+ Object endpointObj = args.get(i++);
+ Map endpoint = null;
+ String modelName;
+ String modelVersion;
+ String modelUrl;
+ if(endpointObj instanceof Map) {
+ endpoint = (Map)endpointObj;
+ modelName = endpoint.get("name") + "";
+ modelVersion = endpoint.get("version") + "";
+ modelUrl = endpoint.get("url") + "";
+ }
+ else {
+ return null;
+ }
+ String modelFunction = "apply";
+ Map<String, String> modelArgs = new HashMap<>();
+ if(args.get(i) instanceof String) {
+ String func = (String)args.get(i);
+ if(endpoint.containsKey("endpoint:" + func)) {
+ modelFunction = "" + endpoint.get("endpoint:" + func);
+ }
+ else {
+ modelFunction = func;
+ }
+ i++;
+ }
+
+ if(args.get(i) instanceof Map) {
+ if(endpoint.containsKey("endpoint:apply")) {
+ modelFunction = "" + endpoint.get("endpoint:apply");
+ }
+ modelArgs = (Map)args.get(i);
+ }
+ if( modelName == null
+ || modelVersion == null
+ || modelFunction == null
+ ) {
+ return null;
+ }
+ ModelCacheKey cacheKey = new ModelCacheKey(modelName, modelVersion, modelFunction, modelArgs);
+ Map<String, Object> ret = resultCache.getIfPresent(cacheKey);
+ if(ret != null) {
+ return ret;
+ }
+ else {
+ String url = modelUrl;
+ if (url.endsWith("/")) {
+ url = url.substring(0, url.length() - 1);
+ }
+ if (modelFunction.startsWith("/")) {
+ modelFunction = modelFunction.substring(1);
+ }
+ try {
+ URL u = new URL(url + "/" + modelFunction);
+
+ String results = RESTUtil.INSTANCE.getRESTJSONResults(u, modelArgs);
+ ret = JSONUtils.INSTANCE.load(results, new TypeReference<Map<String, Object>>() {
+ });
+ resultCache.put(cacheKey, ret);
+ return ret;
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ if (discoverer != null) {
+ try {
+ URL u = new URL(modelUrl);
+ discoverer.blacklist(u);
+ } catch (MalformedURLException e1) {
+ }
+ }
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public synchronized void initialize(Context context) {
+
+ try {
+ Optional<ServiceDiscoverer> discovererOpt = (Optional) (context.getCapability(Context.Capabilities.SERVICE_DISCOVERER));
+ if (discovererOpt.isPresent()) {
+ discoverer = discovererOpt.get();
+ }
+ else {
+ Optional<Object> clientOptional = context.getCapability(Context.Capabilities.ZOOKEEPER_CLIENT);
+ CuratorFramework client = null;
+ if (clientOptional.isPresent() && clientOptional.get() instanceof CuratorFramework) {
+ client = (CuratorFramework) clientOptional.get();
+ } else {
+ throw new IllegalStateException("Unable to initialize function: Cannot find zookeeper client.");
+ }
+ discoverer = createDiscoverer(client);
+ }
+ }
+ catch(Exception ex) {
+ LOG.error(ex.getMessage(), ex);
+ }
+ finally {
+ //We always want to set initialize to true because we don't want to keep trying to initialize over and over
+ isInitialized = true;
+ }
+ }
+
+ @Override
+ public boolean isInitialized() {
+ return isInitialized;
+ }
+ }
+
+ private static ServiceDiscoverer createDiscoverer(CuratorFramework client) throws Exception {
+ MaaSConfig config = ConfigUtil.INSTANCE.read(client, "/metron/maas/config", new MaaSConfig(), MaaSConfig.class);
+ ServiceDiscoverer discoverer = new ServiceDiscoverer(client, config.getServiceRoot());
+ discoverer.start();
+ return discoverer;
+ }
+
+ @Stellar(name="GET_ENDPOINT"
+ , namespace="MAAS"
+ , description="Inspects ZooKeeper and returns a map containing the name, version and url for the model referred to by the input parameters."
+ , params = {
+ "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."
+ )
+ public static class GetEndpoint implements StellarFunction {
+ ServiceDiscoverer discoverer;
+ private boolean isInitialized = false;
+ private boolean isValidState = false;
+
+ @Override
+ public Object apply(List<Object> args, Context context) throws ParseException {
+ if(!isValidState) {
+ LOG.error("Invalid state: Unable to find ServiceDiscoverer service.");
+ return null;
+ }
+ String modelName = null;
+ String modelVersion = null;
+ if(args.size() >= 1) {
+ modelName = args.get(0).toString();
+ }
+ if(args.size() >= 2)
+ {
+ modelVersion = args.get(1).toString();
+ }
+ if(modelName == null) {
+ return null;
+ }
+ try {
+ ModelEndpoint ep = null;
+ if (modelVersion == null) {
+ ep = discoverer.getEndpoint(modelName);
+ } else {
+ ep = discoverer.getEndpoint(modelName, modelVersion);
+ }
+ return ep == null ? null : endpointToMap(ep.getName(), ep.getVersion(), ep.getEndpoint());
+ }
+ catch(Exception ex) {
+ LOG.error("Unable to discover endpoint: {}", ex.getMessage(), ex);
+ return null;
+ }
+ }
+
+ public static Map<String, String> endpointToMap(String name, String version, Endpoint ep) {
+ Map<String, String> ret = new HashMap<>();
+ ret.put("url", ep.getUrl());
+ ret.put("name", name);
+ ret.put("version", version);
+ for(Map.Entry<String, String> kv : ep.getFunctions().entrySet()) {
+ ret.put("endpoint:" + kv.getKey(), kv.getValue());
+ }
+ return ret;
+ }
+
+ @Override
+ public synchronized void initialize(Context context) {
+ try {
+ Optional<Object> clientOptional = context.getCapability(Context.Capabilities.ZOOKEEPER_CLIENT);
+ CuratorFramework client = null;
+ if (clientOptional.isPresent() && clientOptional.get() instanceof CuratorFramework) {
+ client = (CuratorFramework) clientOptional.get();
+ } else {
+ throw new IllegalStateException("Unable to initialize function: Cannot find zookeeper client.");
+ }
+ try {
+ discoverer = createDiscoverer(client);
+ context.addCapability(Context.Capabilities.SERVICE_DISCOVERER, () -> discoverer);
+ isValidState = true;
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ throw new IllegalStateException("Unable to initialize MAAS_GET_ENDPOINT", e);
+ }
+ }
+ finally {
+ isInitialized = true;
+ }
+ }
+
+ @Override
+ public boolean isInitialized() {
+ return isInitialized;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/7f9f86c0/metron-analytics/metron-maas-service/README.md
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-maas-service/README.md b/metron-analytics/metron-maas-service/README.md
index 5ed108b..d234130 100644
--- a/metron-analytics/metron-maas-service/README.md
+++ b/metron-analytics/metron-maas-service/README.md
@@ -126,16 +126,29 @@ Now let's install some prerequisites:
Start Squid via `service squid start`
Now that we have flask and jinja, we can create a mock DGA service to deploy with MaaS:
-* Download the files in [this](https://gist.github.com/cestella/cba10aff0f970078a4c2c8cade3a4d1a) gist into the `/root/mock_dga` directory
-* Make `rest.sh` executable via `chmod +x /root/mock_dga/rest.sh`
+* Download the files in [this](https://gist.github.com/cestella/cba10aff0f970078a4c2c8cade3a4d1a) gist into the `$HOME/mock_dga` directory
+* Make `rest.sh` executable via `chmod +x $HOME/mock_dga/rest.sh`
This service will treat `yahoo.com` and `amazon.com` as legit and everything else as malicious. The contract is that the REST service exposes an endpoint `/apply` and returns back JSON maps with a single key `is_malicious` which can be `malicious` or `legit`.
## Deploy Mock DGA Service via MaaS
+The following presumes that you are a logged in as a user who has a
+home directory in HDFS under `/user/$USER`. If you do not, please create one
+and ensure the permissions are set appropriate:
+```
+su - hdfs -c "hadoop fs -mkdir /user/$USER"
+su - hdfs -c "hadoop fs -chown $USER:$USER /user/$USER"
+```
+Or, in the common case for the `metron` user:
+```
+su - hdfs -c "hadoop fs -mkdir /user/metron"
+su - hdfs -c "hadoop fs -chown metron:metron /user/metron"
+```
+
Now let's start MaaS and deploy the Mock DGA Service:
* Start MaaS via `$METRON_HOME/bin/maas_service.sh -zq node1:2181`
-* Start one instance of the mock DGA model with 512M of memory via `$METRON_HOME/bin/maas_deploy.sh -zq node1:2181 -lmp /root/mock_dga -hmp /user/root/models -mo ADD -m 512 -n dga -v 1.0 -ni 1`
+* Start one instance of the mock DGA model with 512M of memory via `$METRON_HOME/bin/maas_deploy.sh -zq node1:2181 -lmp $HOME/mock_dga -hmp /user/$USER/models -mo ADD -m 512 -n dga -v 1.0 -ni 1`
* As a sanity check:
* Ensure that the model is running via `$METRON_HOME/bin/maas_deploy.sh -zq node1:2181 -mo LIST`. You should see `Model dga @ 1.0` be displayed and under that a url such as (but not exactly) `http://node1:36161`
* Try to hit the model via curl: `curl 'http://localhost:36161/apply?host=caseystella.com'` and ensure that it returns a JSON map indicating the domain is malicious.
@@ -170,8 +183,6 @@ Now that we have a deployed model, let's adjust the configurations for the Squid
* Edit the squid enrichment configuration at `$METRON_HOME/config/zookeeper/enrichments/squid.json` (this file will not exist, so create a new one) to make the threat triage adjust the level of risk based on the model output:
```
{
- "index": "squid",
- "batchSize": 1,
"enrichment" : {
"fieldMap": {}
},
http://git-wip-us.apache.org/repos/asf/metron/blob/7f9f86c0/metron-analytics/metron-maas-service/pom.xml
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-maas-service/pom.xml b/metron-analytics/metron-maas-service/pom.xml
index 0ac9bac..4eeceae 100644
--- a/metron-analytics/metron-maas-service/pom.xml
+++ b/metron-analytics/metron-maas-service/pom.xml
@@ -49,17 +49,6 @@
<version>${global_kryo_serializers_version}</version>
</dependency>
<dependency>
- <groupId>org.apache.metron</groupId>
- <artifactId>stellar-common</artifactId>
- <version>${project.parent.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-auth</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-yarn-server-common</artifactId>
<version>${hadoop.version}</version>
http://git-wip-us.apache.org/repos/asf/metron/blob/7f9f86c0/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/functions/MaaSFunctions.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/functions/MaaSFunctions.java b/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/functions/MaaSFunctions.java
deleted file mode 100644
index eacb64d..0000000
--- a/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/functions/MaaSFunctions.java
+++ /dev/null
@@ -1,324 +0,0 @@
-/**
- * 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.maas.functions;
-
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.google.common.cache.Cache;
-import com.google.common.cache.CacheBuilder;
-import java.lang.invoke.MethodHandles;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.concurrent.TimeUnit;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.metron.maas.config.Endpoint;
-import org.apache.metron.maas.config.MaaSConfig;
-import org.apache.metron.maas.config.ModelEndpoint;
-import org.apache.metron.maas.discovery.ServiceDiscoverer;
-import org.apache.metron.maas.util.ConfigUtil;
-import org.apache.metron.maas.util.RESTUtil;
-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;
-
-public class MaaSFunctions {
- protected static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private static class ModelCacheKey {
- String name;
- String version;
- String method;
- Map<String, String> args;
- public ModelCacheKey(String name, String version, String method, Map<String, String> args) {
- this.name = name;
- this.version = version;
- this.method = method;
- this.args = args;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
-
- ModelCacheKey that = (ModelCacheKey) o;
-
- if (name != null ? !name.equals(that.name) : that.name != null) return false;
- if (version != null ? !version.equals(that.version) : that.version != null) return false;
- if (method != null ? !method.equals(that.method) : that.method != null) return false;
- return args != null ? args.equals(that.args) : that.args == null;
-
- }
-
- @Override
- public int hashCode() {
- int result = name != null ? name.hashCode() : 0;
- result = 31 * result + (version != null ? version.hashCode() : 0);
- result = 31 * result + (method != null ? method.hashCode() : 0);
- result = 31 * result + (args != null ? args.hashCode() : 0);
- return result;
- }
- }
-
- @Stellar(name="MODEL_APPLY"
- , namespace="MAAS"
- , description = "Returns the output of a model deployed via Model as a Service. NOTE: Results are cached locally for 10 minutes."
- , params = { "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."
- )
- public static class ModelApply implements StellarFunction {
- private boolean isInitialized = false;
- private ServiceDiscoverer discoverer;
- private Cache<ModelCacheKey, Map<String, Object> > resultCache;
- public ModelApply() {
- resultCache = CacheBuilder.newBuilder()
- .concurrencyLevel(4)
- .weakKeys()
- .maximumSize(100000)
- .expireAfterWrite(10, TimeUnit.MINUTES)
- .build();
- }
-
- @Override
- public Object apply(List<Object> args, Context context) throws ParseException {
- if(args.size() < 2) {
- throw new ParseException("Unable to execute model_apply. " +
- "Expected arguments: endpoint_map:map, " +
- " [endpoint method:string], model_args:map"
- );
- }
- if(!isInitialized) {
- return null;
- }
- int i = 0;
- if(args.size() == 0) {
- return null;
- }
- Object endpointObj = args.get(i++);
- Map endpoint = null;
- String modelName;
- String modelVersion;
- String modelUrl;
- if(endpointObj instanceof Map) {
- endpoint = (Map)endpointObj;
- modelName = endpoint.get("name") + "";
- modelVersion = endpoint.get("version") + "";
- modelUrl = endpoint.get("url") + "";
- }
- else {
- return null;
- }
- String modelFunction = "apply";
- Map<String, String> modelArgs = new HashMap<>();
- if(args.get(i) instanceof String) {
- String func = (String)args.get(i);
- if(endpoint.containsKey("endpoint:" + func)) {
- modelFunction = "" + endpoint.get("endpoint:" + func);
- }
- else {
- modelFunction = func;
- }
- i++;
- }
-
- if(args.get(i) instanceof Map) {
- if(endpoint.containsKey("endpoint:apply")) {
- modelFunction = "" + endpoint.get("endpoint:apply");
- }
- modelArgs = (Map)args.get(i);
- }
- if( modelName == null
- || modelVersion == null
- || modelFunction == null
- ) {
- return null;
- }
- ModelCacheKey cacheKey = new ModelCacheKey(modelName, modelVersion, modelFunction, modelArgs);
- Map<String, Object> ret = resultCache.getIfPresent(cacheKey);
- if(ret != null) {
- return ret;
- }
- else {
- String url = modelUrl;
- if (url.endsWith("/")) {
- url = url.substring(0, url.length() - 1);
- }
- if (modelFunction.startsWith("/")) {
- modelFunction = modelFunction.substring(1);
- }
- try {
- URL u = new URL(url + "/" + modelFunction);
-
- String results = RESTUtil.INSTANCE.getRESTJSONResults(u, modelArgs);
- ret = JSONUtils.INSTANCE.load(results, new TypeReference<Map<String, Object>>() {
- });
- resultCache.put(cacheKey, ret);
- return ret;
- } catch (Exception e) {
- LOG.error(e.getMessage(), e);
- if (discoverer != null) {
- try {
- URL u = new URL(modelUrl);
- discoverer.blacklist(u);
- } catch (MalformedURLException e1) {
- }
- }
- }
- }
- return null;
- }
-
- @Override
- public synchronized void initialize(Context context) {
-
- try {
- Optional<ServiceDiscoverer> discovererOpt = (Optional) (context.getCapability(Context.Capabilities.SERVICE_DISCOVERER));
- if (discovererOpt.isPresent()) {
- discoverer = discovererOpt.get();
- }
- else {
- Optional<Object> clientOptional = context.getCapability(Context.Capabilities.ZOOKEEPER_CLIENT);
- CuratorFramework client = null;
- if (clientOptional.isPresent() && clientOptional.get() instanceof CuratorFramework) {
- client = (CuratorFramework) clientOptional.get();
- } else {
- throw new IllegalStateException("Unable to initialize function: Cannot find zookeeper client.");
- }
- discoverer = createDiscoverer(client);
- }
- }
- catch(Exception ex) {
- LOG.error(ex.getMessage(), ex);
- }
- finally {
- //We always want to set initialize to true because we don't want to keep trying to initialize over and over
- isInitialized = true;
- }
- }
-
- @Override
- public boolean isInitialized() {
- return isInitialized;
- }
- }
-
- private static ServiceDiscoverer createDiscoverer(CuratorFramework client) throws Exception {
- MaaSConfig config = ConfigUtil.INSTANCE.read(client, "/metron/maas/config", new MaaSConfig(), MaaSConfig.class);
- ServiceDiscoverer discoverer = new ServiceDiscoverer(client, config.getServiceRoot());
- discoverer.start();
- return discoverer;
- }
-
- @Stellar(name="GET_ENDPOINT"
- , namespace="MAAS"
- , description="Inspects ZooKeeper and returns a map containing the name, version and url for the model referred to by the input parameters."
- , params = {
- "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."
- )
- public static class GetEndpoint implements StellarFunction {
- ServiceDiscoverer discoverer;
- private boolean isInitialized = false;
- private boolean isValidState = false;
-
- @Override
- public Object apply(List<Object> args, Context context) throws ParseException {
- if(!isValidState) {
- LOG.error("Invalid state: Unable to find ServiceDiscoverer service.");
- return null;
- }
- String modelName = null;
- String modelVersion = null;
- if(args.size() >= 1) {
- modelName = args.get(0).toString();
- }
- if(args.size() >= 2)
- {
- modelVersion = args.get(1).toString();
- }
- if(modelName == null) {
- return null;
- }
- try {
- ModelEndpoint ep = null;
- if (modelVersion == null) {
- ep = discoverer.getEndpoint(modelName);
- } else {
- ep = discoverer.getEndpoint(modelName, modelVersion);
- }
- return ep == null ? null : endpointToMap(ep.getName(), ep.getVersion(), ep.getEndpoint());
- }
- catch(Exception ex) {
- LOG.error("Unable to discover endpoint: {}", ex.getMessage(), ex);
- return null;
- }
- }
-
- public static Map<String, String> endpointToMap(String name, String version, Endpoint ep) {
- Map<String, String> ret = new HashMap<>();
- ret.put("url", ep.getUrl());
- ret.put("name", name);
- ret.put("version", version);
- for(Map.Entry<String, String> kv : ep.getFunctions().entrySet()) {
- ret.put("endpoint:" + kv.getKey(), kv.getValue());
- }
- return ret;
- }
-
- @Override
- public synchronized void initialize(Context context) {
- try {
- Optional<Object> clientOptional = context.getCapability(Context.Capabilities.ZOOKEEPER_CLIENT);
- CuratorFramework client = null;
- if (clientOptional.isPresent() && clientOptional.get() instanceof CuratorFramework) {
- client = (CuratorFramework) clientOptional.get();
- } else {
- throw new IllegalStateException("Unable to initialize function: Cannot find zookeeper client.");
- }
- try {
- discoverer = createDiscoverer(client);
- context.addCapability(Context.Capabilities.SERVICE_DISCOVERER, () -> discoverer);
- isValidState = true;
- } catch (Exception e) {
- LOG.error(e.getMessage(), e);
- throw new IllegalStateException("Unable to initialize MAAS_GET_ENDPOINT", e);
- }
- }
- finally {
- isInitialized = true;
- }
- }
-
- @Override
- public boolean isInitialized() {
- return isInitialized;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/metron/blob/7f9f86c0/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/service/Client.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/service/Client.java b/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/service/Client.java
index c2d8906..9cabf21 100644
--- a/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/service/Client.java
+++ b/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/service/Client.java
@@ -24,6 +24,7 @@ import java.util.*;
import java.util.function.Function;
import com.google.common.base.Joiner;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
import org.apache.commons.cli.*;
import org.apache.commons.cli.CommandLine;
@@ -558,6 +559,7 @@ public class Client {
// Copy the application master jar to the filesystem
// Create a local resource to point to the destination jar path
FileSystem fs = FileSystem.get(conf);
+ createMaaSDirectory(fs, appId.toString());
Path ajPath = addToLocalResources(fs, appMasterJar, appMasterJarPath, appId.toString(), localResources, null);
// Set the log4j properties if needed
@@ -789,6 +791,18 @@ public class Client {
yarnClient.killApplication(appId);
}
+ private void createMaaSDirectory(FileSystem fs, String appId) throws IOException {
+ for(Path p : ImmutableList.of(new Path(fs.getHomeDirectory(), appName)
+ , new Path(fs.getHomeDirectory(), appName + "/" + appId)
+ )
+ ) {
+ if(!fs.exists(p)) {
+ fs.mkdirs(p);
+ fs.setPermission(p, new FsPermission((short)0755));
+ }
+ }
+ }
+
private Path addToLocalResources(FileSystem fs, String fileSrcPath,
String fileDstPath, String appId, Map<String, LocalResource> localResources,
String resources) throws IOException {
@@ -808,6 +822,7 @@ public class Client {
} else {
fs.copyFromLocalFile(new Path(fileSrcPath), dst);
}
+ fs.setPermission(dst, new FsPermission((short)0755));
FileStatus scFileStatus = fs.getFileStatus(dst);
LocalResource scRsrc =
LocalResource.newInstance(
http://git-wip-us.apache.org/repos/asf/metron/blob/7f9f86c0/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/service/Constants.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/service/Constants.java b/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/service/Constants.java
index ac2c950..d032511 100644
--- a/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/service/Constants.java
+++ b/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/service/Constants.java
@@ -31,5 +31,8 @@ public class Constants {
* Environment key name denoting the timeline domain ID.
*/
public static final String TIMELINEDOMAIN = "TIMELINEDOMAIN";
-
+ /*
+ The filename which communicates the endpoint information for a deployed model
+ */
+ public static final String ENDPOINT_DAT = "endpoint.dat";
}
http://git-wip-us.apache.org/repos/asf/metron/blob/7f9f86c0/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/service/runner/Runner.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/service/runner/Runner.java b/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/service/runner/Runner.java
index cc297d2..8f0b9e5 100644
--- a/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/service/runner/Runner.java
+++ b/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/service/runner/Runner.java
@@ -33,6 +33,7 @@ import org.apache.curator.utils.CloseableUtils;
import org.apache.curator.x.discovery.*;
import org.apache.curator.x.discovery.details.JsonInstanceSerializer;
import org.apache.metron.maas.config.Endpoint;
+import org.apache.metron.maas.service.Constants;
import org.apache.metron.maas.util.ConfigUtil;
import org.apache.metron.maas.config.MaaSConfig;
import org.apache.metron.maas.config.ModelEndpoint;
@@ -202,7 +203,11 @@ public class Runner {
serviceDiscovery.start();
File cwd = new File(script).getParentFile();
- final String cmd = new File(cwd, script).getAbsolutePath();
+ File scriptFile = new File(cwd, script);
+ if(scriptFile.exists() && !scriptFile.canExecute()) {
+ scriptFile.setExecutable(true);
+ }
+ final String cmd = scriptFile.getAbsolutePath();
try {
p = new ProcessBuilder(cmd).directory(cwd).start();
@@ -299,7 +304,7 @@ public class Runner {
private static Endpoint readEndpoint(File cwd) throws Exception {
String content = "";
- File f = new File(cwd, "endpoint.dat");
+ File f = new File(cwd, Constants.ENDPOINT_DAT);
for(int i = 0;i < NUM_ATTEMPTS;i++) {
if(f.exists()) {
try {
http://git-wip-us.apache.org/repos/asf/metron/blob/7f9f86c0/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/submit/ModelSubmission.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/submit/ModelSubmission.java b/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/submit/ModelSubmission.java
index ebfa904..fcae40a 100644
--- a/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/submit/ModelSubmission.java
+++ b/metron-analytics/metron-maas-service/src/main/java/org/apache/metron/maas/submit/ModelSubmission.java
@@ -33,6 +33,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.log4j.PropertyConfigurator;
import org.apache.metron.maas.config.*;
import org.apache.metron.maas.discovery.ServiceDiscoverer;
+import org.apache.metron.maas.service.Constants;
import org.apache.metron.maas.service.Log4jPropertyHelper;
import org.apache.metron.maas.util.ConfigUtil;
import org.apache.metron.maas.queue.Queue;
@@ -247,6 +248,10 @@ public class ModelSubmission {
fs.mkdirs(hdfsPath);
}
for(File f : localDir.listFiles()) {
+ if(f.getName().equals(Constants.ENDPOINT_DAT)) {
+ //skip the endpoint if it exists accidentally, we don't want to localize that.
+ continue;
+ }
Path p = new Path(hdfsPath, f.getName());
FSDataOutputStream out = fs.create(p);
BufferedInputStream in = new BufferedInputStream(new FileInputStream(f));