MINIFICPP-2765 Move GCP Extension to stable C API

Author: martinzink

core-framework/common/include/core/PropertyDefinitionBuilder.h

@@ -27,7 +27,21 @@ namespace org::apache::nifi::minifi::core {
 namespace detail {
 template<typename... Types>
 inline constexpr auto TypeNames = std::array<std::string_view, sizeof...(Types)>{core::className<Types>()...};
-}
+
+template <size_t N>
+struct StringLiteral {
+  char value[N];
+  constexpr StringLiteral(const char (&str)[N]) {  // NOLINT(runtime/explicit)
+    for (size_t i = 0; i < N; ++i) {
+      value[i] = str[i];
+    }
+  }
+};
+
+// A variable template that creates permanent static memory for the span to point to
+template <StringLiteral str>
+inline constexpr auto StaticAllowedType = std::array<std::string_view, 1>{std::string_view{str.value, sizeof(str.value) - 1}};
+}  // namespace detail
 
 template<size_t NumAllowedValues = 0>
 struct PropertyDefinitionBuilder {
@@ -81,6 +95,12 @@ struct PropertyDefinitionBuilder {
     return *this;
   }
 
+  template <detail::StringLiteral TypeName>
+  constexpr PropertyDefinitionBuilder<NumAllowedValues> withAllowedType() {
+    property.allowed_types = detail::StaticAllowedType<TypeName>;
+    return *this;
+  }
+
   constexpr PropertyDefinitionBuilder<NumAllowedValues> withValidator(const PropertyValidator& property_validator) {
     property.validator = gsl::make_not_null(&property_validator);
     return *this;

extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h

@@ -17,10 +17,11 @@
 
 #pragma once
 
-#include <string>
 #include <expected>
+#include <string>
 
 #include "api/core/FlowFile.h"
+#include "api/utils/Proxy.h"
 #include "api/utils/Ssl.h"
 #include "minifi-c.h"
 #include "minifi-cpp/core/PropertyDefinition.h"
@@ -45,6 +46,7 @@ class ProcessContext {
   [[nodiscard]] virtual std::map<std::string, std::string> getDynamicProperties(const FlowFile* flow_file) const = 0;
 
   [[nodiscard]] virtual std::expected<utils::net::SslData, std::error_code> getSslData(std::string_view name) const = 0;
+  [[nodiscard]] virtual std::expected<utils::ProxyData, std::error_code> getProxyData(std::string_view name) const = 0;
 };
 
 class CffiProcessContext : public ProcessContext {
@@ -59,6 +61,7 @@ class CffiProcessContext : public ProcessContext {
   [[nodiscard]] bool hasNonEmptyProperty(std::string_view name) const override;
 
   [[nodiscard]] std::expected<utils::net::SslData, std::error_code> getSslData(std::string_view name) const override;
+  [[nodiscard]] std::expected<utils::ProxyData, std::error_code> getProxyData(std::string_view name) const override;
 
  private:
   [[nodiscard]] std::expected<std::string, std::error_code> getProperty(std::string_view name, const FlowFile* flow_file) const;

extension-framework/cpp-extension-lib/include/api/utils/Proxy.h

@@ -0,0 +1,42 @@
+/**
+* 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 <filesystem>
+#include <optional>
+#include <string>
+
+namespace org::apache::nifi::minifi::api::utils {
+
+enum class ProxyType {
+  DIRECT,
+  HTTP
+};
+
+struct BasicAuthCredentials {
+  std::string username;
+  std::string password;
+};
+
+struct ProxyData {
+  std::string host;
+  uint16_t port;
+  std::optional<BasicAuthCredentials> proxy_credentials;
+  ProxyType proxy_type;
+};
+
+}  // namespace org::apache::nifi::minifi::api::utils::net

extension-framework/cpp-extension-lib/include/api/utils/Ssl.h

@@ -17,12 +17,8 @@
 #pragma once
 
 #include <string>
-#include <memory>
-#include <optional>
 #include <filesystem>
 
-#include "utils/Enum.h"
-
 namespace org::apache::nifi::minifi::api::utils::net {
 
 enum class ClientAuthOption {

extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp

@@ -86,4 +86,34 @@ std::expected<utils::net::SslData, std::error_code> CffiProcessContext::getSslDa
   return ssl_data;
 }
 
+std::expected<utils::ProxyData, std::error_code> CffiProcessContext::getProxyData(const std::string_view name) const {
+  auto proxy_data = utils::ProxyData{};
+
+  if (const auto status = MinifiProcessContextGetProxyData(
+          impl_,
+          utils::minifiStringView(name),
+          [](void* data, const MinifiProxyData* minifi_proxy_data) {
+            auto* proxy = static_cast<utils::ProxyData*>(data);
+            proxy->host = utils::toString(minifi_proxy_data->hostname);
+            proxy->port = minifi_proxy_data->port;
+            if (minifi_proxy_data->password && minifi_proxy_data->username) {
+              proxy->proxy_credentials = utils::BasicAuthCredentials{.username = utils::toString(*minifi_proxy_data->username),
+                  .password = utils::toString(*minifi_proxy_data->password)};
+            } else {
+              proxy->proxy_credentials = std::nullopt;
+            }
+            if (minifi_proxy_data->proxy_type == MINIFI_PROXY_TYPE_HTTP) {
+              proxy->proxy_type = utils::ProxyType::HTTP;
+            } else {
+              proxy->proxy_type = utils::ProxyType::DIRECT;
+            }
+          },
+          &proxy_data);
+      status != MINIFI_STATUS_SUCCESS) {
+    return std::unexpected{utils::make_error_code(status)};
+  }
+
+  return proxy_data;
+}
+
 }  // namespace org::apache::nifi::minifi::api::core

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,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 "../../extension-framework/cpp-extension-lib/include/api/core/Resource.h"
+#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) {
+  const MinifiExtensionDefinition extension_definition{
+    .name = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_NAME)),
+    .version = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_VERSION)),
+    .deinit = nullptr,
+    .user_data = nullptr
+  };
+  auto* extension = MinifiRegisterExtension(extension_context, &extension_definition);
+  minifi::api::core::registerProcessors<minifi::extensions::gcp::DeleteGCSObject,
+      minifi::extensions::gcp::FetchGCSObject,
+      minifi::extensions::gcp::ListGCSBucket,
+      minifi::extensions::gcp::PutGCSObject>(extension);
+  minifi::api::core::registerControllerServices<minifi::extensions::gcp::GCPCredentialsControllerService>(extension);
+}

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