You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@metron.apache.org by ly...@apache.org on 2016/04/11 16:59:18 UTC
incubator-metron git commit: METRON-95 Create data purging script for
HDFS files closes apache/incubator-metron#70
Repository: incubator-metron
Updated Branches:
refs/heads/master a7ea74918 -> 82a586ac8
METRON-95 Create data purging script for HDFS files closes apache/incubator-metron#70
Project: http://git-wip-us.apache.org/repos/asf/incubator-metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-metron/commit/82a586ac
Tree: http://git-wip-us.apache.org/repos/asf/incubator-metron/tree/82a586ac
Diff: http://git-wip-us.apache.org/repos/asf/incubator-metron/diff/82a586ac
Branch: refs/heads/master
Commit: 82a586ac8424affcd081796dc7155afb5ca91b21
Parents: a7ea749
Author: David Lyle <dl...@gmail.com>
Authored: Mon Apr 11 09:58:08 2016 -0500
Committer: David Lyle <dl...@gmail.com>
Committed: Mon Apr 11 09:58:08 2016 -0500
----------------------------------------------------------------------
.../roles/metron_streaming/defaults/main.yml | 27 +-
.../roles/metron_streaming/tasks/hdfs_purge.yml | 50 ++++
.../roles/metron_streaming/tasks/main.yml | 2 +
deployment/roles/metron_streaming/vars/main.yml | 37 ---
.../metron/enrichment/EnrichmentConfigTest.java | 17 ++
.../enrichment/EnrichmentConverterTest.java | 17 ++
metron-streaming/Metron-DataLoads/pom.xml | 7 +-
.../src/main/assembly/assembly.xml | 1 +
.../src/main/bash/prune_hdfs_files.sh | 21 ++
.../metron/dataloads/bulk/HDFSDataPruner.java | 256 +++++++++++++++++++
.../dataloads/bulk/HDFSDataPrunerTest.java | 164 ++++++++++++
metron-streaming/pom.xml | 2 +-
12 files changed, 559 insertions(+), 42 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/82a586ac/deployment/roles/metron_streaming/defaults/main.yml
----------------------------------------------------------------------
diff --git a/deployment/roles/metron_streaming/defaults/main.yml b/deployment/roles/metron_streaming/defaults/main.yml
index a65680e..0eff59f 100644
--- a/deployment/roles/metron_streaming/defaults/main.yml
+++ b/deployment/roles/metron_streaming/defaults/main.yml
@@ -15,12 +15,30 @@
# limitations under the License.
#
---
+metron_jar_name: Metron-Topologies-{{ metron_version }}.jar
+metron_directory: /usr/metron/{{ metron_version }}
+metron_solr_jar_name: Metron-Solr-{{ metron_version }}.jar
+metron_elasticsearch_jar_name: Metron-Elasticsearch-{{ metron_version }}.jar
+
+metron_dataloads_name: Metron-DataLoads-{{ metron_version }}-archive.tar.gz
+metron_dataloads_path: "{{ playbook_dir }}/../../metron-streaming/Metron-DataLoads/target/{{ metron_dataloads_name }}"
+metron_topologies_bundle_name: Metron-Topologies-{{ metron_version }}-archive.tar.gz
+metron_solr_bundle_name: Metron-Solr-{{ metron_version }}-archive.tar.gz
+metron_elasticsearch_bundle_name: Metron-Elasticsearch-{{ metron_version }}-archive.tar.gz
+metron_solr_bundle_path: "{{ playbook_dir }}/../../metron-streaming/Metron-Solr/target/{{ metron_solr_bundle_name }}"
+metron_elasticsearch_bundle_path: "{{ playbook_dir }}/../../metron-streaming/Metron-Elasticsearch/target/{{ metron_elasticsearch_bundle_name }}"
+metron_topologies_bundle_path: "{{ playbook_dir }}/../../metron-streaming/Metron-Topologies/target/{{ metron_topologies_bundle_name }}"
+
config_path: "{{ metron_directory }}/config"
zookeeper_config_path: "{{ config_path }}/zookeeper"
zookeeper_global_config_path: "{{ zookeeper_config_path }}/global.json"
+metron_solr_properties_config_path: "{{ metron_directory }}/config/etc/env/solr.properties"
+metron_elasticsearch_properties_config_path: "{{ metron_directory }}/config/etc/env/elasticsearch.properties"
+hbase_config_path: "/etc/hbase/conf"
+hdfs_config_path: "/etc/hadoop/conf"
+
threat_intel_bulk_load: True
threat_intel_bin: "{{ metron_directory }}/bin/threatintel_bulk_load.sh"
-threat_intel_host: "{{ groups.ambari_master[0] }}"
threat_intel_work_dir: /tmp/ti_bulk
threat_intel_csv_filename: "threat_ip.csv"
threat_intel_csv_filepath: "../roles/metron_streaming/templates/{{ threat_intel_csv_filename }}"
@@ -37,11 +55,14 @@ storm_topologies:
- "{{ metron_directory }}/config/topologies/pcap/remote.yaml"
- "{{ metron_directory }}/config/topologies/enrichment/remote.yaml"
+hdfs_retention_days: 30
+hdfs_bro_purge_cronjob: "{{ metron_directory }}/bin/prune_hdfs_files.sh -f {{ hdfs_url }} -g '/apps/metron/enrichment/indexed/bro_doc/*enrichment-*' -s $(date -d '{{ hdfs_retention_days }} days ago' +%m/%d/%Y) -n 1 >> /var/log/bro-purge/cron-bro-purge.log 2>&1"
+hdfs_yaf_purge_cronjob: "{{ metron_directory }}/bin/prune_hdfs_files.sh -f {{ hdfs_url }} -g '/apps/metron/enrichment/indexed/yaf_doc/*enrichment-*' -s $(date -d '{{ hdfs_retention_days }} days ago' +%m/%d/%Y) -n 1 >> /var/log/yaf-purge/cron-yaf-purge.log 2>&1"
+
elasticsearch_config_path: /etc/elasticsearch
elasticsearch_cluster_name: metron
elasticsearch_transport_port: 9300
-hbase_config_path: "/etc/hbase/conf"
-hdfs_config_path: "/etc/hadoop/conf"
+
metron_hdfs_output_dir: "/apps/metron"
metron_hdfs_rotation_policy: org.apache.storm.hdfs.bolt.rotation.TimedRotationPolicy
metron_hdfs_rotation_policy_count: 1
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/82a586ac/deployment/roles/metron_streaming/tasks/hdfs_purge.yml
----------------------------------------------------------------------
diff --git a/deployment/roles/metron_streaming/tasks/hdfs_purge.yml b/deployment/roles/metron_streaming/tasks/hdfs_purge.yml
new file mode 100644
index 0000000..1f63451
--- /dev/null
+++ b/deployment/roles/metron_streaming/tasks/hdfs_purge.yml
@@ -0,0 +1,50 @@
+#
+# 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.
+#
+---
+- name: Create Log Directories for HDFS Purge
+ file:
+ path: "{{ item }}"
+ state: directory
+ mode: 0755
+ owner: hdfs
+ group: hdfs
+ with_items:
+ - /var/log/bro-purge
+ - /var/log/yaf-purge
+
+- name: Create Empty Log Files for HDFS Purge
+ file:
+ path: "{{ item }}"
+ state: touch
+ owner: hdfs
+ group: hdfs
+ mode: 0644
+ with_items:
+ - /var/log/bro-purge/cron-bro-purge.log
+ - /var/log/yaf-purge/cron-yaf-purge.log
+
+
+- name: Purge HDFS Sensor Data every 30 days.
+ cron:
+ name: "{{ item.name }}"
+ job: "{{ item.job }}"
+ special_time: daily
+ user: hdfs
+ with_items:
+ - { name: "bro_purge", job: "{{ hdfs_bro_purge_cronjob }}" }
+ - { name: "yaf_purge", job: "{{ hdfs_yaf_purge_cronjob }}" }
+
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/82a586ac/deployment/roles/metron_streaming/tasks/main.yml
----------------------------------------------------------------------
diff --git a/deployment/roles/metron_streaming/tasks/main.yml b/deployment/roles/metron_streaming/tasks/main.yml
index b913011..152af56 100644
--- a/deployment/roles/metron_streaming/tasks/main.yml
+++ b/deployment/roles/metron_streaming/tasks/main.yml
@@ -131,3 +131,5 @@
- include: metron_topology.yml
+- include: hdfs_purge.yml
+
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/82a586ac/deployment/roles/metron_streaming/vars/main.yml
----------------------------------------------------------------------
diff --git a/deployment/roles/metron_streaming/vars/main.yml b/deployment/roles/metron_streaming/vars/main.yml
deleted file mode 100644
index f84b949..0000000
--- a/deployment/roles/metron_streaming/vars/main.yml
+++ /dev/null
@@ -1,37 +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.
-#
----
-metron_solr_jar_name: Metron-Solr-{{ metron_version }}.jar
-metron_elasticsearch_jar_name: Metron-Elasticsearch-{{ metron_version }}.jar
-metron_directory: /usr/metron/{{ metron_version }}
-metron_dataloads_name: Metron-DataLoads-{{ metron_version }}-archive.tar.gz
-metron_solr_bundle_name: Metron-Solr-{{ metron_version }}-archive.tar.gz
-metron_elasticsearch_bundle_name: Metron-Elasticsearch-{{ metron_version }}-archive.tar.gz
-metron_topologies_bundle_name: Metron-Topologies-{{ metron_version }}-archive.tar.gz
-metron_dataloads_path: "{{ playbook_dir }}/../../metron-streaming/Metron-DataLoads/target/{{ metron_dataloads_name }}"
-metron_solr_bundle_path: "{{ playbook_dir }}/../../metron-streaming/Metron-Solr/target/{{ metron_solr_bundle_name }}"
-metron_elasticsearch_bundle_path: "{{ playbook_dir }}/../../metron-streaming/Metron-Elasticsearch/target/{{ metron_elasticsearch_bundle_name }}"
-metron_topologies_bundle_path: "{{ playbook_dir }}/../../metron-streaming/Metron-Topologies/target/{{ metron_topologies_bundle_name }}"
-metron_src_config_path: "{{ playbook_dir }}/../../metron-streaming/Metron-Topologies/src/main/resources/Metron_Configs"
-metron_solr_properties_config_path: "{{ metron_directory }}/config/etc/env/solr.properties"
-metron_elasticsearch_properties_config_path: "{{ metron_directory }}/config/etc/env/elasticsearch.properties"
-elasticsearch_config_path: /etc/elasticsearch
-elasticsearch_cluster_name: metron
-elasticsearch_transport_port: 9300
-hbase_config_path: "/etc/hbase/conf"
-hdfs_config_path: "/etc/hadoop/conf"
-metron_hdfs_output_dir: "/apps/metron"
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/82a586ac/metron-streaming/Metron-Common/src/test/java/org/apache/metron/enrichment/EnrichmentConfigTest.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/test/java/org/apache/metron/enrichment/EnrichmentConfigTest.java b/metron-streaming/Metron-Common/src/test/java/org/apache/metron/enrichment/EnrichmentConfigTest.java
index 1f29403..243423c 100644
--- a/metron-streaming/Metron-Common/src/test/java/org/apache/metron/enrichment/EnrichmentConfigTest.java
+++ b/metron-streaming/Metron-Common/src/test/java/org/apache/metron/enrichment/EnrichmentConfigTest.java
@@ -1,3 +1,20 @@
+/**
+ * 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.enrichment;
import org.adrianwalker.multilinestring.Multiline;
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/82a586ac/metron-streaming/Metron-Common/src/test/java/org/apache/metron/hbase/converters/enrichment/EnrichmentConverterTest.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-Common/src/test/java/org/apache/metron/hbase/converters/enrichment/EnrichmentConverterTest.java b/metron-streaming/Metron-Common/src/test/java/org/apache/metron/hbase/converters/enrichment/EnrichmentConverterTest.java
index 9327b68..596f232 100644
--- a/metron-streaming/Metron-Common/src/test/java/org/apache/metron/hbase/converters/enrichment/EnrichmentConverterTest.java
+++ b/metron-streaming/Metron-Common/src/test/java/org/apache/metron/hbase/converters/enrichment/EnrichmentConverterTest.java
@@ -1,3 +1,20 @@
+/**
+ * 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.hbase.converters.enrichment;
import org.apache.hadoop.hbase.client.Put;
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/82a586ac/metron-streaming/Metron-DataLoads/pom.xml
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-DataLoads/pom.xml b/metron-streaming/Metron-DataLoads/pom.xml
index 239881c..53c39da 100644
--- a/metron-streaming/Metron-DataLoads/pom.xml
+++ b/metron-streaming/Metron-DataLoads/pom.xml
@@ -149,7 +149,12 @@
<version>${project.parent.version}</version>
<scope>test</scope>
</dependency>
-
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <version>${global_mockito_version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
<resources>
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/82a586ac/metron-streaming/Metron-DataLoads/src/main/assembly/assembly.xml
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-DataLoads/src/main/assembly/assembly.xml b/metron-streaming/Metron-DataLoads/src/main/assembly/assembly.xml
index 19c5464..c2c384b 100644
--- a/metron-streaming/Metron-DataLoads/src/main/assembly/assembly.xml
+++ b/metron-streaming/Metron-DataLoads/src/main/assembly/assembly.xml
@@ -28,6 +28,7 @@
</excludes>
<fileMode>0755</fileMode>
<lineEnding>unix</lineEnding>
+ <filtered>true</filtered>
</fileSet>
<fileSet>
<directory>${project.basedir}/target</directory>
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/82a586ac/metron-streaming/Metron-DataLoads/src/main/bash/prune_hdfs_files.sh
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-DataLoads/src/main/bash/prune_hdfs_files.sh b/metron-streaming/Metron-DataLoads/src/main/bash/prune_hdfs_files.sh
new file mode 100644
index 0000000..02b12dc
--- /dev/null
+++ b/metron-streaming/Metron-DataLoads/src/main/bash/prune_hdfs_files.sh
@@ -0,0 +1,21 @@
+#!/bin/bash
+#
+# 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.
+#
+
+yarn jar /usr/metron/${project.version}/lib/Metron-DataLoads-${project.version}.jar org.apache.metron.dataloads.bulk.HDFSDataPruner "$@"
+
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/82a586ac/metron-streaming/Metron-DataLoads/src/main/java/org/apache/metron/dataloads/bulk/HDFSDataPruner.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-DataLoads/src/main/java/org/apache/metron/dataloads/bulk/HDFSDataPruner.java b/metron-streaming/Metron-DataLoads/src/main/java/org/apache/metron/dataloads/bulk/HDFSDataPruner.java
new file mode 100644
index 0000000..5fbc27e
--- /dev/null
+++ b/metron-streaming/Metron-DataLoads/src/main/java/org/apache/metron/dataloads/bulk/HDFSDataPruner.java
@@ -0,0 +1,256 @@
+/**
+ * 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.dataloads.bulk;
+
+
+import org.apache.commons.cli.*;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.concurrent.TimeUnit;
+
+public class HDFSDataPruner {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HDFSDataPruner.class);
+ private Date startDate;
+ private long firstTimeMillis;
+ private long lastTimeMillis;
+ private Path globPath;
+ protected FileSystem fileSystem;
+
+ private HDFSDataPruner() {
+ }
+
+
+ HDFSDataPruner(Date startDate, Integer numDays, String fsUri, String globPath) throws IOException {
+
+ this.startDate = dateAtMidnight(startDate);
+ this.lastTimeMillis = startDate.getTime();
+ this.firstTimeMillis = lastTimeMillis - TimeUnit.DAYS.toMillis(numDays);
+ this.globPath = new Path(globPath);
+ Configuration conf = new Configuration();
+ conf.set("fs.defaultFS", fsUri);
+ this.fileSystem = FileSystem.get(conf);
+
+ }
+
+
+ public static void main(String... argv) throws IOException, java.text.ParseException, ClassNotFoundException, InterruptedException {
+
+ /**
+ * Example
+ * start=$(date -d '30 days ago' +%m/%d/%Y)
+ * yarn jar Metron-DataLoads-0.1BETA.jar org.apache.metron.dataloads.bulk.HDFSDataPruner -f hdfs://ec2-52-36-25-217.us-west-2.compute.amazonaws.com:8020 -g '/apps/metron/enrichment/indexed/bro_doc/*enrichment-*' -s $(date -d '30 days ago' +%m/%d/%Y) -n 1;
+ * echo ${start}
+ **/
+
+ Options options = new Options();
+ Options help = new Options();
+
+ {
+ Option o = new Option("h", "help", false, "This screen");
+ o.setRequired(false);
+ help.addOption(o);
+ }
+ {
+ Option o = new Option("s", "start-date", true, "Starting Date (MM/DD/YYYY)");
+ o.setArgName("START_DATE");
+ o.setRequired(true);
+ options.addOption(o);
+ }
+ {
+ Option o = new Option("f", "filesystem", true, "Filesystem uri - e.g. hdfs://host:8020 or file:///");
+ o.setArgName("FILESYSTEM");
+ o.setRequired(true);
+ options.addOption(o);
+ }
+ {
+ Option o = new Option("n", "numdays", true, "Number of days back to purge");
+ o.setArgName("NUMDAYS");
+ o.setRequired(true);
+ options.addOption(o);
+ }
+ {
+ Option o = new Option("g", "glob-string", true, "Glob filemask for files to delete - e.g. /apps/metron/enrichment/bro_doc/file-*");
+ o.setArgName("GLOBSTRING");
+ o.setRequired(true);
+ options.addOption(o);
+ }
+
+ try {
+
+ CommandLineParser parser = new PosixParser();
+ CommandLine cmd = null;
+
+ try {
+
+ cmd = parser.parse(help,argv,true);
+ if( cmd.getOptions().length > 0){
+ final HelpFormatter usageFormatter = new HelpFormatter();
+ usageFormatter.printHelp("HDFSDataPruner", null, options, null, true);
+ System.exit(0);
+ }
+
+ cmd = parser.parse(options, argv);
+
+ } catch (ParseException pe) {
+
+ final HelpFormatter usageFormatter = new HelpFormatter();
+ usageFormatter.printHelp("HDFSDataPruner", null, options, null, true);
+ System.exit(-1);
+
+ }
+
+ String start = cmd.getOptionValue("s");
+ Date startDate = new SimpleDateFormat("MM/dd/yyyy").parse(start);
+ String fileSystemUri = cmd.getOptionValue("f");
+ Integer numDays = Integer.parseInt(cmd.getOptionValue("n"));
+ String globString = cmd.getOptionValue("g");
+
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Running prune with args: " + startDate + " " + numDays + " " + fileSystemUri + " " + globString);
+ }
+
+ HDFSDataPruner pruner = new HDFSDataPruner(startDate, numDays, fileSystemUri, globString);
+
+ LOG.info("Pruned " + pruner.prune() + " files from " + fileSystemUri + globString);
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ System.exit(-1);
+ }
+
+ }
+
+ public Long prune() throws IOException {
+
+ Long filesPruned = new Long(0);
+
+ Date today = dateAtMidnight(new Date());
+
+ if (!today.after(startDate)) {
+ throw new RuntimeException("Prune Start Date must be prior to today");
+ }
+
+ FileStatus[] filesToDelete = fileSystem.globStatus(globPath, new HDFSDataPruner.DateFileFilter(this));
+
+ for (FileStatus fileStatus : filesToDelete) {
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Deleting File: " + fileStatus.getPath());
+ }
+
+ fileSystem.delete(fileStatus.getPath(), false);
+
+ filesPruned++;
+ }
+
+ return filesPruned;
+ }
+
+ private Date dateAtMidnight(Date date) {
+
+ Calendar calendar = Calendar.getInstance();
+
+ calendar.setTime(date);
+ calendar.set(Calendar.HOUR, 0);
+ calendar.set(Calendar.MINUTE, 0);
+ calendar.set(Calendar.SECOND, 0);
+ calendar.set(Calendar.MILLISECOND, 0);
+
+ return calendar.getTime();
+
+ }
+
+ class DateFileFilter extends Configured implements PathFilter {
+
+ HDFSDataPruner pruner;
+ Boolean failOnError = false;
+
+ DateFileFilter(HDFSDataPruner pruner) {
+ this.pruner = pruner;
+ }
+
+ DateFileFilter(HDFSDataPruner pruner, Boolean failOnError) {
+
+ this(pruner);
+ this.failOnError = failOnError;
+
+ }
+
+ @Override
+ public boolean accept(Path path) {
+ try {
+
+ if(pruner.LOG.isDebugEnabled()) {
+ pruner.LOG.debug("ACCEPT - working with file: " + path);
+ }
+
+ if (pruner.fileSystem.isDirectory(path)) {
+ return false;
+
+ }
+ } catch (IOException e) {
+
+ pruner.LOG.error("IOException", e);
+
+ if (failOnError) {
+ throw new RuntimeException(e);
+ }
+
+ return false;
+ }
+
+ try {
+
+ FileStatus file = pruner.fileSystem.getFileStatus(path);
+ long fileModificationTime = file.getModificationTime();
+ boolean accept = false;
+
+ if (fileModificationTime >= pruner.firstTimeMillis && fileModificationTime < pruner.lastTimeMillis) {
+
+ accept = true;
+ }
+
+ return accept;
+
+ } catch (IOException e) {
+
+ pruner.LOG.error("IOException", e);
+
+ if (failOnError) {
+ throw new RuntimeException(e);
+ }
+
+ return false;
+ }
+
+ }
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/82a586ac/metron-streaming/Metron-DataLoads/src/test/java/org/apache/metron/dataloads/bulk/HDFSDataPrunerTest.java
----------------------------------------------------------------------
diff --git a/metron-streaming/Metron-DataLoads/src/test/java/org/apache/metron/dataloads/bulk/HDFSDataPrunerTest.java b/metron-streaming/Metron-DataLoads/src/test/java/org/apache/metron/dataloads/bulk/HDFSDataPrunerTest.java
new file mode 100644
index 0000000..b08c2a5
--- /dev/null
+++ b/metron-streaming/Metron-DataLoads/src/test/java/org/apache/metron/dataloads/bulk/HDFSDataPrunerTest.java
@@ -0,0 +1,164 @@
+/**
+ * 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.dataloads.bulk;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.*;
+
+
+public class HDFSDataPrunerTest {
+
+
+ private static File dataPath = new File("src/test/resources/HDFSDataPrunerTest");
+
+ private Date todaysDate;
+ private Date yesterday = new Date();
+
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+
+ if (dataPath.isDirectory()) {
+ dataPath.delete();
+ }
+
+ if (!dataPath.mkdirs()) {
+ throw new RuntimeException("Couldn't create dataPath at: " + dataPath.getAbsolutePath());
+ }
+
+ dataPath.deleteOnExit();
+
+ }
+
+
+ @Before
+ public void setUp() throws Exception {
+
+ Calendar today = Calendar.getInstance();
+ today.clear(Calendar.HOUR);
+ today.clear(Calendar.MINUTE);
+ today.clear(Calendar.SECOND);
+ todaysDate = today.getTime();
+ yesterday.setTime(todaysDate.getTime() - TimeUnit.DAYS.toMillis(1));
+
+ }
+
+ @Test(expected = RuntimeException.class)
+ public void testFailsOnTodaysDate() throws Exception {
+
+ HDFSDataPruner pruner = new HDFSDataPruner(todaysDate, 30, "file:///", dataPath.getAbsolutePath() + "/file-*");
+ pruner.prune();
+
+ }
+
+ @Test
+ public void testDeletesCorrectFiles() throws Exception {
+
+ createTestFiles();
+
+ HDFSDataPruner pruner = new HDFSDataPruner(yesterday, 30, "file:///", dataPath.getAbsolutePath() + "/file-*");
+
+ Long prunedCount = pruner.prune();
+ assertTrue("Should have pruned 45 files- pruned: " + prunedCount, 45 == prunedCount);
+
+ //Verify first five files remain
+ File[] filesLeft = dataPath.listFiles();
+ File[] filesList = new File[filesLeft.length];
+ for (int i = 0; i < 5; i++) {
+ filesList[i] = new File(dataPath.getPath() + "//file-" + String.format("%02d", i));
+ }
+ assertArrayEquals("First four files should have been left behind", filesLeft, filesList);
+
+
+ }
+
+ @Test(expected = RuntimeException.class)
+ public void testThrowsIsDirectory() throws Exception {
+
+ FileSystem testFS = mock(FileSystem.class);
+ when(testFS.isDirectory((Path) any())).thenThrow(new IOException("Test Exception"));
+
+ HDFSDataPruner pruner = new HDFSDataPruner(yesterday, 30, "file:///", dataPath.getAbsolutePath() + "/file-*");
+ pruner.fileSystem = testFS;
+ HDFSDataPruner.DateFileFilter filter = pruner.new DateFileFilter(pruner, true);
+
+ filter.accept(new Path("foo"));
+
+ }
+
+ @Test(expected = RuntimeException.class)
+ public void testThrowBadFile() throws Exception {
+
+ FileSystem testFS = mock(FileSystem.class);
+ when(testFS.isDirectory((Path) any())).thenReturn(false);
+ when(testFS.getFileStatus((Path) any())).thenThrow(new IOException("Test Exception"));
+
+ HDFSDataPruner pruner = new HDFSDataPruner(yesterday, 30, "file:///", dataPath.getAbsolutePath() + "/file-*");
+ pruner.fileSystem = testFS;
+ HDFSDataPruner.DateFileFilter filter = pruner.new DateFileFilter(pruner, true);
+
+ filter.accept(new Path("foo"));
+
+ }
+
+ private void createTestFiles() throws IOException {
+
+ //create files
+ for (int i = 0; i < 50; i++) {
+ File file = new File(dataPath.getAbsolutePath() + "//file-" + String.format("%02d", i));
+ file.createNewFile();
+ file.deleteOnExit();
+ }
+
+ //Set modification date today - 1 day
+ for (int i = 5; i < 25; i++) {
+ File file = new File(dataPath.getAbsolutePath() + "//file-" + String.format("%02d", i));
+ file.setLastModified(todaysDate.getTime() - TimeUnit.DAYS.toMillis(1));
+ file.deleteOnExit();
+ }
+
+ //Set modification date today - 10 days
+ for (int i = 25; i < 40; i++) {
+ File file = new File(dataPath.getAbsolutePath() + "//file-" + String.format("%02d", i));
+ file.setLastModified(todaysDate.getTime() - TimeUnit.DAYS.toMillis(10));
+ file.deleteOnExit();
+ }
+
+ //Set modification date today - 20 days
+ for (int i = 40; i < 50; i++) {
+ File file = new File(dataPath.getAbsolutePath() + "//file-" + String.format("%02d", i));
+ file.setLastModified(todaysDate.getTime() - TimeUnit.DAYS.toMillis(20));
+ file.deleteOnExit();
+ }
+
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/82a586ac/metron-streaming/pom.xml
----------------------------------------------------------------------
diff --git a/metron-streaming/pom.xml b/metron-streaming/pom.xml
index d4241df..1bfa76d 100644
--- a/metron-streaming/pom.xml
+++ b/metron-streaming/pom.xml
@@ -48,7 +48,7 @@
<global_slf4j_version>1.7.7</global_slf4j_version>
<global_opencsv_version>3.7</global_opencsv_version>
<global_solr_version>5.2.1</global_solr_version>
- <global_mockito_version>1.9.5</global_mockito_version>
+ <global_mockito_version>1.10.16</global_mockito_version>
</properties>
<licenses>
<license>