SDAP-161 MUDROD embedded unit test

core/src/main/java/org/apache/sdap/mudrod/discoveryengine/WeblogDiscoveryEngine.java

@@ -54,7 +54,8 @@ public WeblogDiscoveryEngine(Properties props, ESDriver es, SparkDriver spark) {
   /**
    * Get log file list from a directory
    *
-   * @param logDir path to directory containing logs either local or in HDFS.
+   * @param logDir
+   *          path to directory containing logs either local or in HDFS.
    * @return a list of log files
    */
   public List<String> getFileList(String logDir) {

core/src/main/java/org/apache/sdap/mudrod/driver/ESDriver.java

@@ -490,13 +490,11 @@ protected Client makeClient(Properties props) throws IOException {
 
     Client client = null;
 
-    if (hosts != null && port > 1) {
-      try (TransportClient transportClient = new ESTransportClient(settings)) {
-        for (String host : hosts) {
-          transportClient.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), port));
-        }
-        return transportClient;
-      }
+    if (hosts != null && port > 1) {  
+      TransportClient transportClient = new ESTransportClient(settings);
+      for (String host : hosts)
+        transportClient.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), port));
+      return transportClient;
     } else if (clusterName != null) {
       node = new Node(settings);
       client = node.client();

core/src/main/java/org/apache/sdap/mudrod/main/MudrodEngine.java

@@ -396,7 +396,7 @@ public static void main(String[] args) {
     }
   }
 
-  private static void loadPathConfig(MudrodEngine me, String dataDir) {
+  public static void loadPathConfig(MudrodEngine me, String dataDir) {
     me.props.put(MudrodConstants.ONTOLOGY_INPUT_PATH, dataDir + "SWEET_ocean/");
     me.props.put(MudrodConstants.ONTOLOGY_PATH, dataDir + "ocean_triples.csv");
     me.props.put(MudrodConstants.USER_HISTORY_PATH, dataDir + "userhistorymatrix.csv");

core/src/main/java/org/apache/sdap/mudrod/weblog/pre/ClickStreamGenerator.java

@@ -71,5 +71,4 @@ public Object execute() {
   public Object execute(Object o) {
     return null;
   }
-
 }

core/src/main/java/org/apache/sdap/mudrod/weblog/pre/SessionStatistic.java

@@ -229,7 +229,7 @@ public int processSession(ESDriver es, String sessionId) throws IOException, Int
               String[] keywordList = keywords.split(",");
               for (String item : items) {
                 if (!Arrays.asList(keywordList).contains(item)) {
-                  keywords = keywords + item + ",";
+                  keywords = keywords + "," + item + ",";
                 }
               }
             }

core/src/main/java/org/apache/sdap/mudrod/weblog/structure/log/ApacheAccessLog.java

@@ -67,6 +67,7 @@ public static String parseFromLogLine(String log, Properties props) throws Parse
     String lineJson = "{}";
     matcher = p.matcher(log);
     if (!matcher.matches() || numFields != matcher.groupCount()) {
+      System.out.println("Number of fields does not match.");
       return lineJson;
     }
 

core/src/main/java/org/apache/sdap/mudrod/weblog/structure/session/SessionTree.java

@@ -26,10 +26,13 @@
 
 import java.io.UnsupportedEncodingException;
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.Set;
 import java.util.concurrent.ExecutionException;
 
 /**
@@ -205,7 +208,7 @@ public List<ClickStream> getClickStreamList(Properties props) {
       RequestUrl requestURL = new RequestUrl();
       String viewquery = "";
       try {
-        String infoStr = requestURL.getSearchInfo(viewnode.getRequest());
+        String infoStr = requestURL.getSearchInfo(viewnode.getReferer());
         viewquery = es.customAnalyzing(props.getProperty(MudrodConstants.ES_INDEX_NAME), infoStr);
       } catch (UnsupportedEncodingException | InterruptedException | ExecutionException e) {
         LOG.warn("Exception getting search info. Ignoring...", e);
@@ -222,6 +225,8 @@ public List<ClickStream> getClickStreamList(Properties props) {
 
       if (viewquery != null && !"".equals(viewquery)) {
         String[] queries = viewquery.trim().split(",");
+        Set<String> queryset = new HashSet<>();
+        queryset.addAll(Arrays.asList(queries));
         if (queries.length > 0) {
           for (String query : queries) {
             ClickStream data = new ClickStream(query, dataset, download);

core/src/test/java/org/apache/sdap/mudrod/discoveryengine/WeblogDiscoveryEngineTest.java

@@ -0,0 +1,129 @@
+/*
+ * Licensed 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.sdap.mudrod.discoveryengine;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileReader;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.sdap.mudrod.driver.ESDriver;
+import org.apache.sdap.mudrod.driver.SparkDriver;
+import org.apache.sdap.mudrod.main.AbstractElasticsearchIntegrationTest;
+import org.apache.sdap.mudrod.main.MudrodConstants;
+import org.apache.sdap.mudrod.main.MudrodEngine;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class WeblogDiscoveryEngineTest extends AbstractElasticsearchIntegrationTest {
+
+  private static WeblogDiscoveryEngine weblogEngine = null;
+
+  static String DIR_TESTDATA_ONE = "Testing_Data_1_3dayLog+Meta+Onto";
+
+  @BeforeClass
+  public static void setUp() {
+    MudrodEngine mudrodEngine = new MudrodEngine();
+    Properties props = mudrodEngine.loadConfig();
+    ESDriver es = new ESDriver(props);
+    SparkDriver spark = new SparkDriver(props);
+    String dataDir = getTestDataPath(DIR_TESTDATA_ONE);
+    System.out.println(dataDir);
+    props.setProperty(MudrodConstants.DATA_DIR, dataDir);
+    MudrodEngine.loadPathConfig(mudrodEngine, dataDir);
+    weblogEngine = new WeblogDiscoveryEngine(props, es, spark);
+  }
+
+  @AfterClass
+  public static void tearDown() {
+  }
+
+  private static String getTestDataPath(String testDataDir) {
+    String path = WeblogDiscoveryEngineTest.class.getClassLoader().getResource(testDataDir).toString();
+    if(path.startsWith("file:/")){
+      path = path.replaceAll("file:/", "");
+    }
+    return path;
+  }
+
+  @Test
+  public void testPreprocess() throws IOException {
+    weblogEngine.preprocess();
+    testPreprocessUserHistory();
+    testPreprocessClickStream();
+  }
+
+  private void testPreprocessUserHistory() throws IOException {
+    // compare user history data
+    String userHistorycsvFile = getTestDataPath(DIR_TESTDATA_ONE) + "/userHistoryMatrix.csv";
+    System.out.println(userHistorycsvFile);
+    HashMap<String, List<String>> map = extractPairFromCSV(userHistorycsvFile);
+    Assert.assertEquals("failed in history data result!", "195.219.98.7", String.join(",", map.get("sea surface topography")));
+  }
+
+  private void testPreprocessClickStream() throws IOException {
+    String clickStreamcsvFile = getTestDataPath(DIR_TESTDATA_ONE) + "/clickStreamMatrix.csv";
+    System.out.println(clickStreamcsvFile);
+    HashMap<String, List<String>> map = extractPairFromCSV(clickStreamcsvFile);
+    System.out.println(map);
+    Assert.assertEquals("failed in click stream result!", "\"ostm_l2_ost_ogdr_gps\"", String.join(",", map.get("sea surface topography")));
+  }
+
+  private HashMap extractPairFromCSV(String csvfile){
+    BufferedReader br = null;
+    try {
+      br = new BufferedReader(new FileReader(csvfile));
+    } catch (FileNotFoundException e) {
+      e.printStackTrace();
+    }
+    String line = null;
+    HashMap<String, List<String>> map = new HashMap<>();
+    int i = 0;
+    List<String> header = new LinkedList<>();
+    try {
+      while ((line = br.readLine()) != null) {
+        if (i == 0) {
+          String str[] = line.split(",");
+          for (String s : str) {
+            header.add(s);
+          }
+        } else {
+          String str[] = line.split(",");
+          for (int j = 1; j < str.length; j++) {
+            double value = Double.parseDouble(str[j]);
+            if (value > 0.0) {
+              if (!map.containsKey(str[0])) {
+                map.put(str[0], new ArrayList<>());
+              }
+              map.get(str[0]).add(header.get(j));
+            }
+          }
+        }
+        i += 1;
+      }
+    } catch (IOException e) {
+      e.printStackTrace();
+    }
+
+    return map;
+  }
+}

core/src/test/java/org/apache/sdap/mudrod/driver/EmbeddedElasticsearchServer.java

@@ -0,0 +1,113 @@
+/*
+ * Licensed 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.sdap.mudrod.driver;
+
+import org.apache.commons.io.FileUtils;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.node.Node;
+import org.elasticsearch.node.NodeValidationException;
+import org.elasticsearch.transport.Netty3Plugin;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+
+/**
+ * embedded elasticsearch server.
+ */
+public class EmbeddedElasticsearchServer {
+
+  private static final String DEFAULT_DATA_DIRECTORY = "target/elasticsearch-data";
+
+  private Node node;
+  private final String dataDirectory;
+
+  public EmbeddedElasticsearchServer() {
+    this(DEFAULT_DATA_DIRECTORY);
+  }
+
+  public EmbeddedElasticsearchServer(String dataDirectory) {
+
+    this.dataDirectory = dataDirectory;
+
+    Settings.Builder settingsBuilder = Settings.builder();
+    settingsBuilder.put("http.type", "netty3");
+    settingsBuilder.put("transport.type", "netty3");
+    settingsBuilder.put("cluster.name", "MudrodES").put("http.enabled", "true").put("path.data", dataDirectory).put("path.home", "/");
+
+    Settings settings = settingsBuilder.build();
+    Collection plugins = Arrays.asList(Netty3Plugin.class);
+    node = null;
+    try {
+      node = new PluginConfigurableNode(settings, plugins).start();
+      System.out.println(node.toString());
+    } catch (NodeValidationException e) {
+      e.printStackTrace();
+    }
+
+    System.out.println(node.getNodeEnvironment().nodeId());
+
+
+   /* System.out.println("======= INTEGRATION TEST: START Embedded Elasticsearch Server ========");
+    Settings settings = Settings.builder().put("path.home", this.dataDirectory)
+            .put("transport.type", "local")
+            .put("network.host", "127.0.0.1")
+            .put("cluster.name", "MudrodES")
+            //.put("http.port", "9200-9300")
+            .put("http.type", "netty3")
+            //.put("transport.tcp.port", "9300-9400")
+            .put("http.enabled", false)
+           // .put("client.transport.sniff", true)
+            .build();
+    node = new Node(settings);
+
+
+    Settings set = node.settings();
+    try {
+      node.start();
+    } catch (NodeValidationException e) {
+      // TODO Auto-generated catch block
+      e.printStackTrace();
+    }
+
+    final String clusterName = node.settings().get("cluster.name");
+    final String clusterport = node.settings().get("transport.tcp.port");
+    System.out.println("starting server with cluster-name: "+ clusterport);
+
+    System.out.println(node.client());*/
+
+  }
+
+  public Client getClient() {
+    return node.client();
+  }
+
+  public void shutdown() {
+    try {
+      node.close();
+    } catch (IOException e) {
+      e.printStackTrace();
+    }
+  }
+
+  private void deleteDataDirectory() {
+    try {
+      FileUtils.deleteDirectory(new File(dataDirectory));
+    } catch (IOException e) {
+      throw new RuntimeException("Could not delete data directory of embedded elasticsearch server", e);
+    }
+  }
+}

core/src/test/java/org/apache/sdap/mudrod/driver/PluginConfigurableNode.java

@@ -0,0 +1,27 @@
+/*
+ * Licensed 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.sdap.mudrod.driver;
+
+import java.util.Collection;
+
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.node.Node;
+import org.elasticsearch.node.internal.InternalSettingsPreparer;
+import org.elasticsearch.plugins.Plugin;
+
+public class PluginConfigurableNode extends Node {
+  public PluginConfigurableNode(Settings settings, Collection<Class<? extends Plugin>> classpathPlugins) {
+    super(InternalSettingsPreparer.prepareEnvironment(settings, null), classpathPlugins);
+  }
+}