@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()); }
@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 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 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); }
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); }
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);  }