這篇文章介紹了如何利用Apache Flink的內(nèi)置指標(biāo)系統(tǒng)以及如何使用Prometheus來(lái)高效地監(jiān)控流式應(yīng)用程序。

為什么選擇Prometheus？

隨著深入地了解Prometheus，你會(huì)發(fā)現(xiàn)一些非常好的功能：

• 服務(wù)發(fā)現(xiàn)使配置更加容易。Prometheus支持consul，etcd，kubernetes以及各家公有云廠商自動(dòng)發(fā)現(xiàn)。對(duì)于監(jiān)控目標(biāo)動(dòng)態(tài)發(fā)現(xiàn)，這點(diǎn)特別契合Cloud時(shí)代，應(yīng)用動(dòng)態(tài)擴(kuò)縮的特點(diǎn)。我們無(wú)法想象，在Cloud時(shí)代，需要運(yùn)維不斷更改配置。
• 開(kāi)源社區(qū)建立了數(shù)百個(gè)exporter?；旧虾w了所有基礎(chǔ)設(shè)施和主流中間件。
• 工具庫(kù)可從您的應(yīng)用程序獲取自定義指標(biāo)?；旧现髁鏖_(kāi)發(fā)語(yǔ)言都有對(duì)應(yīng)的工具庫(kù)。
• 它是CNCF旗下的OSS，是繼Kubernetes之后的第二個(gè)畢業(yè)項(xiàng)目。Kubernetes已經(jīng)與Promethues深度結(jié)合，并在其所有服務(wù)中公開(kāi)了Prometheus指標(biāo)。
• Pushgateway，Alermanager等組件，基本上涵蓋了一個(gè)完整的監(jiān)控生命周期。
• Flink官方已經(jīng)提供了對(duì)接Prometheus的jar包，很方便就可以集成。由于本系列文章重點(diǎn)在Flink on Kubernetes， 因此我們所有的操作都是基于這點(diǎn)展開(kāi)。

部署Prometheus

對(duì)k8s不熟悉的同學(xué)，可以查閱k8s相關(guān)文檔。由于部署不是本博客的重點(diǎn)，所以我們直接貼出yaml文件：

---
    apiVersion: v1
    kind: ServiceAccount
    metadata:
      name: monitor
      namespace: kube-system
      labels:
        kubernetes.io/cluster-service: "true"
        addonmanager.kubernetes.io/mode: Reconcile
---
    apiVersion: rbac.authorization.k8s.io/v1
    kind: ClusterRole
    metadata:
      name: monitor
      labels:
        kubernetes.io/cluster-service: "true"
        addonmanager.kubernetes.io/mode: Reconcile 
    rules:
      - apiGroups:
          - ""
        resources:
          - pods
        verbs:
          - get
          - list
          - watch
---
    apiVersion: rbac.authorization.k8s.io/v1
    kind: ClusterRoleBinding
    metadata:
      name: monitor
      labels:
        kubernetes.io/cluster-service: "true"
        addonmanager.kubernetes.io/mode: Reconcile
    roleRef:
      apiGroup: rbac.authorization.k8s.io
      kind: ClusterRole
      name: monitor
    subjects:
    - kind: ServiceAccount
      name: monitor
      namespace: kube-system
---
        apiVersion: v1
        kind: ConfigMap
        metadata:
          labels:
            app: monitor
          name: monitor
          namespace: kube-system
        data:
          prometheus.yml: |-
            global:
                scrape_interval:     10s
                evaluation_interval: 10s
              
            scrape_configs:
              - job_name: kubernetes-pods
                kubernetes_sd_configs:
                - role: pod
                relabel_configs:
                - action: keep
                  regex: true
                  source_labels:
                  - __meta_kubernetes_pod_annotation_prometheus_io_scrape
                - action: replace
                  regex: (.+)
                  source_labels:
                  - __meta_kubernetes_pod_annotation_prometheus_io_path
                  target_label: __metrics_path__
                - action: replace
                  regex: ([^:]+)(?::d+)?;(d+)
                  replacement: $1:$2
                  source_labels:
                  - __address__
                  - __meta_kubernetes_pod_annotation_prometheus_io_port
                  target_label: __address__
                - action: labelmap
                  regex: __meta_kubernetes_pod_label_(.+)
                - action: replace
                  source_labels:
                  - __meta_kubernetes_namespace
                  target_label: kubernetes_namespace
                - action: replace
                  source_labels:
                  - __meta_kubernetes_pod_name
                  target_label: kubernetes_pod_name
        
---
        apiVersion: apps/v1
        kind: StatefulSet
        metadata:
          labels:
            app: monitor
          name: monitor
          namespace: kube-system
        spec:
          serviceName: monitor
          selector:
            matchLabels:
              app: monitor
          replicas: 1
          template:
            metadata:
              labels:
                app: monitor
            spec:
              containers:
              - args:
                - --config.file=/etc/prometheus/prometheus.yml
                - --storage.tsdb.path=/data/prometheus
                - --storage.tsdb.retention.time=10d 
                image: prom/prometheus:v2.19.0
                imagePullPolicy: IfNotPresent
                name: prometheus
                ports:
                - containerPort: 9090
                  protocol: TCP
                readinessProbe:
                  httpGet:
                    path: /-/ready
                    port: 9090
                  initialDelaySeconds: 30
                  timeoutSeconds: 30
                livenessProbe:
                  httpGet:
                    path: /-/healthy
                    port: 9090
                  initialDelaySeconds: 30
                  timeoutSeconds: 30
                resources:
                  limits:
                    cpu: 1000m
                    memory: 2018Mi
                  requests:
                    cpu: 1000m
                    memory: 2018Mi
                volumeMounts:
                - mountPath: /etc/prometheus
                  name: config-volume
                - mountPath: /data
                  name: monitor-persistent-storage
              restartPolicy: Always
              priorityClassName: system-cluster-critical
              serviceAccountName: monitor
              initContainers:
                - name: "init-chown-data"
                  image: "busybox:latest"
                  imagePullPolicy: "IfNotPresent"
                  command: ["chown", "-R", "65534:65534", "/data"]
                  volumeMounts:
                    - name: monitor-persistent-storage
                      mountPath: /data
                      subPath: ""
              volumes:
              - configMap:
                  defaultMode: 420
                  name: monitor
                name: config-volume
          volumeClaimTemplates:
            - metadata:
                name: monitor-persistent-storage
                namespace: kube-system
              spec:
                accessModes:
                  - ReadWriteOnce
                resources:
                  requests:
                    storage: 20Gi
                storageClassName: gp2

---
        apiVersion: v1
        kind: Service
        metadata:
          annotations:
            service.beta.kubernetes.io/aws-load-balancer-type: nlb
          labels:
            app: monitor
          name: monitor
          namespace: kube-system
        spec:
          ports:
          - name: http
            port: 9090
            protocol: TCP
            targetPort: 9090
          selector:
            app: monitor
          type: LoadBalancer

這里我們簡(jiǎn)單說(shuō)下，由于我們想利用Prometheus的Kubernetes的服務(wù)發(fā)現(xiàn)的方式，所以需要RBAC授權(quán)，授權(quán)prometheus 實(shí)例對(duì)集群中的pod有一些讀取權(quán)限。

為什么我們要使用自動(dòng)發(fā)現(xiàn)的方式那？

相比配置文件的方式，自動(dòng)發(fā)現(xiàn)更加靈活。尤其是當(dāng)你使用的是flink on native kubernetes，整個(gè)job manager 和task manager 是根據(jù)作業(yè)的提交自動(dòng)創(chuàng)建的，這種動(dòng)態(tài)性，顯然是配置文件無(wú)法滿足的。

由于我們的集群在eks上，所以大家在使用其他云的時(shí)候，需要略做調(diào)整。

定制鏡像

這里我們基本上使用上一篇文章介紹的demo上，增加監(jiān)控相關(guān)，所以Dockerfile如下：

FROM flink
COPY /plugins/metrics-prometheus/flink-metrics-prometheus-1.11.0.jar /opt/flink/lib
RUN mkdir -p $FLINK_HOME/usrlib
COPY ./examples/streaming/WordCount.jar $FLINK_HOME/usrlib/my-flink-job.jar

Flink 的 Classpath 位于/opt/flink/lib，所以插件的jar包需要放到該目錄下。

作業(yè)提交

由于我們的Pod必須增加一定的標(biāo)識(shí)，從而讓Prometheus實(shí)例可以發(fā)現(xiàn)。所以提交命令稍作更改，如下：

./bin/flink run-application -p 8 -t kubernetes-application 
  -Dkubernetes.cluster-id=my-first-cluster 
  -Dtaskmanager.memory.process.size=2048m 
  -Dkubernetes.taskmanager.cpu=2 
  -Dtaskmanager.numberOfTaskSlots=4 
  -Dkubernetes.container.image=iyacontrol/flink-world-count:v0.0.2 
  -Dkubernetes.container.image.pull-policy=Always 
  -Dkubernetes.namespace=stream 
  -Dkubernetes.jobmanager.service-account=flink 
  -Dkubernetes.rest-service.exposed.type=LoadBalancer 
  -Dkubernetes.rest-service.annotations=service.beta.kubernetes.io/aws-load-balancer-type:nlb,service.beta.kubernetes.io/aws-load-balancer-internal:true 
  -Dkubernetes.jobmanager.annotations=prometheus.io/scrape:true,prometheus.io/port:9249 
  -Dkubernetes.taskmanager.annotations=prometheus.io/scrape:true,prometheus.io/port:9249 
  -Dmetrics.reporters=prom 
  -Dmetrics.reporter.prom.class=org.apache.flink.metrics.prometheus.PrometheusReporter 
  local:///opt/flink/usrlib/my-flink-job.jar

• 給 jobmanager 和 taskmanager 增加了annotations
• 增加了metrcis相關(guān)的配置，指定使用prometheus reporter

關(guān)于prometheus reporter：

參數(shù)：

• port - 可選, Prometheus導(dǎo)出器監(jiān)聽(tīng)的端口，默認(rèn)為9249。為了能夠在一臺(tái)主機(jī)上運(yùn)行報(bào)告程序的多個(gè)實(shí)例（例如，當(dāng)一個(gè)TaskManager與JobManager并置時(shí)），建議使用這樣的端口范圍 9250-9260。
• filterLabelValueCharacters - 可選, 指定是否過(guò)濾標(biāo)簽值字符。如果啟用，則將刪除所有不匹配[a-zA-Z0-9：_]的字符，否則將不刪除任何字符。禁用此選項(xiàng)之前，請(qǐng)確保您的標(biāo)簽值符合Prometheus要求。

效果

提交任務(wù)后，我們看下實(shí)際效果。

首先查看Prometheus 是否發(fā)現(xiàn)了我們的Pod。

然后查看具體的metrics，是否被準(zhǔn)確抓取。

指標(biāo)已經(jīng)收集，后續(xù)大家就可以選擇grafana繪圖了?；蚴窃黾酉鄳?yīng)的報(bào)警規(guī)則。例如：

總結(jié)

當(dāng)然除了Prometheus主動(dòng)發(fā)現(xiàn)Pod，然后定期抓取metrcis的方式，flink 也支持向PushGateway 主動(dòng)push metrcis。

Flink 通過(guò) Reporter 來(lái)向外部系統(tǒng)提供metrcis。通過(guò)在conf/flink-conf.yaml中配置一個(gè)或多個(gè)Reporter ，可以將metrcis公開(kāi)給外部系統(tǒng)。這些Reporter在啟動(dòng)時(shí)將在每個(gè)作業(yè)和任務(wù)管理器上實(shí)例化。

所有Reporter都必須至少具有class或factory.class屬性?？梢?應(yīng)該使用哪個(gè)屬性取決于Reporter的實(shí)現(xiàn)。有關(guān)更多信息，請(qǐng)參見(jiàn)各個(gè)Reporter 配置部分。一些Reporter允許指定報(bào)告間隔。

指定多個(gè)Reporter 的示例配置：

metrics.reporters: my_jmx_reporter,my_other_reporter

metrics.reporter.my_jmx_reporter.factory.class: org.apache.flink.metrics.jmx.JMXReporterFactory
metrics.reporter.my_jmx_reporter.port: 9020-9040
metrics.reporter.my_jmx_reporter.scope.variables.excludes:job_id;task_attempt_num

metrics.reporter.my_other_reporter.class: org.apache.flink.metrics.graphite.GraphiteReporter
metrics.reporter.my_other_reporter.host: 192.168.1.1
metrics.reporter.my_other_reporter.port: 10000

啟動(dòng)Flink時(shí)，必須可以訪問(wèn)包含reporter的jar。支持factory.class屬性的reporter可以作為插件加載。否則，必須將jar放在/lib文件夾中。

你可以通過(guò)實(shí)現(xiàn)org.apache.flink.metrics.reporter.MetricReporter接口來(lái)編寫(xiě)自己的Reporter。

如果 reporter定期發(fā)送報(bào)告，則還必須實(shí)現(xiàn)Scheduled接口。通過(guò)額外實(shí)現(xiàn)MetricReporterFactory，你的reporter也可以作為插件加載。

好啦！今天的分享到這里就結(jié)束了，希望大家持續(xù)關(guān)注馬哥教育官網(wǎng)，每天都會(huì)有大量 優(yōu)質(zhì)內(nèi)容與大家分享！聲明：文章轉(zhuǎn)載于網(wǎng)絡(luò)，版權(quán)歸原作者所有，如有侵權(quán)請(qǐng)及時(shí)聯(lián)系刪除！