MINIFICPP-2849 Implement LMDB based content repository

.github/workflows/ci.yml

@@ -11,7 +11,7 @@ concurrency:
 env:
   DOCKER_CMAKE_FLAGS: -DDOCKER_VERIFY_THREAD=3 -DUSE_SHARED_LIBS= -DSTRICT_GSL_CHECKS=AUDIT -DCI_BUILD=ON -DENABLE_AWS=ON -DENABLE_KAFKA=ON -DENABLE_MQTT=ON -DENABLE_AZURE=ON -DENABLE_SQL=ON \
     -DENABLE_SPLUNK=ON -DENABLE_GCP=ON -DENABLE_OPC=ON -DENABLE_PYTHON_SCRIPTING=ON -DENABLE_LUA_SCRIPTING=ON -DENABLE_KUBERNETES=ON -DENABLE_TEST_PROCESSORS=ON -DENABLE_PROMETHEUS=ON \
-    -DENABLE_ELASTICSEARCH=ON -DENABLE_GRAFANA_LOKI=ON -DENABLE_COUCHBASE=ON -DENABLE_LLAMACPP=ON -DDOCKER_BUILD_ONLY=ON -DMINIFI_PERFORMANCE_TESTS=ON -DMINIFI_RUST=OFF
+    -DENABLE_ELASTICSEARCH=ON -DENABLE_GRAFANA_LOKI=ON -DENABLE_COUCHBASE=ON -DENABLE_LLAMACPP=ON -DENABLE_LMDB=ON -DDOCKER_BUILD_ONLY=ON -DMINIFI_PERFORMANCE_TESTS=ON -DMINIFI_RUST=OFF
   CCACHE_DIR: ${{ GITHUB.WORKSPACE }}/.ccache
   CONAN_LOGIN_USERNAME_NIFI_CONAN: ${{ secrets.CONAN_USERNAME }}
   CONAN_PASSWORD_NIFI_CONAN: ${{ secrets.CONAN_ACCESS_TOKEN }}
@@ -43,6 +43,7 @@ jobs:
         -DENABLE_LIBARCHIVE=ON
         -DENABLE_LLAMACPP=ON
         -DENABLE_KAFKA=ON
+        -DENABLE_LMDB=ON
         -DENABLE_LUA_SCRIPTING=ON
         -DENABLE_LZMA=ON
         -DENABLE_MQTT=ON
@@ -158,6 +159,7 @@ jobs:
         -DENABLE_LIBARCHIVE=ON
         -DENABLE_KAFKA=ON
         -DENABLE_LLAMACPP=ON
+        -DENABLE_LMDB=ON
         -DENABLE_LUA_SCRIPTING=ON
         -DENABLE_LZMA=ON
         -DENABLE_MQTT=ON
@@ -271,10 +273,11 @@ jobs:
         -DENABLE_EXECUTE_PROCESS=ON
         -DENABLE_GCP=ON
         -DENABLE_GRAFANA_LOKI=ON
+        -DENABLE_KAFKA=ON
         -DENABLE_KUBERNETES=ON
         -DENABLE_LIBARCHIVE=ON
         -DENABLE_LLAMACPP=ON
-        -DENABLE_KAFKA=ON
+        -DENABLE_LMDB=ON
         -DENABLE_LUA_SCRIPTING=ON
         -DENABLE_LZMA=ON
         -DENABLE_MQTT=ON

CMakeLists.txt

@@ -744,15 +744,16 @@ endif()
 
 cpack_add_component_group(extensions DISPLAY_NAME "Extensions" EXPANDED)
 set(EXTENSIONS_ENABLED_BY_DEFAULT (
+    minifi-archive-extensions
     minifi-aws
     minifi-azure
     minifi-civet-extensions
     minifi-elasticsearch
     minifi-gcp
     minifi-grafana-loki
-    minifi-archive-extensions
-    minifi-mqtt-extensions
     minifi-kafka
+    minifi-lmdb
+    minifi-mqtt-extensions
     minifi-pdh
     minifi-prometheus
     minifi-rocksdb-repos

CONFIGURE.md

@@ -625,7 +625,7 @@ The Flow File Repository can be configured with the `nifi.flowfile.repository.cl
     # in minifi.properties
     nifi.flowfile.repository.class.name=NoOpRepository  # VolatileFlowFileRepository can also be used which is an alias for NoOpRepository
 
-The Content Repository can be configured with the `nifi.content.repository.class.name` property. If not specified, it uses the `DatabaseContentRepository` class by default, which persists the content in a RocksDB database. `DatabaseContentRepository` is also the default value specified in the minifi.properties file. Alternatively it can be configured to use a `VolatileContentRepository` that keeps the state in memory (so the state gets lost upon restart), or the `FileSystemRepository` to keep the state in regular files.
+The Content Repository can be configured with the `nifi.content.repository.class.name` property. If not specified, it uses the `DatabaseContentRepository` class by default, which persists the content in a RocksDB database. `DatabaseContentRepository` is also the default value specified in the minifi.properties file. Alternatively it can be configured to use a `VolatileContentRepository` that keeps the state in memory (so the state gets lost upon restart), the `FileSystemRepository` to keep the state in regular files, or `LmdbContentRepository` that uses LMDB database as an alternative to RocksDB.
 
 **NOTE:** RocksDB database has a limit of 4GB for the size of a database object. Due to this if you expect to process larger flow files than 4GB you should use the `FileSystemRepository`. The downside of using `FileSystemRepository` is that it does not have the transactional guarantees of the RocksDB repository implementation.
 
@@ -745,6 +745,18 @@ RocksDB options can also be overridden for a specific repository using the `nifi
     nifi.provenance.repository.rocksdb.options.use_direct_reads=false
     nifi.state.storage.rocksdb.options.use_direct_io_for_flush_and_compaction=false
 
+### Configuring LMDB content repository
+
+There is an alternative content repository that can be used: the LMDB database. When `LmdbContentRepository` is set, LMDB is used for storing content, which is a memory-mapped, fast (especially for reads), low-footprint, key-value database. It can be a good alternative on memory-limited edge devices, but each use case should be evaluated separately. The caveats of using this database are the following:
+
+- Single writer only — concurrent writes are serialized, which can be a bottleneck for write-heavy workloads
+- Database file never shrinks automatically — occasional large flow files could keep the database size permanently high on disk and in the reserved virtual address space
+- No compression - stores everything uncompressed on disk
+- The maximum database size must be set upfront, which is done in the minifi.properties file with the following property, with the current default value set to 10 GB:
+
+    # in minifi.properties
+    nifi.content.repository.lmdb.max.db.size=10 GB
+
 #### Shared database
 
 It is also possible to use a single database to store multiple repositories with the `minifidb://` scheme.

bootstrap/cli.py

@@ -63,7 +63,7 @@ def run_conan_install(minifi_options: MinifiOptions, package_manager: PackageMan
         return True
     conan_options = add_conan_options_from_cmake_options(["ENABLE_ALL", "ENABLE_LIBARCHIVE", "ENABLE_ROCKSDB", "ENABLE_SFTP", "ENABLE_PROMETHEUS", "ENABLE_BZIP2", "ENABLE_LZMA", "ENABLE_MQTT",
                                                           "ENABLE_COUCHBASE", "ENABLE_KAFKA", "ENABLE_OPC", "ENABLE_GCP", "ENABLE_GRPC_FOR_LOKI", "ENABLE_BUSTACHE", "ENABLE_KUBERNETES",
-                                                          "ENABLE_AZURE", "ENABLE_LLAMACPP", "ENABLE_AWS", "PORTABLE", "SKIP_TESTS"],
+                                                          "ENABLE_AZURE", "ENABLE_LLAMACPP", "ENABLE_AWS", "ENABLE_LMDB", "PORTABLE", "SKIP_TESTS"],
                                                          minifi_options)
     if minifi_options.custom_malloc is not None and minifi_options.custom_malloc.value not in (None, "OFF"):
         conan_options += f' -o "&:custom_malloc={minifi_options.custom_malloc.value}"'

cmake/GetLMDB.cmake

@@ -0,0 +1,24 @@
+# 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.
+
+if(MINIFI_LMDB_SOURCE STREQUAL "CONAN")
+    message("Using Conan to install LMDB")
+    find_package(lmdb REQUIRED)
+elseif(MINIFI_LMDB_SOURCE STREQUAL "BUILD")
+    message("Using CMake to build LMDB from source")
+    include(LMDB)
+endif()

cmake/LMDB.cmake

@@ -0,0 +1,46 @@
+# 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(FetchContent)
+
+set(PATCH_FILE_1 "${CMAKE_SOURCE_DIR}/thirdparty/lmdb/all/patches/add-cmake-file.patch")
+if (WIN32)
+    set(PATCH_FILE_2 "${CMAKE_SOURCE_DIR}/thirdparty/lmdb/all/patches/fix-windows-symbols.patch")
+    set(PC ${Bash_EXECUTABLE}  -c "set -x &&\
+        (\\\"${Patch_EXECUTABLE}\\\" -p0 -R -s -f --dry-run -i \\\"${PATCH_FILE_1}\\\" || \\\"${Patch_EXECUTABLE}\\\" -p0 -N -i \\\"${PATCH_FILE_1}\\\") &&\
+        (\\\"${Patch_EXECUTABLE}\\\" -p0 -R -s -f --dry-run -i \\\"${PATCH_FILE_2}\\\" || \\\"${Patch_EXECUTABLE}\\\" -p0 -N -i \\\"${PATCH_FILE_2}\\\")")
+else()
+    set(PC ${Bash_EXECUTABLE}  -c "set -x &&\
+        (\\\"${Patch_EXECUTABLE}\\\" -p0 -R -s -f --dry-run -i \\\"${PATCH_FILE_1}\\\" || \\\"${Patch_EXECUTABLE}\\\" -p0 -N -i \\\"${PATCH_FILE_1}\\\")")
+endif()
+
+FetchContent_Declare(
+        lmdb
+        URL      https://github.com/LMDB/lmdb/archive/refs/tags/LMDB_1.0.1.tar.gz
+        URL_HASH SHA256=7ce1db4b8c13f60e0881f12537340c73fb5e0125dba4daa6649a2314341855db
+        PATCH_COMMAND "${PC}"
+        SOURCE_SUBDIR "libraries/liblmdb"
+        SYSTEM
+)
+
+FetchContent_MakeAvailable(lmdb)
+
+set(LMDB_INCLUDE_DIR "${lmdb_SOURCE_DIR}/libraries/liblmdb")
+
+if (NOT TARGET lmdb::lmdb)
+    add_library(lmdb::lmdb ALIAS lmdb)
+endif()

cmake/MiNiFiOptions.cmake

@@ -90,6 +90,7 @@ endif()
 add_minifi_option(ENABLE_ALL "Enables all extensions" OFF)
 add_minifi_option(ENABLE_CIVET "Enables CivetWeb components." ON)
 add_minifi_option(ENABLE_ROCKSDB "Enables the RocksDB extension." ON)
+add_minifi_option(ENABLE_LMDB "Enables the LMDB extension." ON)
 add_minifi_option(ENABLE_LIBARCHIVE "Enables the lib archive extensions." ON)
 add_minifi_option(ENABLE_LZMA "Enables the liblzma build" ON)
 add_minifi_option(ENABLE_BZIP2 "Enables the bzip2 build" ON)
@@ -189,6 +190,7 @@ add_minifi_multi_option(MINIFI_LLAMACPP_SOURCE "Retrieves llama.cpp from provide
 add_minifi_multi_option(MINIFI_DATE_SOURCE "Retrieves date from provided source" "BUILD;CONAN" "${MINIFI_DEFAULT_DEPENDENCY_SOURCE}")
 add_minifi_multi_option(MINIFI_WINFLEXBISON_SOURCE "Retrieves winflexbison from provided source" "BUILD;CONAN" "${MINIFI_DEFAULT_DEPENDENCY_SOURCE}")
 add_minifi_multi_option(MINIFI_AWS_SDK_CPP_SOURCE "Retrieves AWS SDK for C++ from provided source" "BUILD;CONAN" "${MINIFI_DEFAULT_DEPENDENCY_SOURCE}")
+add_minifi_multi_option(MINIFI_LMDB_SOURCE "Retrieves LMDB from provided source" "BUILD;CONAN" "${MINIFI_DEFAULT_DEPENDENCY_SOURCE}")
 
 # Docker options
 

conanfile.py

@@ -42,12 +42,12 @@ class MiNiFiCppMain(ConanFile):
                "enable_rocksdb": [True, False], "enable_sftp": [True, False], "enable_prometheus": [True, False], "enable_bzip2": [True, False], "enable_lzma": [True, False],
                "enable_mqtt": [True, False], "enable_couchbase": [True, False], "enable_kafka": [True, False], "enable_opc": [True, False], "enable_gcp": [True, False],
                "enable_grpc_for_loki": [True, False], "enable_bustache": [True, False], "enable_kubernetes": [True, False], "enable_azure": [True, False], "enable_llamacpp": [True, False],
-               "enable_aws": [True, False], "skip_tests": [True, False], "portable": [True, False]}
+               "enable_aws": [True, False], "enable_lmdb": [True, False], "skip_tests": [True, False], "portable": [True, False]}
 
     default_options = {"shared": False, "fPIC": True, "custom_malloc": False, "enable_all": False, "enable_libarchive": False, "enable_rocksdb": False, "enable_sftp": False,
                        "enable_prometheus": False, "enable_bzip2": False, "enable_lzma": False, "enable_mqtt": False, "enable_couchbase": False, "enable_kafka": False, "enable_opc": False,
                        "enable_gcp": False, "enable_grpc_for_loki": False, "enable_bustache": False, "enable_kubernetes": False, "enable_azure": False, "enable_llamacpp": False, "enable_aws": False,
-                       "skip_tests": False, "portable": True}
+                       "enable_lmdb": False, "skip_tests": False, "portable": True}
 
     exports_sources = shared_sources
 
@@ -104,6 +104,8 @@ def requirements(self):
             self.requires("llama-cpp/b8944@minifi/develop")
         if self.options.enable_all or self.options.enable_aws:
             self.requires("aws-sdk-cpp/1.11.807@minifi/develop")
+        if self.options.enable_all or self.options.enable_lmdb:
+            self.requires("lmdb/1.0.1@minifi/develop")
 
         if self.options.custom_malloc == "jemalloc":
             self.requires("jemalloc/5.3.1")

conf/minifi.properties.in

@@ -54,6 +54,9 @@ nifi.content.repository.class.name=DatabaseContentRepository
 # nifi.database.content.repository.rocksdb.compaction.period=2 min
 # nifi.database.content.repository.optimize.for.small.db.cache.size=8 MB
 
+## Relates to the internal workings of the LMDB backend
+# nifi.content.repository.lmdb.max.db.size=10 GB
+
 # setting this value to "0" enables synchronous deletion
 # nifi.database.content.repository.purge.period = 1 sec
 

core-framework/src/core/BufferedContentSession.cpp

@@ -69,7 +69,7 @@ std::shared_ptr<io::BaseStream> BufferedContentSession::read(const std::shared_p
 
 void BufferedContentSession::commit() {
   for (const auto& resource : managed_resources_) {
-    auto outStream = repository_->write(*resource.first);
+    auto outStream = repository_->write(*resource.first, false);
     if (outStream == nullptr) {
       throw Exception(REPOSITORY_EXCEPTION, "Couldn't open the underlying resource for write: " + resource.first->getContentFullPath());
     }

extensions/lmdb/CMakeLists.txt

@@ -0,0 +1,37 @@
+#
+# 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.
+#
+
+if (NOT (ENABLE_ALL OR ENABLE_LMDB))
+    return()
+endif()
+
+include(GetLMDB)
+
+include(${CMAKE_SOURCE_DIR}/extensions/ExtensionHeader.txt)
+
+file(GLOB SOURCES  "*.cpp")
+
+add_minifi_library(minifi-lmdb SHARED ${SOURCES})
+
+target_include_directories(minifi-lmdb PUBLIC ${LMDB_INCLUDE_DIR})
+target_link_libraries(minifi-lmdb PUBLIC lmdb::lmdb)
+target_link_libraries(minifi-lmdb PUBLIC minifi-api minifi-extension-framework)
+target_link_libraries(minifi-lmdb PRIVATE $<LINK_ONLY:core-minifi>)
+
+register_extension(minifi-lmdb "LMDB" LMDB "This enables persistent repositories using LMDB" "extensions/lmdb/tests")