Playbook：新建子环境的隔离 checklist——一次 ID 撞车污染 10 万条数据的事故复盘

-- 步骤 1: 备份受污染表（CTAS，秒级，不锁主表）
CREATE TABLE service_a.realtime_messages_bak_20260418
  SELECT * FROM service_a.realtime_messages
  WHERE created_at > '2026-03-25';

ALTER TABLE service_a.realtime_messages_bak_20260418
  ADD PRIMARY KEY (id);

-- 步骤 2: 把"待删"行 id 灌进临时表（用 CTE 把跨库 id 撞车的脏数据找出来）
CREATE TABLE service_a._dirty_ids (
  id BIGINT PRIMARY KEY
) ENGINE=InnoDB;

INSERT INTO service_a._dirty_ids (id)
WITH suspicious AS (
  SELECT m.id
  FROM service_a.realtime_messages m
  JOIN service_a.project p ON m.project_id = p.id
  WHERE m.project_id BETWEEN 1 AND 274          -- env-AI 当时 max_id
    AND m.created_at >= '2026-03-25'             -- env-AI 上线日
    AND p.created_at <  '2026-03-25'             -- env-QA 老项目
)
SELECT id FROM suspicious;

-- 步骤 3: 分批 DELETE（每批 1000 行，避免锁表 + binlog 巨大事务）
DROP PROCEDURE IF EXISTS service_a.purge_dirty_messages;

DELIMITER $$
CREATE PROCEDURE service_a.purge_dirty_messages()
BEGIN
  DECLARE done INT DEFAULT 0;
  DECLARE total INT DEFAULT 0;
  SELECT COUNT(*) INTO total FROM service_a._dirty_ids;
  SELECT CONCAT('待清理脏行：', total) AS info;

  REPEAT
    DELETE m FROM service_a.realtime_messages m
    JOIN service_a._dirty_ids d ON m.id = d.id
    LIMIT 1000;

    SET done = ROW_COUNT();
    DELETE FROM service_a._dirty_ids
    WHERE id IN (
      SELECT id FROM (
        SELECT d2.id FROM service_a._dirty_ids d2
        LEFT JOIN service_a.realtime_messages m ON m.id = d2.id
        WHERE m.id IS NULL LIMIT 1000
      ) t
    );

    DO SLEEP(0.2);  -- 给主从同步留缓冲
  UNTIL done = 0 END REPEAT;

  SELECT '清理完成' AS info;
END$$
DELIMITER ;

CALL service_a.purge_dirty_messages();

-- 步骤 4: 校验
SELECT COUNT(*) AS remaining
FROM service_a.realtime_messages m
JOIN service_a.project p ON m.project_id = p.id
WHERE m.project_id BETWEEN 1 AND 274
  AND m.created_at >= '2026-03-25'
  AND p.created_at <  '2026-03-25';
-- 期望输出：remaining = 0

# 文件：env-ai-namespace.yaml
---
apiVersion: v1
kind: Namespace
metadata:
  name: env-ai
  labels:
    goalfy.dev/env: ai
    goalfy.dev/tier: non-prod
    pod-security.kubernetes.io/enforce: baseline
---
apiVersion: v1
kind: ResourceQuota
metadata:
  name: env-ai-quota
  namespace: env-ai
spec:
  hard:
    requests.cpu: "32"
    requests.memory: "64Gi"
    limits.cpu: "64"
    limits.memory: "128Gi"
    persistentvolumeclaims: "20"
    services.loadbalancers: "2"
    pods: "200"
---
apiVersion: v1
kind: LimitRange
metadata:
  name: env-ai-limits
  namespace: env-ai
spec:
  limits:
    - type: Container
      default:
        cpu: "500m"
        memory: "512Mi"
      defaultRequest:
        cpu: "100m"
        memory: "128Mi"
      max:
        cpu: "8"
        memory: "16Gi"
---
# 默认 deny 所有入站和出站
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: default-deny-all
  namespace: env-ai
spec:
  podSelector: {}
  policyTypes: ["Ingress", "Egress"]
---
# 允许 namespace 内部互通 + 出站 DNS + 出站 HTTPS（中间件、外部 API）
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: allow-internal-and-egress
  namespace: env-ai
spec:
  podSelector: {}
  policyTypes: ["Ingress", "Egress"]
  ingress:
    - from:
        - namespaceSelector:
            matchLabels:
              goalfy.dev/env: ai
        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: ingress-nginx
  egress:
    - to:
        - namespaceSelector: {}
          podSelector:
            matchLabels:
              k8s-app: kube-dns
      ports:
        - port: 53
          protocol: UDP
        - port: 53
          protocol: TCP
    - to:
        - ipBlock:
            cidr: 0.0.0.0/0
            except:
              - 169.254.0.0/16
              - 10.0.0.0/8       # 集群内网走 namespaceSelector 单独放行
      ports:
        - port: 443
          protocol: TCP
        - port: 5671              # AMQP TLS
          protocol: TCP
        - port: 9094              # Kafka TLS
          protocol: TCP
        - port: 6379              # Redis（如走专线）
          protocol: TCP
        - port: 3306              # MySQL
          protocol: TCP

kubectl apply -f env-ai-namespace.yaml

# 1. namespace 创建成功
kubectl get ns env-ai -o jsonpath='{.metadata.labels}' | jq
# 期望输出：{"goalfy.dev/env":"ai","goalfy.dev/tier":"non-prod",...}

# 2. ResourceQuota 生效
kubectl describe resourcequota env-ai-quota -n env-ai
# 期望看到 Used / Hard 表格

# 3. NetworkPolicy 默认 deny
kubectl run -n env-ai netcheck --image=nicolaka/netshoot --rm -it --restart=Never -- \
  curl -m 3 https://www.google.com -I
# 期望：先 curl 不通（被 deny-all 拦），applyallow-internal-and-egress 后通

kubectl delete -f env-ai-namespace.yaml

#!/bin/bash
# create-aurora-env-ai.sh
set -euo pipefail

ENV=ai
AWS_REGION=ap-southeast-1
DB_CLUSTER_ID="${ENV}-aurora-mysql"
SUBNET_GROUP="${ENV}-aurora-subnet-group"
SG_ID="sg-xxxxxxxxxxxxx"
MASTER_USER="admin"
MASTER_PWD=$(openssl rand -base64 24)

# 1. 写入凭据到 Secrets Manager
aws secretsmanager create-secret \
  --name "/${ENV}/aurora/admin" \
  --secret-string "{\"username\":\"${MASTER_USER}\",\"password\":\"${MASTER_PWD}\"}" \
  --region "${AWS_REGION}"

# 2. 创建子网组
aws rds create-db-subnet-group \
  --db-subnet-group-name "${SUBNET_GROUP}" \
  --db-subnet-group-description "subnet group for ${ENV}" \
  --subnet-ids subnet-xxxxxxxxxxxxxa subnet-xxxxxxxxxxxxxb subnet-xxxxxxxxxxxxxc \
  --region "${AWS_REGION}"

# 3. 创建 Aurora Serverless v2 cluster
aws rds create-db-cluster \
  --db-cluster-identifier "${DB_CLUSTER_ID}" \
  --engine aurora-mysql \
  --engine-version 8.0.mysql_aurora.3.05.2 \
  --master-username "${MASTER_USER}" \
  --master-user-password "${MASTER_PWD}" \
  --db-subnet-group-name "${SUBNET_GROUP}" \
  --vpc-security-group-ids "${SG_ID}" \
  --serverless-v2-scaling-configuration MinCapacity=0.5,MaxCapacity=4 \
  --backup-retention-period 7 \
  --preferred-backup-window "16:00-17:00" \
  --storage-encrypted \
  --enable-cloudwatch-logs-exports '["error","slowquery"]' \
  --tags "Key=env,Value=${ENV}" "Key=managed-by,Value=ops-checklist" \
  --region "${AWS_REGION}"

# 4. 创建 writer instance
aws rds create-db-instance \
  --db-instance-identifier "${DB_CLUSTER_ID}-writer" \
  --db-cluster-identifier "${DB_CLUSTER_ID}" \
  --engine aurora-mysql \
  --db-instance-class db.serverless \
  --region "${AWS_REGION}"

echo "等待 cluster 状态变 available（约 5-8 分钟）..."
aws rds wait db-cluster-available --db-cluster-identifier "${DB_CLUSTER_ID}" --region "${AWS_REGION}"

ENDPOINT=$(aws rds describe-db-clusters \
  --db-cluster-identifier "${DB_CLUSTER_ID}" \
  --query 'DBClusters[0].Endpoint' --output text \
  --region "${AWS_REGION}")
echo "Aurora endpoint: ${ENDPOINT}"

# 状态 available
aws rds describe-db-clusters --db-cluster-identifier ai-aurora-mysql \
  --query 'DBClusters[0].Status' --output text
# 期望输出：available

# 连接测试
mysql -h "${ENDPOINT}" -u admin -p${MASTER_PWD} -e "SELECT VERSION();"
# 期望输出：8.0.mysql_aurora.3.05.2

aws rds delete-db-instance --db-instance-identifier ai-aurora-mysql-writer --skip-final-snapshot
aws rds delete-db-cluster --db-cluster-identifier ai-aurora-mysql --skip-final-snapshot
aws rds delete-db-subnet-group --db-subnet-group-name ai-aurora-subnet-group

#!/bin/bash
# create-rabbitmq-env-ai.sh
set -euo pipefail

ENV=ai
AWS_REGION=ap-southeast-1
BROKER_NAME="${ENV}-rabbitmq"
SG_ID="sg-yyyyyyyyyyyyy"
USER="admin"
PWD=$(openssl rand -base64 24)

aws secretsmanager create-secret \
  --name "/${ENV}/rabbitmq/admin" \
  --secret-string "{\"username\":\"${USER}\",\"password\":\"${PWD}\"}" \
  --region "${AWS_REGION}"

aws mq create-broker \
  --broker-name "${BROKER_NAME}" \
  --engine-type RABBITMQ \
  --engine-version 3.13 \
  --host-instance-type mq.m7g.medium \
  --deployment-mode SINGLE_INSTANCE \
  --publicly-accessible false \
  --subnet-ids subnet-xxxxxxxxxxxxxa \
  --security-groups "${SG_ID}" \
  --auto-minor-version-upgrade true \
  --users "Username=${USER},Password=${PWD},ConsoleAccess=true" \
  --tags "env=${ENV}" \
  --region "${AWS_REGION}"

echo "broker 创建中（约 10-15 分钟）..."

aws mq describe-broker --broker-id <BROKER_ID> \
  --query 'BrokerState' --output text
# 期望：RUNNING

# 拿 endpoint
aws mq describe-broker --broker-id <BROKER_ID> \
  --query 'BrokerInstances[0].Endpoints'

#!/bin/bash
# create-valkey-env-ai.sh
set -euo pipefail

ENV=ai
AWS_REGION=ap-southeast-1
RG_ID="${ENV}-valkey"

aws elasticache create-cache-subnet-group \
  --cache-subnet-group-name "${RG_ID}-subnet" \
  --cache-subnet-group-description "valkey subnet for ${ENV}" \
  --subnet-ids subnet-xxxxxxxxxxxxxa subnet-xxxxxxxxxxxxxb \
  --region "${AWS_REGION}"

aws elasticache create-replication-group \
  --replication-group-id "${RG_ID}" \
  --replication-group-description "Valkey for ${ENV}" \
  --engine valkey \
  --engine-version 7.2 \
  --cache-node-type cache.t4g.micro \
  --num-cache-clusters 2 \
  --cache-subnet-group-name "${RG_ID}-subnet" \
  --security-group-ids sg-zzzzzzzzzzzzz \
  --automatic-failover-enabled \
  --transit-encryption-enabled \
  --tags "Key=env,Value=${ENV}" \
  --region "${AWS_REGION}"

#!/bin/bash
# create-msk-env-ai.sh
set -euo pipefail

cat > /tmp/msk-config.json <<'EOF'
{
  "BrokerNodeGroupInfo": {
    "InstanceType": "kafka.m7g.large",
    "ClientSubnets": [
      "subnet-xxxxxxxxxxxxxa",
      "subnet-xxxxxxxxxxxxxb",
      "subnet-xxxxxxxxxxxxxc"
    ],
    "SecurityGroups": ["sg-aaaaaaaaaaaaa"]
  },
  "ClusterName": "ai-msk",
  "EncryptionInfo": {
    "EncryptionInTransit": {
      "ClientBroker": "TLS",
      "InCluster": true
    }
  },
  "EnhancedMonitoring": "PER_TOPIC_PER_BROKER",
  "KafkaVersion": "2.8.1",
  "NumberOfBrokerNodes": 3,
  "Tags": { "env": "ai" }
}
EOF

aws kafka create-cluster --cli-input-json file:///tmp/msk-config.json \
  --region ap-southeast-1

#!/bin/bash
# kafka-create-topic.sh
# 用法：./kafka-create-topic.sh <env> <service> <event> [partitions] [replication]

set -euo pipefail

ENV="${1:-}"
SERVICE="${2:-}"
EVENT="${3:-}"
PARTITIONS="${4:-6}"
REPLICATION="${5:-3}"

if [[ -z "$ENV" || -z "$SERVICE" || -z "$EVENT" ]]; then
  echo "用法：$0 <env> <service> <event> [partitions] [replication]"
  exit 1
fi

# 命名校验：env 必须在白名单
case "$ENV" in
  qa|ai|pre|staging|prod) ;;
  *) echo "❌ env 必须是 qa/ai/pre/staging/prod，当前：$ENV"; exit 1 ;;
esac

# service / event 只允许小写字母 + 连字符
if [[ ! "$SERVICE" =~ ^[a-z][a-z0-9-]*$ ]]; then
  echo "❌ service 必须是小写字母+连字符开头：$SERVICE"; exit 1
fi
if [[ ! "$EVENT" =~ ^[a-z][a-z0-9-]*$ ]]; then
  echo "❌ event 必须是小写字母+连字符开头：$EVENT"; exit 1
fi

TOPIC="${ENV}.${SERVICE}.${EVENT}"
BOOTSTRAP="${KAFKA_BOOTSTRAP:?需要设置 KAFKA_BOOTSTRAP}"

echo "[+] 创建 topic：${TOPIC}（partitions=${PARTITIONS}, replication=${REPLICATION}）"
kafka-topics.sh \
  --bootstrap-server "${BOOTSTRAP}" \
  --command-config /etc/kafka/client.properties \
  --create \
  --topic "${TOPIC}" \
  --partitions "${PARTITIONS}" \
  --replication-factor "${REPLICATION}" \
  --config retention.ms=604800000 \
  --config min.insync.replicas=2

# 校验
kafka-topics.sh \
  --bootstrap-server "${BOOTSTRAP}" \
  --command-config /etc/kafka/client.properties \
  --describe --topic "${TOPIC}"

// pkg/mqtopic/topic.go
package mqtopic

import (
    "fmt"
    "os"
    "regexp"
)

var (
    validEnv     = regexp.MustCompile(`^(qa|ai|pre|staging|prod)$`)
    validNameSeg = regexp.MustCompile(`^[a-z][a-z0-9-]*$`)
)

// Name 拼接 topic 名 <env>.<service>.<event>
// 启动时如果 DISPATCH_ENV 没设或不合法，直接 panic 让进程起不来。
func Name(service, event string) string {
    env := os.Getenv("DISPATCH_ENV")
    if !validEnv.MatchString(env) {
        panic(fmt.Sprintf("DISPATCH_ENV invalid: %q (must be qa/ai/pre/staging/prod)", env))
    }
    if !validNameSeg.MatchString(service) || !validNameSeg.MatchString(event) {
        panic(fmt.Sprintf("invalid service/event segment: %s/%s", service, event))
    }
    return fmt.Sprintf("%s.%s.%s", env, service, event)
}

// Group 拼接 consumer group 名 <env>-<service>-<purpose>
func Group(service, purpose string) string {
    env := os.Getenv("DISPATCH_ENV")
    if !validEnv.MatchString(env) {
        panic(fmt.Sprintf("DISPATCH_ENV invalid: %q", env))
    }
    return fmt.Sprintf("%s-%s-%s", env, service, purpose)
}

kafka-topics.sh --bootstrap-server $KAFKA_BOOTSTRAP --list \
  | grep -v '^[a-z]\+\.' && echo "❌ 发现不带 env 前缀的 topic" || echo "✅ 全部 topic 命名合规"

kafka-topics.sh --bootstrap-server $KAFKA_BOOTSTRAP --delete --topic ai.service-foo.realtime

-- 对所有业务表批量设置 AUTO_INCREMENT 起点 1000 万
SET @start = 10000000;
SET @stmt = NULL;

SELECT GROUP_CONCAT(
  CONCAT('ALTER TABLE `', TABLE_NAME, '` AUTO_INCREMENT=', @start)
  SEPARATOR '; '
)
INTO @stmt
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = DATABASE()
  AND AUTO_INCREMENT IS NOT NULL;

PREPARE s FROM @stmt;
EXECUTE s;
DEALLOCATE PREPARE s;

-- 校验
SELECT TABLE_NAME, AUTO_INCREMENT
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = DATABASE()
  AND AUTO_INCREMENT < 10000000;
-- 期望输出：empty set

-- 一次性把所有 sequence 推到 1000 万
DO $$
DECLARE
  r RECORD;
BEGIN
  FOR r IN
    SELECT schemaname, sequencename
    FROM pg_sequences
    WHERE schemaname = current_schema()
  LOOP
    EXECUTE format('SELECT setval(%L, 10000000)',
                   r.schemaname || '.' || r.sequencename);
  END LOOP;
END$$;

-- 校验
SELECT schemaname, sequencename, last_value
FROM pg_sequences
WHERE last_value < 10000000;
-- 期望输出：empty set

---
apiVersion: monitoring.coreos.com/v1
kind: PrometheusRule
metadata:
  name: id-exhaustion
  namespace: monitoring
  labels:
    role: alert-rules
spec:
  groups:
    - name: id-exhaustion
      interval: 5m
      rules:
        - alert: BusinessTableIdNearLimit
          expr: |
            mysql_table_max_id_ratio > 0.7
          for: 30m
          labels:
            severity: warning
          annotations:
            summary: "业务表 id 接近 INT 上限（70%）"
            description: "{{ $labels.env }} / {{ $labels.table }} id 已用 {{ $value | humanizePercentage }}，需要规划 BIGINT 升级"
        - alert: BusinessTableIdRangeOverlap
          expr: |
            multi_env_id_overlap_count > 0
          for: 10m
          labels:
            severity: critical
          annotations:
            summary: "跨环境业务表 id 区间重叠"
            description: "{{ $labels.env_a }} 和 {{ $labels.env_b }} 的 {{ $labels.table }} id 区间存在交集"

-- 跨环境检查 id 区间是否重叠
SELECT 'env-qa'  AS env, MIN(id), MAX(id) FROM qa_db.project
UNION ALL
SELECT 'env-ai'  AS env, MIN(id), MAX(id) FROM ai_db.project
UNION ALL
SELECT 'env-pre' AS env, MIN(id), MAX(id) FROM pre_db.project;
-- 期望：每个环境 [MIN, MAX] 区间互不相交

-- migrations/00001_init.up.sql
-- 所有业务表必须带 env_tag，NOT NULL DEFAULT 'unknown'
CREATE TABLE realtime_messages (
  id           BIGINT       NOT NULL AUTO_INCREMENT,
  project_id   BIGINT       NOT NULL,
  user_id      BIGINT       NOT NULL,
  content      TEXT         NOT NULL,
  env_tag      VARCHAR(16)  NOT NULL DEFAULT 'unknown',
  created_at   TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
  PRIMARY KEY (id),
  KEY idx_env_proj (env_tag, project_id),
  KEY idx_proj_created (project_id, created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 触发器兜底（防应用层漏填）
DELIMITER $$
CREATE TRIGGER realtime_messages_env_tag_default
BEFORE INSERT ON realtime_messages
FOR EACH ROW
BEGIN
  IF NEW.env_tag = 'unknown' OR NEW.env_tag = '' THEN
    SIGNAL SQLSTATE '45000'
      SET MESSAGE_TEXT = 'env_tag must be set explicitly';
  END IF;
END$$
DELIMITER ;

// pkg/mqclient/publisher.go
package mqclient

import (
    "context"
    "encoding/json"
    "errors"
    "os"

    amqp "github.com/rabbitmq/amqp091-go"
)

type Envelope struct {
    DispatchEnv string          `json:"dispatch_env"`
    EventType   string          `json:"event_type"`
    Payload     json.RawMessage `json:"payload"`
}

type Publisher struct {
    ch  *amqp.Channel
    env string
}

func NewPublisher(ch *amqp.Channel) *Publisher {
    env := os.Getenv("DISPATCH_ENV")
    if env == "" {
        panic("DISPATCH_ENV not set, refusing to start publisher")
    }
    return &Publisher{ch: ch, env: env}
}

// Publish 强制注入 dispatch_env，业务代码无法绕过
func (p *Publisher) Publish(ctx context.Context, exchange, routingKey, eventType string, payload any) error {
    raw, err := json.Marshal(payload)
    if err != nil {
        return err
    }
    env := Envelope{
        DispatchEnv: p.env,
        EventType:   eventType,
        Payload:     raw,
    }
    body, err := json.Marshal(env)
    if err != nil {
        return err
    }
    return p.ch.PublishWithContext(ctx, exchange, routingKey, false, false, amqp.Publishing{
        ContentType: "application/json",
        Headers: amqp.Table{
            "x-dispatch-env": p.env,
        },
        Body: body,
    })
}

// pkg/mqclient/consumer.go
package mqclient

import (
    "encoding/json"
    "log/slog"
    "os"

    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promauto"
    amqp "github.com/rabbitmq/amqp091-go"
)

var (
    droppedCrossEnv = promauto.NewCounterVec(prometheus.CounterOpts{
        Name: "dispatcher_dropped_cross_env_total",
        Help: "消息因 dispatch_env 不匹配被丢弃",
    }, []string{"local_env", "msg_env"})

    droppedEmptyEnv = promauto.NewCounter(prometheus.CounterOpts{
        Name: "dispatcher_empty_env_total",
        Help: "消息 dispatch_env 字段为空（异常，必须告警）",
    })
)

type Handler func(payload []byte) error

type Consumer struct {
    ch       *amqp.Channel
    localEnv string
    handlers map[string]Handler
}

func NewConsumer(ch *amqp.Channel) *Consumer {
    env := os.Getenv("DISPATCH_ENV")
    if env == "" {
        panic("DISPATCH_ENV not set")
    }
    return &Consumer{ch: ch, localEnv: env, handlers: map[string]Handler{}}
}

func (c *Consumer) Handle(d amqp.Delivery) {
    var env Envelope
    if err := json.Unmarshal(d.Body, &env); err != nil {
        slog.Error("invalid envelope", "err", err)
        _ = d.Nack(false, false)
        return
    }

    // 显式失败：env 字段为空必须告警
    if env.DispatchEnv == "" {
        droppedEmptyEnv.Inc()
        slog.Error("empty dispatch_env, dropping",
            "event", env.EventType, "msg_id", d.MessageId)
        _ = d.Nack(false, false)
        return
    }

    // 跨环境过滤
    if env.DispatchEnv != c.localEnv {
        droppedCrossEnv.WithLabelValues(c.localEnv, env.DispatchEnv).Inc()
        slog.Warn("cross-env message dropped",
            "local", c.localEnv, "msg_env", env.DispatchEnv)
        _ = d.Ack(false)  // ack 掉，不要回到队列
        return
    }

    handler, ok := c.handlers[env.EventType]
    if !ok {
        slog.Warn("no handler", "event", env.EventType)
        _ = d.Ack(false)
        return
    }

    if err := handler(env.Payload); err != nil {
        slog.Error("handler failed", "err", err)
        _ = d.Nack(false, true)  // 重入队
        return
    }
    _ = d.Ack(false)
}

# 单测
go test ./pkg/mqclient/... -run TestCrossEnvDrop -v

# 跨环境冒烟（手动 publish 一个错环境的消息，期望被丢弃 + metric 增加）
DISPATCH_ENV=qa ./bin/test-publisher --to=ai-broker --env-override=ai
sleep 5
curl -s http://localhost:8080/metrics | grep dispatcher_dropped_cross_env
# 期望看到 dispatcher_dropped_cross_env_total{local_env="ai",msg_env="qa"} 1

---
apiVersion: monitoring.coreos.com/v1
kind: PrometheusRule
metadata:
  name: env-ai-rules
  namespace: monitoring
  labels:
    role: alert-rules
    env: ai     # 关键：每条规则都打 env label，方便 alertmanager 路由
spec:
  groups:
    - name: env-ai
      interval: 30s
      rules:
        - alert: HighErrorRate
          expr: |
            sum by (service) (rate(http_requests_total{namespace="env-ai",code=~"5.."}[5m]))
              / sum by (service) (rate(http_requests_total{namespace="env-ai"}[5m])) > 0.05
          for: 5m
          labels:
            severity: warning
            env: ai
          annotations:
            summary: "env-ai {{ $labels.service }} 错误率 > 5%"

# alertmanager.yaml
route:
  receiver: 'default'
  group_by: ['alertname', 'env']
  routes:
    - matchers:
        - env="prod"
      receiver: 'dingtalk-prod-oncall'
      continue: false
    - matchers:
        - env="pre"
      receiver: 'dingtalk-pre'
      continue: false
    - matchers:
        - env="ai"
      receiver: 'dingtalk-ai-team'
      continue: false
    - matchers:
        - env="qa"
      receiver: 'feishu-qa'
      continue: false

receivers:
  - name: 'default'
    webhook_configs:
      - url: 'http://prometheus-alert:8080/dingtalk/default/send'
  - name: 'dingtalk-prod-oncall'
    webhook_configs:
      - url: 'http://prometheus-alert:8080/dingtalk/prod-oncall/send'
        send_resolved: true
  - name: 'dingtalk-pre'
    webhook_configs:
      - url: 'http://prometheus-alert:8080/dingtalk/pre/send'
        send_resolved: true
  - name: 'dingtalk-ai-team'
    webhook_configs:
      - url: 'http://prometheus-alert:8080/dingtalk/ai-team/send'
        send_resolved: true
  - name: 'feishu-qa'
    webhook_configs:
      - url: 'http://prometheus-alert:8080/feishu/qa/send'
        send_resolved: true

# 触发一个 env=ai 的测试告警
curl -X POST http://alertmanager:9093/api/v1/alerts -H "Content-Type: application/json" -d '[{
  "labels": {"alertname":"TestAlert","env":"ai","severity":"warning"},
  "annotations": {"summary":"test"}
}]'
# 期望：ai-team 钉钉群收到通知；其他群不响

#!/bin/bash
# e2e-isolation-check.sh
set -euo pipefail

ENV="${1:?用法：$0 <env-name>}"
NEIGHBOR="${2:?用法：$0 <env-name> <neighbor-env-name>}"

echo "[1/5] 检查 K8s namespace 隔离"
kubectl get ns "env-${ENV}" -o jsonpath='{.metadata.labels.goalfy\.dev/env}'
echo

echo "[2/5] 检查 RabbitMQ broker 不与邻接环境共用"
ENV_BROKER=$(aws ssm get-parameter --name "/${ENV}/rabbitmq/host" --query 'Parameter.Value' --output text)
NEI_BROKER=$(aws ssm get-parameter --name "/${NEIGHBOR}/rabbitmq/host" --query 'Parameter.Value' --output text)
if [[ "$ENV_BROKER" == "$NEI_BROKER" ]]; then
  echo "❌ ${ENV} 与 ${NEIGHBOR} 共用 RabbitMQ broker：${ENV_BROKER}"; exit 1
fi
echo "✅ ${ENV} broker=${ENV_BROKER}, ${NEIGHBOR} broker=${NEI_BROKER}"

echo "[3/5] 检查 Valkey 实例不共用"
ENV_VALKEY=$(aws ssm get-parameter --name "/${ENV}/valkey/host" --query 'Parameter.Value' --output text)
NEI_VALKEY=$(aws ssm get-parameter --name "/${NEIGHBOR}/valkey/host" --query 'Parameter.Value' --output text)
if [[ "$ENV_VALKEY" == "$NEI_VALKEY" ]]; then
  echo "❌ Valkey 共用：${ENV_VALKEY}"; exit 1
fi
echo "✅ ${ENV} valkey=${ENV_VALKEY}"

echo "[4/5] 检查 Aurora cluster 不共用"
ENV_DB=$(aws rds describe-db-clusters --db-cluster-identifier "${ENV}-aurora-mysql" \
  --query 'DBClusters[0].DBClusterArn' --output text)
NEI_DB=$(aws rds describe-db-clusters --db-cluster-identifier "${NEIGHBOR}-aurora-mysql" \
  --query 'DBClusters[0].DBClusterArn' --output text)
if [[ "$ENV_DB" == "$NEI_DB" ]]; then
  echo "❌ Aurora 共用：${ENV_DB}"; exit 1
fi
echo "✅ ${ENV} aurora=${ENV_DB}"

echo "[5/5] 跨环境扫表巡检"
mysql -h "${NEIGHBOR}-aurora.cluster-xxx.rds.amazonaws.com" \
  -u readonly -p"${NEIGHBOR_RO_PWD}" \
  -D "${NEIGHBOR}_db" <<EOF
SELECT m.project_id, p.user_id, COUNT(*) AS suspicious
FROM realtime_messages m
JOIN project p ON m.project_id = p.id
WHERE m.project_id <= (SELECT MAX(id) FROM ${ENV}_db.project)
  AND m.created_at > '$(date -d '7 days ago' +%F)'
  AND p.created_at < '$(date -d '7 days ago' +%F)'
GROUP BY m.project_id, p.user_id
HAVING suspicious > 0
ORDER BY suspicious DESC
LIMIT 20;
EOF
echo "✅ 扫表完成（如有结果列表，立即停止上线）"

---
apiVersion: batch/v1
kind: CronJob
metadata:
  name: env-isolation-daily-check
  namespace: monitoring
spec:
  schedule: "0 0 * * *"   # UTC 0 点 = 北京 8 点
  concurrencyPolicy: Forbid
  successfulJobsHistoryLimit: 3
  failedJobsHistoryLimit: 7
  jobTemplate:
    spec:
      template:
        spec:
          restartPolicy: OnFailure
          serviceAccountName: env-isolation-checker
          containers:
            - name: checker
              image: <ACCOUNT_ID>.dkr.ecr.ap-southeast-1.amazonaws.com/env-isolation-checker:1.0.0
              env:
                - name: ENVIRONMENTS
                  value: "qa,ai,pre,staging"
                - name: ALERT_WEBHOOK
                  valueFrom:
                    secretKeyRef:
                      name: dingtalk-ops
                      key: webhook
              command: ["/bin/sh", "-c"]
              args:
                - |
                  set -e
                  /app/check-broker-uniqueness.sh
                  /app/check-id-range-overlap.sh
                  /app/check-empty-env-tag.sh

#!/bin/bash
# .git/hooks/pre-commit (或 pre-commit framework)
# 拒绝在 terraform/kustomize 配置里出现两个 env 共用同一个 broker arn

set -euo pipefail

CHANGED=$(git diff --cached --name-only --diff-filter=ACM | grep -E '\.(tf|yaml|yml)$' || true)
[[ -z "$CHANGED" ]] && exit 0

# 提取所有 broker_arn / cluster_arn 引用
DUP=$(echo "$CHANGED" | xargs grep -hoE '(broker_arn|cluster_arn|replication_group_id)\s*=\s*"[^"]+"' \
  | sort | uniq -c | awk '$1 > 1 {print}')

if [[ -n "$DUP" ]]; then
  echo "❌ 发现多处引用同一 broker/cluster/redis：" >&2
  echo "$DUP" >&2
  exit 1
fi

#!/bin/bash
# check-id-start.sh
# 用法：./check-id-start.sh <new-env> <db-host> <db-user> <db-pwd>
set -euo pipefail

NEW_ENV="$1"
NEW_HOST="$2"
USER="$3"
PWD="$4"

OTHER_ENVS=(qa ai pre staging)
GLOBAL_MAX=0

for env in "${OTHER_ENVS[@]}"; do
  [[ "$env" == "$NEW_ENV" ]] && continue
  HOST=$(aws ssm get-parameter --name "/${env}/aurora/host" --query 'Parameter.Value' --output text)
  MAX=$(mysql -h "$HOST" -u "$USER" -p"$PWD" -D "${env}_db" \
        -BNe "SELECT IFNULL(MAX(id),0) FROM project")
  echo "  ${env} max(project.id) = ${MAX}"
  if (( MAX > GLOBAL_MAX )); then GLOBAL_MAX=$MAX; fi
done

REQUIRED=$(( GLOBAL_MAX + 10000000 ))
NEW_AUTO=$(mysql -h "$NEW_HOST" -u "$USER" -p"$PWD" -D "${NEW_ENV}_db" \
  -BNe "SELECT AUTO_INCREMENT FROM information_schema.TABLES WHERE TABLE_SCHEMA='${NEW_ENV}_db' AND TABLE_NAME='project'")

if (( NEW_AUTO < REQUIRED )); then
  echo "❌ ${NEW_ENV}.project AUTO_INCREMENT=${NEW_AUTO}, 要求 >= ${REQUIRED}"
  exit 1
fi
echo "✅ ${NEW_ENV}.project AUTO_INCREMENT=${NEW_AUTO} >= ${REQUIRED}"

-- check-env-tag-coverage.sql
-- 期望：所有业务表都有 env_tag 字段且 NOT NULL
SELECT TABLE_NAME, COLUMN_NAME, IS_NULLABLE, COLUMN_DEFAULT
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = DATABASE()
  AND COLUMN_NAME = 'env_tag';

-- 找出没有 env_tag 字段的表
SELECT t.TABLE_NAME
FROM information_schema.TABLES t
LEFT JOIN information_schema.COLUMNS c
  ON c.TABLE_SCHEMA = t.TABLE_SCHEMA
 AND c.TABLE_NAME = t.TABLE_NAME
 AND c.COLUMN_NAME = 'env_tag'
WHERE t.TABLE_SCHEMA = DATABASE()
  AND t.TABLE_TYPE = 'BASE TABLE'
  AND c.COLUMN_NAME IS NULL;
-- 期望输出：empty set

# 新环境上线申请：env-<NAME>

## PRD 阶段（产品/PM 填）

- [ ] 环境用途（QA / 灰度 / 算法实验 / 多租户隔离）
- [ ] 预期生命周期（≤ 1 个月走方案 B；≥ 3 个月走方案 C；prod 走方案 A）
- [ ] 预期 QPS / 并发用户数 / 月数据增量
- [ ] 是否需要从旧环境同步基础数据（影响 id 起点策略）
- [ ] 是否对外暴露公网（影响 SSL / WAF / 备案）

## 技术评审阶段（架构/SRE 填）

- [ ] 选定方案：A / B / C（默认 C）
- [ ] 月成本预估：$ ____
- [ ] 资源 ID 列表（不能与现有环境重复）：
  - [ ] EKS namespace 名
  - [ ] Aurora cluster identifier
  - [ ] RabbitMQ broker name
  - [ ] Valkey replication group id
  - [ ] MSK cluster name（若独立）
  - [ ] Nacos namespace id
- [ ] 业务表 AUTO_INCREMENT 起点：____ (必须 >= 现有最大 + 1000 万)

## 上线前 dry-run

- [ ] 跑过 `./check-id-start.sh`，输出 ✅
- [ ] 跑过 `./check-env-tag-coverage.sql`，empty set
- [ ] 跑过 `./e2e-isolation-check.sh`，5/5 通过
- [ ] IaC pre-commit 通过（无重复 broker/cluster 引用）
- [ ] alertmanager route 已加新环境分支
- [ ] 钉钉/飞书 webhook 已申请并接入

## 上线后 24 小时巡检

- [ ] T+1h: 业务服务全部 Running，无 CrashLoopBackOff
- [ ] T+4h: dispatcher_empty_env_total 为 0
- [ ] T+12h: 跨环境扫表 SQL 输出 0 行
- [ ] T+24h: id 区间 dashboard 无重叠告警
- [ ] T+24h: 业务侧确认无"灵异"现象

## 责任人

- 提案人：____
- 架构 review：____
- SRE 实施：____
- 业务 owner：____
- 上线日期：____

## 集群合并/迁移 checklist

### 1. 业务 workload
- [ ] Deployment / StatefulSet / DaemonSet 全部清单
- [ ] HPA / PDB
- [ ] CronJob

### 2. 辅助组件（最容易漏）
- [ ] placeholder pod（触发 Karpenter NodePool）
- [ ] node-protector
- [ ] custom scaler
- [ ] init job / migration job
- [ ] sidecar 注入 webhook

### 3. 配置中心默认值
- [ ] Nacos / Apollo 里所有指向旧集群名 / NodePool 名 / broker host 的字段
- [ ] ConfigMap 里硬编码的 region / az / endpoint
- [ ] Secret 里指向旧 KMS key 的引用

### 4. RBAC / IRSA
- [ ] ServiceAccount 注解 eks.amazonaws.com/role-arn 是否可用
- [ ] ClusterRole / RoleBinding 是否带集群名
- [ ] CSI driver / addon 的 IAM 绑定

### 5. 端到端验证
- [ ] 完整链路（创建 → 使用 → 销毁）跑一次
- [ ] 不只看 pod Running，必须验证业务功能

// pkg/mqclient/gateway_filter.go
package mqclient

// GatewayFilter 在 broker 入口对没有 dispatch_env 的消息直接拒收
// 部署在 RabbitMQ shovel 或 Kafka MirrorMaker 链路上做兜底
func GatewayFilter(raw []byte) (forward bool, reason string) {
    var env Envelope
    if err := json.Unmarshal(raw, &env); err != nil {
        return false, "invalid_json"
    }
    if env.DispatchEnv == "" {
        return false, "empty_env"
    }
    if !validEnv.MatchString(env.DispatchEnv) {
        return false, "invalid_env_value"
    }
    return true, ""
}

- alert: EmptyDispatchEnvSeen
  expr: rate(dispatcher_empty_env_total[5m]) > 0
  for: 5m
  labels:
    severity: critical
  annotations:
    summary: "发现 dispatch_env 为空的消息（生产端未注入）"
    description: "本环境 5 分钟内消费到 {{ $value }} 条无 env 标记的消息，必须立刻找出未升级的 producer"