JobSink，在事件發生時觸發長時間執行的背景作業¶

通常，結合 Knative 服務的事件處理預期會在相對較短的時間內（幾分鐘）完成，因為它需要 HTTP 連線保持開啟，否則服務會被縮減。

保持長時間執行的連線開啟會增加失敗的可能性，因此需要重新啟動處理，因為會重試請求。

此限制並不理想，JobSink 是一種資源，可用於建立長時間執行的非同步作業和任務。

JobSink 支援完整的 Kubernetes batch/v1 Job 資源和功能以及 Kubernetes Job 排隊系統，例如 Kueue。

先決條件¶

您必須擁有存取 Kubernetes 叢集的權限，且已安裝 Knative Eventing。

用法¶

當事件傳送至 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