目录

Spark-Opeartor指标体系

概述

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 等,配置好 TYPEHELP,或者 LABEL 等。

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

有两个比较特殊的指标。一个是 sparkAppRunningCountsparkAppExecutorRunningCount,因为他们都是 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日,文中内容可能已过时,请谨慎参考。