# api/v1alpha1/recordingsession_types.go 核心字段示例
type RecordingSessionSpec struct {
  MeetingID       string            `json:"meetingId"`
  RoomID          string            `json:"roomId"`
  LayoutTemplate  string            `json:"layoutTemplate,omitempty"` // 画面布局模板
  OutputConfig    OutputSpec        `json:"output"`                   // 存储/推流目标
  ResourceProfile ResourceProfile   `json:"resourceProfile"`          // 资源画像：CPU/内存/GPU 需求
  ScalingPolicy   ScalingPolicySpec `json:"scalingPolicy,omitempty"`  // 定制化伸缩策略
  // 关键：定义会议预计时长、参会人数上限，供调度器预判
  ExpectedDuration *metav1.Duration `json:"expectedDuration,omitempty"` 
  MaxParticipants  int32            `json:"maxParticipants"`
}

type RecordingSessionStatus struct {
  Phase           SessionPhase      `json:"phase"` // Pending, Scheduling, Running, Compositing, Archiving, Failed, Succeeded
  NodeName        string            `json:"nodeName,omitempty"`
  PodName         string            `json:"podName,omitempty"`
  StartTime       *metav1.Time      `json:"startTime,omitempty"`
  EndTime         *metav1.Time      `json:"endTime,omitempty"`
  CurrentStreams  int32             `json:"currentStreams"`   // 实时参会流数
  HealthStatus    HealthStatus      `json:"healthStatus"`     // 细粒度健康态：Normal, Degraded, Stalled
  FailureReason   string            `json:"failureReason,omitempty"`
  RetryCount      int32             `json:"retryCount"`
}

func (r *RecordingSessionReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    session := &v1alpha1.RecordingSession{}
    if err := r.Get(ctx, req.NamespacedName, session); err != nil {
        return ctrl.Result{}, client.IgnoreNotFound(err)
    }

    // 1. 终态保护：归档中/已成功/已失败 不再处理
    if session.IsTerminal() {
        return ctrl.Result{}, nil
    }

    // 2. 状态机流转逻辑
    switch session.Status.Phase {
    case v1alpha1.PhasePending:
        return r.handleScheduling(ctx, session)      // 调度决策：选节点、创建 Pod
    case v1alpha1.PhaseScheduling:
        return r.handlePodBinding(ctx, session)      // 等待 Pod Running，注入 Sidecar
    case v1alpha1.PhaseRunning:
        return r.handleRuntimeGovernance(ctx, session) // 核心：伸缩、健康检查、自愈
    case v1alpha1.PhaseCompositing:
        return r.handleFinalize(ctx, session)        // 会议结束，触发合流切片、上传归档
    default:
        return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
    }
}

# HELP recording_active_streams 当前活跃音视频流数量
# TYPE recording_active_streams gauge
recording_active_streams{meeting_id="m_123", session_id="s_456"} 42

# HELP recording_stream_bitrate_kbps 单流平均码率
# TYPE recording_stream_bitrate_kbps gauge
recording_stream_bitrate_kbps{meeting_id="m_123"} 1500

# HELP recording_compositing_lag_ms 合流延迟（关键 SLI）
# TYPE recording_compositing_lag_ms gauge
recording_compositing_lag_ms{meeting_id="m_123"} 850

func (r *RecordingSessionReconciler) handleScaling(ctx context.Context, session *v1alpha1.RecordingSession) (ctrl.Result, error) {
    lag := session.GetCompositingLag() // 从 Status 或实时查询获取
    streams := session.Status.CurrentStreams
    
    // 策略 1：合流延迟超阈值 -> 尝试垂直扩容
    if lag > 2000 && session.CanVerticalScale() {
        // 修改 Pod Resource Requests/Limits (需启用 InPlacePodVerticalScaling Feature Gate)
        // 或：标记需重建，Controller 删除 Pod 触发重建到更大规格节点
        return r.triggerVerticalScaleUp(ctx, session)
    }

    // 策略 2：流数激增但节点资源不足 -> 拓扑感知重调度
    if streams > session.Spec.MaxParticipants*80/100 && !session.NodeHasCapacity() {
        return r.triggerReschedule(ctx, session, "NodeResourcePressure")
    }

    // 策略 3：会议结束/人数骤降 -> 缩容回收
    if session.IsMeetingEnded() || streams < session.Spec.MinParticipants {
        return r.triggerScaleDown(ctx, session)
    }
    return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
}

type Healer interface {
    Diagnose(ctx context.Context, session *v1alpha1.RecordingSession) *DiagnosisResult
    Heal(ctx context.Context, session *v1alpha1.RecordingSession, diag *DiagnosisResult) (ctrl.Result, error)
}

// 示例：业务卡顿自愈器
type CompositingStallHealer struct{}

func (h *CompositingStallHealer) Diagnose(ctx context.Context, s *v1alpha1.RecordingSession) *DiagnosisResult {
    if s.Status.HealthStatus == v1alpha1.HealthStalled {
        return &DiagnosisResult{Type: FaultCompositingStall, Severity: LevelP1}
    }
    return nil
}

func (h *CompositingStallHealer) Heal(ctx context.Context, s *v1alpha1.RecordingSession, diag *DiagnosisResult) (ctrl.Result, error) {
    // 1. 尝试软重启：向 Sidecar 发送信号量，重置媒体管道
    if s.Status.RetryCount < 2 {
        if err := r.signalSidecar(ctx, s, syscall.SIGUSR2); err == nil {
            s.Status.RetryCount++
            r.Status().Update(ctx, s)
            return ctrl.Result{RequeueAfter: 30 * time.Second}, nil // 观察恢复情况
        }
    }
    // 2. 硬重建：删除 Pod，利用 StatefulSet/Deployment 或 Operator 直接重建
    // 保留 PVC (录制缓存)，迁移至新节点
    return r.recreatePodWithVolumeRetention(ctx, s)
}

affinity:
  podAntiAffinity:
    preferredDuringSchedulingIgnoredDuringExecution:
    - weight: 100
      podAffinityTerm:
        labelSelector:
          matchExpressions:
          - key: app.kubernetes.io/component
            operator: In
            values: ["recording-worker"]
        topologyKey: kubernetes.io/hostname
  nodeAffinity:
    requiredDuringSchedulingIgnoredDuringExecution:
      nodeSelectorTerms:
      - matchExpressions:
        - key: node-type
          operator: In
          values: ["media-compute"] # 专用媒体计算节点池
        - key: nvidia.com/gpu.present
          operator: Exists           # 需 GPU 加速合流

apiVersion: media.io/v1beta1
kind: RecordingWorkflow
metadata:
  name: webinar-2024-q4-planning
spec:
  workflowRef: "standard-webinar-v2" # 引用预定义模板
  params:
    meetingId: "m_998877"
    mainRoomLayout: "speaker-focus"
    breakoutRooms: 
      - roomId: "br_01"; layout: "grid-4"
      - roomId: "br_02"; layout: "grid-4"
  # 定义全局 SLA
  sla:
    maxTotalDuration: "4h"
    archiveDeadline: "30m" # 会议结束 30 分钟内必须完成归档
---
# 模板示例：标准网络研讨会流程
apiVersion: media.io/v1beta1
kind: WorkflowTemplate
metadata:
  name: standard-webinar-v2
spec:
  dag:
    tasks:
      - name: main-recording
        templateRef: recording-session-template
        arguments:
          roomId: "{{.params.meetingId}}"
          layout: "{{.params.mainRoomLayout}}"
      - name: breakout-recordings
        templateRef: recording-session-template
        # 动态扇出：为每个分组室创建并行录制任务
        withParam: "{{.params.breakoutRooms}}"
        arguments:
          roomId: "{{item.roomId}}"
          layout: "{{item.layout}}"
      - name: compositing-merge
        dependencies: ["main-recording", "breakout-recordings"]
        templateRef: compositing-job-template # 合流切片任务
        arguments:
          inputSources: "{{tasks.main-recording.outputs.artifacts}}, {{tasks.breakout-recordings.outputs.artifacts}}"
      - name: compliance-review
        dependencies: ["compositing-merge"]
        templateRef: manual-approval-template # 合规审批节点
        timeout: "1h"
      - name: cold-archive
        dependencies: ["compliance-review"]
        templateRef: cold-storage-template # 归档至低频存储

// Operator 侧：为每个 RecordingSession 生成唯一 DEK (Data Encryption Key)
func (r *RecordingSessionReconciler) generateEncryptionConfig(ctx context.Context, session *v1alpha1.RecordingSession) (*EncryptionConfig, error) {
    // 1. 向 KMS 请求生成 DEK (返回 Plaintext + Ciphertext)
    dekResp, err := r.kmsClient.GenerateDataKey(ctx, &kms.GenerateDataKeyInput{
        KeyId: aws.String(r.cfg.MasterKeyID),
        KeySpec: types.DataKeySpecAes256,
        EncryptionContext: map[string]string{
            "meeting_id": session.Spec.MeetingID,
            "session_uid": string(session.UID),
            "tenant_id": session.Labels["tenant.io/id"],
        },
    })
    if err != nil { return nil, err }

    // 2. 仅将 CiphertextBlob 注入 Pod 环境变量/Volume
    // 业务进程启动时调用 KMS Decrypt (需 Pod 绑定 IRSA/RRAM Role)
    return &EncryptionConfig{
        Algorithm: "AES-256-GCM",
        EncryptedDEK: base64.StdEncoding.EncodeToString(dekResp.CiphertextBlob),
        KMSKeyID: r.cfg.MasterKeyID,
    }, nil
}

type RetentionPolicy struct {
    // 热存储保留天数
    HotStorageDays int `json:"hotStorageDays"` 
    // 归档存储保留天数
    ColdStorageDays int `json:"coldStorageDays"`
    // 法律冻结标记：true 时冻结所有删除操作
    LegalHold bool `json:"legalHold,omitempty"` 
    // 脱敏规则引用
    DesensitizationRuleRef string `json:"desensitizationRuleRef,omitempty"`
}

// Controller 后台 Reconcile 逻辑片段
func (r *RecordingSessionReconciler) enforceRetention(ctx context.Context, session *v1alpha1.RecordingSession) {
    if session.Status.LegalHold { 
        r.log.Info("Legal hold active, skip retention enforcement", "session", session.Name)
        return 
    }
    
    age := time.Since(session.Status.EndTime.Time)
    // 阶段 1：热 -> 冷 (转存储类)
    if age > time.Duration(session.Spec.RetentionPolicy.HotStorageDays)*24*time.Hour && 
       session.Status.StorageTier == "Hot" {
        r.transitionToColdStorage(ctx, session)
    }
    // 阶段 2：冷 -> 删除 (彻底销毁)
    if age > time.Duration(session.Spec.RetentionPolicy.ColdStorageDays)*24*time.Hour {
        r.secureDelete(ctx, session) // 调用对象存储 DeleteObjectVersions + KMS 禁用 Key
    }
}

# 控制平面集群资源
apiVersion: policy.karmada.io/v1alpha1
kind: PropagationPolicy
metadata:
  name: recording-session-propagation
spec:
  resourceSelectors:
    - apiVersion: media.io/v1alpha1
      kind: RecordingSession
      labelSelector:
        matchLabels:
          # 仅分发标记了跨集群调度的会议
          scheduling.karmada.io/enabled: "true"
  placement:
    clusterAffinity:
      clusterNames: ["cn-beijing-prod", "cn-shanghai-prod", "edge-cdn-01"]
    # 基于实时资源水位的动态调度策略
    replicaScheduling: DynamicWeight
    replicaSchedulingType: Divided
    preference:
      - clusterAffinity:
          clusterNames: ["edge-cdn-01"] # 优先边缘节点，就近拉流
        weight: 80
      - clusterAffinity:
          clusterNames: ["cn-beijing-prod"]
        weight: 20

# PerformanceProfile 关键参数 (需节点重启生效，由运维团队维护 Node Pool)
apiVersion: performance.openshift.io/v2
kind: PerformanceProfile
metadata:
  name: media-compute-profile
spec:
  cpu:
    isolated: "2-31"          # 独占核，剔除核 0,1 给系统
    reserved: "0-1"
  hugepages:
    defaultHugepagesSize: "1G"
    pages:
      - size: "1G"
        count: 16             # 预留 16G HugePages，供 FFmpeg 零拷贝
  realTimeKernel:
    enabled: true             # 启用 RT 内核，降低调度抖动
  nodeSelector:
    node-role.kubernetes.io/media-worker: ""
  machineConfigPoolSelector:
    pools.operator.machineconfiguration.openshift.io/media-worker: ""

# ConfigMap: media-encoding-profile
data:
  h264_high_profile: |
    -c:v h264_nvenc -preset p4 -tune ll -rc:v vbr_hq -cq:v 28
    -b:v 3000k -maxrate 4000k -bufsize 6000k
    -g 60 -keyint_min 60 -sc_threshold 0
    -bf 3 -refs 3 -profile:v high -level 4.2
    -movflags +faststart+frag_keyframe+empty_moov
  # 针对弱网优化的 SVC (Scalable Video Coding) 参数
  svc_screen_share: |
    -c:v libvpx-vp9 -svc 3 -threads 4 -speed 2
    -b:v 1500k -maxrate 2000k -deadline realtime
    -error-resilient 1 -frame-parallel 1

# .gitlab-ci.yml 片段
stages:
  - unit-test
  - integration-test
  - chaos-test   # 关键门禁
  - canary-deploy

chaos_verification:
  stage: chaos-test
  image: chaosiq/chaostoolkit:latest
  script:
    - kind create cluster --name chaos-test
    - helm install operator ./charts/recording-operator -n media-system
    - kubectl apply -f test/chaos/scenarios/  # 批量应用 ChaosExperiment
    - chaosiq run test/chaos/verification.yaml --settings=strict  # 严格模式：任何实验失败即流水线失败
  rules:
    - if: $CI_PIPELINE_SOURCE == "merge_request_event"

// pkg/scheduler/costaware/cost_aware.go
func (p *CostAwarePlugin) Score(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {
    nodeInfo, _ := p.handle.SnapshotSharedLister().NodeInfos().Get(nodeName)
    node := nodeInfo.Node()
    
    // 1. 识别实例类型
    instanceType := node.Labels["node.kubernetes.io/instance-type"]
    lifecycle := node.Labels["node.kubernetes.io/lifecycle"] // "spot" | "on-demand"
    
    // 2. 成本评分 (Spot 权重高)
    baseScore := int64(50)
    if lifecycle == "spot" {
        baseScore += 40 // 优先调度到 Spot
        // 3. 风险惩罚：近 24h 中断率 > 5% 则降权
        if p.interruptionPredictor.GetRisk(instanceType) > 0.05 {
            baseScore -= 30
        }
    }
    
    // 4. 资源碎片率惩罚 (Binpacking)
    allocatable := node.Status.Allocatable
    requested := nodeInfo.Requested
    cpuFrag := 1.0 - float64(requested.Cpu())/float64(allocatable.Cpu().Value())
    baseScore += int64(cpuFrag * 10)
    
    return baseScore, nil
}

# 查看会议实时质量仪表盘 (聚合 Prometheus + 业务 Status)
$ kubectl recording inspect m_123456 --live
MEETING m_123456 | SESSION s_abcdef | PHASE: Running | NODE: cn-bj-192.168.1.50
STREAMS: 48/50 | LAG: 120ms (P99) | CPU: 12.4/16 Core | GPU Mem: 8.2/16 GiB
STORAGE: Hot (12.4 GB) | UPLOAD: 1.2 Gbps
ALERTS: [WARN] Stream #45 packet loss 2.1%

# 一键触发故障演练 (仅限 Staging)
$ kubectl recording chaos inject m_123456 --type=network-partition --duration=30s

# 手动触发归档 (提前结束会议)
$ kubectl recording finalize m_123456 --reason="host_ended"