DEV Community

Aceld
Aceld

Posted on • Edited on

(Part 10)Golang Framework Hands-on - Prometheus Metrics Statistics

#go

Github: https://github.com/aceld/kis-flow
Document: https://github.com/aceld/kis-flow/wiki

Part1-OverView
Part2.1-Project Construction / Basic Modules
Part2.2-Project Construction / Basic Modules
Part3-Data Stream
Part4-Function Scheduling
Part5-Connector
Part6-Configuration Import and Export
Part7-KisFlow Action
Part8-Cache/Params Data Caching and Data Parameters
Part9-Multiple Copies of Flow
Part10-Prometheus Metrics Statistics
Part11-Adaptive Registration of FaaS Parameter Types Based on Reflection

Case1-Quick Start
Case2-Flow Parallel Operation
Case3-Application of KisFlow in Multi-Goroutines
Case4-KisFlow in Message Queue (MQ) Applications

Before diving into this chapter, let's introduce how to start the Prometheus Metrics service. For those unfamiliar with Prometheus, it's advisable to look up additional information. In simple terms, Prometheus is a system monitoring and metrics tool.

As KisFlow is a stream computing framework, metrics such as function scheduling time, total data volume, and algorithm speed are crucial for developers and project teams. These metrics can be recorded using Prometheus Metrics through KisFlow.

Next, we will configure the framework globally, allowing developers to enable Prometheus metrics collection if needed.

10.1 Prometheus Metrics Service

10.1.1 Prometheus Client SDK

First, add the necessary dependency in the kis-flow/go.mod file:

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 )
Enter fullscreen mode Exit fullscreen mode

We use the official Prometheus Golang client SDK. More details can be found in the official README documentation:

Next, let's write a simple Prometheus service that allows external access to KisFlow service metrics. Create a new directory kis-flow/metrics/ for the KisFlow metrics code.

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 starts the Prometheus monitoring service
func RunMetricsService(serverAddr string) error {

// Register the Prometheus monitoring route path http.Handle(common.METRICS_ROUTE, promhttp.Handler()) // Start the HTTP server err := http.ListenAndServe(serverAddr, nil) // Multiple processes cannot listen on the same port if err != nil { log.Logger().ErrorF("RunMetricsService err = %s\n", err) } return err
Enter fullscreen mode Exit fullscreen mode

}

 Define `METRICS_ROUTE` as the monitoring service HTTP route path in `kis-flow/common/const.go`: > kis-flow/common/const.go ```go // ... ... // metrics const ( METRICS_ROUTE string = "/metrics" ) // ... ...
Enter fullscreen mode Exit fullscreen mode

Let's briefly explain the above code. RunMetricsService() starts the Prometheus monitoring HTTP service. The purpose of this service is to provide metrics for the current KisFlow process through HTTP requests. While we haven't collected specific metrics yet, Prometheus will provide default metrics such as the current Go version, GC garbage collection time, memory allocation, etc.

  • serverAddr parameter: This is the address for the Prometheus monitoring service, usually a local address with a port number like "0.0.0.0:20004".
http.Handle(common.METRICS_ROUTE, promhttp.Handler())
Enter fullscreen mode Exit fullscreen mode

This line of code sets "0.0.0.0:20004/metrics" as the metrics entry point.

After writing the above code, remember to pull the relevant dependency package from Prometheus Golang Client SDK, https://github.com/prometheus/client_golang .

$ go mod tidy
Enter fullscreen mode Exit fullscreen mode

After pulling, the current go.mod dependencies will look something like this (with version differences):

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 Unit Testing for Prometheus Server Service Startup Next, let's perform a simple test to verify if the service can start. Create a file named `prometheus_server_test.go` in the `kis-flow/test/` directory: > kis-flow/test/prometheus_server_test.go ```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) } }
Enter fullscreen mode Exit fullscreen mode

Here, the monitoring address is "0.0.0.0:20004". Next, start this unit test case by opening terminal A and navigating to the kis-flow/test/ directory:

$ cd kis-flow/test/ $ go test -test.v -test.paniconexit0 -test.run TestPrometheusServer === RUN TestPrometheusServer
Enter fullscreen mode Exit fullscreen mode

Then, open another terminal B and enter the following command to simulate an HTTP client request:

$ curl http://0.0.0.0:20004/metrics
Enter fullscreen mode Exit fullscreen mode

After that, we should see the monitoring metrics result in terminal B as follows:

# 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
Enter fullscreen mode Exit fullscreen mode

We have already provided configurations for Function, Flow, and Connector in KisFlow, distinguished by kistype. Next, we will implement a global configuration with kistype set to global. In this configuration, we will add settings to enable or disable Prometheus and Metrics collection.

Let's proceed to add global configuration properties to KisFlow.

10.2 KisFlow Global Configuration

10.2.1 Loading Global Configuration Files

The global configuration in YAML format is as follows:

# kistype Global for the global configuration of KisFlow kistype: global # Whether to enable Prometheus monitoring prometheus_enable: true # Whether KisFlow needs to start a separate port listener prometheus_listen: true # The address for Prometheus to listen for metrics prometheus_serve: 0.0.0.0:20004
Enter fullscreen mode Exit fullscreen mode

10.2.2 Struct Definition

Next, based on the configuration protocol above, we'll define the strategy configuration struct for KisFlow and provide some initialization methods. Create a file named kis_global_config.go in the project documentation. Here, we'll define the necessary configuration.

kis-flow/config/kis_global_config.go

package config

type KisGlobalConfig struct {
// kistype Global for the global configuration of KisFlow
KisType string yaml:"kistype"
// Whether to enable Prometheus monitoring
EnableProm bool yaml:"prometheus_enable"
// Whether KisFlow needs to start a separate port listener
PrometheusListen bool yaml:"prometheus_listen"
// The address for Prometheus to listen for metrics
PrometheusServe string yaml:"prometheus_serve"
}

// GlobalConfig is the default global configuration, all are turned off
var GlobalConfig = new(KisGlobalConfig)

 Here, we provide a global `GlobalConfig` object, which is a public variable, making it convenient for other modules to share the global configuration. ### 10.2.3 Configuration File Parsing Next, we'll parse the global configuration and import it. Add the following function in `kis-flow/file/config_import.go`: > kis-flow/file/config_import.go ```go // kisTypeGlobalConfigure parses the Global configuration file in YAML format func kisTypeGlobalConfigure(confData []byte, fileName string, kisType interface{}) error { // Global configuration if err := yaml.Unmarshal(confData, config.GlobalConfig); err != nil { return fmt.Errorf("%s has wrong format kisType = %s", fileName, kisType) } // TODO Initialize Prometheus metrics // TODO Start Prometheus metrics service return nil }
Enter fullscreen mode Exit fullscreen mode

This function loads the global YAML configuration file. After loading, it determines whether to initialize Prometheus metrics monitoring, which we will add later.

Where is kisTypeGlobalConfigure() called? It is invoked during the loading and scanning of local configuration files, similar to other configuration files:

kis-flow/file/config_import.go

// parseConfigWalkYaml parses all configuration files in YAML format and loads the configuration information into allConfig func parseConfigWalkYaml(loadPath string) (*allConfig, error) { // ... ... 
err := filepath.Walk(loadPath, func(filePath string, info os.FileInfo, err error) error { // ... ... // Check if kistype exists if kisType, ok := confMap["kistype"]; !ok { return fmt.Errorf("yaml file %s has no field [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.KisIdTypeConnector: return kisTypeConnConfigure(all, confData, filePath, kisType) // +++++++++++++++++++++++++++++++++ case common.KisIdTypeGlobal: return kisTypeGlobalConfigure(confData, filePath, kisType) // +++++++++++++++++++++++++++++++++ default: return fmt.Errorf("%s sets wrong kistype %s", filePath, kisType) } } }) if err != nil { return nil, err } return all, nil
Enter fullscreen mode Exit fullscreen mode

}

 Here, we add a case for kistype: `KisIdTypeGlobal` to call `kisTypeGlobalConfigure()`. Next, we will create the Metrics module. In this section, we will start by tracking a simple metric: the total amount of data processed by KisFlow (based on the number of source data processed). ## 10.3 Metrics - DataTotal Metric ### 10.3.1 KisMetrics First, create a KisMetrics module by creating the directory `kis-flow/metrics/` and the file `kis_metrics.go`: > kis-flow/metrics/kis_metrics.go ```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 defines the Prometheus metrics for KisFlow type kisMetrics struct { // Total data count DataTotal prometheus.Counter } var Metrics *kisMetrics // RunMetricsService starts the Prometheus monitoring service func RunMetricsService(serverAddr string) error { // Register Prometheus monitoring route http.Handle(common.METRICS_ROUTE, promhttp.Handler()) // Start HTTP server err := http.ListenAndServe(serverAddr, nil) // Multiple processes cannot listen on the same port if err != nil { log.Logger().ErrorF("RunMetricsService err = %s\n", err) } return err } // InitMetrics initializes the metrics func InitMetrics() { Metrics = new(kisMetrics) // Initialize the DataTotal counter Metrics.DataTotal = prometheus.NewCounter(prometheus.CounterOpts{ Name: common.COUNTER_KISFLOW_DATA_TOTAL_NAME, Help: common.COUNTER_KISFLOW_DATA_TOTAL_HELP, }) // Register the metrics prometheus.MustRegister(Metrics.DataTotal) }
Enter fullscreen mode Exit fullscreen mode

  • kisMetrics struct: This struct defines the metrics that KisFlow needs to track. Currently, it only includes one metric, DataTotal, which is of type prometheus.Counter (for more information on the prometheus.Counter type, please refer to the Prometheus documentation).

  • Metrics *kisMetrics: This is a global metrics tracking object for KisFlow, making it publicly accessible for other modules.

  • RunMetricsService(serverAddr string): This function starts the Prometheus service listener, which was already unit tested in previous chapters.

  • InitMetrics(): This function initializes the global object and sets up the metrics. It calls prometheus.MustRegister to register the metrics with Prometheus, which is a necessary step in Prometheus metrics programming.

There are two constants representing the metric display name and description. These are defined as follows:

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 = "Total data processed by all Flows in KisFlow"
Enter fullscreen mode Exit fullscreen mode

)

 ### 10.3.2 DataTotal Metric Tracking To track the total data processed by KisFlow, we need to add the metrics tracking code in the `commitSrcData()` method. This method submits the current Flow's source data, indicating the first time the original source data is submitted to the current Flow. The updated code is as follows: > kis-flow/flow/kis_flow_data.go ```go func (flow *KisFlow) commitSrcData(ctx context.Context) error { // Create a batch of data dataCnt := len(flow.buffer) batch := make(common.KisRowArr, 0, dataCnt) for _, row := range flow.buffer { batch = append(batch, row) } // Clear previous data flow.clearData(flow.data) // Record the original data for the first submission // Since this is the first submission, PrevFunctionId is FirstVirtual because there is no previous Function flow.data[common.FunctionIdFirstVirtual] = batch // Clear the buffer flow.buffer = flow.buffer[0:0] // +++++++++++++++++++++++++++++++ // Track the total data count on the first submission if config.GlobalConfig.EnableProm == true { // Increment the DataTotal metric by the data count 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 }
Enter fullscreen mode Exit fullscreen mode

First, it checks the global configuration to determine if metrics tracking is enabled. If true, it increments the total data count metric with the following code:

metrics.Metrics.DataTotal.Add(float64(dataCnt))
Enter fullscreen mode Exit fullscreen mode

Here, dataCnt is the number of data items being added to the total count.

10.3.3 Starting the Metrics Service

After importing the configuration, we need to start the metrics service. The configuration file is updated as follows:

kis-flow/file/config_import.go

// kisTypeGlobalConfigure parses the Global configuration file in YAML format func kisTypeGlobalConfigure(confData []byte, fileName string, kisType interface{}) error { // Global configuration if ok := yaml.Unmarshal(confData, config.GlobalConfig); ok != nil { return errors.New(fmt.Sprintf("%s is wrong format kisType = %s", fileName, kisType)) } 
// ++++++++++++++++++++ // Start the Metrics service metrics.RunMetrics() return nil
Enter fullscreen mode Exit fullscreen mode

}

 The `RunMetrics()` function is implemented as follows: > kis-flow/metrics/kis_metrics.go ```go // RunMetrics starts the Prometheus metrics service func RunMetrics() { // Initialize Prometheus metrics InitMetrics() if config.GlobalConfig.EnableProm == true && config.GlobalConfig.PrometheusListen == true { // Start the Prometheus metrics service go RunMetricsService(config.GlobalConfig.PrometheusServe) } }
Enter fullscreen mode Exit fullscreen mode

With this setup, after importing the global configuration, it checks whether metrics tracking is enabled. If it is, a new goroutine is started to launch the Prometheus server, listening on the IP and port specified in the configuration file.

Next, we will create a unit test for the DataTotal metric to validate our implementation.

10.4 KisMetrics Unit Testing

10.4.1 Create Global Configuration File

Create a global configuration file kis-flow.yml under kis-flow/test/load_conf/ with the following content:

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

# kistype Global for kisflow global configuration kistype: global # Enable prometheus monitoring prometheus_enable: true # Enable separate kisflow port listening prometheus_listen: true # Prometheus endpoint listening address prometheus_serve: 0.0.0.0:20004

10.4.2 Create Unit Test

Next, create the test case code in kis-flow/test/, and create the kis_metrics_test.go file as follows:

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. Register Function callback business logic kis.Pool().FaaS("funcName1", faas.FuncDemo1Handler) kis.Pool().FaaS("funcName2", faas.FuncDemo2Handler) kis.Pool().FaaS("funcName3", faas.FuncDemo3Handler) // 0. Register ConnectorInit and Connector callback business logic kis.Pool().CaaSInit("ConnName1", caas.InitConnDemo1) kis.Pool().CaaS("ConnName1", "funcName2", common.S, caas.CaasDemoHanler1) // 1. Load configuration file and build Flow if err := file.ConfigImportYaml("/Users/tal/gopath/src/kis-flow/test/load_conf/"); err != nil { panic(err) } // 2. Get Flow flow1 := kis.Pool().GetFlow("flowName1") n := 0 for n < 10 { // 3. Submit raw data _ = flow1.CommitRow("This is Data1 from Test") // 4. Execute flow1 if err := flow1.Run(ctx); err != nil { panic(err) } time.Sleep(1 * time.Second) n++ } select {}
Enter fullscreen mode Exit fullscreen mode

}

 This case works similarly to starting KisFlow usually, except that it includes a loop that starts a stream computation every second and submits a piece of data, looping 10 times. Afterward, we can check the total data amount through the Prometheus monitoring service. The select{} statement prevents the main goroutine from exiting, ensuring that the Prometheus monitoring goroutine continues running. Run the unit test by navigating to `kis-flow/test/` and executing: ```bash go test -test.v -test.paniconexit0 -test.run TestMetricsDataTotal
Enter fullscreen mode Exit fullscreen mode

You will see a lot of log output. Wait for 10 seconds, then open another terminal and enter the following command:

$ curl http://0.0.0.0:20004/metrics
Enter fullscreen mode Exit fullscreen mode

The result will be:

# ... ... # HELP kisflow_data_total KisFlow total data amount of all Flows # TYPE kisflow_data_total counter kisflow_data_total 10 # ... ...
Enter fullscreen mode Exit fullscreen mode

Here, you'll find that the kisflow_data_total metric appears with a result of 10, indicating that our metrics are correctly tracked. Next, we can add more complex metrics for KisFlow.

10.5 Additional Metrics

10.5.1 Metric: Flow Data Total

(1) Define Metric

First, define the metric type as follows:

kis-flow/metrics/kis_metrics.go

// kisMetrics Prometheus monitoring metrics for kisFlow type kisMetrics struct { // Total data amount DataTotal prometheus.Counter // Data total per Flow FlowDataTotal *prometheus.GaugeVec }

FlowDataTotal uses the prometheus.GaugeVec type to distinguish which Flow generates the data.

(2) Initialize and Register Metrics

kis-flow/metrics/kis_metrics.go

func InitMetrics() { Metrics = new(kisMetrics) 
// Initialize DataTotal Counter Metrics.DataTotal = prometheus.NewCounter(prometheus.CounterOpts{ Name: common.COUNTER_KISFLOW_DATA_TOTAL_NAME, Help: common.COUNTER_KISFLOW_DATA_TOTAL_HELP, }) // +++++++++++ // Initialize FlowDataTotal GaugeVec Metrics.FlowDataTotal = prometheus.NewGaugeVec( prometheus.GaugeOpts{ Name: common.GANGE_FLOW_DATA_TOTAL_NAME, Help: common.GANGE_FLOW_DATA_TOTAL_HELP, }, // Label names []string{common.LABEL_FLOW_NAME}, ) // Register Metrics prometheus.MustRegister(Metrics.DataTotal) prometheus.MustRegister(Metrics.FlowDataTotal) // +++
Enter fullscreen mode Exit fullscreen mode

}

 Related constant definitions: > kis-flow/common/const.go ```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 total data amount of all Flows" // +++++++ GANGE_FLOW_DATA_TOTAL_NAME string = "flow_data_total" GANGE_FLOW_DATA_TOTAL_HELP string = "KisFlow data total amount per Flow" )
Enter fullscreen mode Exit fullscreen mode

(3) Add Metric Tracking

We should track the total data amount when submitting raw data.

kis-flow/flow/kis_flow_data.go

func (flow *KisFlow) commitSrcData(ctx context.Context) error { 
// Create batch data dataCnt := len(flow.buffer) batch := make(common.KisRowArr, 0, dataCnt) for _, row := range flow.buffer { batch = append(batch, row) } // Clear previous data flow.clearData(flow.data) // First submission, record flow raw data flow.data[common.FunctionIdFirstVirtual] = batch // Clear buffer flow.buffer = flow.buffer[0:0] // First submission, track data total if config.GlobalConfig.EnableProm == true { // Track data total Metrics.DataTotal metrics.Metrics.DataTotal.Add(float64(dataCnt)) // ++++++++ // Track current Flow data total 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
Enter fullscreen mode Exit fullscreen mode

}

 So the tracking point is in the same position as before, but we add a `flow.Name` label when accumulating the data. ### 10.5.2 Metric: Flow Scheduling Count #### (1) Metric Definition First, define the metric type as follows: > kis-flow/metrics/kis_metrics.go ```go // kisMetrics represents the Prometheus monitoring metrics for kisFlow type kisMetrics struct { // Total data count DataTotal prometheus.Counter // Total data processed by each Flow FlowDataTotal *prometheus.GaugeVec // Total Flow scheduling count FlowScheduleCntsTotal *prometheus.GaugeVec //++++ }
Enter fullscreen mode Exit fullscreen mode

FlowScheduleCntsTotal uses prometheus.GaugeVec type to distinguish which Flow produced the data.

(2) Metric Initialization and Registration

kis-flow/metrics/kis_metrics.go

func InitMetrics() { Metrics = new(kisMetrics) // Initialize DataTotal as a Counter Metrics.DataTotal = prometheus.NewCounter(prometheus.CounterOpts{ Name: common.COUNTER_KISFLOW_DATA_TOTAL_NAME, Help: common.COUNTER_KISFLOW_DATA_TOTAL_HELP, }) // Initialize FlowDataTotal as a GaugeVec Metrics.FlowDataTotal = prometheus.NewGaugeVec( prometheus.GaugeOpts{ Name: common.GANGE_FLOW_DATA_TOTAL_NAME, Help: common.GANGE_FLOW_DATA_TOTAL_HELP, }, // Label name []string{common.LABEL_FLOW_NAME}, ) // +++++++++++++ // Initialize FlowScheduleCntsTotal as a GaugeVec Metrics.FlowScheduleCntsTotal = prometheus.NewGaugeVec( prometheus.GaugeOpts{ Name: common.GANGE_FLOW_SCHE_CNTS_NAME, Help: common.GANGE_FLOW_SCHE_CNTS_HELP, }, // Label name []string{common.LABEL_FLOW_NAME}, ) // Register metrics prometheus.MustRegister(Metrics.DataTotal) prometheus.MustRegister(Metrics.FlowDataTotal) // +++++ prometheus.MustRegister(Metrics.FlowScheduleCntsTotal) }
Enter fullscreen mode Exit fullscreen mode

Define the relevant constants:

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 total data count" GANGE_FLOW_DATA_TOTAL_NAME string = "flow_data_total" GANGE_FLOW_DATA_TOTAL_HELP string = "Total data count for each FlowID in KisFlow" // +++++++ GANGE_FLOW_SCHE_CNTS_NAME string = "flow_schedule_cnts" GANGE_FLOW_SCHE_CNTS_HELP string = "Scheduling count for each FlowID in KisFlow"
Enter fullscreen mode Exit fullscreen mode

)

 #### (3) Metric Data Collection To collect the scheduling count for each Flow, we should collect data in the main entry point `flow.Run()`, as follows: > kis-flow/flow/kis_flow.go ```go // Run starts the stream computation of KisFlow and executes the flow from the initial 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 is disabled in configuration return nil } // Since no Function has been executed yet, PrevFunctionId is set to FirstVirtual because there is no previous Function flow.PrevFunctionId = common.FunctionIdFirstVirtual // Commit the original stream data if err := flow.commitSrcData(ctx); err != nil { return err } // +++++++++++ Metrics if config.GlobalConfig.EnableProm == true { // Collect scheduling count for Flow metrics.Metrics.FlowScheduleCntsTotal.WithLabelValues(flow.Name).Inc() } // ++++++++++++++++++++ // Chain-style stream invocation for fn != nil && flow.abort == false { // Record the current Function being executed in the Flow fid := fn.GetId() flow.ThisFunction = fn flow.ThisFunctionId = fid // Obtain the source data for the current Function to process 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 }
Enter fullscreen mode Exit fullscreen mode

Here, the metric collection occurs before calling the fn.Call() method, and we increment the counter for each Flow execution, grouping by flow.Name.

10.5.3 Metric: Function Scheduling Count

(1) Metric Definition

First, define the metric type as follows:

kis-flow/metrics/kis_metrics.go

// kisMetrics represents the Prometheus monitoring metrics for kisFlow type kisMetrics struct { // Total data count DataTotal prometheus.Counter // Total data processed by each Flow FlowDataTotal *prometheus.GaugeVec // Total Flow scheduling count FlowScheduleCntsTotal *prometheus.GaugeVec // Total Function scheduling count FuncScheduleCntsTotal *prometheus.GaugeVec //++++ }

FuncScheduleCntsTotal uses prometheus.GaugeVec type to distinguish which Function produced the data.

(2) Metric Initialization and Registration

kis-flow/metrics/kis_metrics.go

func InitMetrics() { Metrics = new(kisMetrics) 
// Initialize DataTotal as a Counter Metrics.DataTotal = prometheus.NewCounter(prometheus.CounterOpts{ Name: common.COUNTER_KISFLOW_DATA_TOTAL_NAME, Help: common.COUNTER_KISFLOW_DATA_TOTAL_HELP, }) // Initialize FlowDataTotal as a GaugeVec Metrics.FlowDataTotal = prometheus.NewGaugeVec( prometheus.GaugeOpts{ Name: common.GANGE_FLOW_DATA_TOTAL_NAME, Help: common.GANGE_FLOW_DATA_TOTAL_HELP, }, // Label name []string{common.LABEL_FLOW_NAME}, ) // Initialize FlowScheduleCntsTotal as a GaugeVec Metrics.FlowScheduleCntsTotal = prometheus.NewGaugeVec( prometheus.GaugeOpts{ Name: common.GANGE_FLOW_SCHE_CNTS_NAME, Help: common.GANGE_FLOW_SCHE_CNTS_HELP, }, // Label name []string{common.LABEL_FLOW_NAME}, ) // ++++++++++ // Initialize FuncScheduleCntsTotal as a GaugeVec Metrics.FuncScheduleCntsTotal = prometheus.NewGaugeVec( prometheus.GaugeOpts{ Name: common.GANGE_FUNC_SCHE_CNTS_NAME, Help: common.GANGE_FUNC_SCHE_CNTS_HELP, }, // Label names []string{common.LABEL_FUNCTION_NAME, common.LABEL_FUNCTION_MODE}, ) // Register metrics prometheus.MustRegister(Metrics.DataTotal) prometheus.MustRegister(Metrics.FlowDataTotal) prometheus.MustRegister(Metrics.FlowScheduleCntsTotal) // +++++++ prometheus.MustRegister(Metrics.FuncScheduleCntsTotal)
Enter fullscreen mode Exit fullscreen mode

}

 Define the relevant constants: > kis-flow/common/const.go ```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 total data count" GANGE_FLOW_DATA_TOTAL_NAME string = "flow_data_total" GANGE_FLOW_DATA_TOTAL_HELP string = "Total data count for each FlowID in KisFlow" GANGE_FLOW_SCHE_CNTS_NAME string = "flow_schedule_cnts" GANGE_FLOW_SCHE_CNTS_HELP string = "Scheduling count for each FlowID in KisFlow" // +++++++++ GANGE_FUNC_SCHE_CNTS_NAME string = "func_schedule_cnts" GANGE_FUNC_SCHE_CNTS_HELP string = "Scheduling count for each Function in KisFlow" )
Enter fullscreen mode Exit fullscreen mode

(3) Metric Data Collection

To collect the scheduling count for each Function, we should collect data in the main entry point flow.Run(), as follows:

kis-flow/flow/kis_flow.go

// Run starts the stream computation of KisFlow and executes the flow from the initial 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 is disabled in configuration return nil } // Since no Function has been executed yet, PrevFunctionId is set to FirstVirtual because there is no previous Function flow.PrevFunctionId = common.FunctionIdFirstVirtual // Commit the original stream data if err := flow.commitSrcData(ctx); err != nil { return err } if config.GlobalConfig.EnableProm == true { // Collect scheduling count for Flow metrics.Metrics.FlowScheduleCntsTotal.WithLabelValues(flow.Name).Inc() } // Chain-style stream invocation for fn != nil && flow.abort == false { // Record the current Function being executed in the Flow fid := fn.GetId() flow.ThisFunction = fn flow.ThisFunctionId = fid // ++++++++++++ fName := fn.GetConfig().FName fMode := fn.GetConfig().FMode // +++++++++++++++++++++++++++ if config.GlobalConfig.EnableProm == true { // Collect scheduling count for Function metrics.Metrics.FuncScheduleCntsTotal.WithLabelValues(fName, fMode).Inc() } // ++++++++++++++++++++++++++++ // Obtain the source data for the current Function to process 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
Enter fullscreen mode Exit fullscreen mode

}

 Here, the metric collection occurs before calling the `fn.Call()` method. Each time a Function is scheduled, the counter is incremented, grouped by `fName` and `fMode`. ### 10.5.4 Metric: Function Execution Time #### (1) Metric Definition Define the metric type as follows: > kis-flow/metrics/kis_metrics.go ```go // kisMetrics defines the Prometheus metrics for kisFlow type kisMetrics struct { // Total data count DataTotal prometheus.Counter // Total data processed by each Flow FlowDataTotal *prometheus.GaugeVec // Flow schedule count FlowScheduleCntsTotal *prometheus.GaugeVec // Function schedule count FuncScheduleCntsTotal *prometheus.GaugeVec // Function execution time FunctionDuration *prometheus.HistogramVec //++++ }
Enter fullscreen mode Exit fullscreen mode

FunctionDuration uses the prometheus.HistogramVec type. This type provides distribution statistics across different time intervals, with various buckets representing different time ranges.

(2) Metric Initialization and Registration

kis-flow/metrics/kis_metrics.go

func InitMetrics() { Metrics = new(kisMetrics) 
// Initialize DataTotal Counter Metrics.DataTotal = prometheus.NewCounter(prometheus.CounterOpts{ Name: common.COUNTER_KISFLOW_DATA_TOTAL_NAME, Help: common.COUNTER_KISFLOW_DATA_TOTAL_HELP, }) // Initialize FlowDataTotal GaugeVec Metrics.FlowDataTotal = prometheus.NewGaugeVec( prometheus.GaugeOpts{ Name: common.GAUGE_FLOW_DATA_TOTAL_NAME, Help: common.GAUGE_FLOW_DATA_TOTAL_HELP, }, // Label name []string{common.LABEL_FLOW_NAME}, ) // Initialize FlowScheduleCntsTotal GaugeVec Metrics.FlowScheduleCntsTotal = prometheus.NewGaugeVec( prometheus.GaugeOpts{ Name: common.GAUGE_FLOW_SCHE_CNTS_NAME, Help: common.GAUGE_FLOW_SCHE_CNTS_HELP, }, // Label name []string{common.LABEL_FLOW_NAME}, ) // Initialize FuncScheduleCntsTotal GaugeVec Metrics.FuncScheduleCntsTotal = prometheus.NewGaugeVec( prometheus.GaugeOpts{ Name: common.GAUGE_FUNC_SCHE_CNTS_NAME, Help: common.GAUGE_FUNC_SCHE_CNTS_HELP, }, // Label names []string{common.LABEL_FUNCTION_NAME, common.LABEL_FUNCTION_MODE}, ) // ++++++++++++++++++++++++++ // Initialize 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}, // unit ms, max half a minute }, []string{common.LABEL_FUNCTION_NAME, common.LABEL_FUNCTION_MODE}, ) // Register Metrics prometheus.MustRegister(Metrics.DataTotal) prometheus.MustRegister(Metrics.FlowDataTotal) prometheus.MustRegister(Metrics.FlowScheduleCntsTotal) prometheus.MustRegister(Metrics.FuncScheduleCntsTotal) // +++++++ prometheus.MustRegister(Metrics.FunctionDuration)
Enter fullscreen mode Exit fullscreen mode

}

 Related constant definitions: > kis-flow/common/const.go ```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 = "Total data processed by KisFlow" GAUGE_FLOW_DATA_TOTAL_NAME string = "flow_data_total" GAUGE_FLOW_DATA_TOTAL_HELP string = "Total data processed by each FlowID in KisFlow" GAUGE_FLOW_SCHE_CNTS_NAME string = "flow_schedule_cnts" GAUGE_FLOW_SCHE_CNTS_HELP string = "Flow schedule counts for each FlowID in KisFlow" GAUGE_FUNC_SCHE_CNTS_NAME string = "func_schedule_cnts" GAUGE_FUNC_SCHE_CNTS_HELP string = "Function schedule counts for each Function in KisFlow" HISTOGRAM_FUNCTION_DURATION_NAME string = "func_run_duration" HISTOGRAM_FUNCTION_DURATION_HELP string = "Function execution duration" )
Enter fullscreen mode Exit fullscreen mode

(3) Metric Instrumentation

To measure the execution time of each Function, we should instrument the main entry point of the Flow, flow.Run(), as follows:

kis-flow/flow/kis_flow.go

// Run starts the stream processing of KisFlow, executing from the starting 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 is configured to be disabled return nil } // ++++++++++ Metrics +++++++++ var funcStart time.Time // Since no Function has been executed yet, set PrevFunctionId to FirstVirtual as there is no previous Function flow.PrevFunctionId = common.FunctionIdFirstVirtual // Commit the original stream data if err := flow.commitSrcData(ctx); err != nil { return err } if config.GlobalConfig.EnableProm == true { // Record Flow schedule count metrics.Metrics.FlowScheduleCntsTotal.WithLabelValues(flow.Name).Inc() } // Chain-style stream invocation for fn != nil && flow.abort == false { // Record the current Function being executed in the Flow fid := fn.GetId() flow.ThisFunction = fn flow.ThisFunctionId = fid fName := fn.GetConfig().FName fMode := fn.GetConfig().FMode if config.GlobalConfig.EnableProm == true { // Record Function schedule count metrics.Metrics.FuncScheduleCntsTotal.WithLabelValues(fName, fMode).Inc() // +++++++++++++++ // Record Function execution time start funcStart = time.Now() } // Obtain the source data for the current Function to process 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 } // +++++++++++++++ // Record Function execution duration if config.GlobalConfig.EnableProm == true { // Function execution duration duration := time.Since(funcStart) // Record the current Function execution time metric metrics.Metrics.FunctionDuration.With( prometheus.Labels{ common.LABEL_FUNCTION_NAME: fName, common.LABEL_FUNCTION_MODE: fMode}).Observe(duration.Seconds() * 1000) } // +++++++++++++++ } } return nil
Enter fullscreen mode Exit fullscreen mode

}

 The instrumentation captures the start time before invoking the `Call()` method of the Function and calculates the execution duration after the Function completes. This duration is then recorded in the corresponding bucket of the `HistogramVec` for the Function execution time. ### 10.5.5 Metric: Flow Execution Time #### (1) Metric Definition Define the metric type as follows: > kis-flow/metrics/kis_metrics.go ```go // kisMetrics defines the Prometheus metrics for kisFlow type kisMetrics struct { // Total data count DataTotal prometheus.Counter // Total data processed by each Flow FlowDataTotal *prometheus.GaugeVec // Flow schedule count FlowScheduleCntsTotal *prometheus.GaugeVec // Function schedule count FuncScheduleCntsTotal *prometheus.GaugeVec // Function execution time FunctionDuration *prometheus.HistogramVec // Flow execution time FlowDuration *prometheus.HistogramVec // ++++ }
Enter fullscreen mode Exit fullscreen mode

FlowDuration uses the prometheus.HistogramVec type. This type provides distribution statistics across different time intervals, with various buckets representing different time ranges.

(2) Metric Initialization and Registration

kis-flow/metrics/kis_metrics.go

func InitMetrics() { Metrics = new(kisMetrics) 
// Initialize DataTotal Counter Metrics.DataTotal = prometheus.NewCounter(prometheus.CounterOpts{ Name: common.COUNTER_KISFLOW_DATA_TOTAL_NAME, Help: common.COUNTER_KISFLOW_DATA_TOTAL_HELP, }) // Initialize FlowDataTotal GaugeVec Metrics.FlowDataTotal = prometheus.NewGaugeVec( prometheus.GaugeOpts{ Name: common.GAUGE_FLOW_DATA_TOTAL_NAME, Help: common.GAUGE_FLOW_DATA_TOTAL_HELP, }, // Label name []string{common.LABEL_FLOW_NAME}, ) // Initialize FlowScheduleCntsTotal GaugeVec Metrics.FlowScheduleCntsTotal = prometheus.NewGaugeVec( prometheus.GaugeOpts{ Name: common.GAUGE_FLOW_SCHE_CNTS_NAME, Help: common.GAUGE_FLOW_SCHE_CNTS_HELP, }, // Label name []string{common.LABEL_FLOW_NAME}, ) // Initialize FuncScheduleCntsTotal GaugeVec Metrics.FuncScheduleCntsTotal = prometheus.NewGaugeVec( prometheus.GaugeOpts{ Name: common.GAUGE_FUNC_SCHE_CNTS_NAME, Help: common.GAUGE_FUNC_SCHE_CNTS_HELP, }, // Label names []string{common.LABEL_FUNCTION_NAME, common.LABEL_FUNCTION_MODE}, ) // Initialize 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}, // unit ms, max half a minute }, []string{common.LABEL_FUNCTION_NAME, common.LABEL_FUNCTION_MODE}, ) // ++++++++++++++++++++++++++ // Initialize 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}, // unit ms, max half a minute }, []string{common.LABEL_FLOW_NAME}, ) // Register 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)
Enter fullscreen mode Exit fullscreen mode

}

 Related constant definitions: > kis-flow/common/const.go ```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 = "Total data processed by KisFlow" GAUGE_FLOW_DATA_TOTAL_NAME string = "flow_data_total" GAUGE_FLOW_DATA_TOTAL_HELP string = "Total data processed by each FlowID in KisFlow" GAUGE_FLOW_SCHE_CNTS_NAME string = "flow_schedule_cnts" GAUGE_FLOW_SCHE_CNTS_HELP string = "Flow schedule counts for each FlowID in KisFlow" GAUGE_FUNC_SCHE_CNTS_NAME string = "func_schedule_cnts" GAUGE_FUNC_SCHE_CNTS_HELP string = "Function schedule counts for each Function in KisFlow" HISTOGRAM_FUNCTION_DURATION_NAME string = "func_run_duration" HISTOGRAM_FUNCTION_DURATION_HELP string = "Function execution duration" HISTOGRAM_FLOW_DURATION_NAME string = "flow_run_duration" HISTOGRAM_FLOW_DURATION_HELP string = "Flow execution duration" )
Enter fullscreen mode Exit fullscreen mode

(3) Metric Instrumentation

To measure the execution time of each Flow, we should instrument the main entry point of the Flow, flow.Run(), as follows:

kis-flow/flow/kis_flow.go

// Run starts the stream processing of KisFlow, executing from the starting 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 is configured to be disabled return nil } // ++++++++++ Metrics +++++++++ var funcStart, flowStart time.Time // Since no Function has been executed yet, set PrevFunctionId to FirstVirtual as there is no previous Function flow.PrevFunctionId = common.FunctionIdFirstVirtual // Commit the original stream data if err := flow.commitSrcData(ctx); err != nil { return err } if config.GlobalConfig.EnableProm == true { // Record Flow schedule count metrics.Metrics.FlowScheduleCntsTotal.WithLabelValues(flow.Name).Inc() // Record Flow execution time start flowStart = time.Now() } // Chain-style stream invocation for fn != nil && flow.abort == false { // Record the current Function being executed in the Flow fid := fn.GetId() flow.ThisFunction = fn flow.ThisFunctionId = fid fName := fn.GetConfig().FName fMode := fn.GetConfig().FMode if config.GlobalConfig.EnableProm == true { // Record Function schedule count metrics.Metrics.FuncScheduleCntsTotal.WithLabelValues(fName, fMode).Inc() // Record Function execution time start funcStart = time.Now() } // Obtain the source data for the current Function to process 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 } // Record Function execution duration if config.GlobalConfig.EnableProm == true { // Function execution duration duration := time.Since(funcStart) // Record the current Function execution time metric metrics.Metrics.FunctionDuration.With( prometheus.Labels{ common.LABEL_FUNCTION_NAME: fName, common.LABEL_FUNCTION_MODE: fMode}).Observe(duration.Seconds() * 1000) } } } // Record Flow execution duration if config.GlobalConfig.EnableProm == true { // Flow execution duration duration := time.Since(flowStart) // Record the Flow execution time metric metrics.Metrics.FlowDuration.With( prometheus.Labels{ common.LABEL_FLOW_NAME: flow.Name}).Observe(duration.Seconds() * 1000) } return nil
Enter fullscreen mode Exit fullscreen mode

}

 The instrumentation captures the start time before invoking the first Function and calculates the execution duration after the Flow completes. This duration is then recorded in the corresponding bucket of the `HistogramVec` for the Flow execution time. ## 10.6 KieMetrics Unit Testing (Other Metrics Indicators) ### 10.6.1 Creating Unit Tests We can reuse the previous TestMetricsDataTotal() method for unit test cases, as shown below: > kis-flow/test/kis_metrics_test.go ```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. Register Function callbacks kis.Pool().FaaS("funcName1", faas.FuncDemo1Handler) kis.Pool().FaaS("funcName2", faas.FuncDemo2Handler) kis.Pool().FaaS("funcName3", faas.FuncDemo3Handler) // 0. Register ConnectorInit and Connector callbacks kis.Pool().CaaSInit("ConnName1", caas.InitConnDemo1) kis.Pool().CaaS("ConnName1", "funcName2", common.S, caas.CaasDemoHandler1) // 1. Load configuration files and build Flow if err := file.ConfigImportYaml("/Users/tal/gopath/src/kis-flow/test/load_conf/"); err != nil { panic(err) } // 2. Get Flow flow1 := kis.Pool().GetFlow("flowName1") n := 0 for n < 10 { // 3. Submit raw data _ = flow1.CommitRow("This is Data1 from Test") // 4. Execute flow1 if err := flow1.Run(ctx); err != nil { panic(err) } time.Sleep(1 * time.Second) n++ } select {} }
Enter fullscreen mode Exit fullscreen mode

Execute the unit test by navigating to kis-flow/test/ and running:

go test -test.v -test.paniconexit0 -test.run TestMetricsDataTotal
Enter fullscreen mode Exit fullscreen mode

You will see many log outputs. After waiting for 10 seconds, open another terminal and input the following command:

curl http://0.0.0.0:20004/metrics
Enter fullscreen mode Exit fullscreen mode

You will see the following results:

# HELP flow_data_total KisFlow data count for each FlowID # TYPE flow_data_total gauge flow_data_total{flow_name="flowName1"} 10 # HELP flow_run_duration Flow execution time # 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 Number of times each FlowID is scheduled in KisFlow # TYPE flow_schedule_cnts gauge flow_schedule_cnts{flow_name="flowName1"} 10 # HELP func_run_duration Function execution time # 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"} 13.858197 func_run_duration_count{func_mode="Verify",func_name="funcName1"} 10
Enter fullscreen mode Exit fullscreen mode

This concludes the section on KieMetrics Unit Testing and other Metrics Indicators.

10.7 Grafana Dashboard Display for KisFlow Metrics

With Prometheus metrics collected, we can integrate Grafana to display dashboards for KisFlow stream processing programs. Since each developer's project metrics and dashboard requirements may vary, this document does not provide specific Grafana dashboard configuration files. Instead, here is a sample dashboard for a KisFlow project for demonstration purposes, as shown below:

1

2

3

10.8 [V0.9] Source Code

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

Author: Aceld
GitHub: https://github.com/aceld

KisFlow Open Source Project Address: https://github.com/aceld/kis-flow

Document: https://github.com/aceld/kis-flow/wiki

Part1-OverView
Part2.1-Project Construction / Basic Modules
Part2.2-Project Construction / Basic Modules
Part3-Data Stream
Part4-Function Scheduling
Part5-Connector
Part6-Configuration Import and Export
Part7-KisFlow Action
Part8-Cache/Params Data Caching and Data Parameters
Part9-Multiple Copies of Flow
Part10-Prometheus Metrics Statistics
Part11-Adaptive Registration of FaaS Parameter Types Based on Reflection

Case1-Quick Start
Case2-Flow Parallel Operation
Case3-Application of KisFlow in Multi-Goroutines
Case4-KisFlow in Message Queue (MQ) Applications

Top comments (0)