LogParserTransform extends Transform<StructuredRecord, StructuredRecord> { @Override public void transform(StructuredRecord input, Emitter<StructuredRecord> emitter) throws Exception { String log = getLog(input); if (log == null) { LOG.debug("Couldn't read schema, log message was null"); return; } StructuredRecord output; if (S3_LOG.equals(config.logFormat)) { Matcher logMatcher = S3_LOG_PATTERN.matcher(log); if (!logMatcher.matches() || logMatcher.groupCount() < S3_REGEX_LENGTH) { LOG.debug("Couldn't parse log because log did not match the S3 format, log: {}", log); emitter.emitError(new InvalidEntry<>(31, "Couldn't parse log, because the log did not match the S3 format.", input)); return; } output = parseRequest(logMatcher, S3_INDICES); } else if (CLF_LOG.equals(config.logFormat)) { Matcher logMatcher = CLF_LOG_PATTERN.matcher(log); if (!logMatcher.matches() || logMatcher.groupCount() < CLF_REGEX_LENGTH) { LOG.debug("Couldn't parse log because the log did not match the CLF format. log: {}", log); emitter.emitError(new InvalidEntry<>(31, "Couldn't parse log, because the log did not match the CLF format.", input)); return; } output = parseRequest(logMatcher, CLF_INDICES); } else { if (log.startsWith("#")) { LOG.trace("Log is a comment. Ignoring..."); return; } String[] fields = log.split("\\t"); String uri = fields[7]; String ip = fields[4]; long ts = sdfCloudfront.parse(String.format("%s:%s UTC", fields[0], fields[1])).getTime(); UserAgentStringParser parser = UADetectorServiceFactory.getResourceModuleParser(); ReadableUserAgent userAgent = parser.parse(fields[10]); String browser = userAgent.getFamily().getName(); String device = userAgent.getDeviceCategory().getCategory().getName(); int httpStatus = Integer.parseInt(fields[8]); output = StructuredRecord.builder(LOG_SCHEMA) .set("uri", uri) .set("ip", ip) .set("browser", browser) .set("device", device) .set("httpStatus", httpStatus) .set("ts", ts) .build(); } if (output != null) { emitter.emit(output); } } LogParserTransform(LogParserConfig config); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void prepareRun(StageSubmitterContext context); @Override void transform(StructuredRecord input, Emitter<StructuredRecord> emitter);  }
@Test public void testS3LogTransform() throws Exception { StructuredRecord botRecord = StructuredRecord.builder(STRING_SCHEMA) .set("body", "13a9f69e4a00effd6b4b891dcbcabef632ef9a9da7c localhost " + "[22/Jan/2015:11:03:21 +0000] 122.122.111.11 - 6006CA0AE4 REST.GET.OBJECT " + "ubuntu/this/is/some/folder " + "\"GET /my/uri.gif releases/dists/precise/releases/i18n/Translation-en HTTP/1.1\" " + "403 AccessDenied 231 - 10 - \"-\" \"Debian APT-HTTP/1.3 (0.8.16~exp12ubuntu10. 17)\" -") .build(); StructuredRecord browserRecord = StructuredRecord.builder(STRING_SCHEMA) .set("body", "13a9f69e4a00effd6b4b891dcef632ef9afe38cc8b0 localhost " + "[31/Jan/2015:21:57:57 +0000] 133.133.133.133 - 0E94306589 REST.GET.OBJECT " + "downloads/this/is/another/folder/with/a/file/file.zip " + "\"GET /my/uri.jpg HTTP/1.1\" 304 - - 195750039 198 - " + "\"-\" \"Mozilla/5.0 Gecko/20100115 Firefox/3.6\" -") .build(); MockEmitter<StructuredRecord> emitter = new MockEmitter<>(); S3_TRANSFORM.transform(botRecord, emitter); StructuredRecord output = emitter.getEmitted().get(0); Assert.assertEquals("/my/uri.gif", output.get("uri")); Assert.assertEquals("122.122.111.11", output.get("ip")); Assert.assertEquals("unknown", output.get("browser")); Assert.assertEquals("", output.get("device")); Assert.assertEquals(403, output.<Integer>get("httpStatus").intValue()); Assert.assertEquals(1421924601000L, output.<Long>get("ts").longValue()); S3_TRANSFORM.transform(browserRecord, emitter); output = emitter.getEmitted().get(1); Assert.assertEquals("/my/uri.jpg", output.get("uri")); Assert.assertEquals("133.133.133.133", output.get("ip")); Assert.assertEquals("Firefox", output.get("browser")); Assert.assertEquals("Personal computer", output.get("device")); Assert.assertEquals(304, output.<Integer>get("httpStatus").intValue()); Assert.assertEquals(1422741477000L, output.<Long>get("ts").longValue()); }
@Test public void testCloudfrontLogTransform() throws Exception { String event = "2015-04-17\t13:35:48\tSFO20\t582123\t11.111.111.11\tGET\texample.cloudfront" + ".net\t/coopr-standalone-vm/0.9.8/coopr-standalone-vm-0.9.8.ova\t200\t-\tMozilla/5" + ".0%2520(compatible;%2520Yahoo!%2520Slurp;%2520http: "\t-\tError\tsCmB94WPP5v-QoCyn7Jz1ZLn0kBhzIEkqfFuX2Gh5oA1SA8dsLp-kw==\texample.co\thttp\t264\t0.984"; StructuredRecord record = StructuredRecord.builder(STRING_SCHEMA) .set("body", event) .build(); String comment = "#Fields: date time x-edge-location sc-bytes c-ip cs-method cs(Host) cs-uri-stem sc-status " + "cs(Referer) cs(User-Agent) cs-uri-query cs(Cookie) x-edge-result-type x-edge-request-id x-host-header " + "cs-protocol cs-bytes time-taken"; StructuredRecord commentRecord = StructuredRecord.builder(STRING_SCHEMA) .set("body", comment) .build(); MockEmitter<StructuredRecord> emitter = new MockEmitter<>(); CLOUDFRONT_TRANSFORM.transform(record, emitter); StructuredRecord output = emitter.getEmitted().get(0); Assert.assertEquals("/coopr-standalone-vm/0.9.8/coopr-standalone-vm-0.9.8.ova", output.get("uri")); Assert.assertEquals("11.111.111.11", output.get("ip")); Assert.assertEquals("unknown", output.get("browser")); Assert.assertEquals("", output.get("device")); Assert.assertEquals(200, output.<Integer>get("httpStatus").intValue()); Assert.assertEquals(1429277748000L, output.<Long>get("ts").longValue()); CLOUDFRONT_TRANSFORM.transform(commentRecord, emitter); Assert.assertEquals(1, emitter.getEmitted().size()); }
@Test public void testCLFLogTransform() throws Exception { StructuredRecord record = StructuredRecord.builder(BYTE_SCHEMA) .set("body", ByteBuffer.wrap((Bytes.toBytes("127.0.0.1 - frank [10/Oct/2000:13:55:36 -0700] " + "\"GET /apache_pb.gif HTTP/1.0\" 200 2326 " + "\"http: "(Macintosh; Intel Mac OS X 10_10_3) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/43.0.2357.124 Safari/537.36\"")))) .build(); MockEmitter<StructuredRecord> emitter = new MockEmitter<>(); CLF_TRANSFORM.transform(record, emitter); StructuredRecord output = emitter.getEmitted().get(0); Assert.assertEquals("/apache_pb.gif", output.get("uri")); Assert.assertEquals("127.0.0.1", output.get("ip")); Assert.assertEquals("Chrome", output.get("browser")); Assert.assertEquals("Personal computer", output.get("device")); Assert.assertEquals(200, output.<Integer>get("httpStatus").intValue()); Assert.assertEquals(971211336000L, output.<Long>get("ts").longValue()); }
@Test public void testErrorDatasetForInvalidCLFLog() throws Exception { StructuredRecord record = StructuredRecord.builder(BYTE_SCHEMA) .set("body", ByteBuffer.wrap((Bytes.toBytes("127.0.0.1 - frank [10/Oct/2000:13:55:36 -0700] " + "\"GET /apache_pb.gif HTTP/1.0\" 200 2326 " + "\"http: "")))).build(); MockEmitter<StructuredRecord> emitter = new MockEmitter<>(); CLF_TRANSFORM.transform(record, emitter); Assert.assertEquals(0, emitter.getEmitted().size()); Assert.assertEquals(1, emitter.getErrors().size()); InvalidEntry<StructuredRecord> invalidEntry = emitter.getErrors().get(0); Assert.assertEquals(31, invalidEntry.getErrorCode()); Assert.assertEquals("Error Message", "Couldn't parse log, because the log did not match the CLF format.", invalidEntry.getErrorMsg()); Assert.assertEquals("Error Record", record, invalidEntry.getInvalidRecord()); }
@Test public void testErrorDatasetForInvalidS3Log() throws Exception { StructuredRecord botRecord = StructuredRecord.builder(STRING_SCHEMA) .set("body", "13a9f69e4a00effd6b4b891dcbcabef632ef9a9da7c - 6006CA0AE4 REST.GET.OBJECT " + "ubuntu/this/is/some/folder " + "\"GET /my/uri.gif releases/dists/precise/releases/i18n/Translation-en HTTP/1.1\" " + "403 AccessDenied 231 - 10 - \"-\" \"Debian APT-HTTP/1.3 (0.8.16~exp12ubuntu10. 17)\" -") .build(); MockEmitter<StructuredRecord> emitter = new MockEmitter<>(); S3_TRANSFORM.transform(botRecord, emitter); Assert.assertEquals(0, emitter.getEmitted().size()); Assert.assertEquals(1, emitter.getErrors().size()); InvalidEntry<StructuredRecord> invalidEntry = emitter.getErrors().get(0); Assert.assertEquals(31, invalidEntry.getErrorCode()); Assert.assertEquals("Error Message", "Couldn't parse log, because the log did not match the S3 format.", invalidEntry.getErrorMsg()); Assert.assertEquals("Error Record", botRecord, invalidEntry.getInvalidRecord()); }