以 Serverless 的方式實(shí)現(xiàn) Kubernetes 日志告警

本文轉(zhuǎn)載自公眾號(hào)「KubeSphere 云原生」

作者簡(jiǎn)介：方闐，OpenFunction 開源社區(qū) Maintainer。

當(dāng)我們將容器的日志收集到消息服務(wù)器之后，我們?cè)撊绾翁幚磉@些日志？部署一個(gè)專用的日志處理工作負(fù)載可能會(huì)耗費(fèi)多余的成本，而當(dāng)日志體量驟增、驟降時(shí)亦難以評(píng)估日志處理工作負(fù)載的待機(jī)數(shù)量。本文提供了一種基于 Serverless 的日志處理思路，可以在降低該任務(wù)鏈路成本的同時(shí)提高其靈活性。

我們的大體設(shè)計(jì)是使用 Kafka 服務(wù)器作為日志的接收器，之后以輸入 Kafka 服務(wù)器的日志作為事件，驅(qū)動(dòng) Serverless 工作負(fù)載對(duì)日志進(jìn)行處理。據(jù)此的大致步驟為：

1. 搭建 Kafka 服務(wù)器作為 Kubernetes 集群的日志接收器
2. 部署 OpenFunction 為日志處理工作負(fù)載提供 Serverless 能力
3. 編寫日志處理函數(shù)，抓取特定的日志生成告警消息
4. 配置 Notification Manager^[1] 將告警發(fā)送至 Slack

在這個(gè)場(chǎng)景中，我們會(huì)利用到 OpenFunction^[2] 帶來(lái)的 Serverless 能力。

OpenFunction^[3] 是 KubeSphere 社區(qū)開源的一個(gè) FaaS（Serverless）項(xiàng)目，旨在讓用戶專注于他們的業(yè)務(wù)邏輯，而不必關(guān)心底層運(yùn)行環(huán)境和基礎(chǔ)設(shè)施。該項(xiàng)目當(dāng)前具備以下關(guān)鍵能力：

• 支持通過 dockerfile 或 buildpacks 方式構(gòu)建 OCI 鏡像
• 支持使用 Knative Serving 或 OpenFunctionAsync ( KEDA + Dapr ) 作為 runtime 運(yùn)行 Serverless 工作負(fù)載
• 自帶事件驅(qū)動(dòng)框架

使用 Kafka 作為日志接收器

首先，我們?yōu)?KubeSphere 平臺(tái)開啟 logging 組件（可以參考 啟用可插拔組件^[4] 獲取更多信息）。然后我們使用 strimzi-kafka-operator^[5] 搭建一個(gè)最小化的 Kafka 服務(wù)器。

1. 在 default 命名空間中安裝 strimzi-kafka-operator^[6] ：

helm repo add strimzi https://strimzi.io/charts/
helm install kafka-operator -n default strimzi/strimzi-kafka-operator

2. 運(yùn)行以下命令在 default 命名空間中創(chuàng)建 Kafka 集群和 Kafka Topic，該命令所創(chuàng)建的 Kafka 和 Zookeeper 集群的存儲(chǔ)類型為 ephemeral，使用 emptyDir 進(jìn)行演示。

注意，我們此時(shí)創(chuàng)建了一個(gè)名為 “l(fā)ogs” 的 topic，后續(xù)會(huì)用到它

cat <<EOF | kubectl apply -f -
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: kafka-logs-receiver
  namespace: default
spec:
  kafka:
    version: 2.8.0
    replicas: 1
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      log.message.format.version: '2.8'
      inter.broker.protocol.version: "2.8"
    storage:
      type: ephemeral
  zookeeper:
    replicas: 1
    storage:
      type: ephemeral
  entityOperator:
    topicOperator: {}
    userOperator: {}
---
apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaTopic
metadata:
  name: logs
  namespace: default
  labels:
    strimzi.io/cluster: kafka-logs-receiver
spec:
  partitions: 10
  replicas: 3
  config:
    retention.ms: 7200000
    segment.bytes: 1073741824
EOF

3. 運(yùn)行以下命令查看 Pod 狀態(tài)，并等待 Kafka 和 Zookeeper 運(yùn)行并啟動(dòng)。

$ kubectl get po
NAME                                                   READY   STATUS        RESTARTS   AGE
kafka-logs-receiver-entity-operator-568957ff84-nmtlw   3/3     Running       0          8m42s
kafka-logs-receiver-kafka-0                            1/1     Running       0          9m13s
kafka-logs-receiver-zookeeper-0                        1/1     Running       0          9m46s
strimzi-cluster-operator-687fdd6f77-cwmgm              1/1     Running       0          11m

運(yùn)行以下命令查看 Kafka 集群的元數(shù)據(jù)：

# 啟動(dòng)一個(gè)工具 pod
$ kubectl run utils --image=arunvelsriram/utils -i --tty --rm
# 查看 Kafka 集群的元數(shù)據(jù)
$ kafkacat -L -b kafka-logs-receiver-kafka-brokers:9092

我們將這個(gè) Kafka 服務(wù)器添加為日志接收器。

1. 以 admin 身份登錄 KubeSphere 的 Web 控制臺(tái)。點(diǎn)擊左上角的平臺(tái)管理，然后選擇集群管理。

   如果您啟用了多集群功能^[7]，您可以選擇一個(gè)集群。

2. 在集群管理頁(yè)面，選擇集群設(shè)置下的日志收集。

3. 點(diǎn)擊添加日志接收器并選擇 Kafka。輸入 Kafka 代理地址和端口信息，然后點(diǎn)擊確定繼續(xù)。

4. 運(yùn)行以下命令驗(yàn)證 Kafka 集群是否能從 Fluent Bit 接收日志：

# 啟動(dòng)一個(gè)工具 pod
$ kubectl run utils --image=arunvelsriram/utils -i --tty --rm
# 檢查 logs topic 中的日志情況
$ kafkacat -C -b kafka-logs-receiver-kafka-0.kafka-logs-receiver-kafka-brokers.default.svc:9092 -t logs

部署 OpenFunction

按照概述中的設(shè)計(jì)，我們需要先部署 OpenFunction。OpenFunction 項(xiàng)目引用了很多第三方的項(xiàng)目，如 Knative、Tekton、ShipWright、Dapr、KEDA 等，手動(dòng)安裝較為繁瑣，推薦使用 Prerequisites 文檔^[8] 中的方法，一鍵部署 OpenFunction 的依賴組件。

其中 --with-shipwright 表示部署 shipwright 作為函數(shù)的構(gòu)建驅(qū)動(dòng)--with-openFuncAsync 表示部署 OpenFuncAsync Runtime 作為函數(shù)的負(fù)載驅(qū)動(dòng) 而當(dāng)你的網(wǎng)絡(luò)在訪問 Github 及 Google 受限時(shí)，可以加上 --poor-network 參數(shù)用于下載相關(guān)的組件

$ sh hack/deploy.sh --with-shipwright --with-openFuncAsync --poor-network

部署 OpenFunction：

此處選擇安裝最新的穩(wěn)定版本，你也可以使用開發(fā)版本，參考 Install 文檔^[9]

為了可以正常使用 ShipWright ，我們提供了默認(rèn)的構(gòu)建策略，可以使用以下命令設(shè)置該策略：

$ kubectl apply -f https://raw.githubusercontent.com/OpenFunction/OpenFunction/main/config/strategy/openfunction.yaml

$ kubectl apply -f https://github.com/OpenFunction/OpenFunction/releases/download/v0.3.0/bundle.yaml

編寫日志處理函數(shù)

我們以 創(chuàng)建并部署 WordPress^[10] 為例，搭建一個(gè) WordPress 應(yīng)用作為日志的生產(chǎn)者。該應(yīng)用的工作負(fù)載所在的命名空間為 “demo-project”，Pod 名稱為 “wordpress-v1-f54f697c5-hdn2z”。

當(dāng)請(qǐng)求結(jié)果為 404 時(shí)，我們收到的日志內(nèi)容如下：

{"@timestamp":1629856477.226758,"log":"*.*.*.* - - [25/Aug/2021:01:54:36 +0000] \"GET /notfound HTTP/1.1\" 404 49923 \"-\" \"curl/7.58.0\"\n","time":"2021-08-25T01:54:37.226757612Z","kubernetes":{"pod_name":"wordpress-v1-f54f697c5-hdn2z","namespace_name":"demo-project","container_name":"container-nrdsp1","docker_id":"bb7b48e2883be0c05b22c04b1d1573729dd06223ae0b1676e33a4fac655958a5","container_image":"wordpress:4.8-apache"}}

我們的需求是：當(dāng)一個(gè)請(qǐng)求結(jié)果為 404 時(shí)，發(fā)送一個(gè)告警通知給接收器（可以根據(jù) 配置 Slack 通知^[11] 配置一個(gè) Slack 告警接收器），并記錄命名空間、Pod 名稱、請(qǐng)求路徑、請(qǐng)求方法等信息。按照這個(gè)需求，我們編寫一個(gè)簡(jiǎn)單的處理函數(shù)：

你可以從 OpenFunction Context Spec^[12] 處了解 openfunction-context 的使用方法，這是 OpenFunction 提供給用戶編寫函數(shù)的工具庫(kù) 你可以通過 OpenFunction Samples^[13] 了解更多的 OpenFunction 函數(shù)案例

package logshandler

import (
 "encoding/json"
 "fmt"
 "log"
 "regexp"
 "time"

 ofctx "github.com/OpenFunction/functions-framework-go/openfunction-context"
 alert "github.com/prometheus/alertmanager/template"
)

const (
 HTTPCodeNotFound = "404"
 Namespace        = "demo-project"
 PodName          = "wordpress-v1-[A-Za-z0-9]{9}-[A-Za-z0-9]{5}"
 AlertName        = "404 Request"
 Severity         = "warning"
)

// LogsHandler ctx 參數(shù)提供了用戶函數(shù)在集群語(yǔ)境中的上下文句柄，如 ctx.SendTo 用于將數(shù)據(jù)發(fā)送至指定的目的地
// LogsHandler in 參數(shù)用于將輸入源中的數(shù)據(jù)（如有）以 bytes 的方式傳遞給函數(shù)
func LogsHandler(ctx *ofctx.OpenFunctionContext, in []byte) int {
 content := string(in)
 // 這里我們?cè)O(shè)置了三個(gè)正則表達(dá)式，分別用于匹配 HTTP 返回碼、資源命名空間、資源 Pod 名稱
 matchHTTPCode, _ := regexp.MatchString(fmt.Sprintf(" %s ", HTTPCodeNotFound), content)
 matchNamespace, _ := regexp.MatchString(fmt.Sprintf("namespace_name\":\"%s", Namespace), content)
 matchPodName := regexp.MustCompile(fmt.Sprintf(`(%s)`, PodName)).FindStringSubmatch(content)

 if matchHTTPCode && matchNamespace && matchPodName != nil {
  log.Printf("Match log - Content: %s", content)

  // 如果上述三個(gè)正則表達(dá)式同時(shí)命中，那么我們需要提取日志內(nèi)容中的一些信息，用于填充至告警信息中
  // 這些信息為：404 請(qǐng)求的請(qǐng)求方式（HTTP Method）、請(qǐng)求路徑（HTTP Path）以及 Pod 名稱
  match := regexp.MustCompile(`([A-Z]+) (/\S*) HTTP`).FindStringSubmatch(content)
  if match == nil {
   return 500
  }
  path := match[len(match)-1]
  method := match[len(match)-2]
  podName := matchPodName[len(matchPodName)-1]

  // 收集到關(guān)鍵信息后，我們使用 altermanager 的 Data 結(jié)構(gòu)體組裝告警信息
  notify := &alert.Data{
   Receiver:          "notification_manager",
   Status:            "firing",
   Alerts:            alert.Alerts{},
   GroupLabels:       alert.KV{"alertname": AlertName, "namespace": Namespace},
   CommonLabels:      alert.KV{"alertname": AlertName, "namespace": Namespace, "severity": Severity},
   CommonAnnotations: alert.KV{},
   ExternalURL:       "",
  }
  alt := alert.Alert{
   Status: "firing",
   Labels: alert.KV{
    "alertname": AlertName,
    "namespace": Namespace,
    "severity":  Severity,
    "pod":       podName,
    "path":      path,
    "method":    method,
   },
   Annotations:  alert.KV{},
   StartsAt:     time.Now(),
   EndsAt:       time.Time{},
   GeneratorURL: "",
   Fingerprint:  "",
  }
  notify.Alerts = append(notify.Alerts, alt)
  notifyBytes, _ := json.Marshal(notify)

  // 使用 ctx.SendTo 將內(nèi)容發(fā)送給名為 "notification-manager" 的輸出端（你可以在之后的函數(shù)配置 logs-handler-function.yaml 中找到它的定義）
  if err := ctx.SendTo(notifyBytes, "notification-manager"); err != nil {
   panic(err)
  }
  log.Printf("Send log to notification manager.")
 }
 return 200
}

我們將這個(gè)函數(shù)上傳到代碼倉(cāng)庫(kù)中，記錄代碼倉(cāng)庫(kù)的地址以及代碼在倉(cāng)庫(kù)中的目錄路徑，在下面的創(chuàng)建函數(shù)步驟中我們將使用到這兩個(gè)值。

你可以在 OpenFunction Samples^[14] 中找到這個(gè)案例。

創(chuàng)建函數(shù)

接下來(lái)我們將使用 OpenFunction 構(gòu)建上述的函數(shù)。首先設(shè)置一個(gè)用于訪問鏡像倉(cāng)庫(kù)的秘鑰文件 push-secret（在使用代碼構(gòu)建出 OCI 鏡像后，OpenFunction 會(huì)將該鏡像上傳到用戶的鏡像倉(cāng)庫(kù)中，用于后續(xù)的負(fù)載啟動(dòng)）：

$ REGISTRY_SERVER=https://index.docker.io/v1/ REGISTRY_USER=<your username> REGISTRY_PASSWORD=<your password>
$ kubectl create secret docker-registry push-secret \
    --docker-server=$REGISTRY_SERVER \
    --docker-username=$REGISTRY_USER \
    --docker-password=$REGISTRY_PASSWORD

應(yīng)用函數(shù) logs-handler-function.yaml：

函數(shù)定義中包含了對(duì)兩個(gè)關(guān)鍵組件的使用：

Dapr^[15] 對(duì)應(yīng)用程序屏蔽了復(fù)雜的中間件，使得 logs-handler 可以非常容易地處理 Kafka 中的事件

KEDA^[16] 通過監(jiān)控消息服務(wù)器中的事件流量來(lái)驅(qū)動(dòng) logs-handler 函數(shù)的啟動(dòng)，并且根據(jù) Kafka 中消息的消費(fèi)延時(shí)動(dòng)態(tài)擴(kuò)展 logs-handler 實(shí)例

apiVersion: core.openfunction.io/v1alpha1
kind: Function
metadata:
  name: logs-handler
spec:
  version: "v1.0.0"
  # 這里定義了構(gòu)建后的鏡像的上傳路徑
  image: openfunctiondev/logs-async-handler:v1
  imageCredentials:
    name: push-secret
  build:
    builder: openfunctiondev/go115-builder:v0.2.0
    env:
      FUNC_NAME: "LogsHandler"
    # 這里定義了源代碼的路徑
    # url 為上面提到的代碼倉(cāng)庫(kù)地址
    # sourceSubPath 為代碼在倉(cāng)庫(kù)中的目錄路徑
    srcRepo:
      url: "https://github.com/OpenFunction/samples.git"
      sourceSubPath: "functions/OpenFuncAsync/logs-handler-function/"
  serving:
    # OpenFuncAsync 是 OpenFunction 通過 KEDA+Dapr 實(shí)現(xiàn)的一種由事件驅(qū)動(dòng)的異步函數(shù)運(yùn)行時(shí)
    runtime: "OpenFuncAsync"
    openFuncAsync:
      # 此處定義了函數(shù)的輸入（kafka-receiver）和輸出（notification-manager），與下面 components 中的定義對(duì)應(yīng)關(guān)聯(lián)
      dapr:
        inputs:
          - name: kafka-receiver
            type: bindings
        outputs:
          - name: notification-manager
            type: bindings
            params:
              operation: "post"
              type: "bindings"
        annotations:
          dapr.io/log-level: "debug"
        # 這里完成了上述輸入端和輸出端的具體定義（即 Dapr Components）
        components:
          - name: kafka-receiver
            type: bindings.kafka
            version: v1
            metadata:
              - name: brokers
                value: "kafka-logs-receiver-kafka-brokers:9092"
              - name: authRequired
                value: "false"
              - name: publishTopic
                value: "logs"
              - name: topics
                value: "logs"
              - name: consumerGroup
                value: "logs-handler"
          # 此處為 KubeSphere 的 notification-manager 地址
          - name: notification-manager
            type: bindings.http
            version: v1
            metadata:
              - name: url
                value: http://notification-manager-svc.kubesphere-monitoring-system.svc.cluster.local:19093/api/v2/alerts
      keda:
        scaledObject:
          pollingInterval: 15
          minReplicaCount: 0
          maxReplicaCount: 10
          cooldownPeriod: 30
          # 這里定義了函數(shù)的觸發(fā)器，即 Kafka 服務(wù)器的 “l(fā)ogs” topic
          # 同時(shí)定義了消息堆積閾值（此處為 10），即當(dāng)消息堆積量超過 10，logs-handler 實(shí)例個(gè)數(shù)就會(huì)自動(dòng)擴(kuò)展
          triggers:
            - type: kafka
              metadata:
                topic: logs
                bootstrapServers: kafka-logs-receiver-kafka-brokers.default.svc.cluster.local:9092
                consumerGroup: logs-handler
                lagThreshold: "10"

結(jié)果演示

我們先關(guān)閉 Kafka 日志接收器：在日志收集頁(yè)面，點(diǎn)擊進(jìn)入 Kafka 日志接收器詳情頁(yè)面，然后點(diǎn)擊更多操作并選擇更改狀態(tài)，將其設(shè)置為關(guān)閉。

停用后一段時(shí)間，我們可以觀察到 logs-handler 函數(shù)實(shí)例已經(jīng)收縮到 0 了。

再將 Kafka 日志接收器激活，logs-handler 隨之啟動(dòng)。

$ kubectl get po --watch
NAME                                                     READY   STATUS        RESTARTS   AGE
kafka-logs-receiver-entity-operator-568957ff84-tdrrx     3/3     Running       0          7m27s
kafka-logs-receiver-kafka-0                              1/1     Running       0          7m48s
kafka-logs-receiver-zookeeper-0                          1/1     Running       0          8m12s
logs-handler-serving-kpngc-v100-zcj4q-5f46996f8c-b9d6f   2/2     Terminating   0          34s
strimzi-cluster-operator-687fdd6f77-kc8cv                1/1     Running       0          10m
logs-handler-serving-kpngc-v100-zcj4q-5f46996f8c-b9d6f   2/2     Terminating   0          36s
logs-handler-serving-kpngc-v100-zcj4q-5f46996f8c-b9d6f   0/2     Terminating   0          37s
logs-handler-serving-kpngc-v100-zcj4q-5f46996f8c-b9d6f   0/2     Terminating   0          38s
logs-handler-serving-kpngc-v100-zcj4q-5f46996f8c-b9d6f   0/2     Terminating   0          38s
logs-handler-serving-kpngc-v100-zcj4q-5f46996f8c-9kj2c   0/2     Pending       0          0s
logs-handler-serving-kpngc-v100-zcj4q-5f46996f8c-9kj2c   0/2     Pending       0          0s
logs-handler-serving-kpngc-v100-zcj4q-5f46996f8c-9kj2c   0/2     ContainerCreating   0          0s
logs-handler-serving-kpngc-v100-zcj4q-5f46996f8c-9kj2c   0/2     ContainerCreating   0          2s
logs-handler-serving-kpngc-v100-zcj4q-5f46996f8c-9kj2c   1/2     Running             0          4s
logs-handler-serving-kpngc-v100-zcj4q-5f46996f8c-9kj2c   2/2     Running             0          11s

接著我們向 WordPress 應(yīng)用一個(gè)不存在的路徑發(fā)起請(qǐng)求：

$ curl http://<wp-svc-address>/notfound

可以看到 Slack 中已經(jīng)收到了這條消息（與之對(duì)比的是，當(dāng)我們正常訪問該 WordPress 站點(diǎn)時(shí)， Slack 中并不會(huì)收到告警消息）：

進(jìn)一步探索

• 同步函數(shù)的解決方案

  為了可以正常使用 Knative Serving ，我們需要設(shè)置其網(wǎng)關(guān)的負(fù)載均衡器地址。（你可以使用本機(jī)地址作為 workaround）

  將下面的 "1.2.3.4" 替換為實(shí)際場(chǎng)景中的地址。

  

除了直接由 Kafka 服務(wù)器驅(qū)動(dòng)函數(shù)運(yùn)作（異步方式），OpenFunction 還支持使用自帶的事件框架對(duì)接 Kafka 服務(wù)器，之后以 Sink 的方式驅(qū)動(dòng) Knative 函數(shù)運(yùn)作?？梢詤⒖?OpenFunction Samples^[17] 中的案例。

在該方案中，同步函數(shù)的處理速度較之異步函數(shù)有所降低，當(dāng)然我們同樣可以借助 KEDA 來(lái)觸發(fā) Knative Serving 的 concurrency 機(jī)制，但總體而言缺乏異步函數(shù)的便捷性。（后續(xù)的階段中我們會(huì)優(yōu)化 OpenFunction 的事件框架來(lái)解決同步函數(shù)這方面的缺陷）

由此可見，不同類型的 Serverless 函數(shù)有其擅長(zhǎng)的任務(wù)場(chǎng)景，如一個(gè)有序的控制流函數(shù)就需要由同步函數(shù)而非異步函數(shù)來(lái)處理。

綜述

Serverless 帶來(lái)了我們所期望的對(duì)業(yè)務(wù)場(chǎng)景快速拆解重構(gòu)的能力。

如本案例所示，OpenFunction 不但以 Serverless 的方式提升了日志處理、告警通知鏈路的靈活度，還通過函數(shù)框架將通常對(duì)接 Kafka 時(shí)復(fù)雜的配置步驟簡(jiǎn)化為語(yǔ)義明確的代碼邏輯。同時(shí)，我們也在不斷演進(jìn) OpenFunction，將在之后版本中實(shí)現(xiàn)由自身的 Serverless 能力驅(qū)動(dòng)自身的組件運(yùn)作。

引用鏈接