ArgoEvent云原生的事件驱动架构怎么用

这篇“Argo Event云原生的事件驱动架构怎么用”文章的知识点大部分人都不太理解，所以小编给大家总结了以下内容，内容详细，步骤清晰，具有一定的借鉴价值，希望大家阅读完这篇文章能有所收获，下面我们一起来看看这篇“Argo Event云原生的事件驱动架构怎么用”文章吧。

创新互联公司是专业的定西网站建设公司，定西接单;提供做网站、网站制作,网页设计,网站设计,建网站,PHP网站建设等专业做网站服务;采用PHP框架,可快速的进行定西网站开发网页制作和功能扩展;专业做搜索引擎喜爱的网站,专业的做网站团队,希望更多企业前来合作!

1、Argo Event简介

Event事件大家都很熟悉，可以说Kubernetes就是完全由事件驱动的，不同的controller manager本质就是实现了不同的事件处理函数，比如所有ReplicaSet对象是由ReplicaSetController控制器管理，该控制器通过Informer监听ReplicaSet以及其关联的Pod的事件变化，从而维持运行状态和我们声明spec保持一致。

当然Kubernetes无论是什么Controller，其监听和处理的都是内部事件，而在应用层上我们也有很多外部事件，比如CICD事件、Webhook事件、日志事件等等，如何处理这些事件呢，目前Kubernetes原生是无法实现的。

当然你可以自己实现一个event handler运行在Kubernetes平台，不过实现难度也不小。而Argo Event组件完美解决了这个问题。

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

首先事件源EventSource可以是Webhook、S3、Github、SQS等等，中间会经过一个叫Gateway(新版本叫EventBus)的组件，更准确地说老版本原来gateway的配置功能已经合并到EventSource了，EventBus是新引入的组件，后端默认基于高性能分布式消息中间件NATS[1]实现，当然其他中间件比如Kafka也是可以的。

这个EventBus可以看做是事件的一个消息队列，消息生产者连接EvenSource，EventSource又连接到Sensor。更详细地说EvenSource把事件发送给EvenBus，Sensor会订阅EvenBus的消息队列，EvenBus负责把事件转发到已订阅该事件的Sensor组件，EventSorce在上图中没有体现，具体设计文档可以参考Argo-events Enhancement Proposals[2]。

有些人可能会说为什么EventBus不直接到Trigger，中间引入一个Sensor，这主要是两个原因，一是为了使事件转发和处理松耦合，二是为了实现Trigger事件的参数化，通过Sensor不仅可以实现事件的过滤，还可以实现事件的参数化，比如后面的Trigger是创建一个Kubernetes Pod，那这个Pod的metadata、env等，都可以根据事件内容进行填充。

Sensor组件注册关联了一个或者多个触发器，这些触发器可以触发AWS Lambda事件、Argo Workflow事件、Kubernetes Objects等，通俗简单地说，可以执行Lambda函数，可以动态地创建Kubernetes的对象或者创建前面的介绍的Workflow。

还记得前面介绍的Argo Rollout吗，我们演示了手动promote实现应用发布或者回滚，通过Argo Event就可以很完美地和测试平台或者CI/CD事件结合起来，实现自动应用自动发布或者回滚。

2、一个简单的Webhook例子

关于Argo Event的部署非常简单，直接通过kubecl apply或者helm均可，可以参考文档Installation[3]，这里不再赘述。

Argo Event部署完成后注意还需要部署EventBus，官方推荐使用NATS中间件，文档中有部署NATS stateful的文档。

接下来我们以一个最简单的Webhook事件为例，从而了解Argo Event的几个组件功能以及用法。

首先按照前面的介绍，我们需要先定义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

这个EventSource定义了一个webhook webhook_example，端口为12000，路径为/webhook，一般Webhook为POST方法，因此该Webhhok处理器我们配置只接收POST方法。

为了把这个Webhook EventSource暴露，我们还创建了一个Service，端口也是12000。

此时我们可以手动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           12000/TCP   5m49s
# curl -X POST -d '{}' 10.96.93.24:12000/webhook
success

当然此时由于没有注册任何的Sensor，因此什么都不会发生。

接下来我们定义Sensor:

首先在dependencies中定义了订阅的EventSource以及具体的Webhook，由于一个EventSource可以定义多个Webhook，因此必须同时指定EventSource和Webhook两个参数。

在Trigger中我们定义了对应Action为create一个workflow，这个workflow的spec定义在resource中配置。

最后的parameters部分定义了workflow的参数，这些参数值从event中获取，这里我们会把整个event都当作workflow的input。当然你可以通过dataKey只汲取body部分:dataKey: body.message。

此时我们再次curl这个webhook事件:

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

此时我们获取argo workflow列表发现新创建了一个实例：

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

查看workflow输出如下:

由于我们是把整个event作为workflow input发过去的，因此data内容部分是base64编码，我们可以查看解码后的内容如下:

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

从这里我们也可以看出Event包含两个部分，一个是context，一个是data，data中又包含header部分以及body部分，在parameters中可以通过Key获取任意部分内容。

如上的webhook触发是通过手动curl的，你可以很容易地在github或者bitbucket上配置到webhook中，这样一旦代码有更新就能触发这个事件了。

3、Kubernetes触发AWS Lambda函数

前面的例子中的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事件源为例，这个事件监听Kubernetes的资源状态，比如Pod创建、删除等，这里以创建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

如上例子监听Pods的ADD事件，即创建Pod，filter中过滤只有包含app=my-pod标签的Pod，特别需要注意的是使用的serviceaccount argo-events-sa必须具有Pod的list、watch权限。

接下来我们使用AWS Lambda触发器，Lambda函数已经在AWS提前创建好:

这个Lambda函数很简单，直接返回event本身。

创建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中。

此时我们创建一个新的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

当Pod启动后，我们发现AWS Lambda函数被触发执行:

4、event filter

前面的例子中webhook中所有的事件都会被sensor触发，我们有时不需要处理所有的事件，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，匹配的内容为hello、hey。

5、trigger policy

trigger policy主要用来判断最后触发器执行的结果是成功还是失败，如果是创建Kubernetes资源比如Workflow，可以根据Workflow最终状态决定这个Trigger的执行结果，而如果是触发一个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

如上表示当AWS Lambda返回200或者201时表示Trigger成功。

以上就是关于“Argo Event云原生的事件驱动架构怎么用”这篇文章的内容，相信大家都有了一定的了解，希望小编分享的内容对大家有帮助，若想了解更多相关的知识内容，请关注创新互联行业资讯频道。