[ https://issues.apache.org/jira/browse/NIFI-1420?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15167241#comment-15167241 ]
ASF GitHub Bot commented on NIFI-1420: -------------------------------------- Github user olegz commented on a diff in the pull request: https://github.com/apache/nifi/pull/233#discussion_r54099600 --- Diff: nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/ListenSplunkForwarder.java --- @@ -0,0 +1,208 @@ +/* + * 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.processors.splunk; + +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.SupportsBatching; +import org.apache.nifi.annotation.behavior.WritesAttribute; +import org.apache.nifi.annotation.behavior.WritesAttributes; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.components.AllowableValue; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.flowfile.attributes.CoreAttributes; +import org.apache.nifi.processor.DataUnit; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.listen.AbstractListenEventProcessor; +import org.apache.nifi.processor.util.listen.dispatcher.AsyncChannelDispatcher; +import org.apache.nifi.processor.util.listen.dispatcher.ChannelDispatcher; +import org.apache.nifi.processor.util.listen.dispatcher.DatagramChannelDispatcher; +import org.apache.nifi.processor.util.listen.dispatcher.SocketChannelDispatcher; +import org.apache.nifi.processor.util.listen.event.EventFactory; +import org.apache.nifi.processor.util.listen.event.StandardEvent; +import org.apache.nifi.processor.util.listen.handler.ChannelHandlerFactory; +import org.apache.nifi.processor.util.listen.handler.socket.SocketChannelHandlerFactory; +import org.apache.nifi.processor.util.listen.response.ChannelResponder; +import org.apache.nifi.ssl.SSLContextService; + +import javax.net.ssl.SSLContext; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.SelectableChannel; +import java.nio.channels.SocketChannel; +import java.nio.charset.Charset; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; + +@SupportsBatching +@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN) +@Tags({"listen", "splunk", "tcp", "udp", "logs"}) +@CapabilityDescription("Listens for data from a Splunk forwarder.") +@WritesAttributes({ + @WritesAttribute(attribute="splunk.sender", description="The sending host of the messages."), + @WritesAttribute(attribute="splunk.port", description="The sending port the messages were received over."), + @WritesAttribute(attribute="mime.type", description="The mime.type of the messages which is text/plain.") +}) +public class ListenSplunkForwarder extends AbstractListenEventProcessor<ListenSplunkForwarder.SplunkEvent> { + + public static final AllowableValue TCP_VALUE = new AllowableValue("TCP", "TCP"); + public static final AllowableValue UDP_VALUE = new AllowableValue("UDP", "UDP"); + + public static final PropertyDescriptor PROTOCOL = new PropertyDescriptor + .Builder().name("Protocol") + .description("The protocol for communication.") + .required(true) + .allowableValues(TCP_VALUE, UDP_VALUE) + .defaultValue(TCP_VALUE.getValue()) + .build(); + + public static final PropertyDescriptor SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder() + .name("SSL Context Service") + .description("The Controller Service to use in order to obtain an SSL Context. If this property is set, " + + "messages will be received over a secure connection.") + .required(false) + .identifiesControllerService(SSLContextService.class) + .build(); + + // it is only the array reference that is volatile - not the contents. + private volatile byte[] messageDemarcatorBytes; + + @Override + protected List<PropertyDescriptor> getAdditionalProperties() { + return Arrays.asList( + PROTOCOL, + MAX_CONNECTIONS, + MAX_BATCH_SIZE, + MESSAGE_DELIMITER, + SSL_CONTEXT_SERVICE + ); + } + + @Override + @OnScheduled + public void onScheduled(ProcessContext context) throws IOException { + super.onScheduled(context); + final String msgDemarcator = context.getProperty(MESSAGE_DELIMITER).getValue().replace("\\n", "\n").replace("\\r", "\r").replace("\\t", "\t"); + messageDemarcatorBytes = msgDemarcator.getBytes(charset); + } + + @Override + protected ChannelDispatcher createDispatcher(final ProcessContext context, final BlockingQueue<SplunkEvent> events) + throws IOException { + + final String protocol = context.getProperty(PROTOCOL).getValue(); + final int maxConnections = context.getProperty(MAX_CONNECTIONS).asInteger(); + final int bufferSize = context.getProperty(RECV_BUFFER_SIZE).asDataSize(DataUnit.B).intValue(); + final Charset charSet = Charset.forName(context.getProperty(CHARSET).getValue()); + + // initialize the buffer pool based on max number of connections and the buffer size + final LinkedBlockingQueue<ByteBuffer> bufferPool = new LinkedBlockingQueue<>(maxConnections); + for (int i = 0; i < maxConnections; i++) { + bufferPool.offer(ByteBuffer.allocate(bufferSize)); + } + + final EventFactory<SplunkEvent> eventFactory = new SplunkEventFactory(); + + if (UDP_VALUE.getValue().equals(protocol)) { + return new DatagramChannelDispatcher(eventFactory, bufferPool, events, getLogger()); + } else { + // if an SSLContextService was provided then create an SSLContext to pass down to the dispatcher + SSLContext sslContext = null; + final SSLContextService sslContextService = context.getProperty(SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class); + if (sslContextService != null) { + sslContext = sslContextService.createSSLContext(SSLContextService.ClientAuth.REQUIRED); + } + + final ChannelHandlerFactory<SplunkEvent<SocketChannel>, AsyncChannelDispatcher> handlerFactory = new SocketChannelHandlerFactory<>(); + return new SocketChannelDispatcher(eventFactory, handlerFactory, bufferPool, events, getLogger(), maxConnections, sslContext, charSet); + } + } + + @Override + public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException { + final int maxBatchSize = context.getProperty(MAX_BATCH_SIZE).asInteger(); --- End diff -- Considering that this is using the similar batching approach used in PutKafka, curious if you tested it with run-duration > 0. See comments in https://issues.apache.org/jira/browse/NIFI-1534 as it may be suffering from the same issue. > Splunk Processors > ----------------- > > Key: NIFI-1420 > URL: https://issues.apache.org/jira/browse/NIFI-1420 > Project: Apache NiFi > Issue Type: Improvement > Components: Extensions > Reporter: Bryan Bende > Assignee: Bryan Bende > Priority: Minor > Fix For: 0.6.0 > > > To continue improving NiFi's ability to collect logs, a good integration > point would be to have a processor that could listen for data from a Splunk > forwarder (https://docs.splunk.com/Splexicon:Universalforwarder). Being able > to push log messages to Splunk would also be useful. > Splunk provides an SDK that may be helpful: > https://github.com/splunk/splunk-sdk-java -- This message was sent by Atlassian JIRA (v6.3.4#6332)