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