@Test public void testLogPropertyWithIgnoreRegex() { final LogAttribute logAttribute = new LogAttribute(); final TestRunner runner = TestRunners.newTestRunner(logAttribute); final ProcessContext context = runner.getProcessContext(); final ProcessSession session = runner.getProcessSessionFactory().createSession(); final MockComponentLog LOG = runner.getLogger(); runner.setProperty(LogAttribute.ATTRIBUTES_TO_IGNORE_REGEX, "foo.*"); final Map<String,String> attrs = Maps.newHashMap(); attrs.put("foo", "foo-value"); attrs.put("bar", "bar-value"); attrs.put("foobaz", "foobaz-value"); final MockFlowFile flowFile = runner.enqueue("content", attrs); final String logMessage = logAttribute.processFlowFile(LOG, LogAttribute.DebugLevels.info, flowFile, session, context); assertThat(logMessage, not(containsString("foobaz-value"))); assertThat(logMessage, not(containsString("foo-value"))); assertThat(logMessage, containsString("bar-value")); }
protected String processFlowFile(final ComponentLog logger, final DebugLevels logLevel, final FlowFile flowFile, final ProcessSession session, final ProcessContext context) { final Set<String> attributeKeys = getAttributesToLog(flowFile.getAttributes().keySet(), context); final ComponentLog LOG = getLogger(); final String dashedLine; String logPrefix = context.getProperty(LOG_PREFIX).evaluateAttributeExpressions(flowFile).getValue(); Charset charset = Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions(flowFile).getValue()); if (StringUtil.isBlank(logPrefix)) { dashedLine = StringUtils.repeat('-', 50); } else { logPrefix = StringUtils.abbreviate(logPrefix, 40); logPrefix = StringUtils.center(logPrefix, 40, '-'); dashedLine = StringUtils.repeat('-', 5) + logPrefix + StringUtils.repeat('-', 5); } final StringBuilder message = new StringBuilder(); message.append("logging for flow file ").append(flowFile); message.append("\n"); message.append(dashedLine); message.append("\nStandard FlowFile Attributes"); message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", "entryDate", new Date(flowFile.getEntryDate()))); message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", "lineageStartDate", new Date(flowFile.getLineageStartDate()))); message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", "fileSize", flowFile.getSize())); message.append("\nFlowFile Attribute Map Content"); for (final String key : attributeKeys) { message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", key, flowFile.getAttribute(key))); } message.append("\n"); message.append(dashedLine); final boolean logPayload = context.getProperty(LOG_PAYLOAD).asBoolean(); if (logPayload) { message.append("\n"); if (flowFile.getSize() < ONE_MB) { final FlowFilePayloadCallback callback = new FlowFilePayloadCallback(charset); session.read(flowFile, callback); message.append(callback.getContents()); } else { message.append("\n Not including payload since it is larger than one mb."); } } final String outputMessage = message.toString().trim(); switch (logLevel) { case info: LOG.info(outputMessage); break; case debug: LOG.debug(outputMessage); break; case warn: LOG.warn(outputMessage); break; case trace: LOG.trace(outputMessage); break; case error: LOG.error(outputMessage); break; default: LOG.debug(outputMessage); } return outputMessage; }
LogAttribute extends AbstractProcessor { protected String processFlowFile(final ComponentLog logger, final DebugLevels logLevel, final FlowFile flowFile, final ProcessSession session, final ProcessContext context) { final Set<String> attributeKeys = getAttributesToLog(flowFile.getAttributes().keySet(), context); final ComponentLog LOG = getLogger(); final String dashedLine; String logPrefix = context.getProperty(LOG_PREFIX).evaluateAttributeExpressions(flowFile).getValue(); Charset charset = Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions(flowFile).getValue()); if (StringUtil.isBlank(logPrefix)) { dashedLine = StringUtils.repeat('-', 50); } else { logPrefix = StringUtils.abbreviate(logPrefix, 40); logPrefix = StringUtils.center(logPrefix, 40, '-'); dashedLine = StringUtils.repeat('-', 5) + logPrefix + StringUtils.repeat('-', 5); } final StringBuilder message = new StringBuilder(); message.append("logging for flow file ").append(flowFile); message.append("\n"); message.append(dashedLine); message.append("\nStandard FlowFile Attributes"); message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", "entryDate", new Date(flowFile.getEntryDate()))); message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", "lineageStartDate", new Date(flowFile.getLineageStartDate()))); message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", "fileSize", flowFile.getSize())); message.append("\nFlowFile Attribute Map Content"); for (final String key : attributeKeys) { message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", key, flowFile.getAttribute(key))); } message.append("\n"); message.append(dashedLine); final boolean logPayload = context.getProperty(LOG_PAYLOAD).asBoolean(); if (logPayload) { message.append("\n"); if (flowFile.getSize() < ONE_MB) { final FlowFilePayloadCallback callback = new FlowFilePayloadCallback(charset); session.read(flowFile, callback); message.append(callback.getContents()); } else { message.append("\n Not including payload since it is larger than one mb."); } } final String outputMessage = message.toString().trim(); switch (logLevel) { case info: LOG.info(outputMessage); break; case debug: LOG.debug(outputMessage); break; case warn: LOG.warn(outputMessage); break; case trace: LOG.trace(outputMessage); break; case error: LOG.error(outputMessage); break; default: LOG.debug(outputMessage); } return outputMessage; } }
LogAttribute extends AbstractProcessor { protected String processFlowFile(final ComponentLog logger, final DebugLevels logLevel, final FlowFile flowFile, final ProcessSession session, final ProcessContext context) { final Set<String> attributeKeys = getAttributesToLog(flowFile.getAttributes().keySet(), context); final ComponentLog LOG = getLogger(); final String dashedLine; String logPrefix = context.getProperty(LOG_PREFIX).evaluateAttributeExpressions(flowFile).getValue(); Charset charset = Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions(flowFile).getValue()); if (StringUtil.isBlank(logPrefix)) { dashedLine = StringUtils.repeat('-', 50); } else { logPrefix = StringUtils.abbreviate(logPrefix, 40); logPrefix = StringUtils.center(logPrefix, 40, '-'); dashedLine = StringUtils.repeat('-', 5) + logPrefix + StringUtils.repeat('-', 5); } final StringBuilder message = new StringBuilder(); message.append("logging for flow file ").append(flowFile); message.append("\n"); message.append(dashedLine); message.append("\nStandard FlowFile Attributes"); message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", "entryDate", new Date(flowFile.getEntryDate()))); message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", "lineageStartDate", new Date(flowFile.getLineageStartDate()))); message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", "fileSize", flowFile.getSize())); message.append("\nFlowFile Attribute Map Content"); for (final String key : attributeKeys) { message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", key, flowFile.getAttribute(key))); } message.append("\n"); message.append(dashedLine); final boolean logPayload = context.getProperty(LOG_PAYLOAD).asBoolean(); if (logPayload) { message.append("\n"); if (flowFile.getSize() < ONE_MB) { final FlowFilePayloadCallback callback = new FlowFilePayloadCallback(charset); session.read(flowFile, callback); message.append(callback.getContents()); } else { message.append("\n Not including payload since it is larger than one mb."); } } final String outputMessage = message.toString().trim(); switch (logLevel) { case info: LOG.info(outputMessage); break; case debug: LOG.debug(outputMessage); break; case warn: LOG.warn(outputMessage); break; case trace: LOG.trace(outputMessage); break; case error: LOG.error(outputMessage); break; default: LOG.debug(outputMessage); } return outputMessage; }  }
LogAttribute extends AbstractProcessor { protected String processFlowFile(final ComponentLog logger, final DebugLevels logLevel, final FlowFile flowFile, final ProcessSession session, final ProcessContext context) { final Set<String> attributeKeys = getAttributesToLog(flowFile.getAttributes().keySet(), context); final ComponentLog LOG = getLogger(); final String dashedLine; String logPrefix = context.getProperty(LOG_PREFIX).evaluateAttributeExpressions(flowFile).getValue(); Charset charset = Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions(flowFile).getValue()); if (StringUtil.isBlank(logPrefix)) { dashedLine = StringUtils.repeat('-', 50); } else { logPrefix = StringUtils.abbreviate(logPrefix, 40); logPrefix = StringUtils.center(logPrefix, 40, '-'); dashedLine = StringUtils.repeat('-', 5) + logPrefix + StringUtils.repeat('-', 5); } final StringBuilder message = new StringBuilder(); message.append("logging for flow file ").append(flowFile); message.append("\n"); message.append(dashedLine); message.append("\nStandard FlowFile Attributes"); message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", "entryDate", new Date(flowFile.getEntryDate()))); message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", "lineageStartDate", new Date(flowFile.getLineageStartDate()))); message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", "fileSize", flowFile.getSize())); message.append("\nFlowFile Attribute Map Content"); for (final String key : attributeKeys) { message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", key, flowFile.getAttribute(key))); } message.append("\n"); message.append(dashedLine); final boolean logPayload = context.getProperty(LOG_PAYLOAD).asBoolean(); if (logPayload) { message.append("\n"); if (flowFile.getSize() < ONE_MB) { final FlowFilePayloadCallback callback = new FlowFilePayloadCallback(charset); session.read(flowFile, callback); message.append(callback.getContents()); } else { message.append("\n Not including payload since it is larger than one mb."); } } final String outputMessage = message.toString().trim(); switch (logLevel) { case info: LOG.info(outputMessage); break; case debug: LOG.debug(outputMessage); break; case warn: LOG.warn(outputMessage); break; case trace: LOG.trace(outputMessage); break; case error: LOG.error(outputMessage); break; default: LOG.debug(outputMessage); } return outputMessage; }  @Override Set<Relationship> getRelationships(); @Override void onTrigger(final ProcessContext context, final ProcessSession session); }
LogAttribute extends AbstractProcessor { protected String processFlowFile(final ComponentLog logger, final DebugLevels logLevel, final FlowFile flowFile, final ProcessSession session, final ProcessContext context) { final Set<String> attributeKeys = getAttributesToLog(flowFile.getAttributes().keySet(), context); final ComponentLog LOG = getLogger(); final String dashedLine; String logPrefix = context.getProperty(LOG_PREFIX).evaluateAttributeExpressions(flowFile).getValue(); Charset charset = Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions(flowFile).getValue()); if (StringUtil.isBlank(logPrefix)) { dashedLine = StringUtils.repeat('-', 50); } else { logPrefix = StringUtils.abbreviate(logPrefix, 40); logPrefix = StringUtils.center(logPrefix, 40, '-'); dashedLine = StringUtils.repeat('-', 5) + logPrefix + StringUtils.repeat('-', 5); } final StringBuilder message = new StringBuilder(); message.append("logging for flow file ").append(flowFile); message.append("\n"); message.append(dashedLine); message.append("\nStandard FlowFile Attributes"); message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", "entryDate", new Date(flowFile.getEntryDate()))); message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", "lineageStartDate", new Date(flowFile.getLineageStartDate()))); message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", "fileSize", flowFile.getSize())); message.append("\nFlowFile Attribute Map Content"); for (final String key : attributeKeys) { message.append(String.format("\nKey: '%1$s'\n\tValue: '%2$s'", key, flowFile.getAttribute(key))); } message.append("\n"); message.append(dashedLine); final boolean logPayload = context.getProperty(LOG_PAYLOAD).asBoolean(); if (logPayload) { message.append("\n"); if (flowFile.getSize() < ONE_MB) { final FlowFilePayloadCallback callback = new FlowFilePayloadCallback(charset); session.read(flowFile, callback); message.append(callback.getContents()); } else { message.append("\n Not including payload since it is larger than one mb."); } } final String outputMessage = message.toString().trim(); switch (logLevel) { case info: LOG.info(outputMessage); break; case debug: LOG.debug(outputMessage); break; case warn: LOG.warn(outputMessage); break; case trace: LOG.trace(outputMessage); break; case error: LOG.error(outputMessage); break; default: LOG.debug(outputMessage); } return outputMessage; }  @Override Set<Relationship> getRelationships(); @Override void onTrigger(final ProcessContext context, final ProcessSession session); static final PropertyDescriptor LOG_LEVEL; static final PropertyDescriptor ATTRIBUTES_TO_LOG_CSV; static final PropertyDescriptor ATTRIBUTES_TO_LOG_REGEX; static final PropertyDescriptor ATTRIBUTES_TO_IGNORE_CSV; static final PropertyDescriptor ATTRIBUTES_TO_IGNORE_REGEX; static final PropertyDescriptor LOG_PAYLOAD; static final PropertyDescriptor LOG_PREFIX; static final PropertyDescriptor CHARSET; static final String FIFTY_DASHES; static final long ONE_MB; static final Relationship REL_SUCCESS; }