Move kafka topics to infrastructure config

infrastructure.yaml

@@ -17,6 +17,9 @@ environments:
       allow_partition_increase: true
       allow_partition_decrease: false
 
+    kafka:
+      topics: []
+
     namespaces:
       - name: pulse
         labels:

live/prod/kafka-topics/terragrunt.hcl

@@ -14,14 +14,7 @@ locals {
   env_name     = basename(dirname(get_terragrunt_dir()))
   env_config   = local.infra_config.environments[local.env_name]
 
-  manifest_repo = get_env("TOPIC_MANIFEST_REPO", "local")
+  kafka_topics = try(local.env_config.kafka.topics, [])
-  manifest_path = get_env("TOPIC_MANIFEST_PATH", "")
-  manifest_slug = replace(replace(replace(replace("${local.manifest_repo}:${local.manifest_path}", "/", "__"), ":", "__"), ".", "_"), " ", "_")
-
-  topic_manifest_path = get_env(
-    "TOPIC_MANIFEST_LOCAL_PATH",
-    "${local.repo_root}/.generated/kafka-topics/${local.env_name}/manifest.yaml"
-  )
 
   kafka_cluster_refs = {
     for ref, cfg in local.env_config.kafka_cluster_refs : ref => merge(cfg, {
@@ -34,30 +27,9 @@ terraform {
   source = "${get_terragrunt_dir()}/../../../modules//kafka-topics-yc"
 }
 
-remote_state {
-  backend = "s3"
-  generate = {
-    path      = "backend.tf"
-    if_exists = "overwrite_terragrunt"
-  }
-  config = {
-    endpoint                    = "https://storage.yandexcloud.net"
-    bucket                      = get_env("S3_STATE_BUCKET", get_env("S3_PROD_STATE_BUCKET", "tfstate-terragrunt-prod"))
-    key                         = "live/${local.env_name}/kafka-topics/${local.manifest_slug}/terraform.tfstate"
-    region                      = "ru-central1"
-    access_key                  = get_env("S3_ACCESS_KEY", get_env("AWS_ACCESS_KEY_ID", ""))
-    secret_key                  = get_env("S3_SECRET_KEY", get_env("AWS_SECRET_ACCESS_KEY", ""))
-    skip_region_validation      = true
-    skip_credentials_validation = true
-    skip_metadata_api_check     = true
-    force_path_style            = true
-    skip_bucket_root_access     = true
-  }
-}
-
 inputs = {
   environment          = local.env_name
-  topic_manifest_path  = local.topic_manifest_path
+  topics               = local.kafka_topics
   kafka_cluster_refs   = local.kafka_cluster_refs
   kafka_policy         = local.env_config.kafka_policy
   create_users         = true

modules/kafka-topics-yc/main.tf

@@ -1,7 +1,5 @@
 locals {
-  manifest = yamldecode(file(var.topic_manifest_path))
+  raw_topics = var.topics
-
-  raw_topics = try(local.manifest.kafka.topics, [])
 
   topics = {
     for topic in local.raw_topics : topic.name => {

modules/kafka-topics-yc/variables.tf

@@ -3,9 +3,10 @@ variable "environment" {
   type        = string
 }
 
-variable "topic_manifest_path" {
+variable "topics" {
-  description = "Local path to service-owned Kafka topics manifest."
+  description = "Kafka topics declared in infrastructure.yaml under environments.<env>.kafka.topics."
-  type        = string
+  type        = list(any)
+  default     = []
 }
 
 variable "kafka_cluster_refs" {

scripts/fetch-topic-manifest.sh

@@ -1,52 +0,0 @@
-#!/usr/bin/env bash
-set -euo pipefail
-
-env_name="${TOPIC_ENV:-${ENVIRONMENT:-stage}}"
-repo_root="$(git rev-parse --show-toplevel 2>/dev/null || pwd)"
-output_path="${TOPIC_MANIFEST_LOCAL_PATH:-${repo_root}/.generated/kafka-topics/${env_name}/manifest.yaml}"
-
-mkdir -p "$(dirname "${output_path}")"
-
-if [[ -z "${TOPIC_MANIFEST_PATH:-}" ]]; then
-  echo "TOPIC_MANIFEST_PATH is required for kafka-topics component" >&2
-  exit 1
-fi
-
-if [[ -z "${TOPIC_MANIFEST_REPO:-}" ]]; then
-  if [[ -f "${TOPIC_MANIFEST_PATH}" ]]; then
-    cp "${TOPIC_MANIFEST_PATH}" "${output_path}"
-    echo "Copied local topic manifest to ${output_path}"
-    exit 0
-  fi
-
-  echo "TOPIC_MANIFEST_REPO is required when ${TOPIC_MANIFEST_PATH} is not local" >&2
-  exit 1
-fi
-
-if [[ -z "${TOPIC_MANIFEST_SHA:-}" ]]; then
-  echo "TOPIC_MANIFEST_SHA is required" >&2
-  exit 1
-fi
-
-server_url="${CI_SERVER_URL:-https://gitlab.sarex.io}"
-encoded_path="${TOPIC_MANIFEST_PATH#/}"
-urlencode() {
-  python3 -c 'import sys, urllib.parse; print(urllib.parse.quote(sys.argv[1], safe=""))' "$1"
-}
-
-encoded_project="$(urlencode "${TOPIC_MANIFEST_REPO}")"
-encoded_file="$(urlencode "${encoded_path}")"
-encoded_ref="$(urlencode "${TOPIC_MANIFEST_SHA}")"
-manifest_url="${server_url}/api/v4/projects/${encoded_project}/repository/files/${encoded_file}/raw?ref=${encoded_ref}"
-
-curl_args=(-fsSL)
-if [[ -n "${CI_JOB_TOKEN:-}" ]]; then
-  curl_args+=(-H "JOB-TOKEN: ${CI_JOB_TOKEN}")
-fi
-
-curl "${curl_args[@]}" "${manifest_url}" -o "${output_path}"
-if grep -qiE '^\s*<!doctype html|^\s*<html' "${output_path}"; then
-  echo "Downloaded topic manifest is HTML, not YAML. Check GitLab API auth for ${TOPIC_MANIFEST_REPO}/${encoded_path}@${TOPIC_MANIFEST_SHA}" >&2
-  exit 1
-fi
-echo "Downloaded topic manifest to ${output_path}"

scripts/generate-pipeline.sh

@@ -58,7 +58,6 @@ find live -name "terragrunt.hcl" -not -path "*/.terragrunt-cache/*" | sort | whi
     *) folder_var="YC_FOLDER_ID" ;;
   esac
 
-  # Определяем тег runner'а и правила в зависимости от окружения
   if [ "$env" = "prod" ]; then
     runner_tags="  tags:
     - prod"
@@ -71,17 +70,16 @@ find live -name "terragrunt.hcl" -not -path "*/.terragrunt-cache/*" | sort | whi
     - if: '\$CI_PIPELINE_SOURCE == \"pipeline\"'
     - if: '\$CI_PIPELINE_SOURCE == \"merge_request_event\"'
     - if: '\$CI_COMMIT_BRANCH == \"stage\"'
-    - if: '\$CI_COMMIT_BRANCH =~ /^feature\\\\/.*/'
+    - if: '\$CI_COMMIT_BRANCH =~ /^feature\\/.*/'
     - if: '\$CI_PIPELINE_SOURCE == \"pipeline\" && \$TARGET_ENV == \"stage\"'"
   fi
 
   if [ "$component" = "kafka-topics" ]; then
-    validate_before_script="    - apk add --no-cache bash curl python3 py3-yaml >/dev/null 2>&1 || true; ./scripts/fetch-topic-manifest.sh; ./scripts/validate-kafka-topics.sh; cd \$TG_ROOT && echo \"Working directory:\" && pwd && ls -la"
+    validate_before_script="    - apk add --no-cache bash python3 py3-yaml >/dev/null 2>&1 || true
-    branch_rules="    - if: '\$TF_COMPONENT == \"kafka-topics\" && \$TOPIC_MANIFEST_PATH != \"\"'"
+    - ./scripts/validate-kafka-topics.sh
-  else
+    - cd \$TG_ROOT
-    branch_rules="    - if: '\$TF_COMPONENT == \"kafka-topics\" && \$TOPIC_MANIFEST_PATH != \"\"'
+    - echo \"Working directory:\" && pwd
-      when: never
+    - ls -la"
-${branch_rules}"
   fi
 
   cat >> "$OUTPUT_FILE" << VALIDATE_JOB
@@ -111,19 +109,19 @@ ${needs_section}
 
 VALIDATE_JOB
 
-  # Plan needs и before_script для secrets
   plan_needs="validate-${job_prefix}"
-  plan_before_script="cd \$TG_ROOT && ls -la"
+  plan_before_script="    - cd \$TG_ROOT && ls -la"
   if [ "$component" = "secrets" ]; then
     plan_needs="validate-${job_prefix}
     - plan-${env}-namespace
     - plan-${env}-s3
     - plan-${env}-database"
-    # Для secrets инициализируем зависимости чтобы получить их outputs
+    plan_before_script="    - cd live/${env}/namespace && terragrunt init -upgrade -reconfigure && cd ../s3 && terragrunt init -upgrade -reconfigure && cd ../database && terragrunt init -upgrade -reconfigure && cd ../secrets && ls -la"
-    plan_before_script="cd live/${env}/namespace && terragrunt init -upgrade -reconfigure && cd ../s3 && terragrunt init -upgrade -reconfigure && cd ../database && terragrunt init -upgrade -reconfigure && cd ../secrets && ls -la"
   fi
   if [ "$component" = "kafka-topics" ]; then
-    plan_before_script="apk add --no-cache bash curl python3 py3-yaml >/dev/null 2>&1 || true; ./scripts/fetch-topic-manifest.sh; ./scripts/validate-kafka-topics.sh; cd \$TG_ROOT && ls -la"
+    plan_before_script="    - apk add --no-cache bash python3 py3-yaml >/dev/null 2>&1 || true
+    - ./scripts/validate-kafka-topics.sh
+    - cd \$TG_ROOT && ls -la"
   fi
 
   cat >> "$OUTPUT_FILE" << PLAN_JOB
@@ -138,7 +136,7 @@ ${runner_tags}
   needs:
     - ${plan_needs}
   before_script:
-    - ${plan_before_script}
+${plan_before_script}
   script:
     - terragrunt init -upgrade -reconfigure
     - terragrunt plan -out=tfplan
@@ -154,39 +152,30 @@ ${branch_rules}
 
 PLAN_JOB
 
-  # Apply всегда manual
   when_clause="manual"
 
   apply_needs="plan-${job_prefix}"
-  apply_before_script="cd \$TG_ROOT && ls -la"
+  apply_before_script="    - cd \$TG_ROOT && ls -la"
   if [ "$component" = "secrets" ]; then
     apply_needs="plan-${job_prefix}
     - apply-${env}-namespace
     - apply-${env}-s3
     - apply-${env}-database"
-    # Для secrets инициализируем зависимости чтобы получить их outputs
+    apply_before_script="    - cd live/${env}/namespace && terragrunt init -upgrade -reconfigure && cd ../s3 && terragrunt init -upgrade -reconfigure && cd ../database && terragrunt init -upgrade -reconfigure && cd ../secrets && ls -la"
-    apply_before_script="cd live/${env}/namespace && terragrunt init -upgrade -reconfigure && cd ../s3 && terragrunt init -upgrade -reconfigure && cd ../database && terragrunt init -upgrade -reconfigure && cd ../secrets && ls -la"
+  fi
+  if [ "$component" = "kafka-topics" ]; then
+    apply_before_script="    - apk add --no-cache bash python3 py3-yaml >/dev/null 2>&1 || true
+    - ./scripts/validate-kafka-topics.sh
+    - cd \$TG_ROOT && ls -la"
   fi
 
-  # Apply rules в зависимости от окружения
+  if [ "$env" = "prod" ]; then
-  if [ "$component" = "kafka-topics" ]; then
+    apply_rules="    - if: '\$CI_COMMIT_BRANCH == \"master\"'
-    apply_before_script="apk add --no-cache bash curl python3 py3-yaml >/dev/null 2>&1 || true; ./scripts/fetch-topic-manifest.sh; ./scripts/validate-kafka-topics.sh; cd \$TG_ROOT && ls -la"
-    apply_rules="    - if: '\$TF_COMPONENT == \"kafka-topics\" && \$TOPIC_MANIFEST_PATH != \"\" && \$TOPIC_APPLY == \"false\"'
-      when: never
-    - if: '\$TF_COMPONENT == \"kafka-topics\" && \$TOPIC_MANIFEST_PATH != \"\" && \$TOPIC_APPLY != \"false\"'
-      when: on_success
-    - when: never"
-  elif [ "$env" = "prod" ]; then
-    apply_rules="    - if: '\$TF_COMPONENT == \"kafka-topics\" && \$TOPIC_MANIFEST_PATH != \"\"'
-      when: never
-    - if: '\$CI_COMMIT_BRANCH == \"master\"'
       when: ${when_clause}
     - if: '\$CI_PIPELINE_SOURCE == \"pipeline\" && \$TARGET_ENV == \"prod\"'
       when: ${when_clause}"
   else
-    apply_rules="    - if: '\$TF_COMPONENT == \"kafka-topics\" && \$TOPIC_MANIFEST_PATH != \"\"'
+    apply_rules="    - if: '\$CI_COMMIT_BRANCH == \"stage\"'
-      when: never
-    - if: '\$CI_COMMIT_BRANCH == \"stage\"'
       when: ${when_clause}
     - if: '\$CI_PIPELINE_SOURCE == \"pipeline\" && \$TARGET_ENV == \"stage\"'
       when: ${when_clause}"
@@ -204,7 +193,7 @@ ${runner_tags}
   needs:
     - ${apply_needs}
   before_script:
-    - ${apply_before_script}
+${apply_before_script}
   script:
     - terragrunt init -upgrade -reconfigure
     - terragrunt apply -auto-approve

scripts/validate-kafka-topics.sh

@@ -2,11 +2,10 @@
 set -euo pipefail
 
 repo_root="$(git rev-parse --show-toplevel 2>/dev/null || pwd)"
-env_name="${TOPIC_ENV:-${ENVIRONMENT:-stage}}"
+env_name="${ENVIRONMENT:-$(basename "$(dirname "${TG_ROOT:-live/prod/kafka-topics}")")}"
-manifest_path="${TOPIC_MANIFEST_LOCAL_PATH:-${repo_root}/.generated/kafka-topics/${env_name}/manifest.yaml}"
 infra_path="${INFRASTRUCTURE_CONFIG_PATH:-${repo_root}/infrastructure.yaml}"
 
-python3 - "${manifest_path}" "${infra_path}" "${env_name}" <<'PY'
+python3 - "${infra_path}" "${env_name}" <<'PY'
 import os
 import sys
 
@@ -15,7 +14,7 @@ try:
 except Exception as exc:
     raise SystemExit(f"pyyaml is required for Kafka topics validation: {exc}")
 
-manifest_path, infra_path, env = sys.argv[1:4]
+infra_path, env = sys.argv[1:3]
 
 def load_yaml(path):
     if not os.path.exists(path):
@@ -34,7 +33,6 @@ def env_value(value, env, default=None):
         return default
     return value
 
-manifest = load_yaml(manifest_path)
 infra = load_yaml(infra_path)
 
 env_cfg = (infra.get("environments") or {}).get(env)
@@ -43,17 +41,20 @@ if not isinstance(env_cfg, dict):
 
 cluster_refs = env_cfg.get("kafka_cluster_refs") or {}
 policy = env_cfg.get("kafka_policy") or {}
-topics = ((manifest.get("kafka") or {}).get("topics"))
+topics = ((env_cfg.get("kafka") or {}).get("topics"))
 
+if topics is None:
+    topics = []
 if not isinstance(topics, list):
-    raise SystemExit("kafka.topics must be an array")
+    raise SystemExit(f"environments.{env}.kafka.topics must be an array")
 
 allowed_cleanup = {"delete", "compact"}
 allowed_deletion = {"orphan", "delete"}
 errors = []
+seen_names = set()
 
 for idx, topic in enumerate(topics):
-    prefix = f"kafka.topics[{idx}]"
+    prefix = f"environments.{env}.kafka.topics[{idx}]"
     if not isinstance(topic, dict):
         errors.append(f"{prefix}: must be an object")
         continue
@@ -64,6 +65,11 @@ for idx, topic in enumerate(topics):
 
     if not name:
         errors.append(f"{prefix}.name is required")
+    elif name in seen_names:
+        errors.append(f"{prefix}.name '{name}' is duplicated")
+    else:
+        seen_names.add(name)
+
     if not owner:
         errors.append(f"{prefix}.owner is required")
     if not cluster_ref:
@@ -127,5 +133,5 @@ for idx, topic in enumerate(topics):
 if errors:
     raise SystemExit("Kafka topics validation failed:\n- " + "\n- ".join(errors))
 
-print(f"Kafka topics manifest is valid: {manifest_path}")
+print(f"Kafka topics declaration is valid: {infra_path} env={env} topics={len(topics)}")
 PY