Feature/eav topics prod

infrastructure.yaml

@@ -29,6 +29,78 @@ environments:
             retention.ms: "604800000"
             min.insync.replicas: "1"
           deletionPolicy: orphan
+        - name: eav.attribute.created.v1
+          owner: platform
+          clusterRef: prod
+          partitions: 3
+          replicationFactor: 1
+          config:
+            cleanup.policy: delete
+            retention.ms: "604800000"
+            min.insync.replicas: "1"
+          deletionPolicy: orphan
+          ownerRoles: []
+        - name: eav.attribute.updated.v1
+          owner: platform
+          clusterRef: prod
+          partitions: 3
+          replicationFactor: 1
+          config:
+            cleanup.policy: delete
+            retention.ms: "604800000"
+            min.insync.replicas: "1"
+          deletionPolicy: orphan
+          ownerRoles: []
+        - name: eav.attribute.deleted.v1
+          owner: platform
+          clusterRef: prod
+          partitions: 3
+          replicationFactor: 1
+          config:
+            cleanup.policy: delete
+            retention.ms: "604800000"
+            min.insync.replicas: "1"
+          deletionPolicy: orphan
+          ownerRoles: []
+        - name: eav.value_option.created.v1
+          owner: platform
+          clusterRef: prod
+          partitions: 3
+          replicationFactor: 1
+          config:
+            cleanup.policy: delete
+            retention.ms: "604800000"
+            min.insync.replicas: "1"
+          deletionPolicy: orphan
+          ownerRoles: []
+        - name: eav.value_option.deleted.v1
+          owner: platform
+          clusterRef: prod
+          partitions: 3
+          replicationFactor: 1
+          config:
+            cleanup.policy: delete
+            retention.ms: "604800000"
+            min.insync.replicas: "1"
+          deletionPolicy: orphan
+          ownerRoles: []
+      users:
+        - name: premises
+          clusterRef: prod
+          adoptExisting: true
+          permissions:
+            - topic: premises
+              roles: ["ACCESS_ROLE_CONSUMER", "ACCESS_ROLE_PRODUCER"]
+            - topic: eav.attribute.created.v1
+              roles: ["ACCESS_ROLE_CONSUMER"]
+            - topic: eav.attribute.deleted.v1
+              roles: ["ACCESS_ROLE_CONSUMER"]
+            - topic: eav.attribute.updated.v1
+              roles: ["ACCESS_ROLE_CONSUMER"]
+            - topic: eav.value_option.created.v1
+              roles: ["ACCESS_ROLE_CONSUMER"]
+            - topic: eav.value_option.deleted.v1
+              roles: ["ACCESS_ROLE_CONSUMER"]
 
     rabbitmq:
       management_endpoint: http://default-rabbit-cluster.rabbitmq.svc.cluster.local:15672

live/prod/kafka-topics/terragrunt.hcl

@@ -15,6 +15,7 @@ locals {
   env_config   = local.infra_config.environments[local.env_name]
 
   kafka_topics = try(local.env_config.kafka.topics, [])
+  kafka_users  = try(local.env_config.kafka.users, [])
 
   kafka_cluster_refs = {
     for ref, cfg in local.env_config.kafka_cluster_refs : ref => merge(cfg, {
@@ -30,6 +31,7 @@ terraform {
 inputs = {
   environment          = local.env_name
   topics               = local.kafka_topics
+  users                = local.kafka_users
   kafka_cluster_refs   = local.kafka_cluster_refs
   kafka_policy         = local.env_config.kafka_policy
   create_users         = true

modules/kafka-topics-yc/main.tf

@@ -20,46 +20,83 @@ locals {
 
       config = {
         for key, value in merge(
-          try(var.kafka_cluster_refs[topic.clusterRef].default_topic_config, {}),
+          try(topic.inheritDefaultConfig, true) ? try(var.kafka_cluster_refs[topic.clusterRef].default_topic_config, {}) : {},
           try(topic.config, {})
         ) : key => try(tostring(value[var.environment]), try(tostring(value._default), tostring(value)))
       }
 
       deletion_policy = try(topic.deletionPolicy, "orphan")
+
+      owner_roles = try(topic.ownerRoles, null) == null ? var.default_user_roles : topic.ownerRoles
     }
   }
 
-  owner_clusters = {
-    for item in distinct([
-      for topic in values(local.topics) : "${topic.cluster_ref}:${topic.owner}"
-      ]) : item => {
-      cluster_ref = split(":", item)[0]
-      owner       = split(":", item)[1]
-      cluster_id  = var.kafka_cluster_refs[split(":", item)[0]].cluster_id
-    }
+  user_grants = flatten([
+    for user in var.users : [
+      for permission in try(user.permissions, []) : [
+        for role in permission.roles : {
+          cluster_ref = user.clusterRef
+          user        = user.name
+          topic_name  = permission.topic
+          role        = role
         }
+      ]
+    ]
+  ])
 
-  owner_permissions = {
-    for owner_key, owner in local.owner_clusters : owner_key => flatten([
+  permission_tuples = concat(
+    flatten([
       for topic in values(local.topics) : [
-        for role in var.default_user_roles : {
+        for role in topic.owner_roles : {
+          cluster_ref = topic.cluster_ref
+          user        = topic.owner
           topic_name  = topic.name
           role        = role
         }
       ]
-      if topic.cluster_ref == owner.cluster_ref && topic.owner == owner.owner
-    ])
+    ]),
+    local.user_grants
+  )
+
+  user_clusters = {
+    for item in distinct(concat(
+      [
+        for tuple in local.permission_tuples : "${tuple.cluster_ref}:${tuple.user}"
+      ],
+      [
+        for user in var.users : "${user.clusterRef}:${user.name}"
+      ]
+      )) : item => {
+      cluster_ref = split(":", item)[0]
+      user        = split(":", item)[1]
+      cluster_id  = var.kafka_cluster_refs[split(":", item)[0]].cluster_id
     }
   }
 
+  user_permissions = {
+    for user_key, user in local.user_clusters : user_key => distinct([
+      for tuple in local.permission_tuples : {
+        topic_name = tuple.topic_name
+        role       = tuple.role
+      }
+      if "${tuple.cluster_ref}:${tuple.user}" == user_key
+    ])
+  }
+
+}
+
 resource "random_password" "kafka_user" {
-  for_each = var.create_users ? local.owner_clusters : {}
+  for_each = var.create_users ? local.user_clusters : {}
 
   length  = var.user_password_length
   special = false
   upper   = true
   lower   = true
   numeric = true
+
+  lifecycle {
+    ignore_changes = all
+  }
 }
 
 resource "yandex_mdb_kafka_topic" "this" {
@@ -104,14 +141,14 @@ resource "yandex_mdb_kafka_topic" "this" {
 }
 
 resource "yandex_mdb_kafka_user" "this" {
-  for_each = var.create_users ? local.owner_clusters : {}
+  for_each = var.create_users ? local.user_clusters : {}
 
   cluster_id = each.value.cluster_id
-  name       = each.value.owner
+  name       = each.value.user
   password   = random_password.kafka_user[each.key].result
 
   dynamic "permission" {
-    for_each = try(local.owner_permissions[each.key], [])
+    for_each = try(local.user_permissions[each.key], [])
     content {
       topic_name = permission.value.topic_name
       role       = permission.value.role
@@ -120,6 +157,7 @@ resource "yandex_mdb_kafka_user" "this" {
 
   lifecycle {
     prevent_destroy = true
+    ignore_changes  = [password]
   }
 
   depends_on = [yandex_mdb_kafka_topic.this]

modules/kafka-topics-yc/variables.tf

@@ -5,7 +5,31 @@ variable "environment" {
 
 variable "topics" {
   description = "Kafka topics declared in infrastructure.yaml under environments.<env>.kafka.topics."
-  type        = list(any)
+  type = list(object({
+    name                 = string
+    owner                = string
+    clusterRef           = string
+    partitions           = any
+    replicationFactor    = any
+    config               = optional(any, {})
+    inheritDefaultConfig = optional(bool, true)
+    deletionPolicy       = optional(string, "orphan")
+    ownerRoles           = optional(list(string))
+  }))
+  default = []
+}
+
+variable "users" {
+  description = "User-centric Kafka access declarations from infrastructure.yaml under environments.<env>.kafka.users."
+  type = list(object({
+    name          = string
+    clusterRef    = string
+    adoptExisting = optional(bool, false)
+    permissions = optional(list(object({
+      topic = string
+      roles = list(string)
+    })), [])
+  }))
   default = []
 }