From 4bc0438d2c59df4753a266764ffed27975c28717 Mon Sep 17 00:00:00 2001 From: merrimanr Date: Fri, 3 Aug 2018 16:44:04 -0500 Subject: [PATCH 1/5] initial commit --- .../java/org/apache/metron/pcap/query/PcapCli.java | 2 +- .../main/java/org/apache/metron/pcap/mr/PcapJob.java | 11 +++++++++++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/PcapCli.java b/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/PcapCli.java index 0b06b0c66f..9e8235ff50 100644 --- a/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/PcapCli.java +++ b/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/PcapCli.java @@ -80,7 +80,7 @@ public int run(String[] args) { Pageable results; // write to local FS in the executing directory String execDir = System.getProperty("user.dir"); - + jobRunner.reportStatus(System.out); if ("fixed".equals(jobType)) { FixedCliParser fixedParser = new FixedCliParser(prefixStrategy); FixedPcapConfig config = null; diff --git a/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/mr/PcapJob.java b/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/mr/PcapJob.java index ea2aa29092..e10ffe58a3 100644 --- a/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/mr/PcapJob.java +++ b/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/mr/PcapJob.java @@ -23,6 +23,7 @@ import com.google.common.base.Joiner; import java.io.IOException; +import java.io.PrintStream; import java.lang.invoke.MethodHandles; import java.text.DateFormat; import java.text.SimpleDateFormat; @@ -88,6 +89,7 @@ public class PcapJob implements Statusable { private Timer timer; private long statusInterval; // how often timer thread checks job status. private long completeCheckInterval; // how long we sleep between isDone checks in get() + private PrintStream statusPrintStream; // report status to this print stream public static enum PCAP_COUNTER { MALFORMED_PACKET_COUNT @@ -464,15 +466,24 @@ public synchronized JobStatus getStatus() throws JobException { public Pageable get() throws JobException, InterruptedException { for (; ; ) { JobStatus status = getStatus(); + if (statusPrintStream != null) { + statusPrintStream.println(String.format("Job status = %s, percent complete = %.2f%%", + status.getState(), status.getPercentComplete())); + } if (status.getState() == State.SUCCEEDED || status.getState() == State.KILLED || status.getState() == State.FAILED) { + return getFinalResults(); } Thread.sleep(completeCheckInterval); } } + public void reportStatus(PrintStream printStream) { + statusPrintStream = printStream; + } + private synchronized Pageable getFinalResults() { return new PcapPages(finalResults); } From 4df186c9a2839d867c859b1c3ed4fa4ab5a0084f Mon Sep 17 00:00:00 2001 From: merrimanr Date: Mon, 6 Aug 2018 14:53:14 -0500 Subject: [PATCH 2/5] use Job.monitorAndPrintJob --- .../java/org/apache/metron/pcap/query/PcapCli.java | 8 +++++--- .../main/java/org/apache/metron/pcap/mr/PcapJob.java | 11 ++--------- 2 files changed, 7 insertions(+), 12 deletions(-) diff --git a/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/PcapCli.java b/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/PcapCli.java index 9e8235ff50..65e5e6cda5 100644 --- a/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/PcapCli.java +++ b/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/PcapCli.java @@ -80,7 +80,7 @@ public int run(String[] args) { Pageable results; // write to local FS in the executing directory String execDir = System.getProperty("user.dir"); - jobRunner.reportStatus(System.out); + if ("fixed".equals(jobType)) { FixedCliParser fixedParser = new FixedCliParser(prefixStrategy); FixedPcapConfig config = null; @@ -101,7 +101,8 @@ public int run(String[] args) { PcapOptions.HADOOP_CONF.put(commonConfig, hadoopConf); try { PcapOptions.FILESYSTEM.put(commonConfig, FileSystem.get(hadoopConf)); - results = jobRunner.submit(PcapFinalizerStrategies.CLI, commonConfig).get(); + jobRunner.submit(PcapFinalizerStrategies.CLI, commonConfig); + jobRunner.monitorJob(); } catch (IOException|InterruptedException | JobException e) { LOGGER.error("Failed to execute fixed filter job: {}", e.getMessage(), e); return -1; @@ -125,7 +126,8 @@ public int run(String[] args) { PcapOptions.HADOOP_CONF.put(commonConfig, hadoopConf); try { PcapOptions.FILESYSTEM.put(commonConfig, FileSystem.get(hadoopConf)); - results = jobRunner.submit(PcapFinalizerStrategies.CLI, commonConfig).get(); + jobRunner.submit(PcapFinalizerStrategies.CLI, commonConfig); + jobRunner.monitorJob(); } catch (IOException| InterruptedException | JobException e) { LOGGER.error("Failed to execute fixed filter job: {}", e.getMessage(), e); return -1; diff --git a/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/mr/PcapJob.java b/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/mr/PcapJob.java index e10ffe58a3..58957fc158 100644 --- a/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/mr/PcapJob.java +++ b/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/mr/PcapJob.java @@ -23,7 +23,6 @@ import com.google.common.base.Joiner; import java.io.IOException; -import java.io.PrintStream; import java.lang.invoke.MethodHandles; import java.text.DateFormat; import java.text.SimpleDateFormat; @@ -89,7 +88,6 @@ public class PcapJob implements Statusable { private Timer timer; private long statusInterval; // how often timer thread checks job status. private long completeCheckInterval; // how long we sleep between isDone checks in get() - private PrintStream statusPrintStream; // report status to this print stream public static enum PCAP_COUNTER { MALFORMED_PACKET_COUNT @@ -466,22 +464,17 @@ public synchronized JobStatus getStatus() throws JobException { public Pageable get() throws JobException, InterruptedException { for (; ; ) { JobStatus status = getStatus(); - if (statusPrintStream != null) { - statusPrintStream.println(String.format("Job status = %s, percent complete = %.2f%%", - status.getState(), status.getPercentComplete())); - } if (status.getState() == State.SUCCEEDED || status.getState() == State.KILLED || status.getState() == State.FAILED) { - return getFinalResults(); } Thread.sleep(completeCheckInterval); } } - public void reportStatus(PrintStream printStream) { - statusPrintStream = printStream; + public void monitorJob() throws IOException, InterruptedException { + mrJob.monitorAndPrintJob(); } private synchronized Pageable getFinalResults() { From 3df98443649a3cf6df44f28716cc476ca87b9c9a Mon Sep 17 00:00:00 2001 From: merrimanr Date: Mon, 6 Aug 2018 15:48:25 -0500 Subject: [PATCH 3/5] adjusted pcapcli unit test --- .../java/org/apache/metron/pcap/query/PcapCliTest.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/query/PcapCliTest.java b/metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/query/PcapCliTest.java index c7d6fdf525..07f938ecdc 100644 --- a/metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/query/PcapCliTest.java +++ b/metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/query/PcapCliTest.java @@ -101,7 +101,7 @@ public void runs_fixed_pcap_filter_job_with_default_argument_list() throws Excep PcapCli cli = new PcapCli(jobRunner, prefixStrategy); assertThat("Expect no errors on run", cli.run(args), equalTo(0)); - verify(jobRunner).get(); + verify(jobRunner).monitorJob(); } /** @@ -168,7 +168,7 @@ public void runs_fixed_pcap_filter_job_with_full_argument_list_and_default_datef PcapCli cli = new PcapCli(jobRunner, prefixStrategy); assertThat("Expect no errors on run", cli.run(args), equalTo(0)); - verify(jobRunner).get(); + verify(jobRunner).monitorJob(); } @Test @@ -214,7 +214,7 @@ public void runs_fixed_pcap_filter_job_with_full_argument_list() throws Exceptio PcapCli cli = new PcapCli(jobRunner, prefixStrategy); assertThat("Expect no errors on run", cli.run(args), equalTo(0)); - verify(jobRunner).get(); + verify(jobRunner).monitorJob(); } private long asNanos(String inDate, String format) throws ParseException { @@ -247,7 +247,7 @@ public void runs_query_pcap_filter_job_with_default_argument_list() throws Excep PcapCli cli = new PcapCli(jobRunner, prefixStrategy); assertThat("Expect no errors on run", cli.run(args), equalTo(0)); - verify(jobRunner).get(); + verify(jobRunner).monitorJob(); } @Test @@ -277,7 +277,7 @@ public void runs_query_pcap_filter_job_with_full_argument_list() throws Exceptio PcapCli cli = new PcapCli(jobRunner, prefixStrategy); assertThat("Expect no errors on run", cli.run(args), equalTo(0)); - verify(jobRunner).get(); + verify(jobRunner).monitorJob(); } // INVALID OPTION CHECKS From f17d50fbfed4fd14e5cdd5a718cc973f51d2c4bc Mon Sep 17 00:00:00 2001 From: merrimanr Date: Tue, 7 Aug 2018 13:28:41 -0500 Subject: [PATCH 4/5] pr feedback --- metron-platform/metron-pcap-backend/README.md | 2 ++ .../org/apache/metron/pcap/query/CliParser.java | 4 ++++ .../org/apache/metron/pcap/query/PcapCli.java | 13 +++++++++++-- .../apache/metron/pcap/query/PcapCliTest.java | 16 +++++++++++----- .../apache/metron/pcap/config/PcapConfig.java | 10 ++++++++++ 5 files changed, 38 insertions(+), 7 deletions(-) diff --git a/metron-platform/metron-pcap-backend/README.md b/metron-platform/metron-pcap-backend/README.md index e7960e3c03..2ff20d87ee 100644 --- a/metron-platform/metron-pcap-backend/README.md +++ b/metron-platform/metron-pcap-backend/README.md @@ -132,6 +132,7 @@ usage: Fixed filter options -nr,--num_reducers The number of reducers to use. Default is 10. -h,--help Display help + -ps,--print_status Print the status of the job as it runs -ir,--include_reverse Indicates if filter should check swapped src/dest addresses and IPs -p,--protocol IP Protocol @@ -154,6 +155,7 @@ usage: Query filter options -nr,--num_reducers The number of reducers to use. Default is 10. -h,--help Display help + -ps,--print_status Print the status of the job as it runs -q,--query Query string to use as a filter -st,--start_time (required) Packet start time range. ``` diff --git a/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/CliParser.java b/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/CliParser.java index e6534c5a23..3e45fd433e 100644 --- a/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/CliParser.java +++ b/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/CliParser.java @@ -48,6 +48,7 @@ public Options buildOptions() { options.addOption(newOption("rpf", "records_per_file", true, String.format("Number of records to include in each output pcap file (defaults to %s)", NUM_RECORDS_PER_FILE_DEFAULT))); options.addOption(newOption("et", "end_time", true, "Packet end time range. Default is current system time.")); options.addOption(newOption("df", "date_format", true, "Date format to use for parsing start_time and end_time. Default is to use time in millis since the epoch.")); + options.addOption(newOption("ps", "print_status", false, "Print the status of the job as it runs")); return options; } @@ -118,6 +119,9 @@ public void parse(CommandLine commandLine, PcapConfig config) throws java.text.P //no-op } } + if (commandLine.hasOption("print_status")) { + config.setPrintStatus(true); + } } public void printHelp(String msg, Options opts) { diff --git a/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/PcapCli.java b/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/PcapCli.java index 65e5e6cda5..c6b31bdb30 100644 --- a/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/PcapCli.java +++ b/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/PcapCli.java @@ -30,6 +30,7 @@ import org.apache.metron.common.utils.timestamp.TimestampConverters; import org.apache.metron.job.JobException; import org.apache.metron.job.Pageable; +import org.apache.metron.job.Statusable; import org.apache.metron.pcap.config.PcapOptions; import org.apache.metron.pcap.config.FixedPcapConfig; import org.apache.metron.pcap.config.PcapConfig; @@ -102,7 +103,11 @@ public int run(String[] args) { try { PcapOptions.FILESYSTEM.put(commonConfig, FileSystem.get(hadoopConf)); jobRunner.submit(PcapFinalizerStrategies.CLI, commonConfig); - jobRunner.monitorJob(); + if (config.printStatus()) { + jobRunner.monitorJob(); + LOGGER.info("Finalizing results"); + } + jobRunner.get(); } catch (IOException|InterruptedException | JobException e) { LOGGER.error("Failed to execute fixed filter job: {}", e.getMessage(), e); return -1; @@ -127,7 +132,11 @@ public int run(String[] args) { try { PcapOptions.FILESYSTEM.put(commonConfig, FileSystem.get(hadoopConf)); jobRunner.submit(PcapFinalizerStrategies.CLI, commonConfig); - jobRunner.monitorJob(); + if (config.printStatus()) { + jobRunner.monitorJob(); + LOGGER.info("Finalizing results"); + } + jobRunner.get(); } catch (IOException| InterruptedException | JobException e) { LOGGER.error("Failed to execute fixed filter job: {}", e.getMessage(), e); return -1; diff --git a/metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/query/PcapCliTest.java b/metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/query/PcapCliTest.java index 07f938ecdc..523005d1a7 100644 --- a/metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/query/PcapCliTest.java +++ b/metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/query/PcapCliTest.java @@ -101,7 +101,7 @@ public void runs_fixed_pcap_filter_job_with_default_argument_list() throws Excep PcapCli cli = new PcapCli(jobRunner, prefixStrategy); assertThat("Expect no errors on run", cli.run(args), equalTo(0)); - verify(jobRunner).monitorJob(); + verify(jobRunner).get(); } /** @@ -145,7 +145,8 @@ public void runs_fixed_pcap_filter_job_with_full_argument_list_and_default_datef "-protocol", "6", "-include_reverse", "-num_reducers", "10", - "-records_per_file", "1000" + "-records_per_file", "1000", + "-ps" }; Map query = new HashMap() {{ put(Constants.Fields.SRC_ADDR.getName(), "192.168.1.1"); @@ -169,6 +170,7 @@ public void runs_fixed_pcap_filter_job_with_full_argument_list_and_default_datef PcapCli cli = new PcapCli(jobRunner, prefixStrategy); assertThat("Expect no errors on run", cli.run(args), equalTo(0)); verify(jobRunner).monitorJob(); + verify(jobRunner).get(); } @Test @@ -187,7 +189,8 @@ public void runs_fixed_pcap_filter_job_with_full_argument_list() throws Exceptio "-protocol", "6", "-include_reverse", "-num_reducers", "10", - "-records_per_file", "1000" + "-records_per_file", "1000", + "-ps" }; Map query = new HashMap() {{ put(Constants.Fields.SRC_ADDR.getName(), "192.168.1.1"); @@ -215,6 +218,7 @@ public void runs_fixed_pcap_filter_job_with_full_argument_list() throws Exceptio PcapCli cli = new PcapCli(jobRunner, prefixStrategy); assertThat("Expect no errors on run", cli.run(args), equalTo(0)); verify(jobRunner).monitorJob(); + verify(jobRunner).get(); } private long asNanos(String inDate, String format) throws ParseException { @@ -247,7 +251,7 @@ public void runs_query_pcap_filter_job_with_default_argument_list() throws Excep PcapCli cli = new PcapCli(jobRunner, prefixStrategy); assertThat("Expect no errors on run", cli.run(args), equalTo(0)); - verify(jobRunner).monitorJob(); + verify(jobRunner).get(); } @Test @@ -260,7 +264,8 @@ public void runs_query_pcap_filter_job_with_full_argument_list() throws Exceptio "-base_path", "/base/path", "-base_output_path", "/base/output/path", "-query", "some query string", - "-records_per_file", "1000" + "-records_per_file", "1000", + "-ps" }; String query = "some query string"; @@ -278,6 +283,7 @@ public void runs_query_pcap_filter_job_with_full_argument_list() throws Exceptio PcapCli cli = new PcapCli(jobRunner, prefixStrategy); assertThat("Expect no errors on run", cli.run(args), equalTo(0)); verify(jobRunner).monitorJob(); + verify(jobRunner).get(); } // INVALID OPTION CHECKS diff --git a/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/config/PcapConfig.java b/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/config/PcapConfig.java index 26509be726..d88d5523d9 100644 --- a/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/config/PcapConfig.java +++ b/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/config/PcapConfig.java @@ -31,6 +31,7 @@ public class PcapConfig extends AbstractMapDecorator{ public interface PrefixStrategy extends Function{} private boolean showHelp; + private boolean printStatus; private DateFormat dateFormat; public PcapConfig() { @@ -40,6 +41,7 @@ public PcapConfig() { public PcapConfig(PrefixStrategy prefixStrategy) { this(); setShowHelp(false); + setPrintStatus(false); setBasePath(""); setBaseInterimResultPath(""); setStartTimeMs(-1L); @@ -73,6 +75,14 @@ public void setShowHelp(boolean showHelp) { this.showHelp = showHelp; } + public boolean printStatus() { + return printStatus; + } + + public void setPrintStatus(boolean printStatus) { + this.printStatus = printStatus; + } + public String getBasePath() { return PcapOptions.BASE_PATH.get(this, String.class); } From 8fbdd4f3c9f627bd2ea53d6ca34a709fd591b028 Mon Sep 17 00:00:00 2001 From: merrimanr Date: Thu, 9 Aug 2018 16:00:49 -0500 Subject: [PATCH 5/5] pr feedback --- .../org/apache/metron/pcap/query/CliParser.java | 2 +- .../org/apache/metron/pcap/query/PcapCli.java | 15 ++------------- .../org/apache/metron/pcap/query/PcapCliTest.java | 6 +++--- .../org/apache/metron/pcap/config/PcapConfig.java | 11 +++++------ .../apache/metron/pcap/config/PcapOptions.java | 3 ++- .../java/org/apache/metron/pcap/mr/PcapJob.java | 7 +++++++ 6 files changed, 20 insertions(+), 24 deletions(-) diff --git a/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/CliParser.java b/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/CliParser.java index d07de13cb5..69c725c9a5 100644 --- a/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/CliParser.java +++ b/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/CliParser.java @@ -127,7 +127,7 @@ public void parse(CommandLine commandLine, PcapConfig config) throws java.text.P } } if (commandLine.hasOption("print_status")) { - config.setPrintStatus(true); + config.setPrintJobStatus(true); } } diff --git a/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/PcapCli.java b/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/PcapCli.java index 590324001d..c23f037873 100644 --- a/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/PcapCli.java +++ b/metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/PcapCli.java @@ -30,7 +30,6 @@ import org.apache.metron.common.utils.timestamp.TimestampConverters; import org.apache.metron.job.JobException; import org.apache.metron.job.Pageable; -import org.apache.metron.job.Statusable; import org.apache.metron.pcap.config.PcapOptions; import org.apache.metron.pcap.config.FixedPcapConfig; import org.apache.metron.pcap.config.PcapConfig; @@ -103,12 +102,7 @@ public int run(String[] args) { PcapOptions.HADOOP_CONF.put(commonConfig, hadoopConf); try { PcapOptions.FILESYSTEM.put(commonConfig, FileSystem.get(hadoopConf)); - jobRunner.submit(PcapFinalizerStrategies.CLI, commonConfig); - if (config.printStatus()) { - jobRunner.monitorJob(); - LOGGER.info("Finalizing results"); - } - jobRunner.get(); + results = jobRunner.submit(PcapFinalizerStrategies.CLI, commonConfig).get(); } catch (IOException|InterruptedException | JobException e) { LOGGER.error("Failed to execute fixed filter job: {}", e.getMessage(), e); return -1; @@ -133,12 +127,7 @@ public int run(String[] args) { PcapOptions.HADOOP_CONF.put(commonConfig, hadoopConf); try { PcapOptions.FILESYSTEM.put(commonConfig, FileSystem.get(hadoopConf)); - jobRunner.submit(PcapFinalizerStrategies.CLI, commonConfig); - if (config.printStatus()) { - jobRunner.monitorJob(); - LOGGER.info("Finalizing results"); - } - jobRunner.get(); + results = jobRunner.submit(PcapFinalizerStrategies.CLI, commonConfig).get(); } catch (IOException| InterruptedException | JobException e) { LOGGER.error("Failed to execute fixed filter job: {}", e.getMessage(), e); return -1; diff --git a/metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/query/PcapCliTest.java b/metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/query/PcapCliTest.java index 3b6e744b7e..7c75224940 100644 --- a/metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/query/PcapCliTest.java +++ b/metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/query/PcapCliTest.java @@ -166,12 +166,12 @@ public void runs_fixed_pcap_filter_job_with_full_argument_list_and_default_datef PcapOptions.START_TIME_MS.put(config, 500L); PcapOptions.END_TIME_MS.put(config, 1000L); PcapOptions.NUM_RECORDS_PER_FILE.put(config, 1000); + PcapOptions.PRINT_JOB_STATUS.put(config, true); when(jobRunner.submit(isA(Finalizer.class), argThat(mapContaining(config)))).thenReturn(jobRunner); PcapCli cli = new PcapCli(jobRunner, prefixStrategy); assertThat("Expect no errors on run", cli.run(args), equalTo(0)); - verify(jobRunner).monitorJob(); verify(jobRunner).get(); } @@ -214,12 +214,12 @@ public void runs_fixed_pcap_filter_job_with_full_argument_list() throws Exceptio PcapOptions.START_TIME_MS.put(config, startAsNanos / 1000000L); // needed bc defaults in config PcapOptions.END_TIME_MS.put(config, endAsNanos / 1000000L); // needed bc defaults in config PcapOptions.NUM_RECORDS_PER_FILE.put(config, 1000); + PcapOptions.PRINT_JOB_STATUS.put(config, true); when(jobRunner.submit(isA(Finalizer.class), argThat(mapContaining(config)))).thenReturn(jobRunner); PcapCli cli = new PcapCli(jobRunner, prefixStrategy); assertThat("Expect no errors on run", cli.run(args), equalTo(0)); - verify(jobRunner).monitorJob(); verify(jobRunner).get(); } @@ -279,12 +279,12 @@ public void runs_query_pcap_filter_job_with_full_argument_list() throws Exceptio PcapOptions.START_TIME_MS.put(config, 500L); // needed bc defaults in config PcapOptions.END_TIME_MS.put(config, 1000L); // needed bc defaults in config PcapOptions.NUM_RECORDS_PER_FILE.put(config, 1000); + PcapOptions.PRINT_JOB_STATUS.put(config, true); when(jobRunner.submit(isA(Finalizer.class), argThat(mapContaining(config)))).thenReturn(jobRunner); PcapCli cli = new PcapCli(jobRunner, prefixStrategy); assertThat("Expect no errors on run", cli.run(args), equalTo(0)); - verify(jobRunner).monitorJob(); verify(jobRunner).get(); } diff --git a/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/config/PcapConfig.java b/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/config/PcapConfig.java index d88d5523d9..cbb8170839 100644 --- a/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/config/PcapConfig.java +++ b/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/config/PcapConfig.java @@ -31,7 +31,6 @@ public class PcapConfig extends AbstractMapDecorator{ public interface PrefixStrategy extends Function{} private boolean showHelp; - private boolean printStatus; private DateFormat dateFormat; public PcapConfig() { @@ -41,7 +40,7 @@ public PcapConfig() { public PcapConfig(PrefixStrategy prefixStrategy) { this(); setShowHelp(false); - setPrintStatus(false); + setPrintJobStatus(false); setBasePath(""); setBaseInterimResultPath(""); setStartTimeMs(-1L); @@ -75,12 +74,12 @@ public void setShowHelp(boolean showHelp) { this.showHelp = showHelp; } - public boolean printStatus() { - return printStatus; + public boolean printJobStatus() { + return PcapOptions.PRINT_JOB_STATUS.get(this, Boolean.class); } - public void setPrintStatus(boolean printStatus) { - this.printStatus = printStatus; + public void setPrintJobStatus(boolean printJobStatus) { + PcapOptions.PRINT_JOB_STATUS.put(this, printJobStatus); } public String getBasePath() { diff --git a/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/config/PcapOptions.java b/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/config/PcapOptions.java index 3d7c4f61c8..203c800ee8 100644 --- a/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/config/PcapOptions.java +++ b/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/config/PcapOptions.java @@ -40,7 +40,8 @@ public enum PcapOptions implements ConfigOption { FIELDS("fields"), FILTER_IMPL("filterImpl"), HADOOP_CONF("hadoopConf"), - FILESYSTEM("fileSystem"); + FILESYSTEM("fileSystem"), + PRINT_JOB_STATUS("printJobStatus"); public static final BiFunction STRING_TO_PATH = (s, o) -> o == null ? null : new Path(o.toString()); diff --git a/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/mr/PcapJob.java b/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/mr/PcapJob.java index a2a7ef8267..10f31b4b72 100644 --- a/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/mr/PcapJob.java +++ b/metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/mr/PcapJob.java @@ -467,6 +467,13 @@ public synchronized JobStatus getStatus() throws JobException { */ @Override public Pageable get() throws JobException, InterruptedException { + if (PcapOptions.PRINT_JOB_STATUS.get(configuration, Boolean.class)) { + try { + mrJob.monitorAndPrintJob(); + } catch (IOException e) { + throw new JobException("Could not monitor job status", e); + } + } for (; ; ) { JobStatus status = getStatus(); if (status.getState() == State.SUCCEEDED