szaszm commented on a change in pull request #1044: URL: https://github.com/apache/nifi-minifi-cpp/pull/1044#discussion_r622396192
########## File path: extensions/systemd/ConsumeJournald.cpp ########## @@ -0,0 +1,254 @@ +/** + * + * 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 "ConsumeJournald.h" + +#include <algorithm> + +#include <date/date.h> +#include "spdlog/spdlog.h" // TODO(szaszm): make fmt directly available +#include "utils/GeneralUtils.h" + +namespace org { namespace apache { namespace nifi { namespace minifi { namespace extensions { namespace systemd { + +constexpr const char* ConsumeJournald::CURSOR_KEY; +const core::Relationship ConsumeJournald::Success("success", "Successfully consumed journal messages."); + +const core::Property ConsumeJournald::BatchSize = core::PropertyBuilder::createProperty("Batch Size") + ->withDescription("The maximum number of entries processed in a single execution.") + ->withDefaultValue<size_t>(1000) + ->isRequired(true) + ->build(); + +const core::Property ConsumeJournald::PayloadFormat = core::PropertyBuilder::createProperty("Payload Format") + ->withDescription("Configures flow file content formatting. Raw: only the message. Syslog: similar to syslog or journalctl output.") + ->withDefaultValue<std::string>(PAYLOAD_FORMAT_SYSLOG) + ->withAllowableValues<std::string>({PAYLOAD_FORMAT_RAW, PAYLOAD_FORMAT_SYSLOG}) + ->isRequired(true) + ->build(); + +const core::Property ConsumeJournald::IncludeTimestamp = core::PropertyBuilder::createProperty("Include Timestamp") + ->withDescription("Include message timestamp in the 'timestamp' attribute.") + ->withDefaultValue<bool>(true) + ->isRequired(true) + ->build(); + +const core::Property ConsumeJournald::JournalType = core::PropertyBuilder::createProperty("Journal Type") + ->withDescription("Type of journal to consume.") + ->withDefaultValue<std::string>(JOURNAL_TYPE_SYSTEM) + ->withAllowableValues<std::string>({JOURNAL_TYPE_USER, JOURNAL_TYPE_SYSTEM, JOURNAL_TYPE_BOTH}) + ->isRequired(true) + ->build(); + +const core::Property ConsumeJournald::ProcessOldMessages = core::PropertyBuilder::createProperty("Process Old Messages") + ->withDescription("Process events created before the first usage (schedule) of the processor instance.") + ->withDefaultValue<bool>(false) + ->isRequired(true) + ->build(); + +const core::Property ConsumeJournald::TimestampFormat = core::PropertyBuilder::createProperty("Timestamp Format") + ->withDescription("Format string to use when creating the timestamp attribute or writing messages in the syslog format.") + ->withDefaultValue("%x %X %Z") + ->isRequired(true) + ->build(); + +ConsumeJournald::ConsumeJournald(const std::string &name, const utils::Identifier &id, std::unique_ptr<libwrapper::LibWrapper>&& libwrapper) + :core::Processor{name, id}, libwrapper_{std::move(libwrapper)} +{} + +void ConsumeJournald::initialize() { + setSupportedProperties({BatchSize, PayloadFormat, IncludeTimestamp, JournalType, ProcessOldMessages, TimestampFormat}); + setSupportedRelationships({Success}); + + worker_ = utils::make_unique<Worker>(); +} + +void ConsumeJournald::notifyStop() { + bool running = true; + if (!running_.compare_exchange_strong(running, false, std::memory_order_acq_rel) || !journal_) return; + worker_->enqueue([this] { + journal_ = nullptr; + }).get(); + worker_ = nullptr; +} + +void ConsumeJournald::onSchedule(core::ProcessContext* const context, core::ProcessSessionFactory* const sessionFactory) { + gsl_Expects(context && sessionFactory && !running_ && worker_); + using JournalTypeEnum = systemd::JournalType; + + const auto parse_payload_format = [](const std::string& property_value) -> utils::optional<systemd::PayloadFormat> { + if (utils::StringUtils::equalsIgnoreCase(property_value, PAYLOAD_FORMAT_RAW)) return systemd::PayloadFormat::Raw; + if (utils::StringUtils::equalsIgnoreCase(property_value, PAYLOAD_FORMAT_SYSLOG)) return systemd::PayloadFormat::Syslog; + return utils::nullopt; + }; + const auto parse_journal_type = [](const std::string& property_value) -> utils::optional<JournalTypeEnum> { + if (utils::StringUtils::equalsIgnoreCase(property_value, JOURNAL_TYPE_USER)) return JournalTypeEnum::User; + if (utils::StringUtils::equalsIgnoreCase(property_value, JOURNAL_TYPE_SYSTEM)) return JournalTypeEnum::System; + if (utils::StringUtils::equalsIgnoreCase(property_value, JOURNAL_TYPE_BOTH)) return JournalTypeEnum::Both; + return utils::nullopt; + }; + batch_size_ = context->getProperty<size_t>(BatchSize).value_or(10); + payload_format_ = (context->getProperty(PayloadFormat) | utils::flatMap(parse_payload_format)).value_or(systemd::PayloadFormat::Syslog); + include_timestamp_ = context->getProperty<bool>(IncludeTimestamp).value_or(true); + const auto journal_type = (context->getProperty(JournalType) | utils::flatMap(parse_journal_type)).value_or(JournalTypeEnum::System); + const auto process_old_messages = context->getProperty<bool>(ProcessOldMessages).value_or(false); + timestamp_format_ = [&context] { + auto tf_prop = context->getProperty(TimestampFormat).value_or(TimestampFormat.getDefaultValue()); + if (tf_prop == "ISO" || tf_prop == "ISO 8601" || tf_prop == "ISO8601") return std::string{"%FT%T%Ez"}; + return tf_prop; + }(); + + state_manager_ = context->getStateManager(); + journal_ = worker_->enqueue([this, journal_type]{ return libwrapper_->openJournal(journal_type); }).get(); Review comment: The journal functions are "thread-agnostic", meaning for a given journal handle, they must be called from the same thread. In practice, they probably rely on thread-local storage. See the NOTES of [`man sd_journal_open`](https://www.man7.org/linux/man-pages/man3/sd_journal_open_namespace.3.html). -- 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: us...@infra.apache.org
