You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2025/01/31 09:04:35 UTC
(nifi-minifi-cpp) branch minifi-api updated (001d12c8f -> 47a6468b5)
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a change to branch minifi-api
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
omit 001d12c8f Rebase fix
new 47a6468b5 Rebase fix
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (001d12c8f)
\
N -- N -- N refs/heads/minifi-api (47a6468b5)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
(nifi-minifi-cpp) 01/01: Rebase fix
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch minifi-api
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 47a6468b5c0fc9ea5f1355f4f658e5c5571b610d
Author: Martin Zink <ma...@apache.org>
AuthorDate: Thu Jan 23 16:13:39 2025 +0100
Rebase fix
---
libminifi/include/core/ProcessSession.h | 2 ++
libminifi/src/core/ProcessSession.cpp | 22 ++++++------
libminifi/src/core/state/nodes/FlowInformation.cpp | 32 ++++++++---------
libminifi/test/unit/MetricsTests.cpp | 42 +++++++++++-----------
.../include/minifi-cpp/core/ProcessSession.h | 2 ++
minifi-api/include/minifi-cpp/core/Processor.h | 2 ++
.../include/minifi-cpp/core/ProcessorMetrics.h | 29 +++++++--------
utils/include/core/Processor.h | 4 +--
utils/include/core/ProcessorMetrics.h | 42 +++++++++++-----------
utils/src/core/Processor.cpp | 2 +-
utils/src/core/ProcessorMetrics.cpp | 30 ++++++++--------
11 files changed, 108 insertions(+), 101 deletions(-)
diff --git a/libminifi/include/core/ProcessSession.h b/libminifi/include/core/ProcessSession.h
index b424acbce..38665c662 100644
--- a/libminifi/include/core/ProcessSession.h
+++ b/libminifi/include/core/ProcessSession.h
@@ -110,6 +110,8 @@ class ProcessSessionImpl : public ReferenceContainerImpl, public virtual Process
void importFrom(io::InputStream&& stream, const std::shared_ptr<core::FlowFile> &flow) override;
void import(const std::string& source, const std::shared_ptr<core::FlowFile> &flow, bool keepSource = true, uint64_t offset = 0) override;
+ void import(const std::string& source, std::vector<std::shared_ptr<FlowFile>> &flows, uint64_t offset, char inputDelimiter) override;
+ void import(const std::string& source, std::vector<std::shared_ptr<FlowFile>> &flows, bool keepSource, uint64_t offset, char inputDelimiter) override;
bool exportContent(const std::string &destination, const std::shared_ptr<core::FlowFile> &flow, bool keepContent) override;
diff --git a/libminifi/src/core/ProcessSession.cpp b/libminifi/src/core/ProcessSession.cpp
index 0aa27752a..5526cda69 100644
--- a/libminifi/src/core/ProcessSession.cpp
+++ b/libminifi/src/core/ProcessSession.cpp
@@ -267,7 +267,7 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const io::OutputStreamCallb
auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - start_time);
provenance_report_->modifyContent(flow, details, duration);
if (metrics_) {
- metrics_->bytes_written += stream->size();
+ metrics_->bytesWritten() += stream->size();
}
} catch (const std::exception& exception) {
logger_->log_debug("Caught Exception during process session write, type: {}, what: {}", typeid(exception).name(), exception.what());
@@ -319,7 +319,7 @@ void ProcessSessionImpl::append(const std::shared_ptr<core::FlowFile> &flow, con
}
flow->setSize(flow_file_size + (stream->size() - stream_size_before_callback));
if (metrics_) {
- metrics_->bytes_written += stream->size() - stream_size_before_callback;
+ metrics_->bytesWritten() += stream->size() - stream_size_before_callback;
}
std::stringstream details;
@@ -379,7 +379,7 @@ int64_t ProcessSessionImpl::read(const core::FlowFile& flow_file, const io::Inpu
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to process flowfile content");
}
if (metrics_) {
- metrics_->bytes_read += ret;
+ metrics_->bytesRead() += ret;
}
return ret;
} catch (const std::exception& exception) {
@@ -428,8 +428,8 @@ int64_t ProcessSessionImpl::readWrite(const std::shared_ptr<core::FlowFile> &flo
flow->setOffset(0);
flow->setResourceClaim(output_claim);
if (metrics_) {
- metrics_->bytes_written += read_write_result->bytes_written;
- metrics_->bytes_read += read_write_result->bytes_read;
+ metrics_->bytesWritten() += read_write_result->bytes_written;
+ metrics_->bytesRead() += read_write_result->bytes_read;
}
return read_write_result->bytes_written;
@@ -498,7 +498,7 @@ void ProcessSessionImpl::importFrom(io::InputStream &stream, const std::shared_p
content_stream->close();
if (metrics_) {
- metrics_->bytes_written += content_stream->size();
+ metrics_->bytesWritten() += content_stream->size();
}
std::stringstream details;
details << process_context_->getProcessorNode()->getName() << " modify flow record content " << flow->getUUIDStr();
@@ -562,7 +562,7 @@ void ProcessSessionImpl::import(const std::string& source, const std::shared_ptr
stream->close();
if (metrics_) {
- metrics_->bytes_written += stream->size();
+ metrics_->bytesWritten() += stream->size();
}
input.close();
if (!keepSource) {
@@ -667,7 +667,7 @@ void ProcessSessionImpl::import(const std::string& source, std::vector<std::shar
flowFile->getOffset(), flowFile->getSize(), flowFile->getResourceClaim()->getContentFullPath(), flowFile->getUUIDStr());
stream->close();
if (metrics_) {
- metrics_->bytes_written += stream->size();
+ metrics_->bytesWritten() += stream->size();
}
std::string details = process_context_->getProcessorNode()->getName() + " modify flow record content " + flowFile->getUUIDStr();
auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - start_time);
@@ -953,7 +953,7 @@ void ProcessSessionImpl::commit() {
if (metrics_) {
auto time_delta = std::chrono::steady_clock::now() - commit_start_time;
metrics_->addLastSessionCommitRuntime(std::chrono::duration_cast<std::chrono::milliseconds>(time_delta));
- metrics_->processing_nanos += std::chrono::duration_cast<std::chrono::nanoseconds>(time_delta).count();
+ metrics_->processingNanos() += std::chrono::duration_cast<std::chrono::nanoseconds>(time_delta).count();
}
} catch (const std::exception& exception) {
logger_->log_debug("Caught Exception during process session commit, type: {}, what: {}", typeid(exception).name(), exception.what());
@@ -1165,8 +1165,8 @@ std::shared_ptr<core::FlowFile> ProcessSessionImpl::get() {
ret->setAttribute(SpecialFlowAttribute::FLOW_ID, flow_version->getFlowId());
}
if (metrics_) {
- metrics_->incoming_bytes += ret->getSize();
- ++metrics_->incoming_flow_files;
+ metrics_->incomingBytes() += ret->getSize();
+ ++metrics_->incomingFlowFiles();
}
return ret;
}
diff --git a/libminifi/src/core/state/nodes/FlowInformation.cpp b/libminifi/src/core/state/nodes/FlowInformation.cpp
index f31b26c29..de8be246f 100644
--- a/libminifi/src/core/state/nodes/FlowInformation.cpp
+++ b/libminifi/src/core/state/nodes/FlowInformation.cpp
@@ -81,14 +81,14 @@ std::vector<SerializedResponseNode> FlowInformation::serialize() {
.children = {
{.name = "id", .value = std::string{processor->getUUIDStr()}},
{.name = "groupId", .value = processor->getProcessGroupUUIDStr()},
- {.name = "bytesRead", .value = metrics->bytes_read.load()},
- {.name = "bytesWritten", .value = metrics->bytes_written.load()},
- {.name = "flowFilesIn", .value = metrics->incoming_flow_files.load()},
- {.name = "flowFilesOut", .value = metrics->transferred_flow_files.load()},
- {.name = "bytesIn", .value = metrics->incoming_bytes.load()},
- {.name = "bytesOut", .value = metrics->transferred_bytes.load()},
- {.name = "invocations", .value = metrics->invocations.load()},
- {.name = "processingNanos", .value = metrics->processing_nanos.load()},
+ {.name = "bytesRead", .value = metrics->bytesRead().load()},
+ {.name = "bytesWritten", .value = metrics->bytesWritten().load()},
+ {.name = "flowFilesIn", .value = metrics->incomingFlowFiles().load()},
+ {.name = "flowFilesOut", .value = metrics->transferredFlowFiles().load()},
+ {.name = "bytesIn", .value = metrics->incomingBytes().load()},
+ {.name = "bytesOut", .value = metrics->transferredBytes().load()},
+ {.name = "invocations", .value = metrics->invocations().load()},
+ {.name = "processingNanos", .value = metrics->processingNanos().load()},
{.name = "activeThreadCount", .value = -1},
{.name = "terminatedThreadCount", .value = -1},
{.name = "runStatus", .value = (processor->isRunning() ? "Running" : "Stopped")}
@@ -115,21 +115,21 @@ std::vector<PublishedMetric> FlowInformation::calculateMetrics() {
continue;
}
auto processor_metrics = processor->getMetrics();
- metrics.push_back({"bytes_read", gsl::narrow<double>(processor_metrics->bytes_read.load()),
+ metrics.push_back({"bytes_read", gsl::narrow<double>(processor_metrics->bytesRead().load()),
{{"processor_uuid", processor->getUUIDStr()}, {"processor_name", processor->getName()}, {"metric_class", "FlowInformation"}}});
- metrics.push_back({"bytes_written", gsl::narrow<double>(processor_metrics->bytes_written.load()),
+ metrics.push_back({"bytes_written", gsl::narrow<double>(processor_metrics->bytesWritten().load()),
{{"processor_uuid", processor->getUUIDStr()}, {"processor_name", processor->getName()}, {"metric_class", "FlowInformation"}}});
- metrics.push_back({"flow_files_in", gsl::narrow<double>(processor_metrics->incoming_flow_files.load()),
+ metrics.push_back({"flow_files_in", gsl::narrow<double>(processor_metrics->incomingFlowFiles().load()),
{{"processor_uuid", processor->getUUIDStr()}, {"processor_name", processor->getName()}, {"metric_class", "FlowInformation"}}});
- metrics.push_back({"flow_files_out", gsl::narrow<double>(processor_metrics->transferred_flow_files.load()),
+ metrics.push_back({"flow_files_out", gsl::narrow<double>(processor_metrics->transferredFlowFiles().load()),
{{"processor_uuid", processor->getUUIDStr()}, {"processor_name", processor->getName()}, {"metric_class", "FlowInformation"}}});
- metrics.push_back({"bytes_in", gsl::narrow<double>(processor_metrics->incoming_bytes.load()),
+ metrics.push_back({"bytes_in", gsl::narrow<double>(processor_metrics->incomingBytes().load()),
{{"processor_uuid", processor->getUUIDStr()}, {"processor_name", processor->getName()}, {"metric_class", "FlowInformation"}}});
- metrics.push_back({"bytes_out", gsl::narrow<double>(processor_metrics->transferred_bytes.load()),
+ metrics.push_back({"bytes_out", gsl::narrow<double>(processor_metrics->transferredBytes().load()),
{{"processor_uuid", processor->getUUIDStr()}, {"processor_name", processor->getName()}, {"metric_class", "FlowInformation"}}});
- metrics.push_back({"invocations", gsl::narrow<double>(processor_metrics->invocations.load()),
+ metrics.push_back({"invocations", gsl::narrow<double>(processor_metrics->invocations().load()),
{{"processor_uuid", processor->getUUIDStr()}, {"processor_name", processor->getName()}, {"metric_class", "FlowInformation"}}});
- metrics.push_back({"processing_nanos", gsl::narrow<double>(processor_metrics->processing_nanos.load()),
+ metrics.push_back({"processing_nanos", gsl::narrow<double>(processor_metrics->processingNanos().load()),
{{"processor_uuid", processor->getUUIDStr()}, {"processor_name", processor->getName()}, {"metric_class", "FlowInformation"}}});
metrics.push_back({"is_running", (processor->isRunning() ? 1.0 : 0.0),
{{"processor_uuid", processor->getUUIDStr()}, {"processor_name", processor->getName()}, {"metric_class", "FlowInformation"}}});
diff --git a/libminifi/test/unit/MetricsTests.cpp b/libminifi/test/unit/MetricsTests.cpp
index f8c986af4..806970b07 100644
--- a/libminifi/test/unit/MetricsTests.cpp
+++ b/libminifi/test/unit/MetricsTests.cpp
@@ -287,12 +287,12 @@ TEST_CASE("Test commit runtime processor metrics", "[ProcessorMetrics]") {
REQUIRE(metrics.getAverageSessionCommitRuntime() == 37ms);
}
-class DuplicateContentProcessor : public minifi::core::Processor {
- using minifi::core::Processor::Processor;
+class DuplicateContentProcessor : public minifi::core::ProcessorImpl {
+ using minifi::core::ProcessorImpl::ProcessorImpl;
public:
- DuplicateContentProcessor(std::string_view name, const minifi::utils::Identifier& uuid) : Processor(name, uuid) {}
- explicit DuplicateContentProcessor(std::string_view name) : Processor(name) {}
+ DuplicateContentProcessor(std::string_view name, const minifi::utils::Identifier& uuid) : ProcessorImpl(name, uuid) {}
+ explicit DuplicateContentProcessor(std::string_view name) : ProcessorImpl(name) {}
static constexpr const char* Description = "A processor that creates two more of the same flow file.";
static constexpr auto Properties = std::array<core::PropertyReference, 0>{};
static constexpr auto Success = core::RelationshipDefinition{"success", "Newly created FlowFiles"};
@@ -333,29 +333,29 @@ TEST_CASE("Test processor metrics change after trigger", "[ProcessorMetrics]") {
minifi::test::SingleProcessorTestController test_controller(std::make_unique<DuplicateContentProcessor>("DuplicateContentProcessor"));
test_controller.trigger({minifi::test::InputFlowFileData{"log line 1", {}}});
auto metrics = test_controller.getProcessor()->getMetrics();
- CHECK(metrics->invocations == 1);
- CHECK(metrics->incoming_flow_files == 1);
- CHECK(metrics->transferred_flow_files == 2);
+ CHECK(metrics->invocations() == 1);
+ CHECK(metrics->incomingFlowFiles() == 1);
+ CHECK(metrics->transferredFlowFiles() == 2);
CHECK(metrics->getTransferredFlowFilesToRelationshipCount("success") == 1);
CHECK(metrics->getTransferredFlowFilesToRelationshipCount("original") == 1);
- CHECK(metrics->incoming_bytes == 10);
- CHECK(metrics->transferred_bytes == 30);
- CHECK(metrics->bytes_read == 10);
- CHECK(metrics->bytes_written == 20);
- auto old_nanos = metrics->processing_nanos.load();
- CHECK(metrics->processing_nanos > 0);
+ CHECK(metrics->incomingBytes() == 10);
+ CHECK(metrics->transferredBytes() == 30);
+ CHECK(metrics->bytesRead() == 10);
+ CHECK(metrics->bytesWritten() == 20);
+ auto old_nanos = metrics->processingNanos().load();
+ CHECK(metrics->processingNanos() > 0);
test_controller.trigger({minifi::test::InputFlowFileData{"new log line 2", {}}});
- CHECK(metrics->invocations == 2);
- CHECK(metrics->incoming_flow_files == 2);
- CHECK(metrics->transferred_flow_files == 4);
+ CHECK(metrics->invocations() == 2);
+ CHECK(metrics->incomingFlowFiles() == 2);
+ CHECK(metrics->transferredFlowFiles() == 4);
CHECK(metrics->getTransferredFlowFilesToRelationshipCount("success") == 2);
CHECK(metrics->getTransferredFlowFilesToRelationshipCount("original") == 2);
- CHECK(metrics->incoming_bytes == 24);
- CHECK(metrics->transferred_bytes == 72);
- CHECK(metrics->bytes_read == 24);
- CHECK(metrics->bytes_written == 48);
- CHECK(metrics->processing_nanos > old_nanos);
+ CHECK(metrics->incomingBytes() == 24);
+ CHECK(metrics->transferredBytes() == 72);
+ CHECK(metrics->bytesRead() == 24);
+ CHECK(metrics->bytesWritten() == 48);
+ CHECK(metrics->processingNanos() > old_nanos);
}
diff --git a/minifi-api/include/minifi-cpp/core/ProcessSession.h b/minifi-api/include/minifi-cpp/core/ProcessSession.h
index e19af3b18..196357e21 100644
--- a/minifi-api/include/minifi-cpp/core/ProcessSession.h
+++ b/minifi-api/include/minifi-cpp/core/ProcessSession.h
@@ -97,6 +97,8 @@ class ProcessSession : public virtual ReferenceContainer {
// import from the data source.
virtual void import(const std::string& source, const std::shared_ptr<core::FlowFile> &flow, bool keepSource = true, uint64_t offset = 0) = 0;
+ virtual void import(const std::string& source, std::vector<std::shared_ptr<FlowFile>> &flows, uint64_t offset, char inputDelimiter) = 0;
+ virtual void import(const std::string& source, std::vector<std::shared_ptr<FlowFile>> &flows, bool keepSource, uint64_t offset, char inputDelimiter) = 0;
/**
* Exports the data stream to a file
diff --git a/minifi-api/include/minifi-cpp/core/Processor.h b/minifi-api/include/minifi-cpp/core/Processor.h
index 092df1426..52b9ab6c9 100644
--- a/minifi-api/include/minifi-cpp/core/Processor.h
+++ b/minifi-api/include/minifi-cpp/core/Processor.h
@@ -83,6 +83,8 @@ class Processor : public virtual Connectable, public virtual ConfigurableCompone
virtual void validateAnnotations() const = 0;
virtual annotation::Input getInputRequirement() const = 0;
virtual gsl::not_null<std::shared_ptr<ProcessorMetrics>> getMetrics() const = 0;
+ virtual std::string getProcessGroupUUIDStr() const = 0;
+ virtual void setProcessGroupUUIDStr(const std::string &uuid) = 0;
virtual void updateReachability(const std::lock_guard<std::mutex>& graph_lock, bool force = false) = 0;
virtual const std::unordered_map<Connection*, std::unordered_set<Processor*>>& reachable_processors() const = 0;
diff --git a/minifi-api/include/minifi-cpp/core/ProcessorMetrics.h b/minifi-api/include/minifi-cpp/core/ProcessorMetrics.h
index 5e4825731..8d19c1cd1 100644
--- a/minifi-api/include/minifi-cpp/core/ProcessorMetrics.h
+++ b/minifi-api/include/minifi-cpp/core/ProcessorMetrics.h
@@ -34,23 +34,24 @@ class ProcessorMetrics : public virtual state::response::ResponseNode {
virtual std::chrono::milliseconds getAverageSessionCommitRuntime() const = 0;
virtual std::chrono::milliseconds getLastSessionCommitRuntime() const = 0;
virtual void addLastSessionCommitRuntime(std::chrono::milliseconds runtime) = 0;
+ virtual std::optional<size_t> getTransferredFlowFilesToRelationshipCount(const std::string& relationship) const = 0;
virtual std::atomic<size_t>& invocations() = 0;
virtual const std::atomic<size_t>& invocations() const = 0;
- virtual std::atomic<size_t>& incoming_flow_files() = 0;
- virtual const std::atomic<size_t>& incoming_flow_files() const = 0;
- virtual std::atomic<size_t>& transferred_flow_files() = 0;
- virtual const std::atomic<size_t>& transferred_flow_files() const = 0;
- virtual std::atomic<uint64_t>& incoming_bytes() = 0;
- virtual const std::atomic<uint64_t>& incoming_bytes() const = 0;
- virtual std::atomic<uint64_t>& transferred_bytes() = 0;
- virtual const std::atomic<uint64_t>& transferred_bytes() const = 0;
- virtual std::atomic<uint64_t>& bytes_read() = 0;
- virtual const std::atomic<uint64_t>& bytes_read() const = 0;
- virtual std::atomic<uint64_t>& bytes_written() = 0;
- virtual const std::atomic<uint64_t>& bytes_written() const = 0;
- virtual std::atomic<uint64_t>& processing_nanos() = 0;
- virtual const std::atomic<uint64_t>& processing_nanos() const = 0;
+ virtual std::atomic<size_t>& incomingFlowFiles() = 0;
+ virtual const std::atomic<size_t>& incomingFlowFiles() const = 0;
+ virtual std::atomic<size_t>& transferredFlowFiles() = 0;
+ virtual const std::atomic<size_t>& transferredFlowFiles() const = 0;
+ virtual std::atomic<uint64_t>& incomingBytes() = 0;
+ virtual const std::atomic<uint64_t>& incomingBytes() const = 0;
+ virtual std::atomic<uint64_t>& transferredBytes() = 0;
+ virtual const std::atomic<uint64_t>& transferredBytes() const = 0;
+ virtual std::atomic<uint64_t>& bytesRead() = 0;
+ virtual const std::atomic<uint64_t>& bytesRead() const = 0;
+ virtual std::atomic<uint64_t>& bytesWritten() = 0;
+ virtual const std::atomic<uint64_t>& bytesWritten() const = 0;
+ virtual std::atomic<uint64_t>& processingNanos() = 0;
+ virtual const std::atomic<uint64_t>& processingNanos() const = 0;
};
} // namespace org::apache::nifi::minifi::core
diff --git a/utils/include/core/Processor.h b/utils/include/core/Processor.h
index 336f13a59..4a17fcc9b 100644
--- a/utils/include/core/Processor.h
+++ b/utils/include/core/Processor.h
@@ -167,11 +167,11 @@ class ProcessorImpl : public virtual Processor, public ConnectableImpl, public C
active_tasks_ = 0;
}
- std::string getProcessGroupUUIDStr() const {
+ std::string getProcessGroupUUIDStr() const override {
return process_group_uuid_;
}
- void setProcessGroupUUIDStr(const std::string &uuid) {
+ void setProcessGroupUUIDStr(const std::string &uuid) override {
process_group_uuid_ = uuid;
}
diff --git a/utils/include/core/ProcessorMetrics.h b/utils/include/core/ProcessorMetrics.h
index 3d1ebfb45..030d2ae34 100644
--- a/utils/include/core/ProcessorMetrics.h
+++ b/utils/include/core/ProcessorMetrics.h
@@ -49,27 +49,27 @@ class ProcessorMetricsImpl : public state::response::ResponseNodeImpl, public vi
std::chrono::milliseconds getLastOnTriggerRuntime() const override;
void addLastOnTriggerRuntime(std::chrono::milliseconds runtime) override;
- std::chrono::milliseconds getAverageSessionCommitRuntime() const;
- std::chrono::milliseconds getLastSessionCommitRuntime() const;
- void addLastSessionCommitRuntime(std::chrono::milliseconds runtime);
- std::optional<size_t> getTransferredFlowFilesToRelationshipCount(const std::string& relationship) const;
-
- std::atomic<size_t>& invocations() {return invocations_;}
- const std::atomic<size_t>& invocations() const {return invocations_;}
- std::atomic<size_t>& incoming_flow_files() {return incoming_flow_files_;}
- const std::atomic<size_t>& incoming_flow_files() const {return incoming_flow_files_;}
- std::atomic<size_t>& transferred_flow_files() {return transferred_flow_files_;}
- const std::atomic<size_t>& transferred_flow_files() const {return transferred_flow_files_;}
- std::atomic<uint64_t>& incoming_bytes() {return incoming_bytes_;}
- const std::atomic<uint64_t>& incoming_bytes() const {return incoming_bytes_;}
- std::atomic<uint64_t>& transferred_bytes() {return transferred_bytes_;}
- const std::atomic<uint64_t>& transferred_bytes() const {return transferred_bytes_;}
- std::atomic<uint64_t>& bytes_read() {return bytes_read_;}
- const std::atomic<uint64_t>& bytes_read() const {return bytes_read_;}
- std::atomic<uint64_t>& bytes_written() {return bytes_written_;}
- const std::atomic<uint64_t>& bytes_written() const {return bytes_written_;}
- std::atomic<uint64_t>& processing_nanos() {return processing_nanos_;}
- const std::atomic<uint64_t>& processing_nanos() const {return processing_nanos_;}
+ std::chrono::milliseconds getAverageSessionCommitRuntime() const override;
+ std::chrono::milliseconds getLastSessionCommitRuntime() const override;
+ void addLastSessionCommitRuntime(std::chrono::milliseconds runtime) override;
+ std::optional<size_t> getTransferredFlowFilesToRelationshipCount(const std::string& relationship) const override;
+
+ std::atomic<size_t>& invocations() override {return invocations_;}
+ const std::atomic<size_t>& invocations() const override {return invocations_;}
+ std::atomic<size_t>& incomingFlowFiles() override {return incoming_flow_files_;}
+ const std::atomic<size_t>& incomingFlowFiles() const override {return incoming_flow_files_;}
+ std::atomic<size_t>& transferredFlowFiles() override {return transferred_flow_files_;}
+ const std::atomic<size_t>& transferredFlowFiles() const override {return transferred_flow_files_;}
+ std::atomic<uint64_t>& incomingBytes() override {return incoming_bytes_;}
+ const std::atomic<uint64_t>& incomingBytes() const override {return incoming_bytes_;}
+ std::atomic<uint64_t>& transferredBytes() override {return transferred_bytes_;}
+ const std::atomic<uint64_t>& transferredBytes() const override {return transferred_bytes_;}
+ std::atomic<uint64_t>& bytesRead() override {return bytes_read_;}
+ const std::atomic<uint64_t>& bytesRead() const override {return bytes_read_;}
+ std::atomic<uint64_t>& bytesWritten() override {return bytes_written_;}
+ const std::atomic<uint64_t>& bytesWritten() const override {return bytes_written_;}
+ std::atomic<uint64_t>& processingNanos() override {return processing_nanos_;}
+ const std::atomic<uint64_t>& processingNanos() const override {return processing_nanos_;}
protected:
template<typename ValueType>
diff --git a/utils/src/core/Processor.cpp b/utils/src/core/Processor.cpp
index 1e17f80d4..cf2eb6b53 100644
--- a/utils/src/core/Processor.cpp
+++ b/utils/src/core/Processor.cpp
@@ -201,7 +201,7 @@ void ProcessorImpl::triggerAndCommit(const std::shared_ptr<ProcessContext>& cont
}
void ProcessorImpl::trigger(const std::shared_ptr<ProcessContext>& context, const std::shared_ptr<ProcessSession>& process_session) {
- ++metrics_->invocations;
+ ++metrics_->invocations();
const auto start = std::chrono::steady_clock::now();
onTrigger(*context, *process_session);
metrics_->addLastOnTriggerRuntime(std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - start));
diff --git a/utils/src/core/ProcessorMetrics.cpp b/utils/src/core/ProcessorMetrics.cpp
index 81e66c2c9..f615be769 100644
--- a/utils/src/core/ProcessorMetrics.cpp
+++ b/utils/src/core/ProcessorMetrics.cpp
@@ -49,13 +49,13 @@ std::vector<state::response::SerializedResponseNode> ProcessorMetricsImpl::seria
{.name = "LastOnTriggerRunTime", .value = static_cast<uint64_t>(getLastOnTriggerRuntime().count())},
{.name = "AverageSessionCommitRunTime", .value = static_cast<uint64_t>(getAverageSessionCommitRuntime().count())},
{.name = "LastSessionCommitRunTime", .value = static_cast<uint64_t>(getLastSessionCommitRuntime().count())},
- {.name = "TransferredFlowFiles", .value = static_cast<uint32_t>(transferred_flow_files())},
- {.name = "TransferredBytes", .value = static_cast<uint64_t>(transferred_bytes())},
- {.name = "IncomingFlowFiles", .value = static_cast<uint32_t>(incoming_flow_files())},
- {.name = "IncomingBytes", .value = static_cast<uint64_t>(incoming_bytes())},
- {.name = "BytesRead", .value = static_cast<uint64_t>(bytes_read())},
- {.name = "BytesWritten", .value = static_cast<uint64_t>(bytes_written())},
- {.name = "ProcessingNanos", .value = static_cast<uint64_t>(processing_nanos())}
+ {.name = "TransferredFlowFiles", .value = static_cast<uint32_t>(transferredFlowFiles())},
+ {.name = "TransferredBytes", .value = static_cast<uint64_t>(transferredBytes())},
+ {.name = "IncomingFlowFiles", .value = static_cast<uint32_t>(incomingFlowFiles())},
+ {.name = "IncomingBytes", .value = static_cast<uint64_t>(incomingBytes())},
+ {.name = "BytesRead", .value = static_cast<uint64_t>(bytesRead())},
+ {.name = "BytesWritten", .value = static_cast<uint64_t>(bytesWritten())},
+ {.name = "ProcessingNanos", .value = static_cast<uint64_t>(processingNanos())}
}
};
@@ -83,13 +83,13 @@ std::vector<state::PublishedMetric> ProcessorMetricsImpl::calculateMetrics() {
{"last_onTrigger_runtime_milliseconds", static_cast<double>(getLastOnTriggerRuntime().count()), getCommonLabels()},
{"average_session_commit_runtime_milliseconds", static_cast<double>(getAverageSessionCommitRuntime().count()), getCommonLabels()},
{"last_session_commit_runtime_milliseconds", static_cast<double>(getLastSessionCommitRuntime().count()), getCommonLabels()},
- {"transferred_flow_files", static_cast<double>(transferred_flow_files()), getCommonLabels()},
- {"transferred_bytes", static_cast<double>(transferred_bytes()), getCommonLabels()},
- {"incoming_flow_files", static_cast<double>(incoming_flow_files()), getCommonLabels()},
- {"incoming_bytes", static_cast<double>(incoming_bytes()), getCommonLabels()},
- {"bytes_read", static_cast<double>(bytes_read()), getCommonLabels()},
- {"bytes_written", static_cast<double>(bytes_written()), getCommonLabels()},
- {"processing_nanos", static_cast<double>(processing_nanos()), getCommonLabels()}
+ {"transferred_flow_files", static_cast<double>(transferredFlowFiles()), getCommonLabels()},
+ {"transferred_bytes", static_cast<double>(transferredBytes()), getCommonLabels()},
+ {"incoming_flow_files", static_cast<double>(incomingFlowFiles()), getCommonLabels()},
+ {"incoming_bytes", static_cast<double>(incomingBytes()), getCommonLabels()},
+ {"bytes_read", static_cast<double>(bytesRead()), getCommonLabels()},
+ {"bytes_written", static_cast<double>(bytesWritten()), getCommonLabels()},
+ {"processing_nanos", static_cast<double>(processingNanos()), getCommonLabels()}
};
{
@@ -132,7 +132,7 @@ std::chrono::milliseconds ProcessorMetricsImpl::getLastSessionCommitRuntime() co
return session_commit_runtime_averager_.getLastValue();
}
-std::optional<size_t> ProcessorMetrics::getTransferredFlowFilesToRelationshipCount(const std::string& relationship) const {
+std::optional<size_t> ProcessorMetricsImpl::getTransferredFlowFilesToRelationshipCount(const std::string& relationship) const {
std::lock_guard<std::mutex> lock(transferred_relationships_mutex_);
const auto relationship_it = transferred_relationships_.find(relationship);
if (relationship_it != std::end(transferred_relationships_)) {