容器 | AI | 大数据
所有文章 标签 分类 架构图 关于 DOK文档 科研 WeLink
容器 | AI | 大数据

目录

Spark-Opeartor指标体系

runzhliu 收录于 大数据和机器学习
   约 2602 字   预计阅读 6 分钟 

概述

spark-on-k8s-operator，下文简称 Spark Operator， 背景知识就不介绍太多了，本文主要分享一下 Spark Operator 的指标系统是如何构建的，之后可以按照 Spark Operator 的方法，给自己创建的 Operator 配上指标系统。

/spark-opeartor%E6%8C%87%E6%A0%87%E4%BD%93%E7%B3%BB/image_1e1l81vpp51d7kh1jeqnk1o3um.png

Metrics

Spark Metrics

Spark Operator 目前提供几种自定义的指标，这里自定义的意思是以 Spark Application 这个自定义资源对象 CRD 为监控对象，围绕 Spark Application 创建的一些监控指标，比如 CRD 成功、失败或者正在运行的个数，这样可以让管理员更好的监控和管理 Spark Operator 的运行情况。

下面是目前 Spark Operator 官方提供的一些指标，自定义指标基本都在 sparkapp_metrics.go 里有定义。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

type sparkAppMetrics struct {
    ...
	sparkAppSubmitCount           *prometheus.CounterVec
	sparkAppSuccessCount          *prometheus.CounterVec
	sparkAppFailureCount          *prometheus.CounterVec
	sparkAppFailedSubmissionCount *prometheus.CounterVec
	sparkAppRunningCount          *util.PositiveGauge
	sparkAppSuccessExecutionTime  *prometheus.SummaryVec
	sparkAppFailureExecutionTime  *prometheus.SummaryVec
	sparkAppExecutorRunningCount  *util.PositiveGauge
	sparkAppExecutorFailureCount  *prometheus.CounterVec
	sparkAppExecutorSuccessCount  *prometheus.CounterVec
}

可以看到 Spark Operator 记录了，Spark App 的提交数、成功数、失败数、提交 失败数、当前运行数、运行成功的时间统计、运行失败的时间统计、运行的 Executor 数、失败的 Executor 数以及成功的 Executor 数。

newSparkAppMetrics new 实际就是去注册的意思。按照指标的类型，CounterVec 或者 GaugeVec 等，配置好 TYPE 和 HELP，或者 LABEL 等。

熟悉 Prometheus 的同学应该知道，Counter, Gauge, Summary, Histogram 几种类型，但是我们从上面的 sparkAppMetrics 上还看到了 PositiveGauge。

有两个比较特殊的指标。一个是 sparkAppRunningCount 和 sparkAppExecutorRunningCount，因为他们都是 Gauge 类型的，但是是不会降低到0以下的，所以这里注册的类型，是自定义的 PostiveGauge。

1
2
3
4
5

// spark app 正在运行的计数
sparkAppRunningCount := util.NewPositiveGauge(util.CreateValidMetricNameLabel(prefix, "spark_app_running_count"), "Spark App Running Count via the Operator", validLabels)

// spark app executor 正在运行的计数
sparkAppExecutorRunningCount := util.NewPositiveGauge(util.CreateValidMetricNameLabel(prefix, "spark_app_executor_running_count"), "Spark App Running Executor Count via the Operator", validLabels)

这个是 Spark Operator 自行实现的一个只用于正数的 Gauge。大家应该知道 Gauge 其实是可以亦正亦负的，而这个自定义指标主要是针对那种会随着时间和 CRD 对象的运行情况时而增加时而减少，但是绝对不会跌破0的指标。

1
2
3
4
5

type PositiveGauge struct {
	mux         sync.RWMutex
	name        string
	gaugeMetric *prometheus.GaugeVec
}

实现的原理很简单，就是一个读写锁 mux 和一个指标名 name，以及一个正常的可以亦正亦负的 gaugeMetrics。下面是其构造方法 NewPositiveGauge。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

func NewPositiveGauge(name string, description string, labels []string) *PositiveGauge {
	validLabels := make([]string, len(labels))
	for i, label := range labels {
		validLabels[i] = CreateValidMetricNameLabel("", label)
	}

	gauge := prometheus.NewGaugeVec(
		prometheus.GaugeOpts{
			Name: name,
			Help: description,
		},
		validLabels,
	)

	return &PositiveGauge{
		gaugeMetric: gauge,
		name:        name,
	}
}

WorkQueue Metrics

这里的 WorkQueue Metrics 是指 client-go 库里的 WorkQueue 包里的 metrics.go。因为 Spark Operator 实现的 Controller 里，用到了 rate limiting WorkQueue 这个工作队列。

https://github.com/kubernetes/client-go/blob/master/util/workqueue/metrics.go

Spark Operator 里自定义的 WorkQueueMetrics 主要是用于暴露这个工作队列的指标，其实就是给原来的 workqueque 的指标加上一个 prefix，这样后面收集指标和使用指标时候会更方便。

WorkQueueMetrics 的指标的结构体。

1
2
3

type WorkQueueMetrics struct {
	prefix string
}

下面是 WorkQueque Metrics 的几种类型。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

// Depth Metric for the kubernetes workqueue.
func (p *WorkQueueMetrics) NewDepthMetric(name string) workqueue.GaugeMetric {
    ...
}

// Adds Count Metrics for the kubernetes workqueue.
func (p *WorkQueueMetrics) NewAddsMetric(name string) workqueue.CounterMetric {
    ...
}

// Latency Metric for the kubernetes workqueue.
func (p *WorkQueueMetrics) NewLatencyMetric(name string) workqueue.SummaryMetric {
    ...
}

// WorkDuration Metric for the kubernetes workqueue.
func (p *WorkQueueMetrics) NewWorkDurationMetric(name string) workqueue.SummaryMetric {
    ...
}

// Retry Metric for the kubernetes workqueue.
func (p *WorkQueueMetrics) NewRetriesMetric(name string) workqueue.CounterMetric {
    ...
}

func (p *WorkQueueMetrics) NewUnfinishedWorkSecondsMetric(name string) workqueue.SettableGaugeMetric {
    ...
}

func (p *WorkQueueMetrics) NewLongestRunningProcessorMicrosecondsMetric(name string) workqueue.SettableGaugeMetric {
    ...
}

指标初始化

Spark Metrics 和 WorkQueue Metrics 两部分指标的初始化都在 InitializeMetrics 方法里。Spark Metrics 初始化，是常规的 Prometheus 指标收集初始化的方式WorkQueue Metrics 则是通过 WorkQueue 的 Provider 来填充。

1
2
3
4
5
6
7
8
9

func InitializeMetrics(metricsConfig *MetricConfig) {
	// Start the metrics endpoint for Prometheus to scrape
	http.Handle(metricsConfig.MetricsEndpoint, promhttp.Handler())
	go http.ListenAndServe(fmt.Sprintf(":%s", metricsConfig.MetricsPort), nil)
	glog.Infof("Started Metrics server at localhost:%s%s",   metricsConfig.MetricsPort, metricsConfig.MetricsEndpoint)

	workQueueMetrics := WorkQueueMetrics{prefix: metricsConfig.MetricsPrefix}
	workqueue.SetProvider(&workQueueMetrics)
}

其他

1
2
3
4

func CreateValidMetricNameLabel(prefix, name string) string {
	// "-" is not a valid character for prometheus metric names or labels.
	return strings.Replace(prefix+name, "-", "_", -1)
}

根据 Prometheus 的指引，Metric Name 除了可以用大小写字母和数字以为，还可以用下划线_，但是不能是中划线。CreateValidMetricNameLabel 方法是用来矫正指标名的，以防制造不符合规范的指标名，导致指标无法被 Prometheus 拉取。

The metric name specifies the general feature of a system that is measured (e.g. http_requests_total - the total number of HTTP requests received). It may contain ASCII letters and digits, as well as underscores and colons. It must match the regex [a-zA-Z_:][a-zA-Z0-9_:]*

1
2
3
4
5
6
7
8
9

func RegisterMetric(metric prometheus.Collector) {
	if err := prometheus.Register(metric); err != nil {
		// Ignore AlreadyRegisteredError.
		if _, ok := err.(prometheus.AlreadyRegisteredError); ok {
			return
		}
		glog.Errorf("failed to register metric: %v", err)
	}
}

构建指标体系，需要有一个注册 registry 的过程。传入的就是一个 metrics 类型，然后通过注册接口注册即可。

1
2
3
4
5
6

type MetricConfig struct {
	MetricsEndpoint string
	MetricsPort     string
	MetricsPrefix   string
	MetricsLabels   []string
}

MeticsConfig 是 Spark Operator 的指标配置信息类型，包括 Operator 应用暴露的指标 Endpoint，指标端口 Port，指标的前缀（可以用于快速过滤）以及指标的 Labels（注意是一个数组，意思是指标会被打上这个数组的里的名字作为 Label）。

1
2
3
4
5
6
7

func fetchGaugeValue(m *prometheus.GaugeVec, labels map[string]string) float64 {
	// Hack to get the current value of the metric to support PositiveGauge
	pb := &prometheusmodel.Metric{}

	m.With(labels).Write(pb)
	return pb.GetGauge().GetValue()
}

工作时的指标

很重要的，就是根据 Spark App 的 CRD 对象的状态来输出指标了，这是指标体系最重要的部分。注释部分就是描述了当 newState 处于何种状态时，应该如何去记录指标的变化。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

func (sm *sparkAppMetrics) exportMetrics(oldApp, newApp *v1beta2.SparkApplication) {
	oldState := oldApp.Status.AppState.State
	newState := newApp.Status.AppState.State
	if newState != oldState {
		switch newState {
		case v1beta2.SubmittedState:
		    // sparkAppSubmitCount + 1
		case v1beta2.RunningState:
		    // sparkAppRunningCount + 1
		case v1beta2.SucceedingState:
		    // sparkAppSuccessCount + 1
   		    // sparkAppRunningCount - 1
		case v1beta2.FailingState:
		    // sparkAppFailureCount + 1
		    // sparkAppRunningCount - 1
		case v1beta2.FailedSubmissionState:
		    // sparkAppFailedSubmissionCount + 1
		}
	}

	// Potential Executor status updates
	// 不赘述了
	for executor, newExecState := range newApp.Status.ExecutorState {
		switch newExecState {
		case v1beta2.ExecutorRunningState:
		case v1beta2.ExecutorCompletedState:
		case v1beta2.ExecutorFailedState:
		}
	}
}

指标构建和初始化完成了，也配置好什么时候该输出指标的方法了，然后看看 exportMetrics 这个方法是在什么时候被调用的。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

// updateStatusAndExportMetrics updates the status of the SparkApplication and export the metrics.
func (c *Controller) updateStatusAndExportMetrics(oldApp, newApp *v1beta2.SparkApplication) error {
	// Skip update if nothing changed.
	if equality.Semantic.DeepEqual(oldApp, newApp) {
		return nil
	}

	updatedApp, err := c.updateApplicationStatusWithRetries(oldApp, func(status *v1beta2.SparkApplicationStatus) {
		*status = newApp.Status
	}, c.k8sMinorVersion)

	// Export metrics if the update was successful.
	if err == nil && c.metrics != nil {
	    // 调用
		c.metrics.exportMetrics(oldApp, updatedApp)
	}

	return err
}

当 Spark App CRD 对象在对 status 字段进行更新的时候，除了更新 status 以外，还会调用 exportMetics 方法来输出指标。

显然 updateStatusAndExportMetrics 这个方法是 Controller 在同步 CRD 对象的时候调用的。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

if appToUpdate != nil {
	glog.V(2).Infof("Trying to update SparkApplication %s/%s, from: [%v] to [%v]", app.Namespace, app.Name, app.Status, appToUpdate.Status)
	// 当 appToUpdate 不为 nil，证明需要去更新 CRD 对象的 status 了
	// 这个时候同时输出指标
	err = c.updateStatusAndExportMetrics(app, appToUpdate)
	if err != nil {
		glog.Errorf("failed to update SparkApplication %s/%s: %v", app.Namespace, app.Name, err)
		return err
	}
}

总结

给 Operator 加指标，总体来说不是非常难的事情，难的地方在于判断自己到底需要收集什么指标，做什么样的监控。另外如果需要自定义指标系统，可以参考 Spark Operator 的做法，自定义 CRD 层面的 Metrics，如果还需要监控工作队列，直接通过 client-go 的 workqueue 的接口去做即可。

下面是一份真实的 Spark Operator 输出的指标，可供各位读者参考。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

...
# HELP spark_app_executor_failure_count Spark App Failed Executor Count via the Operator
# TYPE spark_app_executor_failure_count counter
spark_app_executor_failure_count{project="Unknown"} 7
# HELP spark_app_executor_running_count Spark App Running Executor Count via the Operator
# TYPE spark_app_executor_running_count gauge
spark_app_executor_running_count{project="Unknown"} 7
spark_app_executor_running_count{project="demo"} 22
# HELP spark_app_executor_success_count Spark App Successful Executor Count via the Operator
# TYPE spark_app_executor_success_count counter
spark_app_executor_success_count{project="demo"} 65
# HELP spark_app_running_count Spark App Running Count via the Operator
# TYPE spark_app_running_count gauge
spark_app_running_count{project="demo"} 0
# HELP spark_app_submit_count Spark App Submits via the Operator
# TYPE spark_app_submit_count counter
spark_app_submit_count{project="demo"} 1
# HELP spark_app_success_count Spark App Success Count via the Operator
# TYPE spark_app_success_count counter
spark_app_success_count{project="demo"} 1
# HELP spark_app_success_execution_time_microseconds Spark App Successful Execution Runtime via the Operator
# TYPE spark_app_success_execution_time_microseconds summary
spark_app_success_execution_time_microseconds{project="demo",quantile="0.5"} NaN
spark_app_success_execution_time_microseconds{project="demo",quantile="0.9"} NaN
spark_app_success_execution_time_microseconds{project="demo",quantile="0.99"} NaN
spark_app_success_execution_time_microseconds_sum{project="demo"} 5.83e+08
spark_app_success_execution_time_microseconds_count{project="demo"} 1
# HELP spark_application_controller_adds Total number of adds handled by workqueue: spark-application-controller
# TYPE spark_application_controller_adds counter
spark_application_controller_adds 120
# HELP spark_application_controller_depth Current depth of workqueue: spark-application-controller
# TYPE spark_application_controller_depth gauge
spark_application_controller_depth 0
# HELP spark_application_controller_latency Latency for workqueue: spark-application-controller
# TYPE spark_application_controller_latency summary
spark_application_controller_latency{quantile="0.5"} NaN
spark_application_controller_latency{quantile="0.9"} NaN
spark_application_controller_latency{quantile="0.99"} NaN
spark_application_controller_latency_sum 6.150365e+06
spark_application_controller_latency_count 120
# HELP spark_application_controller_longest_running_processor_microseconds Longest running processor microseconds: spark-application-controller
# TYPE spark_application_controller_longest_running_processor_microseconds gauge
spark_application_controller_longest_running_processor_microseconds 0
# HELP spark_application_controller_retries Total number of retries handled by workqueue: spark-application-controller
# TYPE spark_application_controller_retries counter
spark_application_controller_retries 472
# HELP spark_application_controller_unfinished_work_seconds Unfinished work seconds: spark-application-controller
# TYPE spark_application_controller_unfinished_work_seconds gauge
spark_application_controller_unfinished_work_seconds 0
# HELP spark_application_controller_work_duration How long processing an item from workqueue spark-application-controller takes.
# TYPE spark_application_controller_work_duration summary
spark_application_controller_work_duration{quantile="0.5"} NaN
spark_application_controller_work_duration{quantile="0.9"} NaN
spark_application_controller_work_duration{quantile="0.99"} NaN
spark_application_controller_work_duration_sum 1.2046801e+07
spark_application_controller_work_duration_count 120
警告
本文最后更新于 2019年10月9日，文中内容可能已过时，请谨慎参考。
更新于 2019-10-09
 Kubernetes, 大数据, Spark
返回 | 主页