apiVersion: api/v1
kind: Plan
name: binlog-kafka
schedule: "@daily"
failFast: false
tags:
  - binlog
testbed:
  generateName: binlog-kafka-testbed
  resourcePool: brie
  reclaimPolicy:
    gcStrategy:
      onFailure:
        artifact:
          monitor: true
          log: true
        reserveEnv:
          durationOnWorkingHour: "{{meta.parameters.onFailureReserveDurationOnWorkingHour}}"
        notification:
          feishu: >-
            https://open.feishu.cn/open-apis/bot/v2/hook/54a44124-87d9-4104-8f6d-176f09706ca3
      onSuccess:
        notification:
          feishu: >-
            https://open.feishu.cn/open-apis/bot/v2/hook/54a44124-87d9-4104-8f6d-176f09706ca3
  items:
    - name: downstream
      type: MYSQL_NODE
      spec:
        storageClassName: fast-disks
        image: "{{meta.parameters.mysql_image}}"
        port: 3306
        env:
          MYSQL_ALLOW_EMPTY_PASSWORD: "yes"
          MYSQL_ROOT_HOST: "%"
        resources:
          requests:
            memory: 4Gi
            cpu: 4000m
            storage: 100Gi
        volumeMount:
          mountPath: "/var/lib/mysql" # mountPath of "storage" request
    - name: upstream
      type: TIDB_CLUSTER
      spec:
        imagePullPolicy: Always
        pd:
          replicas: 1
          requests:
            cpu: 1000m
            memory: 4Gi
          limits:
            cpu: 1000m
            memory: 4Gi
          baseImage: "{{meta.parameters.pd_base_image}}"
          version: "{{meta.parameters.pd_component_version}}"
        tikv:
          replicas: 1
          requests:
            cpu: 4000m
            memory: 16Gi
          limits:
            cpu: 4000m
            memory: 16Gi
          baseImage: "{{meta.parameters.tikv_base_image}}"
          version: "{{meta.parameters.tikv_component_version}}"
        tidb:
          replicas: 1
          requests:
            cpu: 4000m
            memory: 8Gi
          limits:
            cpu: 4000m
            memory: 8Gi
          baseImage: "{{meta.parameters.tidb_base_image}}"
          version: "{{meta.parameters.tidb_component_version}}"
        pump:
          baseImage: "{{meta.parameters.binlog_base_image}}"
          replicas: 1
          requests:
            storage: 30Gi
          config: |
            addr = "0.0.0.0:8250"
            gc = 7
            heartbeat-interval = 2
          version: "{{meta.parameters.binlog_component_version}}"
        version: "{{meta.parameters.base_image_version}}"
    - name: upstreamdrainer
      type: TIDB_DRAINER
      spec:
        clusterName: upstream
        clusterVersion: "{{meta.parameters.binlog_component_version}}"
        baseImage: "{{meta.parameters.binlog_base_image}}"
        storage: 10Gi
        initialCommitTs: "-1"
        resources:
          requests:
            cpu: 2000m
            memory: 4Gi
        config: |
          detect-interval = 10
          [syncer]
          worker-count = 16
          txn-batch = 20
          disable-dispatch = false
          ignore-schemas = "INFORMATION_SCHEMA,PERFORMANCE_SCHEMA,mysql"
          safe-mode = false
          db-type = "kafka"
          [syncer.to]
          kafka-addrs = "kafka-kafka:9092"
          kafka-max-messages = 1024
          topic-name = "binlog"
    - name: workload
      type: WORKLOAD_NODE
      spec:
        container:
          name: sysbench
          image: "{{meta.parameters.bench_toolset_image}}"
          command:
            - tail
            - "-f"
            - /dev/null
          resources:
            requests:
              memory: 4Gi
              cpu: 4000m
    - name: kafka
      type: KAFKA_CLUSTER
      spec:
        zkspec:
          replicas: 1
          storageClassName: fast-disks
          image: "k8s.gcr.io/kubernetes-zookeeper:1.0-3.4.10"
          resources:
            requests:
              memory: 4Gi
              cpu: 4000m
              storage: 20Gi
        kafkaspec:
          replicas: 2
          storageClassName: fast-disks
          imagePullPolicy: IfNotPresent
          image: "registry-mirror.pingcap.net/wurstmeister/kafka:2.13-2.7.1"
          command:
            - "sh"
            - "-c"
            - "kafka-server-start.sh /opt/kafka/config/server.properties --override broker.id=${HOSTNAME##*-}\
              \ --override listeners=PLAINTEXT://:9092 --override zookeeper.connect=$ZK_CONNECT\
              \ --override log.dir=/var/lib/kafka --override message.max.bytes=1073741824\
              \ --override replica.fetch.max.bytes=1073741824 --override fetch.message.max.bytes=1073741824"
          env:
            KAFKA_HEAP_OPTS: "-Xmx512M -Xms512M -XX:ParallelGCThreads=1"
            KAFKA_OPTS: "-Dlogging.level=INFO"
          resources:
            requests:
              memory: 16Gi
              cpu: 4000m
              storage: 100Gi
    - name: arbiter
      type: WORKLOAD_NODE
      spec:
        container:
          name: arbiter
          image: hub-new.pingcap.net/qa/arbiter:latest
          command:
            - sh
            - "-c"
            - "/arbiter --config /arbiter.toml"
          resources:
            requests:
              memory: 8Gi
              cpu: 4000m
steps:
  - name: binlog-kafka-basic
    caseName: binlog_kafka_basic
    arguments:
      - name: upstream
        value: "{{testbed.upstream}}"
      - name: downstream
        value: "{{testbed.downstream}}"
      - name: upstreamdrainer
        value: "{{testbed.upstreamdrainer}}"
      - name: workload
        value: "{{testbed.workload}}"
      - name: arbiter
        value: "{{testbed.workload}}"
  - name: binlog-kafka-large-transcation
    caseName: binlog_kafka_large_transcation
    arguments:
      - name: upstream
        value: "{{testbed.upstream}}"
      - name: downstream
        value: "{{testbed.downstream}}"
      - name: upstreamdrainer
        value: "{{testbed.upstreamdrainer}}"
      - name: workload
        value: "{{testbed.workload}}"
      - name: arbiter
        value: "{{testbed.workload}}"