tpalfy commented on a change in pull request #4714: URL: https://github.com/apache/nifi/pull/4714#discussion_r543488603
########## File path: nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/SplunkAPICall.java ########## @@ -0,0 +1,227 @@ +/* + * 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 com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.splunk.HttpException; +import com.splunk.RequestMessage; +import com.splunk.ResponseMessage; +import com.splunk.SSLSecurityProtocol; +import com.splunk.Service; +import com.splunk.ServiceArgs; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.AllowableValue; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.processor.AbstractProcessor; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.util.StandardValidators; + +import java.io.IOException; +import java.io.InputStream; +import java.util.Arrays; +import java.util.List; + +abstract class SplunkAPICall extends AbstractProcessor { + private static final String REQUEST_CHANNEL_HEADER_NAME = "X-Splunk-Request-Channel"; + + private static final String HTTP_SCHEME = "http"; + private static final String HTTPS_SCHEME = "https"; + + private static final AllowableValue TLS_1_2_VALUE = new AllowableValue(SSLSecurityProtocol.TLSv1_2.name(), SSLSecurityProtocol.TLSv1_2.name()); Review comment: Minor: `toString()` instead of `name()` could be more appropriate for `displayName`. ########## File path: nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/dto/splunk/SendRawDataResponse.java ########## @@ -0,0 +1,42 @@ +/* + * 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.dto.splunk; + +/** + * Response object for sending raw data directly to HTTP Event Collecto in Splunk. Review comment: Probably a typo: `Collecto` -> `Collector` ########## File path: nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunkHTTP.java ########## @@ -0,0 +1,281 @@ +/* + * 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 com.splunk.RequestMessage; +import com.splunk.ResponseMessage; +import org.apache.commons.io.IOUtils; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.ReadsAttribute; +import org.apache.nifi.annotation.behavior.SystemResource; +import org.apache.nifi.annotation.behavior.SystemResourceConsideration; +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.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.dto.splunk.SendRawDataResponse; +import org.apache.nifi.dto.splunk.SendRawDataSuccessResponse; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; + +import java.io.StringWriter; +import java.io.UnsupportedEncodingException; +import java.net.URLEncoder; +import java.nio.charset.Charset; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED) +@Tags({"splunk", "logs", "http"}) +@CapabilityDescription("Sends flow file content to the specified Splunk server over HTTP or HTTPS. Supports HEC Index Acknowledgement.") +@ReadsAttribute(attribute = "mime.type", description = "Uses as value for HTTP Content-Type header if set.") +@WritesAttributes({ + @WritesAttribute(attribute = "splunk.acknowledgement.id", description = "The indexing acknowledgement id provided by Splunk."), + @WritesAttribute(attribute = "splunk.send.at", description = "The time of sending the put request for Splunk.")}) +@SystemResourceConsideration(resource = SystemResource.MEMORY) +@SeeAlso(QuerySplunkIndexingStatus.class) +public class PutSplunkHTTP extends SplunkAPICall { + private static final String ENDPOINT = "/services/collector/raw"; + + static final PropertyDescriptor SOURCE = new PropertyDescriptor.Builder() + .name("source") + .displayName("Source") + .description("User-defined event source. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor SOURCE_TYPE = new PropertyDescriptor.Builder() + .name("source-type") + .displayName("Source Type") + .description("User-defined event sourcetype. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor HOST = new PropertyDescriptor.Builder() + .name("host") + .displayName("Host") + .description("Specify with the host query string parameter. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor INDEX = new PropertyDescriptor.Builder() + .name("index") + .displayName("Index") + .description("Index name. Specify with the index query string parameter. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder() + .name("character-set") + .displayName("Character Set") + .description("The name of the character set.") + .required(true) + .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR) + .defaultValue(Charset.defaultCharset().name()) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor CONTENT_TYPE = new PropertyDescriptor.Builder() + .name("content-type") + .displayName("Content Type") + .description( + "The media type of the event sent to Splunk. " + + "If not set, \"mime.type\" flow file attribute will be used. " + + "In case of neither of them is specified, this information will not be sent to the server.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final Relationship RELATIONSHIP_SUCCESS = new Relationship.Builder() + .name("success") + .description("FlowFiles that are sent successfully to the destination are sent out this relationship.") + .build(); + + static final Relationship RELATIONSHIP_FAILURE = new Relationship.Builder() + .name("failure") + .description("FlowFiles that failed to send to the destination are sent out this relationship.") + .build(); + + private static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(Arrays.asList( + RELATIONSHIP_SUCCESS, + RELATIONSHIP_FAILURE))); + + @Override + public Set<Relationship> getRelationships() { + return RELATIONSHIPS; + } + + @Override + public List<PropertyDescriptor> getSupportedPropertyDescriptors() { + final List<PropertyDescriptor> result = new ArrayList<>(super.getSupportedPropertyDescriptors()); + result.add(SOURCE); + result.add(SOURCE_TYPE); + result.add(HOST); + result.add(INDEX); + result.add(CONTENT_TYPE); + result.add(CHARSET); + return result; + } + + private volatile String endpoint; + private volatile String contentType; + private volatile String charset; + + @OnScheduled + public void onScheduled(final ProcessContext context) { + super.onScheduled(context); + + if (context.getProperty(CONTENT_TYPE).isSet()) { + contentType = context.getProperty(CONTENT_TYPE).evaluateAttributeExpressions().getValue(); + } + + charset = context.getProperty(CHARSET).evaluateAttributeExpressions().getValue(); + + final Map<String, String> queryParameters = new HashMap<>(); + + if (context.getProperty(SOURCE_TYPE).isSet()) { + queryParameters.put("sourcetype", context.getProperty(SOURCE_TYPE).evaluateAttributeExpressions().getValue()); + } + + if (context.getProperty(SOURCE).isSet()) { + queryParameters.put("source", context.getProperty(SOURCE).evaluateAttributeExpressions().getValue()); + } + + if (context.getProperty(HOST).isSet()) { + queryParameters.put("host", context.getProperty(HOST).evaluateAttributeExpressions().getValue()); + } + + if (context.getProperty(INDEX).isSet()) { + queryParameters.put("index", context.getProperty(INDEX).evaluateAttributeExpressions().getValue()); + } + + endpoint = getEndpoint(queryParameters); + } + + private String getEndpoint(final Map<String, String> queryParameters) { + if (queryParameters.isEmpty()) { + return ENDPOINT; + } + + try { + return URLEncoder.encode(ENDPOINT + '?' + queryParameters.entrySet().stream().map(e -> e.getKey() + '=' + e.getValue()).collect(Collectors.joining("&")), "UTF-8"); + } catch (final UnsupportedEncodingException e) { + getLogger().error("Could not be initialized because of: {}", new Object[] {e.getMessage()}, e); + throw new ProcessException(e); + } + } + + @OnStopped + public void onUnscheduled() { + super.onUnscheduled(); + contentType = null; + charset = null; + endpoint = null; + } + + @Override + public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { + FlowFile flowFile = session.get(); + boolean success = false; + + if (flowFile == null) { + return; + } + + try { + final RequestMessage requestMessage = createRequestMessage(session, flowFile); + final ResponseMessage responseMessage = call(endpoint, requestMessage); + flowFile = session.putAttribute(flowFile, "splunk.status.code", String.valueOf(responseMessage.getStatus())); + + switch (responseMessage.getStatus()) { + case 200: + final SendRawDataSuccessResponse successResponse = unmarshallResult(responseMessage.getContent(), SendRawDataSuccessResponse.class); + + if (successResponse.getCode() == 0) { + flowFile = enrichFlowFile(session, flowFile, successResponse.getAckId()); + success = true; + } else { + flowFile = session.putAttribute(flowFile, "splunk.response.code", String.valueOf(successResponse.getCode())); + getLogger().error("Putting data into Splunk was not successful: ({}) {}", new Object[] {successResponse.getCode(), successResponse.getText()}); + } + + break; + case 503 : // HEC is unhealthy, queues are full + context.yield(); + // fall-through + default: + final SendRawDataResponse response = unmarshallResult(responseMessage.getContent(), SendRawDataResponse.class); + getLogger().error("Putting data into Splunk was not successful: {}", new Object[] {response.getText()}); + } + } catch (final Exception e) { + getLogger().error("Error during communication with Splunk: {}", new Object[] {e.getMessage()}, e); + } finally { + session.transfer(flowFile, success ? RELATIONSHIP_SUCCESS : RELATIONSHIP_FAILURE); + } + } + + private RequestMessage createRequestMessage(final ProcessSession session, final FlowFile flowFile) { + final RequestMessage requestMessage = new RequestMessage("POST"); + final String flowFileContentType = Optional.ofNullable(contentType).orElse(flowFile.getAttribute("mime.type")); + + if (flowFileContentType != null) { + requestMessage.getHeader().put("Content-Type", flowFileContentType); + } + + requestMessage.setContent(extractTextMessageBody(flowFile, session, charset)); + return requestMessage; + } + + private String extractTextMessageBody(final FlowFile flowFile, final ProcessSession session, final String charset) { + final StringWriter writer = new StringWriter(); + session.read(flowFile, in -> IOUtils.copy(in, writer, Charset.forName(charset))); + return writer.toString(); + } + + private FlowFile enrichFlowFile(final ProcessSession session, final FlowFile flowFile, final long ackId) { + final Map<String, String> attributes = new HashMap<>(); + attributes.put(SplunkAPICall.ACKNOWLEDGEMENT_ID_ATTRIBUTE, String.valueOf(ackId)); + attributes.put(SplunkAPICall.SENT_AT_ATTRIBUTE, String.valueOf(System.currentTimeMillis())); Review comment: In reality this is more like when the response came back instead of when the message was sent. Not sure if the difference matters. Is this intentional? ########## File path: nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/QuerySplunkIndexingStatus.java ########## @@ -0,0 +1,236 @@ +/* + * 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 com.splunk.RequestMessage; +import com.splunk.ResponseMessage; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.ReadsAttribute; +import org.apache.nifi.annotation.behavior.ReadsAttributes; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.dto.splunk.EventIndexStatusRequest; +import org.apache.nifi.dto.splunk.EventIndexStatusResponse; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED) +@Tags({"splunk", "logs", "http", "acknowledgement"}) +@CapabilityDescription("Queries Splunk server in order to acquire the status of indexing acknowledgement.") +@ReadsAttributes({ + @ReadsAttribute(attribute = "splunk.acknowledgement.id", description = "The indexing acknowledgement id provided by Splunk."), + @ReadsAttribute(attribute = "splunk.send.at", description = "The time of sending the put request for Splunk.")}) +@SeeAlso(PutSplunkHTTP.class) +public class QuerySplunkIndexingStatus extends SplunkAPICall { + private static final String ENDPOINT = "/services/collector/ack"; + + static final Relationship RELATIONSHIP_ACKNOWLEDGED = new Relationship.Builder() + .name("success") + .description("A FlowFile is transferred into this relationship when the acknowledgement was successful.") + .build(); + + static final Relationship RELATIONSHIP_UNACKNOWLEDGED = new Relationship.Builder() + .name("unacknowledged") + .description("A FlowFile is transferred into this relationship when the acknowledgement was not successful.") Review comment: ```suggestion .description("A FlowFile is transferred to this relationship when the acknowledgement was not successful.") ``` ########## File path: nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunkHTTP.java ########## @@ -0,0 +1,281 @@ +/* + * 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 com.splunk.RequestMessage; +import com.splunk.ResponseMessage; +import org.apache.commons.io.IOUtils; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.ReadsAttribute; +import org.apache.nifi.annotation.behavior.SystemResource; +import org.apache.nifi.annotation.behavior.SystemResourceConsideration; +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.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.dto.splunk.SendRawDataResponse; +import org.apache.nifi.dto.splunk.SendRawDataSuccessResponse; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; + +import java.io.StringWriter; +import java.io.UnsupportedEncodingException; +import java.net.URLEncoder; +import java.nio.charset.Charset; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED) +@Tags({"splunk", "logs", "http"}) +@CapabilityDescription("Sends flow file content to the specified Splunk server over HTTP or HTTPS. Supports HEC Index Acknowledgement.") +@ReadsAttribute(attribute = "mime.type", description = "Uses as value for HTTP Content-Type header if set.") +@WritesAttributes({ + @WritesAttribute(attribute = "splunk.acknowledgement.id", description = "The indexing acknowledgement id provided by Splunk."), + @WritesAttribute(attribute = "splunk.send.at", description = "The time of sending the put request for Splunk.")}) +@SystemResourceConsideration(resource = SystemResource.MEMORY) +@SeeAlso(QuerySplunkIndexingStatus.class) +public class PutSplunkHTTP extends SplunkAPICall { + private static final String ENDPOINT = "/services/collector/raw"; + + static final PropertyDescriptor SOURCE = new PropertyDescriptor.Builder() + .name("source") + .displayName("Source") + .description("User-defined event source. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor SOURCE_TYPE = new PropertyDescriptor.Builder() + .name("source-type") + .displayName("Source Type") + .description("User-defined event sourcetype. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor HOST = new PropertyDescriptor.Builder() + .name("host") + .displayName("Host") + .description("Specify with the host query string parameter. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor INDEX = new PropertyDescriptor.Builder() + .name("index") + .displayName("Index") + .description("Index name. Specify with the index query string parameter. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder() + .name("character-set") + .displayName("Character Set") + .description("The name of the character set.") + .required(true) + .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR) + .defaultValue(Charset.defaultCharset().name()) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor CONTENT_TYPE = new PropertyDescriptor.Builder() + .name("content-type") + .displayName("Content Type") + .description( + "The media type of the event sent to Splunk. " + + "If not set, \"mime.type\" flow file attribute will be used. " + + "In case of neither of them is specified, this information will not be sent to the server.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final Relationship RELATIONSHIP_SUCCESS = new Relationship.Builder() + .name("success") + .description("FlowFiles that are sent successfully to the destination are sent out this relationship.") Review comment: ```suggestion .description("FlowFiles that are sent successfully to the destination are sent to this relationship.") ``` ########## File path: nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/QuerySplunkIndexingStatus.java ########## @@ -0,0 +1,236 @@ +/* + * 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 com.splunk.RequestMessage; +import com.splunk.ResponseMessage; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.ReadsAttribute; +import org.apache.nifi.annotation.behavior.ReadsAttributes; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.dto.splunk.EventIndexStatusRequest; +import org.apache.nifi.dto.splunk.EventIndexStatusResponse; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED) +@Tags({"splunk", "logs", "http", "acknowledgement"}) +@CapabilityDescription("Queries Splunk server in order to acquire the status of indexing acknowledgement.") +@ReadsAttributes({ + @ReadsAttribute(attribute = "splunk.acknowledgement.id", description = "The indexing acknowledgement id provided by Splunk."), + @ReadsAttribute(attribute = "splunk.send.at", description = "The time of sending the put request for Splunk.")}) +@SeeAlso(PutSplunkHTTP.class) +public class QuerySplunkIndexingStatus extends SplunkAPICall { + private static final String ENDPOINT = "/services/collector/ack"; + + static final Relationship RELATIONSHIP_ACKNOWLEDGED = new Relationship.Builder() + .name("success") + .description("A FlowFile is transferred into this relationship when the acknowledgement was successful.") Review comment: ```suggestion .description("A FlowFile is transferred to this relationship when the acknowledgement was successful.") ``` ########## File path: nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunkHTTP.java ########## @@ -0,0 +1,281 @@ +/* + * 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 com.splunk.RequestMessage; +import com.splunk.ResponseMessage; +import org.apache.commons.io.IOUtils; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.ReadsAttribute; +import org.apache.nifi.annotation.behavior.SystemResource; +import org.apache.nifi.annotation.behavior.SystemResourceConsideration; +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.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.dto.splunk.SendRawDataResponse; +import org.apache.nifi.dto.splunk.SendRawDataSuccessResponse; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; + +import java.io.StringWriter; +import java.io.UnsupportedEncodingException; +import java.net.URLEncoder; +import java.nio.charset.Charset; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED) +@Tags({"splunk", "logs", "http"}) +@CapabilityDescription("Sends flow file content to the specified Splunk server over HTTP or HTTPS. Supports HEC Index Acknowledgement.") +@ReadsAttribute(attribute = "mime.type", description = "Uses as value for HTTP Content-Type header if set.") +@WritesAttributes({ + @WritesAttribute(attribute = "splunk.acknowledgement.id", description = "The indexing acknowledgement id provided by Splunk."), + @WritesAttribute(attribute = "splunk.send.at", description = "The time of sending the put request for Splunk.")}) +@SystemResourceConsideration(resource = SystemResource.MEMORY) +@SeeAlso(QuerySplunkIndexingStatus.class) +public class PutSplunkHTTP extends SplunkAPICall { + private static final String ENDPOINT = "/services/collector/raw"; + + static final PropertyDescriptor SOURCE = new PropertyDescriptor.Builder() + .name("source") + .displayName("Source") + .description("User-defined event source. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor SOURCE_TYPE = new PropertyDescriptor.Builder() + .name("source-type") + .displayName("Source Type") + .description("User-defined event sourcetype. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor HOST = new PropertyDescriptor.Builder() + .name("host") + .displayName("Host") + .description("Specify with the host query string parameter. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor INDEX = new PropertyDescriptor.Builder() + .name("index") + .displayName("Index") + .description("Index name. Specify with the index query string parameter. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder() + .name("character-set") + .displayName("Character Set") + .description("The name of the character set.") + .required(true) + .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR) + .defaultValue(Charset.defaultCharset().name()) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor CONTENT_TYPE = new PropertyDescriptor.Builder() + .name("content-type") + .displayName("Content Type") + .description( + "The media type of the event sent to Splunk. " + + "If not set, \"mime.type\" flow file attribute will be used. " + + "In case of neither of them is specified, this information will not be sent to the server.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final Relationship RELATIONSHIP_SUCCESS = new Relationship.Builder() + .name("success") + .description("FlowFiles that are sent successfully to the destination are sent out this relationship.") + .build(); + + static final Relationship RELATIONSHIP_FAILURE = new Relationship.Builder() + .name("failure") + .description("FlowFiles that failed to send to the destination are sent out this relationship.") + .build(); + + private static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(Arrays.asList( + RELATIONSHIP_SUCCESS, + RELATIONSHIP_FAILURE))); + + @Override + public Set<Relationship> getRelationships() { + return RELATIONSHIPS; + } + + @Override + public List<PropertyDescriptor> getSupportedPropertyDescriptors() { + final List<PropertyDescriptor> result = new ArrayList<>(super.getSupportedPropertyDescriptors()); + result.add(SOURCE); + result.add(SOURCE_TYPE); + result.add(HOST); + result.add(INDEX); + result.add(CONTENT_TYPE); + result.add(CHARSET); + return result; + } + + private volatile String endpoint; + private volatile String contentType; + private volatile String charset; + + @OnScheduled + public void onScheduled(final ProcessContext context) { + super.onScheduled(context); + + if (context.getProperty(CONTENT_TYPE).isSet()) { + contentType = context.getProperty(CONTENT_TYPE).evaluateAttributeExpressions().getValue(); + } + + charset = context.getProperty(CHARSET).evaluateAttributeExpressions().getValue(); + + final Map<String, String> queryParameters = new HashMap<>(); + + if (context.getProperty(SOURCE_TYPE).isSet()) { + queryParameters.put("sourcetype", context.getProperty(SOURCE_TYPE).evaluateAttributeExpressions().getValue()); + } + + if (context.getProperty(SOURCE).isSet()) { + queryParameters.put("source", context.getProperty(SOURCE).evaluateAttributeExpressions().getValue()); + } + + if (context.getProperty(HOST).isSet()) { + queryParameters.put("host", context.getProperty(HOST).evaluateAttributeExpressions().getValue()); + } + + if (context.getProperty(INDEX).isSet()) { + queryParameters.put("index", context.getProperty(INDEX).evaluateAttributeExpressions().getValue()); + } + + endpoint = getEndpoint(queryParameters); + } + + private String getEndpoint(final Map<String, String> queryParameters) { + if (queryParameters.isEmpty()) { + return ENDPOINT; + } + + try { + return URLEncoder.encode(ENDPOINT + '?' + queryParameters.entrySet().stream().map(e -> e.getKey() + '=' + e.getValue()).collect(Collectors.joining("&")), "UTF-8"); + } catch (final UnsupportedEncodingException e) { + getLogger().error("Could not be initialized because of: {}", new Object[] {e.getMessage()}, e); + throw new ProcessException(e); + } + } + + @OnStopped + public void onUnscheduled() { + super.onUnscheduled(); + contentType = null; + charset = null; + endpoint = null; + } + + @Override + public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { + FlowFile flowFile = session.get(); + boolean success = false; + + if (flowFile == null) { + return; + } + + try { + final RequestMessage requestMessage = createRequestMessage(session, flowFile); + final ResponseMessage responseMessage = call(endpoint, requestMessage); + flowFile = session.putAttribute(flowFile, "splunk.status.code", String.valueOf(responseMessage.getStatus())); + + switch (responseMessage.getStatus()) { + case 200: + final SendRawDataSuccessResponse successResponse = unmarshallResult(responseMessage.getContent(), SendRawDataSuccessResponse.class); + + if (successResponse.getCode() == 0) { + flowFile = enrichFlowFile(session, flowFile, successResponse.getAckId()); + success = true; + } else { + flowFile = session.putAttribute(flowFile, "splunk.response.code", String.valueOf(successResponse.getCode())); + getLogger().error("Putting data into Splunk was not successful: ({}) {}", new Object[] {successResponse.getCode(), successResponse.getText()}); + } + + break; + case 503 : // HEC is unhealthy, queues are full + context.yield(); + // fall-through + default: + final SendRawDataResponse response = unmarshallResult(responseMessage.getContent(), SendRawDataResponse.class); + getLogger().error("Putting data into Splunk was not successful: {}", new Object[] {response.getText()}); + } + } catch (final Exception e) { + getLogger().error("Error during communication with Splunk: {}", new Object[] {e.getMessage()}, e); + } finally { + session.transfer(flowFile, success ? RELATIONSHIP_SUCCESS : RELATIONSHIP_FAILURE); + } + } + + private RequestMessage createRequestMessage(final ProcessSession session, final FlowFile flowFile) { + final RequestMessage requestMessage = new RequestMessage("POST"); + final String flowFileContentType = Optional.ofNullable(contentType).orElse(flowFile.getAttribute("mime.type")); + + if (flowFileContentType != null) { + requestMessage.getHeader().put("Content-Type", flowFileContentType); + } + + requestMessage.setContent(extractTextMessageBody(flowFile, session, charset)); Review comment: There is a `requestMessage.setContent` that accepts an `OutputStream` instead of a `String`. Seems a better approach considering the typical use-cases of NiFi. ########## File path: nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/QuerySplunkIndexingStatus.java ########## @@ -0,0 +1,236 @@ +/* + * 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 com.splunk.RequestMessage; +import com.splunk.ResponseMessage; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.ReadsAttribute; +import org.apache.nifi.annotation.behavior.ReadsAttributes; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.dto.splunk.EventIndexStatusRequest; +import org.apache.nifi.dto.splunk.EventIndexStatusResponse; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED) +@Tags({"splunk", "logs", "http", "acknowledgement"}) +@CapabilityDescription("Queries Splunk server in order to acquire the status of indexing acknowledgement.") +@ReadsAttributes({ + @ReadsAttribute(attribute = "splunk.acknowledgement.id", description = "The indexing acknowledgement id provided by Splunk."), + @ReadsAttribute(attribute = "splunk.send.at", description = "The time of sending the put request for Splunk.")}) +@SeeAlso(PutSplunkHTTP.class) +public class QuerySplunkIndexingStatus extends SplunkAPICall { + private static final String ENDPOINT = "/services/collector/ack"; + + static final Relationship RELATIONSHIP_ACKNOWLEDGED = new Relationship.Builder() + .name("success") + .description("A FlowFile is transferred into this relationship when the acknowledgement was successful.") + .build(); + + static final Relationship RELATIONSHIP_UNACKNOWLEDGED = new Relationship.Builder() + .name("unacknowledged") + .description("A FlowFile is transferred into this relationship when the acknowledgement was not successful.") + .build(); + + static final Relationship RELATIONSHIP_UNDETERMINED = new Relationship.Builder() + .name("undetermined") + .description( + "A FlowFile is transferred into this relationship when the acknowledgement state is not determined. " + + "Flow files transferred into this relationship might be penalized! " + + "This happens when Splunk returns with HTTP 200 but with false response for the acknowledgement id in the flow file attribute.") + .build(); + + static final Relationship RELATIONSHIP_FAILURE = new Relationship.Builder() + .name("failure") + .description("A FlowFile is transferred into this relationship when the acknowledgement was not successful.") Review comment: Could we add some clarification how this differs from `unacknowledged`? Refer to `Maximum Waiting Time` property with a simple explanation for example. ########## File path: nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/QuerySplunkIndexingStatus.java ########## @@ -0,0 +1,236 @@ +/* + * 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 com.splunk.RequestMessage; +import com.splunk.ResponseMessage; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.ReadsAttribute; +import org.apache.nifi.annotation.behavior.ReadsAttributes; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.dto.splunk.EventIndexStatusRequest; +import org.apache.nifi.dto.splunk.EventIndexStatusResponse; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED) +@Tags({"splunk", "logs", "http", "acknowledgement"}) +@CapabilityDescription("Queries Splunk server in order to acquire the status of indexing acknowledgement.") +@ReadsAttributes({ + @ReadsAttribute(attribute = "splunk.acknowledgement.id", description = "The indexing acknowledgement id provided by Splunk."), + @ReadsAttribute(attribute = "splunk.send.at", description = "The time of sending the put request for Splunk.")}) +@SeeAlso(PutSplunkHTTP.class) +public class QuerySplunkIndexingStatus extends SplunkAPICall { + private static final String ENDPOINT = "/services/collector/ack"; + + static final Relationship RELATIONSHIP_ACKNOWLEDGED = new Relationship.Builder() + .name("success") + .description("A FlowFile is transferred into this relationship when the acknowledgement was successful.") + .build(); + + static final Relationship RELATIONSHIP_UNACKNOWLEDGED = new Relationship.Builder() + .name("unacknowledged") + .description("A FlowFile is transferred into this relationship when the acknowledgement was not successful.") + .build(); + + static final Relationship RELATIONSHIP_UNDETERMINED = new Relationship.Builder() + .name("undetermined") + .description( + "A FlowFile is transferred into this relationship when the acknowledgement state is not determined. " + + "Flow files transferred into this relationship might be penalized! " + + "This happens when Splunk returns with HTTP 200 but with false response for the acknowledgement id in the flow file attribute.") + .build(); + + static final Relationship RELATIONSHIP_FAILURE = new Relationship.Builder() + .name("failure") + .description("A FlowFile is transferred into this relationship when the acknowledgement was not successful.") + .build(); + + private static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(Arrays.asList( + RELATIONSHIP_ACKNOWLEDGED, + RELATIONSHIP_UNACKNOWLEDGED, + RELATIONSHIP_UNDETERMINED, + RELATIONSHIP_FAILURE + ))); + + static final PropertyDescriptor TTL = new PropertyDescriptor.Builder() + .name("ttl") + .displayName("Maximum Waiting Time") + .description( + "The maximum time the service tries to acquire acknowledgement confirmation for an index, from the point of registration. " + + "After the given amount of time, the service considers the index as not acknowledged and moves it into the output buffer as failed acknowledgement.") + .defaultValue("1 hour") + .required(true) + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + + static final PropertyDescriptor MAX_QUERY_SIZE = new PropertyDescriptor.Builder() + .name("max-query-size") + .displayName("Maximum Query Size") + .description( + "The maximum number of acknowledgement identifiers the service query status for in one batch. " + Review comment: This sentence seems to lack a predicate. Maybe ```suggestion "The maximum number of acknowledgement identifiers the service query status(?) contains in one batch. " + ``` ########## File path: nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/QuerySplunkIndexingStatus.java ########## @@ -0,0 +1,236 @@ +/* + * 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 com.splunk.RequestMessage; +import com.splunk.ResponseMessage; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.ReadsAttribute; +import org.apache.nifi.annotation.behavior.ReadsAttributes; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.dto.splunk.EventIndexStatusRequest; +import org.apache.nifi.dto.splunk.EventIndexStatusResponse; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED) +@Tags({"splunk", "logs", "http", "acknowledgement"}) +@CapabilityDescription("Queries Splunk server in order to acquire the status of indexing acknowledgement.") +@ReadsAttributes({ + @ReadsAttribute(attribute = "splunk.acknowledgement.id", description = "The indexing acknowledgement id provided by Splunk."), + @ReadsAttribute(attribute = "splunk.send.at", description = "The time of sending the put request for Splunk.")}) +@SeeAlso(PutSplunkHTTP.class) +public class QuerySplunkIndexingStatus extends SplunkAPICall { + private static final String ENDPOINT = "/services/collector/ack"; + + static final Relationship RELATIONSHIP_ACKNOWLEDGED = new Relationship.Builder() + .name("success") + .description("A FlowFile is transferred into this relationship when the acknowledgement was successful.") + .build(); + + static final Relationship RELATIONSHIP_UNACKNOWLEDGED = new Relationship.Builder() + .name("unacknowledged") + .description("A FlowFile is transferred into this relationship when the acknowledgement was not successful.") + .build(); + + static final Relationship RELATIONSHIP_UNDETERMINED = new Relationship.Builder() + .name("undetermined") + .description( + "A FlowFile is transferred into this relationship when the acknowledgement state is not determined. " + + "Flow files transferred into this relationship might be penalized! " + Review comment: ```suggestion "A FlowFile is transferred to this relationship when the acknowledgement state is not determined. " + "Flow files transferred to this relationship might be penalized! " + ``` ########## File path: nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/QuerySplunkIndexingStatus.java ########## @@ -0,0 +1,236 @@ +/* + * 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 com.splunk.RequestMessage; +import com.splunk.ResponseMessage; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.ReadsAttribute; +import org.apache.nifi.annotation.behavior.ReadsAttributes; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.dto.splunk.EventIndexStatusRequest; +import org.apache.nifi.dto.splunk.EventIndexStatusResponse; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED) +@Tags({"splunk", "logs", "http", "acknowledgement"}) +@CapabilityDescription("Queries Splunk server in order to acquire the status of indexing acknowledgement.") +@ReadsAttributes({ + @ReadsAttribute(attribute = "splunk.acknowledgement.id", description = "The indexing acknowledgement id provided by Splunk."), + @ReadsAttribute(attribute = "splunk.send.at", description = "The time of sending the put request for Splunk.")}) +@SeeAlso(PutSplunkHTTP.class) +public class QuerySplunkIndexingStatus extends SplunkAPICall { + private static final String ENDPOINT = "/services/collector/ack"; + + static final Relationship RELATIONSHIP_ACKNOWLEDGED = new Relationship.Builder() + .name("success") + .description("A FlowFile is transferred into this relationship when the acknowledgement was successful.") + .build(); + + static final Relationship RELATIONSHIP_UNACKNOWLEDGED = new Relationship.Builder() + .name("unacknowledged") + .description("A FlowFile is transferred into this relationship when the acknowledgement was not successful.") + .build(); + + static final Relationship RELATIONSHIP_UNDETERMINED = new Relationship.Builder() + .name("undetermined") + .description( + "A FlowFile is transferred into this relationship when the acknowledgement state is not determined. " + + "Flow files transferred into this relationship might be penalized! " + + "This happens when Splunk returns with HTTP 200 but with false response for the acknowledgement id in the flow file attribute.") + .build(); + + static final Relationship RELATIONSHIP_FAILURE = new Relationship.Builder() + .name("failure") + .description("A FlowFile is transferred into this relationship when the acknowledgement was not successful.") + .build(); + + private static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(Arrays.asList( + RELATIONSHIP_ACKNOWLEDGED, + RELATIONSHIP_UNACKNOWLEDGED, + RELATIONSHIP_UNDETERMINED, + RELATIONSHIP_FAILURE + ))); + + static final PropertyDescriptor TTL = new PropertyDescriptor.Builder() + .name("ttl") + .displayName("Maximum Waiting Time") + .description( + "The maximum time the service tries to acquire acknowledgement confirmation for an index, from the point of registration. " + + "After the given amount of time, the service considers the index as not acknowledged and moves it into the output buffer as failed acknowledgement.") + .defaultValue("1 hour") + .required(true) + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + + static final PropertyDescriptor MAX_QUERY_SIZE = new PropertyDescriptor.Builder() + .name("max-query-size") + .displayName("Maximum Query Size") + .description( + "The maximum number of acknowledgement identifiers the service query status for in one batch. " + + "It is suggested to not set it too low in order to reduce network communication.") Review comment: ```suggestion "It is recommended not to set it too low in order to reduce network communication.") ``` ########## File path: nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/QuerySplunkIndexingStatus.java ########## @@ -0,0 +1,236 @@ +/* + * 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 com.splunk.RequestMessage; +import com.splunk.ResponseMessage; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.ReadsAttribute; +import org.apache.nifi.annotation.behavior.ReadsAttributes; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.dto.splunk.EventIndexStatusRequest; +import org.apache.nifi.dto.splunk.EventIndexStatusResponse; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED) +@Tags({"splunk", "logs", "http", "acknowledgement"}) +@CapabilityDescription("Queries Splunk server in order to acquire the status of indexing acknowledgement.") +@ReadsAttributes({ + @ReadsAttribute(attribute = "splunk.acknowledgement.id", description = "The indexing acknowledgement id provided by Splunk."), + @ReadsAttribute(attribute = "splunk.send.at", description = "The time of sending the put request for Splunk.")}) +@SeeAlso(PutSplunkHTTP.class) +public class QuerySplunkIndexingStatus extends SplunkAPICall { + private static final String ENDPOINT = "/services/collector/ack"; + + static final Relationship RELATIONSHIP_ACKNOWLEDGED = new Relationship.Builder() + .name("success") + .description("A FlowFile is transferred into this relationship when the acknowledgement was successful.") + .build(); + + static final Relationship RELATIONSHIP_UNACKNOWLEDGED = new Relationship.Builder() + .name("unacknowledged") + .description("A FlowFile is transferred into this relationship when the acknowledgement was not successful.") + .build(); + + static final Relationship RELATIONSHIP_UNDETERMINED = new Relationship.Builder() + .name("undetermined") + .description( + "A FlowFile is transferred into this relationship when the acknowledgement state is not determined. " + + "Flow files transferred into this relationship might be penalized! " + + "This happens when Splunk returns with HTTP 200 but with false response for the acknowledgement id in the flow file attribute.") + .build(); + + static final Relationship RELATIONSHIP_FAILURE = new Relationship.Builder() + .name("failure") + .description("A FlowFile is transferred into this relationship when the acknowledgement was not successful.") + .build(); + + private static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(Arrays.asList( + RELATIONSHIP_ACKNOWLEDGED, + RELATIONSHIP_UNACKNOWLEDGED, + RELATIONSHIP_UNDETERMINED, + RELATIONSHIP_FAILURE + ))); + + static final PropertyDescriptor TTL = new PropertyDescriptor.Builder() + .name("ttl") + .displayName("Maximum Waiting Time") + .description( + "The maximum time the service tries to acquire acknowledgement confirmation for an index, from the point of registration. " + + "After the given amount of time, the service considers the index as not acknowledged and moves it into the output buffer as failed acknowledgement.") + .defaultValue("1 hour") + .required(true) + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + + static final PropertyDescriptor MAX_QUERY_SIZE = new PropertyDescriptor.Builder() + .name("max-query-size") + .displayName("Maximum Query Size") + .description( + "The maximum number of acknowledgement identifiers the service query status for in one batch. " + + "It is suggested to not set it too low in order to reduce network communication.") + .defaultValue("10000") + .required(true) + .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR) + .build(); + + private volatile Integer maxQuerySize; + private volatile Integer ttl; + + @Override + public List<PropertyDescriptor> getSupportedPropertyDescriptors() { + final List<PropertyDescriptor> result = new ArrayList<>(); + final List<PropertyDescriptor> common = super.getSupportedPropertyDescriptors(); + result.addAll(common); + result.add(TTL); + result.add(MAX_QUERY_SIZE); + return result; + } + + @Override + public Set<Relationship> getRelationships() { + return RELATIONSHIPS; + } + + @OnScheduled + public void onScheduled(final ProcessContext context) { + super.onScheduled(context); + maxQuerySize = context.getProperty(MAX_QUERY_SIZE).asInteger(); + ttl = context.getProperty(TTL).asTimePeriod(TimeUnit.MILLISECONDS).intValue(); + } + + @OnStopped + public void onUnscheduled() { + super.onUnscheduled(); + maxQuerySize = null; + ttl = null; + } + + @Override + public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { + final RequestMessage requestMessage; + final List<FlowFile> flowFiles = session.get(maxQuerySize); + + if (flowFiles.isEmpty()) { + return; + } + + final long currentTime = System.currentTimeMillis(); + final Map<Long, FlowFile> undetermined = new HashMap<>(); + + for (final FlowFile flowFile : flowFiles) { + final Optional<Long> sentAt = extractLong(flowFile.getAttribute(SplunkAPICall.SENT_AT_ATTRIBUTE)); + final Optional<Long> ackId = extractLong(flowFile.getAttribute(SplunkAPICall.ACKNOWLEDGEMENT_ID_ATTRIBUTE)); + + if (!sentAt.isPresent() || !ackId.isPresent()) { + getLogger().error("Flow file ({}) attributes {} and {} are expected to be set using 64-bit integer values!", + new Object[]{flowFile.getId(), SplunkAPICall.SENT_AT_ATTRIBUTE, SplunkAPICall.ACKNOWLEDGEMENT_ID_ATTRIBUTE}); + session.transfer(flowFile, RELATIONSHIP_FAILURE); + } else if (sentAt.get() + ttl < currentTime) { + session.transfer(flowFile, RELATIONSHIP_UNACKNOWLEDGED); + } else { + undetermined.put(ackId.get(), flowFile); + } + } + + if (undetermined.isEmpty()) { + getLogger().debug("There was no eligible flow file to send request to Splunk."); + return; + } + + try { + requestMessage = createRequestMessage(undetermined); + } catch (final IOException e) { + getLogger().error("Could not prepare Splunk request!", e); + session.transfer(undetermined.values(), RELATIONSHIP_FAILURE); + return; + } + + try { + final ResponseMessage responseMessage = call(ENDPOINT, requestMessage); + + if (responseMessage.getStatus() == 200) { + final EventIndexStatusResponse splunkResponse = unmarshallResult(responseMessage.getContent(), EventIndexStatusResponse.class); + + splunkResponse.getAcks().entrySet().forEach(result -> { + final FlowFile toTransfer = undetermined.get(result.getKey()); + + if (result.getValue()) { + session.transfer(toTransfer, RELATIONSHIP_ACKNOWLEDGED); + } else { + session.penalize(toTransfer); + session.transfer(toTransfer, RELATIONSHIP_UNDETERMINED); + } + }); + } else { + getLogger().error("Query index status was not successful because of ({}) {}", new Object[] {responseMessage.getStatus(), responseMessage.getContent()}); + context.yield(); + session.transfer(undetermined.values(), RELATIONSHIP_UNDETERMINED); + } + } catch (final Exception e) { + getLogger().error("Error during communication with Splunk server!", e); + session.transfer(undetermined.values(), RELATIONSHIP_FAILURE); + } + } + + private RequestMessage createRequestMessage(Map<Long, FlowFile> undetermined) throws IOException { + final RequestMessage requestMessage = new RequestMessage("POST"); + requestMessage.getHeader().put("Content-Type", "application/json"); + requestMessage.setContent(generateContent(undetermined)); + return requestMessage; + } + + private String generateContent(final Map<Long, FlowFile> undetermined) throws IOException { + final EventIndexStatusRequest splunkRequest = new EventIndexStatusRequest(); + splunkRequest.setAcks(new ArrayList<>(undetermined.keySet())); + return marshalRequest(splunkRequest); + } + + private static Optional<Long> extractLong(final String value) { + if (value == null) { + return Optional.empty(); + } + + try { + return Optional.of(Long.valueOf(value)); + } catch (final NumberFormatException e) { + return Optional.empty(); + } Review comment: ```suggestion try { return Optional.ofNullable(value).map(Long::valueOf); } catch (final NumberFormatException e) { return Optional.empty(); } ``` ########## File path: nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/QuerySplunkIndexingStatus.java ########## @@ -0,0 +1,236 @@ +/* + * 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 com.splunk.RequestMessage; +import com.splunk.ResponseMessage; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.ReadsAttribute; +import org.apache.nifi.annotation.behavior.ReadsAttributes; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.dto.splunk.EventIndexStatusRequest; +import org.apache.nifi.dto.splunk.EventIndexStatusResponse; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED) +@Tags({"splunk", "logs", "http", "acknowledgement"}) +@CapabilityDescription("Queries Splunk server in order to acquire the status of indexing acknowledgement.") +@ReadsAttributes({ + @ReadsAttribute(attribute = "splunk.acknowledgement.id", description = "The indexing acknowledgement id provided by Splunk."), + @ReadsAttribute(attribute = "splunk.send.at", description = "The time of sending the put request for Splunk.")}) +@SeeAlso(PutSplunkHTTP.class) +public class QuerySplunkIndexingStatus extends SplunkAPICall { + private static final String ENDPOINT = "/services/collector/ack"; + + static final Relationship RELATIONSHIP_ACKNOWLEDGED = new Relationship.Builder() + .name("success") + .description("A FlowFile is transferred into this relationship when the acknowledgement was successful.") + .build(); + + static final Relationship RELATIONSHIP_UNACKNOWLEDGED = new Relationship.Builder() + .name("unacknowledged") + .description("A FlowFile is transferred into this relationship when the acknowledgement was not successful.") + .build(); + + static final Relationship RELATIONSHIP_UNDETERMINED = new Relationship.Builder() + .name("undetermined") + .description( + "A FlowFile is transferred into this relationship when the acknowledgement state is not determined. " + + "Flow files transferred into this relationship might be penalized! " + + "This happens when Splunk returns with HTTP 200 but with false response for the acknowledgement id in the flow file attribute.") + .build(); + + static final Relationship RELATIONSHIP_FAILURE = new Relationship.Builder() + .name("failure") + .description("A FlowFile is transferred into this relationship when the acknowledgement was not successful.") + .build(); + + private static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(Arrays.asList( + RELATIONSHIP_ACKNOWLEDGED, + RELATIONSHIP_UNACKNOWLEDGED, + RELATIONSHIP_UNDETERMINED, + RELATIONSHIP_FAILURE + ))); + + static final PropertyDescriptor TTL = new PropertyDescriptor.Builder() + .name("ttl") + .displayName("Maximum Waiting Time") + .description( + "The maximum time the service tries to acquire acknowledgement confirmation for an index, from the point of registration. " + + "After the given amount of time, the service considers the index as not acknowledged and moves it into the output buffer as failed acknowledgement.") Review comment: Not really clear what `output buffer` means here. Is it the `failure` relationship itself? If that's the case, why not state that instead? ########## File path: nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunkHTTP.java ########## @@ -0,0 +1,281 @@ +/* + * 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 com.splunk.RequestMessage; +import com.splunk.ResponseMessage; +import org.apache.commons.io.IOUtils; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.ReadsAttribute; +import org.apache.nifi.annotation.behavior.SystemResource; +import org.apache.nifi.annotation.behavior.SystemResourceConsideration; +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.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.dto.splunk.SendRawDataResponse; +import org.apache.nifi.dto.splunk.SendRawDataSuccessResponse; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; + +import java.io.StringWriter; +import java.io.UnsupportedEncodingException; +import java.net.URLEncoder; +import java.nio.charset.Charset; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED) +@Tags({"splunk", "logs", "http"}) +@CapabilityDescription("Sends flow file content to the specified Splunk server over HTTP or HTTPS. Supports HEC Index Acknowledgement.") +@ReadsAttribute(attribute = "mime.type", description = "Uses as value for HTTP Content-Type header if set.") +@WritesAttributes({ + @WritesAttribute(attribute = "splunk.acknowledgement.id", description = "The indexing acknowledgement id provided by Splunk."), + @WritesAttribute(attribute = "splunk.send.at", description = "The time of sending the put request for Splunk.")}) +@SystemResourceConsideration(resource = SystemResource.MEMORY) +@SeeAlso(QuerySplunkIndexingStatus.class) +public class PutSplunkHTTP extends SplunkAPICall { + private static final String ENDPOINT = "/services/collector/raw"; + + static final PropertyDescriptor SOURCE = new PropertyDescriptor.Builder() + .name("source") + .displayName("Source") + .description("User-defined event source. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor SOURCE_TYPE = new PropertyDescriptor.Builder() + .name("source-type") + .displayName("Source Type") + .description("User-defined event sourcetype. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor HOST = new PropertyDescriptor.Builder() + .name("host") + .displayName("Host") + .description("Specify with the host query string parameter. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor INDEX = new PropertyDescriptor.Builder() + .name("index") + .displayName("Index") + .description("Index name. Specify with the index query string parameter. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder() + .name("character-set") + .displayName("Character Set") + .description("The name of the character set.") + .required(true) + .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR) + .defaultValue(Charset.defaultCharset().name()) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor CONTENT_TYPE = new PropertyDescriptor.Builder() + .name("content-type") + .displayName("Content Type") + .description( + "The media type of the event sent to Splunk. " + + "If not set, \"mime.type\" flow file attribute will be used. " + + "In case of neither of them is specified, this information will not be sent to the server.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final Relationship RELATIONSHIP_SUCCESS = new Relationship.Builder() + .name("success") + .description("FlowFiles that are sent successfully to the destination are sent out this relationship.") + .build(); + + static final Relationship RELATIONSHIP_FAILURE = new Relationship.Builder() + .name("failure") + .description("FlowFiles that failed to send to the destination are sent out this relationship.") Review comment: ```suggestion .description("FlowFiles that failed to send to the destination are sent to this relationship.") ``` ########## File path: nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunkHTTP.java ########## @@ -0,0 +1,281 @@ +/* + * 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 com.splunk.RequestMessage; +import com.splunk.ResponseMessage; +import org.apache.commons.io.IOUtils; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.ReadsAttribute; +import org.apache.nifi.annotation.behavior.SystemResource; +import org.apache.nifi.annotation.behavior.SystemResourceConsideration; +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.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.dto.splunk.SendRawDataResponse; +import org.apache.nifi.dto.splunk.SendRawDataSuccessResponse; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; + +import java.io.StringWriter; +import java.io.UnsupportedEncodingException; +import java.net.URLEncoder; +import java.nio.charset.Charset; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED) +@Tags({"splunk", "logs", "http"}) +@CapabilityDescription("Sends flow file content to the specified Splunk server over HTTP or HTTPS. Supports HEC Index Acknowledgement.") +@ReadsAttribute(attribute = "mime.type", description = "Uses as value for HTTP Content-Type header if set.") +@WritesAttributes({ + @WritesAttribute(attribute = "splunk.acknowledgement.id", description = "The indexing acknowledgement id provided by Splunk."), + @WritesAttribute(attribute = "splunk.send.at", description = "The time of sending the put request for Splunk.")}) +@SystemResourceConsideration(resource = SystemResource.MEMORY) +@SeeAlso(QuerySplunkIndexingStatus.class) +public class PutSplunkHTTP extends SplunkAPICall { + private static final String ENDPOINT = "/services/collector/raw"; + + static final PropertyDescriptor SOURCE = new PropertyDescriptor.Builder() + .name("source") + .displayName("Source") + .description("User-defined event source. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor SOURCE_TYPE = new PropertyDescriptor.Builder() + .name("source-type") + .displayName("Source Type") + .description("User-defined event sourcetype. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor HOST = new PropertyDescriptor.Builder() + .name("host") + .displayName("Host") + .description("Specify with the host query string parameter. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor INDEX = new PropertyDescriptor.Builder() + .name("index") + .displayName("Index") + .description("Index name. Specify with the index query string parameter. Sets a default for all events when unspecified.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder() + .name("character-set") + .displayName("Character Set") + .description("The name of the character set.") + .required(true) + .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR) + .defaultValue(Charset.defaultCharset().name()) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final PropertyDescriptor CONTENT_TYPE = new PropertyDescriptor.Builder() + .name("content-type") + .displayName("Content Type") + .description( + "The media type of the event sent to Splunk. " + + "If not set, \"mime.type\" flow file attribute will be used. " + + "In case of neither of them is specified, this information will not be sent to the server.") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + static final Relationship RELATIONSHIP_SUCCESS = new Relationship.Builder() + .name("success") + .description("FlowFiles that are sent successfully to the destination are sent out this relationship.") + .build(); + + static final Relationship RELATIONSHIP_FAILURE = new Relationship.Builder() + .name("failure") + .description("FlowFiles that failed to send to the destination are sent out this relationship.") + .build(); + + private static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(Arrays.asList( + RELATIONSHIP_SUCCESS, + RELATIONSHIP_FAILURE))); + + @Override + public Set<Relationship> getRelationships() { + return RELATIONSHIPS; + } + + @Override + public List<PropertyDescriptor> getSupportedPropertyDescriptors() { + final List<PropertyDescriptor> result = new ArrayList<>(super.getSupportedPropertyDescriptors()); + result.add(SOURCE); + result.add(SOURCE_TYPE); + result.add(HOST); + result.add(INDEX); + result.add(CONTENT_TYPE); + result.add(CHARSET); + return result; + } + + private volatile String endpoint; + private volatile String contentType; + private volatile String charset; + + @OnScheduled + public void onScheduled(final ProcessContext context) { + super.onScheduled(context); + + if (context.getProperty(CONTENT_TYPE).isSet()) { + contentType = context.getProperty(CONTENT_TYPE).evaluateAttributeExpressions().getValue(); + } + + charset = context.getProperty(CHARSET).evaluateAttributeExpressions().getValue(); + + final Map<String, String> queryParameters = new HashMap<>(); + + if (context.getProperty(SOURCE_TYPE).isSet()) { + queryParameters.put("sourcetype", context.getProperty(SOURCE_TYPE).evaluateAttributeExpressions().getValue()); + } + + if (context.getProperty(SOURCE).isSet()) { + queryParameters.put("source", context.getProperty(SOURCE).evaluateAttributeExpressions().getValue()); + } + + if (context.getProperty(HOST).isSet()) { + queryParameters.put("host", context.getProperty(HOST).evaluateAttributeExpressions().getValue()); + } + + if (context.getProperty(INDEX).isSet()) { + queryParameters.put("index", context.getProperty(INDEX).evaluateAttributeExpressions().getValue()); + } + + endpoint = getEndpoint(queryParameters); + } + + private String getEndpoint(final Map<String, String> queryParameters) { + if (queryParameters.isEmpty()) { + return ENDPOINT; + } + + try { + return URLEncoder.encode(ENDPOINT + '?' + queryParameters.entrySet().stream().map(e -> e.getKey() + '=' + e.getValue()).collect(Collectors.joining("&")), "UTF-8"); + } catch (final UnsupportedEncodingException e) { + getLogger().error("Could not be initialized because of: {}", new Object[] {e.getMessage()}, e); + throw new ProcessException(e); + } + } + + @OnStopped + public void onUnscheduled() { + super.onUnscheduled(); + contentType = null; + charset = null; + endpoint = null; + } + + @Override + public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { + FlowFile flowFile = session.get(); + boolean success = false; + + if (flowFile == null) { + return; + } + + try { + final RequestMessage requestMessage = createRequestMessage(session, flowFile); + final ResponseMessage responseMessage = call(endpoint, requestMessage); + flowFile = session.putAttribute(flowFile, "splunk.status.code", String.valueOf(responseMessage.getStatus())); + + switch (responseMessage.getStatus()) { + case 200: + final SendRawDataSuccessResponse successResponse = unmarshallResult(responseMessage.getContent(), SendRawDataSuccessResponse.class); + + if (successResponse.getCode() == 0) { + flowFile = enrichFlowFile(session, flowFile, successResponse.getAckId()); + success = true; + } else { + flowFile = session.putAttribute(flowFile, "splunk.response.code", String.valueOf(successResponse.getCode())); + getLogger().error("Putting data into Splunk was not successful: ({}) {}", new Object[] {successResponse.getCode(), successResponse.getText()}); + } + + break; + case 503 : // HEC is unhealthy, queues are full + context.yield(); + // fall-through + default: + final SendRawDataResponse response = unmarshallResult(responseMessage.getContent(), SendRawDataResponse.class); + getLogger().error("Putting data into Splunk was not successful: {}", new Object[] {response.getText()}); + } + } catch (final Exception e) { Review comment: Suggestion: A main cause of exception could be being unable to unmarshall the result. It may be a good idea to log the response content in that case. Note: For that we would need to know if the unmarshall was successful. Response could be declared before the switch block and checked if it is null here - however response can be of two different type. Not entirely sure why it is better to have two types instead of one though. A single type with nullable `ackId` could work. ---------------------------------------------------------------- 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: [email protected]
