metron-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ceste...@apache.org
Subject metron git commit: METRON-1128: MAAS_GET_ENDPOINT - Unable to resolve function error seen for squid example this closes apache/incubator-metron#714
Date Tue, 29 Aug 2017 13:50:32 GMT
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 <cestella@gmail.com>
Authored: Tue Aug 29 09:50:23 2017 -0400
Committer: cstella <cestella@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));


Mime
View raw message