[GitHub] [nifi-minifi-cpp] martinzink commented on a change in pull request #1044: MINIFICPP-1526 Add ConsumeJournald to consume systemd journal messages
martinzink commented on a change in pull request #1044:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1044#discussion_r634533482
##
File path: CMakeLists.txt
##
@@ -564,6 +560,14 @@ if (ENABLE_ALL OR ENABLE_AZURE)
createExtension(AZURE-EXTENSIONS "AZURE EXTENSIONS" "This enables Azure
support" "extensions/azure" "${TEST_DIR}/azure-tests")
endif()
+## Add the systemd extension
+if(CMAKE_SYSTEM_NAME STREQUAL "Linux")
Review comment:
checked it, compiles without an issue
--
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
[GitHub] [nifi-minifi-cpp] martinzink commented on a change in pull request #1044: MINIFICPP-1526 Add ConsumeJournald to consume systemd journal messages
martinzink commented on a change in pull request #1044:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1044#discussion_r632507305
##
File path: extensions/systemd/ConsumeJournald.cpp
##
@@ -0,0 +1,262 @@
+/**
+ *
+ * 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
+
+#include
+#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(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(PAYLOAD_FORMAT_SYSLOG)
+->withAllowableValues({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(true)
+->isRequired(true)
+->build();
+
+const core::Property ConsumeJournald::JournalType =
core::PropertyBuilder::createProperty("Journal Type")
+->withDescription("Type of journal to consume.")
+->withDefaultValue(JOURNAL_TYPE_SYSTEM)
+->withAllowableValues({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(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)
+:core::Processor{name, id}, libwrapper_{std::move(libwrapper)}
+{}
+
+void ConsumeJournald::initialize() {
+ setSupportedProperties({BatchSize, PayloadFormat, IncludeTimestamp,
JournalType, ProcessOldMessages, TimestampFormat});
+ setSupportedRelationships({Success});
+
+ worker_ = utils::make_unique();
+}
+
+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 {
+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 {
+if (utils::StringUtils::equalsIgnoreCase(property_value,
JOURNAL_TYPE_USER)) return JournalTypeEnum::User;
+if (utils::StringUtils::equalsIgnoreCase(proper
[GitHub] [nifi-minifi-cpp] martinzink commented on a change in pull request #1044: MINIFICPP-1526 Add ConsumeJournald to consume systemd journal messages
martinzink commented on a change in pull request #1044:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1044#discussion_r632507305
##
File path: extensions/systemd/ConsumeJournald.cpp
##
@@ -0,0 +1,262 @@
+/**
+ *
+ * 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
+
+#include
+#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(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(PAYLOAD_FORMAT_SYSLOG)
+->withAllowableValues({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(true)
+->isRequired(true)
+->build();
+
+const core::Property ConsumeJournald::JournalType =
core::PropertyBuilder::createProperty("Journal Type")
+->withDescription("Type of journal to consume.")
+->withDefaultValue(JOURNAL_TYPE_SYSTEM)
+->withAllowableValues({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(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)
+:core::Processor{name, id}, libwrapper_{std::move(libwrapper)}
+{}
+
+void ConsumeJournald::initialize() {
+ setSupportedProperties({BatchSize, PayloadFormat, IncludeTimestamp,
JournalType, ProcessOldMessages, TimestampFormat});
+ setSupportedRelationships({Success});
+
+ worker_ = utils::make_unique();
+}
+
+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 {
+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 {
+if (utils::StringUtils::equalsIgnoreCase(property_value,
JOURNAL_TYPE_USER)) return JournalTypeEnum::User;
+if (utils::StringUtils::equalsIgnoreCase(proper
[GitHub] [nifi-minifi-cpp] martinzink commented on a change in pull request #1044: MINIFICPP-1526 Add ConsumeJournald to consume systemd journal messages
martinzink commented on a change in pull request #1044:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1044#discussion_r632376008
##
File path: CMakeLists.txt
##
@@ -564,6 +560,14 @@ if (ENABLE_ALL OR ENABLE_AZURE)
createExtension(AZURE-EXTENSIONS "AZURE EXTENSIONS" "This enables Azure
support" "extensions/azure" "${TEST_DIR}/azure-tests")
endif()
+## Add the systemd extension
+if(CMAKE_SYSTEM_NAME STREQUAL "Linux")
Review comment:
How does this behave on distros without systemd? e.g Artix
--
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
[GitHub] [nifi-minifi-cpp] martinzink commented on a change in pull request #1044: MINIFICPP-1526 Add ConsumeJournald to consume systemd journal messages
martinzink commented on a change in pull request #1044:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1044#discussion_r632373602
##
File path: extensions/systemd/CMakeLists.txt
##
@@ -0,0 +1,28 @@
+#
+# 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(${CMAKE_SOURCE_DIR}/extensions/ExtensionHeader.txt)
+
+add_library(minifi-systemd STATIC ConsumeJournald.cpp WorkerThread.cpp
libwrapper/LibWrapper.h Common.h libwrapper/LibWrapper.cpp
libwrapper/DlopenWrapper.cpp libwrapper/DlopenWrapper.h)
+set_property(TARGET minifi-systemd PROPERTY POSITION_INDEPENDENT_CODE ON)
+
+target_link_libraries(minifi-systemd ${LIBMINIFI} Threads::Threads date::date)
+
+set(SYSTEMD-EXTENSION minifi-systemd PARENT_SCOPE)
+register_extension(minifi-systemd)
Review comment:
I think, we should add this extension to the linter check as well
```suggestion
register_extension(minifi-systemd)
register_extension_linter(minifi-systemd-extension-linter)
```
this also brings out a couple linter violations which should be fixed as well
--
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
