lordgamez commented on code in PR #2231:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2231#discussion_r4550673337
##########
libminifi/test/unit/JsonFlowSerializerTests.cpp:
##########
@@ -1030,4 +1030,49 @@ TEST_CASE("Parameter provider generated parameter
context is reserialized if Rel
}
}
+TEST_CASE("Reporting tasks are correctly parsed from flow json") {
Review Comment:
An equivalent could be added to YamlFlowSerializerTests.cpp
##########
core-framework/include/core/Resource.h:
##########
@@ -73,6 +74,11 @@ class StaticClassType {
auto factory =
std::unique_ptr<controller::ControllerServiceFactory>(new
controller::ControllerServiceFactoryImpl<Class>(module_name));
getClassLoader().registerClass(construction_name, std::move(factory));
}
+ } else if constexpr (Type == ResourceType::ReportingTask) {
+ for (const auto& construction_name : construction_names_) {
+ auto factory = std::unique_ptr<reporting::ReportingTaskFactory>(new
reporting::ReportingTaskFactoryImpl<Class>(module_name));
+ getClassLoader().registerClass(construction_name, std::move(factory));
Review Comment:
Could this be shorter with make_unique?
```
getClassLoader().registerClass(construction_name,
std::make_unique<reporting::ReportingTaskFactoryImpl<Class>>(module_name));
```
##########
libminifi/src/core/reporting/ReportingTask.cpp:
##########
@@ -0,0 +1,100 @@
+/**
+*
+ * 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 "core/reporting/ReportingTask.h"
+
+
+namespace org::apache::nifi::minifi::core::reporting {
+
+namespace {
+
+class ReportingTaskDescriptorWrapper : public ReportingTaskDescriptor {
+ public:
+ explicit ReportingTaskDescriptorWrapper(ProcessorDescriptor& descriptor):
descriptor_(descriptor) {}
+
+ void setSupportedProperties(std::span<const PropertyReference> properties)
override {
+ descriptor_.setSupportedProperties(properties);
+ }
+
+ void setSupportedProperties(std::span<const Property> properties) override {
+ descriptor_.setSupportedProperties(properties);
+ }
+
+ private:
+ ProcessorDescriptor& descriptor_;
+};
+
+} // namespace
+
+ReportingTask::ReportingTask(std::unique_ptr<ReportingTaskApi> impl):
impl_(std::move(impl)) {}
+
+bool ReportingTask::isWorkAvailable() {
+ return false;
+}
+
+void ReportingTask::restore(const std::shared_ptr<FlowFile>&) {
+ throw std::runtime_error("Not supported");
+}
+
+[[nodiscard]] bool ReportingTask::supportsDynamicProperties() const {
+ return false;
+}
+
+[[nodiscard]] bool ReportingTask::supportsDynamicRelationships() const {
+ return false;
+}
+
+void ReportingTask::initialize(ProcessorDescriptor& descriptor) {
+ ReportingTaskDescriptorWrapper reporting_descriptor(descriptor);
+ impl_->initialize(reporting_descriptor);
+}
+
+bool ReportingTask::isSingleThreaded() const {
+ return true;
+}
+
+void ReportingTask::onTrigger(ProcessContext& context, ProcessSession&) {
+ impl_->onTrigger(context);
+}
+
+void ReportingTask::onSchedule(ProcessContext& context,
ProcessSessionFactory&) {
+ impl_->onSchedule(context);
+ context.setTriggerWhenEmpty(true);
Review Comment:
INPUT_FORBIDDEN is set so this should be redundant
##########
minifi-api/include/minifi-cpp/core/ProcessContext.h:
##########
@@ -46,14 +47,20 @@ class ProcessorInfo {
class Processor;
-class ProcessContext : public virtual core::VariableRegistry, public virtual
utils::EnableSharedFromThis {
+class ProcessContext : public reporting::ReportingTaskContext, public virtual
core::VariableRegistry, public virtual utils::EnableSharedFromThis {
Review Comment:
This inheritance relationship is strange and breaks LSP. ProcessContext
should not inherit from ReportingTaskContext, they should be separate. We
should probably extract a common ComponentContext base.
##########
libminifi/include/core/flow/FlowSchema.h:
##########
@@ -61,9 +61,6 @@ struct FlowSchema {
Keys controller_services;
Keys controller_service_properties;
Keys remote_process_group;
- Keys provenance_reporting;
- Keys provenance_reporting_port_uuid;
- Keys provenance_reporting_batch_size;
Review Comment:
With the batch size removed, is the `batch_size_` in
`SiteToSiteProvenanceReportingTask` ever set?
##########
libminifi/src/core/flow/FlowSchema.cpp:
##########
@@ -57,9 +57,6 @@ FlowSchema FlowSchema::getDefault() {
.controller_services = {"Controller Services"},
.controller_service_properties = {"Properties"},
.remote_process_group = {"Remote Processing Groups", "Remote Process
Groups"},
- .provenance_reporting = {"Provenance Reporting"},
Review Comment:
Provenance Reporting has been removed, but there are still some references
to it in tests, examples and docs like CONFIGURE.md, they should be removed too.
##########
libminifi/src/core/flow/StructuredConfiguration.cpp:
##########
@@ -1149,4 +1078,83 @@ std::string StructuredConfiguration::serialize(const
core::ProcessGroup& process
return flow_serializer_->serialize(process_group, schema_,
sensitive_values_encryptor_, {}, parameter_contexts_);
}
+void StructuredConfiguration::parseReportingTasks(const Node&
reporting_tasks_node, core::ProcessGroup* parent_group) {
+ if (!reporting_tasks_node || !reporting_tasks_node.isSequence()) {
+ return;
+ }
+ for (const auto& reporting_task_node : reporting_tasks_node) {
+ checkRequiredField(reporting_task_node, schema_.name);
+ auto name = reporting_task_node[schema_.name].getString().value();
+ utils::Identifier id;
+ id = getOrGenerateId(reporting_task_node);
+
+ checkRequiredField(reporting_task_node, schema_.type);
+ auto type = reporting_task_node[schema_.type].getString().value();
+
+ auto reporting_task =
createProcessor(utils::string::partAfterLastOccurrenceOf(type, '.'), type,
name, id);
+ if (!reporting_task) {
+ logger_->log_error("Could not create a processor {} with id {}", name,
id.to_string());
+ throw std::invalid_argument("Could not create processor " + name);
+ }
+
+ reporting_task->setFlowIdentifier(flow_version_->getFlowIdentifier());
+
+ auto scheduling_strategy = getOptionalField(reporting_task_node,
schema_.scheduling_strategy, DEFAULT_SCHEDULING_STRATEGY);
+ if (scheduling_strategy == "TIMER_DRIVEN") {
+ reporting_task->setSchedulingStrategy(core::TIMER_DRIVEN);
+ } else {
+ reporting_task->setSchedulingStrategy(core::CRON_DRIVEN);
+ }
+ auto scheduling_period_str = getOptionalField(reporting_task_node,
schema_.scheduling_period, DEFAULT_SCHEDULING_PERIOD_STR);
+ if (scheduling_strategy == "TIMER_DRIVEN") {
+ if (auto scheduling_period =
utils::timeutils::StringToDuration<std::chrono::nanoseconds>(scheduling_period_str))
{
+ reporting_task->setSchedulingPeriod(*scheduling_period);
+ }
+ } else {
+ reporting_task->setCronPeriod(scheduling_period_str);
+ }
+
+ if (auto penalization_node =
reporting_task_node[schema_.penalization_period]) {
+ if (auto penalization_period =
utils::timeutils::StringToDuration<std::chrono::milliseconds>(penalization_node.getString().value()))
{
+ reporting_task->setPenalizationPeriod(penalization_period.value());
+ }
+ }
+
+ if (auto yield_node = reporting_task_node[schema_.proc_yield_period]) {
+ if (auto yield_period =
utils::timeutils::StringToDuration<std::chrono::milliseconds>(yield_node.getString().value()))
{
+ reporting_task->setYieldPeriodMsec(yield_period.value());
+ }
+ }
+
+ if (auto bulletin_level_node =
reporting_task_node[schema_.bulletin_level]) {
+ if (auto bulletin_level = bulletin_level_node.getString().value();
!bulletin_level.empty()) {
+
reporting_task->setLogBulletinLevel(core::logging::mapStringToLogLevel(bulletin_level));
+ }
+ }
+
+ if (auto run_node = reporting_task_node[schema_.runduration_nanos]) {
+ if (auto run_duration_nanos =
parsing::parseIntegral<uint64_t>(run_node.getIntegerAsString().value())) {
+
reporting_task->setRunDurationNano(std::chrono::nanoseconds(*run_duration_nanos));
+ }
+ }
+
+ if (Node properties_node =
reporting_task_node[schema_.processor_properties]) {
Review Comment:
This should be using `schema_.reporting_task_properties`
##########
minifi-api/include/minifi-cpp/core/reporting/ReportingTaskFactory.h:
##########
@@ -0,0 +1,39 @@
+/**
+* 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 <memory>
+#include <string>
+#include <utility>
+
+#include "minifi-cpp/core/reporting/ReportingTaskMetadata.h"
+
+namespace org::apache::nifi::minifi::core::reporting {
+
+class ReportingTaskApi;
+
+class ReportingTaskFactory {
+ public:
+ virtual std::unique_ptr<ReportingTaskApi> create(ReportingTaskMetadata
metadata) = 0;
Review Comment:
The metadata parameter should probably be const ref.
##########
minifi-api/include/minifi-cpp/agent/agent_docs.h:
##########
@@ -38,7 +38,8 @@ enum class ResourceType {
ControllerService,
InternalResource,
DescriptionOnly,
- ParameterProvider
+ ParameterProvider,
+ ReportingTask
Review Comment:
Will the agent_docs be extended in the future to be part of the C2 manifest?
##########
libminifi/src/core/flow/StructuredConfiguration.cpp:
##########
@@ -1149,4 +1078,83 @@ std::string StructuredConfiguration::serialize(const
core::ProcessGroup& process
return flow_serializer_->serialize(process_group, schema_,
sensitive_values_encryptor_, {}, parameter_contexts_);
}
+void StructuredConfiguration::parseReportingTasks(const Node&
reporting_tasks_node, core::ProcessGroup* parent_group) {
+ if (!reporting_tasks_node || !reporting_tasks_node.isSequence()) {
+ return;
+ }
+ for (const auto& reporting_task_node : reporting_tasks_node) {
+ checkRequiredField(reporting_task_node, schema_.name);
+ auto name = reporting_task_node[schema_.name].getString().value();
+ utils::Identifier id;
+ id = getOrGenerateId(reporting_task_node);
+
+ checkRequiredField(reporting_task_node, schema_.type);
+ auto type = reporting_task_node[schema_.type].getString().value();
+
+ auto reporting_task =
createProcessor(utils::string::partAfterLastOccurrenceOf(type, '.'), type,
name, id);
+ if (!reporting_task) {
+ logger_->log_error("Could not create a processor {} with id {}", name,
id.to_string());
+ throw std::invalid_argument("Could not create processor " + name);
+ }
+
+ reporting_task->setFlowIdentifier(flow_version_->getFlowIdentifier());
+
+ auto scheduling_strategy = getOptionalField(reporting_task_node,
schema_.scheduling_strategy, DEFAULT_SCHEDULING_STRATEGY);
+ if (scheduling_strategy == "TIMER_DRIVEN") {
+ reporting_task->setSchedulingStrategy(core::TIMER_DRIVEN);
+ } else {
+ reporting_task->setSchedulingStrategy(core::CRON_DRIVEN);
+ }
+ auto scheduling_period_str = getOptionalField(reporting_task_node,
schema_.scheduling_period, DEFAULT_SCHEDULING_PERIOD_STR);
+ if (scheduling_strategy == "TIMER_DRIVEN") {
+ if (auto scheduling_period =
utils::timeutils::StringToDuration<std::chrono::nanoseconds>(scheduling_period_str))
{
+ reporting_task->setSchedulingPeriod(*scheduling_period);
+ }
+ } else {
+ reporting_task->setCronPeriod(scheduling_period_str);
+ }
+
+ if (auto penalization_node =
reporting_task_node[schema_.penalization_period]) {
+ if (auto penalization_period =
utils::timeutils::StringToDuration<std::chrono::milliseconds>(penalization_node.getString().value()))
{
+ reporting_task->setPenalizationPeriod(penalization_period.value());
+ }
+ }
+
+ if (auto yield_node = reporting_task_node[schema_.proc_yield_period]) {
+ if (auto yield_period =
utils::timeutils::StringToDuration<std::chrono::milliseconds>(yield_node.getString().value()))
{
+ reporting_task->setYieldPeriodMsec(yield_period.value());
+ }
+ }
+
+ if (auto bulletin_level_node =
reporting_task_node[schema_.bulletin_level]) {
+ if (auto bulletin_level = bulletin_level_node.getString().value();
!bulletin_level.empty()) {
+
reporting_task->setLogBulletinLevel(core::logging::mapStringToLogLevel(bulletin_level));
+ }
+ }
+
+ if (auto run_node = reporting_task_node[schema_.runduration_nanos]) {
+ if (auto run_duration_nanos =
parsing::parseIntegral<uint64_t>(run_node.getIntegerAsString().value())) {
+
reporting_task->setRunDurationNano(std::chrono::nanoseconds(*run_duration_nanos));
+ }
+ }
+
+ if (Node properties_node =
reporting_task_node[schema_.processor_properties]) {
+ parsePropertiesNode(properties_node, *reporting_task, name, nullptr);
+ }
+
+ reporting_task->setLoggerCallback([this, processor =
reporting_task.get()](core::logging::LOG_LEVEL level, const std::string&
message) {
+ if (level < processor->getLogBulletinLevel()) {
+ return;
+ }
+ if (bulletin_store_) {
+ bulletin_store_->addProcessorBulletin(*processor, level, message);
+ }
+ });
Review Comment:
I think this is not needed as the `forEachLogger` implementation is empty
for reporting tasks.
##########
behave_framework/src/minifi_behave/minifi/flow_definition.py:
##########
@@ -25,6 +25,9 @@
from .parameter_context import ParameterContext
from .processor import Processor
from .remote_process_group import RemoteProcessGroup
+from .reporting_task import ReportingTask
+from .input_port import InputPort
+from .output_port import OutputPort
Review Comment:
These were already imported
##########
libminifi/test/libtest/unit/ReportingTaskUtils.h:
##########
@@ -0,0 +1,47 @@
+/**
+*
+ * 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 "core/reporting/ReportingTask.h"
+
+namespace org::apache::nifi::minifi::test::utils {
+
+template<typename T>
+std::unique_ptr<core::Processor> make_reporting_task(std::string_view name,
std::optional<minifi::utils::Identifier> uuid = std::nullopt) {
+ if (!uuid) {
+ uuid = minifi::utils::IdGenerator::getIdGenerator()->generate();
+ }
+ auto reporting_task_impl =
std::make_unique<T>(core::reporting::ReportingTaskMetadata{
+ .uuid = uuid.value(),
+ .name = std::string{name},
+ .logger =
minifi::core::logging::LoggerFactory<T>::getLogger(uuid.value())
+ });
+ return
std::make_unique<core::Processor>(minifi::utils::string::partAfterLastOccurrenceOf(core::className<T>(),
':'), name, uuid.value(),
+
std::make_unique<minifi::core::reporting::ReportingTask>(std::move(reporting_task_impl)));
+}
+
+template<typename T, typename ...Args>
+std::unique_ptr<core::Processor> make_custom_reporting_task(Args&&... args) {
Review Comment:
This seems to be unused.
--
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]
