You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by fg...@apache.org on 2023/05/03 13:34:36 UTC
[nifi-minifi-cpp] branch main updated (72e1c3516 -> 6e0878bb4)
This is an automated email from the ASF dual-hosted git repository.
fgerlits pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
from 72e1c3516 MINIFICPP-2103 JNI extension compile fix (libc++)
new 93ef73a00 MINIFICPP-2022 Add valid repository size metrics for all repositories
new 5be21834c MINIFICPP-2099 Only run tests requiring test processors if they are available
new ba60437db MINIFICPP-2100 Add RockyLinux Dockerfile
new 0488c5498 MINIFICPP-2108 Update version number to 0.15.0
new 6e0878bb4 MINIFICPP-2110 Fix default CA path for S3 on CentOS
The 5 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:
CMakeLists.txt | 2 +-
METRICS.md | 40 ++--
README.md | 4 +-
cmake/DockerConfig.cmake | 27 ++-
controller/tests/ControllerTests.cpp | 4 +-
docker/DockerVerify.sh | 16 +-
docker/{centos => rockylinux}/Dockerfile | 27 ++-
.../cluster/checkers/PrometheusChecker.py | 19 +-
docker/test/integration/environment.py | 6 +-
.../features/core_functionality.feature | 3 +-
extensions/aws/processors/S3Processor.cpp | 6 +
extensions/coap/tests/CoapC2VerifyHeartbeat.cpp | 2 +-
extensions/coap/tests/CoapIntegrationBase.h | 6 +-
extensions/http-curl/client/HTTPClient.cpp | 22 +-
extensions/http-curl/client/HTTPClient.h | 1 -
extensions/http-curl/tests/C2NullConfiguration.cpp | 2 +-
extensions/http-curl/tests/C2PauseResumeTest.cpp | 5 +-
.../http-curl/tests/C2VerifyServeResults.cpp | 2 +-
extensions/http-curl/tests/HTTPHandlers.h | 2 +-
extensions/http-curl/tests/HTTPIntegrationBase.h | 2 +-
extensions/http-curl/tests/VerifyInvokeHTTP.h | 2 +-
.../tests/PrometheusMetricsPublisherTest.cpp | 9 +-
.../rocksdb-repos/DatabaseContentRepository.cpp | 19 ++
.../rocksdb-repos/DatabaseContentRepository.h | 3 +
extensions/rocksdb-repos/FlowFileRepository.cpp | 73 +------
extensions/rocksdb-repos/FlowFileRepository.h | 33 +--
extensions/rocksdb-repos/ProvenanceRepository.cpp | 94 +++------
extensions/rocksdb-repos/ProvenanceRepository.h | 25 +--
extensions/rocksdb-repos/RocksDbRepository.cpp | 112 +++++++++++
extensions/rocksdb-repos/RocksDbRepository.h | 63 ++++++
extensions/rocksdb-repos/database/OpenRocksDb.cpp | 13 ++
extensions/rocksdb-repos/database/OpenRocksDb.h | 3 +
libminifi/CMakeLists.txt | 2 +-
libminifi/include/core/ContentRepository.h | 12 +-
libminifi/include/core/Repository.h | 30 +--
libminifi/include/core/RepositoryFactory.h | 5 +-
...izableComponent.h => RepositoryMetricsSource.h} | 27 +--
.../include/core/repository/FileSystemRepository.h | 14 ++
.../core/repository/VolatileContentRepository.h | 18 +-
.../include/core/repository/VolatileRepository.h | 16 +-
.../core/repository/VolatileRepositoryData.h | 18 ++
.../include/core/state/MetricsPublisherStore.h | 5 +-
.../include/core/state/nodes/AgentInformation.h | 73 ++++---
.../include/core/state/nodes/RepositoryMetrics.h | 56 ++++--
.../include/core/state/nodes/ResponseNodeLoader.h | 8 +-
libminifi/include/utils/HTTPUtils.h | 11 +-
libminifi/include/utils/file/FileUtils.h | 14 ++
libminifi/src/core/RepositoryFactory.cpp | 9 +
.../src/core/repository/VolatileRepository.cpp | 6 +-
.../src/core/repository/VolatileRepositoryData.cpp | 1 +
libminifi/src/core/state/MetricsPublisherStore.cpp | 6 +-
.../src/core/state/nodes/ResponseNodeLoader.cpp | 26 +--
.../src/utils/{TestUtils.cpp => HTTPUtils.cpp} | 27 ++-
libminifi/test/flow-tests/TestControllerWithFlow.h | 4 +-
libminifi/test/integration/IntegrationBase.h | 7 +-
.../rocksdb-tests/DBProvenanceRepositoryTests.cpp | 2 +-
libminifi/test/rocksdb-tests/RepoTests.cpp | 221 +++++++++++++++++++++
.../unit/ControllerSocketMetricsPublisherTest.cpp | 2 +-
libminifi/test/unit/FileUtilsTests.cpp | 39 ++++
libminifi/test/unit/MetricsTests.cpp | 140 ++++++-------
libminifi/test/unit/ProvenanceTestHelper.h | 51 ++++-
libminifi/test/unit/ResponseNodeLoaderTests.cpp | 2 +-
minifi_main/MiNiFiMain.cpp | 3 +-
63 files changed, 1006 insertions(+), 496 deletions(-)
copy docker/{centos => rockylinux}/Dockerfile (75%)
create mode 100644 extensions/rocksdb-repos/RocksDbRepository.cpp
create mode 100644 extensions/rocksdb-repos/RocksDbRepository.h
copy libminifi/include/core/{SerializableComponent.h => RepositoryMetricsSource.h} (67%)
copy libminifi/src/utils/{TestUtils.cpp => HTTPUtils.cpp} (61%)
[nifi-minifi-cpp] 01/05: MINIFICPP-2022 Add valid repository size metrics for all repositories
Posted by fg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fgerlits pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 93ef73a00f8775f0bdcd63554ab10fdabb49652f
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Mon Jan 2 15:08:34 2023 +0100
MINIFICPP-2022 Add valid repository size metrics for all repositories
Signed-off-by: Ferenc Gerlits <fg...@gmail.com>
This closes #1490
---
METRICS.md | 40 ++--
controller/tests/ControllerTests.cpp | 4 +-
.../cluster/checkers/PrometheusChecker.py | 19 +-
extensions/coap/tests/CoapIntegrationBase.h | 4 +-
extensions/http-curl/tests/C2PauseResumeTest.cpp | 3 +-
.../tests/PrometheusMetricsPublisherTest.cpp | 9 +-
.../rocksdb-repos/DatabaseContentRepository.cpp | 19 ++
.../rocksdb-repos/DatabaseContentRepository.h | 3 +
extensions/rocksdb-repos/FlowFileRepository.cpp | 73 +------
extensions/rocksdb-repos/FlowFileRepository.h | 33 +--
extensions/rocksdb-repos/ProvenanceRepository.cpp | 94 +++------
extensions/rocksdb-repos/ProvenanceRepository.h | 25 +--
extensions/rocksdb-repos/RocksDbRepository.cpp | 112 +++++++++++
extensions/rocksdb-repos/RocksDbRepository.h | 63 ++++++
extensions/rocksdb-repos/database/OpenRocksDb.cpp | 13 ++
extensions/rocksdb-repos/database/OpenRocksDb.h | 3 +
libminifi/include/core/ContentRepository.h | 12 +-
libminifi/include/core/Repository.h | 30 +--
libminifi/include/core/RepositoryFactory.h | 5 +-
libminifi/include/core/RepositoryMetricsSource.h | 45 +++++
.../include/core/repository/FileSystemRepository.h | 14 ++
.../core/repository/VolatileContentRepository.h | 18 +-
.../include/core/repository/VolatileRepository.h | 16 +-
.../core/repository/VolatileRepositoryData.h | 18 ++
.../include/core/state/MetricsPublisherStore.h | 5 +-
.../include/core/state/nodes/AgentInformation.h | 73 ++++---
.../include/core/state/nodes/RepositoryMetrics.h | 56 ++++--
.../include/core/state/nodes/ResponseNodeLoader.h | 8 +-
libminifi/include/utils/file/FileUtils.h | 14 ++
libminifi/src/core/RepositoryFactory.cpp | 9 +
.../src/core/repository/VolatileRepository.cpp | 6 +-
.../src/core/repository/VolatileRepositoryData.cpp | 1 +
libminifi/src/core/state/MetricsPublisherStore.cpp | 6 +-
.../src/core/state/nodes/ResponseNodeLoader.cpp | 26 +--
libminifi/test/flow-tests/TestControllerWithFlow.h | 4 +-
libminifi/test/integration/IntegrationBase.h | 5 +-
.../rocksdb-tests/DBProvenanceRepositoryTests.cpp | 2 +-
libminifi/test/rocksdb-tests/RepoTests.cpp | 221 +++++++++++++++++++++
.../unit/ControllerSocketMetricsPublisherTest.cpp | 2 +-
libminifi/test/unit/FileUtilsTests.cpp | 39 ++++
libminifi/test/unit/MetricsTests.cpp | 140 ++++++-------
libminifi/test/unit/ProvenanceTestHelper.h | 51 ++++-
libminifi/test/unit/ResponseNodeLoaderTests.cpp | 2 +-
minifi_main/MiNiFiMain.cpp | 3 +-
44 files changed, 931 insertions(+), 417 deletions(-)
diff --git a/METRICS.md b/METRICS.md
index f6c87dbaf..f0cc52d54 100644
--- a/METRICS.md
+++ b/METRICS.md
@@ -100,17 +100,19 @@ QueueMetrics is a system level metric that reports queue metrics for every conne
### RepositoryMetrics
-RepositoryMetrics is a system level metric that reports metrics for the registered repositories (by default flowfile and provenance repository)
+RepositoryMetrics is a system level metric that reports metrics for the registered repositories (by default flowfile, content, and provenance repositories)
-| Metric name | Labels | Description |
-|----------------------|-----------------|---------------------------------------|
-| is_running | repository_name | Is the repository running (1 or 0) |
-| is_full | repository_name | Is the repository full (1 or 0) |
-| repository_size | repository_name | Current size of the repository |
+| Metric name | Labels | Description |
+|---------------------------|-----------------|-------------------------------------------------|
+| is_running | repository_name | Is the repository running (1 or 0) |
+| is_full | repository_name | Is the repository full (1 or 0) |
+| repository_size_bytes | repository_name | Current size of the repository |
+| max_repository_size_bytes | repository_name | Maximum size of the repository (0 if unlimited) |
+| repository_entry_count | repository_name | Current number of entries in the repository |
-| Label | Description |
-|--------------------------|-----------------------------------------------------------------|
-| repository_name | Name of the reported repository |
+| Label | Description |
+|--------------------------|---------------------------------------------------------------------------------------------------------------------------------------|
+| repository_name | Name of the reported repository. There are three repositories present with the following names: `flowfile`, `content` and `provenance` |
### DeviceInfoNode
@@ -145,15 +147,17 @@ FlowInformation is a system level metric that reports component and queue relate
AgentStatus is a system level metric that defines current agent status including repository, component and resource usage information.
-| Metric name | Labels | Description |
-|--------------------------|--------------------------------|------------------------------------------------------------------------------------------------------------|
-| is_running | repository_name | Is the repository running (1 or 0) |
-| is_full | repository_name | Is the repository full (1 or 0) |
-| repository_size | repository_name | Current size of the repository |
-| uptime_milliseconds | - | Agent uptime in milliseconds |
-| is_running | component_uuid, component_name | Check if the component is running (1 or 0) |
-| agent_memory_usage_bytes | - | Memory used by the agent process in bytes |
-| agent_cpu_utilization | - | CPU utilization of the agent process (between 0 and 1). In case of a query error the returned value is -1. |
+| Metric name | Labels | Description |
+|---------------------------|--------------------------------|------------------------------------------------------------------------------------------------------------|
+| is_running | repository_name | Is the repository running (1 or 0) |
+| is_full | repository_name | Is the repository full (1 or 0) |
+| repository_size_bytes | repository_name | Current size of the repository |
+| max_repository_size_bytes | repository_name | Maximum size of the repository (0 if unlimited) |
+| repository_entry_count | repository_name | Current number of entries in the repository |
+| uptime_milliseconds | - | Agent uptime in milliseconds |
+| is_running | component_uuid, component_name | Check if the component is running (1 or 0) |
+| agent_memory_usage_bytes | - | Memory used by the agent process in bytes |
+| agent_cpu_utilization | - | CPU utilization of the agent process (between 0 and 1). In case of a query error the returned value is -1. |
| Label | Description |
|-----------------|----------------------------------------------------------|
diff --git a/controller/tests/ControllerTests.cpp b/controller/tests/ControllerTests.cpp
index 49ad26f79..c88af54df 100644
--- a/controller/tests/ControllerTests.cpp
+++ b/controller/tests/ControllerTests.cpp
@@ -514,7 +514,7 @@ TEST_CASE_METHOD(ControllerTestFixture, "Test manifest getter", "[controllerTest
}
auto reporter = std::make_shared<minifi::c2::ControllerSocketMetricsPublisher>("ControllerSocketMetricsPublisher");
- auto response_node_loader = std::make_shared<minifi::state::response::ResponseNodeLoader>(configuration_, nullptr, nullptr, nullptr);
+ auto response_node_loader = std::make_shared<minifi::state::response::ResponseNodeLoader>(configuration_, std::vector<std::shared_ptr<core::RepositoryMetricsSource>>{}, nullptr);
reporter->initialize(configuration_, response_node_loader);
initalizeControllerSocket(reporter);
@@ -538,7 +538,7 @@ TEST_CASE_METHOD(ControllerTestFixture, "Test jstack getter", "[controllerTests]
}
auto reporter = std::make_shared<minifi::c2::ControllerSocketMetricsPublisher>("ControllerSocketMetricsPublisher");
- auto response_node_loader = std::make_shared<minifi::state::response::ResponseNodeLoader>(configuration_, nullptr, nullptr, nullptr);
+ auto response_node_loader = std::make_shared<minifi::state::response::ResponseNodeLoader>(configuration_, std::vector<std::shared_ptr<core::RepositoryMetricsSource>>{}, nullptr);
reporter->initialize(configuration_, response_node_loader);
initalizeControllerSocket(reporter);
diff --git a/docker/test/integration/cluster/checkers/PrometheusChecker.py b/docker/test/integration/cluster/checkers/PrometheusChecker.py
index 461e9a5e8..bd7daff8b 100644
--- a/docker/test/integration/cluster/checkers/PrometheusChecker.py
+++ b/docker/test/integration/cluster/checkers/PrometheusChecker.py
@@ -47,8 +47,9 @@ class PrometheusChecker:
raise Exception("Metric class '%s' verification is not implemented" % metric_class)
def verify_repository_metrics(self):
- label_list = [{'repository_name': 'provenance'}, {'repository_name': 'flowfile'}]
- return all((self.verify_metrics_exist(['minifi_is_running', 'minifi_is_full', 'minifi_repository_size'], 'RepositoryMetrics', labels) for labels in label_list))
+ label_list = [{'repository_name': 'provenance'}, {'repository_name': 'flowfile'}, {'repository_name': 'content'}]
+ return all((self.verify_metrics_exist(['minifi_is_running', 'minifi_is_full', 'minifi_repository_size_bytes', 'minifi_max_repository_size_bytes', 'minifi_repository_entry_count'], 'RepositoryMetrics', labels) for labels in label_list)) and \
+ all((self.verify_metric_larger_than_zero('minifi_repository_size_bytes', 'RepositoryMetrics', labels) for labels in label_list[1:3]))
def verify_queue_metrics(self):
return self.verify_metrics_exist(['minifi_queue_data_size', 'minifi_queue_data_size_max', 'minifi_queue_size', 'minifi_queue_size_max'], 'QueueMetrics')
@@ -72,12 +73,22 @@ class PrometheusChecker:
return self.verify_metrics_exist(['minifi_physical_mem', 'minifi_memory_usage', 'minifi_cpu_utilization'], 'DeviceInfoNode')
def verify_agent_status_metrics(self):
- label_list = [{'repository_name': 'provenance'}, {'repository_name': 'flowfile'}]
+ label_list = [{'repository_name': 'flowfile'}, {'repository_name': 'content'}]
for labels in label_list:
if not (self.verify_metric_exists('minifi_is_running', 'AgentStatus', labels)
and self.verify_metric_exists('minifi_is_full', 'AgentStatus', labels)
- and self.verify_metric_exists('minifi_repository_size', 'AgentStatus', labels)):
+ and self.verify_metric_exists('minifi_max_repository_size_bytes', 'AgentStatus', labels)
+ and self.verify_metric_larger_than_zero('minifi_repository_size_bytes', 'AgentStatus', labels)
+ and self.verify_metric_exists('minifi_repository_entry_count', 'AgentStatus', labels)):
return False
+
+ # provenance repository is NoOpRepository by default which has zero size
+ if not (self.verify_metric_exists('minifi_is_running', 'AgentStatus', {'repository_name': 'provenance'})
+ and self.verify_metric_exists('minifi_is_full', 'AgentStatus', {'repository_name': 'provenance'})
+ and self.verify_metric_exists('minifi_max_repository_size_bytes', 'AgentStatus', {'repository_name': 'provenance'})
+ and self.verify_metric_exists('minifi_repository_size_bytes', 'AgentStatus', {'repository_name': 'provenance'})
+ and self.verify_metric_exists('minifi_repository_entry_count', 'AgentStatus', {'repository_name': 'provenance'})):
+ return False
return self.verify_metric_exists('minifi_uptime_milliseconds', 'AgentStatus') and \
self.verify_metric_exists('minifi_agent_memory_usage_bytes', 'AgentStatus') and \
self.verify_metric_exists('minifi_agent_cpu_utilization', 'AgentStatus')
diff --git a/extensions/coap/tests/CoapIntegrationBase.h b/extensions/coap/tests/CoapIntegrationBase.h
index b498e1240..e7fbc1e65 100644
--- a/extensions/coap/tests/CoapIntegrationBase.h
+++ b/extensions/coap/tests/CoapIntegrationBase.h
@@ -21,6 +21,7 @@
#include <optional>
#include <string>
#include <utility>
+#include <vector>
#include "../tests/TestServer.h"
#include "CivetServer.h"
@@ -72,7 +73,8 @@ class CoapIntegrationBase : public IntegrationBase {
queryRootProcessGroup(pg);
- auto metrics_publisher_store = std::make_unique<minifi::state::MetricsPublisherStore>(configuration, test_repo, test_flow_repo, yaml_ptr);
+ std::vector<std::shared_ptr<core::RepositoryMetricsSource>> repo_metric_sources{test_repo, test_flow_repo, content_repo};
+ auto metrics_publisher_store = std::make_unique<minifi::state::MetricsPublisherStore>(configuration, repo_metric_sources, yaml_ptr);
auto controller = std::make_unique<minifi::FlowController>(test_repo, test_flow_repo, configuration, std::move(yaml_ptr), content_repo, std::move(metrics_publisher_store));
controller->load();
diff --git a/extensions/http-curl/tests/C2PauseResumeTest.cpp b/extensions/http-curl/tests/C2PauseResumeTest.cpp
index 78bd1c54f..cf8e137a9 100644
--- a/extensions/http-curl/tests/C2PauseResumeTest.cpp
+++ b/extensions/http-curl/tests/C2PauseResumeTest.cpp
@@ -135,7 +135,8 @@ int main(int argc, char **argv) {
auto yaml_ptr = std::make_shared<core::YamlConfiguration>(core::ConfigurationContext{test_repo, content_repo, stream_factory, configuration, args.test_file});
- auto metrics_publisher_store = std::make_unique<minifi::state::MetricsPublisherStore>(configuration, test_repo, test_flow_repo, yaml_ptr);
+ std::vector<std::shared_ptr<core::RepositoryMetricsSource>> repo_metric_sources{test_repo, test_flow_repo, content_repo};
+ auto metrics_publisher_store = std::make_unique<minifi::state::MetricsPublisherStore>(configuration, repo_metric_sources, yaml_ptr);
std::shared_ptr<minifi::FlowController> controller = std::make_shared<minifi::FlowController>(test_repo, test_flow_repo, configuration,
std::move(yaml_ptr), content_repo, std::move(metrics_publisher_store));
diff --git a/extensions/prometheus/tests/PrometheusMetricsPublisherTest.cpp b/extensions/prometheus/tests/PrometheusMetricsPublisherTest.cpp
index 6b39468f5..dcb11082f 100644
--- a/extensions/prometheus/tests/PrometheusMetricsPublisherTest.cpp
+++ b/extensions/prometheus/tests/PrometheusMetricsPublisherTest.cpp
@@ -52,7 +52,9 @@ class PrometheusPublisherTestFixture {
: configuration_(std::make_shared<Configure>()),
provenance_repo_(core::createRepository("provenancerepository")),
flow_file_repo_(core::createRepository("flowfilerepository")),
- response_node_loader_(std::make_shared<state::response::ResponseNodeLoader>(configuration_, provenance_repo_, flow_file_repo_, nullptr)) {
+ content_repo_(core::createContentRepository("volatilecontentrepository")),
+ response_node_loader_(std::make_shared<state::response::ResponseNodeLoader>(configuration_,
+ std::vector<std::shared_ptr<core::RepositoryMetricsSource>>{provenance_repo_, flow_file_repo_, content_repo_}, nullptr)) {
std::unique_ptr<DummyMetricsExposer> dummy_exposer;
if (user_dummy_exposer) {
dummy_exposer = std::make_unique<DummyMetricsExposer>();
@@ -63,8 +65,9 @@ class PrometheusPublisherTestFixture {
protected:
std::shared_ptr<Configure> configuration_;
- std::shared_ptr<core::Repository> provenance_repo_;
- std::shared_ptr<core::Repository> flow_file_repo_;
+ std::shared_ptr<core::RepositoryMetricsSource> provenance_repo_;
+ std::shared_ptr<core::RepositoryMetricsSource> flow_file_repo_;
+ std::shared_ptr<core::RepositoryMetricsSource> content_repo_;
std::shared_ptr<state::response::ResponseNodeLoader> response_node_loader_;
std::unique_ptr<PrometheusMetricsPublisher> publisher_;
DummyMetricsExposer* exposer_ = nullptr;
diff --git a/extensions/rocksdb-repos/DatabaseContentRepository.cpp b/extensions/rocksdb-repos/DatabaseContentRepository.cpp
index 187776778..56c0d2c83 100644
--- a/extensions/rocksdb-repos/DatabaseContentRepository.cpp
+++ b/extensions/rocksdb-repos/DatabaseContentRepository.cpp
@@ -267,6 +267,25 @@ void DatabaseContentRepository::clearOrphans() {
}
}
+uint64_t DatabaseContentRepository::getRepositorySize() const {
+ return (utils::optional_from_ptr(db_.get()) |
+ utils::flatMap([](const auto& db) { return db->open(); }) |
+ utils::flatMap([](const auto& opendb) { return opendb.getApproximateSizes(); })).value_or(0);
+}
+
+uint64_t DatabaseContentRepository::getRepositoryEntryCount() const {
+ return (utils::optional_from_ptr(db_.get()) |
+ utils::flatMap([](const auto& db) { return db->open(); }) |
+ utils::flatMap([](auto&& opendb) -> std::optional<uint64_t> {
+ std::string key_count;
+ opendb.GetProperty("rocksdb.estimate-num-keys", &key_count);
+ if (!key_count.empty()) {
+ return std::stoull(key_count);
+ }
+ return std::nullopt;
+ })).value_or(0);
+}
+
REGISTER_RESOURCE_AS(DatabaseContentRepository, InternalResource, ("DatabaseContentRepository", "databasecontentrepository"));
} // namespace org::apache::nifi::minifi::core::repository
diff --git a/extensions/rocksdb-repos/DatabaseContentRepository.h b/extensions/rocksdb-repos/DatabaseContentRepository.h
index 1305a7c2d..e084a6515 100644
--- a/extensions/rocksdb-repos/DatabaseContentRepository.h
+++ b/extensions/rocksdb-repos/DatabaseContentRepository.h
@@ -74,6 +74,9 @@ class DatabaseContentRepository : public core::ContentRepository {
void start() override;
void stop() override;
+ uint64_t getRepositorySize() const override;
+ uint64_t getRepositoryEntryCount() const override;
+
protected:
bool removeKey(const std::string& content_path) override;
diff --git a/extensions/rocksdb-repos/FlowFileRepository.cpp b/extensions/rocksdb-repos/FlowFileRepository.cpp
index 36999c4e0..b38a7fdce 100644
--- a/extensions/rocksdb-repos/FlowFileRepository.cpp
+++ b/extensions/rocksdb-repos/FlowFileRepository.cpp
@@ -30,6 +30,7 @@
#include "FlowFileRecord.h"
#include "utils/gsl.h"
#include "core/Resource.h"
+#include "utils/OptionalUtils.h"
using namespace std::literals::chrono_literals;
@@ -110,24 +111,6 @@ void FlowFileRepository::deserializeFlowFilesWithNoContentClaim(minifi::internal
}
}
-void FlowFileRepository::printStats() {
- auto opendb = db_->open();
- if (!opendb) {
- return;
- }
- std::string key_count;
- opendb->GetProperty("rocksdb.estimate-num-keys", &key_count);
-
- std::string table_readers;
- opendb->GetProperty("rocksdb.estimate-table-readers-mem", &table_readers);
-
- std::string all_memtables;
- opendb->GetProperty("rocksdb.cur-size-all-mem-tables", &all_memtables);
-
- logger_->log_info("Repository stats: key count: %s, table readers size: %s, all memory tables size: %s",
- key_count, table_readers, all_memtables);
-}
-
void FlowFileRepository::run() {
auto last = std::chrono::steady_clock::now();
while (isRunning()) {
@@ -142,22 +125,6 @@ void FlowFileRepository::run() {
flush();
}
-bool FlowFileRepository::ExecuteWithRetry(const std::function<rocksdb::Status()>& operation) {
- constexpr int RETRY_COUNT = 3;
- std::chrono::milliseconds wait_time = 0ms;
- for (int i=0; i < RETRY_COUNT; ++i) {
- auto status = operation();
- if (status.ok()) {
- logger_->log_trace("Rocksdb operation executed successfully");
- return true;
- }
- logger_->log_error("Rocksdb operation failed: %s", status.ToString());
- wait_time += FLOWFILE_REPOSITORY_RETRY_INTERVAL_INCREMENTS;
- std::this_thread::sleep_for(wait_time);
- }
- return false;
-}
-
void FlowFileRepository::initialize_repository() {
auto opendb = db_->open();
if (!opendb) {
@@ -204,7 +171,6 @@ void FlowFileRepository::initialize_repository() {
void FlowFileRepository::loadComponent(const std::shared_ptr<core::ContentRepository> &content_repo) {
content_repo_ = content_repo;
- repo_size_ = 0;
swap_loader_ = std::make_unique<FlowFileLoader>(gsl::make_not_null(db_.get()), content_repo_);
initialize_repository();
@@ -276,48 +242,11 @@ void FlowFileRepository::setCompactionPeriod(const std::shared_ptr<Configure> &c
}
}
-bool FlowFileRepository::Put(const std::string& key, const uint8_t *buf, size_t bufLen) {
- // persistent to the DB
- auto opendb = db_->open();
- if (!opendb) {
- return false;
- }
- rocksdb::Slice value((const char *) buf, bufLen);
- auto operation = [&key, &value, &opendb]() { return opendb->Put(rocksdb::WriteOptions(), key, value); };
- return ExecuteWithRetry(operation);
-}
-
-bool FlowFileRepository::MultiPut(const std::vector<std::pair<std::string, std::unique_ptr<minifi::io::BufferStream>>>& data) {
- auto opendb = db_->open();
- if (!opendb) {
- return false;
- }
- auto batch = opendb->createWriteBatch();
- for (const auto &item : data) {
- const auto buf = item.second->getBuffer().as_span<const char>();
- rocksdb::Slice value(buf.data(), buf.size());
- if (!batch.Put(item.first, value).ok()) {
- logger_->log_error("Failed to add item to batch operation");
- return false;
- }
- }
- auto operation = [&batch, &opendb]() { return opendb->Write(rocksdb::WriteOptions(), &batch); };
- return ExecuteWithRetry(operation);
-}
-
bool FlowFileRepository::Delete(const std::string& key) {
keys_to_delete_.enqueue({.key = key});
return true;
}
-bool FlowFileRepository::Get(const std::string &key, std::string &value) {
- auto opendb = db_->open();
- if (!opendb) {
- return false;
- }
- return opendb->Get(rocksdb::ReadOptions(), key, &value).ok();
-}
-
void FlowFileRepository::runCompaction() {
do {
if (auto opendb = db_->open()) {
diff --git a/extensions/rocksdb-repos/FlowFileRepository.h b/extensions/rocksdb-repos/FlowFileRepository.h
index 68959ce88..dd1412ada 100644
--- a/extensions/rocksdb-repos/FlowFileRepository.h
+++ b/extensions/rocksdb-repos/FlowFileRepository.h
@@ -23,13 +23,11 @@
#include <list>
#include "utils/file/FileUtils.h"
-#include "rocksdb/db.h"
#include "rocksdb/options.h"
#include "rocksdb/slice.h"
#include "rocksdb/utilities/checkpoint.h"
#include "core/Core.h"
#include "core/logging/LoggerConfiguration.h"
-#include "core/ThreadedRepository.h"
#include "Connection.h"
#include "concurrentqueue.h"
#include "database/RocksDatabase.h"
@@ -40,6 +38,7 @@
#include "range/v3/algorithm/all_of.hpp"
#include "utils/Literals.h"
#include "utils/StoppableThread.h"
+#include "RocksDbRepository.h"
namespace org::apache::nifi::minifi::core::repository {
@@ -53,13 +52,12 @@ constexpr auto FLOWFILE_CHECKPOINT_DIRECTORY = "./flowfile_checkpoint";
constexpr auto MAX_FLOWFILE_REPOSITORY_STORAGE_SIZE = 10_MiB;
constexpr auto MAX_FLOWFILE_REPOSITORY_ENTRY_LIFE_TIME = std::chrono::minutes(10);
constexpr auto FLOWFILE_REPOSITORY_PURGE_PERIOD = std::chrono::seconds(2);
-constexpr auto FLOWFILE_REPOSITORY_RETRY_INTERVAL_INCREMENTS = std::chrono::milliseconds(500);
/**
* Flow File repository
* Design: Extends Repository and implements the run function, using rocksdb as the primary substrate.
*/
-class FlowFileRepository : public ThreadedRepository, public SwapManager {
+class FlowFileRepository : public RocksDbRepository, public SwapManager {
static constexpr std::chrono::milliseconds DEFAULT_COMPACTION_PERIOD = std::chrono::minutes{2};
struct ExpiredFlowFileInfo {
@@ -79,8 +77,8 @@ class FlowFileRepository : public ThreadedRepository, public SwapManager {
std::chrono::milliseconds maxPartitionMillis = MAX_FLOWFILE_REPOSITORY_ENTRY_LIFE_TIME,
int64_t maxPartitionBytes = MAX_FLOWFILE_REPOSITORY_STORAGE_SIZE,
std::chrono::milliseconds purgePeriod = FLOWFILE_REPOSITORY_PURGE_PERIOD)
- : ThreadedRepository(repo_name.length() > 0 ? std::move(repo_name) : core::getClassName<FlowFileRepository>(), std::move(directory), maxPartitionMillis, maxPartitionBytes, purgePeriod),
- logger_(logging::LoggerFactory<FlowFileRepository>::getLogger()) {
+ : RocksDbRepository(repo_name.length() > 0 ? std::move(repo_name) : core::getClassName<FlowFileRepository>(),
+ std::move(directory), maxPartitionMillis, maxPartitionBytes, purgePeriod, logging::LoggerFactory<FlowFileRepository>::getLogger()) {
}
~FlowFileRepository() override {
@@ -95,18 +93,11 @@ class FlowFileRepository : public ThreadedRepository, public SwapManager {
return false;
}
- void flush() override;
-
- virtual void printStats();
-
- bool initialize(const std::shared_ptr<Configure> &configure) override;
-
- bool Put(const std::string& key, const uint8_t *buf, size_t bufLen) override;
- bool MultiPut(const std::vector<std::pair<std::string, std::unique_ptr<minifi::io::BufferStream>>>& data) override;
bool Delete(const std::string& key) override;
bool Delete(const std::shared_ptr<core::CoreComponent>& item) override;
- bool Get(const std::string &key, std::string &value) override;
+ void flush() override;
+ bool initialize(const std::shared_ptr<Configure> &configure) override;
void loadComponent(const std::shared_ptr<core::ContentRepository> &content_repo) override;
bool start() override;
bool stop() override;
@@ -115,27 +106,17 @@ class FlowFileRepository : public ThreadedRepository, public SwapManager {
private:
void run() override;
+ void initialize_repository();
void runCompaction();
void setCompactionPeriod(const std::shared_ptr<Configure> &configure);
- bool ExecuteWithRetry(const std::function<rocksdb::Status()>& operation);
-
- void initialize_repository();
-
void deserializeFlowFilesWithNoContentClaim(minifi::internal::OpenRocksDb& opendb, std::list<ExpiredFlowFileInfo>& flow_files);
- std::thread& getThread() override {
- return thread_;
- }
-
moodycamel::ConcurrentQueue<ExpiredFlowFileInfo> keys_to_delete_;
std::shared_ptr<core::ContentRepository> content_repo_;
- std::unique_ptr<minifi::internal::RocksDatabase> db_;
std::unique_ptr<FlowFileLoader> swap_loader_;
- std::shared_ptr<logging::Logger> logger_;
std::shared_ptr<minifi::Configure> config_;
- std::thread thread_;
std::chrono::milliseconds compaction_period_;
std::unique_ptr<utils::StoppableThread> compaction_thread_;
diff --git a/extensions/rocksdb-repos/ProvenanceRepository.cpp b/extensions/rocksdb-repos/ProvenanceRepository.cpp
index 2680de9d5..15073835e 100644
--- a/extensions/rocksdb-repos/ProvenanceRepository.cpp
+++ b/extensions/rocksdb-repos/ProvenanceRepository.cpp
@@ -23,20 +23,6 @@
namespace org::apache::nifi::minifi::provenance {
-void ProvenanceRepository::printStats() {
- std::string key_count;
- db_->GetProperty("rocksdb.estimate-num-keys", &key_count);
-
- std::string table_readers;
- db_->GetProperty("rocksdb.estimate-table-readers-mem", &table_readers);
-
- std::string all_memtables;
- db_->GetProperty("rocksdb.cur-size-all-mem-tables", &all_memtables);
-
- logger_->log_info("Repository stats: key count: %s, table readers size: %s, all memory tables size: %s",
- key_count, table_readers, all_memtables);
-}
-
void ProvenanceRepository::run() {
size_t count = 0;
while (isRunning()) {
@@ -66,64 +52,45 @@ bool ProvenanceRepository::initialize(const std::shared_ptr<org::apache::nifi::m
}
logger_->log_debug("MiNiFi Provenance Max Storage Time: [%" PRId64 "] ms",
int64_t{max_partition_millis_.count()});
- rocksdb::Options options;
- options.create_if_missing = true;
- options.use_direct_io_for_flush_and_compaction = true;
- options.use_direct_reads = true;
- // Rocksdb write buffers act as a log of database operation: grow till reaching the limit, serialized after
- // This shouldn't go above 16MB and the configured total size of the db should cap it as well
- int64_t max_buffer_size = 16 << 20;
- options.write_buffer_size = gsl::narrow<size_t>(std::min(max_buffer_size, max_partition_bytes_));
- options.max_write_buffer_number = 4;
- options.min_write_buffer_number_to_merge = 1;
- options.compaction_style = rocksdb::CompactionStyle::kCompactionStyleFIFO;
- options.compaction_options_fifo = rocksdb::CompactionOptionsFIFO(max_partition_bytes_, false);
- if (max_partition_millis_ > std::chrono::milliseconds(0)) {
- options.ttl = std::chrono::duration_cast<std::chrono::seconds>(max_partition_millis_).count();
- }
+ auto db_options = [] (minifi::internal::Writable<rocksdb::DBOptions>& db_opts) {
+ db_opts.set(&rocksdb::DBOptions::create_if_missing, true);
+ db_opts.set(&rocksdb::DBOptions::use_direct_io_for_flush_and_compaction, true);
+ db_opts.set(&rocksdb::DBOptions::use_direct_reads, true);
+ };
- logger_->log_info("Write buffer: %llu", options.write_buffer_size);
- logger_->log_info("Max partition bytes: %llu", max_partition_bytes_);
- logger_->log_info("Ttl: %llu", options.ttl);
+ // Rocksdb write buffers act as a log of database operation: grow till reaching the limit, serialized after
+ // This shouldn't go above 16MB and the configured total size of the db should cap it as well
+ auto cf_options = [this] (rocksdb::ColumnFamilyOptions& cf_opts) {
+ int64_t max_buffer_size = 16 << 20;
+ cf_opts.write_buffer_size = gsl::narrow<size_t>(std::min(max_buffer_size, max_partition_bytes_));
+ cf_opts.max_write_buffer_number = 4;
+ cf_opts.min_write_buffer_number_to_merge = 1;
+
+ cf_opts.compaction_style = rocksdb::CompactionStyle::kCompactionStyleFIFO;
+ cf_opts.compaction_options_fifo = rocksdb::CompactionOptionsFIFO(max_partition_bytes_, false);
+ if (max_partition_millis_ > std::chrono::milliseconds(0)) {
+ cf_opts.ttl = std::chrono::duration_cast<std::chrono::seconds>(max_partition_millis_).count();
+ }
+ };
- rocksdb::DB* db;
- rocksdb::Status status = rocksdb::DB::Open(options, directory_, &db);
- if (status.ok()) {
+ db_ = minifi::internal::RocksDatabase::create(db_options, cf_options, directory_);
+ if (db_->open()) {
logger_->log_debug("MiNiFi Provenance Repository database open %s success", directory_);
- db_.reset(db);
} else {
- logger_->log_error("MiNiFi Provenance Repository database open %s failed: %s", directory_, status.ToString());
+ logger_->log_error("MiNiFi Provenance Repository database open %s failed", directory_);
return false;
}
return true;
}
-bool ProvenanceRepository::Put(const std::string& key, const uint8_t *buf, size_t bufLen) {
- // persist to the DB
- rocksdb::Slice value((const char *) buf, bufLen);
- return db_->Put(rocksdb::WriteOptions(), key, value).ok();
-}
-
-bool ProvenanceRepository::MultiPut(const std::vector<std::pair<std::string, std::unique_ptr<minifi::io::BufferStream>>>& data) {
- rocksdb::WriteBatch batch;
- for (const auto &item : data) {
- const auto buf = item.second->getBuffer().as_span<const char>();
- rocksdb::Slice value(buf.data(), buf.size());
- if (!batch.Put(item.first, value).ok()) {
- return false;
- }
- }
- return db_->Write(rocksdb::WriteOptions(), &batch).ok();
-}
-
-bool ProvenanceRepository::Get(const std::string &key, std::string &value) {
- return db_->Get(rocksdb::ReadOptions(), key, &value).ok();
-}
-
bool ProvenanceRepository::getElements(std::vector<std::shared_ptr<core::SerializableComponent>> &records, size_t &max_size) {
- std::unique_ptr<rocksdb::Iterator> it(db_->NewIterator(rocksdb::ReadOptions()));
+ auto opendb = db_->open();
+ if (!opendb) {
+ return false;
+ }
+ std::unique_ptr<rocksdb::Iterator> it(opendb->NewIterator(rocksdb::ReadOptions()));
size_t requested_batch = max_size;
max_size = 0;
for (it->SeekToFirst(); it->Valid(); it->Next()) {
@@ -144,13 +111,6 @@ void ProvenanceRepository::destroy() {
db_.reset();
}
-uint64_t ProvenanceRepository::getKeyCount() const {
- std::string key_count;
- db_->GetProperty("rocksdb.estimate-num-keys", &key_count);
-
- return std::stoull(key_count);
-}
-
REGISTER_RESOURCE_AS(ProvenanceRepository, InternalResource, ("ProvenanceRepository", "provenancerepository"));
} // namespace org::apache::nifi::minifi::provenance
diff --git a/extensions/rocksdb-repos/ProvenanceRepository.h b/extensions/rocksdb-repos/ProvenanceRepository.h
index 7aaeaa34e..ca0b6a8ad 100644
--- a/extensions/rocksdb-repos/ProvenanceRepository.h
+++ b/extensions/rocksdb-repos/ProvenanceRepository.h
@@ -23,14 +23,13 @@
#include <algorithm>
#include <utility>
-#include "rocksdb/db.h"
#include "rocksdb/options.h"
#include "rocksdb/slice.h"
#include "core/Core.h"
#include "core/logging/LoggerConfiguration.h"
-#include "core/ThreadedRepository.h"
#include "provenance/Provenance.h"
#include "utils/Literals.h"
+#include "RocksDbRepository.h"
namespace org::apache::nifi::minifi::provenance {
@@ -39,7 +38,7 @@ constexpr auto MAX_PROVENANCE_STORAGE_SIZE = 10_MiB;
constexpr auto MAX_PROVENANCE_ENTRY_LIFE_TIME = std::chrono::minutes(1);
constexpr auto PROVENANCE_PURGE_PERIOD = std::chrono::milliseconds(2500);
-class ProvenanceRepository : public core::ThreadedRepository {
+class ProvenanceRepository : public core::repository::RocksDbRepository {
public:
ProvenanceRepository(std::string name, const utils::Identifier& /*uuid*/)
: ProvenanceRepository(std::move(name)) {
@@ -50,8 +49,8 @@ class ProvenanceRepository : public core::ThreadedRepository {
std::chrono::milliseconds maxPartitionMillis = MAX_PROVENANCE_ENTRY_LIFE_TIME,
int64_t maxPartitionBytes = MAX_PROVENANCE_STORAGE_SIZE,
std::chrono::milliseconds purgePeriod = PROVENANCE_PURGE_PERIOD)
- : ThreadedRepository(repo_name.length() > 0 ? std::move(repo_name) : core::getClassName<ProvenanceRepository>(),
- directory, maxPartitionMillis, maxPartitionBytes, purgePeriod) {
+ : RocksDbRepository(repo_name.length() > 0 ? std::move(repo_name) : core::getClassName<ProvenanceRepository>(),
+ directory, maxPartitionMillis, maxPartitionBytes, purgePeriod, core::logging::LoggerFactory<ProvenanceRepository>::getLogger()) {
}
~ProvenanceRepository() override {
@@ -62,27 +61,19 @@ class ProvenanceRepository : public core::ThreadedRepository {
EXTENSIONAPI static constexpr bool SupportsDynamicProperties = false;
EXTENSIONAPI static constexpr bool SupportsDynamicRelationships = false;
- void printStats();
-
bool isNoop() const override {
return false;
}
bool initialize(const std::shared_ptr<org::apache::nifi::minifi::Configure> &config) override;
- bool Put(const std::string& key, const uint8_t *buf, size_t bufLen) override;
- bool MultiPut(const std::vector<std::pair<std::string, std::unique_ptr<minifi::io::BufferStream>>>& data) override;
-
bool Delete(const std::string& /*key*/) override {
// The repo is cleaned up by itself, there is no need to delete items.
return true;
}
-
- bool Get(const std::string &key, std::string &value) override;
bool getElements(std::vector<std::shared_ptr<core::SerializableComponent>> &records, size_t &max_size) override;
void destroy();
- uint64_t getKeyCount() const;
// Prevent default copy constructor and assignment operation
// Only support pass by reference or pointer
@@ -93,14 +84,6 @@ class ProvenanceRepository : public core::ThreadedRepository {
private:
// Run function for the thread
void run() override;
-
- std::thread& getThread() override {
- return thread_;
- }
-
- std::unique_ptr<rocksdb::DB> db_;
- std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<ProvenanceRepository>::getLogger();
- std::thread thread_;
};
} // namespace org::apache::nifi::minifi::provenance
diff --git a/extensions/rocksdb-repos/RocksDbRepository.cpp b/extensions/rocksdb-repos/RocksDbRepository.cpp
new file mode 100644
index 000000000..02a735395
--- /dev/null
+++ b/extensions/rocksdb-repos/RocksDbRepository.cpp
@@ -0,0 +1,112 @@
+/**
+ * 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 "RocksDbRepository.h"
+
+using namespace std::literals::chrono_literals;
+
+namespace org::apache::nifi::minifi::core::repository {
+
+void RocksDbRepository::printStats() {
+ auto opendb = db_->open();
+ if (!opendb) {
+ return;
+ }
+ std::string key_count;
+ opendb->GetProperty("rocksdb.estimate-num-keys", &key_count);
+
+ std::string table_readers;
+ opendb->GetProperty("rocksdb.estimate-table-readers-mem", &table_readers);
+
+ std::string all_memtables;
+ opendb->GetProperty("rocksdb.cur-size-all-mem-tables", &all_memtables);
+
+ logger_->log_info("Repository stats: key count: %s, table readers size: %s, all memory tables size: %s",
+ key_count, table_readers, all_memtables);
+}
+
+bool RocksDbRepository::ExecuteWithRetry(const std::function<rocksdb::Status()>& operation) {
+ constexpr int RETRY_COUNT = 3;
+ std::chrono::milliseconds wait_time = 0ms;
+ for (int i=0; i < RETRY_COUNT; ++i) {
+ auto status = operation();
+ if (status.ok()) {
+ logger_->log_trace("Rocksdb operation executed successfully");
+ return true;
+ }
+ logger_->log_error("Rocksdb operation failed: %s", status.ToString());
+ wait_time += FLOWFILE_REPOSITORY_RETRY_INTERVAL_INCREMENTS;
+ std::this_thread::sleep_for(wait_time);
+ }
+ return false;
+}
+
+bool RocksDbRepository::Put(const std::string& key, const uint8_t *buf, size_t bufLen) {
+ auto opendb = db_->open();
+ if (!opendb) {
+ return false;
+ }
+ rocksdb::Slice value((const char *) buf, bufLen);
+ auto operation = [&key, &value, &opendb]() { return opendb->Put(rocksdb::WriteOptions(), key, value); };
+ return ExecuteWithRetry(operation);
+}
+
+bool RocksDbRepository::MultiPut(const std::vector<std::pair<std::string, std::unique_ptr<minifi::io::BufferStream>>>& data) {
+ auto opendb = db_->open();
+ if (!opendb) {
+ return false;
+ }
+ auto batch = opendb->createWriteBatch();
+ for (const auto &item : data) {
+ const auto buf = item.second->getBuffer().as_span<const char>();
+ rocksdb::Slice value(buf.data(), buf.size());
+ if (!batch.Put(item.first, value).ok()) {
+ logger_->log_error("Failed to add item to batch operation");
+ return false;
+ }
+ }
+ auto operation = [&batch, &opendb]() { return opendb->Write(rocksdb::WriteOptions(), &batch); };
+ return ExecuteWithRetry(operation);
+}
+
+bool RocksDbRepository::Get(const std::string &key, std::string &value) {
+ auto opendb = db_->open();
+ if (!opendb) {
+ return false;
+ }
+ return opendb->Get(rocksdb::ReadOptions(), key, &value).ok();
+}
+
+uint64_t RocksDbRepository::getRepositorySize() const {
+ return (utils::optional_from_ptr(db_.get()) |
+ utils::flatMap([](const auto& db) { return db->open(); }) |
+ utils::flatMap([](const auto& opendb) { return opendb.getApproximateSizes(); })).value_or(0);
+}
+
+uint64_t RocksDbRepository::getRepositoryEntryCount() const {
+ return (utils::optional_from_ptr(db_.get()) |
+ utils::flatMap([](const auto& db) { return db->open(); }) |
+ utils::flatMap([](auto&& opendb) -> std::optional<uint64_t> {
+ std::string key_count;
+ opendb.GetProperty("rocksdb.estimate-num-keys", &key_count);
+ if (!key_count.empty()) {
+ return std::stoull(key_count);
+ }
+ return std::nullopt;
+ })).value_or(0);
+}
+
+} // namespace org::apache::nifi::minifi::core::repository
diff --git a/extensions/rocksdb-repos/RocksDbRepository.h b/extensions/rocksdb-repos/RocksDbRepository.h
new file mode 100644
index 000000000..6fd220d8d
--- /dev/null
+++ b/extensions/rocksdb-repos/RocksDbRepository.h
@@ -0,0 +1,63 @@
+/**
+ * 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 <utility>
+#include <vector>
+#include <string>
+#include <memory>
+
+#include "database/RocksDatabase.h"
+#include "core/ThreadedRepository.h"
+
+namespace org::apache::nifi::minifi::core::repository {
+
+constexpr auto FLOWFILE_REPOSITORY_RETRY_INTERVAL_INCREMENTS = std::chrono::milliseconds(500);
+
+class RocksDbRepository : public ThreadedRepository {
+ public:
+ RocksDbRepository(std::string repo_name,
+ std::string directory,
+ std::chrono::milliseconds max_partition_millis,
+ int64_t max_partition_bytes,
+ std::chrono::milliseconds purge_period,
+ std::shared_ptr<logging::Logger> logger)
+ : ThreadedRepository(std::move(repo_name), std::move(directory), max_partition_millis, max_partition_bytes, purge_period),
+ logger_(std::move(logger)) {
+ }
+
+ bool Put(const std::string& key, const uint8_t *buf, size_t bufLen) override;
+ bool MultiPut(const std::vector<std::pair<std::string, std::unique_ptr<minifi::io::BufferStream>>>& data) override;
+ bool Get(const std::string &key, std::string &value) override;
+
+ uint64_t getRepositorySize() const override;
+ uint64_t getRepositoryEntryCount() const override;
+ void printStats();
+
+ protected:
+ bool ExecuteWithRetry(const std::function<rocksdb::Status()>& operation);
+
+ std::thread& getThread() override {
+ return thread_;
+ }
+
+ std::unique_ptr<minifi::internal::RocksDatabase> db_;
+ std::shared_ptr<logging::Logger> logger_;
+ std::thread thread_;
+};
+
+} // namespace org::apache::nifi::minifi::core::repository
diff --git a/extensions/rocksdb-repos/database/OpenRocksDb.cpp b/extensions/rocksdb-repos/database/OpenRocksDb.cpp
index be5f1960c..703c66a39 100644
--- a/extensions/rocksdb-repos/database/OpenRocksDb.cpp
+++ b/extensions/rocksdb-repos/database/OpenRocksDb.cpp
@@ -122,4 +122,17 @@ rocksdb::DB* OpenRocksDb::get() {
return impl_.get();
}
+std::optional<uint64_t> OpenRocksDb::getApproximateSizes() const {
+ const rocksdb::SizeApproximationOptions options{ .include_memtables = true };
+ std::string del_char_str(1, static_cast<char>(127));
+ std::string empty_str;
+ const rocksdb::Range range(empty_str, del_char_str);
+ uint64_t value = 0;
+ auto status = impl_->GetApproximateSizes(options, column_->handle.get(), &range, 1, &value);
+ if (status.ok()) {
+ return value;
+ }
+ return std::nullopt;
+}
+
} // namespace org::apache::nifi::minifi::internal
diff --git a/extensions/rocksdb-repos/database/OpenRocksDb.h b/extensions/rocksdb-repos/database/OpenRocksDb.h
index ebe25aacc..b4d964381 100644
--- a/extensions/rocksdb-repos/database/OpenRocksDb.h
+++ b/extensions/rocksdb-repos/database/OpenRocksDb.h
@@ -21,6 +21,7 @@
#include <memory>
#include <string>
#include <vector>
+#include <optional>
#include "utils/gsl.h"
#include "rocksdb/db.h"
@@ -71,6 +72,8 @@ class OpenRocksDb {
rocksdb::DB* get();
+ std::optional<uint64_t> getApproximateSizes() const;
+
private:
void handleResult(const rocksdb::Status& result);
void handleResult(const std::vector<rocksdb::Status>& results);
diff --git a/libminifi/include/core/ContentRepository.h b/libminifi/include/core/ContentRepository.h
index 17762af26..68230d42b 100644
--- a/libminifi/include/core/ContentRepository.h
+++ b/libminifi/include/core/ContentRepository.h
@@ -26,22 +26,20 @@
#include "properties/Configure.h"
#include "ResourceClaim.h"
#include "StreamManager.h"
-#include "core/Core.h"
#include "ContentSession.h"
+#include "core/RepositoryMetricsSource.h"
+#include "core/Core.h"
namespace org::apache::nifi::minifi::core {
/**
* Content repository definition that extends StreamManager.
*/
-class ContentRepository : public StreamManager<minifi::ResourceClaim>, public utils::EnableSharedFromThis<ContentRepository>, public core::CoreComponent {
+class ContentRepository : public core::CoreComponent, public StreamManager<minifi::ResourceClaim>, public utils::EnableSharedFromThis<ContentRepository>, public core::RepositoryMetricsSource {
public:
explicit ContentRepository(std::string name, const utils::Identifier& uuid = {}) : core::CoreComponent(std::move(name), uuid) {}
~ContentRepository() override = default;
- /**
- * initialize this content repository using the provided configuration.
- */
virtual bool initialize(const std::shared_ptr<Configure> &configure) = 0;
std::string getStoragePath() const override;
@@ -59,6 +57,10 @@ class ContentRepository : public StreamManager<minifi::ResourceClaim>, public ut
bool remove(const minifi::ResourceClaim &streamId) final;
+ std::string getRepositoryName() const override {
+ return getName();
+ }
+
protected:
void removeFromPurgeList();
virtual bool removeKey(const std::string& content_path) = 0;
diff --git a/libminifi/include/core/Repository.h b/libminifi/include/core/Repository.h
index 77fdfbec3..0c74732bc 100644
--- a/libminifi/include/core/Repository.h
+++ b/libminifi/include/core/Repository.h
@@ -29,19 +29,20 @@
#include <utility>
#include <vector>
-#include "Core.h"
#include "ResourceClaim.h"
#include "core/Connectable.h"
#include "core/ContentRepository.h"
#include "core/Property.h"
#include "core/SerializableComponent.h"
#include "core/logging/LoggerFactory.h"
+#include "core/RepositoryMetricsSource.h"
#include "properties/Configure.h"
#include "utils/BackTrace.h"
#include "SwapManager.h"
#include "utils/Literals.h"
#include "utils/StringUtils.h"
#include "utils/TimeUtil.h"
+#include "core/Core.h"
#ifndef WIN32
#include <sys/stat.h>
@@ -54,7 +55,7 @@ constexpr auto MAX_REPOSITORY_STORAGE_SIZE = 10_MiB;
constexpr auto MAX_REPOSITORY_ENTRY_LIFE_TIME = std::chrono::minutes(10);
constexpr auto REPOSITORY_PURGE_PERIOD = std::chrono::milliseconds(2500);
-class Repository : public core::CoreComponent {
+class Repository : public core::CoreComponent, public core::RepositoryMetricsSource {
public:
explicit Repository(std::string repo_name = "Repository",
std::string directory = REPOSITORY_DIRECTORY,
@@ -62,20 +63,15 @@ class Repository : public core::CoreComponent {
int64_t maxPartitionBytes = MAX_REPOSITORY_STORAGE_SIZE,
std::chrono::milliseconds purgePeriod = REPOSITORY_PURGE_PERIOD)
: core::CoreComponent(std::move(repo_name)),
- directory_(std::move(directory)),
max_partition_millis_(maxPartitionMillis),
max_partition_bytes_(maxPartitionBytes),
purge_period_(purgePeriod),
repo_full_(false),
- repo_size_(0),
+ directory_(std::move(directory)),
logger_(logging::LoggerFactory<Repository>::getLogger()) {
}
- virtual bool isRunning() const {
- return true;
- }
-
- virtual bool initialize(const std::shared_ptr<Configure>& /*configure*/) = 0;
+ virtual bool initialize(const std::shared_ptr<Configure> &configure) = 0;
virtual bool start() = 0;
virtual bool stop() = 0;
@@ -116,10 +112,6 @@ class Repository : public core::CoreComponent {
return false;
}
- virtual bool isFull() {
- return repo_full_;
- }
-
virtual bool getElements(std::vector<std::shared_ptr<core::SerializableComponent>>& /*store*/, size_t& /*max_size*/) {
return true;
}
@@ -129,14 +121,14 @@ class Repository : public core::CoreComponent {
virtual void loadComponent(const std::shared_ptr<core::ContentRepository>& /*content_repo*/) {
}
- virtual uint64_t getRepoSize() const {
- return repo_size_;
- }
-
std::string getDirectory() const {
return directory_;
}
+ std::string getRepositoryName() const override {
+ return getName();
+ }
+
// Prevent default copy constructor and assignment operation
// Only support pass by reference or pointer
Repository(const Repository &parent) = delete;
@@ -146,15 +138,13 @@ class Repository : public core::CoreComponent {
std::map<std::string, core::Connectable*> containers_;
std::map<std::string, core::Connectable*> connection_map_;
- std::string directory_;
// max db entry lifetime
std::chrono::milliseconds max_partition_millis_;
// max db size
int64_t max_partition_bytes_;
std::chrono::milliseconds purge_period_;
std::atomic<bool> repo_full_;
-
- std::atomic<uint64_t> repo_size_;
+ std::string directory_;
private:
std::shared_ptr<logging::Logger> logger_;
diff --git a/libminifi/include/core/RepositoryFactory.h b/libminifi/include/core/RepositoryFactory.h
index 60ba5775a..352dec51f 100644
--- a/libminifi/include/core/RepositoryFactory.h
+++ b/libminifi/include/core/RepositoryFactory.h
@@ -16,8 +16,7 @@
* limitations under the License.
*/
-#ifndef LIBMINIFI_INCLUDE_CORE_REPOSITORYFACTORY_H_
-#define LIBMINIFI_INCLUDE_CORE_REPOSITORYFACTORY_H_
+#pragma once
#include <memory>
#include <string>
@@ -45,5 +44,3 @@ std::unique_ptr<core::ContentRepository> createContentRepository(const std::stri
std::unique_ptr<core::Repository> createRepository(const std::string& configuration_class_name, const std::string& repo_name = "");
} // namespace org::apache::nifi::minifi::core
-
-#endif // LIBMINIFI_INCLUDE_CORE_REPOSITORYFACTORY_H_
diff --git a/libminifi/include/core/RepositoryMetricsSource.h b/libminifi/include/core/RepositoryMetricsSource.h
new file mode 100644
index 000000000..cfaeb4ae6
--- /dev/null
+++ b/libminifi/include/core/RepositoryMetricsSource.h
@@ -0,0 +1,45 @@
+/**
+ *
+ * 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 <string>
+
+namespace org::apache::nifi::minifi::core {
+
+class RepositoryMetricsSource {
+ public:
+ virtual ~RepositoryMetricsSource() = default;
+ virtual uint64_t getRepositorySize() const = 0;
+ virtual uint64_t getRepositoryEntryCount() const = 0;
+ virtual std::string getRepositoryName() const = 0;
+
+ virtual uint64_t getMaxRepositorySize() const {
+ return 0;
+ }
+
+ virtual bool isFull() const {
+ return false;
+ }
+
+ virtual bool isRunning() const {
+ return true;
+ }
+};
+
+} // namespace org::apache::nifi::minifi::core
diff --git a/libminifi/include/core/repository/FileSystemRepository.h b/libminifi/include/core/repository/FileSystemRepository.h
index b0c3900cd..99a26b4e3 100644
--- a/libminifi/include/core/repository/FileSystemRepository.h
+++ b/libminifi/include/core/repository/FileSystemRepository.h
@@ -21,10 +21,12 @@
#include <memory>
#include <string>
#include <utility>
+#include <algorithm>
#include "../ContentRepository.h"
#include "properties/Configure.h"
#include "core/logging/LoggerFactory.h"
+#include "utils/file/FileUtils.h"
namespace org::apache::nifi::minifi::core::repository {
@@ -50,6 +52,18 @@ class FileSystemRepository : public core::ContentRepository {
void clearOrphans() override;
+ uint64_t getRepositorySize() const override {
+ return utils::file::path_size(directory_);
+ }
+
+ uint64_t getRepositoryEntryCount() const override {
+ auto dir_it = std::filesystem::recursive_directory_iterator(directory_, std::filesystem::directory_options::skip_permission_denied);
+ return std::count_if(
+ std::filesystem::begin(dir_it),
+ std::filesystem::end(dir_it),
+ [](auto& entry) { return entry.is_regular_file(); });
+ }
+
protected:
bool removeKey(const std::string& content_path) override;
diff --git a/libminifi/include/core/repository/VolatileContentRepository.h b/libminifi/include/core/repository/VolatileContentRepository.h
index a5f6027e9..0690255f7 100644
--- a/libminifi/include/core/repository/VolatileContentRepository.h
+++ b/libminifi/include/core/repository/VolatileContentRepository.h
@@ -24,12 +24,12 @@
#include "AtomicRepoEntries.h"
#include "io/AtomicEntryStream.h"
#include "../ContentRepository.h"
-#include "core/repository/VolatileRepository.h"
#include "properties/Configure.h"
#include "core/Connectable.h"
#include "core/logging/LoggerFactory.h"
#include "utils/GeneralUtils.h"
#include "VolatileRepositoryData.h"
+#include "utils/Literals.h"
namespace org::apache::nifi::minifi::core::repository {
/**
@@ -58,6 +58,22 @@ class VolatileContentRepository : public core::ContentRepository {
}
}
+ uint64_t getRepositorySize() const override {
+ return repo_data_.getRepositorySize();
+ }
+
+ uint64_t getMaxRepositorySize() const override {
+ return repo_data_.getMaxRepositorySize();
+ }
+
+ uint64_t getRepositoryEntryCount() const override {
+ return master_list_.size();
+ }
+
+ bool isFull() const override {
+ return repo_data_.isFull();
+ }
+
/**
* Initialize the volatile content repo
* @param configure configuration
diff --git a/libminifi/include/core/repository/VolatileRepository.h b/libminifi/include/core/repository/VolatileRepository.h
index 6277b2a12..f22ace0ad 100644
--- a/libminifi/include/core/repository/VolatileRepository.h
+++ b/libminifi/include/core/repository/VolatileRepository.h
@@ -81,8 +81,20 @@ class VolatileRepository : public core::ThreadedRepository {
*/
bool Get(const std::string& key, std::string &value) override;
- uint64_t getRepoSize() const override {
- return repo_data_.current_size;
+ uint64_t getRepositorySize() const override {
+ return repo_data_.getRepositorySize();
+ }
+
+ uint64_t getMaxRepositorySize() const override {
+ return repo_data_.getMaxRepositorySize();
+ }
+
+ uint64_t getRepositoryEntryCount() const override {
+ return repo_data_.getRepositoryEntryCount();
+ }
+
+ bool isFull() const override {
+ return repo_data_.isFull();
}
protected:
diff --git a/libminifi/include/core/repository/VolatileRepositoryData.h b/libminifi/include/core/repository/VolatileRepositoryData.h
index a428126c5..86d8161b7 100644
--- a/libminifi/include/core/repository/VolatileRepositoryData.h
+++ b/libminifi/include/core/repository/VolatileRepositoryData.h
@@ -35,8 +35,26 @@ struct VolatileRepositoryData {
void initialize(const std::shared_ptr<Configure> &configure, const std::string& repo_name);
void clear();
+
+ uint64_t getRepositorySize() const {
+ return current_size;
+ }
+
+ uint64_t getMaxRepositorySize() const {
+ return max_size;
+ }
+
+ uint64_t getRepositoryEntryCount() const {
+ return current_entry_count;
+ }
+
+ bool isFull() const {
+ return current_size >= max_size;
+ }
+
// current size of the volatile repo.
std::atomic<size_t> current_size;
+ std::atomic<size_t> current_entry_count;
// value vector that exists for non blocking iteration over
// objects that store data for this repo instance.
std::vector<AtomicEntry<std::string>*> value_vector;
diff --git a/libminifi/include/core/state/MetricsPublisherStore.h b/libminifi/include/core/state/MetricsPublisherStore.h
index 90eab8aa1..b4fafaa92 100644
--- a/libminifi/include/core/state/MetricsPublisherStore.h
+++ b/libminifi/include/core/state/MetricsPublisherStore.h
@@ -21,6 +21,7 @@
#include <string>
#include <utility>
#include <memory>
+#include <vector>
#include "MetricsPublisher.h"
#include "core/state/nodes/ResponseNodeLoader.h"
@@ -31,8 +32,8 @@ namespace org::apache::nifi::minifi::state {
class MetricsPublisherStore {
public:
- MetricsPublisherStore(std::shared_ptr<Configure> configuration, std::shared_ptr<core::Repository> provenance_repo,
- std::shared_ptr<core::Repository> flow_file_repo, std::shared_ptr<core::FlowConfiguration> flow_configuration);
+ MetricsPublisherStore(std::shared_ptr<Configure> configuration, const std::vector<std::shared_ptr<core::RepositoryMetricsSource>>& repository_metric_sources,
+ std::shared_ptr<core::FlowConfiguration> flow_configuration);
void initialize(core::controller::ControllerServiceProvider* controller, state::StateMonitor* update_sink);
void loadMetricNodes(core::ProcessGroup* root);
void clearMetricNodes();
diff --git a/libminifi/include/core/state/nodes/AgentInformation.h b/libminifi/include/core/state/nodes/AgentInformation.h
index d38465295..0d18c721e 100644
--- a/libminifi/include/core/state/nodes/AgentInformation.h
+++ b/libminifi/include/core/state/nodes/AgentInformation.h
@@ -61,6 +61,7 @@
#include "core/AgentIdentificationProvider.h"
#include "utils/Export.h"
#include "SupportedOperations.h"
+#include "core/RepositoryMetricsSource.h"
namespace org::apache::nifi::minifi::state::response {
@@ -423,13 +424,13 @@ class AgentStatus : public StateMonitorNode {
return "AgentStatus";
}
- void setRepositories(const std::map<std::string, std::shared_ptr<core::Repository>> &repositories) {
+ void setRepositories(const std::map<std::string, std::shared_ptr<core::RepositoryMetricsSource>> &repositories) {
repositories_ = repositories;
}
- void addRepository(const std::shared_ptr<core::Repository> &repo) {
+ void addRepository(const std::shared_ptr<core::RepositoryMetricsSource> &repo) {
if (nullptr != repo) {
- repositories_.insert(std::make_pair(repo->getName(), repo));
+ repositories_.insert(std::make_pair(repo->getRepositoryName(), repo));
}
}
@@ -454,9 +455,11 @@ class AgentStatus : public StateMonitorNode {
std::vector<PublishedMetric> calculateMetrics() override {
std::vector<PublishedMetric> metrics;
for (const auto& [_, repo] : repositories_) {
- metrics.push_back({"is_running", (repo->isRunning() ? 1.0 : 0.0), {{"metric_class", getName()}, {"repository_name", repo->getName()}}});
- metrics.push_back({"is_full", (repo->isFull() ? 1.0 : 0.0), {{"metric_class", getName()}, {"repository_name", repo->getName()}}});
- metrics.push_back({"repository_size", static_cast<double>(repo->getRepoSize()), {{"metric_class", getName()}, {"repository_name", repo->getName()}}});
+ metrics.push_back({"is_running", (repo->isRunning() ? 1.0 : 0.0), {{"metric_class", getName()}, {"repository_name", repo->getRepositoryName()}}});
+ metrics.push_back({"is_full", (repo->isFull() ? 1.0 : 0.0), {{"metric_class", getName()}, {"repository_name", repo->getRepositoryName()}}});
+ metrics.push_back({"repository_size_bytes", static_cast<double>(repo->getRepositorySize()), {{"metric_class", getName()}, {"repository_name", repo->getRepositoryName()}}});
+ metrics.push_back({"max_repository_size_bytes", static_cast<double>(repo->getMaxRepositorySize()), {{"metric_class", getName()}, {"repository_name", repo->getRepositoryName()}}});
+ metrics.push_back({"repository_entry_count", static_cast<double>(repo->getRepositoryEntryCount()), {{"metric_class", getName()}, {"repository_name", repo->getRepositoryName()}}});
}
if (nullptr != monitor_) {
auto uptime = monitor_->getUptime();
@@ -488,26 +491,36 @@ class AgentStatus : public StateMonitorNode {
repositories.name = "repositories";
for (const auto& repo : repositories_) {
- SerializedResponseNode repoNode;
- repoNode.collapsible = false;
- repoNode.name = repo.first;
-
- SerializedResponseNode queuesize;
- queuesize.name = "size";
- queuesize.value = repo.second->getRepoSize();
-
- SerializedResponseNode isRunning;
- isRunning.name = "running";
- isRunning.value = repo.second->isRunning();
-
- SerializedResponseNode isFull;
- isFull.name = "full";
- isFull.value = repo.second->isFull();
-
- repoNode.children.push_back(queuesize);
- repoNode.children.push_back(isRunning);
- repoNode.children.push_back(isFull);
- repositories.children.push_back(repoNode);
+ SerializedResponseNode repo_node;
+ repo_node.collapsible = false;
+ repo_node.name = repo.first;
+
+ SerializedResponseNode repo_size;
+ repo_size.name = "size";
+ repo_size.value = repo.second->getRepositorySize();
+
+ SerializedResponseNode max_repo_size;
+ max_repo_size.name = "maxSize";
+ max_repo_size.value = repo.second->getMaxRepositorySize();
+
+ SerializedResponseNode repo_entry_count;
+ repo_entry_count.name = "entryCount";
+ repo_entry_count.value = repo.second->getRepositoryEntryCount();
+
+ SerializedResponseNode is_running;
+ is_running.name = "running";
+ is_running.value = repo.second->isRunning();
+
+ SerializedResponseNode is_full;
+ is_full.name = "full";
+ is_full.value = repo.second->isFull();
+
+ repo_node.children.push_back(repo_size);
+ repo_node.children.push_back(max_repo_size);
+ repo_node.children.push_back(repo_entry_count);
+ repo_node.children.push_back(is_running);
+ repo_node.children.push_back(is_full);
+ repositories.children.push_back(repo_node);
}
return repositories;
}
@@ -578,7 +591,7 @@ class AgentStatus : public StateMonitorNode {
return resource_consumption;
}
- std::map<std::string, std::shared_ptr<core::Repository>> repositories_;
+ std::map<std::string, std::shared_ptr<core::RepositoryMetricsSource>> repositories_;
MINIFIAPI static utils::ProcessCpuUsageTracker cpu_load_tracker_;
MINIFIAPI static std::mutex cpu_load_tracker_mutex_;
@@ -608,9 +621,9 @@ class AgentMonitor {
AgentMonitor()
: monitor_(nullptr) {
}
- void addRepository(const std::shared_ptr<core::Repository> &repo) {
+ void addRepository(const std::shared_ptr<core::RepositoryMetricsSource> &repo) {
if (nullptr != repo) {
- repositories_.insert(std::make_pair(repo->getName(), repo));
+ repositories_.insert(std::make_pair(repo->getRepositoryName(), repo));
}
}
@@ -619,7 +632,7 @@ class AgentMonitor {
}
protected:
- std::map<std::string, std::shared_ptr<core::Repository>> repositories_;
+ std::map<std::string, std::shared_ptr<core::RepositoryMetricsSource>> repositories_;
state::StateMonitor* monitor_ = nullptr;
};
diff --git a/libminifi/include/core/state/nodes/RepositoryMetrics.h b/libminifi/include/core/state/nodes/RepositoryMetrics.h
index 599730bef..bc8dada72 100644
--- a/libminifi/include/core/state/nodes/RepositoryMetrics.h
+++ b/libminifi/include/core/state/nodes/RepositoryMetrics.h
@@ -26,6 +26,7 @@
#include "../nodes/MetricsBase.h"
#include "Connection.h"
+#include "core/RepositoryMetricsSource.h"
namespace org::apache::nifi::minifi::state::response {
@@ -54,33 +55,42 @@ class RepositoryMetrics : public ResponseNode {
return "RepositoryMetrics";
}
- void addRepository(const std::shared_ptr<core::Repository> &repo) {
+ void addRepository(const std::shared_ptr<core::RepositoryMetricsSource> &repo) {
if (nullptr != repo) {
- repositories_.insert(std::make_pair(repo->getName(), repo));
+ repositories_.push_back(repo);
}
}
std::vector<SerializedResponseNode> serialize() override {
std::vector<SerializedResponseNode> serialized;
- for (auto conn : repositories_) {
- auto repo = conn.second;
+ for (const auto& repo : repositories_) {
SerializedResponseNode parent;
- parent.name = repo->getName();
- SerializedResponseNode datasize;
- datasize.name = "running";
- datasize.value = repo->isRunning();
+ parent.name = repo->getRepositoryName();
+ SerializedResponseNode is_running;
+ is_running.name = "running";
+ is_running.value = repo->isRunning();
- SerializedResponseNode datasizemax;
- datasizemax.name = "full";
- datasizemax.value = repo->isFull();
+ SerializedResponseNode is_full;
+ is_full.name = "full";
+ is_full.value = repo->isFull();
- SerializedResponseNode queuesize;
- queuesize.name = "size";
- queuesize.value = std::to_string(repo->getRepoSize());
+ SerializedResponseNode repo_size;
+ repo_size.name = "size";
+ repo_size.value = std::to_string(repo->getRepositorySize());
- parent.children.push_back(datasize);
- parent.children.push_back(datasizemax);
- parent.children.push_back(queuesize);
+ SerializedResponseNode max_repo_size;
+ max_repo_size.name = "maxSize";
+ max_repo_size.value = std::to_string(repo->getMaxRepositorySize());
+
+ SerializedResponseNode repo_entry_count;
+ repo_entry_count.name = "entryCount";
+ repo_entry_count.value = repo->getRepositoryEntryCount();
+
+ parent.children.push_back(is_running);
+ parent.children.push_back(is_full);
+ parent.children.push_back(repo_size);
+ parent.children.push_back(max_repo_size);
+ parent.children.push_back(repo_entry_count);
serialized.push_back(parent);
}
@@ -89,16 +99,18 @@ class RepositoryMetrics : public ResponseNode {
std::vector<PublishedMetric> calculateMetrics() override {
std::vector<PublishedMetric> metrics;
- for (const auto& [_, repo] : repositories_) {
- metrics.push_back({"is_running", (repo->isRunning() ? 1.0 : 0.0), {{"metric_class", getName()}, {"repository_name", repo->getName()}}});
- metrics.push_back({"is_full", (repo->isFull() ? 1.0 : 0.0), {{"metric_class", getName()}, {"repository_name", repo->getName()}}});
- metrics.push_back({"repository_size", static_cast<double>(repo->getRepoSize()), {{"metric_class", getName()}, {"repository_name", repo->getName()}}});
+ for (const auto& repo : repositories_) {
+ metrics.push_back({"is_running", (repo->isRunning() ? 1.0 : 0.0), {{"metric_class", getName()}, {"repository_name", repo->getRepositoryName()}}});
+ metrics.push_back({"is_full", (repo->isFull() ? 1.0 : 0.0), {{"metric_class", getName()}, {"repository_name", repo->getRepositoryName()}}});
+ metrics.push_back({"repository_size_bytes", static_cast<double>(repo->getRepositorySize()), {{"metric_class", getName()}, {"repository_name", repo->getRepositoryName()}}});
+ metrics.push_back({"max_repository_size_bytes", static_cast<double>(repo->getMaxRepositorySize()), {{"metric_class", getName()}, {"repository_name", repo->getRepositoryName()}}});
+ metrics.push_back({"repository_entry_count", static_cast<double>(repo->getRepositoryEntryCount()), {{"metric_class", getName()}, {"repository_name", repo->getRepositoryName()}}});
}
return metrics;
}
protected:
- std::map<std::string, std::shared_ptr<core::Repository>> repositories_;
+ std::vector<std::shared_ptr<core::RepositoryMetricsSource>> repositories_;
};
} // namespace org::apache::nifi::minifi::state::response
diff --git a/libminifi/include/core/state/nodes/ResponseNodeLoader.h b/libminifi/include/core/state/nodes/ResponseNodeLoader.h
index 719d2e9a5..af330585c 100644
--- a/libminifi/include/core/state/nodes/ResponseNodeLoader.h
+++ b/libminifi/include/core/state/nodes/ResponseNodeLoader.h
@@ -33,13 +33,14 @@
#include "utils/gsl.h"
#include "utils/Id.h"
#include "utils/expected.h"
+#include "core/RepositoryMetricsSource.h"
namespace org::apache::nifi::minifi::state::response {
class ResponseNodeLoader {
public:
- ResponseNodeLoader(std::shared_ptr<Configure> configuration, std::shared_ptr<core::Repository> provenance_repo,
- std::shared_ptr<core::Repository> flow_file_repo, std::shared_ptr<core::FlowConfiguration> flow_configuration);
+ ResponseNodeLoader(std::shared_ptr<Configure> configuration, std::vector<std::shared_ptr<core::RepositoryMetricsSource>> repository_metric_sources,
+ std::shared_ptr<core::FlowConfiguration> flow_configuration);
void setNewConfigRoot(core::ProcessGroup* root);
void clearConfigRoot();
@@ -70,8 +71,7 @@ class ResponseNodeLoader {
std::unordered_map<std::string, std::vector<SharedResponseNode>> component_metrics_;
std::unordered_map<std::string, SharedResponseNode> system_metrics_;
std::shared_ptr<Configure> configuration_;
- std::shared_ptr<core::Repository> provenance_repo_;
- std::shared_ptr<core::Repository> flow_file_repo_;
+ std::vector<std::shared_ptr<core::RepositoryMetricsSource>> repository_metric_sources_;
std::shared_ptr<core::FlowConfiguration> flow_configuration_;
core::controller::ControllerServiceProvider* controller_{};
state::StateMonitor* update_sink_{};
diff --git a/libminifi/include/utils/file/FileUtils.h b/libminifi/include/utils/file/FileUtils.h
index db37d861b..ba778f583 100644
--- a/libminifi/include/utils/file/FileUtils.h
+++ b/libminifi/include/utils/file/FileUtils.h
@@ -206,6 +206,20 @@ inline bool is_directory(const std::filesystem::path &path) {
return false;
}
+inline uint64_t path_size(const std::filesystem::path& path) {
+ uint64_t size = 0;
+ if (std::filesystem::is_regular_file(path)) {
+ return utils::file::file_size(path);
+ } else if (utils::file::is_directory(path)) {
+ for (const std::filesystem::directory_entry& entry : std::filesystem::recursive_directory_iterator(path, std::filesystem::directory_options::skip_permission_denied)) {
+ if (entry.is_regular_file()) {
+ size += entry.file_size();
+ }
+ }
+ }
+ return size;
+}
+
inline bool exists(const std::filesystem::path &path) {
std::error_code ec;
bool result = std::filesystem::exists(path, ec);
diff --git a/libminifi/src/core/RepositoryFactory.cpp b/libminifi/src/core/RepositoryFactory.cpp
index 8455043cb..bc79e136d 100644
--- a/libminifi/src/core/RepositoryFactory.cpp
+++ b/libminifi/src/core/RepositoryFactory.cpp
@@ -37,6 +37,7 @@ std::unique_ptr<core::ContentRepository> createContentRepository(const std::stri
auto return_obj = core::ClassLoader::getDefaultClassLoader().instantiate<core::ContentRepository>(class_name_lc,
class_name_lc);
if (return_obj) {
+ return_obj->setName(repo_name);
return return_obj;
}
if (class_name_lc == "volatilecontentrepository") {
@@ -68,6 +69,14 @@ class NoOpThreadedRepository : public core::ThreadedRepository {
stop();
}
+ uint64_t getRepositorySize() const override {
+ return 0;
+ }
+
+ uint64_t getRepositoryEntryCount() const override {
+ return 0;
+ }
+
private:
void run() override {
}
diff --git a/libminifi/src/core/repository/VolatileRepository.cpp b/libminifi/src/core/repository/VolatileRepository.cpp
index 5171376cd..161f9f357 100644
--- a/libminifi/src/core/repository/VolatileRepository.cpp
+++ b/libminifi/src/core/repository/VolatileRepository.cpp
@@ -71,6 +71,9 @@ bool VolatileRepository::Put(const std::string& key, const uint8_t *buf, size_t
}
} while (!updated);
repo_data_.current_size += size;
+ if (repo_data_.current_entry_count < repo_data_.max_count) {
+ ++repo_data_.current_entry_count;
+ }
logger_->log_debug("VolatileRepository -- put %zu %" PRIu32, repo_data_.current_size.load(), current_index_.load());
return true;
@@ -92,6 +95,7 @@ bool VolatileRepository::Delete(const std::string& key) {
RepoValue<std::string> value;
if (ent->getValue(key, value)) {
repo_data_.current_size -= value.size();
+ --repo_data_.current_entry_count;
logger_->log_debug("Delete and pushed into purge_list from volatile");
emplace(value);
return true;
@@ -102,10 +106,8 @@ bool VolatileRepository::Delete(const std::string& key) {
bool VolatileRepository::Get(const std::string &key, std::string &value) {
for (auto ent : repo_data_.value_vector) {
- // let the destructor do the cleanup
RepoValue<std::string> repo_value;
if (ent->getValue(key, repo_value)) {
- repo_data_.current_size -= value.size();
repo_value.emplace(value);
return true;
}
diff --git a/libminifi/src/core/repository/VolatileRepositoryData.cpp b/libminifi/src/core/repository/VolatileRepositoryData.cpp
index a5ab41aff..09633ff40 100644
--- a/libminifi/src/core/repository/VolatileRepositoryData.cpp
+++ b/libminifi/src/core/repository/VolatileRepositoryData.cpp
@@ -23,6 +23,7 @@ namespace org::apache::nifi::minifi::core::repository {
VolatileRepositoryData::VolatileRepositoryData(uint32_t max_count, size_t max_size)
: current_size(0),
+ current_entry_count(0),
max_count(max_count),
max_size(max_size) {
}
diff --git a/libminifi/src/core/state/MetricsPublisherStore.cpp b/libminifi/src/core/state/MetricsPublisherStore.cpp
index d4622b8ea..c837d264a 100644
--- a/libminifi/src/core/state/MetricsPublisherStore.cpp
+++ b/libminifi/src/core/state/MetricsPublisherStore.cpp
@@ -22,10 +22,10 @@
namespace org::apache::nifi::minifi::state {
-MetricsPublisherStore::MetricsPublisherStore(std::shared_ptr<Configure> configuration, std::shared_ptr<core::Repository> provenance_repo,
- std::shared_ptr<core::Repository> flow_file_repo, std::shared_ptr<core::FlowConfiguration> flow_configuration)
+MetricsPublisherStore::MetricsPublisherStore(std::shared_ptr<Configure> configuration, const std::vector<std::shared_ptr<core::RepositoryMetricsSource>>& repository_metric_sources,
+ std::shared_ptr<core::FlowConfiguration> flow_configuration)
: configuration_(configuration),
- response_node_loader_(std::make_shared<response::ResponseNodeLoader>(std::move(configuration), std::move(provenance_repo), std::move(flow_file_repo), std::move(flow_configuration))) {
+ response_node_loader_(std::make_shared<response::ResponseNodeLoader>(std::move(configuration), repository_metric_sources, std::move(flow_configuration))) {
}
void MetricsPublisherStore::initialize(core::controller::ControllerServiceProvider* controller, state::StateMonitor* update_sink) {
diff --git a/libminifi/src/core/state/nodes/ResponseNodeLoader.cpp b/libminifi/src/core/state/nodes/ResponseNodeLoader.cpp
index aeed88e99..66fe7f213 100644
--- a/libminifi/src/core/state/nodes/ResponseNodeLoader.cpp
+++ b/libminifi/src/core/state/nodes/ResponseNodeLoader.cpp
@@ -32,12 +32,11 @@
namespace org::apache::nifi::minifi::state::response {
-ResponseNodeLoader::ResponseNodeLoader(std::shared_ptr<Configure> configuration, std::shared_ptr<core::Repository> provenance_repo,
- std::shared_ptr<core::Repository> flow_file_repo, std::shared_ptr<core::FlowConfiguration> flow_configuration)
- : configuration_(std::move(configuration)),
- provenance_repo_(std::move(provenance_repo)),
- flow_file_repo_(std::move(flow_file_repo)),
- flow_configuration_(std::move(flow_configuration)) {
+ResponseNodeLoader::ResponseNodeLoader(std::shared_ptr<Configure> configuration, std::vector<std::shared_ptr<core::RepositoryMetricsSource>> repository_metric_sources,
+ std::shared_ptr<core::FlowConfiguration> flow_configuration)
+ : configuration_(std::move(configuration)),
+ repository_metric_sources_(std::move(repository_metric_sources)),
+ flow_configuration_(std::move(flow_configuration)) {
}
void ResponseNodeLoader::clearConfigRoot() {
@@ -124,8 +123,9 @@ std::vector<SharedResponseNode> ResponseNodeLoader::getResponseNodes(const std::
void ResponseNodeLoader::initializeRepositoryMetrics(const SharedResponseNode& response_node) const {
auto repository_metrics = dynamic_cast<RepositoryMetrics*>(response_node.get());
if (repository_metrics != nullptr) {
- repository_metrics->addRepository(provenance_repo_);
- repository_metrics->addRepository(flow_file_repo_);
+ for (const auto& repo : repository_metric_sources_) {
+ repository_metrics->addRepository(repo);
+ }
}
}
@@ -155,8 +155,9 @@ void ResponseNodeLoader::initializeAgentIdentifier(const SharedResponseNode& res
void ResponseNodeLoader::initializeAgentMonitor(const SharedResponseNode& response_node) const {
auto monitor = dynamic_cast<state::response::AgentMonitor*>(response_node.get());
if (monitor != nullptr) {
- monitor->addRepository(provenance_repo_);
- monitor->addRepository(flow_file_repo_);
+ for (const auto& repo : repository_metric_sources_) {
+ monitor->addRepository(repo);
+ }
monitor->setStateMonitor(update_sink_);
}
}
@@ -176,8 +177,9 @@ void ResponseNodeLoader::initializeAgentNode(const SharedResponseNode& response_
void ResponseNodeLoader::initializeAgentStatus(const SharedResponseNode& response_node) const {
auto agent_status = dynamic_cast<state::response::AgentStatus*>(response_node.get());
if (agent_status != nullptr) {
- agent_status->addRepository(provenance_repo_);
- agent_status->addRepository(flow_file_repo_);
+ for (const auto& repo : repository_metric_sources_) {
+ agent_status->addRepository(repo);
+ }
agent_status->setStateMonitor(update_sink_);
}
}
diff --git a/libminifi/test/flow-tests/TestControllerWithFlow.h b/libminifi/test/flow-tests/TestControllerWithFlow.h
index ced53fea9..760d903e3 100644
--- a/libminifi/test/flow-tests/TestControllerWithFlow.h
+++ b/libminifi/test/flow-tests/TestControllerWithFlow.h
@@ -21,6 +21,7 @@
#include <string>
#include <memory>
#include <utility>
+#include <vector>
#include "FlowController.h"
#include "unit/ProvenanceTestHelper.h"
@@ -68,7 +69,8 @@ class TestControllerWithFlow: public TestController {
auto flow = std::make_shared<core::YamlConfiguration>(core::ConfigurationContext{ff_repo, content_repo, stream_factory, configuration_, yaml_path_.string()});
auto root = flow->getRoot();
root_ = root.get();
- auto metrics_publisher_store = std::make_unique<minifi::state::MetricsPublisherStore>(configuration_, prov_repo, ff_repo, flow);
+ std::vector<std::shared_ptr<core::RepositoryMetricsSource>> repo_metric_sources{prov_repo, ff_repo, content_repo};
+ auto metrics_publisher_store = std::make_unique<minifi::state::MetricsPublisherStore>(configuration_, repo_metric_sources, flow);
metrics_publisher_store_ = metrics_publisher_store.get();
controller_ = std::make_shared<minifi::FlowController>(prov_repo, ff_repo, configuration_, std::move(flow), content_repo, std::move(metrics_publisher_store));
controller_->load(std::move(root));
diff --git a/libminifi/test/integration/IntegrationBase.h b/libminifi/test/integration/IntegrationBase.h
index 84bfedc7e..00631ffdf 100644
--- a/libminifi/test/integration/IntegrationBase.h
+++ b/libminifi/test/integration/IntegrationBase.h
@@ -24,6 +24,8 @@
#include <optional>
#include <string>
#include <utility>
+#include <vector>
+
#include "core/logging/Logger.h"
#include "core/ProcessGroup.h"
#include "core/yaml/YamlConfiguration.h"
@@ -195,7 +197,8 @@ void IntegrationBase::run(const std::optional<std::filesystem::path>& test_file_
running = true;
};
- auto metrics_publisher_store = std::make_unique<minifi::state::MetricsPublisherStore>(configuration, test_repo, test_flow_repo, flow_config);
+ std::vector<std::shared_ptr<core::RepositoryMetricsSource>> repo_metric_sources{test_repo, test_flow_repo, content_repo};
+ auto metrics_publisher_store = std::make_unique<minifi::state::MetricsPublisherStore>(configuration, repo_metric_sources, flow_config);
flowController_ = std::make_unique<minifi::FlowController>(test_repo, test_flow_repo, configuration,
std::move(flow_config), content_repo, std::move(metrics_publisher_store), filesystem, request_restart);
flowController_->load();
diff --git a/libminifi/test/rocksdb-tests/DBProvenanceRepositoryTests.cpp b/libminifi/test/rocksdb-tests/DBProvenanceRepositoryTests.cpp
index db437fb61..ef061d290 100644
--- a/libminifi/test/rocksdb-tests/DBProvenanceRepositoryTests.cpp
+++ b/libminifi/test/rocksdb-tests/DBProvenanceRepositoryTests.cpp
@@ -52,7 +52,7 @@ void verifyMaxKeyCount(const minifi::provenance::ProvenanceRepository& repo, uin
for (int i = 0; i < 50; ++i) {
std::this_thread::sleep_for(100ms);
- k = std::min(k, repo.getKeyCount());
+ k = std::min(k, repo.getRepositoryEntryCount());
if (k < keyCount) {
break;
}
diff --git a/libminifi/test/rocksdb-tests/RepoTests.cpp b/libminifi/test/rocksdb-tests/RepoTests.cpp
index 5d921fff0..4a187cca6 100644
--- a/libminifi/test/rocksdb-tests/RepoTests.cpp
+++ b/libminifi/test/rocksdb-tests/RepoTests.cpp
@@ -27,6 +27,7 @@
#include "core/RepositoryFactory.h"
#include "FlowFileRecord.h"
#include "FlowFileRepository.h"
+#include "ProvenanceRepository.h"
#include "provenance/Provenance.h"
#include "properties/Configure.h"
#include "../unit/ProvenanceTestHelper.h"
@@ -34,6 +35,9 @@
#include "../Catch.h"
#include "utils/gsl.h"
#include "utils/IntegrationTestUtils.h"
+#include "core/repository/VolatileFlowFileRepository.h"
+#include "core/repository/VolatileProvenanceRepository.h"
+#include "DatabaseContentRepository.h"
using namespace std::literals::chrono_literals;
@@ -515,4 +519,221 @@ TEST_CASE("FlowFileRepository synchronously pushes existing flow files") {
}
}
+TEST_CASE("Test getting flow file repository size properties", "[TestGettingRepositorySize]") {
+ LogTestController::getInstance().setDebug<core::repository::FlowFileRepository>();
+ LogTestController::getInstance().setDebug<minifi::provenance::ProvenanceRepository>();
+ LogTestController::getInstance().setDebug<core::repository::VolatileFlowFileRepository>();
+ LogTestController::getInstance().setDebug<core::repository::VolatileProvenanceRepository>();
+ TestController testController;
+ auto dir = testController.createTempDirectory();
+
+ std::shared_ptr<core::Repository> repository;
+ auto expected_is_full = false;
+ uint64_t expected_max_repo_size = 0;
+ SECTION("FlowFileRepository") {
+ repository = std::make_shared<core::repository::FlowFileRepository>("ff", dir.string(), 0ms, 0, 1ms);
+ }
+
+ SECTION("ProvenanceRepository") {
+ repository = std::make_shared<minifi::provenance::ProvenanceRepository>("ff", dir.string(), 0ms, 0, 1ms);
+ }
+
+ SECTION("VolatileFlowFileRepository") {
+ repository = std::make_shared<core::repository::VolatileFlowFileRepository>("ff", dir.string(), 0ms, 10, 1ms);
+ expected_is_full = true;
+ expected_max_repo_size = 7;
+ }
+
+ SECTION("VolatileProvenanceRepository") {
+ repository = std::make_shared<core::repository::VolatileProvenanceRepository>("ff", dir.string(), 0ms, 10, 1ms);
+ expected_is_full = true;
+ expected_max_repo_size = 7;
+ }
+ auto configuration = std::make_shared<minifi::Configure>();
+ repository->initialize(configuration);
+
+ auto flow_file = std::make_shared<minifi::FlowFileRecord>();
+
+ for (auto i = 0; i < 100; ++i) {
+ flow_file->addAttribute("key" + std::to_string(i), "testattributevalue" + std::to_string(i));
+ }
+
+ auto original_size = repository->getRepositorySize();
+ using org::apache::nifi::minifi::utils::verifyEventHappenedInPollTime;
+ REQUIRE(verifyEventHappenedInPollTime(std::chrono::seconds(5), [&original_size, &repository] {
+ auto old_size = original_size;
+ original_size = repository->getRepositorySize();
+ return old_size == original_size;
+ },
+ std::chrono::milliseconds(50)));
+ REQUIRE(true == flow_file->Persist(repository));
+ auto flow_file_2 = std::make_shared<minifi::FlowFileRecord>();
+ REQUIRE(true == flow_file_2->Persist(repository));
+
+ repository->flush();
+ repository->stop();
+
+ auto new_size = repository->getRepositorySize();
+ REQUIRE(verifyEventHappenedInPollTime(std::chrono::seconds(5), [&new_size, &repository] {
+ auto old_size = new_size;
+ new_size = repository->getRepositorySize();
+ return old_size == new_size;
+ },
+ std::chrono::milliseconds(50)));
+ REQUIRE(new_size > original_size);
+ REQUIRE(expected_is_full == repository->isFull());
+ REQUIRE(expected_max_repo_size == repository->getMaxRepositorySize());
+ REQUIRE(2 == repository->getRepositoryEntryCount());
+}
+
+TEST_CASE("Test getting noop repository size properties", "[TestGettingRepositorySize]") {
+ TestController testController;
+ auto dir = testController.createTempDirectory();
+
+ auto repository = minifi::core::createRepository("NoOpRepository", "ff");
+
+ repository->initialize(std::make_shared<minifi::Configure>());
+
+ auto flow_file = std::make_shared<minifi::FlowFileRecord>();
+
+ flow_file->addAttribute("key", "testattributevalue");
+
+ repository->flush();
+ repository->stop();
+
+ REQUIRE(repository->getRepositorySize() == 0);
+ REQUIRE(!repository->isFull());
+ REQUIRE(repository->getMaxRepositorySize() == 0);
+ REQUIRE(repository->getRepositoryEntryCount() == 0);
+}
+
+TEST_CASE("Test getting content repository size properties", "[TestGettingRepositorySize]") {
+ LogTestController::getInstance().setDebug<core::ContentRepository>();
+ LogTestController::getInstance().setDebug<core::repository::FileSystemRepository>();
+ LogTestController::getInstance().setDebug<core::repository::VolatileContentRepository>();
+ LogTestController::getInstance().setDebug<core::repository::DatabaseContentRepository>();
+ TestController testController;
+ auto dir = testController.createTempDirectory();
+
+ auto repository = std::make_shared<core::repository::FlowFileRepository>("ff", dir.string(), 0ms, 0, 1ms);
+
+ auto content_repo_dir = testController.createTempDirectory();
+ auto configuration = std::make_shared<minifi::Configure>();
+ configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, content_repo_dir.string());
+ std::string content = "content";
+ configuration->set(minifi::Configure::nifi_volatile_repository_options_content_max_bytes, std::to_string(content.size()));
+
+ std::shared_ptr<core::ContentRepository> content_repo;
+ auto expected_is_full = false;
+ uint64_t expected_max_repo_size = 0;
+ SECTION("FileSystemRepository") {
+ content_repo = std::make_shared<core::repository::FileSystemRepository>();
+ }
+
+ SECTION("VolatileContentRepository") {
+ content_repo = std::make_shared<core::repository::VolatileContentRepository>("content");
+ expected_is_full = true;
+ expected_max_repo_size = content.size();
+ }
+
+ SECTION("DatabaseContentRepository") {
+ content_repo = std::make_shared<core::repository::DatabaseContentRepository>();
+ }
+
+ content_repo->initialize(configuration);
+
+ repository->initialize(configuration);
+ repository->loadComponent(content_repo);
+ auto original_content_repo_size = content_repo->getRepositorySize();
+
+ auto flow_file = std::make_shared<minifi::FlowFileRecord>();
+
+ auto content_session = content_repo->createSession();
+ auto claim = content_session->create();
+ auto stream = content_session->write(claim);
+ stream->write(gsl::make_span(content).as_span<const std::byte>());
+ flow_file->setResourceClaim(claim);
+ flow_file->setSize(stream->size());
+ flow_file->setOffset(0);
+
+ stream->close();
+ content_session->commit();
+
+ repository->flush();
+ repository->stop();
+
+ using org::apache::nifi::minifi::utils::verifyEventHappenedInPollTime;
+ REQUIRE(verifyEventHappenedInPollTime(std::chrono::seconds(5), [&original_content_repo_size, &content_repo] {
+ auto new_content_repo_size = content_repo->getRepositorySize();
+ return new_content_repo_size > original_content_repo_size;
+ },
+ std::chrono::milliseconds(50)));
+
+ REQUIRE(expected_is_full == content_repo->isFull());
+ REQUIRE(expected_max_repo_size == content_repo->getMaxRepositorySize());
+ REQUIRE(1 == content_repo->getRepositoryEntryCount());
+}
+
+TEST_CASE("Flow file repositories can be stopped", "[TestRepoIsRunning]") {
+ LogTestController::getInstance().setDebug<core::repository::FlowFileRepository>();
+ LogTestController::getInstance().setDebug<minifi::provenance::ProvenanceRepository>();
+ LogTestController::getInstance().setDebug<core::repository::VolatileFlowFileRepository>();
+ LogTestController::getInstance().setDebug<core::repository::VolatileProvenanceRepository>();
+ TestController testController;
+ auto dir = testController.createTempDirectory();
+
+ std::shared_ptr<core::Repository> repository;
+ SECTION("FlowFileRepository") {
+ repository = std::make_shared<core::repository::FlowFileRepository>("ff", dir.string(), 0ms, 0, 1ms);
+ }
+
+ SECTION("ProvenanceRepository") {
+ repository = std::make_shared<minifi::provenance::ProvenanceRepository>("ff", dir.string(), 0ms, 0, 1ms);
+ }
+
+ SECTION("VolatileFlowFileRepository") {
+ repository = std::make_shared<core::repository::VolatileFlowFileRepository>("ff", dir.string(), 0ms, 10, 1ms);
+ }
+
+ SECTION("VolatileProvenanceRepository") {
+ repository = std::make_shared<core::repository::VolatileProvenanceRepository>("ff", dir.string(), 0ms, 10, 1ms);
+ }
+
+ SECTION("NoOpRepository") {
+ repository = minifi::core::createRepository("NoOpRepository", "ff");
+ }
+
+ repository->initialize(std::make_shared<minifi::Configure>());
+
+ REQUIRE(!repository->isRunning());
+ repository->start();
+ REQUIRE(repository->isRunning());
+ repository->stop();
+ REQUIRE(!repository->isRunning());
+}
+
+TEST_CASE("Content repositories are always running", "[TestRepoIsRunning]") {
+ LogTestController::getInstance().setDebug<core::ContentRepository>();
+ LogTestController::getInstance().setDebug<core::repository::FileSystemRepository>();
+ LogTestController::getInstance().setDebug<core::repository::VolatileContentRepository>();
+ LogTestController::getInstance().setDebug<core::repository::DatabaseContentRepository>();
+ TestController testController;
+ auto dir = testController.createTempDirectory();
+
+ std::shared_ptr<core::ContentRepository> content_repo;
+ SECTION("FileSystemRepository") {
+ content_repo = std::make_shared<core::repository::FileSystemRepository>();
+ }
+
+ SECTION("VolatileContentRepository") {
+ content_repo = std::make_shared<core::repository::VolatileContentRepository>("content");
+ }
+
+ SECTION("DatabaseContentRepository") {
+ content_repo = std::make_shared<core::repository::DatabaseContentRepository>();
+ }
+
+ REQUIRE(content_repo->isRunning());
+}
+
} // namespace
diff --git a/libminifi/test/unit/ControllerSocketMetricsPublisherTest.cpp b/libminifi/test/unit/ControllerSocketMetricsPublisherTest.cpp
index 71a336ac2..8db4e2f00 100644
--- a/libminifi/test/unit/ControllerSocketMetricsPublisherTest.cpp
+++ b/libminifi/test/unit/ControllerSocketMetricsPublisherTest.cpp
@@ -63,7 +63,7 @@ class ControllerSocketMetricsPublisherTestFixture {
public:
ControllerSocketMetricsPublisherTestFixture()
: configuration_(std::make_shared<Configure>()),
- response_node_loader_(std::make_shared<state::response::ResponseNodeLoader>(configuration_, nullptr, nullptr, nullptr)),
+ response_node_loader_(std::make_shared<state::response::ResponseNodeLoader>(configuration_, std::vector<std::shared_ptr<core::RepositoryMetricsSource>>{}, nullptr)),
test_response_node_(std::make_shared<TestQueueMetrics>()),
controller_socket_metrics_publisher_("test_publisher") {
controller_socket_metrics_publisher_.initialize(configuration_, response_node_loader_);
diff --git a/libminifi/test/unit/FileUtilsTests.cpp b/libminifi/test/unit/FileUtilsTests.cpp
index fc87866b7..d537f6f27 100644
--- a/libminifi/test/unit/FileUtilsTests.cpp
+++ b/libminifi/test/unit/FileUtilsTests.cpp
@@ -477,3 +477,42 @@ TEST_CASE("FileUtils::get_relative_path", "[TestGetRelativePath]") {
REQUIRE(*FileUtils::get_relative_path(path, base_path / "") == std::filesystem::path("subdir") / "file.log");
REQUIRE(*FileUtils::get_relative_path(base_path, base_path) == ".");
}
+
+TEST_CASE("FileUtils::path_size", "[TestPathSize]") {
+ auto writeToFile = [](const std::filesystem::path& path) {
+ std::ofstream test_file_stream(path, std::ios::out | std::ios::binary);
+ test_file_stream << "foo\n";
+ test_file_stream.flush();
+ };
+
+ TestController test_controller;
+ REQUIRE(FileUtils::path_size({""}) == 0);
+ REQUIRE(FileUtils::path_size({"/random/non-existent/dir"}) == 0);
+
+ auto dir = test_controller.createTempDirectory();
+ REQUIRE(FileUtils::path_size(dir) == 0);
+
+ auto test_file = dir / "test_file.log";
+ writeToFile(test_file);
+
+ REQUIRE(FileUtils::path_size(test_file) == 4);
+ REQUIRE(FileUtils::path_size(dir) == 4);
+
+ auto subdir = dir / "subdir";
+ REQUIRE(utils::file::create_dir(subdir) == 0);
+
+ REQUIRE(FileUtils::path_size(dir) == 4);
+
+ auto subdir_test_file = subdir / "test_file2.log";
+ writeToFile(subdir_test_file);
+
+ REQUIRE(FileUtils::path_size(dir) == 8);
+
+ auto subsubdir = subdir / "subsubdir";
+ REQUIRE(utils::file::create_dir(subsubdir) == 0);
+
+ auto subsubdir_test_file = subsubdir / "test_file3.log";
+ writeToFile(subsubdir_test_file);
+
+ REQUIRE(FileUtils::path_size(dir) == 12);
+}
diff --git a/libminifi/test/unit/MetricsTests.cpp b/libminifi/test/unit/MetricsTests.cpp
index 972365fd4..af6d9e963 100644
--- a/libminifi/test/unit/MetricsTests.cpp
+++ b/libminifi/test/unit/MetricsTests.cpp
@@ -26,16 +26,22 @@
#include "repository/VolatileContentRepository.h"
#include "ProvenanceTestHelper.h"
#include "../DummyProcessor.h"
+#include "range/v3/algorithm/find_if.hpp"
using namespace std::literals::chrono_literals;
namespace org::apache::nifi::minifi::test {
+void checkSerializedValue(const std::vector<org::apache::nifi::minifi::state::response::SerializedResponseNode>& children, const std::string& name, const std::string& expected_value) {
+ auto it = ranges::find_if(children, [&](const auto& child) { return child.name == name; });
+ REQUIRE(it != children.end());
+ REQUIRE(expected_value == it->value.to_string());
+}
+
TEST_CASE("QueueMetricsTestNoConnections", "[c2m2]") {
minifi::state::response::QueueMetrics metrics;
REQUIRE("QueueMetrics" == metrics.getName());
-
REQUIRE(metrics.serialize().empty());
}
@@ -58,40 +64,24 @@ TEST_CASE("QueueMetricsTestConnections", "[c2m3]") {
metrics.updateConnection(connection.get());
- REQUIRE(1 == metrics.serialize().size());
+ auto seialized_metrics = metrics.serialize();
+ REQUIRE(1 == seialized_metrics.size());
minifi::state::response::SerializedResponseNode resp = metrics.serialize().at(0);
REQUIRE("testconnection" == resp.name);
-
REQUIRE(4 == resp.children.size());
- minifi::state::response::SerializedResponseNode datasize = resp.children.at(0);
-
- REQUIRE("datasize" == datasize.name);
- REQUIRE("0" == datasize.value.to_string());
-
- minifi::state::response::SerializedResponseNode datasizemax = resp.children.at(1);
-
- REQUIRE("datasizemax" == datasizemax.name);
- REQUIRE("1024" == datasizemax.value);
-
- minifi::state::response::SerializedResponseNode queued = resp.children.at(2);
-
- REQUIRE("queued" == queued.name);
- REQUIRE("0" == queued.value.to_string());
-
- minifi::state::response::SerializedResponseNode queuedmax = resp.children.at(3);
-
- REQUIRE("queuedmax" == queuedmax.name);
- REQUIRE("1024" == queuedmax.value.to_string());
+ checkSerializedValue(resp.children, "datasize", "0");
+ checkSerializedValue(resp.children, "datasizemax", "1024");
+ checkSerializedValue(resp.children, "queued", "0");
+ checkSerializedValue(resp.children, "queuedmax", "1024");
}
TEST_CASE("RepositorymetricsNoRepo", "[c2m4]") {
minifi::state::response::RepositoryMetrics metrics;
REQUIRE("RepositoryMetrics" == metrics.getName());
-
REQUIRE(metrics.serialize().empty());
}
@@ -109,23 +99,13 @@ TEST_CASE("RepositorymetricsHaveRepo", "[c2m4]") {
minifi::state::response::SerializedResponseNode resp = metrics.serialize().at(0);
REQUIRE("repo_name" == resp.name);
+ REQUIRE(5 == resp.children.size());
- REQUIRE(3 == resp.children.size());
-
- minifi::state::response::SerializedResponseNode running = resp.children.at(0);
-
- REQUIRE("running" == running.name);
- REQUIRE("false" == running.value.to_string());
-
- minifi::state::response::SerializedResponseNode full = resp.children.at(1);
-
- REQUIRE("full" == full.name);
- REQUIRE("false" == full.value);
-
- minifi::state::response::SerializedResponseNode size = resp.children.at(2);
-
- REQUIRE("size" == size.name);
- REQUIRE("0" == size.value);
+ checkSerializedValue(resp.children, "running", "false");
+ checkSerializedValue(resp.children, "full", "false");
+ checkSerializedValue(resp.children, "size", "0");
+ checkSerializedValue(resp.children, "maxSize", "0");
+ checkSerializedValue(resp.children, "entryCount", "0");
}
repo->start();
@@ -135,26 +115,16 @@ TEST_CASE("RepositorymetricsHaveRepo", "[c2m4]") {
minifi::state::response::SerializedResponseNode resp = metrics.serialize().at(0);
REQUIRE("repo_name" == resp.name);
+ REQUIRE(5 == resp.children.size());
- REQUIRE(3 == resp.children.size());
-
- minifi::state::response::SerializedResponseNode running = resp.children.at(0);
-
- REQUIRE("running" == running.name);
- REQUIRE("true" == running.value.to_string());
-
- minifi::state::response::SerializedResponseNode full = resp.children.at(1);
-
- REQUIRE("full" == full.name);
- REQUIRE("false" == full.value);
-
- minifi::state::response::SerializedResponseNode size = resp.children.at(2);
-
- REQUIRE("size" == size.name);
- REQUIRE("0" == size.value);
+ checkSerializedValue(resp.children, "running", "true");
+ checkSerializedValue(resp.children, "full", "false");
+ checkSerializedValue(resp.children, "size", "0");
+ checkSerializedValue(resp.children, "maxSize", "0");
+ checkSerializedValue(resp.children, "entryCount", "0");
}
- repo->setFull();
+ repo->stop();
{
REQUIRE(1 == metrics.serialize().size());
@@ -162,50 +132,54 @@ TEST_CASE("RepositorymetricsHaveRepo", "[c2m4]") {
minifi::state::response::SerializedResponseNode resp = metrics.serialize().at(0);
REQUIRE("repo_name" == resp.name);
+ REQUIRE(5 == resp.children.size());
- REQUIRE(3 == resp.children.size());
-
- minifi::state::response::SerializedResponseNode running = resp.children.at(0);
-
- REQUIRE("running" == running.name);
- REQUIRE("true" == running.value.to_string());
-
- minifi::state::response::SerializedResponseNode full = resp.children.at(1);
-
- REQUIRE("full" == full.name);
- REQUIRE("true" == full.value.to_string());
+ checkSerializedValue(resp.children, "running", "false");
+ checkSerializedValue(resp.children, "full", "false");
+ checkSerializedValue(resp.children, "size", "0");
+ checkSerializedValue(resp.children, "maxSize", "0");
+ checkSerializedValue(resp.children, "entryCount", "0");
+ }
+}
- minifi::state::response::SerializedResponseNode size = resp.children.at(2);
+TEST_CASE("VolatileRepositorymetricsCanBeFull", "[c2m4]") {
+ minifi::state::response::RepositoryMetrics metrics;
- REQUIRE("size" == size.name);
- REQUIRE("0" == size.value.to_string());
- }
+ REQUIRE("RepositoryMetrics" == metrics.getName());
- repo->stop();
+ auto repo = std::make_shared<TestVolatileRepository>();
+ metrics.addRepository(repo);
{
REQUIRE(1 == metrics.serialize().size());
minifi::state::response::SerializedResponseNode resp = metrics.serialize().at(0);
REQUIRE("repo_name" == resp.name);
+ REQUIRE(5 == resp.children.size());
- REQUIRE(3 == resp.children.size());
-
- minifi::state::response::SerializedResponseNode running = resp.children.at(0);
+ checkSerializedValue(resp.children, "running", "false");
+ checkSerializedValue(resp.children, "full", "false");
+ checkSerializedValue(resp.children, "size", "0");
+ checkSerializedValue(resp.children, "maxSize", std::to_string(static_cast<int64_t>(TEST_MAX_REPOSITORY_STORAGE_SIZE * 0.75)));
+ checkSerializedValue(resp.children, "entryCount", "0");
+ }
- REQUIRE("running" == running.name);
- REQUIRE("false" == running.value.to_string());
+ repo->setFull();
- minifi::state::response::SerializedResponseNode full = resp.children.at(1);
+ {
+ REQUIRE(1 == metrics.serialize().size());
- REQUIRE("full" == full.name);
- REQUIRE("true" == full.value);
+ minifi::state::response::SerializedResponseNode resp = metrics.serialize().at(0);
- minifi::state::response::SerializedResponseNode size = resp.children.at(2);
+ REQUIRE("repo_name" == resp.name);
+ REQUIRE(5 == resp.children.size());
- REQUIRE("size" == size.name);
- REQUIRE("0" == size.value);
+ checkSerializedValue(resp.children, "running", "false");
+ checkSerializedValue(resp.children, "full", "true");
+ checkSerializedValue(resp.children, "size", std::to_string(static_cast<int64_t>(TEST_MAX_REPOSITORY_STORAGE_SIZE * 0.75)));
+ checkSerializedValue(resp.children, "maxSize", std::to_string(static_cast<int64_t>(TEST_MAX_REPOSITORY_STORAGE_SIZE * 0.75)));
+ checkSerializedValue(resp.children, "entryCount", "10000");
}
}
diff --git a/libminifi/test/unit/ProvenanceTestHelper.h b/libminifi/test/unit/ProvenanceTestHelper.h
index f2d04ed70..3f5a16c72 100644
--- a/libminifi/test/unit/ProvenanceTestHelper.h
+++ b/libminifi/test/unit/ProvenanceTestHelper.h
@@ -28,6 +28,7 @@
#include <utility>
#include <vector>
#include "core/repository/VolatileContentRepository.h"
+#include "core/repository/VolatileFlowFileRepository.h"
#include "core/Processor.h"
#include "core/ThreadedRepository.h"
#include "Connection.h"
@@ -39,21 +40,19 @@
using namespace std::literals::chrono_literals;
+const int64_t TEST_MAX_REPOSITORY_STORAGE_SIZE = 100;
+
template <typename T_BaseRepository>
class TestRepositoryBase : public T_BaseRepository {
public:
TestRepositoryBase()
- : T_BaseRepository("repo_name", "./dir", 1s, 100, 0ms) {
+ : T_BaseRepository("repo_name", "./dir", 1s, TEST_MAX_REPOSITORY_STORAGE_SIZE, 0ms) {
}
bool initialize(const std::shared_ptr<org::apache::nifi::minifi::Configure> &) override {
return true;
}
- void setFull() {
- T_BaseRepository::repo_full_ = true;
- }
-
bool isNoop() const override {
return false;
}
@@ -125,6 +124,30 @@ class TestRepository : public TestRepositoryBase<org::apache::nifi::minifi::core
bool stop() override {
return true;
}
+
+ uint64_t getRepositorySize() const override {
+ return 0;
+ }
+
+ uint64_t getRepositoryEntryCount() const override {
+ return 0;
+ }
+};
+
+class TestVolatileRepository : public TestRepositoryBase<org::apache::nifi::minifi::core::repository::VolatileFlowFileRepository> {
+ public:
+ bool start() override {
+ return true;
+ }
+
+ bool stop() override {
+ return true;
+ }
+
+ void setFull() {
+ repo_data_.current_size = repo_data_.max_size;
+ repo_data_.current_entry_count = repo_data_.max_count;
+ }
};
class TestThreadedRepository : public TestRepositoryBase<org::apache::nifi::minifi::core::ThreadedRepository> {
@@ -142,13 +165,21 @@ class TestThreadedRepository : public TestRepositoryBase<org::apache::nifi::mini
return thread_;
}
+ uint64_t getRepositorySize() const override {
+ return 0;
+ }
+
+ uint64_t getRepositoryEntryCount() const override {
+ return 0;
+ }
+
std::thread thread_;
};
class TestFlowRepository : public org::apache::nifi::minifi::core::ThreadedRepository {
public:
TestFlowRepository()
- : org::apache::nifi::minifi::core::ThreadedRepository("ff", "./dir", 1s, 100, 0ms) {
+ : org::apache::nifi::minifi::core::ThreadedRepository("ff", "./dir", 1s, TEST_MAX_REPOSITORY_STORAGE_SIZE, 0ms) {
}
bool initialize(const std::shared_ptr<org::apache::nifi::minifi::Configure> &) override {
@@ -178,6 +209,14 @@ class TestFlowRepository : public org::apache::nifi::minifi::core::ThreadedRepos
}
}
+ uint64_t getRepositorySize() const override {
+ return 0;
+ }
+
+ uint64_t getRepositoryEntryCount() const override {
+ return 0;
+ }
+
private:
void run() override {
}
diff --git a/libminifi/test/unit/ResponseNodeLoaderTests.cpp b/libminifi/test/unit/ResponseNodeLoaderTests.cpp
index 1c5d41ea1..f0926523e 100644
--- a/libminifi/test/unit/ResponseNodeLoaderTests.cpp
+++ b/libminifi/test/unit/ResponseNodeLoaderTests.cpp
@@ -36,7 +36,7 @@ class ResponseNodeLoaderTestFixture {
prov_repo_(std::make_shared<TestRepository>()),
ff_repository_(std::make_shared<TestRepository>()),
content_repo_(std::make_shared<minifi::core::repository::VolatileContentRepository>()),
- response_node_loader_(configuration_, prov_repo_, ff_repository_, nullptr) {
+ response_node_loader_(configuration_, {prov_repo_, ff_repository_, content_repo_}, nullptr) {
ff_repository_->initialize(configuration_);
content_repo_->initialize(configuration_);
auto uuid1 = addProcessor<minifi::processors::WriteToFlowFileTestProcessor>("WriteToFlowFileTestProcessor1");
diff --git a/minifi_main/MiNiFiMain.cpp b/minifi_main/MiNiFiMain.cpp
index 29476495c..1c97cb757 100644
--- a/minifi_main/MiNiFiMain.cpp
+++ b/minifi_main/MiNiFiMain.cpp
@@ -371,7 +371,8 @@ int main(int argc, char **argv) {
.path = configure->get(minifi::Configure::nifi_flow_configuration_file),
.filesystem = filesystem}, nifi_configuration_class_name);
- auto metrics_publisher_store = std::make_unique<minifi::state::MetricsPublisherStore>(configure, prov_repo, flow_repo, flow_configuration);
+ std::vector<std::shared_ptr<core::RepositoryMetricsSource>> repo_metric_sources{prov_repo, flow_repo, content_repo};
+ auto metrics_publisher_store = std::make_unique<minifi::state::MetricsPublisherStore>(configure, repo_metric_sources, flow_configuration);
const auto controller = std::make_unique<minifi::FlowController>(
prov_repo, flow_repo, configure, std::move(flow_configuration), content_repo, std::move(metrics_publisher_store), filesystem, request_restart);
[nifi-minifi-cpp] 05/05: MINIFICPP-2110 Fix default CA path for S3 on CentOS
Posted by fg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fgerlits pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 6e0878bb4447fafff7103a81c16c05f0435d0381
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Tue Apr 25 17:02:50 2023 +0200
MINIFICPP-2110 Fix default CA path for S3 on CentOS
Signed-off-by: Ferenc Gerlits <fg...@gmail.com>
This closes #1565
---
extensions/aws/processors/S3Processor.cpp | 6 +++
extensions/coap/tests/CoapC2VerifyHeartbeat.cpp | 2 +-
extensions/coap/tests/CoapIntegrationBase.h | 2 +-
extensions/http-curl/client/HTTPClient.cpp | 22 +----------
extensions/http-curl/client/HTTPClient.h | 1 -
extensions/http-curl/tests/C2NullConfiguration.cpp | 2 +-
extensions/http-curl/tests/C2PauseResumeTest.cpp | 2 +-
.../http-curl/tests/C2VerifyServeResults.cpp | 2 +-
extensions/http-curl/tests/HTTPHandlers.h | 2 +-
extensions/http-curl/tests/HTTPIntegrationBase.h | 2 +-
extensions/http-curl/tests/VerifyInvokeHTTP.h | 2 +-
libminifi/include/utils/HTTPUtils.h | 11 ++++--
libminifi/src/utils/HTTPUtils.cpp | 43 ++++++++++++++++++++++
libminifi/test/integration/IntegrationBase.h | 2 +-
14 files changed, 68 insertions(+), 33 deletions(-)
diff --git a/extensions/aws/processors/S3Processor.cpp b/extensions/aws/processors/S3Processor.cpp
index 695bc9a10..7e8e55579 100644
--- a/extensions/aws/processors/S3Processor.cpp
+++ b/extensions/aws/processors/S3Processor.cpp
@@ -26,6 +26,7 @@
#include "AWSCredentialsService.h"
#include "properties/Properties.h"
#include "utils/StringUtils.h"
+#include "utils/HTTPUtils.h"
namespace org::apache::nifi::minifi::aws::processors {
@@ -123,6 +124,11 @@ void S3Processor::onSchedule(const std::shared_ptr<core::ProcessContext>& contex
} else {
throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Communications Timeout missing or invalid");
}
+
+ static const auto default_ca_path = minifi::utils::getDefaultCAPath();
+ if (default_ca_path) {
+ client_config_->caFile = default_ca_path->string();
+ }
}
std::optional<CommonProperties> S3Processor::getCommonELSupportedProperties(
diff --git a/extensions/coap/tests/CoapC2VerifyHeartbeat.cpp b/extensions/coap/tests/CoapC2VerifyHeartbeat.cpp
index 23007e233..aba295de2 100644
--- a/extensions/coap/tests/CoapC2VerifyHeartbeat.cpp
+++ b/extensions/coap/tests/CoapC2VerifyHeartbeat.cpp
@@ -104,7 +104,7 @@ class VerifyCoAPServer : public CoapIntegrationBase {
std::string scheme;
std::string path;
- parse_http_components(url, port, scheme, path);
+ minifi::utils::parse_http_components(url, port, scheme, path);
uint16_t newport = std::stoi(port) + 2;
auto new_port_str = std::to_string(newport);
diff --git a/extensions/coap/tests/CoapIntegrationBase.h b/extensions/coap/tests/CoapIntegrationBase.h
index e7fbc1e65..5ded576e6 100644
--- a/extensions/coap/tests/CoapIntegrationBase.h
+++ b/extensions/coap/tests/CoapIntegrationBase.h
@@ -94,7 +94,7 @@ class CoapIntegrationBase : public IntegrationBase {
void CoapIntegrationBase::setUrl(std::string url, CivetHandler *handler) {
std::string path;
- parse_http_components(url, port, scheme, path);
+ minifi::utils::parse_http_components(url, port, scheme, path);
CivetCallbacks callback{};
if (url.find("localhost") != std::string::npos) {
if (server != nullptr) {
diff --git a/extensions/http-curl/client/HTTPClient.cpp b/extensions/http-curl/client/HTTPClient.cpp
index 21a1d037e..4d2452269 100644
--- a/extensions/http-curl/client/HTTPClient.cpp
+++ b/extensions/http-curl/client/HTTPClient.cpp
@@ -30,6 +30,7 @@
#include "utils/RegexUtils.h"
#include "range/v3/algorithm/all_of.hpp"
#include "range/v3/action/transform.hpp"
+#include "utils/HTTPUtils.h"
using namespace std::literals::chrono_literals;
@@ -418,7 +419,7 @@ void HTTPClient::configure_secure_connection() {
curl_easy_setopt(http_session_.get(), CURLOPT_CAINFO, nullptr);
curl_easy_setopt(http_session_.get(), CURLOPT_CAPATH, nullptr);
} else {
- static const auto default_ca_path = getDefaultCAPath();
+ static const auto default_ca_path = utils::getDefaultCAPath();
if (default_ca_path)
logger_->log_debug("Using CA certificate file \"%s\"", default_ca_path->string());
@@ -469,25 +470,6 @@ std::string HTTPClient::replaceInvalidCharactersInHttpHeaderFieldName(std::strin
return field_name;
}
-std::optional<std::filesystem::path> HTTPClient::getDefaultCAPath() {
-#ifndef WIN32
- const std::vector<std::filesystem::path> possible_ca_paths = {
- "/etc/ssl/certs/ca-certificates.crt",
- "/etc/pki/tls/certs/ca-bundle.crt",
- "/usr/share/ssl/certs/ca-bundle.crt",
- "/usr/local/share/certs/ca-root-nss.crt",
- "/etc/ssl/cert.pem"
- };
-
- for (const auto& possible_ca_path : possible_ca_paths) {
- if (std::filesystem::exists(possible_ca_path)) {
- return possible_ca_path;
- }
- }
-#endif
- return std::nullopt;
-}
-
void HTTPClient::CurlEasyCleanup::operator()(CURL* curl) const {
curl_easy_cleanup(curl);
}
diff --git a/extensions/http-curl/client/HTTPClient.h b/extensions/http-curl/client/HTTPClient.h
index acc650380..4f90c3b9d 100644
--- a/extensions/http-curl/client/HTTPClient.h
+++ b/extensions/http-curl/client/HTTPClient.h
@@ -188,7 +188,6 @@ class HTTPClient : public utils::BaseHTTPClient, public core::Connectable {
static bool isValidHttpHeaderField(std::string_view field_name);
static std::string replaceInvalidCharactersInHttpHeaderFieldName(std::string field_name);
- static std::optional<std::filesystem::path> getDefaultCAPath();
private:
static int onProgress(void *client, curl_off_t dltotal, curl_off_t dlnow, curl_off_t ultotal, curl_off_t ulnow);
diff --git a/extensions/http-curl/tests/C2NullConfiguration.cpp b/extensions/http-curl/tests/C2NullConfiguration.cpp
index 4bbbdf670..ef02b1bec 100644
--- a/extensions/http-curl/tests/C2NullConfiguration.cpp
+++ b/extensions/http-curl/tests/C2NullConfiguration.cpp
@@ -75,7 +75,7 @@ class VerifyC2Server : public HTTPIntegrationBase {
std::string port;
std::string scheme;
std::string path;
- parse_http_components(url, port, scheme, path);
+ minifi::utils::parse_http_components(url, port, scheme, path);
configuration->set(Configuration::nifi_c2_enable, "true");
configuration->set(Configuration::nifi_c2_agent_class, "test");
configuration->set(Configuration::nifi_c2_agent_protocol_class, "RESTSender");
diff --git a/extensions/http-curl/tests/C2PauseResumeTest.cpp b/extensions/http-curl/tests/C2PauseResumeTest.cpp
index cf8e137a9..0ac86416e 100644
--- a/extensions/http-curl/tests/C2PauseResumeTest.cpp
+++ b/extensions/http-curl/tests/C2PauseResumeTest.cpp
@@ -154,7 +154,7 @@ int main(int argc, char **argv) {
std::string scheme;
std::string path;
std::unique_ptr<TestServer> server;
- parse_http_components(url, port, scheme, path);
+ minifi::utils::parse_http_components(url, port, scheme, path);
server = std::make_unique<TestServer>(port, path, &responder);
harness.setUrl(args.url, &responder);
diff --git a/extensions/http-curl/tests/C2VerifyServeResults.cpp b/extensions/http-curl/tests/C2VerifyServeResults.cpp
index acc0ff0e3..e8262278c 100644
--- a/extensions/http-curl/tests/C2VerifyServeResults.cpp
+++ b/extensions/http-curl/tests/C2VerifyServeResults.cpp
@@ -72,7 +72,7 @@ class VerifyC2Server : public HTTPIntegrationBase {
std::string port;
std::string scheme;
std::string path;
- parse_http_components(url, port, scheme, path);
+ minifi::utils::parse_http_components(url, port, scheme, path);
configuration->set(org::apache::nifi::minifi::Configuration::nifi_c2_enable, "true");
configuration->set(org::apache::nifi::minifi::Configuration::nifi_c2_agent_class, "test");
configuration->set(org::apache::nifi::minifi::Configuration::nifi_c2_agent_heartbeat_reporter_classes, "RESTReceiver");
diff --git a/extensions/http-curl/tests/HTTPHandlers.h b/extensions/http-curl/tests/HTTPHandlers.h
index f95d372bc..aaeb37bee 100644
--- a/extensions/http-curl/tests/HTTPHandlers.h
+++ b/extensions/http-curl/tests/HTTPHandlers.h
@@ -95,7 +95,7 @@ class PeerResponder : public ServerAwareHandler {
explicit PeerResponder(std::string base_url) {
(void)base_url; // unused in release builds
std::string scheme;
- assert(parse_http_components(base_url, port, scheme, path));
+ assert(minifi::utils::parse_http_components(base_url, port, scheme, path));
}
bool handleGet(CivetServer* /*server*/, struct mg_connection *conn) override {
diff --git a/extensions/http-curl/tests/HTTPIntegrationBase.h b/extensions/http-curl/tests/HTTPIntegrationBase.h
index 47dc5c83b..f7c7733a9 100644
--- a/extensions/http-curl/tests/HTTPIntegrationBase.h
+++ b/extensions/http-curl/tests/HTTPIntegrationBase.h
@@ -80,7 +80,7 @@ class HTTPIntegrationBase : public IntegrationBase {
void HTTPIntegrationBase::setUrl(const std::string &url, ServerAwareHandler *handler) {
std::string url_port, url_scheme, url_path;
- parse_http_components(url, url_port, url_scheme, url_path);
+ minifi::utils::parse_http_components(url, url_port, url_scheme, url_path);
if (server) {
if (url_port != "0" && url_port != port) {
throw std::logic_error("Inconsistent port requirements");
diff --git a/extensions/http-curl/tests/VerifyInvokeHTTP.h b/extensions/http-curl/tests/VerifyInvokeHTTP.h
index 2996489cd..32993e9db 100644
--- a/extensions/http-curl/tests/VerifyInvokeHTTP.h
+++ b/extensions/http-curl/tests/VerifyInvokeHTTP.h
@@ -55,7 +55,7 @@ class VerifyInvokeHTTP : public HTTPIntegrationBase {
throw std::logic_error("Url is already set");
}
std::string port, scheme, path;
- parse_http_components(url, port, scheme, path);
+ minifi::utils::parse_http_components(url, port, scheme, path);
path_ = path;
HTTPIntegrationBase::setUrl(url, handler);
}
diff --git a/libminifi/include/utils/HTTPUtils.h b/libminifi/include/utils/HTTPUtils.h
index 708525e98..f748b1d54 100644
--- a/libminifi/include/utils/HTTPUtils.h
+++ b/libminifi/include/utils/HTTPUtils.h
@@ -16,14 +16,17 @@
* limitations under the License.
*/
-#ifndef LIBMINIFI_INCLUDE_UTILS_HTTPUTILS_H_
-#define LIBMINIFI_INCLUDE_UTILS_HTTPUTILS_H_
+#pragma once
#include <string>
+#include <optional>
+#include <filesystem>
#include "io/ClientSocket.h"
#include "utils/RegexUtils.h"
+namespace org::apache::nifi::minifi::utils {
+
/**
This function, unfortunately, assumes that we're parsing http components of a local host. On windows this is problematic
so we convert localhost to our local hostname.
@@ -49,4 +52,6 @@ inline bool parse_http_components(const std::string &url, std::string &port, std
return false;
}
-#endif // LIBMINIFI_INCLUDE_UTILS_HTTPUTILS_H_
+std::optional<std::filesystem::path> getDefaultCAPath();
+
+} // namespace org::apache::nifi::minifi::utils
diff --git a/libminifi/src/utils/HTTPUtils.cpp b/libminifi/src/utils/HTTPUtils.cpp
new file mode 100644
index 000000000..30b84150f
--- /dev/null
+++ b/libminifi/src/utils/HTTPUtils.cpp
@@ -0,0 +1,43 @@
+/**
+ *
+ * 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 "utils/HTTPUtils.h"
+
+#include <vector>
+
+namespace org::apache::nifi::minifi::utils {
+
+std::optional<std::filesystem::path> getDefaultCAPath() {
+#ifndef WIN32
+ const std::vector<std::filesystem::path> possible_ca_paths = {
+ "/etc/ssl/certs/ca-certificates.crt",
+ "/etc/pki/tls/certs/ca-bundle.crt",
+ "/usr/share/ssl/certs/ca-bundle.crt",
+ "/usr/local/share/certs/ca-root-nss.crt",
+ "/etc/ssl/cert.pem"
+ };
+
+ for (const auto& possible_ca_path : possible_ca_paths) {
+ if (std::filesystem::exists(possible_ca_path)) {
+ return possible_ca_path;
+ }
+ }
+#endif
+ return std::nullopt;
+}
+
+} // namespace org::apache::nifi::minifi::utils
diff --git a/libminifi/test/integration/IntegrationBase.h b/libminifi/test/integration/IntegrationBase.h
index 00631ffdf..05cd07b17 100644
--- a/libminifi/test/integration/IntegrationBase.h
+++ b/libminifi/test/integration/IntegrationBase.h
@@ -267,7 +267,7 @@ cmd_args parse_cmdline_args_with_url(int argc, char ** argv) {
#ifdef WIN32
if (url.find("localhost") != std::string::npos) {
std::string port, scheme, path;
- parse_http_components(url, port, scheme, path);
+ minifi::utils::parse_http_components(url, port, scheme, path);
url = scheme + "://" + org::apache::nifi::minifi::io::Socket::getMyHostName() + ":" + port + path;
}
#endif
[nifi-minifi-cpp] 04/05: MINIFICPP-2108 Update version number to 0.15.0
Posted by fg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fgerlits pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 0488c5498833a899f2a4a43c6b6e401014235ebd
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Thu Apr 20 15:51:18 2023 +0200
MINIFICPP-2108 Update version number to 0.15.0
Signed-off-by: Ferenc Gerlits <fg...@gmail.com>
This closes #1563
---
CMakeLists.txt | 2 +-
README.md | 4 ++--
libminifi/CMakeLists.txt | 2 +-
3 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 1bc7f382c..ce9589bfa 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -24,7 +24,7 @@ if (CMAKE_VERSION VERSION_GREATER_EQUAL 3.24)
cmake_policy(SET CMP0135 NEW) # policy to set the timestamps of extracted contents to the time of extraction
endif()
-project(nifi-minifi-cpp VERSION 0.14.0)
+project(nifi-minifi-cpp VERSION 0.15.0)
set(PROJECT_NAME "nifi-minifi-cpp")
# Optional build number for linux distribution targets' tar.gz output
diff --git a/README.md b/README.md
index 33dde804a..ebc3d6713 100644
--- a/README.md
+++ b/README.md
@@ -433,7 +433,7 @@ advanced features.
CPack: Install projects
CPack: - Install directory: ~/Development/code/apache/nifi-minifi-cpp
CPack: Create package
- CPack: - package: ~/Development/code/apache/nifi-minifi-cpp/build/nifi-minifi-cpp-0.12.0.tar.gz generated.
+ CPack: - package: ~/Development/code/apache/nifi-minifi-cpp/build/nifi-minifi-cpp-0.15.0.tar.gz generated.
```
- Create a source assembly located in your build directory with suffix -source.tar.gz
@@ -445,7 +445,7 @@ advanced features.
CPack: Install projects
CPack: - Install directory: ~/Development/code/apache/nifi-minifi-cpp
CPack: Create package
- CPack: - package: ~/Development/code/apache/nifi-minifi-cpp/build/nifi-minifi-cpp-0.12.0-source.tar.gz generated.
+ CPack: - package: ~/Development/code/apache/nifi-minifi-cpp/build/nifi-minifi-cpp-0.15.0-source.tar.gz generated.
```
### Building a docker image
diff --git a/libminifi/CMakeLists.txt b/libminifi/CMakeLists.txt
index 883948c39..31d9c1ce5 100644
--- a/libminifi/CMakeLists.txt
+++ b/libminifi/CMakeLists.txt
@@ -22,7 +22,7 @@ cmake_policy(SET CMP0096 NEW) # policy to preserve the leading zeros in PROJECT_
if (CMAKE_VERSION VERSION_GREATER_EQUAL 3.24)
cmake_policy(SET CMP0135 NEW) # policy to set the timestamps of extracted contents to the time of extraction
endif()
-project(nifi-libcore-minifi VERSION 0.14.0)
+project(nifi-libcore-minifi VERSION 0.15.0)
set(PROJECT_NAME "nifi-libcore-minifi")
if (WIN32)
[nifi-minifi-cpp] 03/05: MINIFICPP-2100 Add RockyLinux Dockerfile
Posted by fg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fgerlits pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit ba60437dbfd1007dd611028da881fe315c365535
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Fri Apr 14 11:23:29 2023 +0200
MINIFICPP-2100 Add RockyLinux Dockerfile
Signed-off-by: Ferenc Gerlits <fg...@gmail.com>
This closes #1558
---
cmake/DockerConfig.cmake | 15 ++++++++++
docker/rockylinux/Dockerfile | 68 ++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 83 insertions(+)
diff --git a/cmake/DockerConfig.cmake b/cmake/DockerConfig.cmake
index df8ea6431..8721cdd47 100644
--- a/cmake/DockerConfig.cmake
+++ b/cmake/DockerConfig.cmake
@@ -122,6 +122,21 @@ add_custom_target(
-c CMAKE_BUILD_TYPE=${CMAKE_BUILD_TYPE}
WORKING_DIRECTORY ${CMAKE_SOURCE_DIR}/docker/)
+add_custom_target(
+ rockylinux
+ COMMAND ${CMAKE_SOURCE_DIR}/docker/DockerBuild.sh
+ -u 1000
+ -g 1000
+ -v ${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}.${PROJECT_VERSION_PATCH}
+ -o ${MINIFI_DOCKER_OPTIONS_STR}
+ -l ${CMAKE_BINARY_DIR}
+ -d rockylinux
+ -c BUILD_NUMBER=${BUILD_NUMBER}
+ -c DOCKER_CCACHE_DUMP_LOCATION=${DOCKER_CCACHE_DUMP_LOCATION}
+ -c DOCKER_SKIP_TESTS=${DOCKER_SKIP_TESTS}
+ -c CMAKE_BUILD_TYPE=${CMAKE_BUILD_TYPE}
+ WORKING_DIRECTORY ${CMAKE_SOURCE_DIR}/docker/)
+
if (EXISTS ${CMAKE_SOURCE_DIR}/docker/test/integration/features)
add_subdirectory(${CMAKE_SOURCE_DIR}/docker/test/integration/features)
diff --git a/docker/rockylinux/Dockerfile b/docker/rockylinux/Dockerfile
new file mode 100644
index 000000000..5ab695323
--- /dev/null
+++ b/docker/rockylinux/Dockerfile
@@ -0,0 +1,68 @@
+# 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.
+#
+FROM rockylinux:8
+LABEL maintainer="Apache NiFi <de...@nifi.apache.org>"
+
+ARG MINIFI_VERSION
+ARG UID=1000
+ARG GID=1000
+
+# MINIFI_OPTIONS will be passed directly to cmake
+# use it to define cmake options (e.g. -DENABLE_AWS=ON -DENABLE_AZURE=ON)
+ARG MINIFI_OPTIONS=""
+ARG CMAKE_BUILD_TYPE=Release
+ARG DOCKER_SKIP_TESTS=ON
+ARG DOCKER_MAKE_TARGET="all"
+
+ENV MINIFI_BASE_DIR /opt/minifi
+ENV MINIFI_HOME $MINIFI_BASE_DIR/nifi-minifi-cpp-$MINIFI_VERSION
+ENV USER minificpp
+
+RUN mkdir -p $MINIFI_BASE_DIR
+COPY . ${MINIFI_BASE_DIR}
+
+# Install the system dependencies needed for a build
+# gpsd-devel and ccache are in EPEL
+RUN dnf -y install epel-release && dnf -y install sudo git which make libarchive ccache ca-certificates cmake && \
+ if echo "$MINIFI_OPTIONS" | grep -q "ENABLE_GPS=ON"; then dnf -y install gpsd-devel; fi && \
+ if echo "$MINIFI_OPTIONS" | grep -q "ENABLE_JNI=ON"; then dnf -y install java-1.8.0-openjdk maven; fi && \
+ if echo "$MINIFI_OPTIONS" | grep -q "ENABLE_PCAP=ON"; then dnf -y install libpcap-devel; fi && \
+ if echo "$MINIFI_OPTIONS" | grep -q "ENABLE_USB_CAMERA=ON"; then dnf -y install libpng-devel libusbx-devel; fi && \
+ if echo "$MINIFI_OPTIONS" | grep -q "ENABLE_OPENCV=ON" || echo "$MINIFI_OPTIONS" | grep -q "ENABLE_BUSTACHE=ON"; then dnf -y install boost-devel; fi && \
+ if echo "$MINIFI_OPTIONS" | grep -q "ENABLE_LUA_SCRIPTING=ON"; then dnf -y --enablerepo=powertools install lua-devel; fi && \
+ if echo "$MINIFI_OPTIONS" | grep -q "ENABLE_PYTHON_SCRIPTING=ON"; then dnf -y install python3-devel; fi && \
+ if echo "$MINIFI_OPTIONS" | grep -q "ENABLE_SFTP=ON" && [ "${DOCKER_SKIP_TESTS}" == "OFF" ]; then dnf -y install java-1.8.0-openjdk maven; fi
+
+RUN cd $MINIFI_BASE_DIR && \
+ ./bootstrap.sh -t && \
+ ln -s /usr/bin/ccache /usr/lib64/ccache/c++
+
+# Setup minificpp user
+RUN groupadd -g ${GID} ${USER} && useradd -g ${GID} ${USER} && \
+ chown -R ${USER}:${USER} ${MINIFI_BASE_DIR}
+
+USER ${USER}
+
+# Perform the build
+RUN cd $MINIFI_BASE_DIR && \
+ cd build && \
+ source /opt/rh/gcc-toolset-12/enable && \
+ export PATH=/usr/lib64/ccache${PATH:+:${PATH}} && \
+ export CCACHE_DIR=${MINIFI_BASE_DIR}/.ccache && \
+ cmake3 -DSTATIC_BUILD= -DSKIP_TESTS=${DOCKER_SKIP_TESTS} ${MINIFI_OPTIONS} -DAWS_ENABLE_UNITY_BUILD=OFF -DCMAKE_BUILD_TYPE="${CMAKE_BUILD_TYPE}" .. && \
+ make -j "$(nproc)" ${DOCKER_MAKE_TARGET}
[nifi-minifi-cpp] 02/05: MINIFICPP-2099 Only run tests requiring test processors if they are available
Posted by fg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fgerlits pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 5be21834c1dcd7901f745a758466bd0e3ea5b5a4
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Thu Apr 13 18:16:49 2023 +0200
MINIFICPP-2099 Only run tests requiring test processors if they are available
Signed-off-by: Ferenc Gerlits <fg...@gmail.com>
This closes #1557
---
cmake/DockerConfig.cmake | 12 ++++++------
docker/DockerVerify.sh | 16 +++++++++++++++-
docker/test/integration/environment.py | 6 +++++-
.../test/integration/features/core_functionality.feature | 3 +--
4 files changed, 27 insertions(+), 10 deletions(-)
diff --git a/cmake/DockerConfig.cmake b/cmake/DockerConfig.cmake
index 7740d2faf..df8ea6431 100644
--- a/cmake/DockerConfig.cmake
+++ b/cmake/DockerConfig.cmake
@@ -127,27 +127,27 @@ if (EXISTS ${CMAKE_SOURCE_DIR}/docker/test/integration/features)
add_custom_target(
docker-verify-all
- COMMAND ${CMAKE_SOURCE_DIR}/docker/DockerVerify.sh ${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}.${PROJECT_VERSION_PATCH} ${ALL_BEHAVE_TESTS})
+ COMMAND ${CMAKE_SOURCE_DIR}/docker/DockerVerify.sh --enable_test_processors ON ${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}.${PROJECT_VERSION_PATCH} ${ALL_BEHAVE_TESTS})
add_custom_target(
docker-verify
- COMMAND ${CMAKE_SOURCE_DIR}/docker/DockerVerify.sh ${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}.${PROJECT_VERSION_PATCH} ${ENABLED_BEHAVE_TESTS})
+ COMMAND ${CMAKE_SOURCE_DIR}/docker/DockerVerify.sh --enable_test_processors ${ENABLE_TEST_PROCESSORS} ${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}.${PROJECT_VERSION_PATCH} ${ENABLED_BEHAVE_TESTS})
add_custom_target(
docker-verify-q1
- COMMAND ${CMAKE_SOURCE_DIR}/docker/DockerVerify.sh ${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}.${PROJECT_VERSION_PATCH} ${ENABLED_BEHAVE_TESTS_FIRST_QUADRANT})
+ COMMAND ${CMAKE_SOURCE_DIR}/docker/DockerVerify.sh --enable_test_processors ${ENABLE_TEST_PROCESSORS} ${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}.${PROJECT_VERSION_PATCH} ${ENABLED_BEHAVE_TESTS_FIRST_QUADRANT})
add_custom_target(
docker-verify-q2
- COMMAND ${CMAKE_SOURCE_DIR}/docker/DockerVerify.sh ${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}.${PROJECT_VERSION_PATCH} ${ENABLED_BEHAVE_TESTS_SECOND_QUADRANT})
+ COMMAND ${CMAKE_SOURCE_DIR}/docker/DockerVerify.sh --enable_test_processors ${ENABLE_TEST_PROCESSORS} ${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}.${PROJECT_VERSION_PATCH} ${ENABLED_BEHAVE_TESTS_SECOND_QUADRANT})
add_custom_target(
docker-verify-q3
- COMMAND ${CMAKE_SOURCE_DIR}/docker/DockerVerify.sh ${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}.${PROJECT_VERSION_PATCH} ${ENABLED_BEHAVE_TESTS_THIRD_QUADRANT})
+ COMMAND ${CMAKE_SOURCE_DIR}/docker/DockerVerify.sh --enable_test_processors ${ENABLE_TEST_PROCESSORS} ${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}.${PROJECT_VERSION_PATCH} ${ENABLED_BEHAVE_TESTS_THIRD_QUADRANT})
add_custom_target(
docker-verify-q4
- COMMAND ${CMAKE_SOURCE_DIR}/docker/DockerVerify.sh ${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}.${PROJECT_VERSION_PATCH} ${ENABLED_BEHAVE_TESTS_LAST_QUADRANT})
+ COMMAND ${CMAKE_SOURCE_DIR}/docker/DockerVerify.sh --enable_test_processors ${ENABLE_TEST_PROCESSORS} ${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}.${PROJECT_VERSION_PATCH} ${ENABLED_BEHAVE_TESTS_LAST_QUADRANT})
endif()
include(VerifyPythonCompatibility)
diff --git a/docker/DockerVerify.sh b/docker/DockerVerify.sh
index c1cbd593c..49ea8fd6c 100755
--- a/docker/DockerVerify.sh
+++ b/docker/DockerVerify.sh
@@ -28,12 +28,13 @@ die()
_positionals=()
_arg_feature_path=('' )
_arg_image_tag_prefix=
+_enable_test_processors=OFF
print_help()
{
printf '%s\n' "Runs the provided behave tests in a containerized environment"
- printf 'Usage: %s [--image-tag-prefix <arg>] [-h|--help] <minifi_version> <feature_path-1> [<feature_path-2>] ... [<feature_path-n>] ...\n' "$0"
+ printf 'Usage: %s [--image-tag-prefix <arg>] [--enable_test_processors <ON|OFF>] [-h|--help] <minifi_version> <feature_path-1> [<feature_path-2>] ... [<feature_path-n>] ...\n' "$0"
printf '\t%s\n' "<minifi_version>: the version of minifi"
printf '\t%s\n' "<feature_path>: feature files to run"
printf '\t%s\n' "--image-tag-prefix: optional prefix to the docker tag (no default)"
@@ -56,6 +57,14 @@ parse_commandline()
--image-tag-prefix=*)
_arg_image_tag_prefix="${_key##--image-tag-prefix=}"
;;
+ --enable_test_processors)
+ test $# -lt 2 && die "Missing value for the optional argument '$_key'." 1
+ _enable_test_processors="$2"
+ shift
+ ;;
+ --enable_test_processors=*)
+ _enable_test_processors="${_key##--enable_test_processors=}"
+ ;;
-h|--help)
print_help
exit 0
@@ -157,6 +166,11 @@ export TEST_DIRECTORY
# Add --no-logcapture to see logs interleaved with the test output
BEHAVE_OPTS=(-f pretty --logging-level INFO --logging-clear-handlers)
+if [[ "${_enable_test_processors}" == "ON" ]]; then
+ BEHAVE_OPTS+=(--define "test_processors=ON")
+else
+ BEHAVE_OPTS+=(--define "test_processors=OFF")
+fi
# Specify feature or scenario to run a specific test e.g.:
# behave "${BEHAVE_OPTS[@]}" "features/file_system_operations.feature"
diff --git a/docker/test/integration/environment.py b/docker/test/integration/environment.py
index 8f245afd2..3261b71c6 100644
--- a/docker/test/integration/environment.py
+++ b/docker/test/integration/environment.py
@@ -33,12 +33,16 @@ def before_scenario(context, scenario):
scenario.skip("Marked with @skip")
return
+ if "requires_test_processors" in scenario.effective_tags and context.config.userdata.get('test_processors') != "ON":
+ scenario.skip("Test processors are not available, skipping test scenario")
+ return
+
logging.info("Integration test setup at {time:%H:%M:%S.%f}".format(time=datetime.datetime.now()))
context.test = MiNiFi_integration_test(context)
def after_scenario(context, scenario):
- if "skip" in scenario.effective_tags:
+ if scenario.status == "skipped":
logging.info("Scenario was skipped, no need for clean up.")
return
diff --git a/docker/test/integration/features/core_functionality.feature b/docker/test/integration/features/core_functionality.feature
index 8802c8355..b6682d5ed 100644
--- a/docker/test/integration/features/core_functionality.feature
+++ b/docker/test/integration/features/core_functionality.feature
@@ -38,14 +38,13 @@ Feature: Core flow functionalities
Then at least one flowfile with the content "first_custom_text" is placed in the monitored directory in less than 20 seconds
And at least one flowfile with the content "second_custom_text" is placed in the monitored directory in less than 20 seconds
-
Scenario: The default configuration uses RocksDB for both the flow file and content repositories
Given a GenerateFlowFile processor with the "Data Format" property set to "Text"
When the MiNiFi instance starts up
Then the Minifi logs contain the following message: "Using plaintext FlowFileRepository" in less than 5 seconds
And the Minifi logs contain the following message: "Using plaintext DatabaseContentRepository" in less than 1 second
-
+ @requires_test_processors
Scenario: Processors are destructed when agent is stopped
Given a transient MiNiFi flow with the name "transient-minifi" is set up
And a LogOnDestructionProcessor processor with the name "logOnDestruction" in the "transient-minifi" flow