PcapCli { public int run(String[] args) { if (args.length < 1) { printBasicHelp(); return -1; } String jobType = args[0]; String[] commandArgs = Arrays.copyOfRange(args, 1, args.length); Configuration hadoopConf = new Configuration(); String[] otherArgs = null; try { otherArgs = new GenericOptionsParser(hadoopConf, commandArgs).getRemainingArgs(); } catch (IOException e) { LOGGER.error("Failed to configure hadoop with provided options: {}", e.getMessage(), e); return -1; } PcapConfig commonConfig = null; Pageable<Path> results; String execDir = System.getProperty("user.dir"); if ("fixed".equals(jobType)) { FixedCliParser fixedParser = new FixedCliParser(prefixStrategy); FixedPcapConfig config = null; try { config = fixedParser.parse(otherArgs); commonConfig = config; PcapOptions.FINAL_OUTPUT_PATH.put(commonConfig, new Path(execDir)); } catch (ParseException | java.text.ParseException e) { System.err.println(e.getMessage()); System.err.flush(); fixedParser.printHelp(); return -1; } if (config.showHelp()) { fixedParser.printHelp(); return 0; } PcapOptions.FILTER_IMPL.put(commonConfig, new FixedPcapFilter.Configurator()); config.getYarnQueue().ifPresent(s -> hadoopConf.set(MRJobConfig.QUEUE_NAME, s)); PcapOptions.HADOOP_CONF.put(commonConfig, hadoopConf); try { PcapOptions.FILESYSTEM.put(commonConfig, FileSystem.get(hadoopConf)); 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; } } else if ("query".equals(jobType)) { QueryCliParser queryParser = new QueryCliParser(prefixStrategy); QueryPcapConfig config = null; try { config = queryParser.parse(otherArgs); commonConfig = config; PcapOptions.FINAL_OUTPUT_PATH.put(commonConfig, new Path(execDir)); } catch (ParseException | java.text.ParseException e) { System.err.println(e.getMessage()); queryParser.printHelp(); return -1; } if (config.showHelp()) { queryParser.printHelp(); return 0; } PcapOptions.FILTER_IMPL.put(commonConfig, new FixedPcapFilter.Configurator()); config.getYarnQueue().ifPresent(s -> hadoopConf.set(MRJobConfig.QUEUE_NAME, s)); PcapOptions.HADOOP_CONF.put(commonConfig, hadoopConf); try { PcapOptions.FILESYSTEM.put(commonConfig, FileSystem.get(hadoopConf)); 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; } } else { printBasicHelp(); return -1; } return 0; } PcapCli(PcapJob jobRunner, PcapConfig.PrefixStrategy prefixStrategy); static void main(String[] args); int run(String[] args); void printBasicHelp(); static final PcapConfig.PrefixStrategy PREFIX_STRATEGY; }
@Test public void runs_fixed_pcap_filter_job_with_default_argument_list() throws Exception { String[] args = { "fixed", "-start_time", "500", "-ip_src_addr", "192.168.1.1", "-ip_dst_addr", "192.168.1.2", "-ip_src_port", "8081", "-ip_dst_port", "8082", "-protocol", "6", "-packet_filter", "`casey`" }; HashMap<String, String> query = new HashMap<String, String>() {{ put(Constants.Fields.SRC_ADDR.getName(), "192.168.1.1"); put(Constants.Fields.DST_ADDR.getName(), "192.168.1.2"); put(Constants.Fields.SRC_PORT.getName(), "8081"); put(Constants.Fields.DST_PORT.getName(), "8082"); put(Constants.Fields.PROTOCOL.getName(), "6"); put(Constants.Fields.INCLUDES_REVERSE_TRAFFIC.getName(), "false"); put(PcapHelper.PacketFields.PACKET_FILTER.getName(), "`casey`"); }}; FixedPcapConfig config = new FixedPcapConfig(prefixStrategy); PcapOptions.BASE_PATH.put(config, BASE_INPUT_PATH_DEFAULT); PcapOptions.BASE_INTERIM_RESULT_PATH.put(config, BASE_INTERIM_RESULT_PATH_DEFAULT); PcapOptions.FIELDS.put(config, query); PcapOptions.NUM_REDUCERS.put(config, 10); PcapOptions.START_TIME_MS.put(config, 500L); 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).get(); }
@Test public void runs_fixed_pcap_filter_job_with_full_argument_list_and_default_dateformat() throws Exception { String[] args = { "fixed", "-start_time", "500", "-end_time", "1000", "-base_path", "/base/path", "-base_output_path", "/base/output/path", "-ip_src_addr", "192.168.1.1", "-ip_dst_addr", "192.168.1.2", "-ip_src_port", "8081", "-ip_dst_port", "8082", "-protocol", "6", "-include_reverse", "-num_reducers", "10", "-records_per_file", "1000", "-finalizer_threads", "10" }; Map<String, String> query = new HashMap<String, String>() {{ put(Constants.Fields.SRC_ADDR.getName(), "192.168.1.1"); put(Constants.Fields.DST_ADDR.getName(), "192.168.1.2"); put(Constants.Fields.SRC_PORT.getName(), "8081"); put(Constants.Fields.DST_PORT.getName(), "8082"); put(Constants.Fields.PROTOCOL.getName(), "6"); put(Constants.Fields.INCLUDES_REVERSE_TRAFFIC.getName(), "true"); }}; FixedPcapConfig config = new FixedPcapConfig(prefixStrategy); PcapOptions.BASE_PATH.put(config, "/base/path"); PcapOptions.BASE_INTERIM_RESULT_PATH.put(config, "/base/output/path"); PcapOptions.FIELDS.put(config, query); PcapOptions.NUM_REDUCERS.put(config, 10); 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); PcapOptions.FINALIZER_THREADPOOL_SIZE.put(config, "10"); 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).get(); }
@Test public void runs_fixed_pcap_filter_job_with_full_argument_list() throws Exception { String[] args = { "fixed", "-start_time", "2016-06-13-18:35.00", "-end_time", "2016-06-15-18:35.00", "-date_format", "yyyy-MM-dd-HH:mm.ss", "-base_path", "/base/path", "-base_output_path", "/base/output/path", "-ip_src_addr", "192.168.1.1", "-ip_dst_addr", "192.168.1.2", "-ip_src_port", "8081", "-ip_dst_port", "8082", "-protocol", "6", "-include_reverse", "-num_reducers", "10", "-records_per_file", "1000", "-yq", "pcap", "-finalizer_threads", "10" }; Map<String, String> query = new HashMap<String, String>() {{ put(Constants.Fields.SRC_ADDR.getName(), "192.168.1.1"); put(Constants.Fields.DST_ADDR.getName(), "192.168.1.2"); put(Constants.Fields.SRC_PORT.getName(), "8081"); put(Constants.Fields.DST_PORT.getName(), "8082"); put(Constants.Fields.PROTOCOL.getName(), "6"); put(Constants.Fields.INCLUDES_REVERSE_TRAFFIC.getName(), "true"); }}; long startAsNanos = asNanos("2016-06-13-18:35.00", "yyyy-MM-dd-HH:mm.ss"); long endAsNanos = asNanos("2016-06-15-18:35.00", "yyyy-MM-dd-HH:mm.ss"); FixedPcapConfig config = new FixedPcapConfig(prefixStrategy); PcapOptions.BASE_PATH.put(config, "/base/path"); PcapOptions.BASE_INTERIM_RESULT_PATH.put(config, "/base/output/path"); PcapOptions.FIELDS.put(config, query); PcapOptions.NUM_REDUCERS.put(config, 10); PcapOptions.START_TIME_MS.put(config, startAsNanos / 1000000L); PcapOptions.END_TIME_MS.put(config, endAsNanos / 1000000L); PcapOptions.NUM_RECORDS_PER_FILE.put(config, 1000); PcapOptions.PRINT_JOB_STATUS.put(config, true); PcapOptions.HADOOP_CONF.put(config, new HashMap<String, Object>() {{ put(MRJobConfig.QUEUE_NAME, "pcap"); }}); PcapOptions.FINALIZER_THREADPOOL_SIZE.put(config, "10"); 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).get(); }
@Test public void runs_query_pcap_filter_job_with_default_argument_list() throws Exception { String[] args = { "query", "-start_time", "500", "-query", "some query string" }; String query = "some query string"; FixedPcapConfig config = new FixedPcapConfig(prefixStrategy); PcapOptions.BASE_PATH.put(config, BASE_INPUT_PATH_DEFAULT); PcapOptions.BASE_INTERIM_RESULT_PATH.put(config, BASE_INTERIM_RESULT_PATH_DEFAULT); PcapOptions.FIELDS.put(config, query); PcapOptions.NUM_REDUCERS.put(config, 10); PcapOptions.START_TIME_MS.put(config, 500L); PcapOptions.FINALIZER_THREADPOOL_SIZE.put(config, "1"); 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).get(); }
@Test public void runs_query_pcap_filter_job_with_full_argument_list() throws Exception { String[] args = { "query", "-start_time", "500", "-end_time", "1000", "-num_reducers", "10", "-base_path", "/base/path", "-base_output_path", "/base/output/path", "-query", "some query string", "-records_per_file", "1000", "-finalizer_threads", "10" }; String query = "some query string"; FixedPcapConfig config = new FixedPcapConfig(prefixStrategy); PcapOptions.BASE_PATH.put(config, "/base/path"); PcapOptions.BASE_INTERIM_RESULT_PATH.put(config, "/base/output/path"); PcapOptions.FIELDS.put(config, query); PcapOptions.NUM_REDUCERS.put(config, 10); 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); PcapOptions.FINALIZER_THREADPOOL_SIZE.put(config, "10"); 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).get(); }