VPCFlowLogParser { public static void parseToRecord(Map<String, Object> map, String line) throws InterceptException { Matcher matcher = FLOW_RECORD_PATTERN.matcher(line); if (matcher.matches()) { try { map.put(VERSION, Short.parseShort(matcher.group(1))); map.put(ACCOUNT_ID, matcher.group(2)); map.put(INTERFACE_ID, matcher.group(3)); if (matcher.group(14).charAt(0) == STATUS_CODE_ZERO) { map.put(SRCADDR, matcher.group(4)); map.put(DSTADDR, matcher.group(5)); map.put(SRCPORT, Integer.parseInt(matcher.group(6))); map.put(DSTPORT, Integer.parseInt(matcher.group(7))); map.put(PROTOCOL, matcher.group(8).charAt(0)); map.put(PACKETS, Integer.parseInt(matcher.group(9))); map.put(BYTES, Integer.parseInt(matcher.group(10))); } map.put(START, Integer.parseInt(matcher.group(11))); map.put(END, Integer.parseInt(matcher.group(12))); map.put(ACCEPTED, matcher.group(13).equals(ACCEPT)); map.put(LOG_STATUS, (byte) matcher.group(14).charAt(0)); } catch (Exception e) { throw new InterceptException(e); } } else { throw new InterceptException("Line doesn't match flow record:" + line); } } VPCFlowLogParser(); List<Map<String, Object>> parseFlowLogMap(String json); static void parseToRecord(Map<String, Object> map, String line); List<Event> parseFlowLogJson(EventFactory factory, String json); static void parseToRecord(Event event, String line); static VPCFlowLogRecord parseToRecord(String line);  }
@Test public void testRecordParsing() throws InterceptException { String val = "2 123456789101 eni-g123abcd 100.100.10.2 50.90.30.21 43895 443 6 13 1836 1469126675 1469126733 ACCEPT OK"; VPCFlowLogRecord record = VPCFlowLogParser.parseToRecord(val); assertNotNull(record); assertEquals((short) 2, record.getVersion()); assertEquals("123456789101", record.getAccountId()); assertEquals("eni-g123abcd", record.getInterfaceId()); assertEquals("100.100.10.2", record.getSrcAddr()); assertEquals("50.90.30.21", record.getDstAddr()); assertEquals(43895, record.getSrcPort()); assertEquals(443, record.getDstPort()); assertEquals("6".charAt(0), record.getProtocol()); assertEquals(13, record.getPackets()); assertEquals(1836, record.getBytes()); assertEquals(1469126675, record.getStartTs()); assertEquals(1469126733, record.getEndTs()); assertEquals((byte) "O".charAt(0), record.getLogStatus()); System.out.println(new Gson().toJson(record)); }
@Test public void testRecordParsingInvalid() { String val = "2 123456789101 eni-g123abcd 100.100.10.2 50.90.30.21 43895 443 6 13 1836 1469126675 1469126733 RE OK"; try { VPCFlowLogParser.parseToRecord(val); fail("Should throw an exception and not pass"); } catch (InterceptException e) { } }
@Test public void testRecordParsingSkipData() throws InterceptException { String val = "2 123456789101 eni-g123abcd - - - - - - - 1431280876 1431280934 - SKIPDATA"; VPCFlowLogRecord record = VPCFlowLogParser.parseToRecord(val); assertNotNull(record); assertEquals((short) 2, record.getVersion()); assertEquals("123456789101", record.getAccountId()); assertEquals("eni-g123abcd", record.getInterfaceId()); assertEquals(1431280876, record.getStartTs()); assertEquals(1431280934, record.getEndTs()); assertEquals((byte) "S".charAt(0), record.getLogStatus()); }
@Test public void testRecordParsingNoData() throws InterceptException { String val = "2 123456789010 eni-1a2b3c4d - - - - - - - 1431280876 1431280934 - NODATA"; VPCFlowLogRecord record = VPCFlowLogParser.parseToRecord(val); assertNotNull(record); assertEquals((short) 2, record.getVersion()); assertEquals("123456789010", record.getAccountId()); assertEquals("eni-1a2b3c4d", record.getInterfaceId()); assertEquals(1431280876, record.getStartTs()); assertEquals(1431280934, record.getEndTs()); assertEquals((byte) "N".charAt(0), record.getLogStatus()); }