[PPANTT-168] feat: introduce ingestion infra (#2514)

* [PPANTT-137] feat: updated modules and workload identity for aks-leonardo and payopts domain

* [PPANTT-137] feat: updated modules and workload identity for aks-leonardo and payopts domain

* [PPANTT-168] feat: Introducing infra domain for gpd ingestion

* [PPANTT-168] feat: Introducing debezium connector, updated module version

* [PPANTT-168] feat: updated eventhub config and kafka-connect.yaml

* fix

* fix

* fix

* fix

* fix

* [PPANTT-168] feat: updated terraform vars

* fix

* fix

* [PPANTT-168] feat: updated 05_debezium_connect.tf, debezium-secrets.yaml, kafka-connect.yaml

* fix

* [PPANTT-168] feat: updated 05_debezium_connect.tf, debezium-secrets.yaml, kafka-connect.yaml

* fix

* fix wait chk resources

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* align config

* change evh to public

* fix

* add mng to cdc-gpd-connection-string

* fix

* fi

* deploy UAT

* [PPANTT-171] feat: introduced ingestion manager to the status page

* [PPANTT-186] feat: introduced alerts for ingestion manager functions

* fix

* [PPANTT-186] feat: introduced alerts for ingestion manager functions

* fix

* [PPANTT-185] feat: introduced logging

* [PPANTT-168] feat: Updated postgres flex config

* fix new name gpd db

* fix new name gpd db

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* apply gps-common prod

* apply gps-app prod

---------

Co-authored-by: pasqualespica <[email protected]>
Co-authored-by: Pasquale Spica <[email protected]>

src/domains/gps-app/00_data.tf

@@ -29,7 +29,7 @@ data "azurerm_key_vault_secret" "gpd_db_pwd" {
 # }
 
 data "azurerm_postgresql_flexible_server" "postgres_flexible_server_private" {
-  count               = 1 # forced count
+  count               = var.env_short == "p" ? 1 : 0  # NEWGPD-DB : DEPRECATED to remove after switch to new WEU gpd  
   name                = format("%s-gpd-pgflex", local.product)
   resource_group_name = format("%s-pgres-flex-rg", local.product)
 }

src/domains/gps-app/05_debezium_connect.tf

@@ -11,6 +11,11 @@ data "azurerm_key_vault_secret" "pgres_gpd_cdc_pwd" {
   key_vault_id = data.azurerm_key_vault.kv.id
 }
 
+data "azurerm_key_vault_secret" "otel_headers" {
+  name         = "elastic-apm-secret-token"
+  key_vault_id = data.azurerm_key_vault.kv.id
+}
+
 data "azurerm_eventhub_namespace_authorization_rule" "cdc_connection_string" {
   name                = "cdc-gpd-connection-string"
   namespace_name      = "pagopa-${var.env_short}-itn-observ-gpd-evh"
@@ -19,6 +24,7 @@ data "azurerm_eventhub_namespace_authorization_rule" "cdc_connection_string" {
 
 # https://github.com/strimzi/strimzi-kafka-operator/tree/main/helm-charts/helm3/strimzi-kafka-operator
 resource "helm_release" "strimzi-kafka-operator" {
+  count      = var.gpd_cdc_enabled ? 1 : 0
   name       = "strimzi-kafka-operator"
   chart      = "strimzi-kafka-operator"
   repository = "oci://quay.io/strimzi-helm"
@@ -59,41 +65,51 @@ locals {
   # https://learn.microsoft.com/it-it/azure/event-hubs/event-hubs-kafka-connect-debezium#configure-kafka-connect-for-event-hubs
 
   kafka_connect_yaml = templatefile("${path.module}/yaml/kafka-connect.yaml", {
-    namespace          = "gps" # kubernetes_namespace.namespace.metadata[0].name
-    replicas           = var.replicas
-    request_memory     = var.request_memory
-    request_cpu        = var.request_cpu
-    limits_memory      = var.limits_memory
-    limits_cpu         = var.limits_cpu
-    bootstrap_servers  = "pagopa-${var.env_short}-itn-observ-gpd-evh.servicebus.windows.net:9093"
-    container_registry = var.container_registry
+    namespace                = "gps" # kubernetes_namespace.namespace.metadata[0].name
+    replicas                 = var.replicas
+    request_memory           = var.request_memory
+    request_cpu              = var.request_cpu
+    limits_memory            = var.limits_memory
+    limits_cpu               = var.limits_cpu
+    bootstrap_servers        = "pagopa-${var.env_short}-itn-observ-gpd-evh.servicebus.windows.net:9093"
+    container_registry       = var.container_registry
+    otlp_endpoint            = "http://otel-collector.elastic-system.svc:4317"
+    otlp_resource_attributes = "service.name=gpddebeziumconnectorkotl,deployment.environment=${var.env}"
+    otlp_headers             = data.azurerm_key_vault_secret.otel_headers.value
   })
 
   postgres_connector_yaml = templatefile("${path.module}/yaml/postgres-connector.yaml", {
     namespace         = "gps" # kubernetes_namespace.namespace.metadata[0].name
-    postgres_hostname = "pagopa-${var.env_short}-gpd-pgflex.postgres.database.azure.com"
+    postgres_hostname = "pagopa-${var.env_short}-${var.location_short}-gpd-pgflex.postgres.database.azure.com"
 
     postgres_port         = 5432
     postgres_db_name      = var.postgres_db_name
     postgres_topic_prefix = "cdc-raw-auto"
     postgres_username     = data.azurerm_key_vault_secret.pgres_gpd_cdc_login.value
     postgres_password     = data.azurerm_key_vault_secret.pgres_gpd_cdc_pwd.value
     tasks_max             = var.tasks_max
+    max_threads           = var.max_threads
   })
 
 }
 
 resource "kubectl_manifest" "debezium_role" {
+  count = var.gpd_cdc_enabled ? 1 : 0
+
   force_conflicts = true
   yaml_body       = local.debezium_role_yaml
 }
 
 resource "kubectl_manifest" "debezium_secrets" {
+  count = var.gpd_cdc_enabled ? 1 : 0
+
   force_conflicts = true
   yaml_body       = local.debezium_secrets_yaml
 }
 
 resource "kubectl_manifest" "debezoum_rbac" {
+  count = var.gpd_cdc_enabled ? 1 : 0
+
   # depends_on      = [kubectl_manifest.debezium_role, kubectl_manifest.debezium_secrets]
   depends_on      = [kubectl_manifest.debezium_role]
   force_conflicts = true
@@ -119,6 +135,8 @@ resource "kubectl_manifest" "debezoum_rbac" {
 # }
 
 resource "kubectl_manifest" "kafka_connect" {
+  count = var.gpd_cdc_enabled ? 1 : 0
+
   depends_on = [
     helm_release.strimzi-kafka-operator
   ]
@@ -127,6 +145,9 @@ resource "kubectl_manifest" "kafka_connect" {
 }
 
 resource "null_resource" "wait_kafka_connect" {
+
+  count = var.gpd_cdc_enabled ? 1 : 0
+
   depends_on = [
     kubectl_manifest.kafka_connect
   ]
@@ -137,6 +158,8 @@ resource "null_resource" "wait_kafka_connect" {
 }
 
 resource "kubectl_manifest" "postgres_connector" {
+  count = var.gpd_cdc_enabled ? 1 : 0
+
   depends_on = [
     helm_release.strimzi-kafka-operator
   ]
@@ -145,6 +168,8 @@ resource "kubectl_manifest" "postgres_connector" {
 }
 
 resource "null_resource" "wait_postgres_connector" {
+  count = var.gpd_cdc_enabled ? 1 : 0
+
   depends_on = [
     kubectl_manifest.kafka_connect
   ]

src/domains/gps-app/99_variables.tf

@@ -544,8 +544,20 @@ variable "limits_cpu" {
   default     = "0.5"
 }
 
+variable "max_threads" {
+  type        = number
+  description = "Number of max_threads"
+  default     = 1
+}
+
 variable "gh_runner_job_location" {
   type        = string
   description = "(Optional) The GH runner container app job location. Consistent with the container app environment location"
   default     = "westeurope"
 }
+
+variable "gpd_cdc_enabled" {
+  type        = bool
+  description = "Enable CDC for GDP"
+  default     = false
+}

src/domains/gps-app/env/weu-dev/terraform.tfvars

@@ -65,7 +65,7 @@ pgbouncer_enabled = false
 # WISP-dismantling-cfg
 create_wisp_converter = true
 
-### debezium kafka conn
+### debezium zookeeper_yaml
 zookeeper_replicas       = "1"
 zookeeper_request_memory = "512Mi"
 zookeeper_request_cpu    = "0.5"
@@ -74,11 +74,15 @@ zookeeper_limits_cpu     = "0.5"
 zookeeper_jvm_xms        = "512m"
 zookeeper_jvm_xmx        = "512m"
 zookeeper_storage_size   = "100Gi"
-replicas                 = 1
-request_cpu              = 0.5
-request_memory           = "512Mi"
-limits_memory            = "512Mi"
-limits_cpu               = 0.5
-postgres_db_name         = "apd"
-tasks_max                = "1"
-container_registry       = "pagopadcommonacr.azurecr.io"
+
+### debezium kafka_connect_yaml
+replicas           = 1
+request_cpu        = 0.5
+limits_cpu         = 0.5
+request_memory     = "512Mi"
+limits_memory      = "512Mi"
+postgres_db_name   = "apd"
+tasks_max          = "1"
+container_registry = "pagopadcommonacr.azurecr.io"
+max_threads        = 1
+gpd_cdc_enabled    = true

src/domains/gps-app/env/weu-prod/terraform.tfvars

@@ -113,6 +113,7 @@ fn_app_storage_account_info = {
   advanced_threat_protection_enable = true
 }
 
+
 ### debezium kafka conn
 zookeeper_replicas       = 3
 zookeeper_request_memory = "512Mi"
@@ -122,11 +123,15 @@ zookeeper_limits_cpu     = 1
 zookeeper_jvm_xms        = "512m"
 zookeeper_jvm_xmx        = "1024m"
 zookeeper_storage_size   = "100Gi"
-replicas                 = 3
-request_cpu              = 0.5
-request_memory           = "512Mi"
-limits_memory            = "1024Mi"
-limits_cpu               = 1
-postgres_db_name         = "apd"
-tasks_max                = "1"
-container_registry       = "pagopapcommonacr.azurecr.io"
+
+### debezium kafka_connect_yaml
+replicas           = 2
+request_cpu        = 0.5
+limits_cpu         = 2
+request_memory     = "512Mi"
+limits_memory      = "3072Mi"
+postgres_db_name   = "apd"
+tasks_max          = "1"
+container_registry = "pagopapcommonacr.azurecr.io"
+max_threads        = 10
+gpd_cdc_enabled    = false

src/domains/gps-app/env/weu-uat/terraform.tfvars

@@ -65,7 +65,7 @@ pgbouncer_enabled = true
 # WISP-dismantling-cfg
 create_wisp_converter = true
 
-### debezium kafka conn
+### debezium zookeeper_yaml
 zookeeper_replicas       = 3
 zookeeper_request_memory = "512Mi"
 zookeeper_request_cpu    = 0.5
@@ -74,11 +74,15 @@ zookeeper_limits_cpu     = 1
 zookeeper_jvm_xms        = "512m"
 zookeeper_jvm_xmx        = "1024m"
 zookeeper_storage_size   = "100Gi"
-replicas                 = 1
-request_cpu              = 0.5
-request_memory           = "512Mi"
-limits_memory            = "1024Mi"
-limits_cpu               = 1
-postgres_db_name         = "apd"
-tasks_max                = "1"
-container_registry       = "pagopaucommonacr.azurecr.io"
+
+### debezium kafka_connect_yaml
+replicas           = 1
+request_cpu        = 0.5
+limits_cpu         = 2
+request_memory     = "512Mi"
+limits_memory      = "3072Mi"
+postgres_db_name   = "apd"
+tasks_max          = "1"
+container_registry = "pagopaucommonacr.azurecr.io"
+max_threads        = 10
+gpd_cdc_enabled    = true

src/domains/gps-app/yaml/kafka-connect.yaml

@@ -45,6 +45,11 @@ spec:
     internal.value.converter: org.apache.kafka.connect.json.JsonConverter
     internal.key.converter.schemas.enable: false
     internal.value.converter.schemas.enable: false
+    otel.traces.exporter: otlp
+    otel.propagators: tracecontext
+    otel.exporter.otlp.endpoint: "${otlp_endpoint}"
+    otel.resource.attributes: "${otlp_resource_attributes}"
+    otel.expoerter.oltp.header: "${otlp_headers}"
   build:
     output:
       type: docker

src/domains/gps-app/yaml/postgres-connector.yaml

@@ -24,5 +24,6 @@ spec:
     schema.include.list: "apd"
     table.include.list: "apd.payment_option,apd.payment_position,apd.transfer"
     plugin.name: "pgoutput"
+    snapshot.max.threads: ${max_threads}
     publication.autocreate.mode: "disabled" # shall be create before 
     # publication.autocreate.mode: "filtered"  # create it

src/domains/gps-common/01_network.tf

@@ -5,7 +5,7 @@ module "storage_account_snet" {
   address_prefixes                              = var.gpd_sftp_cidr_subnet_gpd_storage_account
   resource_group_name                           = local.vnet_resource_group_name
   virtual_network_name                          = data.azurerm_virtual_network.vnet.name
-  service_endpoints                             = ["Microsoft.Web", "Microsoft.AzureCosmosDB", "Microsoft.EventHub"]
+  service_endpoints                             = ["Microsoft.Web", "Microsoft.AzureCosmosDB", "Microsoft.EventHub", "Microsoft.Storage"]
   private_link_service_network_policies_enabled = var.gpd_sftp_sa_snet_private_link_service_network_policies_enabled
 }
 

src/domains/gps-common/02_security.tf

@@ -547,45 +547,54 @@ resource "azurerm_key_vault_secret" "cdc-raw-auto_apd_transfer-rx_kv" {
   content_type = "text/plain"
   key_vault_id = module.key_vault.id
 }
+
+# ##########################
 # CDC GDP out eventhub
+# ##########################
 data "azurerm_eventhub_authorization_rule" "gpd_ingestion_apd_payment_option_tx" {
+  count               = var.gpd_cdc_enabled ? 1 : 0
   name                = "gpd-ingestion.apd.payment_option-tx"
   namespace_name      = "pagopa-${var.env_short}-itn-observ-gpd-evh"
   eventhub_name       = "gpd-ingestion.apd.payment_option"
   resource_group_name = "pagopa-${var.env_short}-itn-observ-evh-rg"
 }
 
 resource "azurerm_key_vault_secret" "gpd_ingestion_apd_payment_option_tx_kv" {
+  count        = var.gpd_cdc_enabled ? 1 : 0
   name         = "payment-option-topic-output-conn-string"
-  value        = data.azurerm_eventhub_authorization_rule.gpd_ingestion_apd_payment_option_tx.primary_connection_string
+  value        = data.azurerm_eventhub_authorization_rule.gpd_ingestion_apd_payment_option_tx[0].primary_connection_string
   content_type = "text/plain"
   key_vault_id = module.key_vault.id
 }
 
 data "azurerm_eventhub_authorization_rule" "gpd_ingestion_apd_payment_position_tx" {
+  count               = var.gpd_cdc_enabled ? 1 : 0
   name                = "gpd-ingestion.apd.payment_position-tx"
   namespace_name      = "pagopa-${var.env_short}-itn-observ-gpd-evh"
   eventhub_name       = "gpd-ingestion.apd.payment_position"
   resource_group_name = "pagopa-${var.env_short}-itn-observ-evh-rg"
 }
 
 resource "azurerm_key_vault_secret" "gpd_ingestion_apd_payment_position_tx_kv" {
+  count        = var.gpd_cdc_enabled ? 1 : 0
   name         = "payment-position-topic-output-conn-string"
-  value        = data.azurerm_eventhub_authorization_rule.gpd_ingestion_apd_payment_position_tx.primary_connection_string
+  value        = data.azurerm_eventhub_authorization_rule.gpd_ingestion_apd_payment_position_tx[0].primary_connection_string
   content_type = "text/plain"
   key_vault_id = module.key_vault.id
 }
 
 data "azurerm_eventhub_authorization_rule" "gpd_ingestion_apd_payment_option_transfer_tx" {
+  count               = var.gpd_cdc_enabled ? 1 : 0
   name                = "gpd-ingestion.apd.transfer-tx"
   namespace_name      = "pagopa-${var.env_short}-itn-observ-gpd-evh"
   eventhub_name       = "gpd-ingestion.apd.transfer"
   resource_group_name = "pagopa-${var.env_short}-itn-observ-evh-rg"
 }
 
 resource "azurerm_key_vault_secret" "gpd_ingestion_apd_payment_option_transfer_tx_kv" {
+  count        = var.gpd_cdc_enabled ? 1 : 0
   name         = "transfer-topic-output-conn-string"
-  value        = data.azurerm_eventhub_authorization_rule.gpd_ingestion_apd_payment_option_transfer_tx.primary_connection_string
+  value        = data.azurerm_eventhub_authorization_rule.gpd_ingestion_apd_payment_option_transfer_tx[0].primary_connection_string
   content_type = "text/plain"
   key_vault_id = module.key_vault.id
-}
+}

src/domains/gps-common/03_cosmosdb.tf

@@ -53,10 +53,11 @@ module "gps_cosmosdb_account" {
   allowed_virtual_network_subnet_ids = var.cosmos_gps_db_params.public_network_access_enabled ? var.env_short == "d" ? [] : [data.azurerm_subnet.aks_subnet.id] : [data.azurerm_subnet.aks_subnet.id]
 
   # private endpoint
-  private_endpoint_sql_name = "${local.project}-cosmos-sql-endpoint"
-  private_endpoint_enabled  = var.cosmos_gps_db_params.private_endpoint_enabled
-  subnet_id                 = module.gps_cosmosdb_snet.id
-  private_dns_zone_sql_ids  = [data.azurerm_private_dns_zone.cosmos.id]
+  private_endpoint_sql_name           = "${local.project}-cosmos-sql-endpoint" # forced after update module vers
+  private_service_connection_sql_name = "${local.project}-cosmos-sql-endpoint" # forced after update module vers
+  private_endpoint_enabled            = var.cosmos_gps_db_params.private_endpoint_enabled
+  subnet_id                           = module.gps_cosmosdb_snet.id
+  private_dns_zone_sql_ids            = [data.azurerm_private_dns_zone.cosmos.id]
 
   tags = var.tags
 }