JobSink，在事件发生时触发长时间运行的后台作业¶

通常，事件处理与 Knative 服务的结合预计将在相对较短的时间内（分钟）完成，因为它需要 HTTP 连接保持打开状态，否则服务将缩减。

保持长时间运行的连接打开会增加失败的可能性，因此处理需要重新启动，因为请求将被重试。

此限制并不理想，JobSink 是一种资源，您可以使用它创建长时间运行的异步作业和任务。

JobSink 支持完整的 Kubernetes batch/v1 作业资源和功能以及 Kubernetes 作业排队系统，例如 Kueue。

先决条件¶

您必须能够访问安装了 Knative Eventing 的 Kubernetes 集群。

用法¶

当事件被发送到 JobSink 时，Eventing 会创建一个 Job 并将接收到的事件作为 JSON 文件安装在 /etc/jobsink-event/event 中。

创建一个 JobSink

apiVersion: sinks.knative.dev/v1alpha1
kind: JobSink
metadata:
  name: job-sink-logger
spec:
  job:
    spec:
      completions: 1
      parallelism: 1
      template:
        spec:
          restartPolicy: Never
          containers:
            - name: main
              image: docker.io/library/bash:5
              command: [ "cat" ]
              args:
                - "/etc/jobsink-event/event"

应用 JobSink 资源
```
kubectl apply -f <job-sink-file.yaml>
```

验证 JobSink 是否已准备好

kubectl get jobsinks.sinks.knative.dev

示例输出

NAME              URL                                                                          AGE   READY   REASON
job-sink-logger   http://job-sink.knative-eventing.svc.cluster.local/default/job-sink-logger   5s    True

触发 JobSink

kubectl run curl --image=curlimages/curl --rm=true --restart=Never -ti -- -X POST -v \
   -H "content-type: application/json"  \
   -H "ce-specversion: 1.0" \
   -H "ce-source: my/curl/command" \
   -H "ce-type: my.demo.event" \
   -H "ce-id: 123" \
   -d '{"details":"JobSinkDemo"}' \
   http://job-sink.knative-eventing.svc.cluster.local/default/job-sink-logger

验证是否创建了 Job 并打印事件

kubectl logs job-sink-loggerszoi6-dqbtq

示例输出

{"specversion":"1.0","id":"123","source":"my/curl/command","type":"my.demo.event","datacontenttype":"application/json","data":{"details":"JobSinkDemo"}}

JobSink 幂等性¶

JobSink 将为每个不同的接收事件创建作业。

事件通过事件的 source 和 id 属性的组合来唯一标识。

如果收到具有相同 source 和 id 属性的事件，并且已经存在作业，则不会创建另一个 Job。

读取事件文件¶

您可以使用任何 CloudEvents JSON 反序列化器读取文件并将其反序列化。

例如，以下代码段使用 CloudEvents Go SDK 读取事件并对其进行处理。

package mytask

import (
    "encoding/json"
    "fmt"
    "os"

    cloudevents "github.com/cloudevents/sdk-go/v2"
)

func handleEvent() error {
    eventBytes, err := os.ReadFile("/etc/jobsink-event/event")
    if err != nil {
        return err
    }

    event := &cloudevents.Event{}
    if err := json.Unmarshal(eventBytes, event); err != nil {
        return err
    }

    // Process event ...
    fmt.Println(event)

    return nil
}

从不同的事件源触发作业¶

JobSink 可以被任何事件源或触发器触发。

例如，您可以使用 KafkaSource 在 Kafka 记录被发送到 Kafka 主题时触发 Job。

apiVersion: sources.knative.dev/v1beta1
kind: KafkaSource
metadata:
  name: kafka-source
spec:
  bootstrapServers:
    - my-cluster-kafka-bootstrap.kafka:9092
  topics:
    - knative-demo-topic
  sink:
    ref:
      apiVersion: sinks.knative.dev/v1alpha1
      kind: JobSink
      name: job-sink-logger

或者，当 Knative Broker 接收事件时使用 Trigger

apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata:
  name: my-job-sink-trigger
spec:
  broker: my-broker
  filter:
    attributes:
      type: dev.knative.foo.bar
      myextension: my-extension-value
    subscriber:
      ref:
        apiVersion: sinks.knative.dev/v1alpha1
        kind: JobSink
        name: job-sink-logger

或者，甚至可以作为 Knative Broker 的死信接收器

apiVersion: eventing.knative.dev/v1
kind: Broker
metadata:
  name: my-broker
spec:
  # ...

  delivery:
    deadLetterSink:
      ref:
        apiVersion: sinks.knative.dev/v1alpha1
        kind: JobSink
        name: job-sink-logger
    retry: 5
    backoffPolicy: exponential
    backoffDelay: "PT1S"

自定义事件文件目录¶

apiVersion: sinks.knative.dev/v1alpha1
kind: JobSink
metadata:
  name: job-sink-custom-mount-path
spec:
  job:
    spec:
      completions: 1
      parallelism: 1
      template:
        spec:
          restartPolicy: Never
          containers:
            - name: main
              image: docker.io/library/bash:5
              command: [ "bash" ]
              args:
                - -c
                - echo "Hello world!" && sleep 5

              # The event will be available in a file at `/etc/custom-path/event`
              volumeMounts:
                - name: "jobsink-event"
                  mountPath: "/etc/custom-path"
                  readOnly: true

清理已完成的作业¶

要清理已完成的作业，您可以设置 spec.job.spec.ttlSecondsAfterFinished: 600 字段，Kubernetes 将在 600 秒（10 分钟）后删除已完成的作业。

JobSink 示例¶

JobSink 成功示例¶

apiVersion: sinks.knative.dev/v1alpha1
kind: JobSink
metadata:
  name: job-sink-success
spec:
  job:
    metadata:
      labels:
        my-label: my-value
    spec:
      completions: 12
      parallelism: 3
      template:
        spec:
          restartPolicy: Never
          containers:
            - name: main
              image: docker.io/library/bash:5
              command: [ "bash" ]
              args:
                - -c
                - echo "Hello world!" && sleep 5
      backoffLimit: 6
      podFailurePolicy:
        rules:
          - action: FailJob
            onExitCodes:
              containerName: main      # optional
              operator: In             # one of: In, NotIn
              values: [ 42 ]
          - action: Ignore             # one of: Ignore, FailJob, Count
            onPodConditions:
              - type: DisruptionTarget   # indicates Pod disruption

JobSink 失败示例¶

apiVersion: sinks.knative.dev/v1alpha1
kind: JobSink
metadata:
  name: job-sink-failure
spec:
  job:
    metadata:
      labels:
        my-label: my-value
    spec:
      completions: 12
      parallelism: 3
      template:
        spec:
          restartPolicy: Never
          containers:
            - name: main
              image: docker.io/library/bash:5
              command: [ "bash" ]        # example command simulating a bug which triggers the FailJob action
              args:
                - -c
                - echo "Hello world!" && sleep 5 && exit 42
      backoffLimit: 6
      podFailurePolicy:
        rules:
          - action: FailJob
            onExitCodes:
              containerName: main      # optional
              operator: In             # one of: In, NotIn
              values: [ 42 ]
          - action: Ignore             # one of: Ignore, FailJob, Count
            onPodConditions:
              - type: DisruptionTarget   # indicates Pod disruption