1 Overview

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

2 Metrics

2.1 Spark Metrics

Spark Operator 目前提供几种自定义的指标，这里自定义的意思是以 Spark Application 这个自定义资源对象为监控对象，围绕 Spark Application 创建的一些监控指标，来让 Spark Operator 的维护者，更好的监控 Operator 中 CRD 对象的情况。

下面是目前 Spark Operator 的指标。自定义指标基本都在 sparkapp_metrics.go 里定义。

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。

sparkAppRunningCount := util.NewPositiveGauge(util.CreateValidMetricNameLabel(prefix, "spark_app_running_count"), "Spark App Running Count via the Operator", validLabels)
sparkAppExecutorRunningCount := util.NewPositiveGauge(util.CreateValidMetricNameLabel(prefix, "spark_app_executor_running_count"), "Spark App Running Executor Count via the Operator", validLabels)

这个是 Spark Operator 自实现的一个只用于正数的 Gauge，因为大家都知道 Gauge 其实是可以亦正亦负的。这个指标收集一些会增长会减少，但是不会跌破0的类型。

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

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

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,
    }
}

2.2 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 的指标的结构体。

type WorkQueueMetrics struct {
    prefix string
}

下面是 Workqueque Metrics 的几种类型。

// 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 {
    ...
}

2.3 指标初始化

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

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)
}

2.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_:]*

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 类型，然后通过注册接口注册即可。

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

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

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()
}

2.5 工作时的指标

很重要的，就是根据 Spark App 的 CRD 对象的状态来输出指标了，这是指标体系最重要的部分。这里看不懂，前面都白看。

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 这个方法是在什么时候被调用的。

// 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 对象的时候调用的。

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
    }
}

3 Summary

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

下面是一份真实的 Spark Operator 输出的指标，供参考。

...
# 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

ITPub博客

Spark Opeartor的指标体系