Argo Event云原生的事件驅(qū)動(dòng)架構(gòu)怎么用

這篇“Argo Event云原生的事件驅(qū)動(dòng)架構(gòu)怎么用”文章的知識(shí)點(diǎn)大部分人都不太理解，所以小編給大家總結(jié)了以下內(nèi)容，內(nèi)容詳細(xì)，步驟清晰，具有一定的借鑒價(jià)值，希望大家閱讀完這篇文章能有所收獲，下面我們一起來看看這篇“Argo Event云原生的事件驅(qū)動(dòng)架構(gòu)怎么用”文章吧。

1、Argo Event簡(jiǎn)介

Event事件大家都很熟悉，可以說Kubernetes就是完全由事件驅(qū)動(dòng)的，不同的controller manager本質(zhì)就是實(shí)現(xiàn)了不同的事件處理函數(shù)，比如所有ReplicaSet對(duì)象是由ReplicaSetController控制器管理，該控制器通過Informer監(jiān)聽ReplicaSet以及其關(guān)聯(lián)的Pod的事件變化，從而維持運(yùn)行狀態(tài)和我們聲明spec保持一致。

當(dāng)然Kubernetes無論是什么Controller，其監(jiān)聽和處理的都是內(nèi)部事件，而在應(yīng)用層上我們也有很多外部事件，比如CICD事件、Webhook事件、日志事件等等，如何處理這些事件呢，目前Kubernetes原生是無法實(shí)現(xiàn)的。

當(dāng)然你可以自己實(shí)現(xiàn)一個(gè)event handler運(yùn)行在Kubernetes平臺(tái)，不過實(shí)現(xiàn)難度也不小。而Argo Event組件完美解決了這個(gè)問題。

如圖是Argo Event官方提供的的流程圖：

首先事件源EventSource可以是Webhook、S3、Github、SQS等等，中間會(huì)經(jīng)過一個(gè)叫Gateway(新版本叫EventBus)的組件，更準(zhǔn)確地說老版本原來gateway的配置功能已經(jīng)合并到EventSource了，EventBus是新引入的組件，后端默認(rèn)基于高性能分布式消息中間件NATS[1]實(shí)現(xiàn)，當(dāng)然其他中間件比如Kafka也是可以的。

這個(gè)EventBus可以看做是事件的一個(gè)消息隊(duì)列，消息生產(chǎn)者連接EvenSource，EventSource又連接到Sensor。更詳細(xì)地說EvenSource把事件發(fā)送給EvenBus，Sensor會(huì)訂閱EvenBus的消息隊(duì)列，EvenBus負(fù)責(zé)把事件轉(zhuǎn)發(fā)到已訂閱該事件的Sensor組件，EventSorce在上圖中沒有體現(xiàn)，具體設(shè)計(jì)文檔可以參考Argo-events Enhancement Proposals[2]。

有些人可能會(huì)說為什么EventBus不直接到Trigger，中間引入一個(gè)Sensor，這主要是兩個(gè)原因，一是為了使事件轉(zhuǎn)發(fā)和處理松耦合，二是為了實(shí)現(xiàn)Trigger事件的參數(shù)化，通過Sensor不僅可以實(shí)現(xiàn)事件的過濾，還可以實(shí)現(xiàn)事件的參數(shù)化，比如后面的Trigger是創(chuàng)建一個(gè)Kubernetes Pod，那這個(gè)Pod的metadata、env等，都可以根據(jù)事件內(nèi)容進(jìn)行填充。

Sensor組件注冊(cè)關(guān)聯(lián)了一個(gè)或者多個(gè)觸發(fā)器，這些觸發(fā)器可以觸發(fā)AWS Lambda事件、Argo Workflow事件、Kubernetes Objects等，通俗簡(jiǎn)單地說，可以執(zhí)行Lambda函數(shù)，可以動(dòng)態(tài)地創(chuàng)建Kubernetes的對(duì)象或者創(chuàng)建前面的介紹的Workflow。

還記得前面介紹的Argo Rollout嗎，我們演示了手動(dòng)promote實(shí)現(xiàn)應(yīng)用發(fā)布或者回滾，通過Argo Event就可以很完美地和測(cè)試平臺(tái)或者CI/CD事件結(jié)合起來，實(shí)現(xiàn)自動(dòng)應(yīng)用自動(dòng)發(fā)布或者回滾。

2、一個(gè)簡(jiǎn)單的Webhook例子

關(guān)于Argo Event的部署非常簡(jiǎn)單，直接通過kubecl apply或者h(yuǎn)elm均可，可以參考文檔Installation[3]，這里不再贅述。

Argo Event部署完成后注意還需要部署EventBus，官方推薦使用NATS中間件，文檔中有部署NATS stateful的文檔。

接下來我們以一個(gè)最簡(jiǎn)單的Webhook事件為例，從而了解Argo Event的幾個(gè)組件功能以及用法。

首先按照前面的介紹，我們需要先定義EventSource：

apiVersion: argoproj.io/v1alpha1
kind: EventSource
metadata:
  name: webhook
spec:
  service:
    ports:
      - port: 12000
        targetPort: 12000
  webhook:
    webhook_example:
      port: "12000"
      endpoint: /webhook
      method: POST

這個(gè)EventSource定義了一個(gè)webhook webhook_example，端口為12000，路徑為/webhook，一般Webhook為POST方法，因此該Webhhok處理器我們配置只接收POST方法。

為了把這個(gè)Webhook EventSource暴露，我們還創(chuàng)建了一個(gè)Service，端口也是12000。

此時(shí)我們可以手動(dòng)curl該Service:

# kubectl get svc -l eventsource-name=webhook
NAME                      TYPE        CLUSTER-IP    EXTERNAL-IP   PORT(S)     AGE
webhook-eventsource-svc   ClusterIP   10.96.93.24   <none>        12000/TCP   5m49s
# curl -X POST -d '{}' 10.96.93.24:12000/webhook
success

當(dāng)然此時(shí)由于沒有注冊(cè)任何的Sensor，因此什么都不會(huì)發(fā)生。

接下來我們定義Sensor:

首先在dependencies中定義了訂閱的EventSource以及具體的Webhook，由于一個(gè)EventSource可以定義多個(gè)Webhook，因此必須同時(shí)指定EventSource和Webhook兩個(gè)參數(shù)。

在Trigger中我們定義了對(duì)應(yīng)Action為create一個(gè)workflow，這個(gè)workflow的spec定義在resource中配置。

最后的parameters部分定義了workflow的參數(shù)，這些參數(shù)值從event中獲取，這里我們會(huì)把整個(gè)event都當(dāng)作workflow的input。當(dāng)然你可以通過dataKey只汲取body部分:dataKey: body.message。

此時(shí)我們?cè)俅蝐url這個(gè)webhook事件:

curl -X POST -d '{"message": "HelloWorld!"}' 10.96.93.24:12000/webhook

此時(shí)我們獲取argo workflow列表發(fā)現(xiàn)新創(chuàng)建了一個(gè)實(shí)例：

# argo list
NAME            STATUS      AGE   DURATION   PRIORITY
webhook-8xt4s   Succeeded   1m    18s        0

查看workflow輸出如下:

由于我們是把整個(gè)event作為workflow input發(fā)過去的，因此data內(nèi)容部分是base64編碼，我們可以查看解碼后的內(nèi)容如下:

{
  "header": {
    "Accept": [
      "*/*"
    ],
    "Content-Length": [
      "26"
    ],
    "Content-Type": [
      "application/x-www-form-urlencoded"
    ],
    "User-Agent": [
      "curl/7.58.0"
    ]
  },
  "body": {
    "message": "HelloWorld!"
  }
}

從這里我們也可以看出Event包含兩個(gè)部分，一個(gè)是context，一個(gè)是data，data中又包含header部分以及body部分，在parameters中可以通過Key獲取任意部分內(nèi)容。

如上的webhook觸發(fā)是通過手動(dòng)curl的，你可以很容易地在github或者bitbucket上配置到webhook中，這樣一旦代碼有更新就能觸發(fā)這個(gè)事件了。

3、Kubernetes觸發(fā)AWS Lambda函數(shù)

前面的例子中的EventSource使用了Webhook，除了Webhook，Argo Event還支持很多的EventSource，比如:

• amqp

• aws-sns

• aws-sqs

• github/gitlab

• hdfs

• kafka

• redis

• Kubernetes resource

• ...

Trigger也同樣支持很多，比如:

• aws lambda

• amqp

• kafka

• ...

如上官方都提供了非常豐富的例子，可以參考argo events examples[4]。

這里以Kubernetes resource事件源為例，這個(gè)事件監(jiān)聽Kubernetes的資源狀態(tài)，比如Pod創(chuàng)建、刪除等，這里以創(chuàng)建Pod為例:

apiVersion: argoproj.io/v1alpha1
kind: EventSource
metadata:
  name: k8s-resource-demo
spec:
  template:
    serviceAccountName: argo-events-sa
  resource:
    pod_demo:
      namespace: argo-events
      version: v1
      resource: pods
      eventTypes:
        - ADD
      filter:
        afterStart: true
        labels:
          - key: app
            operation: "=="
            value: my-pod

如上例子監(jiān)聽Pods的ADD事件，即創(chuàng)建Pod，filter中過濾只有包含app=my-pod標(biāo)簽的Pod，特別需要注意的是使用的serviceaccount argo-events-sa必須具有Pod的list、watch權(quán)限。

接下來我們使用AWS Lambda觸發(fā)器，Lambda函數(shù)已經(jīng)在AWS提前創(chuàng)建好:

這個(gè)Lambda函數(shù)很簡(jiǎn)單，直接返回event本身。

創(chuàng)建Sensor如下：

apiVersion: argoproj.io/v1alpha1
kind: Sensor
metadata:
  name: aws-lambda-trigger-demo
spec:
  template:
    serviceAccountName: argo-events-sa
  dependencies:
    - name: test-dep
      eventSourceName: k8s-resource-demo
      eventName: pod_demo
  triggers:
    - template:
        name: lambda-trigger
        awsLambda:
          functionName: hello
          accessKey:
            name: aws-secret
            key: accesskey
          secretKey:
            name: aws-secret
            key: secretkey
          namespace: argo-events
          region: cn-northwest-1
          payload:
            - src:
                dependencyName: test-dep
                dataKey: body.name
              dest: name

如上AWS access key和access secret需要提前放到aws-secret中。

此時(shí)我們創(chuàng)建一個(gè)新的Pod my-pod:

apiVersion: v1
kind: Pod
metadata:
  labels:
    app: my-pod
  name: my-pod
spec:
  containers:
  - image: nginx
    name: my-pod
  dnsPolicy: ClusterFirst
  restartPolicy: Always

當(dāng)Pod啟動(dòng)后，我們發(fā)現(xiàn)AWS Lambda函數(shù)被觸發(fā)執(zhí)行:

4、event filter

前面的例子中webhook中所有的事件都會(huì)被sensor觸發(fā)，我們有時(shí)不需要處理所有的事件，Argo Event支持基于data以及context過濾，比如我們只處理message為hello或者為hey的事件，其他消息忽略，只需要在原來的dependencies中test-dep增加filter即可:

dependencies:
- name: test-dep
  eventSourceName: webhook
  eventName: webhook_example
  filters:
  - name: data-filter
    data:
    - path: body.message
      type: string
      value:
      - "hello"
      - "hey"

filter指定了基于data過濾，過濾的字段為body.message，匹配的內(nèi)容為hello、hey。

5、trigger policy

trigger policy主要用來判斷最后觸發(fā)器執(zhí)行的結(jié)果是成功還是失敗，如果是創(chuàng)建Kubernetes資源比如Workflow，可以根據(jù)Workflow最終狀態(tài)決定這個(gè)Trigger的執(zhí)行結(jié)果，而如果是觸發(fā)一個(gè)HTTP或者AWS Lambda，則需要自定義policy status。

awsLambda:
  functionName: hello
  accessKey:
    name: aws-secret
    key: accesskey
  secretKey:
    name: aws-secret
    key: secretkey
  namespace: argo-events
  region: us-east-1
  payload:
  - src:
      dependencyName: test-dep
      dataKey: body.message
    dest: message
  policy:
    status:
        allow:
         - 200
         - 201

如上表示當(dāng)AWS Lambda返回200或者201時(shí)表示Trigger成功。

以上就是關(guān)于“Argo Event云原生的事件驅(qū)動(dòng)架構(gòu)怎么用”這篇文章的內(nèi)容，相信大家都有了一定的了解，希望小編分享的內(nèi)容對(duì)大家有幫助，若想了解更多相關(guān)的知識(shí)內(nèi)容，請(qǐng)關(guān)注億速云行業(yè)資訊頻道。