Playbook：AWS MSK Serverless 迁回 Provisioned——什么时候、为什么、怎么迁

#!/bin/bash
# msk-cost-pull.sh - 用 Cost Explorer 拉过去 30 天 MSK 账单按 usage type 分桶
# 用法：./msk-cost-pull.sh [region]
# 前置：aws cli v2，IAM 拥有 ce:GetCostAndUsage 权限

set -euo pipefail

REGION="${1:-ap-southeast-1}"
END=$(date -u +%Y-%m-%d)
START=$(date -u -d '30 days ago' +%Y-%m-%d)

command -v jq >/dev/null || { echo "需要 jq"; exit 1; }

aws ce get-cost-and-usage \
  --time-period "Start=${START},End=${END}" \
  --granularity DAILY \
  --metrics "UnblendedCost" \
  --filter '{"And":[{"Dimensions":{"Key":"SERVICE","Values":["Amazon Managed Streaming for Apache Kafka"]}},{"Dimensions":{"Key":"REGION","Values":["'"${REGION}"'"]}}]}' \
  --group-by Type=DIMENSION,Key=USAGE_TYPE \
  --output json \
  | jq -r '
      .ResultsByTime
      | map(.Groups[])
      | group_by(.Keys[0])
      | map({usage_type: .[0].Keys[0], cost_usd: (map(.Metrics.UnblendedCost.Amount | tonumber) | add | .*100 | round / 100)})
      | sort_by(-.cost_usd)
      | (["usage_type","cost_usd"] | @tsv),
        (.[] | [.usage_type, .cost_usd] | @tsv)
    ' \
  | column -t -s $'\t'

usage_type                            cost_usd
APS1-Kafka-Serverless-Hours           540.00
APS1-Kafka-Serverless-WriteThroughput 0.18
APS1-Kafka-Serverless-Storage         0.04

#!/usr/bin/env python3
# msk_estimator.py - Serverless vs Provisioned 月成本估算
# 用法：python3 msk_estimator.py --brokers 2 --instance kafka.t3.small --ebs-gb 100

import argparse

# 价格参考（2026-04，ap-southeast-1，不含税）
HOUR = 730
BROKER_HOURLY = {
    "kafka.t3.small":   0.0456,
    "kafka.m5.large":   0.21,
    "kafka.m5.xlarge":  0.42,
    "kafka.m5.2xlarge": 0.84,
}
EBS_GP3_PER_GB_MONTH = 0.0928
SERVERLESS_CLUSTER_HOURLY = 0.75
SERVERLESS_INGRESS_PER_GB = 0.10
SERVERLESS_EGRESS_PER_GB  = 0.05
SERVERLESS_STORAGE_GB_MO  = 0.10
SERVERLESS_PARTITION_HOUR = 0.0015

def parse_args():
    p = argparse.ArgumentParser()
    p.add_argument("--brokers", type=int, default=2)
    p.add_argument("--instance", default="kafka.t3.small")
    p.add_argument("--ebs-gb", type=int, default=100)
    p.add_argument("--ingress-gb-mo", type=float, default=1.0)
    p.add_argument("--egress-gb-mo",  type=float, default=1.0)
    p.add_argument("--storage-gb",    type=float, default=2.0)
    p.add_argument("--partitions",    type=int,   default=6)
    return p.parse_args()

def provisioned(args):
    if args.instance not in BROKER_HOURLY:
        raise SystemExit(f"unknown instance {args.instance}")
    broker = BROKER_HOURLY[args.instance] * HOUR * args.brokers
    ebs    = args.ebs_gb * EBS_GP3_PER_GB_MONTH * args.brokers
    return {"broker": broker, "ebs": ebs, "total": broker + ebs}

def serverless(args):
    cluster   = SERVERLESS_CLUSTER_HOURLY * HOUR
    ingress   = args.ingress_gb_mo * SERVERLESS_INGRESS_PER_GB
    egress    = args.egress_gb_mo  * SERVERLESS_EGRESS_PER_GB
    storage   = args.storage_gb    * SERVERLESS_STORAGE_GB_MO
    partition = args.partitions    * SERVERLESS_PARTITION_HOUR * HOUR
    return {
        "cluster": cluster, "ingress": ingress, "egress": egress,
        "storage": storage, "partition": partition,
        "total": cluster + ingress + egress + storage + partition,
    }

def main():
    args = parse_args()
    p = provisioned(args)
    s = serverless(args)
    print(f"Provisioned ({args.brokers}x {args.instance}, {args.ebs_gb}GB EBS):")
    for k, v in p.items():
        print(f"  {k:10s} ${v:7.2f}")
    print(f"\nServerless (ingress {args.ingress_gb_mo}GB, partitions {args.partitions}):")
    for k, v in s.items():
        print(f"  {k:10s} ${v:7.2f}")
    delta = s["total"] - p["total"]
    pct   = delta / s["total"] * 100 if s["total"] else 0
    print(f"\n月节省 ${delta:.2f} ({pct:.1f}%)")

if __name__ == "__main__":
    main()

broker 数 ≥ ceil(峰值吞吐 MB/s / 单 broker 可承载 MB/s) × replication_factor / partition_per_broker
单 broker 可承载 ≈ 实例规格 × 0.6（留 burst 余量）
partition 数 ≥ max(消费者副本数, broker 数 × 2, 峰值吞吐 / 5 MB/s)

{
  "ClusterName": "svc-foo-qa-kafka-v2",
  "KafkaVersion": "2.8.1",
  "NumberOfBrokerNodes": 2,
  "BrokerNodeGroupInfo": {
    "InstanceType": "kafka.t3.small",
    "ClientSubnets": [
      "subnet-aaaaaaaaaaaaaaaaa",
      "subnet-bbbbbbbbbbbbbbbbb"
    ],
    "SecurityGroups": [
      "sg-xxxxxxxxxxxxxxxxx"
    ],
    "StorageInfo": {
      "EbsStorageInfo": {
        "VolumeSize": 100,
        "ProvisionedThroughput": {
          "Enabled": false
        }
      }
    },
    "ConnectivityInfo": {
      "PublicAccess": {
        "Type": "DISABLED"
      }
    }
  },
  "ClientAuthentication": {
    "Sasl": {
      "Iam": { "Enabled": true }
    },
    "Unauthenticated": { "Enabled": false }
  },
  "EncryptionInfo": {
    "EncryptionAtRest": {
      "DataVolumeKMSKeyId": "alias/aws/kafka"
    },
    "EncryptionInTransit": {
      "ClientBroker": "TLS",
      "InCluster": true
    }
  },
  "EnhancedMonitoring": "PER_TOPIC_PER_BROKER",
  "OpenMonitoring": {
    "Prometheus": {
      "JmxExporter":  { "EnabledInBroker": true },
      "NodeExporter": { "EnabledInBroker": true }
    }
  },
  "LoggingInfo": {
    "BrokerLogs": {
      "CloudWatchLogs": {
        "Enabled": true,
        "LogGroup": "/aws/msk/svc-foo-qa-kafka-v2"
      }
    }
  },
  "Tags": {
    "Environment": "qa",
    "Service": "svc-foo",
    "ManagedBy": "playbook"
  }
}

#!/bin/bash
# msk-create.sh - 创建 Provisioned 集群并等待 ACTIVE
# 用法：./msk-create.sh cluster-config.json [region]
# 前置：kafka:CreateCluster / DescribeCluster 权限；ClientSubnets 至少 2 个不同 AZ

set -euo pipefail

CFG="${1:?cluster-config.json}"
REGION="${2:-ap-southeast-1}"

command -v jq >/dev/null || { echo "需要 jq"; exit 1; }

ARN=$(aws kafka create-cluster --cli-input-json "file://${CFG}" --region "${REGION}" \
        --query 'ClusterArn' --output text)
echo "已创建 ClusterArn=${ARN}"

while true; do
  STATE=$(aws kafka describe-cluster --cluster-arn "${ARN}" --region "${REGION}" \
            --query 'ClusterInfo.State' --output text)
  echo "$(date -u +%H:%M:%S) state=${STATE}"
  case "${STATE}" in
    ACTIVE)  break ;;
    FAILED)  echo "创建失败"; exit 2 ;;
    CREATING|UPDATING) sleep 30 ;;
    *)       echo "未知状态 ${STATE}"; exit 3 ;;
  esac
done

aws kafka get-bootstrap-brokers --cluster-arn "${ARN}" --region "${REGION}" \
  --query 'BootstrapBrokerStringSaslIam' --output text

已创建 ClusterArn=arn:aws:kafka:ap-southeast-1:<ACCOUNT_ID>:cluster/svc-foo-qa-kafka-v2/...
14:02:11 state=CREATING
...（约 18-25 分钟）...
14:21:43 state=ACTIVE
b-1.svcfooqakafkav2.xxxxxx.c4.kafka.ap-southeast-1.amazonaws.com:9098,b-2...

---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: kafka-admin
  namespace: svc-foo
  annotations:
    eks.amazonaws.com/role-arn: arn:aws:iam::<ACCOUNT_ID>:role/svc-foo-msk-admin
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: kafka-producer
  namespace: svc-foo
  annotations:
    eks.amazonaws.com/role-arn: arn:aws:iam::<ACCOUNT_ID>:role/svc-foo-msk-producer
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: kafka-consumer
  namespace: svc-foo
  annotations:
    eks.amazonaws.com/role-arn: arn:aws:iam::<ACCOUNT_ID>:role/svc-foo-msk-consumer

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Federated": "arn:aws:iam::<ACCOUNT_ID>:oidc-provider/oidc.eks.ap-southeast-1.amazonaws.com/id/<OIDC_ID>"
      },
      "Action": "sts:AssumeRoleWithWebIdentity",
      "Condition": {
        "StringEquals": {
          "oidc.eks.ap-southeast-1.amazonaws.com/id/<OIDC_ID>:sub": "system:serviceaccount:svc-foo:kafka-admin",
          "oidc.eks.ap-southeast-1.amazonaws.com/id/<OIDC_ID>:aud": "sts.amazonaws.com"
        }
      }
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "kafka-cluster:Connect",
        "kafka-cluster:DescribeCluster",
        "kafka-cluster:AlterCluster",
        "kafka-cluster:DescribeTopic",
        "kafka-cluster:CreateTopic",
        "kafka-cluster:DeleteTopic",
        "kafka-cluster:AlterTopic",
        "kafka-cluster:DescribeTopicDynamicConfiguration",
        "kafka-cluster:AlterTopicDynamicConfiguration",
        "kafka-cluster:DescribeGroup",
        "kafka-cluster:AlterGroup",
        "kafka-cluster:DeleteGroup"
      ],
      "Resource": [
        "arn:aws:kafka:ap-southeast-1:<ACCOUNT_ID>:cluster/svc-foo-qa-kafka-v2/*",
        "arn:aws:kafka:ap-southeast-1:<ACCOUNT_ID>:topic/svc-foo-qa-kafka-v2/*/*",
        "arn:aws:kafka:ap-southeast-1:<ACCOUNT_ID>:group/svc-foo-qa-kafka-v2/*/*"
      ]
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["kafka-cluster:Connect"],
      "Resource": "arn:aws:kafka:ap-southeast-1:<ACCOUNT_ID>:cluster/svc-foo-qa-kafka-v2/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "kafka-cluster:WriteData",
        "kafka-cluster:DescribeTopic",
        "kafka-cluster:WriteDataIdempotently"
      ],
      "Resource": "arn:aws:kafka:ap-southeast-1:<ACCOUNT_ID>:topic/svc-foo-qa-kafka-v2/*/message-*"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["kafka-cluster:Connect"],
      "Resource": "arn:aws:kafka:ap-southeast-1:<ACCOUNT_ID>:cluster/svc-foo-qa-kafka-v2/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "kafka-cluster:ReadData",
        "kafka-cluster:DescribeTopic"
      ],
      "Resource": "arn:aws:kafka:ap-southeast-1:<ACCOUNT_ID>:topic/svc-foo-qa-kafka-v2/*/message-*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "kafka-cluster:DescribeGroup",
        "kafka-cluster:AlterGroup"
      ],
      "Resource": "arn:aws:kafka:ap-southeast-1:<ACCOUNT_ID>:group/svc-foo-qa-kafka-v2/*/*"
    }
  ]
}

# 列出 ns 内所有 SA → role 映射，确认 3 个 SA 都注解到位
kubectl -n svc-foo get sa -o json | python3 -c "
import json, sys
for i in json.load(sys.stdin)['items']:
    name = i['metadata']['name']
    role = i['metadata'].get('annotations', {}).get('eks.amazonaws.com/role-arn', 'NONE')
    if role != 'NONE':
        print(f'{name}: {role}')
"
# 起一个临时 Pod 用 admin SA 试连，能 list 即通
kubectl -n svc-foo run iam-check --rm -it --restart=Never \
  --image=amazon/aws-cli:2.15.0 \
  --overrides='{"spec":{"serviceAccountName":"kafka-admin"}}' \
  -- sts get-caller-identity

#!/bin/bash
# msk-create-topic.sh - 在新集群上建 topic
# 用法：./msk-create-topic.sh <new-bootstrap> <topic> <partitions>
# 前置：当前 kubectl context 指向迁移目标 EKS

set -euo pipefail
NEW_BS="${1:?bootstrap}"
TOPIC="${2:?topic}"
PART="${3:?partitions}"

kubectl -n svc-foo run kafka-cli --image=confluentinc/cp-kafka:7.5.0 \
  --restart=Never \
  --overrides='{"spec":{"serviceAccountName":"kafka-admin"}}' \
  --command -- sleep 600

trap 'kubectl -n svc-foo delete pod kafka-cli --ignore-not-found' EXIT

kubectl -n svc-foo wait --for=condition=Ready pod/kafka-cli --timeout=60s

kubectl -n svc-foo exec kafka-cli -- bash -c "
  set -e
  curl -sL -o /tmp/iam.jar \
    https://github.com/aws/aws-msk-iam-auth/releases/download/v2.2.0/aws-msk-iam-auth-2.2.0-all.jar
  export CLASSPATH=/tmp/iam.jar
  cat > /tmp/client.properties <<'EOF'
security.protocol=SASL_SSL
sasl.mechanism=AWS_MSK_IAM
sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required;
sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler
EOF
  kafka-topics --bootstrap-server '${NEW_BS}' --command-config /tmp/client.properties \
    --create --topic '${TOPIC}' --partitions '${PART}' --replication-factor 2 \
    --config retention.ms=604800000 --config min.insync.replicas=1
  kafka-topics --bootstrap-server '${NEW_BS}' --command-config /tmp/client.properties --describe --topic '${TOPIC}'
"

spring:
  kafka:
    bootstrap-servers:
      - b-1.svcfooqakafkav2.xxxxxx.c4.kafka.ap-southeast-1.amazonaws.com:9098
      - b-2.svcfooqakafkav2.xxxxxx.c4.kafka.ap-southeast-1.amazonaws.com:9098
    properties:
      security.protocol: SASL_SSL
      sasl.mechanism: AWS_MSK_IAM
      sasl.jaas.config: software.amazon.msk.auth.iam.IAMLoginModule required;
      sasl.client.callback.handler.class: software.amazon.msk.auth.iam.IAMClientCallbackHandler
    consumer:
      group-id: svc-foo-consumer
      auto-offset-reset: earliest
      enable-auto-commit: false
    producer:
      acks: all
      retries: 5
      properties:
        enable.idempotence: true

<dependency>
  <groupId>software.amazon.msk</groupId>
  <artifactId>aws-msk-iam-auth</artifactId>
  <version>2.2.0</version>
</dependency>

public class MskIamCallbackHandler implements AuthenticateCallbackHandler {
    private String region;

    @Override
    public void configure(Map<String, ?> cfg, String mechanism, List<AppConfigurationEntry> jaas) {
        this.region = (String) cfg.getOrDefault(AWSConfigConstants.AWS_REGION, "ap-southeast-1");
    }

    @Override
    public void handle(Callback[] callbacks) throws IOException {
        for (Callback cb : callbacks) {
            if (cb instanceof OAuthBearerTokenCallback) {
                String token = MSKAuthTokenProvider.generateAuthToken(Region.of(region));
                ((OAuthBearerTokenCallback) cb).token(new BasicOAuthBearerToken(token, ...));
            }
        }
    }

    @Override public void close() {}
}

package kafkaclient

import (
	"context"
	"strings"
	"time"

	"github.com/IBM/sarama"
	"github.com/aws/aws-msk-iam-sasl-signer-go/signer"
)

type mskTokenProvider struct {
	region string
}

func (m *mskTokenProvider) Token() (*sarama.AccessToken, error) {
	token, _, err := signer.GenerateAuthToken(context.Background(), m.region)
	if err != nil {
		return nil, err
	}
	return &sarama.AccessToken{Token: token}, nil
}

// NewConfig 注意：brokers 必须是逗号分隔字符串切分后的 []string，不要把整段塞进一个元素
func NewConfig(brokersCSV, region string) (sarama.Client, error) {
	brokers := strings.Split(brokersCSV, ",")
	for i := range brokers {
		brokers[i] = strings.TrimSpace(brokers[i])
	}

	cfg := sarama.NewConfig()
	cfg.Version = sarama.V2_8_1_0
	cfg.Net.SASL.Enable = true
	cfg.Net.SASL.Mechanism = sarama.SASLTypeOAuth
	cfg.Net.SASL.TokenProvider = &mskTokenProvider{region: region}
	cfg.Net.TLS.Enable = true
	cfg.Net.DialTimeout = 10 * time.Second
	cfg.Producer.RequiredAcks = sarama.WaitForAll
	cfg.Producer.Idempotent = true
	cfg.Producer.Retry.Max = 5
	cfg.Net.MaxOpenRequests = 1

	return sarama.NewClient(brokers, cfg)
}

# kafka_client.py
import socket
from typing import Tuple

from aws_msk_iam_sasl_signer import MSKAuthTokenProvider
from confluent_kafka import Consumer, Producer

REGION = "ap-southeast-1"

def oauth_cb(_oauth_config) -> Tuple[str, float]:
    token, expiry_ms = MSKAuthTokenProvider.generate_auth_token(REGION)
    # confluent-kafka 期望返回 (token, expiry_seconds_since_epoch)
    return token, expiry_ms / 1000.0

def make_producer(bootstrap: str) -> Producer:
    return Producer({
        "bootstrap.servers": bootstrap,
        "security.protocol": "SASL_SSL",
        "sasl.mechanism":    "OAUTHBEARER",
        "oauth_cb":          oauth_cb,
        "client.id":         socket.gethostname(),
        "enable.idempotence": True,
        "acks": "all",
        "compression.type": "lz4",
    })

def make_consumer(bootstrap: str, group_id: str) -> Consumer:
    return Consumer({
        "bootstrap.servers": bootstrap,
        "security.protocol": "SASL_SSL",
        "sasl.mechanism":    "OAUTHBEARER",
        "oauth_cb":          oauth_cb,
        "group.id":          group_id,
        "auto.offset.reset": "earliest",
        "enable.auto.commit": False,
    })

# Nacos publish 新增字段
kafka.bootstrap_servers_new=<NEW_BS>
kafka.dual_consume=true

# rollout consumer
kubectl -n svc-foo rollout restart sts/consumer
kubectl -n svc-foo rollout status sts/consumer --timeout=10m

# 旧集群 lag 正常（offset 增长）
kafka-consumer-groups --bootstrap-server $OLD_BS --command-config /tmp/old.properties \
  --describe --group svc-foo-consumer

# 新集群 lag = 0（暂时没消息）
kafka-consumer-groups --bootstrap-server $NEW_BS --command-config /tmp/new.properties \
  --describe --group svc-foo-consumer

# producer dual write 简化示意
def send(payload: bytes, key: bytes):
    fut_old = old_producer.produce(TOPIC, value=payload, key=key)
    fut_new = new_producer.produce(TOPIC, value=payload, key=key)
    old_producer.poll(0)
    new_producer.poll(0)
    # 任意一边成功即视为成功，避免拖慢主流程

# Nacos
kafka.bootstrap_servers_new=<NEW_BS>
kafka.dual_produce=true
kubectl -n svc-foo rollout restart deploy/producer

# 新集群 BytesInPerSec 起来
aws cloudwatch get-metric-statistics --namespace AWS/Kafka \
  --metric-name BytesInPerSec --dimensions \
  Name="Cluster Name",Value=svc-foo-qa-kafka-v2 \
  Name="Broker ID",Value=1 \
  --start-time $(date -u -d '5 minutes ago' +%FT%TZ) \
  --end-time $(date -u +%FT%TZ) \
  --period 60 --statistics Sum --region ap-southeast-1

# 新集群消费 lag 应该开始增长，说明消息进得来

# Nacos
kafka.dual_consume=false
kafka.bootstrap_servers=<NEW_BS>     # 主 endpoint 切到新集群
kubectl -n svc-foo rollout restart sts/consumer

# 旧集群消费 lag 暂停增长但 offset 不动（因为没人订阅了），约 30 分钟后过期消失
kafka-consumer-groups --bootstrap-server $OLD_BS --command-config /tmp/old.properties \
  --describe --group svc-foo-consumer
# 输出 has no active members 是正常的

# Nacos
kafka.dual_produce=false
kafka.bootstrap_servers=<NEW_BS>
kubectl -n svc-foo rollout restart deploy/producer

# 旧集群 BytesInPerSec 应该 30 分钟内归零
aws cloudwatch get-metric-statistics --namespace AWS/Kafka \
  --metric-name BytesInPerSec --dimensions \
  Name="Cluster Name",Value=svc-foo-qa-kafka-old \
  --start-time $(date -u -d '30 minutes ago' +%FT%TZ) \
  --end-time $(date -u +%FT%TZ) \
  --period 60 --statistics Sum --region ap-southeast-1

# 应用日志中无 KafkaConnectionError、SASL handshake failed
kubectl -n svc-foo logs -l app=producer --tail=200 | grep -iE 'kafka|sasl|error' || echo OK

# 检查任何 pod 是否还在连旧 broker
kubectl -n svc-foo exec deploy/producer -- bash -c \
  'cat /proc/net/tcp | awk "{print \$3}" | sort -u' | python3 -c '
import sys
for line in sys.stdin:
    line=line.strip()
    if not line or line=="local_address": continue
    ip = ".".join(str(int(line.split(":")[0][i:i+2],16)) for i in (6,4,2,0))
    print(ip)
' | sort -u

# 删除旧集群
aws kafka delete-cluster --cluster-arn $OLD_ARN --region ap-southeast-1

#!/bin/bash
# schema-export.sh - 把一个 Glue Registry 下所有 schema 导出到本地 JSON
# 用法：./schema-export.sh <registry-name> <out-dir> [region]

set -euo pipefail
REG="${1:?registry}"
OUT="${2:?out-dir}"
REGION="${3:-ap-southeast-1}"

mkdir -p "${OUT}"
command -v jq >/dev/null || { echo "需要 jq"; exit 1; }

aws glue list-schemas --registry-id RegistryName="${REG}" --region "${REGION}" \
    --max-results 100 \
    --query 'Schemas[].SchemaName' --output json | jq -r '.[]' | while read -r SCHEMA; do
  echo "导出 ${SCHEMA}"
  aws glue list-schema-versions \
      --schema-id RegistryName="${REG}",SchemaName="${SCHEMA}" \
      --region "${REGION}" --max-results 100 \
      --query 'Schemas[].VersionNumber' --output json | jq -r '.[]' | while read -r VER; do
    aws glue get-schema-version \
        --schema-id RegistryName="${REG}",SchemaName="${SCHEMA}" \
        --schema-version-number VersionNumber="${VER}" \
        --region "${REGION}" \
        > "${OUT}/${SCHEMA}.v${VER}.json"
  done
done

echo "完成，文件落 ${OUT}/"
ls -1 "${OUT}/"

#!/bin/bash
# schema-import.sh - 按版本顺序导入到目标 Registry
# 用法：./schema-import.sh <new-registry> <in-dir> [region]

set -euo pipefail
REG="${1:?registry}"
IN="${2:?in-dir}"
REGION="${3:-ap-southeast-1}"

aws glue create-registry --registry-name "${REG}" --region "${REGION}" 2>/dev/null || true

# 按 schema 名字 + 版本号排序，逐个 register / put
ls "${IN}" | sed -E 's/\.v[0-9]+\.json$//' | sort -u | while read -r SCHEMA; do
  ls "${IN}" | grep "^${SCHEMA}\.v" | sort -t v -k2 -n | while read -r FILE; do
    DEF=$(jq -r '.SchemaDefinition' "${IN}/${FILE}")
    DT=$(jq -r '.DataFormat'        "${IN}/${FILE}")
    if aws glue get-schema --schema-id RegistryName="${REG}",SchemaName="${SCHEMA}" \
         --region "${REGION}" >/dev/null 2>&1; then
      aws glue register-schema-version \
        --schema-id RegistryName="${REG}",SchemaName="${SCHEMA}" \
        --schema-definition "${DEF}" --region "${REGION}" >/dev/null
    else
      aws glue create-schema \
        --registry-id RegistryName="${REG}" \
        --schema-name "${SCHEMA}" --data-format "${DT}" \
        --compatibility BACKWARD --schema-definition "${DEF}" \
        --region "${REGION}" >/dev/null
    fi
    echo "imported ${FILE}"
  done
done

# Java/Spring 端示意
spring:
  kafka:
    properties:
      schema.registry.url: ""    # 走 AWS Glue 而非 Confluent SR 时留空
      registry.name:        svc-foo-prod-registry
      avro.serialization.required.field: true

spring:
  kafka:
    properties:
      schema.registry.url: https://schema-registry.example.aws.com
      basic.auth.credentials.source: USER_INFO
      basic.auth.user.info: <key>:<secret>

#!/bin/bash
# msk-rollback.sh - 回滚到旧集群
# 用法：./msk-rollback.sh <env> [phase]
# phase: 1|2|3|4 表示当前所处阶段，未传则全量回滚

set -euo pipefail
ENV="${1:?env}"
PHASE="${2:-full}"
BACKUP_DIR="${HOME}/ops-archive/$(date +%F)-msk-rollback-${ENV}"
mkdir -p "${BACKUP_DIR}"

case "${PHASE}" in
  1)
    # 阶段 1 = 仅消费者改了双订阅，回滚只需关闭 dual_consume
    nacos-cli set "${ENV}/svc-foo/kafka.dual_consume" false
    kubectl -n svc-foo rollout restart sts/consumer
    ;;
  2|3)
    # 阶段 2/3 = 已经 dual write 或消费者切新，回滚需要把所有 endpoint 切回旧 + 关 dual
    nacos-cli set "${ENV}/svc-foo/kafka.bootstrap_servers" "${OLD_BS}"
    nacos-cli set "${ENV}/svc-foo/kafka.dual_produce" false
    nacos-cli set "${ENV}/svc-foo/kafka.dual_consume" false
    kubectl -n svc-foo rollout restart deploy/producer sts/consumer
    ;;
  4|full)
    # 阶段 4 = 已经只写新，旧集群可能没流量但 IAM 还在，全量回滚
    nacos-cli set "${ENV}/svc-foo/kafka.bootstrap_servers" "${OLD_BS}"
    nacos-cli set "${ENV}/svc-foo/kafka.dual_produce" false
    nacos-cli set "${ENV}/svc-foo/kafka.dual_consume" false
    kubectl -n svc-foo rollout restart deploy/producer sts/consumer
    # 确认旧 IAM policy 仍含旧集群 ARN（追加模式不需要改）
    aws iam get-role-policy --role-name svc-foo-msk-producer \
      --policy-name msk-access --query 'PolicyDocument' \
      --output json > "${BACKUP_DIR}/producer-policy.json"
    ;;
  *)
    echo "unknown phase ${PHASE}"; exit 1 ;;
esac

echo "回滚动作已下发，验证："
echo "  kubectl -n svc-foo logs -l app=consumer --tail=200 | grep -iE 'kafka|sasl|error'"
echo "  kafka-consumer-groups --bootstrap-server ${OLD_BS} --describe --group svc-foo-consumer"

svc-foo-msk SA → role/svc-foo-msk-access      (consumer 用，policy 含通配符 *)
msk-sa       SA → role/svc-foo-prod-msk-role  (producer/gateway 用，硬编码旧集群 ARN)

# 列出 ns 下所有 SA → role 映射
kubectl -n svc-foo get sa -o json | python3 -c "
import json, sys
for i in json.load(sys.stdin)['items']:
    name = i['metadata']['name']
    role = i['metadata'].get('annotations', {}).get('eks.amazonaws.com/role-arn', 'NONE')
    if role != 'NONE':
        print(f'{name}: {role}')
"

# 对每个非 NONE 的 role 都追加新集群 ARN 到 policy（追加模式，旧 ARN 保留）
for ROLE in svc-foo-msk-access svc-foo-prod-msk-role; do
  aws iam get-role-policy --role-name "${ROLE}" --policy-name msk-access \
    --query 'PolicyDocument' --output json > "/tmp/${ROLE}.json"
  jq '.Statement[].Resource += [
        "arn:aws:kafka:ap-southeast-1:<ACCOUNT_ID>:cluster/svc-foo-qa-kafka-v2/*",
        "arn:aws:kafka:ap-southeast-1:<ACCOUNT_ID>:topic/svc-foo-qa-kafka-v2/*/*",
        "arn:aws:kafka:ap-southeast-1:<ACCOUNT_ID>:group/svc-foo-qa-kafka-v2/*/*"
      ]' "/tmp/${ROLE}.json" > "/tmp/${ROLE}.new.json"
  aws iam put-role-policy --role-name "${ROLE}" --policy-name msk-access \
    --policy-document "file:///tmp/${ROLE}.new.json"
done

panic: runtime error: too many colons in address b-1.svcfooqakafkav2.xxxxxx.c4.kafka.ap-southeast-1.amazonaws.com:9098,b-2.svcfooqakafkav2.xxxxxx.c4.kafka.ap-southeast-1.amazonaws.com:9098

goroutine 1 [running]:
net.parseAddr(...)
github.com/IBM/sarama.(*Broker).Open(...)

 Kafka:
-  Addrs: ["b-1.xxx:9098,b-2.xxx:9098"]
+  Addrs:
+    - "b-1.xxx:9098"
+    - "b-2.xxx:9098"

import re

def split_addrs(content: str) -> str:
    def repl(m):
        hosts = m.group(1).split(",")
        if len(hosts) > 1:
            return "Addrs: [" + ", ".join(f'"{h.strip()}"' for h in hosts) + "]"
        return m.group(0)
    return re.sub(r'Addrs:\s*\["([^"]+)"\]', repl, content)

brokers := strings.Split(strings.TrimSpace(brokersCSV), ",")
for i := range brokers {
    brokers[i] = strings.TrimSpace(brokers[i])
}
client, err := sarama.NewClient(brokers, cfg)

#!/usr/bin/env python3
# capacity_plan.py - 给定当前流量 / 目标 12 个月增长，反算最小 broker 数
# 用法：python3 capacity_plan.py --current-mbs 1.5 --growth 3 --partitions 6

import argparse, math

INSTANCE_THROUGHPUT_MBS = {
    "kafka.t3.small":   3,   # burstable，长时间峰值 3 MB/s
    "kafka.m5.large":   30,
    "kafka.m5.xlarge":  60,
    "kafka.m5.2xlarge": 120,
}
RF = 2  # replication factor

def plan(current_mbs, growth, partitions):
    target = current_mbs * growth
    rows = []
    for inst, cap in INSTANCE_THROUGHPUT_MBS.items():
        per_broker = cap * 0.6  # 留 burst 余量
        brokers_for_throughput = math.ceil(target * RF / per_broker)
        brokers_for_partitions = math.ceil(partitions / 50)  # 每 broker 50 partition
        brokers = max(2, brokers_for_throughput, brokers_for_partitions)
        rows.append((inst, brokers, brokers * cap * 0.6))
    print(f"目标 {target:.1f} MB/s，partition {partitions}：")
    for inst, b, cap in rows:
        print(f"  {inst:20s} brokers={b}  容量~{cap:.0f} MB/s")

if __name__ == "__main__":
    p = argparse.ArgumentParser()
    p.add_argument("--current-mbs", type=float, required=True)
    p.add_argument("--growth", type=float, default=3.0)
    p.add_argument("--partitions", type=int, default=6)
    a = p.parse_args()
    plan(a.current_mbs, a.growth, a.partitions)

# Schema Registry
aws glue list-registries --region $REGION
aws glue list-schemas    --region $REGION --max-results 100

# MSK Connect
aws kafkaconnect list-connectors --region $REGION

# CloudWatch alarm referencing old cluster
aws cloudwatch describe-alarms --region $REGION \
  --query "MetricAlarms[?contains(Dimensions[?Name=='Cluster Name'].Value | [0], 'old-cluster')].AlarmName" \
  --output table

# IAM policy 含旧 ARN 的角色
aws iam list-policies --scope Local \
  --query 'Policies[?contains(PolicyName, `msk`)].Arn' --output text \
  | xargs -I{} aws iam get-policy-version --policy-arn {} --version-id v1 \
      --query 'PolicyVersion.Document' --output json