Add Document Type property

Author: lordgamez

PROCESSORS.md

@@ -2218,6 +2218,7 @@ In the list below, the names of required properties appear in bold. Any other pr
 | **Bucket Name**                          | default       |                                                    | The name of bucket to access.<br/>**Supports Expression Language: true**                                                                                                                     |
 | Scope Name                               |               |                                                    | Scope to use inside the bucket. If not specified, the _default scope is used.<br/>**Supports Expression Language: true**                                                                     |
 | Collection Name                          |               |                                                    | Collection to use inside the bucket scope. If not specified, the _default collection is used.<br/>**Supports Expression Language: true**                                                     |
+| **Document Type**                        | Json          | Json<br/>Binary<br/>String                         | Content type to store data as.                                                                                                                                                               |
 | Document Id                              |               |                                                    | A static, fixed Couchbase document id, or an expression to construct the Couchbase document id. If not specified, the FlowFile UUID will be used.<br/>**Supports Expression Language: true** |
 | **Persist To**                           | NONE          | NONE<br/>ACTIVE<br/>ONE<br/>TWO<br/>THREE<br/>FOUR | Durability constraint about disk persistence.                                                                                                                                                |
 | **Replicate To**                         | NONE          | NONE<br/>ONE<br/>TWO<br/>THREE                     | Durability constraint about replication.                                                                                                                                                     |

docker/requirements.txt

@@ -12,3 +12,4 @@ azure-storage-blob==12.13.0
 prometheus-api-client==0.5.0
 humanfriendly==10.0
 requests<2.29  # https://github.com/docker/docker-py/issues/3113
+couchbase==4.3.2

docker/test/integration/cluster/DockerTestCluster.py

@@ -35,6 +35,7 @@
 from .checkers.SplunkChecker import SplunkChecker
 from .checkers.GrafanaLokiChecker import GrafanaLokiChecker
 from .checkers.ModbusChecker import ModbusChecker
+from .checkers.CouchbaseChecker import CouchbaseChecker
 from utils import get_peak_memory_usage, get_minifi_pid, get_memory_usage, retry_check
 
 
@@ -54,6 +55,7 @@ def __init__(self, context, feature_id):
         self.grafana_loki_checker = GrafanaLokiChecker()
         self.minifi_controller_executor = MinifiControllerExecutor(self.container_communicator)
         self.modbus_checker = ModbusChecker(self.container_communicator)
+        self.couchbase_checker = CouchbaseChecker()
 
     def cleanup(self):
         self.container_store.cleanup()
@@ -424,3 +426,6 @@ def set_value_on_plc_with_modbus(self, container_name, modbus_cmd):
 
     def enable_ssl_in_nifi(self):
         self.container_store.enable_ssl_in_nifi()
+
+    def is_data_present_in_couchbase(self, doc_id: str, bucket_name: str, expected_data: str, expected_data_type: str):
+        return self.couchbase_checker.is_data_present_in_couchbase(doc_id, bucket_name, expected_data, expected_data_type)

docker/test/integration/cluster/containers/CouchbaseServerContainer.py

@@ -51,4 +51,5 @@ def deploy(self):
             detach=True,
             name=self.name,
             network=self.network.name,
+            ports={'11210/tcp': 11210},
             entrypoint=self.command)

docker/test/integration/features/MiNiFi_integration_test_driver.py

@@ -445,3 +445,6 @@ def set_value_on_plc_with_modbus(self, container_name, modbus_cmd):
 
     def enable_ssl_in_nifi(self):
         self.cluster.enable_ssl_in_nifi()
+
+    def check_is_data_present_on_couchbase(self, doc_id: str, bucket_name: str, expected_data: str, expected_data_type: str):
+        assert self.cluster.is_data_present_in_couchbase(doc_id, bucket_name, expected_data, expected_data_type)

docker/test/integration/features/couchbase.feature

@@ -18,12 +18,13 @@ Feature: Executing Couchbase operations from MiNiFi-C++
   Background:
     Given the content of "/tmp/output" is monitored
 
-  Scenario: A MiNiFi instance can insert data to test bucket with PutCouchbaseKey processor
+  Scenario: A MiNiFi instance can insert json data to test bucket with PutCouchbaseKey processor
     Given a Couchbase server is set up
     And a GetFile processor with the "Input Directory" property set to "/tmp/input"
     And a file with the content '{"field1": "value1", "field2": "value2"}' is present in '/tmp/input'
     And a PutCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
     And the "Document Id" property of the PutCouchbaseKey processor is set to "test_doc_id"
+    And the "Document Type" property of the PutCouchbaseKey processor is set to "Json"
     And a LogAttribute processor with the "FlowFiles To Log" property set to "0"
     And a CouchbaseClusterService is setup up for PutCouchbaseKey with the name "CouchbaseClusterService"
 
@@ -39,3 +40,28 @@ Feature: Executing Couchbase operations from MiNiFi-C++
     And the Minifi logs match the following regex: "key:couchbase.doc.sequence.number value:[1-9][0-9]*" in less than 1 seconds
     And the Minifi logs match the following regex: "key:couchbase.partition.uuid value:[1-9][0-9]*" in less than 1 seconds
     And the Minifi logs match the following regex: "key:couchbase.partition.id value:[1-9][0-9]*" in less than 1 seconds
+    And a document with id "test_doc_id" in bucket "test_bucket" is present with data '{"field1": "value1", "field2": "value2"}' of type "Json" in Couchbase
+
+  Scenario: A MiNiFi instance can insert binary data to test bucket with PutCouchbaseKey processor
+    Given a Couchbase server is set up
+    And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+    And a file with the content '{"field1": "value1"}' is present in '/tmp/input'
+    And a PutCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
+    And the "Document Id" property of the PutCouchbaseKey processor is set to "test_doc_id"
+    And the "Document Type" property of the PutCouchbaseKey processor is set to "Binary"
+    And a LogAttribute processor with the "FlowFiles To Log" property set to "0"
+    And a CouchbaseClusterService is setup up for PutCouchbaseKey with the name "CouchbaseClusterService"
+
+    And the "success" relationship of the GetFile processor is connected to the PutCouchbaseKey
+    And the "success" relationship of the PutCouchbaseKey processor is connected to the LogAttribute
+
+    When a Couchbase server is started
+    And all instances start up
+
+    Then the Minifi logs contain the following message: "key:couchbase.bucket value:test_bucket" in less than 60 seconds
+    And the Minifi logs contain the following message: "key:couchbase.doc.id value:test_doc_id" in less than 1 seconds
+    And the Minifi logs match the following regex: "key:couchbase.doc.cas value:[1-9][0-9]*" in less than 1 seconds
+    And the Minifi logs match the following regex: "key:couchbase.doc.sequence.number value:[1-9][0-9]*" in less than 1 seconds
+    And the Minifi logs match the following regex: "key:couchbase.partition.uuid value:[1-9][0-9]*" in less than 1 seconds
+    And the Minifi logs match the following regex: "key:couchbase.partition.id value:[1-9][0-9]*" in less than 1 seconds
+    And a document with id "test_doc_id" in bucket "test_bucket" is present with data '{"field1": "value1"}' of type "Binary" in Couchbase

docker/test/integration/features/steps/steps.py

@@ -1377,3 +1377,8 @@ def step_impl(context, processor_name, service_name):
     processor = context.test.get_node_by_name(processor_name)
     processor.controller_services.append(couchbase_cluster_controller_service)
     processor.set_property("Couchbase Cluster Controller Service", couchbase_cluster_controller_service.name)
+
+
+@then("a document with id \"{doc_id}\" in bucket \"{bucket_name}\" is present with data '{data}' of type \"{data_type}\" in Couchbase")
+def step_impl(context, doc_id: str, bucket_name: str, data: str, data_type: str):
+    context.test.check_is_data_present_on_couchbase(doc_id, bucket_name, data, data_type)

extensions/couchbase/controllerservices/CouchbaseClusterService.cpp

@@ -17,6 +17,9 @@
  */
 
 #include "CouchbaseClusterService.h"
+#include "couchbase/codec/raw_binary_transcoder.hxx"
+#include "couchbase/codec/raw_string_transcoder.hxx"
+#include "couchbase/codec/raw_json_transcoder.hxx"
 
 #include "core/Resource.h"
 
@@ -39,13 +42,22 @@ nonstd::expected<::couchbase::collection, CouchbaseErrorType> CouchbaseClient::g
 }
 
 nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> CouchbaseClient::upsert(const CouchbaseCollection& collection,
-      const std::string& document_id, const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options) {
+    CouchbaseValueType document_type, const std::string& document_id, const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options) {
   auto collection_result = getCollection(collection);
   if (!collection_result.has_value()) {
     return nonstd::make_unexpected(collection_result.error());
   }
 
-  auto [upsert_err, upsert_resp] = collection_result->upsert<::couchbase::codec::raw_binary_transcoder>(document_id, buffer, options).get();
+  std::pair<::couchbase::error, ::couchbase::mutation_result> result;
+  if (document_type == CouchbaseValueType::Json) {
+    result = collection_result->upsert<::couchbase::codec::raw_json_transcoder>(document_id, buffer, options).get();
+  } else if (document_type == CouchbaseValueType::String) {
+    std::string data_str(reinterpret_cast<const char*>(buffer.data()), buffer.size());
+    result = collection_result->upsert<::couchbase::codec::raw_string_transcoder>(document_id, data_str, options).get();
+  } else {
+    result = collection_result->upsert<::couchbase::codec::raw_binary_transcoder>(document_id, buffer, options).get();
+  }
+  auto& [upsert_err, upsert_resp] = result;
   if (upsert_err.ec()) {
     // ambiguous_timeout should not be retried as we do not know if the insert was successful or not
     if (getErrorType(upsert_err.ec()) == CouchbaseErrorType::TEMPORARY && upsert_err.ec().value() != static_cast<int>(::couchbase::errc::common::ambiguous_timeout)) {

extensions/couchbase/controllerservices/CouchbaseClusterService.h

@@ -29,8 +29,6 @@
 #include "couchbase/cluster.hxx"
 #include "core/ProcessContext.h"
 #include "core/logging/LoggerConfiguration.h"
-#include "couchbase/codec/raw_binary_transcoder.hxx"
-#include "couchbase/error.hxx"
 
 namespace org::apache::nifi::minifi::couchbase {
 
@@ -48,6 +46,12 @@ struct CouchbaseUpsertResult {
   std::uint16_t partition_id{0};
 };
 
+enum class CouchbaseValueType {
+  Json,
+  Binary,
+  String
+};
+
 enum class CouchbaseErrorType {
   FATAL,
   TEMPORARY,
@@ -59,8 +63,8 @@ class CouchbaseClient {
     : connection_string_(std::move(connection_string)), username_(std::move(username)), password_(std::move(password)), logger_(logger) {
   }
 
-  nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> upsert(const CouchbaseCollection& collection, const std::string& document_id, const std::vector<std::byte>& buffer,
-    const ::couchbase::upsert_options& options);
+  nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> upsert(const CouchbaseCollection& collection, CouchbaseValueType document_type, const std::string& document_id,
+    const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options);
   std::optional<CouchbaseErrorType> establishConnection();
   void close();
 
@@ -142,10 +146,10 @@ class CouchbaseClusterService : public core::controller::ControllerService {
     }
   }
 
-  virtual nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> upsert(const CouchbaseCollection& collection,
+  virtual nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> upsert(const CouchbaseCollection& collection, CouchbaseValueType document_type,
       const std::string& document_id, const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options) {
     gsl_Expects(client_);
-    return client_->upsert(collection, document_id, buffer, options);
+    return client_->upsert(collection, document_type, document_id, buffer, options);
   }
 
   static gsl::not_null<std::shared_ptr<CouchbaseClusterService>> getFromProperty(const core::ProcessContext& context, const core::PropertyReference& property);

extensions/couchbase/processors/PutCouchbaseKey.cpp

@@ -25,6 +25,7 @@ namespace org::apache::nifi::minifi::couchbase::processors {
 
 void PutCouchbaseKey::onSchedule(core::ProcessContext& context, core::ProcessSessionFactory&) {
   couchbase_cluster_service_ = controllers::CouchbaseClusterService::getFromProperty(context, PutCouchbaseKey::CouchbaseClusterControllerService);
+  document_type_ = utils::parseEnumProperty<CouchbaseValueType>(context, PutCouchbaseKey::DocumentType);
   persist_to_ = utils::parseEnumProperty<::couchbase::persist_to>(context, PutCouchbaseKey::PersistTo);
   replicate_to_ = utils::parseEnumProperty<::couchbase::replicate_to>(context, PutCouchbaseKey::ReplicateTo);
 }
@@ -61,7 +62,7 @@ void PutCouchbaseKey::onTrigger(core::ProcessContext& context, core::ProcessSess
   ::couchbase::upsert_options options;
   options.durability(persist_to_, replicate_to_);
   auto result = session.readBuffer(flow_file);
-  if (auto upsert_result = couchbase_cluster_service_->upsert(collection, document_id, result.buffer, options)) {
+  if (auto upsert_result = couchbase_cluster_service_->upsert(collection, document_type_, document_id, result.buffer, options)) {
     session.putAttribute(*flow_file, "couchbase.bucket", upsert_result->bucket_name);
     session.putAttribute(*flow_file, "couchbase.doc.id", document_id);
     session.putAttribute(*flow_file, "couchbase.doc.cas", std::to_string(upsert_result->cas));