【prometheus】-07 Kubernetes雲原生監控之kube-state-metrics叢集資源監控
2021-09-06
【prometheus】-06 Kubernetes雲原生監控之cAdvisor容器資源監控
2021-09-01
【prometheus】-05 Kubernetes雲原生監控之節點性能監控
2021-08-30
【prometheus】-04 輕松搞定Prometheus Eureka服務發現
2021-08-25
【prometheus】-03 輕松搞定Prometheus檔案服務發現
2021-08-23
【prometheus】-02 一張圖徹底搞懂Prometheus服務發現機制
2021-08-18
【prometheus】- 01 雲原生時代的監控系統入門
2021-08-16
Prometheus服務發現機制之Kubernetes
概述
分析過雲原生監控接入方案，下面開始看下雲原生服務發現機制。Prometheus本身就是作為雲原生監控出現的，是以對雲原生服務發現支援具有天然優勢。
Kubernetes
服務發現協定允許使用
Kubernetes Rest API
檢索出
Prometheus
需要監控的
targets
，并且跟着叢集狀态進行同步變更。
kubernetes_sd_configs
表示基于
Kubernetes
進行服務發現，服務發現目标類型使用
role
表示，比如：
role=service
，表示針對
Kubernetes
中的
service
資源對象，進行具體的服務發現操作。
kubernetes_sd_configs
支援的
role
包括：
node、service、pod、endpoints、ingress
。
原理
基于
Kubernetes
進行服務發現，主要針對
Kubernetes
中的
service、pod、node
等資源對象進行服務發現，
Prometheus
使用
client-go
對
role
中指定的資源對象進行監聽。一般
Prometheus
部署在
Kubernetes
叢集中的話，
Prometheus
可以直接利用指定的
Service Account
對
Kubernetes API
進行通路。若
Prometheus
在
Kubernetes
叢集之外，則
kubernetes_sd_configs
還需指定監控叢集的
API Server
的
URL
以及相關的認證資訊，進而能夠建立對應叢集的
Client
。
“
client-go是kubernetes官方提供的go語言的用戶端庫，go應用使用該庫可以通路kubernetes的API Server，這樣我們就能通過程式設計來對kubernetes資源進行增删改查操作。
配置示例：
- job_name: kubernetes-pod
metrics_path: /metrics
kubernetes_sd_configs:
- role: pod
namespaces:
names:
- 'test01'
api_server: https://apiserver.simon:6443
bearer_token_file: d:/token.k8s
tls_config:
insecure_skip_verify: true
bearer_token_file: d:/token.k8s
tls_config:
insecure_skip_verify: true
協定分析
Kubernetes
服務發現大緻原理如下圖：
1、通過
clientset
通路
API Server
，根據
role
配置擷取不同的叢集資源對象；
2、通過
List & Watch
機制，注冊監聽事件：
p.informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: func(o interface{}) {
podAddCount.Inc()
p.enqueue(o)
},
DeleteFunc: func(o interface{}) {
podDeleteCount.Inc()
p.enqueue(o)
},
UpdateFunc: func(_, o interface{}) {
podUpdateCount.Inc()
p.enqueue(o)
},
})
通過
informer.AddEventHandler
函數可以為叢集資源添加資源事件回調方法，支援3種資源事件回調方法：
AddFunc、DeleteFunc、UpdateFunc
，分别對應新增資源、修改資源和删除資源時事件觸發。
3、資源變更注冊回調方法中，将目标資源對象轉成
key
放入到隊列
queue
中，如下
pod
資源：
func (p *Pod) enqueue(obj interface{}) {
//obj是pod資源對象，通過DeletionHandlingMetaNamespaceKeyFunc将其轉換成key
//比如key=test01/nginx-deployment-5ffc5bf56c-n2pl8，即namespace/pod_name格式
key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
if err != nil {
return
}
p.queue.Add(key)
}
4、背景
goroutines
無限循環執行
process
邏輯，
process
邏輯中就是不停從
queue
中提取資料進行處理，比如
pod.go
對應邏輯如下：
func (p *Pod) process(ctx context.Context, ch chan<- []*targetgroup.Group) bool {
keyObj, quit := p.queue.Get()
if quit {
return false
}
defer p.queue.Done(keyObj)
key := keyObj.(string)
//與 MetaNamespaceKeyFunc() 功能相反的是 SplitMetaNamespaceKey() 函數，它将傳入的 Key 分解，傳回對象所在的命名空間和對象名稱。
namespace, name, err := cache.SplitMetaNamespaceKey(key)
if err != nil {
return true
}
//根據key擷取資源對象obj
o, exists, err := p.store.GetByKey(key)
if err != nil {
return true
}
if !exists {
//pod被删除時，exists=false
// 然後發送targets為空的tg，即移除
send(ctx, ch, &targetgroup.Group{Source: podSourceFromNamespaceAndName(namespace, name)})
return true
}
pod, err := convertToPod(o)
if err != nil {
level.Error(p.logger).Log("msg", "converting to Pod object failed", "err", err)
return true
}
//p.buildPod(pod):将資源對象資訊轉成target groups
send(ctx, ch, p.buildPod(pod))
return true
}
大緻邏輯：
a、根據從
queue
中提取的
key
，使用
p.store.GetByKey(key)
擷取對應的資源對象，比如
pod、service
等對象；
b、如果對象不存在，則表示資源對象被删除，則建立一個
targets
集合為空的
target groups
，這樣
Scrape Manager
就會移除
targets
；
c、使用
buildXXX(obj)
将資源對象解析成
target groups
，如
buildNode()、buildPod()
等；
d、最後使用
send()
方法将解析的
target groups
通過通道
channel
傳遞出去，最終傳遞給
Scrape Manager
，這樣
target groups
中
targets
将被
Prometheus
抓取監控資料。
pod
資源的
target groups
結構如下示例，每個pod對象都會被解析成
target groups
，其中包含
targets
集合、
labels
标簽集合：
Discovery建立
1、假如我們定義如下抓取作業：
- job_name: kubernetes-nodes-cadvisor
metrics_path: /metrics
scheme: https
kubernetes_sd_configs:
- role: node
api_server: https://apiserver.simon:6443
bearer_token_file: d:/token.k8s
tls_config:
insecure_skip_verify: true
bearer_token_file: d:/token.k8s
tls_config:
insecure_skip_verify: true
relabel_configs:
# 将标簽(.*)作為新标簽名，原有值不變
- action: labelmap
regex: __meta_kubernetes_node_label_(.*)
# 修改NodeIP:10250為APIServerIP:6443
- action: replace
regex: (.*)
source_labels: ["__address__"]
target_label: __address__
replacement: 192.168.52.151:6443
- action: replace
source_labels: [__meta_kubernetes_node_name]
target_label: __metrics_path__
regex: (.*)
replacement: /api/v1/nodes/${1}/proxy/metrics/cadvisor
會被解析成
kubernetes.SDConfig
如下：
kubernetes.SDConfig
定義如下：
type SDConfig struct {
APIServer config.URL `yaml:"api_server,omitempty"`
Role Role `yaml:"role"`
HTTPClientConfig config.HTTPClientConfig `yaml:",inline"`
NamespaceDiscovery NamespaceDiscovery `yaml:"namespaces,omitempty"`
Selectors []SelectorConfig `yaml:"selectors,omitempty"`
}
2、
Discovery
建立
//建立Clientset，可看成操作Kubernetes API的用戶端
c, err := kubernetes.NewForConfig(kcfg)
if err != nil {
return nil, err
}
return &Discovery{
client: c,
logger: l,
role: conf.Role,
namespaceDiscovery: &conf.NamespaceDiscovery,
discoverers: make([]discovery.Discoverer, 0),
selectors: mapSelector(conf.Selectors),
}, nil
3、
Discovery
建立完成，最後會調用
Discovery.Run()
啟動服務發現：
func (d *Discovery) Run(ctx context.Context, ch chan<- []*targetgroup.Group) {
d.Lock()
namespaces := d.getNamespaces()
switch d.role {
case RoleEndpointSlice:
role=endpointslice邏輯
case RoleEndpoint:
role=endpoints邏輯
case RolePod:
role=pod邏輯
case RoleService:
role=service邏輯
case RoleIngress:
role=ingress邏輯
case RoleNode:
role=node邏輯
default:
level.Error(d.logger).Log("msg", "unknown Kubernetes discovery kind", "role", d.role)
}
var wg sync.WaitGroup
for _, dd := range d.discoverers {
wg.Add(1)
go func(d discovery.Discoverer) {
defer wg.Done()
d.Run(ctx, ch)
}(dd)
}
d.Unlock()
wg.Wait()
<-ctx.Done()
}
4、注冊叢集資源對象監聽事件回調邏輯：
for _, namespace := range namespaces {
p := d.client.CoreV1().Pods(namespace)
plw := &cache.ListWatch{
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
options.FieldSelector = d.selectors.pod.field
options.LabelSelector = d.selectors.pod.label
return p.List(ctx, options)
},
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
options.FieldSelector = d.selectors.pod.field
options.LabelSelector = d.selectors.pod.label
return p.Watch(ctx, options)
},
}
pod := NewPod(
log.With(d.logger, "role", "pod"),
cache.NewSharedInformer(plw, &apiv1.Pod{}, resyncPeriod),
)
d.discoverers = append(d.discoverers, pod)
go pod.informer.Run(ctx.Done())
}