METRON-1722: PcapCLI should print progress to stdout

metron-platform/metron-pcap-backend/README.md

@@ -132,6 +132,7 @@ usage: Fixed filter options
  -nr,--num_reducers <arg>        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 <arg>             IP Protocol
@@ -154,6 +155,7 @@ usage: Query filter options
  -nr,--num_reducers <arg>        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 <arg>                Query string to use as a filter
  -st,--start_time <arg>          (required) Packet start time range.
 ```

metron-platform/metron-pcap-backend/src/main/java/org/apache/metron/pcap/query/CliParser.java

@@ -55,6 +55,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;
   }
 
@@ -125,6 +126,9 @@ public void parse(CommandLine commandLine, PcapConfig config) throws java.text.P
         //no-op
       }
     }
+    if (commandLine.hasOption("print_status")) {
+      config.setPrintJobStatus(true);
+    }
   }
 
   public void printHelp(String msg, Options opts) {

metron-platform/metron-pcap-backend/src/test/java/org/apache/metron/pcap/query/PcapCliTest.java

@@ -147,7 +147,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<String, String> query = new HashMap<String, String>() {{
       put(Constants.Fields.SRC_ADDR.getName(), "192.168.1.1");
@@ -165,6 +166,7 @@ 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);
 
@@ -189,7 +191,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<String, String> query = new HashMap<String, String>() {{
       put(Constants.Fields.SRC_ADDR.getName(), "192.168.1.1");
@@ -211,6 +214,7 @@ 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);
 
@@ -262,7 +266,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";
@@ -274,6 +279,7 @@ 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);
 

metron-platform/metron-pcap/src/main/java/org/apache/metron/pcap/config/PcapConfig.java

@@ -40,6 +40,7 @@ public PcapConfig() {
   public PcapConfig(PrefixStrategy prefixStrategy) {
     this();
     setShowHelp(false);
+    setPrintJobStatus(false);
     setBasePath("");
     setBaseInterimResultPath("");
     setStartTimeMs(-1L);
@@ -73,6 +74,14 @@ public void setShowHelp(boolean showHelp) {
     this.showHelp = showHelp;
   }
 
+  public boolean printJobStatus() {
+    return PcapOptions.PRINT_JOB_STATUS.get(this, Boolean.class);
+  }
+
+  public void setPrintJobStatus(boolean printJobStatus) {
+    PcapOptions.PRINT_JOB_STATUS.put(this, printJobStatus);
+  }
+
   public String getBasePath() {
     return PcapOptions.BASE_PATH.get(this, String.class);
   }

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, Object, Path> STRING_TO_PATH =
       (s, o) -> o == null ? null : new Path(o.toString());

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<Path> 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
@@ -478,6 +485,10 @@ public Pageable<Path> get() throws JobException, InterruptedException {
     }
   }
 
+  public void monitorJob() throws IOException, InterruptedException {
+    mrJob.monitorAndPrintJob();
+  }
+
   private synchronized Pageable<Path> getFinalResults() {
     return new PcapPages(finalResults);
   }