Repository: nifi Updated Branches: refs/heads/master aa947e4d3 -> 56ad14ad1
NIFI-4977 Adding expression language support to the Sender properties of PutSyslog This closes #2547. Signed-off-by: Bryan Bende <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/56ad14ad Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/56ad14ad Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/56ad14ad Branch: refs/heads/master Commit: 56ad14ad1eb5bc1d716f6d2664618a07fd049e07 Parents: aa947e4 Author: Joe Percivall <[email protected]> Authored: Wed Mar 14 20:39:04 2018 -0400 Committer: Bryan Bende <[email protected]> Committed: Mon Mar 26 15:49:52 2018 -0400 ---------------------------------------------------------------------- .../standard/AbstractSyslogProcessor.java | 9 ++-- .../nifi/processors/standard/ListenSyslog.java | 52 ++++++++++---------- .../nifi/processors/standard/PutSyslog.java | 31 +++++++----- .../processors/standard/TestListenSyslog.java | 2 +- .../nifi/processors/standard/TestPutSyslog.java | 47 ++++++++++++++++++ 5 files changed, 99 insertions(+), 42 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/56ad14ad/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractSyslogProcessor.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractSyslogProcessor.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractSyslogProcessor.java index a8a6bef..f7319f2 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractSyslogProcessor.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractSyslogProcessor.java @@ -38,23 +38,26 @@ public abstract class AbstractSyslogProcessor extends AbstractProcessor { .build(); public static final PropertyDescriptor PORT = new PropertyDescriptor .Builder().name("Port") - .description("The port for Syslog communication.") + .description("The port for Syslog communication. Note that Expression language is not evaluated per FlowFile.") .required(true) .addValidator(StandardValidators.PORT_VALIDATOR) + .expressionLanguageSupported(true) .build(); public static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder() .name("Character Set") - .description("Specifies the character set of the Syslog messages") + .description("Specifies the character set of the Syslog messages. Note that Expression language is not evaluated per FlowFile.") .required(true) .defaultValue("UTF-8") .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR) + .expressionLanguageSupported(true) .build(); public static final PropertyDescriptor TIMEOUT = new PropertyDescriptor.Builder() .name("Timeout") - .description("The timeout for connecting to and communicating with the syslog server. Does not apply to UDP") + .description("The timeout for connecting to and communicating with the syslog server. Does not apply to UDP. Note that Expression language is not evaluated per FlowFile.") .required(false) .defaultValue("10 seconds") .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .expressionLanguageSupported(true) .build(); http://git-wip-us.apache.org/repos/asf/nifi/blob/56ad14ad/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenSyslog.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenSyslog.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenSyslog.java index d841e15..fb239ee 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenSyslog.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenSyslog.java @@ -16,28 +16,6 @@ */ package org.apache.nifi.processors.standard; -import static org.apache.nifi.processor.util.listen.ListenerProperties.NETWORK_INTF_NAME; - -import java.io.IOException; -import java.io.OutputStream; -import java.net.InetAddress; -import java.net.NetworkInterface; -import java.nio.ByteBuffer; -import java.nio.channels.SelectableChannel; -import java.nio.channels.SocketChannel; -import java.nio.charset.Charset; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; -import java.util.HashMap; -import java.util.HashSet; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.TimeUnit; -import javax.net.ssl.SSLContext; import org.apache.commons.lang3.StringUtils; import org.apache.nifi.annotation.behavior.InputRequirement; import org.apache.nifi.annotation.behavior.SupportsBatching; @@ -77,6 +55,29 @@ import org.apache.nifi.security.util.SslContextFactory; import org.apache.nifi.ssl.RestrictedSSLContextService; import org.apache.nifi.ssl.SSLContextService; +import javax.net.ssl.SSLContext; +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.NetworkInterface; +import java.nio.ByteBuffer; +import java.nio.channels.SelectableChannel; +import java.nio.channels.SocketChannel; +import java.nio.charset.Charset; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; + +import static org.apache.nifi.processor.util.listen.ListenerProperties.NETWORK_INTF_NAME; + @SupportsBatching @InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN) @Tags({"syslog", "listen", "udp", "tcp", "logs"}) @@ -114,6 +115,7 @@ public class ListenSyslog extends AbstractSyslogProcessor { .defaultValue("10000") .required(true) .build(); + public static final PropertyDescriptor RECV_BUFFER_SIZE = new PropertyDescriptor.Builder() .name("Receive Buffer Size") .displayName("Receive Buffer Size") @@ -285,13 +287,13 @@ public class ListenSyslog extends AbstractSyslogProcessor { @OnScheduled public void onScheduled(final ProcessContext context) throws IOException { - final int port = context.getProperty(PORT).asInteger(); + final int port = context.getProperty(PORT).evaluateAttributeExpressions().asInteger(); final int bufferSize = context.getProperty(RECV_BUFFER_SIZE).asDataSize(DataUnit.B).intValue(); final int maxChannelBufferSize = context.getProperty(MAX_SOCKET_BUFFER_SIZE).asDataSize(DataUnit.B).intValue(); final int maxMessageQueueSize = context.getProperty(MAX_MESSAGE_QUEUE_SIZE).asInteger(); final String protocol = context.getProperty(PROTOCOL).getValue(); final String nicIPAddressStr = context.getProperty(NETWORK_INTF_NAME).evaluateAttributeExpressions().getValue(); - final String charSet = context.getProperty(CHARSET).getValue(); + final String charSet = context.getProperty(CHARSET).evaluateAttributeExpressions().getValue(); final String msgDemarcator = context.getProperty(MESSAGE_DELIMITER).getValue().replace("\\n", "\n").replace("\\r", "\r").replace("\\t", "\t"); messageDemarcatorBytes = msgDemarcator.getBytes(Charset.forName(charSet)); @@ -413,7 +415,7 @@ public class ListenSyslog extends AbstractSyslogProcessor { final int maxBatchSize = context.getProperty(MAX_BATCH_SIZE).asInteger(); - final String port = context.getProperty(PORT).getValue(); + final String port = context.getProperty(PORT).evaluateAttributeExpressions().getValue(); final String protocol = context.getProperty(PROTOCOL).getValue(); final Map<String, String> defaultAttributes = new HashMap<>(4); http://git-wip-us.apache.org/repos/asf/nifi/blob/56ad14ad/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSyslog.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSyslog.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSyslog.java index 63f17ba..81b016f 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSyslog.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSyslog.java @@ -73,33 +73,38 @@ public class PutSyslog extends AbstractSyslogProcessor { public static final PropertyDescriptor HOSTNAME = new PropertyDescriptor.Builder() .name("Hostname") - .description("The ip address or hostname of the Syslog server.") + .description("The ip address or hostname of the Syslog server. Note that Expression language is not evaluated per FlowFile.") .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) .defaultValue("localhost") .required(true) + .expressionLanguageSupported(true) .build(); + public static final PropertyDescriptor MAX_SOCKET_SEND_BUFFER_SIZE = new PropertyDescriptor.Builder() .name("Max Size of Socket Send Buffer") .description("The maximum size of the socket send buffer that should be used. This is a suggestion to the Operating System " + "to indicate how big the socket buffer should be. If this value is set too low, the buffer may fill up before " + - "the data can be read, and incoming data will be dropped.") + "the data can be read, and incoming data will be dropped. Note that Expression language is not evaluated per FlowFile.") .addValidator(StandardValidators.DATA_SIZE_VALIDATOR) .defaultValue("1 MB") .required(true) + .expressionLanguageSupported(true) .build(); public static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor .Builder().name("Batch Size") - .description("The number of incoming FlowFiles to process in a single execution of this processor.") + .description("The number of incoming FlowFiles to process in a single execution of this processor. Note that Expression language is not evaluated per FlowFile.") .required(true) .defaultValue("25") .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR) + .expressionLanguageSupported(true) .build(); public static final PropertyDescriptor IDLE_EXPIRATION = new PropertyDescriptor .Builder().name("Idle Connection Expiration") - .description("The amount of time a connection should be held open without being used before closing the connection.") + .description("The amount of time a connection should be held open without being used before closing the connection. Note that Expression language is not evaluated per FlowFile.") .required(true) .defaultValue("5 seconds") .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .expressionLanguageSupported(true) .build(); public static final PropertyDescriptor MSG_PRIORITY = new PropertyDescriptor .Builder().name("Message Priority") @@ -223,11 +228,11 @@ public class PutSyslog extends AbstractSyslogProcessor { } protected ChannelSender createSender(final ProcessContext context) throws IOException { - final int port = context.getProperty(PORT).asInteger(); - final String host = context.getProperty(HOSTNAME).getValue(); + final int port = context.getProperty(PORT).evaluateAttributeExpressions().asInteger(); + final String host = context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue(); final String protocol = context.getProperty(PROTOCOL).getValue(); - final int maxSendBuffer = context.getProperty(MAX_SOCKET_SEND_BUFFER_SIZE).asDataSize(DataUnit.B).intValue(); - final int timeout = context.getProperty(TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue(); + final int maxSendBuffer = context.getProperty(MAX_SOCKET_SEND_BUFFER_SIZE).evaluateAttributeExpressions().asDataSize(DataUnit.B).intValue(); + final int timeout = context.getProperty(TIMEOUT).evaluateAttributeExpressions().asTimePeriod(TimeUnit.MILLISECONDS).intValue(); final SSLContextService sslContextService = context.getProperty(SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class); return createSender(sslContextService, protocol, host, port, maxSendBuffer, timeout); } @@ -299,11 +304,11 @@ public class PutSyslog extends AbstractSyslogProcessor { @Override public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException { final String protocol = context.getProperty(PROTOCOL).getValue(); - final int batchSize = context.getProperty(BATCH_SIZE).asInteger(); + final int batchSize = context.getProperty(BATCH_SIZE).evaluateAttributeExpressions().asInteger(); final List<FlowFile> flowFiles = session.get(batchSize); if (flowFiles == null || flowFiles.isEmpty()) { - final PruneResult result = pruneIdleSenders(context.getProperty(IDLE_EXPIRATION).asTimePeriod(TimeUnit.MILLISECONDS).longValue()); + final PruneResult result = pruneIdleSenders(context.getProperty(IDLE_EXPIRATION).evaluateAttributeExpressions().asTimePeriod(TimeUnit.MILLISECONDS).longValue()); // yield if we closed an idle connection, or if there were no connections in the first place if (result.getNumClosed() > 0 || (result.getNumClosed() == 0 && result.getNumConsidered() == 0)) { context.yield(); @@ -329,11 +334,11 @@ public class PutSyslog extends AbstractSyslogProcessor { } } - final String port = context.getProperty(PORT).getValue(); - final String host = context.getProperty(HOSTNAME).getValue(); + final String port = context.getProperty(PORT).evaluateAttributeExpressions().getValue(); + final String host = context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue(); final String transitUri = new StringBuilder().append(protocol).append("://").append(host).append(":").append(port).toString(); final AtomicReference<IOException> exceptionHolder = new AtomicReference<>(null); - final Charset charSet = Charset.forName(context.getProperty(CHARSET).getValue()); + final Charset charSet = Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions().getValue()); try { for (FlowFile flowFile : flowFiles) { http://git-wip-us.apache.org/repos/asf/nifi/blob/56ad14ad/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenSyslog.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenSyslog.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenSyslog.java index 2c199c1..66c3a83 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenSyslog.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenSyslog.java @@ -253,7 +253,7 @@ public class TestListenSyslog { public List<PropertyDescriptor> getSupportedPropertyDescriptors() { final List<PropertyDescriptor> properties = new ArrayList<>(super.getSupportedPropertyDescriptors()); properties.remove(PORT); - properties.add(new PropertyDescriptor.Builder().name(PORT.getName()).addValidator(Validator.VALID).build()); + properties.add(new PropertyDescriptor.Builder().name(PORT.getName()).expressionLanguageSupported(true).addValidator(Validator.VALID).build()); return properties; } http://git-wip-us.apache.org/repos/asf/nifi/blob/56ad14ad/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutSyslog.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutSyslog.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutSyslog.java index 60fe37d..42569d9 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutSyslog.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutSyslog.java @@ -115,6 +115,53 @@ public class TestPutSyslog { } @Test + public void testValidELPropertiesTcp() { + final String pri = "34"; + final String version = "1"; + final String stamp = "2003-10-11T22:14:15.003Z"; + final String host = "mymachine.example.com"; + final String body = "su - ID47 - BOM'su root' failed for lonvick on /dev/pts/8"; + + final String expectedMessage = "<" + pri + ">" + version + " " + stamp + " " + host + " " + body; + + runner.setProperty(PutSyslog.HOSTNAME, "${'hostname'}"); + runner.setProperty(PutSyslog.PORT, "${port}"); + runner.setProperty(PutSyslog.CHARSET, "${charset}"); + runner.setProperty(PutSyslog.TIMEOUT, "${timeout}"); + runner.setProperty(PutSyslog.MAX_SOCKET_SEND_BUFFER_SIZE, "${maxSocketSenderBufferSize}"); + runner.setProperty(PutSyslog.IDLE_EXPIRATION, "${idleExpiration}"); + runner.setProperty(PutSyslog.PROTOCOL, PutSyslog.TCP_VALUE); + runner.setProperty(PutSyslog.MSG_PRIORITY, pri); + runner.setProperty(PutSyslog.MSG_VERSION, version); + runner.setProperty(PutSyslog.MSG_TIMESTAMP, stamp); + runner.setProperty(PutSyslog.MSG_HOSTNAME, host); + runner.setProperty(PutSyslog.MSG_BODY, body); + + runner.assertValid(); + runner.setVariable("hostname", "hostname"); + runner.setVariable("port", "10443"); + runner.setVariable("charset", "UTF-8"); + runner.setVariable("timeout", "10 secs"); + runner.setVariable("maxSocketSenderBufferSize", "10 mb"); + runner.setVariable("idleExpiration", "10 secs"); + + runner.enqueue("incoming data".getBytes(Charset.forName("UTF-8"))); + runner.run(); + + runner.assertAllFlowFilesTransferred(PutSyslog.REL_SUCCESS, 1); + Assert.assertEquals(1, sender.messages.size()); + Assert.assertEquals(expectedMessage, sender.messages.get(0).replace("\n", "")); + + final List<ProvenanceEventRecord> events = runner.getProvenanceEvents(); + Assert.assertNotNull(events); + Assert.assertEquals(1, events.size()); + + final ProvenanceEventRecord event = events.get(0); + Assert.assertEquals(ProvenanceEventType.SEND, event.getEventType()); + Assert.assertEquals("TCP://hostname:10443", event.getTransitUri()); + } + + @Test public void testValidMessageStaticPropertiesNoVersion() { final String pri = "34"; final String stamp = "2003-10-11T22:14:15.003Z";
