You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by GitBox <gi...@apache.org> on 2021/04/30 15:14:54 UTC
[GitHub] [nifi] exceptionfactory opened a new pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
exceptionfactory opened a new pull request #5044:
URL: https://github.com/apache/nifi/pull/5044
#### Description of PR
NIFI-8462 Refactors the `PutSyslog` and `ListenSyslog` Processors to use Netty for socket communication. This approach replaces the use of custom socket channel dispatcher classes in `nifi-processor-utils`. Leveraging the Netty `SslHandler` also removes the use of custom `SSLSocket` and `SSLEngine` handling in `nifi-security-socket-ssl` classes. This approach resolves latent issues with TLS 1.3 communication in these processors, described in NIFI-7468.
The new `nifi-event-transport` module abstracts communication handling for both UDP and TCP communication, providing interfaces and classes that can be leveraged in other components. Refactoring other components to leverage Netty should eventually remove the need for error-prone `SSLEngine` handling in custom classes.
Changes also include refactored unit tests for `PutSyslog` and `ListenSyslog`, eliminating the use of mocked senders and removing unnecessary integration tests.
In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:
### For all changes:
- [X] Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
- [X] Does your PR title start with **NIFI-XXXX** where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
- [X] Has your PR been rebased against the latest commit within the target branch (typically `main`)?
- [X] Is your initial contribution a single, squashed commit? _Additional commits in response to PR reviewer feedback should be made on this branch and pushed to allow change tracking. Do not `squash` or use `--force` when pushing to allow for clean monitoring of changes._
### For code changes:
- [X] Have you ensured that the full suite of tests is executed via `mvn -Pcontrib-check clean install` at the root `nifi` folder?
- [X] Have you written or updated unit tests to verify your changes?
- [X] Have you verified that the full build is successful on JDK 8?
- [X] Have you verified that the full build is successful on JDK 11?
- [ ] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)?
- [ ] If applicable, have you updated the `LICENSE` file, including the main `LICENSE` file under `nifi-assembly`?
- [ ] If applicable, have you updated the `NOTICE` file, including the main `NOTICE` file found under `nifi-assembly`?
- [ ] If adding new Properties, have you added `.displayName` in addition to .name (programmatic access) for each of the new properties?
### For documentation related changes:
- [X] Have you ensured that format looks appropriate for the output in which it is rendered?
### Note:
Please ensure that once the PR is submitted, you check GitHub Actions CI for build issues and submit an update to your PR as soon as possible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a change in pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on a change in pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#discussion_r632540020
##########
File path: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSyslog.java
##########
@@ -222,213 +224,97 @@ protected void init(final ProcessorInitializationContext context) {
}
@OnScheduled
- public void onScheduled(final ProcessContext context) throws IOException {
- // initialize the queue of senders, one per task, senders will get created on the fly in onTrigger
- this.senderPool = new LinkedBlockingQueue<>(context.getMaxConcurrentTasks());
- }
-
- protected ChannelSender createSender(final ProcessContext context) throws IOException {
- final int port = context.getProperty(PORT).evaluateAttributeExpressions().asInteger();
- final String host = context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue();
+ public void onScheduled(final ProcessContext context) throws InterruptedException {
+ eventSender = getEventSender(context);
final String protocol = context.getProperty(PROTOCOL).getValue();
- 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);
- }
-
- // visible for testing to override and provide a mock sender if desired
- protected ChannelSender createSender(final SSLContextService sslContextService, final String protocol, final String host,
- final int port, final int maxSendBufferSize, final int timeout)
- throws IOException {
-
- ChannelSender sender;
- if (protocol.equals(UDP_VALUE.getValue())) {
- sender = new DatagramChannelSender(host, port, maxSendBufferSize, getLogger());
- } else {
- // if an SSLContextService is provided then we make a secure sender
- if (sslContextService != null) {
- final SSLContext sslContext = sslContextService.createContext();
- sender = new SSLSocketChannelSender(host, port, maxSendBufferSize, sslContext, getLogger());
- } else {
- sender = new SocketChannelSender(host, port, maxSendBufferSize, getLogger());
- }
- }
- sender.setTimeout(timeout);
- sender.open();
- return sender;
+ final String hostname = context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue();
+ final int port = context.getProperty(PORT).evaluateAttributeExpressions().asInteger();
+ transitUri = String.format("%s://%s:%s", protocol, hostname, port);
}
@OnStopped
- public void onStopped() {
- if (senderPool != null) {
- ChannelSender sender = senderPool.poll();
- while (sender != null) {
- sender.close();
- sender = senderPool.poll();
- }
+ public void onStopped() throws Exception {
+ if (eventSender != null) {
+ eventSender.close();
}
}
- private PruneResult pruneIdleSenders(final long idleThreshold){
- int numClosed = 0;
- int numConsidered = 0;
-
- long currentTime = System.currentTimeMillis();
- final List<ChannelSender> putBack = new ArrayList<>();
-
- // if a connection hasn't been used with in the threshold then it gets closed
- ChannelSender sender;
- while ((sender = senderPool.poll()) != null) {
- numConsidered++;
- if (currentTime > (sender.getLastUsed() + idleThreshold)) {
- getLogger().debug("Closing idle connection...");
- sender.close();
- numClosed++;
- } else {
- putBack.add(sender);
- }
- }
-
- // re-queue senders that weren't idle, but if the queue is full then close the sender
- for (ChannelSender putBackSender : putBack) {
- boolean returned = senderPool.offer(putBackSender);
- if (!returned) {
- putBackSender.close();
- }
- }
-
- return new PruneResult(numClosed, numConsidered);
- }
-
@Override
- public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
- final String protocol = context.getProperty(PROTOCOL).getValue();
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
final int batchSize = context.getProperty(BATCH_SIZE).evaluateAttributeExpressions().asInteger();
-
final List<FlowFile> flowFiles = session.get(batchSize);
Review comment:
That's a good point, I removed the yield when I removed the unnecessary pruning of idle senders, but it still makes sense to yield in the absence of Flow Files.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a change in pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on a change in pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#discussion_r632532916
##########
File path: nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/netty/EventLoopGroupFactory.java
##########
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.event.transport.netty;
+
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.util.concurrent.DefaultThreadFactory;
+
+import java.util.Objects;
+import java.util.concurrent.ThreadFactory;
+
+/**
+ * Event Loop Group Factory for standardized instance creation
+ */
+class EventLoopGroupFactory {
+ private static final String DEFAULT_THREAD_NAME_PREFIX = "NettyEventLoopGroup";
+
+ private static final boolean DAEMON_THREAD_ENABLED = true;
+
+ private String threadNamePrefix = DEFAULT_THREAD_NAME_PREFIX;
+
+ private int workerThreads;
+
+ /**
+ * Set Thread Name Prefix used in Netty NioEventLoopGroup defaults to NettyChannel
+ *
+ * @param threadNamePrefix Thread Name Prefix
+ */
+ public void setThreadNamePrefix(final String threadNamePrefix) {
+ this.threadNamePrefix = Objects.requireNonNull(threadNamePrefix, "Thread Name Prefix required");
+ }
+
+ /**
+ * Set Worker Threads used in Netty NioEventLoopGroup with 0 interpreted as the default based on available processors
+ *
+ * @param workerThreads NioEventLoopGroup Worker Threads
+ */
+ public void setWorkerThreads(final int workerThreads) {
+ this.workerThreads = workerThreads;
+ }
+
+ EventLoopGroup getEventLoopGroup() {
Review comment:
Good question, I originally intended for this base class and the inheriting subclasses to be scoped as narrowly as possible so that implementations would remain in this package. Changing this to protected would maintain that intent while allowing a bit more flexibility, so will change to protected.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] ottobackwards commented on a change in pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
ottobackwards commented on a change in pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#discussion_r624525871
##########
File path: nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/netty/ByteArrayMessageNettyEventServerFactory.java
##########
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.event.transport.netty;
+
+import io.netty.buffer.Unpooled;
+import io.netty.handler.codec.DelimiterBasedFrameDecoder;
+import io.netty.handler.codec.bytes.ByteArrayDecoder;
+import org.apache.nifi.event.transport.configuration.TransportProtocol;
+import org.apache.nifi.event.transport.message.ByteArrayMessage;
+import org.apache.nifi.event.transport.netty.channel.ByteArrayMessageChannelHandler;
+import org.apache.nifi.event.transport.netty.codec.DatagramByteArrayMessageDecoder;
+import org.apache.nifi.event.transport.netty.channel.LogExceptionChannelHandler;
+import org.apache.nifi.event.transport.netty.codec.SocketByteArrayMessageDecoder;
+import org.apache.nifi.logging.ComponentLog;
+
+import java.util.Arrays;
+import java.util.concurrent.BlockingQueue;
+
+/**
+ * Netty Event Server Factory for Byte Array Messages
+ */
+public class ByteArrayMessageNettyEventServerFactory extends NettyEventServerFactory {
+ private static final boolean STRIP_DELIMITER = true;
+
+ /**
+ * Netty Event Server Factory with configurable delimiter and queue of Byte Array Messages
+ *
+ * @param log Component Log
+ * @param address Remote Address
+ * @param port Remote Port Number
+ * @param protocol Channel Protocol
+ * @param delimiter Message Delimiter
+ * @param maxFrameLength Maximum Frame Length for delimited TCP messages
+ * @param messages Blocking Queue for events received
+ */
+ public ByteArrayMessageNettyEventServerFactory(final ComponentLog log,
+ final String address,
+ final int port,
+ final TransportProtocol protocol,
+ final byte[] delimiter,
+ final int maxFrameLength,
+ final BlockingQueue<ByteArrayMessage> messages) {
+ super(address, port, protocol);
+
+ final LogExceptionChannelHandler logExceptionChannelHandler = new LogExceptionChannelHandler(log);
+ final ByteArrayMessageChannelHandler byteArrayMessageChannelHandler = new ByteArrayMessageChannelHandler(messages);
+
+ if (TransportProtocol.UDP.equals(protocol)) {
+ setHandlerSupplier(() -> Arrays.asList(
+ logExceptionChannelHandler,
+ new DatagramByteArrayMessageDecoder(),
+ byteArrayMessageChannelHandler
+ ));
+ } else {
Review comment:
would this be more re-usable if you could set the FrameDecoder? Or would you be expected to re-implement the factory?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a change in pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on a change in pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#discussion_r624557495
##########
File path: nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/netty/NettyEventSender.java
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.event.transport.netty;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.pool.ChannelPool;
+import io.netty.util.concurrent.Future;
+import org.apache.nifi.event.transport.EventException;
+import org.apache.nifi.event.transport.EventSender;
+
+import java.net.SocketAddress;
+
+/**
+ * Netty Event Sender with Channel Pool
+ *
+ * @param <T> Event Type
+ */
+class NettyEventSender<T> implements EventSender<T> {
+ private final EventLoopGroup group;
+
+ private final ChannelPool channelPool;
+
+ private final SocketAddress remoteAddress;
+
+ /**
+ * Netty Channel Event Sender with Event Loop Group and Channel Pool
+ *
+ * @param group Event Loop Group
+ * @param channelPool Channel Pool
+ * @param remoteAddress Remote Address
+ */
+ NettyEventSender(final EventLoopGroup group, final ChannelPool channelPool, final SocketAddress remoteAddress) {
+ this.group = group;
+ this.channelPool = channelPool;
+ this.remoteAddress = remoteAddress;
+ }
+
+ /**
+ * Send Event using Channel acquired from Channel Pool
+ *
+ * @param event Event
+ */
+ @Override
+ public void sendEvent(final T event) {
+ try {
+ final Future<Channel> futureChannel = channelPool.acquire().sync();
+ final Channel channel = futureChannel.get();
+ try {
Review comment:
The `FixedChannelPool` configured in `NettyEventSenderFactory` includes a `ChannelHealthChecker` that checks whether the channel is active and connected. That should minimize the possible types of exceptions thrown in this try block, but the finally block is intended to ensure that the `Channel` is always returned to the pool.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] thenatog commented on pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
thenatog commented on pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#issuecomment-847488079
Tested this out using the template found here: https://community.cloudera.com/t5/Community-Articles/NiFi-Send-to-syslog/ta-p/248638 and found that PutSyslog and ListenSyslog are working. Reviewed the code and it looks good.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] gresockj commented on a change in pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
gresockj commented on a change in pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#discussion_r632070182
##########
File path: nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/message/ByteArrayMessage.java
##########
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.event.transport.message;
+
+/**
+ * Byte Array Message with Sender
+ */
+public class ByteArrayMessage {
+ private byte[] message;
Review comment:
Any benefit in making these final, since there's no getters?
##########
File path: nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/netty/NettyEventSenderFactory.java
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.event.transport.netty;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.pool.ChannelHealthChecker;
+import io.netty.channel.pool.ChannelPool;
+import io.netty.channel.pool.ChannelPoolHandler;
+import io.netty.channel.pool.FixedChannelPool;
+import io.netty.channel.socket.nio.NioDatagramChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import org.apache.nifi.event.transport.EventSender;
+import org.apache.nifi.event.transport.EventSenderFactory;
+import org.apache.nifi.event.transport.configuration.TransportProtocol;
+import org.apache.nifi.event.transport.netty.channel.pool.InitializingChannelPoolHandler;
+import org.apache.nifi.event.transport.netty.channel.ssl.ClientSslStandardChannelInitializer;
+import org.apache.nifi.event.transport.netty.channel.StandardChannelInitializer;
+
+import javax.net.ssl.SSLContext;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Supplier;
+
+/**
+ * Netty Event Sender Factory
+ */
+public class NettyEventSenderFactory<T> extends EventLoopGroupFactory implements EventSenderFactory<T> {
+ private static final int MAX_PENDING_ACQUIRES = 1024;
+
+ private Duration timeout = Duration.ofSeconds(30);
+
+ private int maxConnections = Runtime.getRuntime().availableProcessors() * 2;
+
+ private Supplier<List<ChannelHandler>> handlerSupplier = () -> Collections.emptyList();
+
+ private final String address;
Review comment:
Can you move the final ones to the top, just to make it easier to visually distinguish the mutable vs. immutable fields?
##########
File path: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenSyslog.java
##########
@@ -152,7 +145,7 @@
"The maximum number of Syslog events to add to a single FlowFile. If multiple events are available, they will be concatenated along with "
+ "the <Message Delimiter> up to this configured maximum number of messages")
.addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
- .expressionLanguageSupported(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
Review comment:
If you enable EL, you'll need to call evaluateAttributeExpressions when you use the property below. Also, I'm not sure FLOWFILE_ATTRIBUTES is appropriate here, since ListenSyslog is annotated with INPUT_FORBIDDEN.
##########
File path: nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/netty/EventLoopGroupFactory.java
##########
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.event.transport.netty;
+
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.util.concurrent.DefaultThreadFactory;
+
+import java.util.Objects;
+import java.util.concurrent.ThreadFactory;
+
+/**
+ * Event Loop Group Factory for standardized instance creation
+ */
+class EventLoopGroupFactory {
+ private static final String DEFAULT_THREAD_NAME_PREFIX = "NettyEventLoopGroup";
+
+ private static final boolean DAEMON_THREAD_ENABLED = true;
+
+ private String threadNamePrefix = DEFAULT_THREAD_NAME_PREFIX;
+
+ private int workerThreads;
+
+ /**
+ * Set Thread Name Prefix used in Netty NioEventLoopGroup defaults to NettyChannel
+ *
+ * @param threadNamePrefix Thread Name Prefix
+ */
+ public void setThreadNamePrefix(final String threadNamePrefix) {
+ this.threadNamePrefix = Objects.requireNonNull(threadNamePrefix, "Thread Name Prefix required");
+ }
+
+ /**
+ * Set Worker Threads used in Netty NioEventLoopGroup with 0 interpreted as the default based on available processors
+ *
+ * @param workerThreads NioEventLoopGroup Worker Threads
+ */
+ public void setWorkerThreads(final int workerThreads) {
+ this.workerThreads = workerThreads;
+ }
+
+ EventLoopGroup getEventLoopGroup() {
Review comment:
Why package scope instead of protected?
##########
File path: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSyslog.java
##########
@@ -222,213 +224,97 @@ protected void init(final ProcessorInitializationContext context) {
}
@OnScheduled
- public void onScheduled(final ProcessContext context) throws IOException {
- // initialize the queue of senders, one per task, senders will get created on the fly in onTrigger
- this.senderPool = new LinkedBlockingQueue<>(context.getMaxConcurrentTasks());
- }
-
- protected ChannelSender createSender(final ProcessContext context) throws IOException {
- final int port = context.getProperty(PORT).evaluateAttributeExpressions().asInteger();
- final String host = context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue();
+ public void onScheduled(final ProcessContext context) throws InterruptedException {
+ eventSender = getEventSender(context);
final String protocol = context.getProperty(PROTOCOL).getValue();
- 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);
- }
-
- // visible for testing to override and provide a mock sender if desired
- protected ChannelSender createSender(final SSLContextService sslContextService, final String protocol, final String host,
- final int port, final int maxSendBufferSize, final int timeout)
- throws IOException {
-
- ChannelSender sender;
- if (protocol.equals(UDP_VALUE.getValue())) {
- sender = new DatagramChannelSender(host, port, maxSendBufferSize, getLogger());
- } else {
- // if an SSLContextService is provided then we make a secure sender
- if (sslContextService != null) {
- final SSLContext sslContext = sslContextService.createContext();
- sender = new SSLSocketChannelSender(host, port, maxSendBufferSize, sslContext, getLogger());
- } else {
- sender = new SocketChannelSender(host, port, maxSendBufferSize, getLogger());
- }
- }
- sender.setTimeout(timeout);
- sender.open();
- return sender;
+ final String hostname = context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue();
+ final int port = context.getProperty(PORT).evaluateAttributeExpressions().asInteger();
+ transitUri = String.format("%s://%s:%s", protocol, hostname, port);
}
@OnStopped
- public void onStopped() {
- if (senderPool != null) {
- ChannelSender sender = senderPool.poll();
- while (sender != null) {
- sender.close();
- sender = senderPool.poll();
- }
+ public void onStopped() throws Exception {
+ if (eventSender != null) {
+ eventSender.close();
}
}
- private PruneResult pruneIdleSenders(final long idleThreshold){
- int numClosed = 0;
- int numConsidered = 0;
-
- long currentTime = System.currentTimeMillis();
- final List<ChannelSender> putBack = new ArrayList<>();
-
- // if a connection hasn't been used with in the threshold then it gets closed
- ChannelSender sender;
- while ((sender = senderPool.poll()) != null) {
- numConsidered++;
- if (currentTime > (sender.getLastUsed() + idleThreshold)) {
- getLogger().debug("Closing idle connection...");
- sender.close();
- numClosed++;
- } else {
- putBack.add(sender);
- }
- }
-
- // re-queue senders that weren't idle, but if the queue is full then close the sender
- for (ChannelSender putBackSender : putBack) {
- boolean returned = senderPool.offer(putBackSender);
- if (!returned) {
- putBackSender.close();
- }
- }
-
- return new PruneResult(numClosed, numConsidered);
- }
-
@Override
- public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
- final String protocol = context.getProperty(PROTOCOL).getValue();
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
final int batchSize = context.getProperty(BATCH_SIZE).evaluateAttributeExpressions().asInteger();
-
final List<FlowFile> flowFiles = session.get(batchSize);
Review comment:
I noticed that PutSyslog continually runs even with no incoming flowfiles, probably because the logic to yield and return if flowFiles is null or empty has been removed. What do you think about adding that check back?
##########
File path: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSyslog.java
##########
@@ -150,6 +149,7 @@
"messages will be sent over a secure connection.")
.required(false)
.identifiesControllerService(SSLContextService.class)
+ .dependsOn(PROTOCOL, TCP_VALUE)
Review comment:
I noticed that in ListenSyslog, the SSL Context Service property is always available, but in PutSyslog it's conditional on TCP being set. Can you add the condition to both?
Also, if you have an SSL Context Service selected and then change it back to UDP, the processor is marked as invalid even though the SSL Context Service property is no longer available to be configured. Is it possible to unset this property if protocol is set to UDP, or is that just a limitation of the current "dependsOn" functionality?
##########
File path: nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/netty/NettyEventSenderFactory.java
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.event.transport.netty;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.pool.ChannelHealthChecker;
+import io.netty.channel.pool.ChannelPool;
+import io.netty.channel.pool.ChannelPoolHandler;
+import io.netty.channel.pool.FixedChannelPool;
+import io.netty.channel.socket.nio.NioDatagramChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import org.apache.nifi.event.transport.EventSender;
+import org.apache.nifi.event.transport.EventSenderFactory;
+import org.apache.nifi.event.transport.configuration.TransportProtocol;
+import org.apache.nifi.event.transport.netty.channel.pool.InitializingChannelPoolHandler;
+import org.apache.nifi.event.transport.netty.channel.ssl.ClientSslStandardChannelInitializer;
+import org.apache.nifi.event.transport.netty.channel.StandardChannelInitializer;
+
+import javax.net.ssl.SSLContext;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Supplier;
+
+/**
+ * Netty Event Sender Factory
+ */
+public class NettyEventSenderFactory<T> extends EventLoopGroupFactory implements EventSenderFactory<T> {
+ private static final int MAX_PENDING_ACQUIRES = 1024;
+
+ private Duration timeout = Duration.ofSeconds(30);
+
+ private int maxConnections = Runtime.getRuntime().availableProcessors() * 2;
+
+ private Supplier<List<ChannelHandler>> handlerSupplier = () -> Collections.emptyList();
+
+ private final String address;
+
+ private final int port;
+
+ private final TransportProtocol protocol;
+
+ private SSLContext sslContext;
+
+ public NettyEventSenderFactory(final String address, final int port, final TransportProtocol protocol) {
+ this.address = address;
+ this.port = port;
+ this.protocol = protocol;
+ }
+
+ /**
+ * Set Channel Handler Supplier
+ *
+ * @param handlerSupplier Channel Handler Supplier
+ */
+ public void setHandlerSupplier(final Supplier<List<ChannelHandler>> handlerSupplier) {
+ this.handlerSupplier = Objects.requireNonNull(handlerSupplier);
+ }
+
+ /**
+ * Set SSL Context to enable TLS Channel Handler
+ *
+ * @param sslContext SSL Context
+ */
+ public void setSslContext(final SSLContext sslContext) {
+ this.sslContext = sslContext;
+ }
+
+ /**
+ * Set Timeout for Connections and Communication
+ *
+ * @param timeout Timeout Duration
+ */
+ public void setTimeout(final Duration timeout) {
+ this.timeout = Objects.requireNonNull(timeout, "Timeout required");
+ }
+
+ /**
+ * Set Maximum Connections for Channel Pool
+ *
+ * @param maxConnections Maximum Number of connections defaults to available processors multiplied by 2
+ */
+ public void setMaxConnections(final int maxConnections) {
+ this.maxConnections = maxConnections;
+ }
+
+ /**
+ * Get Event Sender with connected Channel
+ *
+ * @return Connected Event Sender
+ */
+ public EventSender<T> getEventSender() {
+ final Bootstrap bootstrap = new Bootstrap();
+ bootstrap.remoteAddress(new InetSocketAddress(address, port));
+ final EventLoopGroup group = getEventLoopGroup();
+ bootstrap.group(group);
+
+ if (TransportProtocol.UDP.equals(protocol)) {
+ bootstrap.channel(NioDatagramChannel.class);
+ } else {
+ bootstrap.channel(NioSocketChannel.class);
+ }
+
+ setChannelOptions(bootstrap);
+ return getConfiguredEventSender(bootstrap);
+ }
+
+ private void setChannelOptions(final Bootstrap bootstrap) {
+ final int timeoutMilliseconds = (int) timeout.toMillis();
+ bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, timeoutMilliseconds);
+ }
+
+ private EventSender<T> getConfiguredEventSender(final Bootstrap bootstrap) {
+ final SocketAddress remoteAddress = bootstrap.config().remoteAddress();
+ final ChannelPool channelPool = getChannelPool(bootstrap);
+ return new NettyEventSender<>(bootstrap.config().group(), channelPool, remoteAddress);
+ }
+
+ private ChannelPool getChannelPool(final Bootstrap bootstrap) {
+ final ChannelInitializer<Channel> channelInitializer = getChannelInitializer();
+ final ChannelPoolHandler handler = new InitializingChannelPoolHandler(channelInitializer);
+ return new FixedChannelPool(bootstrap,
+ handler,
+ ChannelHealthChecker.ACTIVE,
+ FixedChannelPool.AcquireTimeoutAction.FAIL,
+ timeout.toMillis(),
+ maxConnections,
+ MAX_PENDING_ACQUIRES);
+ }
+
+ private ChannelInitializer<Channel> getChannelInitializer() {
+ StandardChannelInitializer<Channel> channelInitializer;
+ if (sslContext == null) {
Review comment:
Ternary here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#issuecomment-841257313
Thanks for the thorough review and helpful feedback @gresockj! I added a commit including the changes and rebased on main to incorporate recent updates.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a change in pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on a change in pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#discussion_r632533499
##########
File path: nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/netty/NettyEventSenderFactory.java
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.event.transport.netty;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.pool.ChannelHealthChecker;
+import io.netty.channel.pool.ChannelPool;
+import io.netty.channel.pool.ChannelPoolHandler;
+import io.netty.channel.pool.FixedChannelPool;
+import io.netty.channel.socket.nio.NioDatagramChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import org.apache.nifi.event.transport.EventSender;
+import org.apache.nifi.event.transport.EventSenderFactory;
+import org.apache.nifi.event.transport.configuration.TransportProtocol;
+import org.apache.nifi.event.transport.netty.channel.pool.InitializingChannelPoolHandler;
+import org.apache.nifi.event.transport.netty.channel.ssl.ClientSslStandardChannelInitializer;
+import org.apache.nifi.event.transport.netty.channel.StandardChannelInitializer;
+
+import javax.net.ssl.SSLContext;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Supplier;
+
+/**
+ * Netty Event Sender Factory
+ */
+public class NettyEventSenderFactory<T> extends EventLoopGroupFactory implements EventSenderFactory<T> {
+ private static final int MAX_PENDING_ACQUIRES = 1024;
+
+ private Duration timeout = Duration.ofSeconds(30);
+
+ private int maxConnections = Runtime.getRuntime().availableProcessors() * 2;
+
+ private Supplier<List<ChannelHandler>> handlerSupplier = () -> Collections.emptyList();
+
+ private final String address;
Review comment:
Sure, thanks for the suggestion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a change in pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on a change in pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#discussion_r632529701
##########
File path: nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/message/ByteArrayMessage.java
##########
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.event.transport.message;
+
+/**
+ * Byte Array Message with Sender
+ */
+public class ByteArrayMessage {
+ private byte[] message;
Review comment:
Good point, will make the change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a change in pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on a change in pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#discussion_r632534325
##########
File path: nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/netty/NettyEventSenderFactory.java
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.event.transport.netty;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.pool.ChannelHealthChecker;
+import io.netty.channel.pool.ChannelPool;
+import io.netty.channel.pool.ChannelPoolHandler;
+import io.netty.channel.pool.FixedChannelPool;
+import io.netty.channel.socket.nio.NioDatagramChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import org.apache.nifi.event.transport.EventSender;
+import org.apache.nifi.event.transport.EventSenderFactory;
+import org.apache.nifi.event.transport.configuration.TransportProtocol;
+import org.apache.nifi.event.transport.netty.channel.pool.InitializingChannelPoolHandler;
+import org.apache.nifi.event.transport.netty.channel.ssl.ClientSslStandardChannelInitializer;
+import org.apache.nifi.event.transport.netty.channel.StandardChannelInitializer;
+
+import javax.net.ssl.SSLContext;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Supplier;
+
+/**
+ * Netty Event Sender Factory
+ */
+public class NettyEventSenderFactory<T> extends EventLoopGroupFactory implements EventSenderFactory<T> {
+ private static final int MAX_PENDING_ACQUIRES = 1024;
+
+ private Duration timeout = Duration.ofSeconds(30);
+
+ private int maxConnections = Runtime.getRuntime().availableProcessors() * 2;
+
+ private Supplier<List<ChannelHandler>> handlerSupplier = () -> Collections.emptyList();
+
+ private final String address;
+
+ private final int port;
+
+ private final TransportProtocol protocol;
+
+ private SSLContext sslContext;
+
+ public NettyEventSenderFactory(final String address, final int port, final TransportProtocol protocol) {
+ this.address = address;
+ this.port = port;
+ this.protocol = protocol;
+ }
+
+ /**
+ * Set Channel Handler Supplier
+ *
+ * @param handlerSupplier Channel Handler Supplier
+ */
+ public void setHandlerSupplier(final Supplier<List<ChannelHandler>> handlerSupplier) {
+ this.handlerSupplier = Objects.requireNonNull(handlerSupplier);
+ }
+
+ /**
+ * Set SSL Context to enable TLS Channel Handler
+ *
+ * @param sslContext SSL Context
+ */
+ public void setSslContext(final SSLContext sslContext) {
+ this.sslContext = sslContext;
+ }
+
+ /**
+ * Set Timeout for Connections and Communication
+ *
+ * @param timeout Timeout Duration
+ */
+ public void setTimeout(final Duration timeout) {
+ this.timeout = Objects.requireNonNull(timeout, "Timeout required");
+ }
+
+ /**
+ * Set Maximum Connections for Channel Pool
+ *
+ * @param maxConnections Maximum Number of connections defaults to available processors multiplied by 2
+ */
+ public void setMaxConnections(final int maxConnections) {
+ this.maxConnections = maxConnections;
+ }
+
+ /**
+ * Get Event Sender with connected Channel
+ *
+ * @return Connected Event Sender
+ */
+ public EventSender<T> getEventSender() {
+ final Bootstrap bootstrap = new Bootstrap();
+ bootstrap.remoteAddress(new InetSocketAddress(address, port));
+ final EventLoopGroup group = getEventLoopGroup();
+ bootstrap.group(group);
+
+ if (TransportProtocol.UDP.equals(protocol)) {
+ bootstrap.channel(NioDatagramChannel.class);
+ } else {
+ bootstrap.channel(NioSocketChannel.class);
+ }
+
+ setChannelOptions(bootstrap);
+ return getConfiguredEventSender(bootstrap);
+ }
+
+ private void setChannelOptions(final Bootstrap bootstrap) {
+ final int timeoutMilliseconds = (int) timeout.toMillis();
+ bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, timeoutMilliseconds);
+ }
+
+ private EventSender<T> getConfiguredEventSender(final Bootstrap bootstrap) {
+ final SocketAddress remoteAddress = bootstrap.config().remoteAddress();
+ final ChannelPool channelPool = getChannelPool(bootstrap);
+ return new NettyEventSender<>(bootstrap.config().group(), channelPool, remoteAddress);
+ }
+
+ private ChannelPool getChannelPool(final Bootstrap bootstrap) {
+ final ChannelInitializer<Channel> channelInitializer = getChannelInitializer();
+ final ChannelPoolHandler handler = new InitializingChannelPoolHandler(channelInitializer);
+ return new FixedChannelPool(bootstrap,
+ handler,
+ ChannelHealthChecker.ACTIVE,
+ FixedChannelPool.AcquireTimeoutAction.FAIL,
+ timeout.toMillis(),
+ maxConnections,
+ MAX_PENDING_ACQUIRES);
+ }
+
+ private ChannelInitializer<Channel> getChannelInitializer() {
+ StandardChannelInitializer<Channel> channelInitializer;
+ if (sslContext == null) {
Review comment:
Sounds good, will update.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] thenatog commented on pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
thenatog commented on pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#issuecomment-847937054
+1 will merge.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] thenatog closed pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
thenatog closed pull request #5044:
URL: https://github.com/apache/nifi/pull/5044
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#issuecomment-830687467
> If someone was going to use these new classes, it might be good to have a readme or something for this part of the code
Thanks for the feedback @ottobackwards! Providing some additional documentation would be helpful, although I think it would be better to wait until other components start leveraging this module. I attempted to provide some high-level abstractions in the `EventSender` and `EventServer` interfaces, but even these interfaces and implementations are geared sending and receiving byte arrays for Syslog messages. There should be some reuse of these classes for other processors, such as `PutTCP` and `ListenTCP`, among others. If there are particular components that could use some additional description, feel free to point those out. However, I expect some changes when it comes to leveraging similar approaches with other components.
Please let me know if you have any additional comments or questions!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a change in pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on a change in pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#discussion_r632536283
##########
File path: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSyslog.java
##########
@@ -150,6 +149,7 @@
"messages will be sent over a secure connection.")
.required(false)
.identifiesControllerService(SSLContextService.class)
+ .dependsOn(PROTOCOL, TCP_VALUE)
Review comment:
Good catch, I had intended to update both processors to make SSL Context Service depend on TCP.
I noticed the issue with SSL Context Service and changing from TCP to UDP. It seems like that is something that might make sense to address at the framework level.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a change in pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on a change in pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#discussion_r624557852
##########
File path: nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/netty/ByteArrayMessageNettyEventServerFactory.java
##########
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.event.transport.netty;
+
+import io.netty.buffer.Unpooled;
+import io.netty.handler.codec.DelimiterBasedFrameDecoder;
+import io.netty.handler.codec.bytes.ByteArrayDecoder;
+import org.apache.nifi.event.transport.configuration.TransportProtocol;
+import org.apache.nifi.event.transport.message.ByteArrayMessage;
+import org.apache.nifi.event.transport.netty.channel.ByteArrayMessageChannelHandler;
+import org.apache.nifi.event.transport.netty.codec.DatagramByteArrayMessageDecoder;
+import org.apache.nifi.event.transport.netty.channel.LogExceptionChannelHandler;
+import org.apache.nifi.event.transport.netty.codec.SocketByteArrayMessageDecoder;
+import org.apache.nifi.logging.ComponentLog;
+
+import java.util.Arrays;
+import java.util.concurrent.BlockingQueue;
+
+/**
+ * Netty Event Server Factory for Byte Array Messages
+ */
+public class ByteArrayMessageNettyEventServerFactory extends NettyEventServerFactory {
+ private static final boolean STRIP_DELIMITER = true;
+
+ /**
+ * Netty Event Server Factory with configurable delimiter and queue of Byte Array Messages
+ *
+ * @param log Component Log
+ * @param address Remote Address
+ * @param port Remote Port Number
+ * @param protocol Channel Protocol
+ * @param delimiter Message Delimiter
+ * @param maxFrameLength Maximum Frame Length for delimited TCP messages
+ * @param messages Blocking Queue for events received
+ */
+ public ByteArrayMessageNettyEventServerFactory(final ComponentLog log,
+ final String address,
+ final int port,
+ final TransportProtocol protocol,
+ final byte[] delimiter,
+ final int maxFrameLength,
+ final BlockingQueue<ByteArrayMessage> messages) {
+ super(address, port, protocol);
+
+ final LogExceptionChannelHandler logExceptionChannelHandler = new LogExceptionChannelHandler(log);
+ final ByteArrayMessageChannelHandler byteArrayMessageChannelHandler = new ByteArrayMessageChannelHandler(messages);
+
+ if (TransportProtocol.UDP.equals(protocol)) {
+ setHandlerSupplier(() -> Arrays.asList(
+ logExceptionChannelHandler,
+ new DatagramByteArrayMessageDecoder(),
+ byteArrayMessageChannelHandler
+ ));
+ } else {
Review comment:
That's a good question. The initial purpose of this particular `EventServerFactory` implementation was to abstract the configuration details necessary for constructing an `EventServer` capable of receiving delimited byte arrays. Netty provides a number of message decoders as well as the ability to create custom decoders capable of returning more structured objects. I would expect different implementations of `NettyEventServerFactory` for different use cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] gresockj commented on a change in pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
gresockj commented on a change in pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#discussion_r632555033
##########
File path: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenSyslog.java
##########
@@ -152,7 +145,7 @@
"The maximum number of Syslog events to add to a single FlowFile. If multiple events are available, they will be concatenated along with "
+ "the <Message Delimiter> up to this configured maximum number of messages")
.addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
- .expressionLanguageSupported(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
Review comment:
It might still make sense to let it be ExpressionLanguage.VARIABLE_REGISTRY if you want to allow for the additional flexibility, though it seems like a setting that wouldn't likely be externalized to the variable registry.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] ottobackwards commented on a change in pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
ottobackwards commented on a change in pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#discussion_r624528290
##########
File path: nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/netty/NettyEventSender.java
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.event.transport.netty;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.pool.ChannelPool;
+import io.netty.util.concurrent.Future;
+import org.apache.nifi.event.transport.EventException;
+import org.apache.nifi.event.transport.EventSender;
+
+import java.net.SocketAddress;
+
+/**
+ * Netty Event Sender with Channel Pool
+ *
+ * @param <T> Event Type
+ */
+class NettyEventSender<T> implements EventSender<T> {
+ private final EventLoopGroup group;
+
+ private final ChannelPool channelPool;
+
+ private final SocketAddress remoteAddress;
+
+ /**
+ * Netty Channel Event Sender with Event Loop Group and Channel Pool
+ *
+ * @param group Event Loop Group
+ * @param channelPool Channel Pool
+ * @param remoteAddress Remote Address
+ */
+ NettyEventSender(final EventLoopGroup group, final ChannelPool channelPool, final SocketAddress remoteAddress) {
+ this.group = group;
+ this.channelPool = channelPool;
+ this.remoteAddress = remoteAddress;
+ }
+
+ /**
+ * Send Event using Channel acquired from Channel Pool
+ *
+ * @param event Event
+ */
+ @Override
+ public void sendEvent(final T event) {
+ try {
+ final Future<Channel> futureChannel = channelPool.acquire().sync();
+ final Channel channel = futureChannel.get();
+ try {
Review comment:
Can the channel not be active here? Can it not be writeable?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a change in pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on a change in pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#discussion_r632537625
##########
File path: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenSyslog.java
##########
@@ -152,7 +145,7 @@
"The maximum number of Syslog events to add to a single FlowFile. If multiple events are available, they will be concatenated along with "
+ "the <Message Delimiter> up to this configured maximum number of messages")
.addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
- .expressionLanguageSupported(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
Review comment:
Good catch, that should not have been changed, will revert back.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] ottobackwards commented on pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
ottobackwards commented on pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#issuecomment-830652000
If someone was going to use these new classes, it might be good to have a readme or something for this part of the code
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a change in pull request #5044: NIFI-8462 Refactored PutSyslog and ListenSyslog using Netty
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on a change in pull request #5044:
URL: https://github.com/apache/nifi/pull/5044#discussion_r632574971
##########
File path: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenSyslog.java
##########
@@ -152,7 +145,7 @@
"The maximum number of Syslog events to add to a single FlowFile. If multiple events are available, they will be concatenated along with "
+ "the <Message Delimiter> up to this configured maximum number of messages")
.addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
- .expressionLanguageSupported(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
Review comment:
Yes, that would be an option, but agree that there doesn't seem to be much value in making it configurable through the variable registry.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org