Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
arpadboda closed pull request #1703: MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() URL: https://github.com/apache/nifi-minifi-cpp/pull/1703 -- 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: issues-unsubscr...@nifi.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
szaszm commented on PR #1703: URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#issuecomment-1853035680 rebased -- 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: issues-unsubscr...@nifi.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
martinzink commented on PR #1703: URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#issuecomment-1851842349 @szaszm Could you rebase and fix conflicts from https://github.com/apache/nifi-minifi-cpp/pull/1693 ? -- 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: issues-unsubscr...@nifi.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
szaszm commented on code in PR #1703:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1415251638
##
extensions/standard-processors/RollingWindow.h:
##
@@ -0,0 +1,71 @@
+/**
+ * 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
+#include
+#include
+
+namespace org::apache::nifi::minifi::processors::standard::utils {
+
+namespace detail {
+template
+struct priority_queue : std::priority_queue {
+ using std::priority_queue::priority_queue;
+
+ // Expose the underlying container
+ const Container& get_container() const & { return this->c; }
+ Container get_container() && { return std::move(this->c); }
+};
+} // namespace detail
+
+template
+class RollingWindow {
+ public:
+ struct Entry {
+Timestamp timestamp{};
+Value value{};
+ };
+ struct EntryComparator {
+// greater-than, because std::priority_queue order is reversed. This way,
top() is the oldest entry.
+bool operator()(const Entry& lhs, const Entry& rhs) const {
+ return lhs.timestamp > rhs.timestamp;
+}
+ };
+
+ void removeOlderThan(Timestamp timestamp) {
+while (!state_.empty() && state_.top().timestamp < timestamp) {
+ state_.pop();
+}
+ }
+
+ /** Remove the oldest entries until the size is <= size. */
+ void shrinkToSize(size_t size) {
+while (state_.size() > size && !state_.empty()) {
Review Comment:
You're right. When answering last time, I somehow didn't realize that the
lowest size goes is 0, and `>0` can't be 0. I'm changing it.
--
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: issues-unsubscr...@nifi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
fgerlits commented on code in PR #1703:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1415097648
##
extensions/standard-processors/RollingWindow.h:
##
@@ -0,0 +1,71 @@
+/**
+ * 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
+#include
+#include
+
+namespace org::apache::nifi::minifi::processors::standard::utils {
+
+namespace detail {
+template
+struct priority_queue : std::priority_queue {
+ using std::priority_queue::priority_queue;
+
+ // Expose the underlying container
+ const Container& get_container() const & { return this->c; }
+ Container get_container() && { return std::move(this->c); }
+};
+} // namespace detail
+
+template
+class RollingWindow {
+ public:
+ struct Entry {
+Timestamp timestamp{};
+Value value{};
+ };
+ struct EntryComparator {
+// greater-than, because std::priority_queue order is reversed. This way,
top() is the oldest entry.
+bool operator()(const Entry& lhs, const Entry& rhs) const {
+ return lhs.timestamp > rhs.timestamp;
+}
+ };
+
+ void removeOlderThan(Timestamp timestamp) {
+while (!state_.empty() && state_.top().timestamp < timestamp) {
+ state_.pop();
+}
+ }
+
+ /** Remove the oldest entries until the size is <= size. */
+ void shrinkToSize(size_t size) {
+while (state_.size() > size && !state_.empty()) {
Review Comment:
I meant that the first check, `state_.size() > size`, already implies
`state_.size() > 0`, which implies `!state_.empty()`. But yeah, it's just a
few extra (lightweight) operations which the compiler may be able to optimize
out. If you think it's more readable this way, I'm OK with keeping it as it is.
--
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: issues-unsubscr...@nifi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
szaszm commented on code in PR #1703:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1414716543
##
libminifi/src/core/extension/ExtensionManager.cpp:
##
@@ -72,9 +72,13 @@ bool ExtensionManager::initialize(const
std::shared_ptr& config) {
}));
for (const auto& candidate : candidates) {
auto library = internal::asDynamicLibrary(candidate);
- if (!library || !library->verify(logger_)) {
+ if (!library) {
continue;
}
+ if (!library->verify(logger_)) {
+logger_->log_warn("Skipping library '{}' at '{}': failed verification,
different build?",
+library->name, library->getFullPath());
+ }
Review Comment:
Yup, that was an oversight, thanks for catching it. fixed in
[bdf1cd3](https://github.com/apache/nifi-minifi-cpp/pull/1703/commits/bdf1cd3b5f27b9c4d759ebc23b1a05c259905ef9)
--
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: issues-unsubscr...@nifi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
szaszm commented on code in PR #1703:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1414719449
##
libminifi/include/core/AbstractProcessor.h:
##
@@ -0,0 +1,59 @@
+/**
+ * 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
+#include
+#include
+#include "range/v3/view/transform.hpp"
+#include "core/Annotation.h"
+#include "core/Core.h"
+#include "core/Processor.h"
+#include "core/PropertyDefinition.h"
+#include "core/RelationshipDefinition.h"
+
+namespace org::apache::nifi::minifi::core {
+template
+class AbstractProcessor : public Processor {
+ public:
+ using Processor::Processor;
+
+ void initialize() final {
+static_assert(std::is_same_v);
+static_assert(std::is_same_v);
+setSupportedProperties(ProcessorT::Properties);
+setSupportedRelationships(ProcessorT::Relationships);
+ }
+
+ void onSchedule(core::ProcessContext*, core::ProcessSessionFactory*)
override = 0;
+ void onTrigger(core::ProcessContext*, core::ProcessSession*) override = 0;
+
+ bool supportsDynamicProperties() const noexcept final { return
ProcessorT::SupportsDynamicProperties; }
+ bool supportsDynamicRelationships() const noexcept final { return
ProcessorT::SupportsDynamicRelationships; }
+ minifi::core::annotation::Input getInputRequirement() const noexcept final {
return ProcessorT::InputRequirement; }
+ bool isSingleThreaded() const noexcept final { return
ProcessorT::IsSingleThreaded; }
+ std::string getProcessorType() const final {
+constexpr auto class_name = className();
+constexpr auto last_colon_index = class_name.find_last_of(':');
+constexpr auto end = class_name.substr(last_colon_index + 1);
+if constexpr (last_colon_index == std::string_view::npos) {
+ return std::string{class_name};
+}
Review Comment:
inlined `end` to the return statement in
[bdf1cd3](https://github.com/apache/nifi-minifi-cpp/pull/1703/commits/bdf1cd3b5f27b9c4d759ebc23b1a05c259905ef9)
--
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: issues-unsubscr...@nifi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
szaszm commented on code in PR #1703:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1414719168
##
extensions/standard-processors/processors/AttributeRollingWindow.cpp:
##
@@ -0,0 +1,122 @@
+/**
+ * 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 "AttributeRollingWindow.h"
+#include
+#include
+#include "fmt/format.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "core/Resource.h"
+#include "utils/expected.h"
+#include "utils/OptionalUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+void AttributeRollingWindow::onSchedule(core::ProcessContext* context,
core::ProcessSessionFactory*) {
+ gsl_Expects(context);
+ time_window_ = context->getProperty(TimeWindow)
+ | utils::transform(::TimePeriodValue::getMilliseconds);
+ window_length_ = context->getProperty(WindowLength)
+ | utils::filter([](uint64_t value) { return value > 0; })
+ | utils::transform([](uint64_t value) { return size_t{value}; });
+ if (!time_window_ && !window_length_) {
+throw minifi::Exception{ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Either
'Time window' or 'Window length' must be set"};
+ }
+ attribute_name_prefix_ = (context->getProperty(AttributeNamePrefix)
+ | utils::orElse([] {
+throw minifi::Exception{ExceptionType::PROCESS_SCHEDULE_EXCEPTION,
"'Attribute name prefix' must be set"};
+ })).value();
+ gsl_Ensures(runningInvariant());
+}
+
+void AttributeRollingWindow::onTrigger(core::ProcessContext* context,
core::ProcessSession* session) {
+ gsl_Expects(context && session && runningInvariant());
+ const auto flow_file = session->get();
+ if (!flow_file) { yield(); return; }
+ gsl_Assert(flow_file);
+ const auto current_value_opt = context->getProperty(ValueToTrack, flow_file);
+ if (!current_value_opt) {
+logger_->log_warn("Missing value to track, flow file uuid: {}",
flow_file->getUUIDStr());
+session->transfer(flow_file, Failure);
+return;
+ }
+ const auto current_value = [_value_opt] {
+try {
+ return std::stod(*current_value_opt);
+} catch (const std::exception& ex) {
+ throw minifi::Exception{ExceptionType::PROCESSOR_EXCEPTION,
+ fmt::format("Failed to convert 'Value to track' of '{}' to double",
*current_value_opt)};
+}
+ }();
+ // copy: so we can release the lock sooner
+ const auto state_copy = [&, now = std::chrono::system_clock::now()] {
+const std::lock_guard lg{state_mutex_};
+state_.add(now, current_value);
+if (window_length_) {
+ state_.shrinkToSize(*window_length_);
+} else {
+ gsl_Assert(time_window_);
+ state_.removeOlderThan(now - *time_window_);
+}
+return state_.getEntries();
+ }();
+ const auto sorted_values = [_copy] {
+auto values = state_copy |
ranges::views::transform((state_)::Entry::value) |
ranges::to;
+std::sort(std::begin(values), std::end(values));
+return values;
+ }();
+ calculateAndSetAttributes(*flow_file, sorted_values);
+ session->transfer(flow_file, Success);
+}
+
+/**
+ * Calculate statistical properties of the values in the rolling window and
set them as attributes on the flow file.
+ * Properties: count, value (sum), mean (average), median, variance, stddev
+ */
+void AttributeRollingWindow::calculateAndSetAttributes(core::FlowFile
_file,
+std::span sorted_values) const {
+ const auto attribute_name = [this](std::string_view suffix) {
+return utils::string::join_pack(attribute_name_prefix_, suffix);
+ };
+ const auto set_aggregate = [_file, attribute_name](std::string_view
name, double value) {
+flow_file.setAttribute(attribute_name(name), std::to_string(value));
+ };
+ set_aggregate("count", sorted_values.size());
+ const auto sum = std::accumulate(std::begin(sorted_values),
std::end(sorted_values), 0.0);
+ set_aggregate("value", sum);
+ const auto mean = sum / gsl::narrow_cast(sorted_values.size());
+ set_aggregate("mean", mean);
+ set_aggregate("median", [&] {
+const auto mid = sorted_values.size() / 2;
+return sorted_values.size() % 2 == 0
+? std::midpoint(sorted_values[mid], sorted_values[mid - 1]) // even
number of values: average the two
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
szaszm commented on code in PR #1703:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1414718950
##
extensions/standard-processors/processors/AttributeRollingWindow.cpp:
##
@@ -0,0 +1,122 @@
+/**
+ * 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 "AttributeRollingWindow.h"
+#include
+#include
+#include "fmt/format.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "core/Resource.h"
+#include "utils/expected.h"
+#include "utils/OptionalUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+void AttributeRollingWindow::onSchedule(core::ProcessContext* context,
core::ProcessSessionFactory*) {
+ gsl_Expects(context);
+ time_window_ = context->getProperty(TimeWindow)
+ | utils::transform(::TimePeriodValue::getMilliseconds);
+ window_length_ = context->getProperty(WindowLength)
+ | utils::filter([](uint64_t value) { return value > 0; })
+ | utils::transform([](uint64_t value) { return size_t{value}; });
+ if (!time_window_ && !window_length_) {
+throw minifi::Exception{ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Either
'Time window' or 'Window length' must be set"};
+ }
+ attribute_name_prefix_ = (context->getProperty(AttributeNamePrefix)
+ | utils::orElse([] {
+throw minifi::Exception{ExceptionType::PROCESS_SCHEDULE_EXCEPTION,
"'Attribute name prefix' must be set"};
+ })).value();
+ gsl_Ensures(runningInvariant());
+}
+
+void AttributeRollingWindow::onTrigger(core::ProcessContext* context,
core::ProcessSession* session) {
+ gsl_Expects(context && session && runningInvariant());
+ const auto flow_file = session->get();
+ if (!flow_file) { yield(); return; }
+ gsl_Assert(flow_file);
+ const auto current_value_opt = context->getProperty(ValueToTrack, flow_file);
+ if (!current_value_opt) {
+logger_->log_warn("Missing value to track, flow file uuid: {}",
flow_file->getUUIDStr());
+session->transfer(flow_file, Failure);
+return;
+ }
+ const auto current_value = [_value_opt] {
+try {
+ return std::stod(*current_value_opt);
+} catch (const std::exception& ex) {
+ throw minifi::Exception{ExceptionType::PROCESSOR_EXCEPTION,
+ fmt::format("Failed to convert 'Value to track' of '{}' to double",
*current_value_opt)};
+}
+ }();
+ // copy: so we can release the lock sooner
+ const auto state_copy = [&, now = std::chrono::system_clock::now()] {
+const std::lock_guard lg{state_mutex_};
+state_.add(now, current_value);
+if (window_length_) {
+ state_.shrinkToSize(*window_length_);
+} else {
+ gsl_Assert(time_window_);
+ state_.removeOlderThan(now - *time_window_);
+}
+return state_.getEntries();
+ }();
+ const auto sorted_values = [_copy] {
+auto values = state_copy |
ranges::views::transform((state_)::Entry::value) |
ranges::to;
+std::sort(std::begin(values), std::end(values));
+return values;
+ }();
+ calculateAndSetAttributes(*flow_file, sorted_values);
+ session->transfer(flow_file, Success);
+}
+
+/**
+ * Calculate statistical properties of the values in the rolling window and
set them as attributes on the flow file.
+ * Properties: count, value (sum), mean (average), median, variance, stddev
Review Comment:
fixed in
[bdf1cd3](https://github.com/apache/nifi-minifi-cpp/pull/1703/commits/bdf1cd3b5f27b9c4d759ebc23b1a05c259905ef9)
(added min, max)
--
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: issues-unsubscr...@nifi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
szaszm commented on code in PR #1703:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1414718897
##
extensions/standard-processors/processors/AttributeRollingWindow.cpp:
##
@@ -0,0 +1,122 @@
+/**
+ * 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 "AttributeRollingWindow.h"
+#include
+#include
+#include "fmt/format.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "core/Resource.h"
+#include "utils/expected.h"
+#include "utils/OptionalUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+void AttributeRollingWindow::onSchedule(core::ProcessContext* context,
core::ProcessSessionFactory*) {
+ gsl_Expects(context);
+ time_window_ = context->getProperty(TimeWindow)
+ | utils::transform(::TimePeriodValue::getMilliseconds);
+ window_length_ = context->getProperty(WindowLength)
+ | utils::filter([](uint64_t value) { return value > 0; })
+ | utils::transform([](uint64_t value) { return size_t{value}; });
+ if (!time_window_ && !window_length_) {
+throw minifi::Exception{ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Either
'Time window' or 'Window length' must be set"};
+ }
+ attribute_name_prefix_ = (context->getProperty(AttributeNamePrefix)
+ | utils::orElse([] {
+throw minifi::Exception{ExceptionType::PROCESS_SCHEDULE_EXCEPTION,
"'Attribute name prefix' must be set"};
+ })).value();
+ gsl_Ensures(runningInvariant());
+}
+
+void AttributeRollingWindow::onTrigger(core::ProcessContext* context,
core::ProcessSession* session) {
+ gsl_Expects(context && session && runningInvariant());
+ const auto flow_file = session->get();
+ if (!flow_file) { yield(); return; }
+ gsl_Assert(flow_file);
+ const auto current_value_opt = context->getProperty(ValueToTrack, flow_file);
+ if (!current_value_opt) {
+logger_->log_warn("Missing value to track, flow file uuid: {}",
flow_file->getUUIDStr());
+session->transfer(flow_file, Failure);
+return;
+ }
+ const auto current_value = [_value_opt] {
+try {
+ return std::stod(*current_value_opt);
+} catch (const std::exception& ex) {
+ throw minifi::Exception{ExceptionType::PROCESSOR_EXCEPTION,
+ fmt::format("Failed to convert 'Value to track' of '{}' to double",
*current_value_opt)};
+}
+ }();
Review Comment:
fixed in
[bdf1cd3](https://github.com/apache/nifi-minifi-cpp/pull/1703/commits/bdf1cd3b5f27b9c4d759ebc23b1a05c259905ef9)
##
extensions/standard-processors/processors/AttributeRollingWindow.cpp:
##
@@ -0,0 +1,122 @@
+/**
+ * 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 "AttributeRollingWindow.h"
+#include
+#include
+#include "fmt/format.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "core/Resource.h"
+#include "utils/expected.h"
+#include "utils/OptionalUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+void AttributeRollingWindow::onSchedule(core::ProcessContext* context,
core::ProcessSessionFactory*) {
+ gsl_Expects(context);
+ time_window_ = context->getProperty(TimeWindow)
+ | utils::transform(::TimePeriodValue::getMilliseconds);
+ window_length_ = context->getProperty(WindowLength)
+ | utils::filter([](uint64_t value) { return value > 0; })
+ | utils::transform([](uint64_t value) { return size_t{value}; });
+ if (!time_window_ && !window_length_) {
+throw
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
szaszm commented on code in PR #1703:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1414718702
##
extensions/standard-processors/processors/AttributeRollingWindow.h:
##
@@ -0,0 +1,117 @@
+/**
+ * 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
+#include
+#include
+#include
+#include
+#include
+
+#include "core/AbstractProcessor.h"
+#include "core/Annotation.h"
+#include "core/logging/LoggerFactory.h"
+#include "core/PropertyDefinitionBuilder.h"
+#include "core/PropertyType.h"
+#include "core/RelationshipDefinition.h"
+#include "RollingWindow.h"
+#include "StateManager.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+class AttributeRollingWindow final : public
core::AbstractProcessor {
+ public:
+ using core::AbstractProcessor::AbstractProcessor;
+
+ EXTENSIONAPI static constexpr auto Description = "Track a Rolling Window
based on evaluating an Expression Language "
+ "expression on each FlowFile. Each FlowFile will be emitted with the
count of FlowFiles and total aggregate value"
+ "of values processed in the current window.";
+
+ EXTENSIONAPI static constexpr auto ValueToTrack =
core::PropertyDefinitionBuilder<>::createProperty("Value to track")
+ .withDescription("The expression on which to evaluate each FlowFile. The
result of the expression will be added "
+ "to the rolling window value.")
+ .isRequired(true)
+ .supportsExpressionLanguage(true)
+ .build();
+ EXTENSIONAPI static constexpr auto TimeWindow =
core::PropertyDefinitionBuilder<>::createProperty("Time window")
+ .withDescription("The amount of time for a rolling window. The format of
the value is expected to be a "
+ "count followed by a time unit. For example 5 millis, 10 secs, 1
min, 3 hours, 2 days, etc.")
+ .withPropertyType(core::StandardPropertyTypes::TIME_PERIOD_TYPE)
+ .build();
+ EXTENSIONAPI static constexpr auto WindowLength =
core::PropertyDefinitionBuilder<>::createProperty("Window length")
+ .withDescription("The window length in number of values. Takes
precedence over 'Time window'. If set to zero, "
+ "the 'Time window' property is used instead.")
+ .isRequired(true)
+ .withDefaultValue("0")
+ .withPropertyType(core::StandardPropertyTypes::UNSIGNED_INT_TYPE)
+ .build();
+ EXTENSIONAPI static constexpr auto AttributeNamePrefix =
core::PropertyDefinitionBuilder<>::createProperty("Attribute name prefix")
+ .withDescription("The prefix to add to the generated attribute names.
For example, if this is set to 'rolling.window.', "
+ "then the full attribute names will be
'rolling.window.value', 'rolling.window.count', etc.")
+ .isRequired(true)
+ .withDefaultValue("rolling.window.")
+ .build();
+ EXTENSIONAPI static constexpr auto Properties =
std::array{
+ValueToTrack,
+TimeWindow,
+WindowLength,
+AttributeNamePrefix
+ };
+
+ EXTENSIONAPI static constexpr auto Success =
core::RelationshipDefinition{"success", "All FlowFiles that are "
+ "successfully processed are routed to this relationship."};
+ EXTENSIONAPI static constexpr auto Failure =
core::RelationshipDefinition{"failure", "When a FlowFile fails, "
+ "it is routed here."};
+ EXTENSIONAPI static constexpr auto Relationships = std::array{Success,
Failure};
+
+ EXTENSIONAPI static constexpr auto Count =
core::OutputAttributeDefinition<1>{"count", {Success}, "Number of the
values in the rolling window"};
+ EXTENSIONAPI static constexpr auto Value =
core::OutputAttributeDefinition<1>{"value", {Success}, "Sum of the
values in the rolling window"};
+ EXTENSIONAPI static constexpr auto Mean =
core::OutputAttributeDefinition<1>{"mean", {Success}, "Mean of the
values in the rolling window"};
+ EXTENSIONAPI static constexpr auto Median =
core::OutputAttributeDefinition<1>{"median", {Success}, "Median of the
values in the rolling window"};
+ EXTENSIONAPI static constexpr auto Variance =
core::OutputAttributeDefinition<1>{"variance", {Success}, "Variance of
the values in the rolling window"};
+ EXTENSIONAPI static constexpr auto Stddev =
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
szaszm commented on code in PR #1703: URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1414718561 ## PROCESSORS.md: ## @@ -147,6 +148,44 @@ In the list below, the names of required properties appear in bold. Any other pr | success | success operational on the flow record | +## AttributeRollingWindow + +### Description + +Track a Rolling Window based on evaluating an Expression Language expression on each FlowFile. Each FlowFile will be emitted with the count of FlowFiles and total aggregate valueof values processed in the current window. Review Comment: fixed in [bdf1cd3](https://github.com/apache/nifi-minifi-cpp/pull/1703/commits/bdf1cd3b5f27b9c4d759ebc23b1a05c259905ef9) -- 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: issues-unsubscr...@nifi.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
szaszm commented on code in PR #1703:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1414716543
##
libminifi/src/core/extension/ExtensionManager.cpp:
##
@@ -72,9 +72,13 @@ bool ExtensionManager::initialize(const
std::shared_ptr& config) {
}));
for (const auto& candidate : candidates) {
auto library = internal::asDynamicLibrary(candidate);
- if (!library || !library->verify(logger_)) {
+ if (!library) {
continue;
}
+ if (!library->verify(logger_)) {
+logger_->log_warn("Skipping library '{}' at '{}': failed verification,
different build?",
+library->name, library->getFullPath());
+ }
Review Comment:
Yup, that was an oversight, thanks for catching it
--
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: issues-unsubscr...@nifi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
szaszm commented on code in PR #1703: URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1414711038 ## PROCESSORS.md: ## @@ -147,6 +148,44 @@ In the list below, the names of required properties appear in bold. Any other pr | success | success operational on the flow record | +## AttributeRollingWindow + +### Description + +Track a Rolling Window based on evaluating an Expression Language expression on each FlowFile. Each FlowFile will be emitted with the count of FlowFiles and total aggregate valueof values processed in the current window. + +### Properties + +In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language. + +| Name | Default Value | Allowable Values | Description | +|---|-|--|--| +| **Value to track**| | | The expression on which to evaluate each FlowFile. The result of the expression will be added to the rolling window value.**Supports Expression Language: true**| +| Time window | | | The amount of time for a rolling window. The format of the value is expected to be a count followed by a time unit. For example 5 millis, 10 secs, 1 min, 3 hours, 2 days, etc. | +| **Window length** | 0 | | The window length in number of values. Takes precedence over 'Time window'. If set to zero, the 'Time window' property is used instead. | +| **Attribute name prefix** | rolling.window. | | The prefix to add to the generated attribute names. For example, if this is set to 'rolling.window.', then the full attribute names will be 'rolling.window.value', 'rolling.window.count', etc. | + +### Relationships + +| Name| Description | +|-|| +| success | All FlowFiles that are successfully processed are routed to this relationship. | +| failure | When a FlowFile fails, it is routed here. | + +### Output Attributes + +| Attribute| Relationship | Description | +|--|--|| +| count| success | Number of the values in the rolling window | +| value| success | Sum of the values in the rolling window | Review Comment: The processor already quite different from the NiFi version, but I prefer not to deviate in this aspect. -- 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: issues-unsubscr...@nifi.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
szaszm commented on code in PR #1703:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1414710024
##
libminifi/include/core/AbstractProcessor.h:
##
@@ -0,0 +1,59 @@
+/**
+ * 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
+#include
+#include
+#include "range/v3/view/transform.hpp"
+#include "core/Annotation.h"
+#include "core/Core.h"
+#include "core/Processor.h"
+#include "core/PropertyDefinition.h"
+#include "core/RelationshipDefinition.h"
+
+namespace org::apache::nifi::minifi::core {
+template
+class AbstractProcessor : public Processor {
+ public:
+ using Processor::Processor;
+
+ void initialize() final {
+static_assert(std::is_same_v);
+static_assert(std::is_same_v);
+setSupportedProperties(ProcessorT::Properties);
+setSupportedRelationships(ProcessorT::Relationships);
+ }
+
+ void onSchedule(core::ProcessContext*, core::ProcessSessionFactory*)
override = 0;
+ void onTrigger(core::ProcessContext*, core::ProcessSession*) override = 0;
+
+ bool supportsDynamicProperties() const noexcept final { return
ProcessorT::SupportsDynamicProperties; }
+ bool supportsDynamicRelationships() const noexcept final { return
ProcessorT::SupportsDynamicRelationships; }
+ minifi::core::annotation::Input getInputRequirement() const noexcept final {
return ProcessorT::InputRequirement; }
+ bool isSingleThreaded() const noexcept final { return
ProcessorT::IsSingleThreaded; }
+ std::string getProcessorType() const final {
+constexpr auto class_name = className();
+constexpr auto last_colon_index = class_name.find_last_of(':');
+constexpr auto end = class_name.substr(last_colon_index + 1);
+if constexpr (last_colon_index == std::string_view::npos) {
+ return std::string{class_name};
+}
Review Comment:
Why would it not compile? I have nothing against switching it, but I think
all compliant compilers should have constexpr string_view::substr.
--
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: issues-unsubscr...@nifi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
szaszm commented on code in PR #1703:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1414704269
##
extensions/standard-processors/processors/AttributeRollingWindow.cpp:
##
@@ -0,0 +1,122 @@
+/**
+ * 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 "AttributeRollingWindow.h"
+#include
+#include
+#include "fmt/format.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "core/Resource.h"
+#include "utils/expected.h"
+#include "utils/OptionalUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+void AttributeRollingWindow::onSchedule(core::ProcessContext* context,
core::ProcessSessionFactory*) {
+ gsl_Expects(context);
+ time_window_ = context->getProperty(TimeWindow)
+ | utils::transform(::TimePeriodValue::getMilliseconds);
+ window_length_ = context->getProperty(WindowLength)
+ | utils::filter([](uint64_t value) { return value > 0; })
+ | utils::transform([](uint64_t value) { return size_t{value}; });
Review Comment:
It's using direct-list-initialization, so narrowing conversions will fail to
compile. I originally didn't consider 32 bit platforms, but it would probably
be better to not break the build there. I'm changing it to gsl::narrow, so a
narrowing conversion will terminate.
--
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: issues-unsubscr...@nifi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
szaszm commented on code in PR #1703:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1414703188
##
extensions/standard-processors/RollingWindow.h:
##
@@ -0,0 +1,71 @@
+/**
+ * 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
+#include
+#include
+
+namespace org::apache::nifi::minifi::processors::standard::utils {
+
+namespace detail {
+template
+struct priority_queue : std::priority_queue {
+ using std::priority_queue::priority_queue;
+
+ // Expose the underlying container
+ const Container& get_container() const & { return this->c; }
+ Container get_container() && { return std::move(this->c); }
+};
+} // namespace detail
+
+template
+class RollingWindow {
+ public:
+ struct Entry {
+Timestamp timestamp{};
+Value value{};
+ };
+ struct EntryComparator {
+// greater-than, because std::priority_queue order is reversed. This way,
top() is the oldest entry.
+bool operator()(const Entry& lhs, const Entry& rhs) const {
+ return lhs.timestamp > rhs.timestamp;
+}
+ };
+
+ void removeOlderThan(Timestamp timestamp) {
+while (!state_.empty() && state_.top().timestamp < timestamp) {
+ state_.pop();
+}
+ }
+
+ /** Remove the oldest entries until the size is <= size. */
+ void shrinkToSize(size_t size) {
+while (state_.size() > size && !state_.empty()) {
Review Comment:
While the AttributeRollingWindow processor never calls this with a size of
0, the interface contract of RollingWindow allows shrinking to 0. In the case
of AttributeRollingWindow, the optimizer can probably prove that size is never
zero, and get rid of the extra check after inlining.
--
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: issues-unsubscr...@nifi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
szaszm commented on code in PR #1703:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1414699559
##
extensions/standard-processors/processors/AttributeRollingWindow.cpp:
##
@@ -0,0 +1,121 @@
+/**
+ * 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 "AttributeRollingWindow.h"
+#include
+#include
+#include "fmt/format.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "core/Resource.h"
+#include "utils/expected.h"
+#include "utils/OptionalUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+void AttributeRollingWindow::onSchedule(core::ProcessContext* context,
core::ProcessSessionFactory*) {
+ gsl_Expects(context);
+ time_window_ = context->getProperty(TimeWindow)
+ | utils::transform(::TimePeriodValue::getMilliseconds);
+ window_length_ = context->getProperty(WindowLength)
+ | utils::filter([](size_t value) { return value > 0; });
+ if (!time_window_ && !window_length_) {
+throw minifi::Exception{ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Either
'Time window' or 'Window length' must be set"};
+ }
+ attribute_name_prefix_ = (context->getProperty(AttributeNamePrefix)
+ | utils::orElse([] {
+throw minifi::Exception{ExceptionType::PROCESS_SCHEDULE_EXCEPTION,
"'Attribute name prefix' must be set"};
+ })).value();
+ gsl_Ensures(runningInvariant());
+}
+
+void AttributeRollingWindow::onTrigger(core::ProcessContext* context,
core::ProcessSession* session) {
+ gsl_Expects(context && session && runningInvariant());
+ const auto flow_file = session->get();
+ if (!flow_file) { yield(); return; }
+ gsl_Assert(flow_file);
+ const auto current_value_opt = context->getProperty(ValueToTrack, flow_file);
+ if (!current_value_opt) {
+logger_->log_warn("Missing value to track, flow file uuid: {}",
flow_file->getUUIDStr());
+session->transfer(flow_file, Failure);
+return;
+ }
+ const auto current_value = [_value_opt] {
+try {
+ return std::stod(*current_value_opt);
+} catch (const std::exception& ex) {
+ throw minifi::Exception{ExceptionType::PROCESSOR_EXCEPTION,
+ fmt::format("Failed to convert 'Value to track' of '{}' to double",
*current_value_opt)};
+}
+ }();
+ // copy: so we can release the lock sooner
+ const auto state_copy = [&, now = std::chrono::system_clock::now()] {
+const std::lock_guard lg{state_mutex_};
+state_.add(now, current_value);
+if (window_length_) {
+ state_.shrinkToSize(*window_length_);
+} else {
+ gsl_Assert(time_window_);
+ state_.removeOlderThan(now - *time_window_);
+}
+return state_.getEntries();
+ }();
+ const auto sorted_values = [_copy] {
+auto values = state_copy |
ranges::views::transform((state_)::Entry::value) |
ranges::to;
+std::sort(std::begin(values), std::end(values));
+return values;
+ }();
+ calculateAndSetAttributes(*flow_file, sorted_values);
+ session->transfer(flow_file, Success);
+}
+
+/**
+ * Calculate statistical properties of the values in the rolling window and
set them as attributes on the flow file.
+ * Properties: count, value (sum), mean (average), median, variance, stddev
+ */
+void AttributeRollingWindow::calculateAndSetAttributes(core::FlowFile
_file,
+std::span sorted_values) const {
+ const auto attribute_name = [this](std::string_view suffix) {
+return utils::string::join_pack(attribute_name_prefix_, suffix);
+ };
+ const auto set_aggregate = [_file, attribute_name](std::string_view
name, double value) {
+flow_file.setAttribute(attribute_name(name), std::to_string(value));
+ };
+ set_aggregate("count", sorted_values.size());
+ const auto sum = std::accumulate(std::begin(sorted_values),
std::end(sorted_values), 0.0);
+ set_aggregate("value", sum);
+ const auto mean = sum / gsl::narrow_cast(sorted_values.size());
+ set_aggregate("mean", mean);
+ set_aggregate("median", [&] {
+const auto mid = sorted_values.size() / 2;
+return sorted_values.size() % 2 == 0
+? std::midpoint(sorted_values[mid], sorted_values[mid - 1]) // even
number of values: average the two middle values
+: sorted_values[mid]; // odd number of values:
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
szaszm commented on code in PR #1703:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1414698106
##
extensions/standard-processors/processors/AttributeRollingWindow.cpp:
##
@@ -0,0 +1,121 @@
+/**
+ * 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 "AttributeRollingWindow.h"
+#include
+#include
+#include "fmt/format.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "core/Resource.h"
+#include "utils/expected.h"
+#include "utils/OptionalUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+void AttributeRollingWindow::onSchedule(core::ProcessContext* context,
core::ProcessSessionFactory*) {
+ gsl_Expects(context);
+ time_window_ = context->getProperty(TimeWindow)
+ | utils::transform(::TimePeriodValue::getMilliseconds);
+ window_length_ = context->getProperty(WindowLength)
+ | utils::filter([](size_t value) { return value > 0; });
+ if (!time_window_ && !window_length_) {
+throw minifi::Exception{ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Either
'Time window' or 'Window length' must be set"};
+ }
+ attribute_name_prefix_ = (context->getProperty(AttributeNamePrefix)
+ | utils::orElse([] {
+throw minifi::Exception{ExceptionType::PROCESS_SCHEDULE_EXCEPTION,
"'Attribute name prefix' must be set"};
+ })).value();
+ gsl_Ensures(runningInvariant());
+}
+
+void AttributeRollingWindow::onTrigger(core::ProcessContext* context,
core::ProcessSession* session) {
+ gsl_Expects(context && session && runningInvariant());
+ const auto flow_file = session->get();
+ if (!flow_file) { yield(); return; }
+ gsl_Assert(flow_file);
+ const auto current_value_opt = context->getProperty(ValueToTrack, flow_file);
+ if (!current_value_opt) {
+logger_->log_warn("Missing value to track, flow file uuid: {}",
flow_file->getUUIDStr());
+session->transfer(flow_file, Failure);
+return;
+ }
+ const auto current_value = [_value_opt] {
+try {
+ return std::stod(*current_value_opt);
+} catch (const std::exception& ex) {
+ throw minifi::Exception{ExceptionType::PROCESSOR_EXCEPTION,
+ fmt::format("Failed to convert 'Value to track' of '{}' to double",
*current_value_opt)};
+}
+ }();
+ // copy: so we can release the lock sooner
+ const auto state_copy = [&, now = std::chrono::system_clock::now()] {
+const std::lock_guard lg{state_mutex_};
+state_.add(now, current_value);
+if (window_length_) {
+ state_.shrinkToSize(*window_length_);
+} else {
+ gsl_Assert(time_window_);
+ state_.removeOlderThan(now - *time_window_);
+}
+return state_.getEntries();
+ }();
+ const auto sorted_values = [_copy] {
+auto values = state_copy |
ranges::views::transform((state_)::Entry::value) |
ranges::to;
+std::sort(std::begin(values), std::end(values));
+return values;
+ }();
+ calculateAndSetAttributes(*flow_file, sorted_values);
+ session->transfer(flow_file, Success);
+}
+
+/**
+ * Calculate statistical properties of the values in the rolling window and
set them as attributes on the flow file.
+ * Properties: count, value (sum), mean (average), median, variance, stddev
+ */
+void AttributeRollingWindow::calculateAndSetAttributes(core::FlowFile
_file,
+std::span sorted_values) const {
+ const auto attribute_name = [this](std::string_view suffix) {
+return utils::string::join_pack(attribute_name_prefix_, suffix);
+ };
+ const auto set_aggregate = [_file, attribute_name](std::string_view
name, double value) {
+flow_file.setAttribute(attribute_name(name), std::to_string(value));
+ };
+ set_aggregate("count", sorted_values.size());
+ const auto sum = std::accumulate(std::begin(sorted_values),
std::end(sorted_values), 0.0);
Review Comment:
I prefer the standard library versions when available, because the headers
are more lightweight, and it's not much more typing.
--
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: issues-unsubscr...@nifi.apache.org
For queries about this service, please contact Infrastructure
Re: [PR] MINIFICPP-2265 Implement AttributeRollingWindow and EL nextInt() [nifi-minifi-cpp]
fgerlits commented on code in PR #1703:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1703#discussion_r1414065111
##
extensions/standard-processors/processors/AttributeRollingWindow.cpp:
##
@@ -0,0 +1,121 @@
+/**
+ * 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 "AttributeRollingWindow.h"
+#include
+#include
+#include "fmt/format.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "core/Resource.h"
+#include "utils/expected.h"
+#include "utils/OptionalUtils.h"
+
+namespace org::apache::nifi::minifi::processors {
+
+void AttributeRollingWindow::onSchedule(core::ProcessContext* context,
core::ProcessSessionFactory*) {
+ gsl_Expects(context);
+ time_window_ = context->getProperty(TimeWindow)
+ | utils::transform(::TimePeriodValue::getMilliseconds);
+ window_length_ = context->getProperty(WindowLength)
+ | utils::filter([](size_t value) { return value > 0; });
+ if (!time_window_ && !window_length_) {
+throw minifi::Exception{ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Either
'Time window' or 'Window length' must be set"};
+ }
+ attribute_name_prefix_ = (context->getProperty(AttributeNamePrefix)
+ | utils::orElse([] {
+throw minifi::Exception{ExceptionType::PROCESS_SCHEDULE_EXCEPTION,
"'Attribute name prefix' must be set"};
+ })).value();
+ gsl_Ensures(runningInvariant());
+}
+
+void AttributeRollingWindow::onTrigger(core::ProcessContext* context,
core::ProcessSession* session) {
+ gsl_Expects(context && session && runningInvariant());
+ const auto flow_file = session->get();
+ if (!flow_file) { yield(); return; }
+ gsl_Assert(flow_file);
+ const auto current_value_opt = context->getProperty(ValueToTrack, flow_file);
+ if (!current_value_opt) {
+logger_->log_warn("Missing value to track, flow file uuid: {}",
flow_file->getUUIDStr());
+session->transfer(flow_file, Failure);
+return;
+ }
+ const auto current_value = [_value_opt] {
+try {
+ return std::stod(*current_value_opt);
+} catch (const std::exception& ex) {
+ throw minifi::Exception{ExceptionType::PROCESSOR_EXCEPTION,
+ fmt::format("Failed to convert 'Value to track' of '{}' to double",
*current_value_opt)};
+}
+ }();
+ // copy: so we can release the lock sooner
+ const auto state_copy = [&, now = std::chrono::system_clock::now()] {
+const std::lock_guard lg{state_mutex_};
+state_.add(now, current_value);
+if (window_length_) {
+ state_.shrinkToSize(*window_length_);
+} else {
+ gsl_Assert(time_window_);
+ state_.removeOlderThan(now - *time_window_);
+}
+return state_.getEntries();
+ }();
+ const auto sorted_values = [_copy] {
+auto values = state_copy |
ranges::views::transform((state_)::Entry::value) |
ranges::to;
+std::sort(std::begin(values), std::end(values));
+return values;
+ }();
+ calculateAndSetAttributes(*flow_file, sorted_values);
+ session->transfer(flow_file, Success);
+}
+
+/**
+ * Calculate statistical properties of the values in the rolling window and
set them as attributes on the flow file.
+ * Properties: count, value (sum), mean (average), median, variance, stddev
+ */
+void AttributeRollingWindow::calculateAndSetAttributes(core::FlowFile
_file,
+std::span sorted_values) const {
+ const auto attribute_name = [this](std::string_view suffix) {
+return utils::string::join_pack(attribute_name_prefix_, suffix);
+ };
+ const auto set_aggregate = [_file, attribute_name](std::string_view
name, double value) {
+flow_file.setAttribute(attribute_name(name), std::to_string(value));
+ };
+ set_aggregate("count", sorted_values.size());
+ const auto sum = std::accumulate(std::begin(sorted_values),
std::end(sorted_values), 0.0);
Review Comment:
I would use `ranges::accumulate`:
```suggestion
const auto sum = ranges::accumulate(sorted_values, 0.0);
```
##
extensions/standard-processors/processors/AttributeRollingWindow.cpp:
##
@@ -0,0 +1,121 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional
