fix: reset some metrics when loader resume (#558)

dm/dm-ansible/scripts/dm.json

@@ -1789,7 +1789,7 @@
           "tableColumn": "",
           "targets": [
             {
-              "expr": "dm_loader_data_size_count{task=\"$task\",instance=\"$instance\"}",
+              "expr": "dm_loader_data_size_gauge{task=\"$task\",instance=\"$instance\"}",
               "format": "time_series",
               "intervalFactor": 2,
               "refId": "A"
@@ -2017,7 +2017,7 @@
           "tableColumn": "",
           "targets": [
             {
-              "expr": "dm_loader_table_count{task=\"$task\",instance=\"$instance\"}",
+              "expr": "dm_loader_table_gauge{task=\"$task\",instance=\"$instance\"}",
               "format": "time_series",
               "intervalFactor": 2,
               "refId": "A"
@@ -2093,7 +2093,7 @@
           "tableColumn": "",
           "targets": [
             {
-              "expr": "dm_loader_data_file_count{task=\"$task\",instance=\"$instance\"}",
+              "expr": "dm_loader_data_file_gauge{task=\"$task\",instance=\"$instance\"}",
               "format": "time_series",
               "intervalFactor": 2,
               "refId": "A"

loader/loader.go

@@ -807,6 +807,8 @@ func (l *Loader) prepareDbFiles(files map[string]struct{}) error {
 }
 
 func (l *Loader) prepareTableFiles(files map[string]struct{}) error {
+	var tablesNumber float64
+
 	for file := range files {
 		if !strings.HasSuffix(file, "-schema.sql") {
 			continue
@@ -839,15 +841,18 @@ func (l *Loader) prepareTableFiles(files map[string]struct{}) error {
 		if _, ok := tables[table]; ok {
 			return terror.ErrLoadUnitDuplicateTableFile.Generate(file)
 		}
-		tableCounter.WithLabelValues(l.cfg.Name).Inc()
+		tablesNumber++
 		tables[table] = make(DataFiles, 0, 16)
 		l.totalFileCount.Add(1) // for table
 	}
 
+	tableGauge.WithLabelValues(l.cfg.Name).Set(tablesNumber)
 	return nil
 }
 
 func (l *Loader) prepareDataFiles(files map[string]struct{}) error {
+	var dataFilesNumber float64
+
 	for file := range files {
 		if !strings.HasSuffix(file, ".sql") || strings.Contains(file, "-schema.sql") ||
 			strings.Contains(file, "-schema-create.sql") {
@@ -892,11 +897,12 @@ func (l *Loader) prepareDataFiles(files map[string]struct{}) error {
 		l.totalFileCount.Add(1) // for data
 
 		dataFiles = append(dataFiles, file)
-		dataFileCounter.WithLabelValues(l.cfg.Name).Inc()
+		dataFilesNumber++
 		tables[table] = dataFiles
 	}
 
-	dataSizeCounter.WithLabelValues(l.cfg.Name).Add(float64(l.totalDataSize.Get()))
+	dataFileGauge.WithLabelValues(l.cfg.Name).Set(dataFilesNumber)
+	dataSizeGauge.WithLabelValues(l.cfg.Name).Set(float64(l.totalDataSize.Get()))
 	return nil
 }
 

loader/metrics.go

@@ -45,27 +45,27 @@ var (
 			Buckets:   prometheus.ExponentialBuckets(0.0005, 2, 16),
 		}, []string{"task"})
 
-	dataFileCounter = prometheus.NewCounterVec(
-		prometheus.CounterOpts{
+	dataFileGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
 			Namespace: "dm",
 			Subsystem: "loader",
-			Name:      "data_file_count",
+			Name:      "data_file_gauge",
 			Help:      "data files in total",
 		}, []string{"task"})
 
-	tableCounter = prometheus.NewCounterVec(
-		prometheus.CounterOpts{
+	tableGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
 			Namespace: "dm",
 			Subsystem: "loader",
-			Name:      "table_count",
+			Name:      "table_gauge",
 			Help:      "tables in total",
 		}, []string{"task"})
 
-	dataSizeCounter = prometheus.NewCounterVec(
-		prometheus.CounterOpts{
+	dataSizeGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
 			Namespace: "dm",
 			Subsystem: "loader",
-			Name:      "data_size_count",
+			Name:      "data_size_gauge",
 			Help:      "data size in total",
 		}, []string{"task"})
 
@@ -92,9 +92,9 @@ func RegisterMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(tidbExecutionErrorCounter)
 	registry.MustRegister(txnHistogram)
 	registry.MustRegister(queryHistogram)
-	registry.MustRegister(dataFileCounter)
-	registry.MustRegister(tableCounter)
-	registry.MustRegister(dataSizeCounter)
+	registry.MustRegister(dataFileGauge)
+	registry.MustRegister(tableGauge)
+	registry.MustRegister(dataSizeGauge)
 	registry.MustRegister(progressGauge)
 	registry.MustRegister(loaderExitWithErrorCounter)
 }