Golang框架实战-KisFlow流式计算框架(11)-Prometheus Metrics统计

Golang框架实战-KisFlow流式计算框架专栏

Golang框架实战-KisFlow流式计算框架(1)-概述
Golang框架实战-KisFlow流式计算框架(2)-项目构建/基础模块-(上)
Golang框架实战-KisFlow流式计算框架(3)-项目构建/基础模块-(下)
Golang框架实战-KisFlow流式计算框架(4)-数据流
Golang框架实战-KisFlow流式计算框架(5)-Function调度
Golang框架实战-KisFlow流式计算框架(6)-Connector
Golang框架实战-KisFlow流式计算框架(7)-配置导入与导出
Golang框架实战-KisFlow流式计算框架(8)-KisFlow Action
Golang框架实战-KisFlow流式计算框架(9)-Cache/Params 数据缓存与数据参数
Golang框架实战-KisFlow流式计算框架(10)-Flow多副本

在介绍本章之前，我们先普及一下Prometheus Metrics的服务启动办法。 有关Prometheus是个什么东东，希望大家可以去额外补充下知识，我用一句大白话来解释就是，系统的监控指标。

那么KisFlow作为流式计算框架，那么有关每个Function的调度时间、总体的数据量、算法速度等等指标可能也是项目中或者开发者所要关注的一些数据，那么这些数据，经过KisFlow，可以通过Prometheus的Metrics打点来进行记录。

所以接下来我们可以通过全局的配置，开发者可以选择是否开启Prometheus的数据埋点能力。

10.1 Prometheus Metrics服务

10.1.1 prometheus client sdk

首先在kis-flow/go.mod中，新增require：

module kis-flow

go 1.18

require (
    github.com/google/uuid v1.5.0
    github.com/patrickmn/go-cache v2.1.0+incompatible
    github.com/prometheus/client_golang v1.14.0  //++++++++
    gopkg.in/yaml.v3 v3.0.1
)

这里我们采用prometheus的官方Golang客户端SDK。https://github.com/prometheus/client_golang 有关具体的介绍，参考官方的README文档：https://github.com/prometheus/client_golang/blob/main/README.md

其次，我们先简单的写一个prometheus的服务，可以让外界得到KisFlow服务的相关指标数据等。 在kis-flow/下新建kis-flow/metrics/目录，作为KisFlow统计指标的代码部分。

kis-flow/metrics/kis_metrics.go

package metrics

import (
    "github.com/prometheus/client_golang/prometheus/promhttp"
    "kis-flow/common"
    "kis-flow/log"
    "net/http"
)


// RunMetricsService 启动Prometheus监控服务
func RunMetricsService(serverAddr string) error {

    // 注册Prometheus 监控路由路径
    http.Handle(common.METRICS_ROUTE, promhttp.Handler())

    // 启动HttpServer
    err := http.ListenAndServe(serverAddr, nil) //多个进程不可监听同一个端口
    if err != nil {
        log.Logger().ErrorF("RunMetricsService err = %s\n", err)
    }

    return err
}

其中METRICS_ROUTE作为监控服务的http路由路径，定义在kis-flow/common/const.go中：
如下：

kis-flow/common/const.go

// ... ...

// metrics
const (
    METRICS_ROUTE string = "/metrics"
)

// ... ...

接下来来简单说明下上述的代码，RunMetricsService() 是启动prometheus监控的http服务代码，为什么要启动这个服务，目的是，我们可以通过http的请求来获取kisflow目前进程的运行指标，那么都有哪些指标，现在我们还没有进行统计，prometheus会默认提供当前进程的go版本号、gc垃圾回收时间、内存分配等等基础统计指标信息。

  • serverAddr参数： 这个作为prometheus监控的地址，一般是本地地址加上一个端口号：如"0.0.0.0:20004"。
http.Handle(common.METRICS_ROUTE, promhttp.Handler())

这行代码则表示，"0.0.0.0:20004/metrics" 为获取指标入口。

上述代码写完之后，别忘了拉去一下https://github.com/prometheus/client_golang 相关的依赖包。

$ go mod tidy

拉取之后，当前的go.mod 的依赖大致如下(会有版本号的区别)：

kis-flow/go.mod

module kis-flow

go 1.18

require (
    github.com/google/uuid v1.5.0
    github.com/patrickmn/go-cache v2.1.0+incompatible
    github.com/prometheus/client_golang v1.14.0
    gopkg.in/yaml.v3 v3.0.1
)

require (
    github.com/beorn7/perks v1.0.1 // indirect
    github.com/cespare/xxhash/v2 v2.1.2 // indirect
    github.com/golang/protobuf v1.5.2 // indirect
    github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
    github.com/prometheus/client_model v0.3.0 // indirect
    github.com/prometheus/common v0.37.0 // indirect
    github.com/prometheus/procfs v0.8.0 // indirect
    golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a // indirect
    google.golang.org/protobuf v1.28.1 // indirect
)

10.1.2 prometheus server 服务启动单元测试

接下来来简单测试下服务是否可以启动。

在kis-flow/test/下创建prometheus_server_test.go文件：

kis-flow/test/prometheus_server_test.go

package test

import (
    "kis-flow/metrics"
    "testing"
)

func TestPrometheusServer(t *testing.T) {

    err := metrics.RunMetricsService("0.0.0.0:20004")
    if err != nil {
        panic(err)
    }
}

这里的监控地址为"0.0.0.0:20004"。接下来来启动本单元测试用例，打开一个终端A，cd到kis-flow/test/目录下：

 $ go test -test.v -test.paniconexit0 -test.run TestPrometheusServer
 === RUN   TestPrometheusServer

然后打开另一个终端B，输入如下指令，模拟http客户端进行请求：

 $ curl http://0.0.0.0:20004/metrics

之后我们在终端B得到监控指标的结果如下：

# HELP go_gc_duration_seconds A summary of the pause duration of garbage collection cycles.
# TYPE go_gc_duration_seconds summary
go_gc_duration_seconds{quantile="0"} 0
go_gc_duration_seconds{quantile="0.25"} 0
go_gc_duration_seconds{quantile="0.5"} 0
go_gc_duration_seconds{quantile="0.75"} 0
go_gc_duration_seconds{quantile="1"} 0
go_gc_duration_seconds_sum 0
go_gc_duration_seconds_count 0
# HELP go_goroutines Number of goroutines that currently exist.
# TYPE go_goroutines gauge
go_goroutines 8
# HELP go_info Information about the Go environment.
# TYPE go_info gauge
go_info{version="go1.18.8"} 1
# HELP go_memstats_alloc_bytes Number of bytes allocated and still in use.
# TYPE go_memstats_alloc_bytes gauge
go_memstats_alloc_bytes 3.2364e+06
# HELP go_memstats_alloc_bytes_total Total number of bytes allocated, even if freed.
# TYPE go_memstats_alloc_bytes_total counter
go_memstats_alloc_bytes_total 3.2364e+06
# HELP go_memstats_buck_hash_sys_bytes Number of bytes used by the profiling bucket hash table.
# TYPE go_memstats_buck_hash_sys_bytes gauge
go_memstats_buck_hash_sys_bytes 1.446507e+06
# HELP go_memstats_frees_total Total number of frees.
# TYPE go_memstats_frees_total counter
go_memstats_frees_total 0
# HELP go_memstats_gc_sys_bytes Number of bytes used for garbage collection system metadata.
# TYPE go_memstats_gc_sys_bytes gauge
go_memstats_gc_sys_bytes 3.561224e+06
# HELP go_memstats_heap_alloc_bytes Number of heap bytes allocated and still in use.
# TYPE go_memstats_heap_alloc_bytes gauge
go_memstats_heap_alloc_bytes 3.2364e+06
# HELP go_memstats_heap_idle_bytes Number of heap bytes waiting to be used.
# TYPE go_memstats_heap_idle_bytes gauge
go_memstats_heap_idle_bytes 4.636672e+06
# HELP go_memstats_heap_inuse_bytes Number of heap bytes that are in use.
# TYPE go_memstats_heap_inuse_bytes gauge
go_memstats_heap_inuse_bytes 3.260416e+06
# HELP go_memstats_heap_objects Number of allocated objects.
# TYPE go_memstats_heap_objects gauge
go_memstats_heap_objects 21294
# HELP go_memstats_heap_released_bytes Number of heap bytes released to OS.
# TYPE go_memstats_heap_released_bytes gauge
go_memstats_heap_released_bytes 4.636672e+06
# HELP go_memstats_heap_sys_bytes Number of heap bytes obtained from system.
# TYPE go_memstats_heap_sys_bytes gauge
go_memstats_heap_sys_bytes 7.897088e+06
# HELP go_memstats_last_gc_time_seconds Number of seconds since 1970 of last garbage collection.
# TYPE go_memstats_last_gc_time_seconds gauge
go_memstats_last_gc_time_seconds 0
# HELP go_memstats_lookups_total Total number of pointer lookups.
# TYPE go_memstats_lookups_total counter
go_memstats_lookups_total 0
# HELP go_memstats_mallocs_total Total number of mallocs.
# TYPE go_memstats_mallocs_total counter
go_memstats_mallocs_total 21294
# HELP go_memstats_mcache_inuse_bytes Number of bytes in use by mcache structures.
# TYPE go_memstats_mcache_inuse_bytes gauge
go_memstats_mcache_inuse_bytes 9600
# HELP go_memstats_mcache_sys_bytes Number of bytes used for mcache structures obtained from system.
# TYPE go_memstats_mcache_sys_bytes gauge
go_memstats_mcache_sys_bytes 15600
# HELP go_memstats_mspan_inuse_bytes Number of bytes in use by mspan structures.
# TYPE go_memstats_mspan_inuse_bytes gauge
go_memstats_mspan_inuse_bytes 46376
# HELP go_memstats_mspan_sys_bytes Number of bytes used for mspan structures obtained from system.
# TYPE go_memstats_mspan_sys_bytes gauge
go_memstats_mspan_sys_bytes 48960
# HELP go_memstats_next_gc_bytes Number of heap bytes when next garbage collection will take place.
# TYPE go_memstats_next_gc_bytes gauge
go_memstats_next_gc_bytes 4.194304e+06
# HELP go_memstats_other_sys_bytes Number of bytes used for other system allocations.
# TYPE go_memstats_other_sys_bytes gauge
go_memstats_other_sys_bytes 1.171301e+06
# HELP go_memstats_stack_inuse_bytes Number of bytes in use by the stack allocator.
# TYPE go_memstats_stack_inuse_bytes gauge
go_memstats_stack_inuse_bytes 491520
# HELP go_memstats_stack_sys_bytes Number of bytes obtained from system for stack allocator.
# TYPE go_memstats_stack_sys_bytes gauge
go_memstats_stack_sys_bytes 491520
# HELP go_memstats_sys_bytes Number of bytes obtained from system.
# TYPE go_memstats_sys_bytes gauge
go_memstats_sys_bytes 1.46322e+07
# HELP go_threads Number of OS threads created.
# TYPE go_threads gauge
go_threads 7
# HELP promhttp_metric_handler_requests_in_flight Current number of scrapes being served.
# TYPE promhttp_metric_handler_requests_in_flight gauge
promhttp_metric_handler_requests_in_flight 1
# HELP promhttp_metric_handler_requests_total Total number of scrapes by HTTP status code.
# TYPE promhttp_metric_handler_requests_total counter
promhttp_metric_handler_requests_total{code="200"} 1
promhttp_metric_handler_requests_total{code="500"} 0
promhttp_metric_handler_requests_total{code="503"} 0

我们已经给KisFlow提供了Function,Flow,Connector等的配置是通过kistype来区分的。
接下来我们来实现kistype等于global全局配置，在这个配置里，我们来设置是否启动Prometheus和Metrics统计的开关。
接下来给KisFlow加上全局配置文件的属性加载。

10.2 KisFlow全局配置

10.2.1 全局配置文件加载

全局配置的yaml的格式如下：

#kistype Global为kisflow的全局配置
kistype: global
#是否启动prometheus监控
prometheus_enable: true
#是否需要kisflow单独启动端口监听
prometheus_listen: true
#prometheus取点监听地址
prometheus_serve: 0.0.0.0:20004

10.2.2 结构体定义

接下来我们根据上述的配置协议，来定义KisFlow的策略配置结构体，并且提供一些响应的初始化方法。
我们在项目文档中创建kis_global_config.go文件，在这里我们将需要的Config定义实现。

kis-flow/config/kis_global_config.go

package config

type KisGlobalConfig struct {
    //kistype Global为kisflow的全局配置
    KisType string `yaml:"kistype"`
    //是否启动prometheus监控
    EnableProm bool `yaml:"prometheus_enable"`
    //是否需要kisflow单独启动端口监听
    PrometheusListen bool `yaml:"prometheus_listen"`
    //prometheus取点监听地址
    PrometheusServe string `yaml:"prometheus_serve"`
}

// GlobalConfig 默认全局配置，全部均为关闭
var GlobalConfig = new(KisGlobalConfig)

这里提供了一个全局的GlobalConfig对象，并且是公有变量，方便其他模块共享全局配置。

10.2.3 配置文件解析

接下来，针对全局配置做做导入配置的解析，在kif-flow/flie/config_import.go中，添加如下函数：

kis-flow/file/config_import.go

// kisTypeGlobalConfigure 解析Global配置文件，yaml格式
func kisTypeGlobalConfigure(confData []byte, fileName string, kisType interface{}) error {
    // 全局配置
    if ok := yaml.Unmarshal(confData, config.GlobalConfig); ok != nil {
        return errors.New(fmt.Sprintf("%s is wrong format kisType = %s", fileName, kisType))
    }

    // TODO 初始化Prometheus指标


    // TODO 启动Prometheus指标Metrics服务

    return nil
}

这里加载全局的yaml配置文件，加载之后，判断是否要启动初始化Prometheus指标监控，这个稍后我们再添加。

那么kisTypeGlobalConfigure()在哪里被调度，和其他的配置文件一样，在加载扫描本地配置文件的时候，被调度即可，如下：

kis-flow/file/config_import.go

// parseConfigWalkYaml 全盘解析配置文件，yaml格式, 讲配置信息解析到allConfig中
func parseConfigWalkYaml(loadPath string) (*allConfig, error) {
    // ... ...

    err := filepath.Walk(loadPath, func(filePath string, info os.FileInfo, err error) error {
        // ... ... 

        // 判断kisType是否存在
        if kisType, ok := confMap["kistype"]; !ok {
            return errors.New(fmt.Sprintf("yaml file %s has no file [kistype]!", filePath))
        } else {
            switch kisType {
            case common.KisIdTypeFlow:
                return kisTypeFlowConfigure(all, confData, filePath, kisType)

            case common.KisIdTypeFunction:
                return kisTypeFuncConfigure(all, confData, filePath, kisType)

            case common.KisIdTypeConnnector:
                return kisTypeConnConfigure(all, confData, filePath, kisType)

            // +++++++++++++++++++++++++++++++++
            case common.KisIdTypeGlobal:
                return kisTypeGlobalConfigure(confData, filePath, kisType)
            // +++++++++++++++++++++++++++++++++

            default:
                return errors.New(fmt.Sprintf("%s set wrong kistype %s", filePath, kisType))
            }
        }
    })

    if err != nil {
        return nil, err
    }

    return all, nil
}

在这里，我们增加kistype的Case：KisIdTypeGlobal来调用kisTypeGlobalConfigure()。

接下来，我们来创建Mertrics统计模块，本节先统计一个最简单的指标，KisFlow当前处理过的数据总量（以处理的源数据数量为准）

10.3 Metrics统计指标-DataTotal全量数据

10.3.1 KisMertics

首先创建一个KisMrtics模块，创建目录kis-flow/metrics/，并且创建文件kis_metrics.go文件：

kis-flow/metrics/kis_metrics.go

package metrics

import (
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promhttp"
    "kis-flow/common"
    "kis-flow/log"
    "net/http"
)

// kisMetrics kisFlow的Prometheus监控指标
type kisMetrics struct {
    //数据数量总量
    DataTotal prometheus.Counter
}

var Metrics *kisMetrics

// RunMetricsService 启动Prometheus监控服务
func RunMetricsService(serverAddr string) error {

    // 注册Prometheus 监控路由路径
    http.Handle(common.METRICS_ROUTE, promhttp.Handler())

    // 启动HttpServer
    err := http.ListenAndServe(serverAddr, nil) //多个进程不可监听同一个端口
    if err != nil {
        log.Logger().ErrorF("RunMetricsService err = %s\n", err)
    }

    return err
}

func InitMetrics() {
    Metrics = new(kisMetrics)

    // DataTotal初始化Counter
    Metrics.DataTotal = prometheus.NewCounter(prometheus.CounterOpts{
        Name: common.COUNTER_KISFLOW_DATA_TOTAL_NAME,
        Help: common.COUNTER_KISFLOW_DATA_TOTAL_HELP,
    })

    // 注册Metrics
    prometheus.MustRegister(Metrics.DataTotal)
}
  • kisMetrics struct: 为KisFlow的需要统计的指标，目前只有一个统计指标DataTotal，类型为prometheus.Counter（有关prometheus.Counter类型的说明请参考：有关prometheus.Counter指标的概念）
  • Metrics *kisMetrics：是KisFlow一个全局的指标统计对象，公有，方便其他模块获取。
  • RunMetricsService(serverAddr string)： 为启动prometheus服务监听，在之前的章节我们已经对这部分做了单元测试。
  • InitMetrics(): 为初始化全局对象和一些指标的初始化，最后需要调用prometheus.MustRegister将指标注册到prometheus中，这是prometheus统计编程的必要基本过程。

这里面有两个常量，分别表示指标显示名称和含义，我们定义在下面:

kis-flow/common/const.go

// metrics
const (
    METRICS_ROUTE string = "/metrics"

    COUNTER_KISFLOW_DATA_TOTAL_NAME string = "kisflow_data_total"
    COUNTER_KISFLOW_DATA_TOTAL_HELP string = "KisFlow全部Flow的数据总量"
)

10.3.2 DataTotal指标统计

作为KisFlow的处理全量数据，我们只需要在commitSrcData ()方法中统计就可以了，commitSrcData() 提交当前Flow的数据源数据, 表示首次提交当前Flow的原始数据源 ，表示数据首次进入KisFlow中，下面我们添加代码如下：

kis-flow/flow/kis_flow_data.go

func (flow *KisFlow) commitSrcData(ctx context.Context) error {

    // 制作批量数据batch
    dataCnt := len(flow.buffer)
    batch := make(common.KisRowArr, 0, dataCnt)

    for _, row := range flow.buffer {
        batch = append(batch, row)
    }

    // 清空之前所有数据
    flow.clearData(flow.data)

    // 首次提交，记录flow原始数据
    // 因为首次提交，所以PrevFunctionId为FirstVirtual 因为没有上一层Function
    flow.data[common.FunctionIdFirstVirtual] = batch

    // 清空缓冲Buf
    flow.buffer = flow.buffer[0:0]

    // +++++++++++++++++++++++++++++++
    // 首次提交数据源数据，进行统计数据总量
    if config.GlobalConfig.EnableProm == true {
        // 统计数据总量 Metrics.DataTota 指标累计加1
        metrics.Metrics.DataTotal.Add(float64(dataCnt))
    }
    // ++++++++++++++++++++++++++++++

    log.Logger().DebugFX(ctx, "====> After CommitSrcData, flow_name = %s, flow_id = %s\nAll Level Data =\n %+v\n", flow.Name, flow.Id, flow.data)

    return nil
}

先根据全局配置判断是否统计指标，如果为true则，通过下面代码来对metrics的全量数据进行统计:

        metrics.Metrics.DataTotal.Add(float64(dataCnt))

dataCnt为累计增加的数量。

10.3.3 Metrics启动

在导入配置之后，我们需要启动metrics服务，调度如下：

kis-flow/file/config_import.go

// kisTypeGlobalConfigure 解析Global配置文件，yaml格式
func kisTypeGlobalConfigure(confData []byte, fileName string, kisType interface{}) error {
    // 全局配置
    if ok := yaml.Unmarshal(confData, config.GlobalConfig); ok != nil {
        return errors.New(fmt.Sprintf("%s is wrong format kisType = %s", fileName, kisType))
    }

    // ++++++++++++++++++++
    // 启动Metrics服务
    metrics.RunMetrics()

    return nil
}

其中RunMetrics()是实现如下：

kis-flow/metrics/kis_metrics.go

// RunMetrics 启动Prometheus指标服务
func RunMetrics() {
    // 初始化Prometheus指标
    InitMetrics()

    if config.GlobalConfig.EnableProm == true && config.GlobalConfig.PrometheusListen == true {
        // 启动Prometheus指标Metrics服务
        go RunMetricsService(config.GlobalConfig.PrometheusServe)
    }
}

这样，在导入全局配置后，看是否开启统计，如果统计，我们就会开一个协程来启动PrometheusServe，监听的ip和端口会在配置文件里进行配置。

接下来我们先对DataTotal指标做一个单元测试，来进行验证。

10.4 KisMetrics单元测试

10.4.1 全局配置文件创建

创建一个全局配置文件kis-flow.yml 在kis-flow/test/load_conf/下，内容如下：

kis-flow/test/load_conf/kis-flow.yml

#kistype Global为kisflow的全局配置
kistype: global
#是否启动prometheus监控
prometheus_enable: true
#是否需要kisflow单独启动端口监听
prometheus_listen: true
#prometheus取点监听地址
prometheus_serve: 0.0.0.0:20004

10.4.2 新建单元测试

接下来创建测试用例代码，在kis-flow/test/下，创建kis_metrics_test.go文件，如下：

kis-flow/test/kis_metrics_test.go

package test

import (
    "context"
    "kis-flow/common"
    "kis-flow/file"
    "kis-flow/kis"
    "kis-flow/test/caas"
    "kis-flow/test/faas"
    "testing"
    "time"
)

func TestMetricsDataTotal(t *testing.T) {
    ctx := context.Background()

    // 0. 注册Function 回调业务
    kis.Pool().FaaS("funcName1", faas.FuncDemo1Handler)
    kis.Pool().FaaS("funcName2", faas.FuncDemo2Handler)
    kis.Pool().FaaS("funcName3", faas.FuncDemo3Handler)

    // 0. 注册ConnectorInit 和 Connector 回调业务
    kis.Pool().CaaSInit("ConnName1", caas.InitConnDemo1)
    kis.Pool().CaaS("ConnName1", "funcName2", common.S, caas.CaasDemoHanler1)

    // 1. 加载配置文件并构建Flow
    if err := file.ConfigImportYaml("/Users/tal/gopath/src/kis-flow/test/load_conf/"); err != nil {
        panic(err)
    }

    // 2. 获取Flow
    flow1 := kis.Pool().GetFlow("flowName1")
    
    n := 0

    for n < 10 {
        // 3. 提交原始数据
        _ = flow1.CommitRow("This is Data1 from Test")

        // 4. 执行flow1
        if err := flow1.Run(ctx); err != nil {
            panic(err)
        }

        time.Sleep(1 * time.Second)
        n++
    }
    
    select {}
}

这个Case和我们一般启动KisFlow一样，只不过，这里面会出现一个for循环，每割1秒回启动一次流式计算，并且提交一条数据，一共循环10次，之后我们可以通过prometheus的监听服务来查看数据的总量。
最后加select{}的目的是为了防止主协程退出，导致prometheus的监听子协程连同退出。

执行单元测试，cd到kis-flow/test/下，执行：

go test -test.v -test.paniconexit0 -test.run TestMetricsDataTotal

会得到很多日志输出，我们等待10s，之后再开启一个终端，输入如下指令：

 $ curl http://0.0.0.0:20004/metrics

得到如下结果：

# ... ...
# HELP kisflow_data_total KisFlow全部Flow的数据总量
# TYPE kisflow_data_total counter
kisflow_data_total 10
# ... ...

其中我们会发现，kisflow_data_total指标已经出现，结果是10，说明我们的metrics指标已经可以统计了，那么接下来我们就可以再基于这个逻辑，新增一些其他KisFlow需要关心的比较复杂的指标。

接下来我们来统计其他的一些关键指标，包括：Flow处理数据总量、Flow被调度次数、Function被调度次数、Flow执行时间、Function执行时间等。

10.5 Metrics统计指标-其他统计指标

10.5.1 指标：Flow处理数据总量

（1）指标定义

首先定义指标类型，如下：

kis-flow/metrics/kis_metrics.go

// kisMetrics kisFlow的Prometheus监控指标
type kisMetrics struct {
    // 数据数量总量
    DataTotal prometheus.Counter
    // 各Flow处理数据总量
    FlowDataTotal *prometheus.GaugeVec
}

FlowDataTotal 采用 prometheus.GaugeVec类型。主要是为了区分是哪个Flow产生的数据。

（2）指标初始化及注册

kis-flow/metrics/kis_metrics.go

func InitMetrics() {
    Metrics = new(kisMetrics)

    // DataTotal初始化Counter
    Metrics.DataTotal = prometheus.NewCounter(prometheus.CounterOpts{
        Name: common.COUNTER_KISFLOW_DATA_TOTAL_NAME,
        Help: common.COUNTER_KISFLOW_DATA_TOTAL_HELP,
    })

    // +++++++++++
    // FlowDataTotal初始化GaugeVec
    Metrics.FlowDataTotal = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: common.GANGE_FLOW_DATA_TOTAL_NAME,
            Help: common.GANGE_FLOW_DATA_TOTAL_HELP,
        },
        // 标签名称
        []string{common.LABEL_FLOW_NAME},
    )

    // 注册Metrics
    prometheus.MustRegister(Metrics.DataTotal)
    prometheus.MustRegister(Metrics.FlowDataTotal) // +++
}

相关常量定义：

kis-flow/common/const.go


// metrics
const (
    METRICS_ROUTE string = "/metrics"

    // ++++++++
    LABEL_FLOW_NAME     string = "flow_name"
    LABEL_FLOW_ID       string = "flow_id"
    LABEL_FUNCTION_NAME string = "func_name"
    LABEL_FUNCTION_MODE string = "func_mode"

    COUNTER_KISFLOW_DATA_TOTAL_NAME string = "kisflow_data_total"
    COUNTER_KISFLOW_DATA_TOTAL_HELP string = "KisFlow全部Flow的数据总量"

    // +++++++ 
    GANGE_FLOW_DATA_TOTAL_NAME string = "flow_data_total"
    GANGE_FLOW_DATA_TOTAL_HELP string = "KisFlow各个FlowID数据流的数据数量总量"
)

（3）统计指标埋点

作为flow的数据总量，我们应该在数据每次提交源数据的时候进行统计即可。

kis-flow/flow/kis_flow_data.go

func (flow *KisFlow) commitSrcData(ctx context.Context) error {

    // 制作批量数据batch
    dataCnt := len(flow.buffer)
    batch := make(common.KisRowArr, 0, dataCnt)

    for _, row := range flow.buffer {
        batch = append(batch, row)
    }

    // 清空之前所有数据
    flow.clearData(flow.data)

    // 首次提交，记录flow原始数据
    // 因为首次提交，所以PrevFunctionId为FirstVirtual 因为没有上一层Function
    flow.data[common.FunctionIdFirstVirtual] = batch

    // 清空缓冲Buf
    flow.buffer = flow.buffer[0:0]

    // 首次提交数据源数据，进行统计数据总量
    if config.GlobalConfig.EnableProm == true {
        // 统计数据总量 Metrics.DataTota 指标累计加1
        metrics.Metrics.DataTotal.Add(float64(dataCnt))

        // ++++++++
        //统计当前Flow数量指标
        metrics.Metrics.FlowDataTotal.WithLabelValues(flow.Name).Add(float64(dataCnt))
    }

    log.Logger().DebugFX(ctx, "====> After CommitSrcData, flow_name = %s, flow_id = %s\nAll Level Data =\n %+v\n", flow.Name, flow.Id, flow.data)

    return nil
}

所以埋点的位置这之前的统计从数据量埋点的位置一样，只不过在累加数据的时候，加上flow.Name标签。

10.5.2 指标：Flow被调度次数

（1）指标定义

首先定义指标类型，如下：

kis-flow/metrics/kis_metrics.go

// kisMetrics kisFlow的Prometheus监控指标
type kisMetrics struct {
    // 数据数量总量
    DataTotal prometheus.Counter
    // 各Flow处理数据总量
    FlowDataTotal *prometheus.GaugeVec
    // Flow被调度次数
    FlowScheduleCntsToTal *prometheus.GaugeVec //++++
}

FlowScheduleCntsToTal 采用 prometheus.GaugeVec类型。主要是为了区分是哪个Flow产生的数据。

（2）指标初始化及注册

kis-flow/metrics/kis_metrics.go

func InitMetrics() {
    Metrics = new(kisMetrics)

    // DataTotal初始化Counter
    Metrics.DataTotal = prometheus.NewCounter(prometheus.CounterOpts{
        Name: common.COUNTER_KISFLOW_DATA_TOTAL_NAME,
        Help: common.COUNTER_KISFLOW_DATA_TOTAL_HELP,
    })

    // FlowDataTotal初始化GaugeVec
    Metrics.FlowDataTotal = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: common.GANGE_FLOW_DATA_TOTAL_NAME,
            Help: common.GANGE_FLOW_DATA_TOTAL_HELP,
        },
        // 标签名称
        []string{common.LABEL_FLOW_NAME},
    )

    // +++++++++++++
    // FlowScheduleCntsToTal初始化GaugeVec
    Metrics.FlowScheduleCntsToTal = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: common.GANGE_FLOW_SCHE_CNTS_NAME,
            Help: common.GANGE_FLOW_SCHE_CNTS_HELP,
        },
        //标签名称
        []string{common.LABEL_FLOW_NAME},
    )

    // 注册Metrics
    prometheus.MustRegister(Metrics.DataTotal)
    prometheus.MustRegister(Metrics.FlowDataTotal) 
    // +++++
    prometheus.MustRegister(Metrics.FlowScheduleCntsToTal)
}

相关常量定义：

kis-flow/common/const.go


// metrics
const (
    METRICS_ROUTE string = "/metrics"

    LABEL_FLOW_NAME     string = "flow_name"
    LABEL_FLOW_ID       string = "flow_id"
    LABEL_FUNCTION_NAME string = "func_name"
    LABEL_FUNCTION_MODE string = "func_mode"

    COUNTER_KISFLOW_DATA_TOTAL_NAME string = "kisflow_data_total"
    COUNTER_KISFLOW_DATA_TOTAL_HELP string = "KisFlow全部Flow的数据总量"

    GANGE_FLOW_DATA_TOTAL_NAME string = "flow_data_total"
    GANGE_FLOW_DATA_TOTAL_HELP string = "KisFlow各个FlowID数据流的数据数量总量"

    // +++++++
    GANGE_FLOW_SCHE_CNTS_NAME string = "flow_schedule_cnts"
    GANGE_FLOW_SCHE_CNTS_HELP string = "KisFlow各个FlowID被调度的次数"
)

（3）统计指标埋点

如果统计每个Flow的调度次数，我们应该在启动Flow的主入口flow.Run()进行统计，如下：

kis-flow/flow/kis_flow.go

// Run 启动KisFlow的流式计算, 从起始Function开始执行流
func (flow *KisFlow) Run(ctx context.Context) error {

    var fn kis.Function

    fn = flow.FlowHead
    flow.abort = false

    if flow.Conf.Status == int(common.FlowDisable) {
        // flow被配置关闭
        return nil
    }

    // 因为此时还没有执行任何Function, 所以PrevFunctionId为FirstVirtual 因为没有上一层Function
    flow.PrevFunctionId = common.FunctionIdFirstVirtual

    // 提交数据流原始数据
    if err := flow.commitSrcData(ctx); err != nil {
        return err
    }

    // +++++++++++ Metrics
    if config.GlobalConfig.EnableProm == true {
        // 统计Flow的调度次数
        metrics.Metrics.FlowScheduleCntsToTal.WithLabelValues(flow.Name).Inc()
    }
    // ++++++++++++++++++++

    // 流式链式调用
    for fn != nil && flow.abort == false {

        // flow记录当前执行到的Function 标记
        fid := fn.GetId()
        flow.ThisFunction = fn
        flow.ThisFunctionId = fid

        // 得到当前Function要处理与的源数据
        if inputData, err := flow.getCurData(); err != nil {
            log.Logger().ErrorFX(ctx, "flow.Run(): getCurData err = %s\n", err.Error())
            return err
        } else {
            flow.inPut = inputData
        }

        if err := fn.Call(ctx, flow); err != nil {
            // Error
            return err
        } else {
            // Success
            fn, err = flow.dealAction(ctx, fn)
            if err != nil {
                return err
            }

        }
    }

    return nil
}

所以埋点的位置这之前的统计从数据量埋点的位置一样，只不过在累加数据的时候，加上flow.Name标签。

10.5.3 指标：Function被调度次数

（1）指标定义

首先定义指标类型，如下：

kis-flow/metrics/kis_metrics.go

// kisMetrics kisFlow的Prometheus监控指标
type kisMetrics struct {
    // 数据数量总量
    DataTotal prometheus.Counter
    // 各Flow处理数据总量
    FlowDataTotal *prometheus.GaugeVec
    // Flow被调度次数
    FlowScheduleCntsToTal *prometheus.GaugeVec 
    // Function被调度次数
    FuncScheduleCntsTotal *prometheus.GaugeVec //++++
}

FuncScheduleCntsTotal 采用 prometheus.GaugeVec类型。主要是为了区分是哪个Function产生的数据。

（2）指标初始化及注册

kis-flow/metrics/kis_metrics.go

func InitMetrics() {
    Metrics = new(kisMetrics)

    // DataTotal初始化Counter
    Metrics.DataTotal = prometheus.NewCounter(prometheus.CounterOpts{
        Name: common.COUNTER_KISFLOW_DATA_TOTAL_NAME,
        Help: common.COUNTER_KISFLOW_DATA_TOTAL_HELP,
    })

    // FlowDataTotal初始化GaugeVec
    Metrics.FlowDataTotal = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: common.GANGE_FLOW_DATA_TOTAL_NAME,
            Help: common.GANGE_FLOW_DATA_TOTAL_HELP,
        },
        // 标签名称
        []string{common.LABEL_FLOW_NAME},
    )

    // FlowScheduleCntsToTal初始化GaugeVec
    Metrics.FlowScheduleCntsToTal = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: common.GANGE_FLOW_SCHE_CNTS_NAME,
            Help: common.GANGE_FLOW_SCHE_CNTS_HELP,
        },
        //标签名称
        []string{common.LABEL_FLOW_NAME},
    )

    // ++++++++++
    // FuncScheduleCntsTotal初始化GaugeVec
    Metrics.FuncScheduleCntsTotal = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: common.GANGE_FUNC_SCHE_CNTS_NAME,
            Help: common.GANGE_FUNC_SCHE_CNTS_HELP,
        },
        //标签名称
        []string{common.LABEL_FUNCTION_NAME, common.LABEL_FUNCTION_MODE},
    )

    // 注册Metrics
    prometheus.MustRegister(Metrics.DataTotal)
    prometheus.MustRegister(Metrics.FlowDataTotal) 
    prometheus.MustRegister(Metrics.FlowScheduleCntsToTal)
    // +++++++
    prometheus.MustRegister(Metrics.FuncScheduleCntsTotal)
}

相关常量定义：

kis-flow/common/const.go


// metrics
const (
    METRICS_ROUTE string = "/metrics"

    LABEL_FLOW_NAME     string = "flow_name"
    LABEL_FLOW_ID       string = "flow_id"
    LABEL_FUNCTION_NAME string = "func_name"
    LABEL_FUNCTION_MODE string = "func_mode"

    COUNTER_KISFLOW_DATA_TOTAL_NAME string = "kisflow_data_total"
    COUNTER_KISFLOW_DATA_TOTAL_HELP string = "KisFlow全部Flow的数据总量"

    GANGE_FLOW_DATA_TOTAL_NAME string = "flow_data_total"
    GANGE_FLOW_DATA_TOTAL_HELP string = "KisFlow各个FlowID数据流的数据数量总量"

    GANGE_FLOW_SCHE_CNTS_NAME string = "flow_schedule_cnts"
    GANGE_FLOW_SCHE_CNTS_HELP string = "KisFlow各个FlowID被调度的次数"

    // +++++++++ 
    GANGE_FUNC_SCHE_CNTS_NAME string = "func_schedule_cnts"
    GANGE_FUNC_SCHE_CNTS_HELP string = "KisFlow各个Function被调度的次数"
)

（3）统计指标埋点

如果统计每个Function的调度次数，我们应该在启动Flow的主入口flow.Run()进行统计，如下：

kis-flow/flow/kis_flow.go

// Run 启动KisFlow的流式计算, 从起始Function开始执行流
func (flow *KisFlow) Run(ctx context.Context) error {

    var fn kis.Function

    fn = flow.FlowHead
    flow.abort = false

    if flow.Conf.Status == int(common.FlowDisable) {
        // flow被配置关闭
        return nil
    }

    // 因为此时还没有执行任何Function, 所以PrevFunctionId为FirstVirtual 因为没有上一层Function
    flow.PrevFunctionId = common.FunctionIdFirstVirtual

    // 提交数据流原始数据
    if err := flow.commitSrcData(ctx); err != nil {
        return err
    }

    if config.GlobalConfig.EnableProm == true {
        // 统计Flow的调度次数
        metrics.Metrics.FlowScheduleCntsToTal.WithLabelValues(flow.Name).Inc()
    }

    // 流式链式调用
    for fn != nil && flow.abort == false {

        // flow记录当前执行到的Function 标记
        fid := fn.GetId()
        flow.ThisFunction = fn
        flow.ThisFunctionId = fid

        // ++++++++++++
        fName := fn.GetConfig().FName
        fMode := fn.GetConfig().FMode

        // +++++++++++++++++++++++++++
        if config.GlobalConfig.EnableProm == true {
            // 统计Function调度次数
            metrics.Metrics.FuncScheduleCntsTotal.WithLabelValues(fName, fMode).Inc()
        }
        // ++++++++++++++++++++++++++++

        // 得到当前Function要处理与的源数据
        if inputData, err := flow.getCurData(); err != nil {
            log.Logger().ErrorFX(ctx, "flow.Run(): getCurData err = %s\n", err.Error())
            return err
        } else {
            flow.inPut = inputData
        }

        if err := fn.Call(ctx, flow); err != nil {
            // Error
            return err
        } else {
            // Success
            fn, err = flow.dealAction(ctx, fn)
            if err != nil {
                return err
            }
        }
    }

    return nil
}

在埋点的位置为循环调度function的时候，每次在执行Funciton的Call()方法之前进行调度数据统计，并且按照fName和fMode进行分组。

10.5.4 指标：Function执行时间

（1）指标定义

定义指标类型，如下：

kis-flow/metrics/kis_metrics.go

// kisMetrics kisFlow的Prometheus监控指标
type kisMetrics struct {
    // 数据数量总量
    DataTotal prometheus.Counter
    // 各Flow处理数据总量
    FlowDataTotal *prometheus.GaugeVec
    // Flow被调度次数
    FlowScheduleCntsToTal *prometheus.GaugeVec 
    // Function被调度次数
    FuncScheduleCntsTotal *prometheus.GaugeVec 
    // Function执行时间
    FunctionDuration *prometheus.HistogramVec //++++
}

FunctionDuration 采用 prometheus.HistogramVec类型。这是一个不同区间值的分布统计，不同的时间区间会落到相应的Bucket中。

（2）指标初始化及注册

kis-flow/metrics/kis_metrics.go

func InitMetrics() {
    Metrics = new(kisMetrics)

    // DataTotal初始化Counter
    Metrics.DataTotal = prometheus.NewCounter(prometheus.CounterOpts{
        Name: common.COUNTER_KISFLOW_DATA_TOTAL_NAME,
        Help: common.COUNTER_KISFLOW_DATA_TOTAL_HELP,
    })

    // FlowDataTotal初始化GaugeVec
    Metrics.FlowDataTotal = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: common.GANGE_FLOW_DATA_TOTAL_NAME,
            Help: common.GANGE_FLOW_DATA_TOTAL_HELP,
        },
        // 标签名称
        []string{common.LABEL_FLOW_NAME},
    )

    // FlowScheduleCntsToTal初始化GaugeVec
    Metrics.FlowScheduleCntsToTal = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: common.GANGE_FLOW_SCHE_CNTS_NAME,
            Help: common.GANGE_FLOW_SCHE_CNTS_HELP,
        },
        //标签名称
        []string{common.LABEL_FLOW_NAME},
    )

    // FuncScheduleCntsTotal初始化GaugeVec
    Metrics.FuncScheduleCntsTotal = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: common.GANGE_FUNC_SCHE_CNTS_NAME,
            Help: common.GANGE_FUNC_SCHE_CNTS_HELP,
        },
        //标签名称
        []string{common.LABEL_FUNCTION_NAME, common.LABEL_FUNCTION_MODE},
    )

    // ++++++++++++++++++++++++++
    // FunctionDuration初始化HistogramVec
    Metrics.FunctionDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
        Name:    common.HISTOGRAM_FUNCTION_DURATION_NAME,
        Help:    common.HISTOGRAM_FUNCTION_DURATION_HELP,
        Buckets: []float64{0.005, 0.01, 0.03, 0.08, 0.1, 0.5, 1.0, 5.0, 10, 100, 1000, 5000, 30000}, //单位ms,最大半分钟
    },
        []string{common.LABEL_FUNCTION_NAME, common.LABEL_FUNCTION_MODE},
    )

    // 注册Metrics
    prometheus.MustRegister(Metrics.DataTotal)
    prometheus.MustRegister(Metrics.FlowDataTotal) 
    prometheus.MustRegister(Metrics.FlowScheduleCntsToTal)
    prometheus.MustRegister(Metrics.FuncScheduleCntsTotal)
    // +++++++
    prometheus.MustRegister(Metrics.FunctionDuration)
}

相关常量定义：

kis-flow/common/const.go


// metrics
const (
    METRICS_ROUTE string = "/metrics"

    LABEL_FLOW_NAME     string = "flow_name"
    LABEL_FLOW_ID       string = "flow_id"
    LABEL_FUNCTION_NAME string = "func_name"
    LABEL_FUNCTION_MODE string = "func_mode"

    COUNTER_KISFLOW_DATA_TOTAL_NAME string = "kisflow_data_total"
    COUNTER_KISFLOW_DATA_TOTAL_HELP string = "KisFlow全部Flow的数据总量"

    GANGE_FLOW_DATA_TOTAL_NAME string = "flow_data_total"
    GANGE_FLOW_DATA_TOTAL_HELP string = "KisFlow各个FlowID数据流的数据数量总量"

    GANGE_FLOW_SCHE_CNTS_NAME string = "flow_schedule_cnts"
    GANGE_FLOW_SCHE_CNTS_HELP string = "KisFlow各个FlowID被调度的次数"

    GANGE_FUNC_SCHE_CNTS_NAME string = "func_schedule_cnts"
    GANGE_FUNC_SCHE_CNTS_HELP string = "KisFlow各个Function被调度的次数"
    // ++++++++
    HISTOGRAM_FUNCTION_DURATION_NAME string = "func_run_duration"
    HISTOGRAM_FUNCTION_DURATION_HELP string = "Function执行耗时"
)

（3）统计指标埋点

如果统计每个Function的调度实行时长，我们应该在启动Flow的主入口flow.Run()进行统计，如下：

kis-flow/flow/kis_flow.go

// Run 启动KisFlow的流式计算, 从起始Function开始执行流
func (flow *KisFlow) Run(ctx context.Context) error {

    var fn kis.Function

    fn = flow.FlowHead
    flow.abort = false

    if flow.Conf.Status == int(common.FlowDisable) {
        // flow被配置关闭
        return nil
    }
    
    // ++++++++++ Metrics +++++++++
    var funcStart time.Time

    // 因为此时还没有执行任何Function, 所以PrevFunctionId为FirstVirtual 因为没有上一层Function
    flow.PrevFunctionId = common.FunctionIdFirstVirtual

    // 提交数据流原始数据
    if err := flow.commitSrcData(ctx); err != nil {
        return err
    }

    if config.GlobalConfig.EnableProm == true {
        // 统计Flow的调度次数
        metrics.Metrics.FlowScheduleCntsToTal.WithLabelValues(flow.Name).Inc()
    }

    // 流式链式调用
    for fn != nil && flow.abort == false {

        // flow记录当前执行到的Function 标记
        fid := fn.GetId()
        flow.ThisFunction = fn
        flow.ThisFunctionId = fid

        fName := fn.GetConfig().FName
        fMode := fn.GetConfig().FMode

        if config.GlobalConfig.EnableProm == true {
            // 统计Function调度次数
            metrics.Metrics.FuncScheduleCntsTotal.WithLabelValues(fName, fMode).Inc()
            
            // +++++++++++++++
            // 统计Function 耗时 记录开始时间
            funcStart = time.Now()
        }

        // 得到当前Function要处理与的源数据
        if inputData, err := flow.getCurData(); err != nil {
            log.Logger().ErrorFX(ctx, "flow.Run(): getCurData err = %s\n", err.Error())
            return err
        } else {
            flow.inPut = inputData
        }

        if err := fn.Call(ctx, flow); err != nil {
            // Error
            return err
        } else {
            // Success
            fn, err = flow.dealAction(ctx, fn)
            if err != nil {
                return err
            }

            // +++++++++++++++
            // 统计Function 耗时
            if config.GlobalConfig.EnableProm == true {
                // Function消耗时间
                duration := time.Since(funcStart)

                // 统计当前Function统计指标,做时间统计
                metrics.Metrics.FunctionDuration.With(
                    prometheus.Labels{
                        common.LABEL_FUNCTION_NAME: fName,
                        common.LABEL_FUNCTION_MODE: fMode}).Observe(duration.Seconds() * 1000)
            }
            // +++++++++++++++

        }
    }

    return nil
}

在埋点的位置每次在执行Funciton的Call()方法之前进行起始时间记录，然后在执行Function之后，算出执行时间，左后进行统计，按照相对应的时间区间，放入到响应的HistogramVec中的bucket中。

10.5.5 指标：Flow执行时间

（1）指标定义

定义指标类型，如下：

kis-flow/metrics/kis_metrics.go

// kisMetrics kisFlow的Prometheus监控指标
type kisMetrics struct {
    // 数据数量总量
    DataTotal prometheus.Counter
    // 各Flow处理数据总量
    FlowDataTotal *prometheus.GaugeVec
    // Flow被调度次数
    FlowScheduleCntsToTal *prometheus.GaugeVec 
    // Function被调度次数
    FuncScheduleCntsTotal *prometheus.GaugeVec 
    // Function执行时间
    FunctionDuration *prometheus.HistogramVec
    // Flow执行时间
    FlowDuration *prometheus.HistogramVec // ++++
}

FlowDuration 采用 prometheus.HistogramVec类型。这是一个不同区间值的分布统计，不同的时间区间会落到相应的Bucket中。

（2）指标初始化及注册

kis-flow/metrics/kis_metrics.go

func InitMetrics() {
    Metrics = new(kisMetrics)

    // DataTotal初始化Counter
    Metrics.DataTotal = prometheus.NewCounter(prometheus.CounterOpts{
        Name: common.COUNTER_KISFLOW_DATA_TOTAL_NAME,
        Help: common.COUNTER_KISFLOW_DATA_TOTAL_HELP,
    })

    // FlowDataTotal初始化GaugeVec
    Metrics.FlowDataTotal = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: common.GANGE_FLOW_DATA_TOTAL_NAME,
            Help: common.GANGE_FLOW_DATA_TOTAL_HELP,
        },
        // 标签名称
        []string{common.LABEL_FLOW_NAME},
    )

    // FlowScheduleCntsToTal初始化GaugeVec
    Metrics.FlowScheduleCntsToTal = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: common.GANGE_FLOW_SCHE_CNTS_NAME,
            Help: common.GANGE_FLOW_SCHE_CNTS_HELP,
        },
        //标签名称
        []string{common.LABEL_FLOW_NAME},
    )

    // FuncScheduleCntsTotal初始化GaugeVec
    Metrics.FuncScheduleCntsTotal = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: common.GANGE_FUNC_SCHE_CNTS_NAME,
            Help: common.GANGE_FUNC_SCHE_CNTS_HELP,
        },
        //标签名称
        []string{common.LABEL_FUNCTION_NAME, common.LABEL_FUNCTION_MODE},
    )

    // FunctionDuration初始化HistogramVec
    Metrics.FunctionDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
        Name:    common.HISTOGRAM_FUNCTION_DURATION_NAME,
        Help:    common.HISTOGRAM_FUNCTION_DURATION_HELP,
        Buckets: []float64{0.005, 0.01, 0.03, 0.08, 0.1, 0.5, 1.0, 5.0, 10, 100, 1000, 5000, 30000}, //单位ms,最大半分钟
    },
        []string{common.LABEL_FUNCTION_NAME, common.LABEL_FUNCTION_MODE},
    )
    

    // +++++++++++++
    // FlowDuration初始化HistogramVec
    Metrics.FlowDuration = prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    common.HISTOGRAM_FLOW_DURATION_NAME,
            Help:    common.HISTOGRAM_FLOW_DURATION_HELP,
            Buckets: []float64{0.005, 0.01, 0.03, 0.08, 0.1, 0.5, 1.0, 5.0, 10, 100, 1000, 5000, 30000, 60000}, //单位ms,最大1分钟
        },
        []string{common.LABEL_FLOW_NAME},
    )

    // 注册Metrics
    prometheus.MustRegister(Metrics.DataTotal)
    prometheus.MustRegister(Metrics.FlowDataTotal) 
    prometheus.MustRegister(Metrics.FlowScheduleCntsToTal)
    prometheus.MustRegister(Metrics.FuncScheduleCntsTotal)
    prometheus.MustRegister(Metrics.FunctionDuration)
    prometheus.MustRegister(Metrics.FlowDuration) // +++++
}

相关常量定义：

kis-flow/common/const.go


// metrics
const (
    METRICS_ROUTE string = "/metrics"

    LABEL_FLOW_NAME     string = "flow_name"
    LABEL_FLOW_ID       string = "flow_id"
    LABEL_FUNCTION_NAME string = "func_name"
    LABEL_FUNCTION_MODE string = "func_mode"

    COUNTER_KISFLOW_DATA_TOTAL_NAME string = "kisflow_data_total"
    COUNTER_KISFLOW_DATA_TOTAL_HELP string = "KisFlow全部Flow的数据总量"

    GANGE_FLOW_DATA_TOTAL_NAME string = "flow_data_total"
    GANGE_FLOW_DATA_TOTAL_HELP string = "KisFlow各个FlowID数据流的数据数量总量"

    GANGE_FLOW_SCHE_CNTS_NAME string = "flow_schedule_cnts"
    GANGE_FLOW_SCHE_CNTS_HELP string = "KisFlow各个FlowID被调度的次数"

    GANGE_FUNC_SCHE_CNTS_NAME string = "func_schedule_cnts"
    GANGE_FUNC_SCHE_CNTS_HELP string = "KisFlow各个Function被调度的次数"
    HISTOGRAM_FUNCTION_DURATION_NAME string = "func_run_duration"
    HISTOGRAM_FUNCTION_DURATION_HELP string = "Function执行耗时"

    // ++++++++
    HISTOGRAM_FLOW_DURATION_NAME string = "flow_run_duration"
    HISTOGRAM_FLOW_DURATION_HELP string = "Flow执行耗时"
)

（3）统计指标埋点

如果统计每个Flow的调度实行时长，我们应该在启动Flow的主入口flow.Run()进行统计，如下：

kis-flow/flow/kis_flow.go

// Run 启动KisFlow的流式计算, 从起始Function开始执行流
func (flow *KisFlow) Run(ctx context.Context) error {

    var fn kis.Function

    fn = flow.FlowHead
    flow.abort = false

    if flow.Conf.Status == int(common.FlowDisable) {
        // flow被配置关闭
        return nil
    }
    
    var funcStart time.Time
    // ++++++++++ Metrics +++++++++
    var flowStart time.Time


    // 因为此时还没有执行任何Function, 所以PrevFunctionId为FirstVirtual 因为没有上一层Function
    flow.PrevFunctionId = common.FunctionIdFirstVirtual

    // 提交数据流原始数据
    if err := flow.commitSrcData(ctx); err != nil {
        return err
    }

    if config.GlobalConfig.EnableProm == true {
        // 统计Flow的调度次数
        metrics.Metrics.FlowScheduleCntsToTal.WithLabelValues(flow.Name).Inc()
        
        // +++++++
        // 统计Flow的执行消耗时长
        flowStart = time.Now()
    }

    // 流式链式调用
    for fn != nil && flow.abort == false {

        // flow记录当前执行到的Function 标记
        fid := fn.GetId()
        flow.ThisFunction = fn
        flow.ThisFunctionId = fid

        fName := fn.GetConfig().FName
        fMode := fn.GetConfig().FMode

        if config.GlobalConfig.EnableProm == true {
            // 统计Function调度次数
            metrics.Metrics.FuncScheduleCntsTotal.WithLabelValues(fName, fMode).Inc()
            
            // 统计Function 耗时 记录开始时间
            funcStart = time.Now()
        }

        // 得到当前Function要处理与的源数据
        if inputData, err := flow.getCurData(); err != nil {
            log.Logger().ErrorFX(ctx, "flow.Run(): getCurData err = %s\n", err.Error())
            return err
        } else {
            flow.inPut = inputData
        }

        if err := fn.Call(ctx, flow); err != nil {
            // Error
            return err
        } else {
            // Success
            fn, err = flow.dealAction(ctx, fn)
            if err != nil {
                return err
            }

            // 统计Function 耗时
            if config.GlobalConfig.EnableProm == true {
                // Function消耗时间
                duration := time.Since(funcStart)

                // 统计当前Function统计指标,做时间统计
                metrics.Metrics.FunctionDuration.With(
                    prometheus.Labels{
                        common.LABEL_FUNCTION_NAME: fName,
                        common.LABEL_FUNCTION_MODE: fMode}).Observe(duration.Seconds() * 1000)
            }
        }
    }

    // +++++++++++++++++++++++++
    // Metrics
    if config.GlobalConfig.EnableProm == true {
        // 统计Flow执行耗时
        duration := time.Since(flowStart)
        metrics.Metrics.FlowDuration.WithLabelValues(flow.Name).Observe(duration.Seconds() * 1000)
    }

    return nil
}

在埋点的位置在flow进入Run()方法之后，进行起始时间记录，然后在Run()最后进行duration统计，统计办法与Function的统计时长类似。

接下来我们先对DataTotal指标做一个单元测试，来进行验证。

10.6 KieMetrics单元测试(其他Metrics指标)

10.6.1 新建单元测试

单元测试用例我们复用之前的TestMetricsDataTotal()方法即可，如下：

kis-flow/test/kis_metrics_test.go

package test

import (
    "context"
    "kis-flow/common"
    "kis-flow/file"
    "kis-flow/kis"
    "kis-flow/test/caas"
    "kis-flow/test/faas"
    "testing"
    "time"
)

func TestMetricsDataTotal(t *testing.T) {
    ctx := context.Background()

    // 0. 注册Function 回调业务
    kis.Pool().FaaS("funcName1", faas.FuncDemo1Handler)
    kis.Pool().FaaS("funcName2", faas.FuncDemo2Handler)
    kis.Pool().FaaS("funcName3", faas.FuncDemo3Handler)

    // 0. 注册ConnectorInit 和 Connector 回调业务
    kis.Pool().CaaSInit("ConnName1", caas.InitConnDemo1)
    kis.Pool().CaaS("ConnName1", "funcName2", common.S, caas.CaasDemoHanler1)

    // 1. 加载配置文件并构建Flow
    if err := file.ConfigImportYaml("/Users/tal/gopath/src/kis-flow/test/load_conf/"); err != nil {
        panic(err)
    }

    // 2. 获取Flow
    flow1 := kis.Pool().GetFlow("flowName1")
    
    n := 0

    for n < 10 {
        // 3. 提交原始数据
        _ = flow1.CommitRow("This is Data1 from Test")

        // 4. 执行flow1
        if err := flow1.Run(ctx); err != nil {
            panic(err)
        }

        time.Sleep(1 * time.Second)
        n++
    }
    
    select {}
}

执行单元测试，cd到kis-flow/test/下，执行：

go test -test.v -test.paniconexit0 -test.run TestMetricsDataTotal

会得到很多日志输出，我们等待10s，之后再开启一个终端，输入如下指令：

 $ curl http://0.0.0.0:20004/metrics

得到如下结果：

# HELP flow_data_total KisFlow各个FlowID数据流的数据数量总量
# TYPE flow_data_total gauge
flow_data_total{flow_name="flowName1"} 10
# HELP flow_run_duration Flow执行耗时
# TYPE flow_run_duration histogram
flow_run_duration_bucket{flow_name="flowName1",le="0.005"} 0
flow_run_duration_bucket{flow_name="flowName1",le="0.01"} 0
flow_run_duration_bucket{flow_name="flowName1",le="0.03"} 0
flow_run_duration_bucket{flow_name="flowName1",le="0.08"} 0
flow_run_duration_bucket{flow_name="flowName1",le="0.1"} 0
flow_run_duration_bucket{flow_name="flowName1",le="0.5"} 0
flow_run_duration_bucket{flow_name="flowName1",le="1"} 0
flow_run_duration_bucket{flow_name="flowName1",le="5"} 9
flow_run_duration_bucket{flow_name="flowName1",le="10"} 10
flow_run_duration_bucket{flow_name="flowName1",le="100"} 10
flow_run_duration_bucket{flow_name="flowName1",le="1000"} 10
flow_run_duration_bucket{flow_name="flowName1",le="5000"} 10
flow_run_duration_bucket{flow_name="flowName1",le="30000"} 10
flow_run_duration_bucket{flow_name="flowName1",le="60000"} 10
flow_run_duration_bucket{flow_name="flowName1",le="+Inf"} 10
flow_run_duration_sum{flow_name="flowName1"} 29.135023
flow_run_duration_count{flow_name="flowName1"} 10
# HELP flow_schedule_cnts KisFlow各个FlowID被调度的次数
# TYPE flow_schedule_cnts gauge
flow_schedule_cnts{flow_name="flowName1"} 10
# HELP func_run_duration Function执行耗时
# TYPE func_run_duration histogram
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="0.005"} 0
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="0.01"} 0
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="0.03"} 0
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="0.08"} 0
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="0.1"} 0
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="0.5"} 0
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="1"} 0
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="5"} 9
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="10"} 10
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="100"} 10
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="1000"} 10
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="5000"} 10
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="30000"} 10
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="+Inf"} 10
func_run_duration_sum{func_mode="Calculate",func_name="funcName3"} 20.925857
func_run_duration_count{func_mode="Calculate",func_name="funcName3"} 10
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="0.005"} 0
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="0.01"} 0
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="0.03"} 0
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="0.08"} 0
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="0.1"} 0
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="0.5"} 0
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="1"} 1
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="5"} 10
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="10"} 10
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="100"} 10
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="1000"} 10
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="5000"} 10
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="30000"} 10
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="+Inf"} 10
func_run_duration_sum{func_mode="Save",func_name="funcName2"} 27.026124
func_run_duration_count{func_mode="Save",func_name="funcName2"} 10
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="0.005"} 0
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="0.01"} 0
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="0.03"} 0
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="0.08"} 0
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="0.1"} 0
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="0.5"} 5
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="1"} 10
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="5"} 10
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="10"} 10
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="100"} 10
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="1000"} 10
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="5000"} 10
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="30000"} 10
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="+Inf"} 10
func_run_duration_sum{func_mode="Verify",func_name="funcName1"} 4.811095
func_run_duration_count{func_mode="Verify",func_name="funcName1"} 10
# HELP func_schedule_cnts KisFlow各个Function被调度的次数
# TYPE func_schedule_cnts gauge
func_schedule_cnts{func_mode="Calculate",func_name="funcName3"} 10
func_schedule_cnts{func_mode="Save",func_name="funcName2"} 10
func_schedule_cnts{func_mode="Verify",func_name="funcName1"} 10

# HELP kisflow_data_total KisFlow全部Flow的数据总量
# TYPE kisflow_data_total counter
kisflow_data_total 10

# ... ...

其中我们会发现，我们之前的统计指标均已经出现了：

  • kisflow_data_total:为总数据量，目前是10条数据，因为我们一共Commit提交了10条源数据。
  • flow_data_total:为flow的数据总量，目前我们只启动了flowName1，该数据被通缉到了标签flowName1中。
flow_data_total{flow_name="flowName1"} 10
  • flow_schedule_cnts:为flow的调度次数，因为我们一共执行了10次 flow.Run()方法，所以这个调度次数是10。
# HELP flow_schedule_cnts KisFlow各个FlowID被调度的次数
# TYPE flow_schedule_cnts gauge
flow_schedule_cnts{flow_name="flowName1"} 10
  • func_schedule_cnts： 为各个Function的被调度次数，这里因为每个Flow会关联3个Function，所以每个Function的调度次数应该和Flow1的调度次数相同，都是10。
# HELP func_schedule_cnts KisFlow各个Function被调度的次数
# TYPE func_schedule_cnts gauge
func_schedule_cnts{func_mode="Calculate",func_name="funcName3"} 10
func_schedule_cnts{func_mode="Save",func_name="funcName2"} 10
func_schedule_cnts{func_mode="Verify",func_name="funcName1"} 10
  • func_run_duration_bucket: 为Function的执行耗时分布统计。（有关HISTOGRAM 的统计方式比较复杂，这里就不赘述了，有兴趣的开发者可以去查阅一些相关资料。） 
# HELP func_run_duration Function执行耗时
# TYPE func_run_duration histogram
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="0.005"} 0
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="0.01"} 0
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="0.03"} 0
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="0.08"} 0
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="0.1"} 0
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="0.5"} 0
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="1"} 0
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="5"} 9
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="10"} 10
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="100"} 10
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="1000"} 10
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="5000"} 10
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="30000"} 10
func_run_duration_bucket{func_mode="Calculate",func_name="funcName3",le="+Inf"} 10
func_run_duration_sum{func_mode="Calculate",func_name="funcName3"} 20.925857
func_run_duration_count{func_mode="Calculate",func_name="funcName3"} 10
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="0.005"} 0
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="0.01"} 0
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="0.03"} 0
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="0.08"} 0
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="0.1"} 0
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="0.5"} 0
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="1"} 1
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="5"} 10
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="10"} 10
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="100"} 10
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="1000"} 10
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="5000"} 10
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="30000"} 10
func_run_duration_bucket{func_mode="Save",func_name="funcName2",le="+Inf"} 10
func_run_duration_sum{func_mode="Save",func_name="funcName2"} 27.026124
func_run_duration_count{func_mode="Save",func_name="funcName2"} 10
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="0.005"} 0
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="0.01"} 0
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="0.03"} 0
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="0.08"} 0
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="0.1"} 0
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="0.5"} 5
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="1"} 10
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="5"} 10
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="10"} 10
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="100"} 10
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="1000"} 10
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="5000"} 10
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="30000"} 10
func_run_duration_bucket{func_mode="Verify",func_name="funcName1",le="+Inf"} 10
func_run_duration_sum{func_mode="Verify",func_name="funcName1"} 4.811095
func_run_duration_count{func_mode="Verify",func_name="funcName1"} 10
  • flow_run_duration_bucket: 为为Flow的执行耗时分布统计。（有关HISTOGRAM 的统计方式比较复杂，这里就不赘述了，有兴趣的开发者可以去查阅一些相关资料。） 
# HELP flow_run_duration Flow执行耗时
# TYPE flow_run_duration histogram
flow_run_duration_bucket{flow_name="flowName1",le="0.005"} 0
flow_run_duration_bucket{flow_name="flowName1",le="0.01"} 0
flow_run_duration_bucket{flow_name="flowName1",le="0.03"} 0
flow_run_duration_bucket{flow_name="flowName1",le="0.08"} 0
flow_run_duration_bucket{flow_name="flowName1",le="0.1"} 0
flow_run_duration_bucket{flow_name="flowName1",le="0.5"} 0
flow_run_duration_bucket{flow_name="flowName1",le="1"} 0
flow_run_duration_bucket{flow_name="flowName1",le="5"} 9
flow_run_duration_bucket{flow_name="flowName1",le="10"} 10
flow_run_duration_bucket{flow_name="flowName1",le="100"} 10
flow_run_duration_bucket{flow_name="flowName1",le="1000"} 10
flow_run_duration_bucket{flow_name="flowName1",le="5000"} 10
flow_run_duration_bucket{flow_name="flowName1",le="30000"} 10
flow_run_duration_bucket{flow_name="flowName1",le="60000"} 10
flow_run_duration_bucket{flow_name="flowName1",le="+Inf"} 10
flow_run_duration_sum{flow_name="flowName1"} 29.135023
flow_run_duration_count{flow_name="flowName1"} 10

10.7 有关KisFlow的Metrics的Grafana看板展示

既然有了Prometheus的指标统计，我们可以给KisFlow的流式计算程序结合Grafana进行看板展示。
由于各个开发者的项目的统计指标和看板要求不一定相同，这里本文就不提供具体的Grafana看板的配置文件了，下面提供一个KisFlow的项目看板，作为演示参考，如下：

10.8 【V0.9】 源代码

https://github.com/aceld/kis-flow/releases/tag/v0.9

作者:刘丹冰Aceld github: https://github.com/aceld
KisFlow开源项目地址：https://github.com/aceld/kis-flow

Golang框架实战-KisFlow流式计算框架专栏

Golang框架实战-KisFlow流式计算框架(1)-概述
Golang框架实战-KisFlow流式计算框架(2)-项目构建/基础模块-(上)
Golang框架实战-KisFlow流式计算框架(3)-项目构建/基础模块-(下)
Golang框架实战-KisFlow流式计算框架(4)-数据流
Golang框架实战-KisFlow流式计算框架(5)-Function调度
Golang框架实战-KisFlow流式计算框架(6)-Connector
Golang框架实战-KisFlow流式计算框架(7)-配置导入与导出
Golang框架实战-KisFlow流式计算框架(8)-KisFlow Action
Golang框架实战-KisFlow流式计算框架(10)-Flow多副本

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 211,948评论 6赞 492
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 90,371评论 3赞 385
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 157,490评论 0赞 348
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 56,521评论 1赞 284
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 65,627评论 6赞 386
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 49,842评论 1赞 290
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,997评论 3赞 408
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 37,741评论 0赞 268
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,203评论 1赞 303
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,534评论 2赞 327
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,673评论 1赞 341
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,339评论 4赞 330
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,955评论 3赞 313
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,770评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,000评论 1赞 266
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 46,394评论 2赞 360
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 43,562评论 2赞 349

推荐阅读更多精彩内容