創作人：張妙成
審稿人：田雪松
項目背景
PaaS 下管理了大量叢集，監控和告警能快速的讓開發維護人員，知道系統已經發生故障，并且輔助高效排障。
但是無法提前預知叢集的健康狀況，開發人員和維護人員均無法在故障前及時作出調整。為了幫助使用者及時的知道叢集的健康狀态，更好使用 Elasticsearch 叢集，可以定期對叢集進行名額檢查并給出相應報告。巡檢作業及時發現叢集的健康問題，叢集的配置是否合理，提前主動發現問題，能有效保證叢集的穩定性、健狀性，進而減少業務中斷時間保證服務品質。
為了解決叢集健康狀态提前預知困難的問題，可以通過抽取一些名額，進行定時檢查達到健康診斷的目的。
巡檢主要是對叢集的各個名額檢查，給出一份全方位的報告，并提供一定的推薦解決、優化方案。如阿裡的 EYOU 平台（阿裡雲 Elasticsearch 智能診斷系統）會系統的在 Elasticsearch 公有雲進行各個名額的檢查，并給出相應報告，極大的減小了風險，降低了維護成本。
智能管理系統不是一個獨立的檢查系統，而是一個與其他系統相結合的閉環系統，獨立的巡檢子產品對各項名額進行檢查分析，将結果通過 PaaS 系統展示給使用者，并在 PaaS 中給予入口，用以幫助使用者手動再次觸發檢查，增強實時性，提高使用者體驗。
本文将介紹智能巡檢系統在整個 Elasticsearch 相關系統中的位置與意義，并從名額分析選取、異常标準的角度，主要闡述智能巡檢系統的設計與實作。
巡檢系統的結構
整個應用的架構如上：
- Elasticsearch 叢集在 K8S 環境中（實際生産大多是 K8S 環境與實體機、虛拟機環境共存，這裡簡化成最終要達到的統一環境），由 PaaS 平台進行統一管理。
- PaaS 的資訊資料主要是與 DB 互動（PaaS 是與 DB 的唯一互動入口），使用者主要與 PaaS平台進行互動。
- 智能巡檢系統資訊收集子產品（一組 Python Job）主要是 K8S 環境中的 Elasticsearch cluster、主控端進行互動，資料報告資訊通過 PaaS 平台存入 DB。
- 監控使用 VictoriaMetrics（ Prometheus 的高可用方案）作為存儲，grafana 作為前端展示頁面。監控可以配置 Elasticsearch 各項名額，其中與智能巡檢相關是巡檢異常數量的監控面闆，用來給 OPS 觀察巡檢亞健康叢集異常點的修複（優化）情況。
- PaaS 提供入口手動觸發再次檢查。
名額選取簡介
巡檢的名額、異常門檻值與告警配置的主要差別是，檢的名額項會更加關心可能引發故障的某些現象和配置，參考門檻值相對告警配置會更加寬松。巡檢主要是通過名額的采集分析，得出一份相對全面的報告和推薦解決方案。為了報告的全面性與分析的準确性，巡檢的名額項會與告警配置有一定相似或重複。
告警與巡檢需要解決的問題不同，告警的目的是将異常名額恢複到正常狀态，響應的實時性要求較高，而巡檢的目的是預防故障、消除隐患、優化叢集性能，以報告的形式推到平台和使用者，不需要使用者主動響應，隻需解決問題後重新觸發巡檢。為推進優化，可将巡檢報告中非健康名額配置成監控面闆、告警。
叢集健康程度可以從幾個方面表現：cluster 層面、node 層面、shard 層面、index 層面、jvm 層面、threadpool 層面。如下為參考名額：
| 子產品 | 名額項 | 異常門檻值（參考） |
|---|---|---|
| cluster 層面 | 1.cluster status 2.pengding_task3.cpu_util（極差）4.query（極差）5.一次bulk請求的數量 | 1.health為red、yellow2. pending_task數量count > 100 3.cpu_util(max) - cpu_util(min) > 50%4.query(max)/query(min) > 25.indexing_total(stacked) /s > 1000且indexing_total / thread_pool_write_completed < 100 |
| node 層面 | 1.uptime2.free_disk3.cpu_util4.node上shard數量 | 1.date_now - uptime < 1h2.free_disk < 30%3.cpu_util > 90%4.count(shard) = 0 |
| shard 層面 | 1.number2.size of per shard | 1.每GB的heap超過20個shards2.搜尋類叢集（tag）單個shard_size > 20g，日志類叢集單個 shard_size > 50g |
| index 層面 | 1.replica2.dynamic mapping3.refresh_interval4.indices.refresh.total5.max_result_window | 1.存在index無replica2.dynamic != false && dynamic != strict3.refresh_interval = -14.refresh_count > 80/min5.max_result_window > 10000 |
| jvm 層面 | 1.jvm heap使用率2.jdk version一緻性3.heap segment memory4.Full GC | 1.heap_util > 90%2.各個節點 jdk 版本出現不一緻3.segment memory占用heap > heap_size的20% 4.出現full gc |
| threadpool 層面 | 1.bulk reject數量2.search reject數量 | 1.bulk rejected>0 2.search rejected>0 |
Elasticsearch 功能強大、使用友善，也就意味着對使用者來說有很多的預設設定，使用者使用的自由度很高，也就意味着開放的能力豐富，使用者的使用對叢集健康程度有着很大的影響。
是以名額選取需要從兩個角度，一是現有的現象名額，二是常用不合理的配置名額。接下來對選取的名額進行簡要逐一分析。
名額分析
cluster 層面名額分析
cluster status
叢集健康狀态，檢測到叢集狀态非 green，則說明巡檢異常結果未處理，或者突發情況導緻，此時巡檢的意義是快速的給出推薦解決方案，讓運維、叢集 owner（開發）能夠有處理的方案，并非一味依賴告警等待運維處理，最大限度的減少異常帶來的影響；
非 green 狀态下，首先檢查的是節點個數是否符合預期，節點數量正常情況下，通常通過explain API進行分析。
GET /_cluster/allocation/explain
{
"index":"index_name",
"shard": 1,
"primary":false
} 分片 allocation 是通過配置設定器和決策器來決定的，explain API 通過決策器的資訊來展現unassigned shard 的異常原因，allocation 原理圖如下：
詳細流程圖連結：
https://www.processon.com/view/link/5f43b268e401fd5f24852544該部分隻要非 green 狀态即為異常，智能巡檢系統會分析出異常原因，并給出推薦解決方案到最終的報告中，如果是出現 red 場景或者 OOM 導緻掉節點場景，則會由告警平台即時通知到使用者和 OPS，使用者和 OPS 可以通過在 PAAS 平台記錄的報告，快速檢視狀态異常的分析；由于巡檢系統非實時，可以手動觸發智能巡檢系統，快速得到最新分析報告與推薦解決方案。
推薦解決方案由兩部分組成，
- 一些特定場景分析後的經驗建議
- 該叢集所有unassigned shard通過explain API的查到的結果集。
示例代碼見示例代碼：
https://www.teambition.com/project/601f63c9997fc9a15fb8e683/app/5eba5fba6a92214d420a3219/workspaces/601f659743c5e10046459556/docs/606285c1eaa1190001e688b1#608588d89645b900464132e3pengding_task
pending_task 反應了 master 節點尚未執行的叢集級别的更改任務（例如：建立索引，更新映射，配置設定分片）的清單。pending_task的任務是分級别的（優先級排序：IMMEDIATE>URGENT>HIGH>NORMAL>LOW>LANGUID），隻有當上一級别的任務執行完畢後，才會執行下一級别的任務，即當出現 HIGH 級别以上的 pending_task 任務時，備份和建立索引等低級别任務将延遲執行。
pending_task 過多會給 master 節點造成壓力，大叢集（大資料量、高并發）情況下，容易造成節點被踢出叢集，甚至叢集不響應的情況。pending_task 積壓的場景一般出現在大叢集中，由于 task 不能快速處理完，會長時間處于積壓狀态，且會越積壓越多，是以異常門檻值可以設定一個較大值，根據經驗值，設定 pending_task 數量大于 100 為異常；
GET _cat/pending_tasks 節點最大 cpu_util 與最小 cpu_util 之差（極差）
cpu_util 是判斷 Elasticsearch 叢集健康程度的重要名額，直接影響叢集的吞吐量、請求的響應時間。一般情況下叢集各個節點 cpu_util 普遍過高的場景比較好處理，增加配置即可讓叢集恢複健康狀态。
而單節點/部分節點出現 cpu_util 過高的情況則不容易定位，如 shard 配置設定不均勻、routing 設定不合理、主控端異常都有可能引發該現象，且可能在程式上線一段時間後才出現，這時巡檢結果對預警與分析有重大意義。
GET _cat/nodes?h=ip,cpu 為了能夠有效起到預警作用，cpu_util 極差的門檻值不宜過大，同時應該考慮到角色分離、冷熱分離等場景，将極差的計算範圍限制到同角色節點之間。
data節點最大 qps 與最小 qps 之差（極差）
節點 qps 名額受到 shard 配置設定、routing 的影響，極差過大代表叢集資料配置設定不均，或有參數幹預，叢集負載不均衡。
為該名額設定門檻值時，同樣需要考慮到角色分離場景，同時由于業務的不同，不同叢集 qps 相差巨大，可以通過換算成瞬時流量的百分比來做極差計算。
GET /_nodes/stats/indices,ingest/search 一次 bulk 請求的數量
bulk 請求适用于大寫入場景，由于減少了大量的連接配接，bulk 的效率遠高于單條 index。一次 bulk 請求涉及到的 doc 數量對性能有較大影響，過小容易造成線程池堆積，過大容易造成逾時。
該名額的門檻值設定，僅針對寫入/更新頻率較大的叢集，由于叢集配置的差異，可給每個配置區間内設定一個最小值作為巡檢的門檻值。
node 層面名額分析
uptime
叢集節點重新開機往往是對叢集性能有着較大影響，接收到的流量會異常，同時減少了一部分吞吐量，對于大叢集而言，重建緩存、shard allocation 會對叢集的響應造成影響。叢集重新開機的原因以及重新開機帶來的影響對于叢集都是一個隐患。
GET _cat/nodes?h=ip,name,node.role,master,uptime 如果啟動時間距離目前時間 1h 以内，則節點發生重新開機，提示使用者檢查重新開機原因、評估帶來的影響。
free_disk
磁盤剩餘空間最直覺的就是影響資料寫入，到達水位線（預設95%）後會進入 read only 狀态，其次磁盤空間還會帶來其他隐患，比如無法操作 forcemerge、甚至 deleteByQuery 也無法完成等。
GET _cat/allocation 出于成本與使用率的考慮，預期狀态磁盤占比不應過低，一般 50% 以内為安全值，由于不同叢集的特性可能會有差異，可以設定 70% 作為異常門檻值。
cpu_util
CPU 對叢集性能影響極大，高 CPU 的場景下，響應時間增加，可能出現大量逾時（讀寫異常）。是以需要平衡 cpu 使用率（一般通過超分、調整配置兩種方式）。
GET _cat/nodes?h=ip,name,cpu 由于業務流量存在周期性波峰波谷，是以門檻值需要在安全範圍内盡量設定大一點，例如設定門檻值為"cpu_util = 90%"。
node上shard數
節點未配置設定 shard，則需要确認資源容量配置設定是否合理，該名額主要針對分片設定不合理導緻的資源浪費。一般發生在分片數量設定不合理、遷移過程（存在 exclude）的場景下。故門檻值可以設定為 "count(shard) = 0"。
GET _cat/allocation shard層面名額分析
number
過多或過少的 shard 在一定場景下都會影響查詢和寫入性能。官方建議的合理的設定數量：每GB 的 heap 不超過 20個 shards; 比如 20GB heap，400個 shards, 30GB heap，不超過 600個shards。Elasticsearch 7.0 版本開始，叢集中每個節點預設限制 1000個shard。
門檻值按照官方建議值設定即可。
GET _cat/shards?h=index,shard,prirep,state,docs,store,ip,node size of per shard
大量的小 shard 會影響寫入和查詢性能，且在同資料量情況下占用更多的記憶體和磁盤。單 shard 過大則有更多的弊端，例如查詢耗時變長、不易于恢複、遷移、容易造成叢集壓力不均衡等。
通常單個 shard 的大小建議在 10GB - 65GB 之間 （經驗值參考：搜尋類控制在 20GB，日志類控制在 50BG），查詢 API 同上。
官方建議shard size、count值參考： https://www.elastic.co/guide/en/elasticsearch/reference/current/size-your-shards.html
Index 層面名額分析
replica
所有叢集的 index 都應該有副本分片，沒有副本分片的 index 在節點 crash 時會丢失資料。
GET index_name/_settings 當 number_of_replicas 為 0 時候異常情況。
動态 mapping
dynamic mapping 設定為 true 會使得 mapping 變得不可維護，且 mapping 源資料由 master 維護、分發，大量變更可能導緻 master 壓力過大，在高峰情況下，可能會使得積壓大量task，引發叢集不響應、踢出節點等問題。
GET /*/_mapping?format=json 巡檢需要檢查出 "dynamic=true（或預設）"的索引的叢集，标記為異常。
refresh_interval
索引 refresh 頻率是影響性能的一個因素，受到 refresh_interval 參數與 buffer 大小的影響，由于業務場景的差異，對 refresh 的設定可能大不相同，可将叢集類型大緻分為搜尋類型與資料分析類型，根據類型的不同設定差異化的門檻值，且叢集不應該出現"refresh_interval = -1"的設定。
GET /*/_settings?include_defaults=true indices.refresh.total
refresh 的頻率影響着 segment 的生成速度與大小，而 segment 過多往往影響查詢性能，并且需要消耗更多的記憶體和磁盤空間。由于預設值為"refresh_interval = 1s"，不考慮 buffer 的影響可以認為 refresh 頻率為 60/min，故巡檢門檻值可以設定到比預設值稍高，例如：count(refresh) = 80/min。
GET /_nodes/stats/indices,ingest/refresh max_result_window
max_result_window 為單次請求傳回 doc 的最大值，預設為 10000，該預設值的限制可以覆寫到所有正常的業務場景。一般是深度分頁、全量查詢、job 查詢可能導緻傳回 doc 數大于10000，觸發異常，而這些場景可以由 scroll、search after 來完成。故該名額門檻值可以設定成該參數預設值。
GET /*/_settings?include_defaults=true jvm 層面名額分析
jvm heap 使用率
jvm 堆的使用率過高有着 OutOfMemory 的風險，并使得 GC 頻率過高，影響請求響應時間。由于使用的 G1 收集器，首次 GC 收集會在預估 GC 時間達到預定值的時候開始觸發，則 heap 使用率的穩定值也随着參數設定而産生較大差異。而該參數主要是為了預防 OutOfMemory 異常，是以該名額門檻值可以設定一個較大值，例如 "heap > 90"。
GET /_nodes/stats/indices,jvm jdk version 一緻性
由于 Elasticsearch 的分布式屬性，叢集存在多節點，每個節點一個單獨的執行個體，需要保證 jdk 版本一緻。
jvm heap segment memory
segment memory 常駐 heap 記憶體，是以 segment memory 的增長會壓縮其他對象的記憶體空間。segment memory 是每個 segment 倒排詞典上層的一個字首索引，即 FST 結構，該字首索引會在 segment 不斷的累積下逐漸增多。
為了防止其對 heap 記憶體過多的占用，需要對該值繼續檢查限制，由于 FST 結構對字首索引進行大量壓縮，正常狀态下對 heap 占用較低，巡檢門檻值也可以設定較低，例如 20% heap_size。
GET /_nodes/stats/indices,ingest/segments full gc
Elasticsearch 7.x 預設使用的 G1 垃圾收集器，是以一般會是 Young GC 或 Mixed GC，如果mixed GC 無法跟上新對象配置設定記憶體的速度，導緻老年代填滿無法繼續進行 Mixed GC，于是使用 full GC 來收集整個 heap。G1 不提供 full GC，使用的是 serial old GC。是以該 full GC 是單線程串行的，且 stop the world，這對業務來說是緻命的。是以該巡檢的門檻值為"count(full gc) > 0"。
GET /_nodes/stats/indices,jvm threadpool 層面名額分析
bulk reject 數量
bulk 出現 reject 意味着線程池中線程被完全占用，且隊列也已經占滿。該名額門檻值可設定為 "count(bulk rejected)>0"。
GET _cat/thread_pool search reject 數量
search 出現 reject 意味着線程池中線程被完全占用，且隊列也已經占滿。該名額門檻值可設定為 "count(search rejected)>0"，查詢 API 同上。
結語
本章詳細介紹了智能巡檢系統的結構與名額選取、門檻值确認，并給出 cluster status 名額采集分析的完整示例代碼。有興趣的讀者可以将智能巡檢系統通過 python 腳本或者 operator 的方式實作。由于 PAAS 系統下管理的大量差異巨大的 Elasticsearch 叢集，巡檢系統實作主要難點與重點，是抽象出合理的名額以及精細化實作。
上述介紹中可以看到部分名額項是固定場景下必現，且解決方案唯一且簡單，例如 shard_limit場景，一般通過 API 調整 total_shards_per_node 參數即可恢複，是以這些場景可以設計成自動化修複，達到簡單的“自愈”，解放開發維護人員的生産力。
示例代碼
# -*- coding:utf-8 -*-
"""
analyse cluster status exception,
比對常用的case, 傳回explain完整結果
"""
import espaas_api
import argparse
import traceback
import json
import logging
import gevent
import time
import datetime
import requests
from libs.log import initlog
from gevent import monkey
monkey.patch_all()
timeout = 10
# 7.10 elasticsearch 有16種決策器
dict = {
'same_shard': '預設配置下一個節點不允許配置設定同一shard的多個副本分片', # 不參與統一處理
'shards_limit': '節點限制單索引shard數，調整相應索引的total_shards_per_node參數',
'disk_threshold': '磁盤空間達到水位線無法配置設定shard，調整磁盤容量、節點數或者清理磁盤',
'max_retry': '達到allocate最大重試次數，嘗試調用API手動retry',
'awareness': '副本配置設定過多，大于awareness的配置',
'cluster_rebalance': '叢集正在Rebalance，可忽略',
'concurrent_rebalance': '叢集目前正在Rebalance，可忽略',
'node_version': '配置設定到的節點版本不一緻，請聯系管理者調整',
'replica_after_primary_active': '主分片未active，等待并觀察主分片',
'filter': '未通過filter，請檢查配置',
'enable': '叢集enable限制',
'throttling': '叢集正在恢複，需要按照優先級恢複 primary > replica',
'rebalance_only_when_active': 'Only allow rebalancing when all shards are active within the shard replication group',
'resize': 'An allocation decider that ensures we allocate the shards of a target index for resize operations next to the source primaries',
'restore_in_progress': 'This allocation decider prevents shards that have failed to be restored from a snapshot to be allocated',
'snapshot_in_progress': 'This allocation decider prevents shards that are currently been snapshotted to be moved to other nodes'
}
def check_exception_status_reason_advice(cluster):
# 省略結果資料入庫代碼
res = {
"status": "",
"reason": "",
"advice": "",
"detail": ""
}
# 1.health API
# 2.node count check
url = gen_url(cluster) + "/_cat/health?format=json"
r = requests.get(url, auth=(cluster["username"], cluster["password"]), timeout=timeout)
if r.status_code == 200:
try:
content = json.loads(r.content)
cluster_status_res = content[0]
res["status"] = cluster_status_res["status"]
if cluster_status_res["status"] == "green":
return res
if cluster_status_res["node.total"] != cluster["nodes_count"]:
res["reason"] = "node_left"
res["advice"] = "請檢查k8s環境中node數, 使用`kubectl describe po pod_name`檢查crash reason"
return res
except:
traceback.print_exc()
else:
res["status"] = "unknown"
return res
# 3.get UNASSIGNED shard
get_all_shard_url = gen_url(cluster) + "_cat/shards?format=json"
get_all_shard_res = requests.get(get_all_shard_url, auth=(cluster["username"], cluster["password"]),
timeout=timeout)
unassigned_shard_list = []
unassigned_shard_final_list = []
if get_all_shard_res.status_code == 200:
try:
shards = json.loads(get_all_shard_res.content)
for shard in shards:
if shard["state"] == "UNASSIGNED":
unassigned_shard_list.append(shard)
if len(unassigned_shard_list) > 20:
unassigned_shard_final_list = unassigned_shard_list[0:20]
else:
unassigned_shard_final_list = unassigned_shard_list
except:
traceback.print_exc()
# 4.explain
# 5.經驗值提取出常用建議
decider_list = []
explanation_list = []
has_advice = 0
explain_res_all = []
for shard_tmp in unassigned_shard_final_list:
index_name_tmp = shard_tmp["index"]
shard_pos = shard_tmp["shard"]
shard_type = shard_tmp["prirep"]
is_primary = False
if shard_type == "p":
is_primary = True
explain_url = gen_url(cluster) + "/_cluster/allocation/explain"
payload = {
"index": index_name_tmp,
"shard": shard_pos,
"primary": is_primary
}
explain_res = requests.post(explain_url, json.dumps(payload),
auth=(cluster["username"], cluster["password"]),
headers={"Content-Type": "application/json"}).json()
explain_res_all.append(explain_res)
# 整理單個分片所有的decider資訊
same_shard_count = 0
for decisions_tmp in explain_res["node_allocation_decisions"]:
for decider_tmp in decisions_tmp['deciders']:
decider_list.append(decider_tmp['decider'])
explanation_list.append(decider_tmp['explanation'])
if decider_tmp['decider'] == "same_shard":
same_shard_count = same_shard_count + 1
tmp_string = ''
explanation_list_string = tmp_string.join(explanation_list)
if explanation_list_string.find("ik_max_word") != -1:
res["reason"] = "max_retry"
res["advice"] = "可能觸發6.8.5Ik分詞器bug，請使用其他相容版本或根據issue fix"
has_advice = 1
break
if explanation_list_string.find("cluster.routing.allocation.enable=primaries") == -1 \
and explanation_list_string.find("cluster.routing.allocation.enable=replicas") == -1:
print '正常'
else:
res["reason"] = "enable"
res["advice"] = "cancel cluster.routing.allocation.enable=replicas/primaries settings"
has_advice = 1
break
if 'same_shard' in decider_list and same_shard_count == cluster["nodes_count"]:
res["reason"] = "same_shard"
res["advice"] = "shard副本設定過多，調整副本數量"
has_advice = 1
break
# 取decider與ES決策器keys的交集
decider_and = list(set(decider_list).intersection(set(dict.keys())))
reason_and = ''
advice_and = ''
if list(set(decider_list).intersection(set(dict.keys()))):
for x in decider_and:
if x != 'same_shard':
reason_and += x + ';'
advice_and += dict[x] + ';'
res["reason"] = reason_and
res["advice"] = advice_and
if has_advice == 0:
res["detail"] = explain_res_all
return res
def gen_url(cluster):
url = 'http://' + cluster["domain"] + ':' + str(cluster["http_port"]) + cluster["http_path"]
if url.endswith("/"):
url = url[:-1]
return url
def main(cluster_id=None):
try:
if cluster_id:
clusters = espaas_api.get("cluster", "/api/get_cluster_by?cluster_id=%s" % cluster_id)["data"]
else:
clusters = espaas_api.get("cluster", "/api/all_status_exception_cluster")["data"]
jobs = []
for cluster in clusters:
try:
jobs.append(gevent.spawn(check_exception_status_reason_advice, cluster))
except:
traceback.print_exc()
logging.info("send job number: %s" % len(jobs))
gevent.joinall(jobs)
except:
traceback.print_exc()
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument("-l", default="-", help="log file")
parser.add_argument("--level", default="info")
args = parser.parse_args()
initlog(level=args.level, log=args.l)
main() 創作人簡介：
張妙成，ES PAAS 平台開發，雲計算技術愛好者。
個人部落格：
https://blog.csdn.net/qq_33999844?spm=1001.2014.3001.5343