使用 Prometheus 監(jiān)控 Flink

作者：iyacontrol

來源：SegmentFault 思否社區(qū)?

---

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

為什么選擇Prometheus？

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

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

Flink官方已經(jīng)提供了對接Prometheus的jar包，很方便就可以集成。由于本系列文章重點在Flink on Kubernetes， 因此我們所有的操作都是基于這點展開。

部署Prometheus

對k8s不熟悉的同學(xué)，可以查閱k8s相關(guā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

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

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

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

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

---

定制鏡像

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

? ? ? ? ? ? ? ?

FROM flinkCOPY /plugins/metrics-prometheus/flink-metrics-prometheus-1.11.0.jar /opt/flink/libRUN mkdir -p $FLINK_HOME/usrlibCOPY ./examples/streaming/WordCount.jar $FLINK_HOME/usrlib/my-flink-job.jar

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

作業(yè)提交

---

由于我們的Pod必須增加一定的標(biāo)識，從而讓Prometheus實例可以發(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)聽的端口，默認(rèn)為9249。為了能夠在一臺主機上運行報告程序的多個實例（例如，當(dāng)一個TaskManager與JobManager并置時），建議使用這樣的端口范圍 9250-9260。
• filterLabelValueCharacters - 可選, 指定是否過濾標(biāo)簽值字符。如果啟用，則將刪除所有不匹配[a-zA-Z0-9：_]的字符，否則將不刪除任何字符。禁用此選項之前，請確保您的標(biāo)簽值符合Prometheus要求。

---

效果

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

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

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

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

---

總結(jié)

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

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

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

指定多個Reporter 的示例配置：

? ? ? ? ? ? ? ?

metrics.reporters: my_jmx_reporter,my_other_reporter
metrics.reporter.my_jmx_reporter.factory.class: org.apache.flink.metrics.jmx.JMXReporterFactorymetrics.reporter.my_jmx_reporter.port: 9020-9040metrics.reporter.my_jmx_reporter.scope.variables.excludes:job_id;task_attempt_num
metrics.reporter.my_other_reporter.class: org.apache.flink.metrics.graphite.GraphiteReportermetrics.reporter.my_other_reporter.host: 192.168.1.1metrics.reporter.my_other_reporter.port: 10000

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

你可以通過實現(xiàn)org.apache.flink.metrics.reporter.MetricReporter接口來編寫自己的Reporter。如果 reporter定期發(fā)送報告，則還必須實現(xiàn)Scheduled接口。通過額外實現(xiàn)MetricReporterFactory，你的reporter也可以作為插件加載。

---

點擊左下角閱讀原文，到?SegmentFault 思否社區(qū)?和文章作者展開更多互動和交流。