szaszm commented on a change in pull request #1219: URL: https://github.com/apache/nifi-minifi-cpp/pull/1219#discussion_r767729620
########## File path: extensions/splunk/PutSplunkHTTP.h ########## @@ -0,0 +1,54 @@ +/** + * 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. + */ + +#pragma once + +#include <string> +#include <memory> + +#include "SplunkHECProcessor.h" +#include "utils/gsl.h" + +namespace org::apache::nifi::minifi::extensions::splunk { + +class PutSplunkHTTP final : public SplunkHECProcessor { Review comment: Please specify `InputRequirements` by overriding `getInputRequirements`. ########## File path: extensions/splunk/QuerySplunkIndexingStatus.cpp ########## @@ -0,0 +1,194 @@ +/** + * 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. + */ + + +#include "QuerySplunkIndexingStatus.h" + +#include <unordered_map> +#include <utility> + +#include "SplunkAttributes.h" + +#include "core/Resource.h" +#include "client/HTTPClient.h" +#include "utils/HTTPClient.h" +#include "utils/TimeUtil.h" + +#include "rapidjson/document.h" +#include "rapidjson/stringbuffer.h" +#include "rapidjson/writer.h" + +namespace org::apache::nifi::minifi::extensions::splunk { + +const core::Property QuerySplunkIndexingStatus::MaximumWaitingTime(core::PropertyBuilder::createProperty("Maximum Waiting Time") + ->withDescription("The maximum time the processor tries to acquire acknowledgement confirmation for an index, from the point of registration. " + "After the given amount of time, the processor considers the index as not acknowledged and transfers the FlowFile to the \"unacknowledged\" relationship.") + ->withDefaultValue("1 hour")->isRequired(true)->build()); + +const core::Property QuerySplunkIndexingStatus::MaxQuerySize(core::PropertyBuilder::createProperty("Maximum Query Size") + ->withDescription("The maximum number of acknowledgement identifiers the outgoing query contains in one batch. " + "It is recommended not to set it too low in order to reduce network communication.") + ->withDefaultValue("1000")->isRequired(true)->build()); + +const core::Relationship QuerySplunkIndexingStatus::Acknowledged("acknowledged", + "A FlowFile is transferred to this relationship when the acknowledgement was successful."); + +const core::Relationship QuerySplunkIndexingStatus::Unacknowledged("unacknowledged", + "A FlowFile is transferred to this relationship when the acknowledgement was not successful. " + "This can happen when the acknowledgement did not happened within the time period set for Maximum Waiting Time. " + "FlowFiles with acknowledgement id unknown for the Splunk server will be transferred to this relationship after the Maximum Waiting Time is reached."); + +const core::Relationship QuerySplunkIndexingStatus::Undetermined("undetermined", + "A FlowFile is transferred to this relationship when the acknowledgement state is not determined. " + "FlowFiles transferred to 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."); + +const core::Relationship QuerySplunkIndexingStatus::Failure("failure", + "A FlowFile is transferred to this relationship when the acknowledgement was not successful due to errors during the communication, " + "or if the flowfile was missing the acknowledgement id"); + +void QuerySplunkIndexingStatus::initialize() { + SplunkHECProcessor::initialize(); + setSupportedRelationships({Acknowledged, Unacknowledged, Undetermined, Failure}); + updateSupportedProperties({MaximumWaitingTime, MaxQuerySize}); +} Review comment: There may be value in listing all of the supported properties here instead of touching the set twice. It's definitely more declarative style, which makes reasoning easier IMO. Related discussion: https://github.com/apache/nifi-minifi-cpp/pull/1158#discussion_r710018162 ########## File path: libminifi/include/utils/TimeUtil.h ########## @@ -37,6 +37,24 @@ namespace minifi { namespace utils { namespace timeutils { +/** + * Converts the time point to the elapsed time since epoch + * @returns TimeUnit since epoch + */ +template<typename TimeUnit, typename TimePoint> +uint64_t getTimeStamp(const TimePoint& time_point) { Review comment: Timestamp is used as a single word in the majority of cases, so it should be capitalized as such. ```suggestion uint64_t getTimestamp(const TimePoint& time_point) { ``` ########## File path: extensions/splunk/SplunkHECProcessor.h ########## @@ -0,0 +1,61 @@ +/** + * 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. + */ + +#pragma once +#include <string> +#include <memory> + +#include "core/Processor.h" + + +namespace org::apache::nifi::minifi::utils { +class HTTPClient; +} + +namespace org::apache::nifi::minifi::extensions::splunk { + +class SplunkHECProcessor : public core::Processor { + public: + EXTENSIONAPI static const core::Property Hostname; + EXTENSIONAPI static const core::Property Port; + EXTENSIONAPI static const core::Property Token; + EXTENSIONAPI static const core::Property SplunkRequestChannel; + EXTENSIONAPI static const core::Property SSLContext; + + explicit SplunkHECProcessor(const std::string& name, const utils::Identifier& uuid = {}) + : Processor(name, uuid) { + } + ~SplunkHECProcessor() override {}; Review comment: ```suggestion ~SplunkHECProcessor() override = default; ``` ########## File path: docker/test/integration/minifi/core/SplunkContainer.py ########## @@ -0,0 +1,26 @@ +import logging +from .Container import Container + + +class SplunkContainer(Container): + def __init__(self, name, vols, network, image_store): + super().__init__(name, 'splunk', vols, network, image_store) + + def get_startup_finished_log_entry(self): + return "Ansible playbook complete, will begin streaming splunkd_stderr.log" + + def deploy(self): + if not self.set_deployed(): + return + + logging.info('Creating and running Splunk docker container...') + self.client.containers.run( + self.image_store.get_image(self.get_engine()), + detach=True, + name=self.name, + network=self.network.name, + environment=[ + "SPLUNK_START_ARGS=--accept-license", Review comment: What license are we accepting here without a prompt? I found an Apache License 2.0 on their page, but it seems strange that they would require accepting an open source license in such a verbose way. ########## File path: extensions/splunk/SplunkHECProcessor.cpp ########## @@ -0,0 +1,81 @@ +/** + * 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. + */ + +#include "SplunkHECProcessor.h" +#include "client/HTTPClient.h" +#include "utils/HTTPClient.h" + +namespace org::apache::nifi::minifi::extensions::splunk { + +const core::Property SplunkHECProcessor::Hostname(core::PropertyBuilder::createProperty("Hostname") + ->withDescription("The ip address or hostname of the Splunk server.") + ->isRequired(true)->build()); + +const core::Property SplunkHECProcessor::Port(core::PropertyBuilder::createProperty("Port") + ->withDescription("The HTTP Event Collector HTTP Port Number.") + ->withDefaultValue("8088")->isRequired(true)->build()); + +const core::Property SplunkHECProcessor::Token(core::PropertyBuilder::createProperty("Token") + ->withDescription("HTTP Event Collector token starting with the string Splunk. For example \'Splunk 1234578-abcd-1234-abcd-1234abcd\'") + ->isRequired(true)->build()); + +const core::Property SplunkHECProcessor::SplunkRequestChannel(core::PropertyBuilder::createProperty("Splunk Request Channel") + ->withDescription("Identifier of the used request channel.")->isRequired(true)->build()); + +const core::Property SplunkHECProcessor::SSLContext(core::PropertyBuilder::createProperty("SSL Context Service") + ->withDescription("The SSL Context Service used to provide client certificate " + "information for TLS/SSL (https) connections.") + ->isRequired(false)->withExclusiveProperty("Remote URL", "^http:.*$") + ->asType<minifi::controllers::SSLContextService>()->build()); + +void SplunkHECProcessor::initialize() { + setSupportedProperties({Hostname, Port, Token, SplunkRequestChannel}); +} + +void SplunkHECProcessor::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>&) { + gsl_Expects(context); + if (!context->getProperty(Hostname.getName(), hostname_)) + throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Failed to get Hostname"); + + if (!context->getProperty(Port.getName(), port_)) + throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Failed to get Port"); + + if (!context->getProperty(Token.getName(), token_)) + throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Failed to get Token"); + + if (!context->getProperty(SplunkRequestChannel.getName(), request_channel_)) + throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Failed to get SplunkRequestChannel"); +} + +std::string SplunkHECProcessor::getUrl() const { + return hostname_ + ":" + port_; Review comment: This is not a URL. You may want to add a protocol prefix or rename the function. (network location?) ########## File path: extensions/splunk/PutSplunkHTTP.cpp ########## @@ -0,0 +1,180 @@ +/** + * 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. + */ + + +#include "PutSplunkHTTP.h" + +#include <vector> +#include <utility> + +#include "SplunkAttributes.h" + +#include "core/Resource.h" +#include "utils/StringUtils.h" +#include "client/HTTPClient.h" +#include "utils/HTTPClient.h" +#include "utils/TimeUtil.h" + +#include "rapidjson/document.h" + + +namespace org::apache::nifi::minifi::extensions::splunk { + +const core::Property PutSplunkHTTP::Source(core::PropertyBuilder::createProperty("Source") + ->withDescription("Basic field describing the source of the event. If unspecified, the event will use the default defined in splunk.") + ->supportsExpressionLanguage(true)->build()); + +const core::Property PutSplunkHTTP::SourceType(core::PropertyBuilder::createProperty("Source Type") + ->withDescription("Basic field describing the source type of the event. If unspecified, the event will use the default defined in splunk.") + ->supportsExpressionLanguage(true)->build()); + +const core::Property PutSplunkHTTP::Host(core::PropertyBuilder::createProperty("Host") + ->withDescription("Basic field describing the host of the event. If unspecified, the event will use the default defined in splunk.") + ->supportsExpressionLanguage(true)->build()); + +const core::Property PutSplunkHTTP::Index(core::PropertyBuilder::createProperty("Index") + ->withDescription("Identifies the index where to send the event. If unspecified, the event will use the default defined in splunk.") + ->supportsExpressionLanguage(true)->build()); + +const core::Property PutSplunkHTTP::ContentType(core::PropertyBuilder::createProperty("Content Type") + ->withDescription("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.") + ->supportsExpressionLanguage(true)->build()); + + +const core::Relationship PutSplunkHTTP::Success("success", "FlowFiles that are sent successfully to the destination are sent to this relationship."); +const core::Relationship PutSplunkHTTP::Failure("failure", "FlowFiles that failed to send to the destination are sent to this relationship."); + +void PutSplunkHTTP::initialize() { + SplunkHECProcessor::initialize(); + setSupportedRelationships({Success, Failure}); + updateSupportedProperties({Source, SourceType, Host, Index, ContentType}); +} + +void PutSplunkHTTP::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>& sessionFactory) { + SplunkHECProcessor::onSchedule(context, sessionFactory); +} + + +namespace { +std::optional<std::string> getContentType(core::ProcessContext& context, const gsl::not_null<std::shared_ptr<core::FlowFile>>& flow_file) { Review comment: A const ref to a flow file should be enough here. If not, then getAttribute needs fixing. ```suggestion std::optional<std::string> getContentType(core::ProcessContext& context, const core::FlowFile& flow_file) { ``` ########## File path: extensions/splunk/PutSplunkHTTP.cpp ########## @@ -0,0 +1,180 @@ +/** + * 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. + */ + + +#include "PutSplunkHTTP.h" + +#include <vector> +#include <utility> + +#include "SplunkAttributes.h" + +#include "core/Resource.h" +#include "utils/StringUtils.h" +#include "client/HTTPClient.h" +#include "utils/HTTPClient.h" +#include "utils/TimeUtil.h" + +#include "rapidjson/document.h" + + +namespace org::apache::nifi::minifi::extensions::splunk { + +const core::Property PutSplunkHTTP::Source(core::PropertyBuilder::createProperty("Source") + ->withDescription("Basic field describing the source of the event. If unspecified, the event will use the default defined in splunk.") + ->supportsExpressionLanguage(true)->build()); + +const core::Property PutSplunkHTTP::SourceType(core::PropertyBuilder::createProperty("Source Type") + ->withDescription("Basic field describing the source type of the event. If unspecified, the event will use the default defined in splunk.") + ->supportsExpressionLanguage(true)->build()); + +const core::Property PutSplunkHTTP::Host(core::PropertyBuilder::createProperty("Host") + ->withDescription("Basic field describing the host of the event. If unspecified, the event will use the default defined in splunk.") + ->supportsExpressionLanguage(true)->build()); + +const core::Property PutSplunkHTTP::Index(core::PropertyBuilder::createProperty("Index") + ->withDescription("Identifies the index where to send the event. If unspecified, the event will use the default defined in splunk.") + ->supportsExpressionLanguage(true)->build()); + +const core::Property PutSplunkHTTP::ContentType(core::PropertyBuilder::createProperty("Content Type") + ->withDescription("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.") + ->supportsExpressionLanguage(true)->build()); + + +const core::Relationship PutSplunkHTTP::Success("success", "FlowFiles that are sent successfully to the destination are sent to this relationship."); +const core::Relationship PutSplunkHTTP::Failure("failure", "FlowFiles that failed to send to the destination are sent to this relationship."); + +void PutSplunkHTTP::initialize() { + SplunkHECProcessor::initialize(); + setSupportedRelationships({Success, Failure}); + updateSupportedProperties({Source, SourceType, Host, Index, ContentType}); +} + +void PutSplunkHTTP::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>& sessionFactory) { + SplunkHECProcessor::onSchedule(context, sessionFactory); +} + + +namespace { +std::optional<std::string> getContentType(core::ProcessContext& context, const gsl::not_null<std::shared_ptr<core::FlowFile>>& flow_file) { + std::optional<std::string> content_type = context.getProperty(PutSplunkHTTP::ContentType); + if (content_type.has_value()) + return content_type; + return flow_file->getAttribute("mime.key"); +} + + +std::string getEndpoint(core::ProcessContext& context, const gsl::not_null<std::shared_ptr<core::FlowFile>>& flow_file) { + std::stringstream endpoint; + endpoint << "/services/collector/raw"; + std::vector<std::string> parameters; + std::string prop_value; + if (context.getProperty(PutSplunkHTTP::SourceType, prop_value, flow_file)) { + parameters.push_back("sourcetype=" + prop_value); + } + if (context.getProperty(PutSplunkHTTP::Source, prop_value, flow_file)) { + parameters.push_back("source=" + prop_value); + } + if (context.getProperty(PutSplunkHTTP::Host, prop_value, flow_file)) { + parameters.push_back("host=" + prop_value); + } + if (context.getProperty(PutSplunkHTTP::Index, prop_value, flow_file)) { + parameters.push_back("index=" + prop_value); + } + if (!parameters.empty()) { + endpoint << "?" << utils::StringUtils::join("&", parameters); + } + return endpoint.str(); +} + +bool addAttributesFromClientResponse(core::FlowFile& flow_file, utils::HTTPClient& client) { + rapidjson::Document response_json; + rapidjson::ParseResult parse_result = response_json.Parse<rapidjson::kParseStopWhenDoneFlag>(client.getResponseBody().data()); + bool result = true; + if (parse_result.IsError()) + return false; + + if (response_json.HasMember("code") && response_json["code"].IsInt()) + flow_file.addAttribute(SPLUNK_RESPONSE_CODE, std::to_string(response_json["code"].GetInt())); + else + result = false; + + if (response_json.HasMember("ackId") && response_json["ackId"].IsUint64()) + flow_file.addAttribute(SPLUNK_ACK_ID, std::to_string(response_json["ackId"].GetUint64())); + else + result = false; + + return result; +} + +bool enrichFlowFileWithAttributes(core::FlowFile& flow_file, utils::HTTPClient& client) { + flow_file.addAttribute(SPLUNK_STATUS_CODE, std::to_string(client.getResponseCode())); + flow_file.addAttribute(SPLUNK_RESPONSE_TIME, std::to_string(utils::timeutils::getTimeStamp<std::chrono::milliseconds>(std::chrono::system_clock::now()))); + + return addAttributesFromClientResponse(flow_file, client) && client.getResponseCode() == 200; +} + +void setFlowFileAsPayload(core::ProcessSession& session, + core::ProcessContext& context, + utils::HTTPClient& client, + const gsl::not_null<std::shared_ptr<core::FlowFile>>& flow_file, + const std::unique_ptr<utils::ByteInputCallBack>& payload_callback, + const std::unique_ptr<utils::HTTPUploadCallback>& payload_callback_obj) { + session.read(flow_file, payload_callback.get()); + payload_callback_obj->ptr = payload_callback.get(); + payload_callback_obj->pos = 0; + client.appendHeader("Content-Length", std::to_string(flow_file->getSize())); + + client.setUploadCallback(payload_callback_obj.get()); + client.setSeekFunction(payload_callback_obj.get()); + + auto content_type = getContentType(context, flow_file); + if (content_type.has_value()) + client.setContentType(content_type.value()); +} +} // namespace + +void PutSplunkHTTP::onTrigger(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSession>& session) { + gsl_Expects(context && session); + + auto ff = session->get(); + if (!ff) { + context->yield(); + return; + } + auto flow_file = gsl::not_null(std::move(ff)); + + utils::HTTPClient client(getUrl() + getEndpoint(*context, flow_file), getSSLContextService(*context)); + setHeaders(client); + + std::unique_ptr<utils::ByteInputCallBack> payload_callback = std::make_unique<utils::ByteInputCallBack>(); + std::unique_ptr<utils::HTTPUploadCallback> payload_callback_obj = std::make_unique<utils::HTTPUploadCallback>(); Review comment: Don't repeat the type after `make_unique`, use `auto` instead. ```suggestion const auto payload_callback = std::make_unique<utils::ByteInputCallBack>(); const auto payload_callback_obj = std::make_unique<utils::HTTPUploadCallback>(); ``` ########## File path: docker/Dockerfile ########## @@ -113,8 +114,8 @@ RUN cmake -DSTATIC_BUILD= -DSKIP_TESTS=true -DENABLE_ALL="${ENABLE_ALL}" -DENABL -DENABLE_COAP="${ENABLE_COAP}" -DENABLE_SQL="${ENABLE_SQL}" -DENABLE_MQTT="${ENABLE_MQTT}" -DENABLE_PCAP="${ENABLE_PCAP}" \ -DENABLE_LIBRDKAFKA="${ENABLE_LIBRDKAFKA}" -DENABLE_SENSORS="${ENABLE_SENSORS}" -DENABLE_USB_CAMERA="${ENABLE_USB_CAMERA}" \ -DENABLE_TENSORFLOW="${ENABLE_TENSORFLOW}" -DENABLE_AWS="${ENABLE_AWS}" -DENABLE_BUSTACHE="${ENABLE_BUSTACHE}" -DENABLE_SFTP="${ENABLE_SFTP}" \ - -DENABLE_OPENWSMAN="${ENABLE_OPENWSMAN}" -DENABLE_AZURE="${ENABLE_AZURE}" -DENABLE_NANOFI=${ENABLE_NANOFI} -DENABLE_SYSTEMD=OFF \ Review comment: extra space added here ########## File path: extensions/splunk/PutSplunkHTTP.cpp ########## @@ -0,0 +1,180 @@ +/** + * 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. + */ + + +#include "PutSplunkHTTP.h" + +#include <vector> +#include <utility> + +#include "SplunkAttributes.h" + +#include "core/Resource.h" +#include "utils/StringUtils.h" +#include "client/HTTPClient.h" +#include "utils/HTTPClient.h" +#include "utils/TimeUtil.h" + +#include "rapidjson/document.h" + + +namespace org::apache::nifi::minifi::extensions::splunk { + +const core::Property PutSplunkHTTP::Source(core::PropertyBuilder::createProperty("Source") + ->withDescription("Basic field describing the source of the event. If unspecified, the event will use the default defined in splunk.") + ->supportsExpressionLanguage(true)->build()); + +const core::Property PutSplunkHTTP::SourceType(core::PropertyBuilder::createProperty("Source Type") + ->withDescription("Basic field describing the source type of the event. If unspecified, the event will use the default defined in splunk.") + ->supportsExpressionLanguage(true)->build()); + +const core::Property PutSplunkHTTP::Host(core::PropertyBuilder::createProperty("Host") + ->withDescription("Basic field describing the host of the event. If unspecified, the event will use the default defined in splunk.") + ->supportsExpressionLanguage(true)->build()); + +const core::Property PutSplunkHTTP::Index(core::PropertyBuilder::createProperty("Index") + ->withDescription("Identifies the index where to send the event. If unspecified, the event will use the default defined in splunk.") + ->supportsExpressionLanguage(true)->build()); + +const core::Property PutSplunkHTTP::ContentType(core::PropertyBuilder::createProperty("Content Type") + ->withDescription("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.") + ->supportsExpressionLanguage(true)->build()); + + +const core::Relationship PutSplunkHTTP::Success("success", "FlowFiles that are sent successfully to the destination are sent to this relationship."); +const core::Relationship PutSplunkHTTP::Failure("failure", "FlowFiles that failed to send to the destination are sent to this relationship."); + +void PutSplunkHTTP::initialize() { + SplunkHECProcessor::initialize(); + setSupportedRelationships({Success, Failure}); + updateSupportedProperties({Source, SourceType, Host, Index, ContentType}); +} + +void PutSplunkHTTP::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>& sessionFactory) { + SplunkHECProcessor::onSchedule(context, sessionFactory); +} + + +namespace { +std::optional<std::string> getContentType(core::ProcessContext& context, const gsl::not_null<std::shared_ptr<core::FlowFile>>& flow_file) { + std::optional<std::string> content_type = context.getProperty(PutSplunkHTTP::ContentType); + if (content_type.has_value()) + return content_type; + return flow_file->getAttribute("mime.key"); +} + + +std::string getEndpoint(core::ProcessContext& context, const gsl::not_null<std::shared_ptr<core::FlowFile>>& flow_file) { + std::stringstream endpoint; + endpoint << "/services/collector/raw"; + std::vector<std::string> parameters; + std::string prop_value; + if (context.getProperty(PutSplunkHTTP::SourceType, prop_value, flow_file)) { + parameters.push_back("sourcetype=" + prop_value); + } + if (context.getProperty(PutSplunkHTTP::Source, prop_value, flow_file)) { + parameters.push_back("source=" + prop_value); + } + if (context.getProperty(PutSplunkHTTP::Host, prop_value, flow_file)) { + parameters.push_back("host=" + prop_value); + } + if (context.getProperty(PutSplunkHTTP::Index, prop_value, flow_file)) { + parameters.push_back("index=" + prop_value); + } + if (!parameters.empty()) { + endpoint << "?" << utils::StringUtils::join("&", parameters); + } + return endpoint.str(); +} + +bool addAttributesFromClientResponse(core::FlowFile& flow_file, utils::HTTPClient& client) { + rapidjson::Document response_json; + rapidjson::ParseResult parse_result = response_json.Parse<rapidjson::kParseStopWhenDoneFlag>(client.getResponseBody().data()); + bool result = true; + if (parse_result.IsError()) + return false; + + if (response_json.HasMember("code") && response_json["code"].IsInt()) + flow_file.addAttribute(SPLUNK_RESPONSE_CODE, std::to_string(response_json["code"].GetInt())); + else + result = false; + + if (response_json.HasMember("ackId") && response_json["ackId"].IsUint64()) + flow_file.addAttribute(SPLUNK_ACK_ID, std::to_string(response_json["ackId"].GetUint64())); + else + result = false; + + return result; +} + +bool enrichFlowFileWithAttributes(core::FlowFile& flow_file, utils::HTTPClient& client) { + flow_file.addAttribute(SPLUNK_STATUS_CODE, std::to_string(client.getResponseCode())); + flow_file.addAttribute(SPLUNK_RESPONSE_TIME, std::to_string(utils::timeutils::getTimeStamp<std::chrono::milliseconds>(std::chrono::system_clock::now()))); + + return addAttributesFromClientResponse(flow_file, client) && client.getResponseCode() == 200; +} + +void setFlowFileAsPayload(core::ProcessSession& session, + core::ProcessContext& context, + utils::HTTPClient& client, + const gsl::not_null<std::shared_ptr<core::FlowFile>>& flow_file, + const std::unique_ptr<utils::ByteInputCallBack>& payload_callback, + const std::unique_ptr<utils::HTTPUploadCallback>& payload_callback_obj) { Review comment: This function only needs references to the actual object. No need to force the caller to use unique_ptr, just use normal references instead. ```suggestion utils::ByteInputCallBack& payload_callback, utils::HTTPUploadCallback& payload_callback_obj) { ``` -- 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. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
