Argo Events #

# 创建命名空间
kubectl create namespace argo-events

# 安装 EventSource、Sensor、EventBus 控制器
kubectl apply -f https://raw.githubusercontent.com/argoproj/argo-events/stable/manifests/install.yaml

# 可选：安装 Validating Admission Webhook
kubectl apply -f https://raw.githubusercontent.com/argoproj/argo-events/stable/manifests/install-validating-webhook.yaml

# 部署 NATS Native EventBus
kubectl apply -n argo-events \
  -f https://raw.githubusercontent.com/argoproj/argo-events/stable/examples/eventbus/native.yaml

# 确认 EventBus Pod 运行
kubectl -n argo-events get pods | grep eventbus

# EventBus 示例（NATS Native）
apiVersion: argoproj.io/v1alpha1
kind: EventBus
metadata:
  name: default
spec:
  nats:
    native:
      replicas: 3              # 最少 3 个副本
      auth: token               # 认证策略：none 或 token
      # persistence:             # 持久化存储（可选）
      #   storageClassName: standard
      #   accessMode: ReadWriteOnce
      #   volumeSize: 10Gi

apiVersion: argoproj.io/v1alpha1
kind: EventSource
metadata:
  name: webhook
spec:
  service:
    ports:
      - port: 12000
        targetPort: 12000
  webhook:
    example:
      port: "12000"            # HTTP 服务器端口
      endpoint: /example       # 监听路径
      method: POST             # 允许的 HTTP 方法

kubectl apply -n argo-events -f webhook-eventsource.yaml

# EventSource Controller 会自动创建 Pod 和 Service
kubectl -n argo-events get pods | grep webhook
kubectl -n argo-events get svc | grep webhook

# Sensor RBAC（允许触发 Argo Workflow）
kubectl apply -n argo-events \
  -f https://raw.githubusercontent.com/argoproj/argo-events/master/examples/rbac/sensor-rbac.yaml

# Workflow RBAC（允许 Argo Workflow 运行）
kubectl apply -n argo-events \
  -f https://raw.githubusercontent.com/argoproj/argo-events/master/examples/rbac/workflow-rbac.yaml

apiVersion: argoproj.io/v1alpha1
kind: Sensor
metadata:
  name: webhook
spec:
  template:
    serviceAccountName: operate-workflow-sa    # 触发 Argo Workflow 需要的 SA
  dependencies:
    - name: test-dep                          # 事件依赖名称
      eventSourceName: webhook                # 关联的 EventSource
      eventName: example                      # EventSource 中的事件名称
  triggers:
    - template:
        name: webhook-workflow-trigger
        k8s:
          operation: create                   # 操作类型：create/update/delete
          source:
            resource:                         # 要创建的 K8s 资源
              apiVersion: argoproj.io/v1alpha1
              kind: Workflow
              metadata:
                generateName: webhook-        # 自动生成名称
              spec:
                entrypoint: whalesay
                arguments:
                  parameters:
                    - name: message
                      value: hello world       # 默认值，会被事件 payload 覆盖
                templates:
                  - name: whalesay
                    inputs:
                      parameters:
                        - name: message
                    container:
                      image: docker/whalesay:latest
                      command: [cowsay]
                      args: ["{{inputs.parameters.message}}"]
          parameters:                          # 参数绑定
            - src:
                dependencyName: test-dep
                dataKey: body                  # 从事件 payload 的 body 提取数据
              dest: spec.arguments.parameters.0.value

# 端口转发到 EventSource Service
kubectl -n argo-events port-forward svc/webhook-eventsource-svc 12000:12000

# 发送 POST 请求触发事件
curl -d '{"message":"this is my first webhook"}' \
  -H "Content-Type: application/json" \
  -X POST http://localhost:12000/example

# 验证 Argo Workflow 被触发
kubectl -n argo-events get workflows | grep webhook

# 查看 Workflow 日志
kubectl -n argo-events logs -f $(kubectl -n argo-events get workflows -l argo-events=sensor-name -o name | head -1)

apiVersion: argoproj.io/v1alpha1
kind: EventSource
metadata:
  name: webhook
spec:
  service:
    ports:
      - port: 12000
        targetPort: 12000
  webhook:
    example:
      port: "12000"
      endpoint: /example
      method: POST
    # 一个 EventSource 可运行多个 HTTP 服务器
    # example-secure:
    #   port: "13000"
    #   endpoint: /secure
    #   method: POST
    #   serverCertSecret:        # TLS 证书
    #     name: my-secret
    #     key: cert-key
    #   serverKeySecret:
    #     name: my-secret
    #     key: pk-key

apiVersion: argoproj.io/v1alpha1
kind: EventSource
metadata:
  name: calendar
spec:
  calendar:
    # 每 10 秒触发一次
    example-with-interval:
      interval: 10s

    # Cron 表达式：每小时的第 30 分钟触发
    example-with-schedule:
      schedule: "30 * * * *"
      timezone: "Asia/Shanghai"       # 时区
      metadata:                        # 随事件一起发送的附加数据
        hello: world

    # 带补发机制的 Cron（错过的事件会被补发）
    example-with-catchup:
      schedule: "* * * * *"
      persistence:
        catchup:
          enabled: true
          maxDuration: 5m
        configMap:
          createIfNotExist: true
          name: calendar-state

    # 排除特定日期
    schedule-with-exclusion:
      schedule: "30 * * * *"
      exclusionDates:
        - "EXDATE:20250101T150405Z"

apiVersion: argoproj.io/v1alpha1
kind: EventSource
metadata:
  name: k8s-events
spec:
  template:
    serviceAccountName: resource-watcher-sa    # 需要 list/watch 权限
  resource:
    # 监听 Deployment 变化
    example-deployment:
      namespace: default
      group: "apps"
      version: "v1"
      resource: "deployments"
      eventTypes:
        - ADD              # 资源创建
        - UPDATE           # 资源更新
        - DELETE           # 资源删除

# NATS JetStream EventBus
apiVersion: argoproj.io/v1alpha1
kind: EventBus
metadata:
  name: default
spec:
  jetstream:
    version: latest
    replicas: 3                    # 最少 3 个副本
    persistence:                   # 持久化
      storageClassName: standard
      accessMode: ReadWriteOnce
      volumeSize: 10Gi
    # containerTemplate:           # 资源限制
    #   resources:
    #     requests:
    #       cpu: 100m
    #       memory: 128Mi
    # streamReplicas: 3            # Stream 副本数
    # maxAge: 72h                  # 消息最大保留时间

# Kafka EventBus
apiVersion: argoproj.io/v1alpha1
kind: EventBus
metadata:
  name: default
spec:
  kafka:
    brokers:
      - "kafka-broker-1:9092"
      - "kafka-broker-2:9092"
      - "kafka-broker-3:9092"
    topic: argo-events
    consumerGroup: argo-events-sensor
    # TLS 配置（可选）
    # tls:
    #   caCertPath: /path/to/ca.crt
    #   clientCertPath: /path/to/client.crt
    #   clientKeyPath: /path/to/client.key

apiVersion: argoproj.io/v1alpha1
kind: Sensor
metadata:
  name: my-sensor
spec:
  template:
    serviceAccountName: operate-workflow-sa    # 触发 K8s 资源时需要
    # containerTemplate:                       # 资源限制
    #   resources:
    #     requests:
    #       memory: 128Mi
  dependencies:                                # 事件依赖列表
    - name: dep-1                              # 依赖名称（必须唯一）
      eventSourceName: webhook                 # EventSource 名称
      eventName: example                       # EventSource 中的事件名
      filters:                                 # 过滤条件（可选）
        - name: data-filter
          data:
            path: body.type                    # 从事件 payload 提取的路径
            type: string
            value:
              - "push"                         # 匹配的值（支持多个 OR 匹配）
  triggers:                                    # 触发器列表
    - template: { ... }
  # errorOnFailedRound: true                   # 触发失败时是否标记 Sensor 为错误

spec:
  dependencies:
    # AND 逻辑：所有依赖都满足才触发
    - name: git-push
      eventSourceName: github
      eventName: push
      filters:
        - name: repo-filter
          data:
            path: body.repository.name
            type: string
            value: ["my-repo"]

    - name: schedule
      eventSourceName: calendar
      eventName: nightly

  triggers:
    - template:
        name: ci-trigger
        conditions: "git-push && schedule"     # 依赖条件（AND 逻辑）
        # conditions: "git-push || schedule"   # OR 逻辑
        k8s:
          operation: create
          source: { ... }

dependencies:
  - name: filtered-webhook
    eventSourceName: webhook
    eventName: example
    filters:
      # 数据过滤：从 payload 提取路径并匹配值
      - name: data-filter
        data:
          path: body.event_type               # JSON 路径
          type: string
          value:
            - "push"
            - "pull_request"
          comparator: "="                      # 比较运算符

      # 上下文过滤：从 CloudEvents context 提取
      - name: context-filter
        context:
          source: "webhook"                    # 匹配事件源名称
          type: "webhook"                     # 匹配事件类型

      # 时间过滤
      - name: time-filter
        time:
          start: "2025-01-01T00:00:00Z"
          stop: "2025-12-31T23:59:59Z"

triggers:
  - template:
      name: workflow-trigger
      conditions: "git-push"           # 可选：满足条件时触发
      argoWorkflow:
        operation: create             # create | submit | resubmit | retry | resume | suspend
        source:
          resource:
            apiVersion: argoproj.io/v1alpha1
            kind: Workflow
            metadata:
              generateName: ci-
            spec:
              entrypoint: build-and-deploy
              arguments:
                parameters:
                  - name: repo-url
                    value: ""
                  - name: revision
                    value: ""
              templates:
                - name: build-and-deploy
                  inputs:
                    parameters:
                      - name: repo-url
                      - name: revision
                  steps:
                    - - name: checkout
                        template: checkout
                        arguments:
                          parameters:
                            - name: repo-url
                              value: "{{inputs.parameters.repo-url}}"
                            - name: revision
                              value: "{{inputs.parameters.revision}}"
        parameters:
          - src:
              dependencyName: git-push
              dataKey: body.repository.clone_url
            dest: spec.arguments.parameters.0.value
          - src:
              dependencyName: git-push
              dataKey: body.ref        # 提取分支名
            dest: spec.arguments.parameters.1.value

triggers:
  - template:
      name: k8s-trigger
      k8s:
        operation: create            # create | update | delete
        source:
          resource:
            apiVersion: apps/v1
            kind: Deployment
            metadata:
              generateName: triggered-deployment-
            spec:
              replicas: 1
              selector:
                matchLabels:
                  app: triggered
              template:
                metadata:
                  labels:
                    app: triggered
                spec:
                  containers:
                    - name: app
                      image: myapp:latest

triggers:
  - template:
      name: http-trigger
      http:
        url: http://my-server.example.com/api/process
        method: POST
        parameters:
          - src:
              dependencyName: webhook
              dataKey: body.message
            dest: query.message       # 注入到 query 参数
        payload:
          - src:
              dependencyName: webhook
              dataKey: body
            dest: payload           # 注入到 request body

triggers:
  - template:
      name: slack-notification
      slack:
        message: "Deployment triggered by Argo Events!"
        slackToken:                  # K8s Secret 包含 Slack API Token
          name: slack-secret
          key: token
        channel: "#deployments"

{
  "context": {
    "type": "webhook",
    "specversion": "0.3",
    "source": "webhook",
    "id": "unique_event_id",
    "time": "2025-01-01T00:00:00Z",
    "datacontenttype": "application/json",
    "subject": "example"
  },
  "data": {
    "header": {                       // HTTP 请求头
      "Content-Type": ["application/json"]
    },
    "body": {                         // HTTP 请求体
      "message": "hello world"
    }
  }
}

parameters:
  # 1. 从事件 Data 提取（最常用）
  - src:
      dependencyName: test-dep
      dataKey: body.message           # 从 data.body.message 提取
    dest: spec.arguments.parameters.0.value

  # 2. 从事件 Context 提取
  - src:
      dependencyName: test-dep
      contextKey: type                # 从 context.type 提取
    dest: spec.arguments.parameters.1.value

  # 3. 使用默认值（当 key 不存在时）
  - src:
      dependencyName: test-dep
      dataKey: body.nonexistent       # 不存在的 key
    dest: spec.arguments.parameters.2.value
    value: "default-value"           # 默认值

  # 4. Sprig 模板变换（对提取的数据做处理）
  - src:
      dependencyName: test-dep
      dataTemplate: "{{ .Input.body.message | title }}"    # 首字母大写
    dest: spec.arguments.parameters.3.value

  - src:
      dependencyName: test-dep
      contextTemplate: "{{ .Input.subject | title }}"
    dest: spec.arguments.parameters.4.value

  # 5. 使用 useRawData（保留原始类型：number/boolean/json）
  - src:
      dependencyName: test-dep
      dataKey: body.count
    dest: spec.arguments.parameters.5.value
    useRawData: true                  # 不设则解析为 string

  # 6. append/prepend 操作（追加到现有值）
  - src:
      dependencyName: test-dep
      dataKey: body.message
    dest: metadata.generateName
    operation: append                 # append | prepend

# Resource EventSource 的 RBAC
apiVersion: v1
kind: ServiceAccount
metadata:
  name: resource-watcher-sa
  namespace: argo-events
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: deployments-watcher
  namespace: argo-events
rules:
  - apiGroups: ["apps"]
    resources: ["deployments"]
    verbs: ["list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: deployments-watcher-binding
  namespace: argo-events
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: Role
  name: deployments-watcher
subjects:
  - kind: ServiceAccount
    name: resource-watcher-sa
    namespace: argo-events

triggers:
  - template:
      name: argocd-sync
      k8s:
        operation: create
        source:
          resource:
            apiVersion: argoproj.io/v1alpha1
            kind: Application
            metadata:
              generateName: sync-trigger-
            spec:
              # ...

triggers:
  - template:
      name: rollout-trigger
      argoRollout:
        # 触发 Rollout 的 promote/abort/restart 等操作
        rolloutName: my-rollout
        # ...

# Helm 安装
helm repo add argo https://argoproj.github.io/argo-helm
helm install argo-events argo/argo-events -n argo-events --create-namespace

# Kustomize 安装
# kustomization.yaml:
# bases:
# - github.com/argoproj/argo-events/manifests/cluster-install

# 查看 EventSource 状态
kubectl -n argo-events get eventsource <name> -o yaml

# 查看 Sensor 状态
kubectl -n argo-events get sensor <name> -o yaml

# 查看 EventBus 状态
kubectl -n argo-events get eventbus default -o yaml

# 查看 EventSource Pod 日志
kubectl -n argo-events logs -l eventsource-name=<name>

# 查看 Sensor Pod 日志
kubectl -n argo-events logs -l sensor-name=<name>

# 查看 Controller 日志
kubectl -n argo-events logs -l app=argo-events-controller

# 验证资源配置（lint）
argo-events lint eventsource.yaml sensor.yaml

# 端口转发测试 Webhook
kubectl -n argo-events port-forward svc/<eventsource-svc> 12000:12000

# 发送测试请求
curl -d '{"message":"test"}' -H "Content-Type: application/json" \
  -X POST http://localhost:12000/example

# 检查触发的 Workflow
kubectl -n argo-events get workflows