MINIFICPP-2765 Move GCP Extension to stable C API

Author: martinzink

extensions/gcp/CMakeLists.txt

@@ -30,8 +30,8 @@ add_minifi_library(minifi-gcp SHARED ${SOURCES})
 if (NOT WIN32)
     target_compile_options(minifi-gcp PRIVATE -Wno-error=deprecated-declarations)  # Suppress deprecation warnings for std::rel_ops usage
 endif()
-target_link_libraries(minifi-gcp ${LIBMINIFI} google-cloud-cpp::storage)
-target_include_directories(minifi-gcp SYSTEM PUBLIC ${google-cloud-cpp_INCLUDE_DIRS})
+target_link_libraries(minifi-gcp minifi-cpp-extension-lib google-cloud-cpp::storage)
 
-register_extension(minifi-gcp "GCP EXTENSIONS" GCP-EXTENSIONS "This enables Google Cloud Platform support" "extensions/gcp/tests")
+target_include_directories(minifi-gcp SYSTEM PUBLIC ${google-cloud-cpp_INCLUDE_DIRS})
 
+register_c_api_extension(minifi-gcp "GCP EXTENSIONS" GCP-EXTENSIONS "This enables Google Cloud Platform support" "extensions/gcp/tests")

extensions/gcp/ExtensionInitializer.cpp

@@ -0,0 +1,52 @@
+/**
+ * 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 "api/core/Resource.h"
+#include "api/utils/minifi-c-utils.h"
+#include "processors/DeleteGCSObject.h"
+#include "processors/FetchGCSObject.h"
+#include "processors/ListGCSBucket.h"
+#include "processors/PutGCSObject.h"
+
+#define MKSOC(x) #x
+#define MAKESTRING(x) MKSOC(x)  // NOLINT(cppcoreguidelines-macro-usage)
+
+namespace minifi = org::apache::nifi::minifi;
+
+CEXTENSIONAPI const uint32_t MinifiApiVersion = MINIFI_API_VERSION;
+
+CEXTENSIONAPI void MinifiInitExtension(MinifiExtensionContext* extension_context) {
+  MinifiExtensionCreateInfo ext_create_info{.name = minifi::api::utils::toStringView(MAKESTRING(EXTENSION_NAME)),
+      .version = minifi::api::utils::toStringView(MAKESTRING(EXTENSION_VERSION)),
+      .deinit = nullptr,
+      .user_data = nullptr};
+  auto* extension = MinifiCreateExtension(extension_context, &ext_create_info);
+  minifi::api::core::useProcessorClassDescription<minifi::extensions::gcp::DeleteGCSObject>([&](const MinifiProcessorClassDefinition& description) {
+    MinifiRegisterProcessor(extension, &description);
+  });
+  minifi::api::core::useProcessorClassDescription<minifi::extensions::gcp::FetchGCSObject>([&](const MinifiProcessorClassDefinition& description) {
+    MinifiRegisterProcessor(extension, &description);
+  });
+  minifi::api::core::useProcessorClassDescription<minifi::extensions::gcp::ListGCSBucket>([&](const MinifiProcessorClassDefinition& description) {
+    MinifiRegisterProcessor(extension, &description);
+  });
+  minifi::api::core::useProcessorClassDescription<minifi::extensions::gcp::PutGCSObject>([&](const MinifiProcessorClassDefinition& description) {
+    MinifiRegisterProcessor(extension, &description);
+  });
+  minifi::api::core::useControllerServiceClassDescription<minifi::extensions::gcp::GCPCredentialsControllerService>(
+      [&](const MinifiControllerServiceClassDefinition& description) { MinifiRegisterControllerService(extension, &description); });
+}

extensions/gcp/GCPAttributes.h

@@ -19,8 +19,9 @@
 
 #include <string_view>
 
+#include "api/core/FlowFile.h"
+#include "api/core/ProcessSession.h"
 #include "google/cloud/storage/object_metadata.h"
-#include "minifi-cpp/core/FlowFile.h"
 
 namespace org::apache::nifi::minifi::extensions::gcp {
 
@@ -50,32 +51,32 @@ constexpr std::string_view GCS_SELF_LINK_ATTR = "gcs.self.link";
 constexpr std::string_view GCS_ENCRYPTION_ALGORITHM_ATTR = "gcs.encryption.algorithm";
 constexpr std::string_view GCS_ENCRYPTION_SHA256_ATTR = "gcs.encryption.sha256";
 
-inline void setAttributesFromObjectMetadata(core::FlowFile& flow_file, const ::google::cloud::storage::ObjectMetadata& object_metadata) {
-  flow_file.setAttribute(GCS_BUCKET_ATTR, object_metadata.bucket());
-  flow_file.setAttribute(GCS_OBJECT_NAME_ATTR, object_metadata.name());
-  flow_file.setAttribute(GCS_SIZE_ATTR, std::to_string(object_metadata.size()));
-  flow_file.setAttribute(GCS_CRC32C_ATTR, object_metadata.crc32c());
-  flow_file.setAttribute(GCS_MD5_ATTR, object_metadata.md5_hash());
-  flow_file.setAttribute(GCS_CONTENT_ENCODING_ATTR, object_metadata.content_encoding());
-  flow_file.setAttribute(GCS_CONTENT_LANGUAGE_ATTR, object_metadata.content_language());
-  flow_file.setAttribute(GCS_CONTENT_DISPOSITION_ATTR, object_metadata.content_disposition());
-  flow_file.setAttribute(GCS_CREATE_TIME_ATTR, std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(object_metadata.time_created().time_since_epoch()).count()));
-  flow_file.setAttribute(GCS_UPDATE_TIME_ATTR, std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(object_metadata.updated().time_since_epoch()).count()));
-  flow_file.setAttribute(GCS_DELETE_TIME_ATTR, std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(object_metadata.time_deleted().time_since_epoch()).count()));
-  flow_file.setAttribute(GCS_MEDIA_LINK_ATTR, object_metadata.media_link());
-  flow_file.setAttribute(GCS_SELF_LINK_ATTR, object_metadata.self_link());
-  flow_file.setAttribute(GCS_ETAG_ATTR, object_metadata.etag());
-  flow_file.setAttribute(GCS_GENERATED_ID, object_metadata.id());
-  flow_file.setAttribute(GCS_META_GENERATION, std::to_string(object_metadata.metageneration()));
-  flow_file.setAttribute(GCS_GENERATION, std::to_string(object_metadata.generation()));
-  flow_file.setAttribute(GCS_STORAGE_CLASS, object_metadata.storage_class());
+inline void setAttributesFromObjectMetadata(api::core::FlowFile& flow_file, const ::google::cloud::storage::ObjectMetadata& object_metadata, api::core::ProcessSession& session) {
+  session.setAttribute(flow_file, GCS_BUCKET_ATTR, object_metadata.bucket());
+  session.setAttribute(flow_file, GCS_OBJECT_NAME_ATTR, object_metadata.name());
+  session.setAttribute(flow_file, GCS_SIZE_ATTR, std::to_string(object_metadata.size()));
+  session.setAttribute(flow_file, GCS_CRC32C_ATTR, object_metadata.crc32c());
+  session.setAttribute(flow_file, GCS_MD5_ATTR, object_metadata.md5_hash());
+  session.setAttribute(flow_file, GCS_CONTENT_ENCODING_ATTR, object_metadata.content_encoding());
+  session.setAttribute(flow_file, GCS_CONTENT_LANGUAGE_ATTR, object_metadata.content_language());
+  session.setAttribute(flow_file, GCS_CONTENT_DISPOSITION_ATTR, object_metadata.content_disposition());
+  session.setAttribute(flow_file, GCS_CREATE_TIME_ATTR, std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(object_metadata.time_created().time_since_epoch()).count()));
+  session.setAttribute(flow_file, GCS_UPDATE_TIME_ATTR, std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(object_metadata.updated().time_since_epoch()).count()));
+  session.setAttribute(flow_file, GCS_DELETE_TIME_ATTR, std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(object_metadata.time_deleted().time_since_epoch()).count()));
+  session.setAttribute(flow_file, GCS_MEDIA_LINK_ATTR, object_metadata.media_link());
+  session.setAttribute(flow_file, GCS_SELF_LINK_ATTR, object_metadata.self_link());
+  session.setAttribute(flow_file, GCS_ETAG_ATTR, object_metadata.etag());
+  session.setAttribute(flow_file, GCS_GENERATED_ID, object_metadata.id());
+  session.setAttribute(flow_file, GCS_META_GENERATION, std::to_string(object_metadata.metageneration()));
+  session.setAttribute(flow_file, GCS_GENERATION, std::to_string(object_metadata.generation()));
+  session.setAttribute(flow_file, GCS_STORAGE_CLASS, object_metadata.storage_class());
   if (object_metadata.has_customer_encryption()) {
-    flow_file.setAttribute(GCS_ENCRYPTION_ALGORITHM_ATTR, object_metadata.customer_encryption().encryption_algorithm);
-    flow_file.setAttribute(GCS_ENCRYPTION_SHA256_ATTR, object_metadata.customer_encryption().key_sha256);
+    session.setAttribute(flow_file, GCS_ENCRYPTION_ALGORITHM_ATTR, object_metadata.customer_encryption().encryption_algorithm);
+    session.setAttribute(flow_file, GCS_ENCRYPTION_SHA256_ATTR, object_metadata.customer_encryption().key_sha256);
   }
   if (object_metadata.has_owner()) {
-    flow_file.setAttribute(GCS_OWNER_ENTITY_ATTR, object_metadata.owner().entity);
-    flow_file.setAttribute(GCS_OWNER_ENTITY_ID_ATTR, object_metadata.owner().entity_id);
+    session.setAttribute(flow_file, GCS_OWNER_ENTITY_ATTR, object_metadata.owner().entity);
+    session.setAttribute(flow_file, GCS_OWNER_ENTITY_ID_ATTR, object_metadata.owner().entity_id);
   }
 }
 

extensions/gcp/controllerservices/GCPCredentialsControllerService.cpp

@@ -18,34 +18,36 @@
 
 #include "GCPCredentialsControllerService.h"
 
-#include "core/Resource.h"
 #include "google/cloud/storage/client.h"
-#include "utils/ProcessorConfigUtils.h"
-#include "utils/file/FileUtils.h"
 
 namespace org::apache::nifi::minifi::extensions::gcp {
 
-void GCPCredentialsControllerService::initialize() {
-  setSupportedProperties(Properties);
+namespace {
+// TODO(MINIFICPP-2763) use utils::file::get_content instead
+std::string get_content(const std::filesystem::path& file_name) {
+  std::ifstream file(file_name, std::ifstream::binary);
+  std::string content((std::istreambuf_iterator<char>(file)), std::istreambuf_iterator<char>());
+  return content;
+}
 }
 
-std::shared_ptr<google::cloud::Credentials> GCPCredentialsControllerService::createCredentialsFromJsonPath() const {
-  const auto json_path = getProperty(JsonFilePath.name);
+std::shared_ptr<google::cloud::Credentials> GCPCredentialsControllerService::createCredentialsFromJsonPath(api::core::ControllerServiceContext& ctx) const {
+  const auto json_path = ctx.getProperty(JsonFilePath.name);
   if (!json_path) {
     logger_->log_error("Missing or invalid {}", JsonFilePath.name);
     return nullptr;
   }
 
-  if (!utils::file::exists(*json_path)) {
+  if (std::error_code ec; !std::filesystem::exists(*json_path, ec) || ec) {
     logger_->log_error("JSON file for GCP credentials '{}' does not exist", *json_path);
     return nullptr;
   }
 
-  return google::cloud::MakeServiceAccountCredentials(utils::file::get_content(*json_path));
+  return google::cloud::MakeServiceAccountCredentials(get_content(*json_path));
 }
 
-std::shared_ptr<google::cloud::Credentials> GCPCredentialsControllerService::createCredentialsFromJsonContents() const {
-  auto json_contents = getProperty(JsonContents.name);
+std::shared_ptr<google::cloud::Credentials> GCPCredentialsControllerService::createCredentialsFromJsonContents(api::core::ControllerServiceContext& ctx) const {
+  auto json_contents = ctx.getProperty(JsonContents.name);
   if (!json_contents) {
     logger_->log_error("Missing or invalid {}", JsonContents.name);
     return nullptr;
@@ -54,9 +56,9 @@ std::shared_ptr<google::cloud::Credentials> GCPCredentialsControllerService::cre
   return google::cloud::MakeServiceAccountCredentials(*json_contents);
 }
 
-void GCPCredentialsControllerService::onEnable() {
+MinifiStatus GCPCredentialsControllerService::enableImpl(api::core::ControllerServiceContext& ctx) {
   std::optional<CredentialsLocation> credentials_location;
-  if (const auto value = getProperty(CredentialsLoc.name)) {
+  if (const auto value = ctx.getProperty(CredentialsLoc.name)) {
     credentials_location = magic_enum::enum_cast<CredentialsLocation>(*value);
   }
   if (!credentials_location) {
@@ -68,15 +70,15 @@ void GCPCredentialsControllerService::onEnable() {
   } else if (*credentials_location == CredentialsLocation::USE_COMPUTE_ENGINE_CREDENTIALS) {
     credentials_ = google::cloud::MakeComputeEngineCredentials();
   } else if (*credentials_location == CredentialsLocation::USE_JSON_FILE) {
-    credentials_ = createCredentialsFromJsonPath();
+    credentials_ = createCredentialsFromJsonPath(ctx);
   } else if (*credentials_location == CredentialsLocation::USE_JSON_CONTENTS) {
-    credentials_ = createCredentialsFromJsonContents();
+    credentials_ = createCredentialsFromJsonContents(ctx);
   } else if (*credentials_location == CredentialsLocation::USE_ANONYMOUS_CREDENTIALS) {
     credentials_ = google::cloud::MakeInsecureCredentials();
   }
   if (!credentials_)
     logger_->log_error("Couldn't create valid credentials");
+  return MINIFI_STATUS_SUCCESS;
 }
 
-REGISTER_RESOURCE(GCPCredentialsControllerService, ControllerService);
 }  // namespace org::apache::nifi::minifi::extensions::gcp

extensions/gcp/controllerservices/GCPCredentialsControllerService.h

@@ -18,17 +18,15 @@
 #pragma once
 
 #include <filesystem>
-#include <string>
 #include <memory>
+#include <string>
 
-#include "core/controller/ControllerServiceBase.h"
-#include "minifi-cpp/core/logging/Logger.h"
-#include "core/logging/LoggerFactory.h"
-#include "minifi-cpp/core/PropertyDefinition.h"
+#include "api/core/ControllerServiceImpl.h"
+#include "api/utils/Export.h"
 #include "core/PropertyDefinitionBuilder.h"
-#include "utils/Enum.h"
-
 #include "google/cloud/credentials.h"
+#include "minifi-cpp/core/PropertyDefinition.h"
+#include "utils/Enum.h"
 
 namespace org::apache::nifi::minifi::extensions::gcp {
 enum class CredentialsLocation {
@@ -63,7 +61,7 @@ constexpr customize_t enum_name<CredentialsLocation>(CredentialsLocation value)
 
 namespace org::apache::nifi::minifi::extensions::gcp {
 
-class GCPCredentialsControllerService : public core::controller::ControllerServiceBase, public core::controller::ControllerServiceHandle {
+class GCPCredentialsControllerService : public api::core::ControllerServiceImpl {
  public:
   EXTENSIONAPI static constexpr const char* Description = "Manages the credentials for Google Cloud Platform. This allows for multiple Google Cloud Platform related processors "
       "to reference this single controller service so that Google Cloud Platform credentials can be managed and controlled in a central location.";
@@ -91,21 +89,16 @@ class GCPCredentialsControllerService : public core::controller::ControllerServi
 
 
   EXTENSIONAPI static constexpr bool SupportsDynamicProperties = false;
-  ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_CONTROLLER_SERVICES
-
-  using ControllerServiceBase::ControllerServiceBase;
-
-  void initialize() override;
 
-  void onEnable() override;
+  using ControllerServiceImpl::ControllerServiceImpl;
 
-  [[nodiscard]] ControllerServiceHandle* getControllerServiceHandle() override {return this;}
+  MinifiStatus enableImpl(api::core::ControllerServiceContext& ctx) override;
 
   [[nodiscard]] const auto& getCredentials() const { return credentials_; }
 
  protected:
-  [[nodiscard]] std::shared_ptr<google::cloud::Credentials> createCredentialsFromJsonPath() const;
-  [[nodiscard]] std::shared_ptr<google::cloud::Credentials> createCredentialsFromJsonContents() const;
+  [[nodiscard]] std::shared_ptr<google::cloud::Credentials> createCredentialsFromJsonPath(api::core::ControllerServiceContext& ctx) const;
+  [[nodiscard]] std::shared_ptr<google::cloud::Credentials> createCredentialsFromJsonContents(api::core::ControllerServiceContext& ctx) const;
 
 
   std::shared_ptr<google::cloud::Credentials> credentials_;

extensions/gcp/processors/DeleteGCSObject.cpp

@@ -17,69 +17,62 @@
 
 #include "DeleteGCSObject.h"
 
-#include "utils/ProcessorConfigUtils.h"
 
 #include "../GCPAttributes.h"
-#include "minifi-cpp/core/FlowFile.h"
-#include "minifi-cpp/core/ProcessContext.h"
-#include "core/ProcessSession.h"
-#include "core/Resource.h"
+#include "api/core/ProcessContext.h"
+#include "api/core/ProcessSession.h"
+#include "api/core/Resource.h"
+#include "api/utils/ProcessorConfigUtils.h"
 
 namespace gcs = ::google::cloud::storage;
 
 namespace org::apache::nifi::minifi::extensions::gcp {
-void DeleteGCSObject::initialize() {
-  setSupportedProperties(Properties);
-  setSupportedRelationships(Relationships);
-}
 
-void DeleteGCSObject::onTrigger(core::ProcessContext& context, core::ProcessSession& session) {
+MinifiStatus DeleteGCSObject::onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) {
   gsl_Expects(gcp_credentials_);
 
   auto flow_file = session.get();
   if (!flow_file) {
-    context.yield();
-    return;
+    return MINIFI_STATUS_PROCESSOR_YIELD;
   }
 
-  auto bucket = context.getProperty(Bucket, flow_file.get());
+  auto bucket = api::utils::parseOptionalProperty(context, Bucket, &flow_file);
   if (!bucket || bucket->empty()) {
     logger_->log_error("Missing bucket name");
-    session.transfer(flow_file, Failure);
-    return;
+    session.transfer(std::move(flow_file), Failure);
+    return MINIFI_STATUS_SUCCESS;
   }
-  auto object_name = context.getProperty(Key, flow_file.get());
+  auto object_name = api::utils::parseOptionalProperty(context, Key, &flow_file);
   if (!object_name || object_name->empty()) {
     logger_->log_error("Missing object name");
-    session.transfer(flow_file, Failure);
-    return;
+    session.transfer(std::move(flow_file), Failure);
+    return MINIFI_STATUS_SUCCESS;
   }
 
   gcs::Generation generation;
-  if (const auto object_generation_str =  context.getProperty(ObjectGeneration, flow_file.get()); object_generation_str && !object_generation_str->empty()) {
+  if (auto object_generation_str = api::utils::parseOptionalProperty(context, ObjectGeneration, &flow_file); object_generation_str && !object_generation_str->empty()) {
     if (const auto geni64 = parsing::parseIntegral<int64_t>(*object_generation_str)) {
       generation = gcs::Generation{*geni64};
     } else {
       logger_->log_error("Invalid generation: {}", *object_generation_str);
-      session.transfer(flow_file, Failure);
-      return;
+      session.transfer(std::move(flow_file), Failure);
+    return MINIFI_STATUS_SUCCESS;
     }
   }
 
   auto status = getClient().DeleteObject(*bucket, *object_name, generation, gcs::IfGenerationNotMatch(0));
 
   if (!status.ok()) {
-    flow_file->setAttribute(GCS_STATUS_MESSAGE, status.message());
-    flow_file->setAttribute(GCS_ERROR_REASON, status.error_info().reason());
-    flow_file->setAttribute(GCS_ERROR_DOMAIN, status.error_info().domain());
+    session.setAttribute(flow_file, GCS_STATUS_MESSAGE, status.message());
+    session.setAttribute(flow_file, GCS_ERROR_REASON, status.error_info().reason());
+    session.setAttribute(flow_file, GCS_ERROR_DOMAIN, status.error_info().domain());
     logger_->log_error("Failed to delete {} object from {} bucket on Google Cloud Storage {} {}", *object_name, *bucket, status.message(), status.error_info().reason());
-    session.transfer(flow_file, Failure);
-    return;
+    session.transfer(std::move(flow_file), Failure);
+    return MINIFI_STATUS_SUCCESS;
   }
 
-  session.transfer(flow_file, Success);
+  session.transfer(std::move(flow_file), Success);
+  return MINIFI_STATUS_SUCCESS;
 }
 
-REGISTER_RESOURCE(DeleteGCSObject, Processor);
-
 }  // namespace org::apache::nifi::minifi::extensions::gcp