pingcap · ti-chi-bot · Mar 25, 2025
diff --git a/pkg/ddl/index.go b/pkg/ddl/index.go
@@ -2291,6 +2291,7 @@ func (w *worker) updateJobRowCount(taskKey string, jobID int64) {
 	w.getReorgCtx(jobID).setRowCount(rowCount)
 }
 
+<<<<<<< HEAD
 // submitAndWaitTask submits a task and wait for it to finish.
 func submitAndWaitTask(ctx context.Context, taskKey string, taskType proto.TaskType, concurrency int, targetScope string, taskMeta []byte) error {
 	task, err := handle.SubmitTask(ctx, taskKey, taskType, concurrency, targetScope, taskMeta)
@@ -2300,6 +2301,8 @@ func submitAndWaitTask(ctx context.Context, taskKey string, taskType proto.TaskT
 	return handle.WaitTaskDoneOrPaused(ctx, task.ID)
 }
 
+=======
+>>>>>>> b08a92082fe (disttask: fix task metrics for distributed execute framework panel (#59940))
 func getNextPartitionInfo(reorg *reorgInfo, t table.PartitionedTable, currPhysicalTableID int64) (int64, kv.Key, kv.Key, error) {
 	pi := t.Meta().GetPartitionInfo()
 	if pi == nil {

diff --git a/pkg/disttask/framework/handle/BUILD.bazel b/pkg/disttask/framework/handle/BUILD.bazel
@@ -8,7 +8,6 @@ go_library(
     deps = [
         "//pkg/disttask/framework/proto",
         "//pkg/disttask/framework/storage",
-        "//pkg/metrics",
         "//pkg/util/backoff",
         "//pkg/util/logutil",
         "@com_github_pingcap_errors//:errors",

diff --git a/pkg/disttask/framework/handle/handle.go b/pkg/disttask/framework/handle/handle.go
@@ -21,7 +21,6 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/pkg/disttask/framework/proto"
 	"github.com/pingcap/tidb/pkg/disttask/framework/storage"
-	"github.com/pingcap/tidb/pkg/metrics"
 	"github.com/pingcap/tidb/pkg/util/backoff"
 	"github.com/pingcap/tidb/pkg/util/logutil"
 	"go.uber.org/zap"
@@ -77,7 +76,6 @@ func SubmitTask(ctx context.Context, taskKey string, taskType proto.TaskType, co
 	if err != nil {
 		return nil, err
 	}
-	metrics.UpdateMetricsForAddTask(&task.TaskBase)
 
 	NotifyTaskChange()
 	return task, nil

diff --git a/pkg/disttask/framework/mock/scheduler_mock.go b/pkg/disttask/framework/mock/scheduler_mock.go
diff --git a/pkg/disttask/framework/scheduler/BUILD.bazel b/pkg/disttask/framework/scheduler/BUILD.bazel
@@ -23,7 +23,6 @@ go_library(
         "//pkg/domain/infosync",
         "//pkg/kv",
         "//pkg/lightning/log",
-        "//pkg/metrics",
         "//pkg/sessionctx",
         "//pkg/util",
         "//pkg/util/backoff",

diff --git a/pkg/disttask/framework/scheduler/collector.go b/pkg/disttask/framework/scheduler/collector.go
@@ -23,10 +23,10 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
-var subtaskCollector = newCollector()
+var disttaskCollector = newCollector()
 
 func init() {
-	prometheus.MustRegister(subtaskCollector)
+	prometheus.MustRegister(disttaskCollector)
 }
 
 // Because the exec_id of a subtask may change, after all tasks
@@ -37,13 +37,20 @@ func init() {
 // Therefore, a custom collector is used.
 type collector struct {
 	subtaskInfo atomic.Pointer[[]*proto.SubtaskBase]
+	taskInfo    atomic.Pointer[[]*proto.TaskBase]
 
+	tasks           *prometheus.Desc
 	subtasks        *prometheus.Desc
 	subtaskDuration *prometheus.Desc
 }
 
 func newCollector() *collector {
 	return &collector{
+		tasks: prometheus.NewDesc(
+			"tidb_disttask_task_status",
+			"Number of tasks.",
+			[]string{"task_type", "status"}, nil,
+		),
 		subtasks: prometheus.NewDesc(
 			"tidb_disttask_subtasks",
 			"Number of subtasks.",
@@ -59,12 +66,45 @@ func newCollector() *collector {
 
 // Describe implements the prometheus.Collector interface.
 func (c *collector) Describe(ch chan<- *prometheus.Desc) {
+	ch <- c.tasks
 	ch <- c.subtasks
 	ch <- c.subtaskDuration
 }
 
 // Collect implements the prometheus.Collector interface.
 func (c *collector) Collect(ch chan<- prometheus.Metric) {
+	c.collectTasks(ch)
+	c.collectSubtasks(ch)
+}
+
+func (c *collector) collectTasks(ch chan<- prometheus.Metric) {
+	p := c.taskInfo.Load()
+	if p == nil {
+		return
+	}
+	tasks := *p
+	// task type => state => cnt
+	taskTypeStateCnt := make(map[string]map[string]int)
+	for _, task := range tasks {
+		tp := task.Type.String()
+		if _, ok := taskTypeStateCnt[tp]; !ok {
+			taskTypeStateCnt[tp] = make(map[string]int)
+		}
+		state := task.State.String()
+		taskTypeStateCnt[tp][state]++
+	}
+	for tp, stateCnt := range taskTypeStateCnt {
+		for state, cnt := range stateCnt {
+			ch <- prometheus.MustNewConstMetric(c.tasks, prometheus.GaugeValue,
+				float64(cnt),
+				tp,
+				state,
+			)
+		}
+	}
+}
+
+func (c *collector) collectSubtasks(ch chan<- prometheus.Metric) {
 	p := c.subtaskInfo.Load()
 	if p == nil {
 		return

diff --git a/pkg/disttask/framework/scheduler/interface.go b/pkg/disttask/framework/scheduler/interface.go
@@ -29,6 +29,8 @@ type TaskManager interface {
 	// to make sure lower rank tasks can be scheduled if resource is enough.
 	// The returned tasks are sorted by task order, see proto.Task.
 	GetTopUnfinishedTasks(ctx context.Context) ([]*proto.TaskBase, error)
+	// GetAllTasks gets all tasks with basic columns.
+	GetAllTasks(ctx context.Context) ([]*proto.TaskBase, error)
 	// GetAllSubtasks gets all subtasks with basic columns.
 	GetAllSubtasks(ctx context.Context) ([]*proto.SubtaskBase, error)
 	GetTasksInStates(ctx context.Context, states ...any) (task []*proto.Task, err error)

diff --git a/pkg/disttask/framework/scheduler/scheduler_manager.go b/pkg/disttask/framework/scheduler/scheduler_manager.go
@@ -24,7 +24,6 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/pkg/disttask/framework/handle"
 	"github.com/pingcap/tidb/pkg/disttask/framework/proto"
-	"github.com/pingcap/tidb/pkg/metrics"
 	tidbutil "github.com/pingcap/tidb/pkg/util"
 	"github.com/pingcap/tidb/pkg/util/intest"
 	"github.com/pingcap/tidb/pkg/util/syncutil"
@@ -284,8 +283,6 @@ func (sm *Manager) startSchedulers(schedulableTasks []*proto.TaskBase) error {
 			sm.logger.Info("start scheduler without allocating slots",
 				zap.Int64("task-id", task.ID), zap.Stringer("state", task.State))
 		}
-
-		metrics.UpdateMetricsForScheduleTask(task)
 		sm.startScheduler(task, allocateSlots, reservedExecID)
 	}
 	return nil
@@ -362,7 +359,6 @@ func (sm *Manager) startScheduler(basicTask *proto.TaskBase, allocateSlots bool,
 			handle.NotifyTaskChange()
 			sm.logger.Info("task scheduler exist", zap.Int64("task-id", task.ID))
 		}()
-		metrics.UpdateMetricsForRunTask(task)
 		scheduler.ScheduleTask()
 		sm.finishCh <- struct{}{}
 	})
@@ -436,7 +432,6 @@ func (sm *Manager) cleanupFinishedTasks(tasks []*proto.Task) error {
 			// if task doesn't register cleanup function, mark it as cleaned.
 			cleanedTasks = append(cleanedTasks, task)
 		}
-		metrics.UpdateMetricsForFinishTask(task)
 	}
 	if firstErr != nil {
 		sm.logger.Warn("cleanup routine failed", zap.Error(errors.Trace(firstErr)))
@@ -465,13 +460,17 @@ func (sm *Manager) collectLoop() {
 }
 
 func (sm *Manager) collect() {
+	tasks, err := sm.taskMgr.GetAllTasks(sm.ctx)
+	if err != nil {
+		sm.logger.Warn("get all tasks failed", zap.Error(err))
+	}
 	subtasks, err := sm.taskMgr.GetAllSubtasks(sm.ctx)
 	if err != nil {
 		sm.logger.Warn("get all subtasks failed", zap.Error(err))
 		return
 	}
-
-	subtaskCollector.subtaskInfo.Store(&subtasks)
+	disttaskCollector.taskInfo.Store(&tasks)
+	disttaskCollector.subtaskInfo.Store(&subtasks)
 }
 
 // MockScheduler mock one scheduler for one task, only used for tests.

diff --git a/pkg/disttask/framework/storage/task_table.go b/pkg/disttask/framework/storage/task_table.go
@@ -785,6 +785,22 @@ func (mgr *TaskManager) GetSubtasksWithHistory(ctx context.Context, taskID int64
 	return subtasks, nil
 }
 
+// GetAllTasks gets all tasks with basic columns.
+func (mgr *TaskManager) GetAllTasks(ctx context.Context) ([]*proto.TaskBase, error) {
+	rs, err := mgr.ExecuteSQLWithNewSession(ctx, `select `+basicTaskColumns+` from mysql.tidb_global_task t`)
+	if err != nil {
+		return nil, err
+	}
+	if len(rs) == 0 {
+		return nil, nil
+	}
+	tasks := make([]*proto.TaskBase, 0, len(rs))
+	for _, r := range rs {
+		tasks = append(tasks, row2TaskBasic(r))
+	}
+	return tasks, nil
+}
+
 // GetAllSubtasks gets all subtasks with basic columns.
 func (mgr *TaskManager) GetAllSubtasks(ctx context.Context) ([]*proto.SubtaskBase, error) {
 	rs, err := mgr.ExecuteSQLWithNewSession(ctx, `select `+basicSubtaskColumns+` from mysql.tidb_background_subtask`)

diff --git a/pkg/disttask/importinto/job.go b/pkg/disttask/importinto/job.go
@@ -28,8 +28,11 @@ import (
 	"github.com/pingcap/tidb/pkg/domain/infosync"
 	"github.com/pingcap/tidb/pkg/executor/importer"
 	"github.com/pingcap/tidb/pkg/kv"
+<<<<<<< HEAD
 	"github.com/pingcap/tidb/pkg/lightning/checkpoints"
 	"github.com/pingcap/tidb/pkg/metrics"
+=======
+>>>>>>> b08a92082fe (disttask: fix task metrics for distributed execute framework panel (#59940))
 	"github.com/pingcap/tidb/pkg/sessionctx"
 	"github.com/pingcap/tidb/pkg/util/logutil"
 	"github.com/tikv/client-go/v2/util"
@@ -105,8 +108,6 @@ func doSubmitTask(ctx context.Context, plan *importer.Plan, stmt string, instanc
 		return 0, nil, err
 	}
 
-	metrics.UpdateMetricsForAddTask(task)
-
 	logutil.BgLogger().Info("job submitted to task queue",
 		zap.Int64("job-id", jobID),
 		zap.Int64("task-id", task.ID),

diff --git a/pkg/metrics/BUILD.bazel b/pkg/metrics/BUILD.bazel
@@ -29,7 +29,6 @@ go_library(
     importpath = "github.com/pingcap/tidb/pkg/metrics",
     visibility = ["//visibility:public"],
     deps = [
-        "//pkg/disttask/framework/proto",
         "//pkg/lightning/metric",
         "//pkg/parser/terror",
         "//pkg/timer/metrics",

diff --git a/pkg/metrics/disttask.go b/pkg/metrics/disttask.go
@@ -15,55 +15,16 @@
 package metrics
 
 import (
-	"fmt"
-	"time"
-
-	"github.com/pingcap/tidb/pkg/disttask/framework/proto"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
-const (
-	lblTaskStatus = "status"
-	lblTaskType   = "task_type"
-	lblTaskID     = "task_id"
-	lblSubTaskID  = "subtask_id"
-	lblExecID     = "exec_id"
-)
-
-// status for task
-const (
-	SchedulingStatus = "scheduling"
-	WaitingStatus    = "waiting"
-	RunningStatus    = "running"
-	CompletedStatus  = "completed"
-)
-
 var (
-	//DistTaskGauge is the gauge of dist task count.
-	DistTaskGauge *prometheus.GaugeVec
-	//DistTaskStartTimeGauge is the gauge of dist task count.
-	DistTaskStartTimeGauge *prometheus.GaugeVec
 	// DistTaskUsedSlotsGauge is the gauge of used slots on executor node.
 	DistTaskUsedSlotsGauge *prometheus.GaugeVec
 )
 
 // InitDistTaskMetrics initializes disttask metrics.
 func InitDistTaskMetrics() {
-	DistTaskGauge = NewGaugeVec(
-		prometheus.GaugeOpts{
-			Namespace: "tidb",
-			Subsystem: "disttask",
-			Name:      "task_status",
-			Help:      "Gauge of disttask.",
-		}, []string{lblTaskType, lblTaskStatus})
-
-	DistTaskStartTimeGauge = NewGaugeVec(
-		prometheus.GaugeOpts{
-			Namespace: "tidb",
-			Subsystem: "disttask",
-			Name:      "start_time",
-			Help:      "Gauge of start_time of disttask.",
-		}, []string{lblTaskType, lblTaskStatus, lblTaskID})
 	DistTaskUsedSlotsGauge = NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "tidb",
@@ -72,30 +33,3 @@ func InitDistTaskMetrics() {
 			Help:      "Gauge of used slots on a executor node.",
 		}, []string{"service_scope"})
 }
-
-// UpdateMetricsForAddTask update metrics when a task is added
-func UpdateMetricsForAddTask(task *proto.TaskBase) {
-	DistTaskGauge.WithLabelValues(task.Type.String(), WaitingStatus).Inc()
-	DistTaskStartTimeGauge.WithLabelValues(task.Type.String(), WaitingStatus, fmt.Sprint(task.ID)).Set(float64(time.Now().UnixMicro()))
-}
-
-// UpdateMetricsForScheduleTask update metrics when a task is added
-func UpdateMetricsForScheduleTask(task *proto.TaskBase) {
-	DistTaskGauge.WithLabelValues(task.Type.String(), WaitingStatus).Dec()
-	DistTaskGauge.WithLabelValues(task.Type.String(), SchedulingStatus).Inc()
-	DistTaskStartTimeGauge.DeleteLabelValues(task.Type.String(), WaitingStatus, fmt.Sprint(task.ID))
-	DistTaskStartTimeGauge.WithLabelValues(task.Type.String(), SchedulingStatus, fmt.Sprint(task.ID)).SetToCurrentTime()
-}
-
-// UpdateMetricsForRunTask update metrics when a task starts running
-func UpdateMetricsForRunTask(task *proto.Task) {
-	DistTaskStartTimeGauge.DeleteLabelValues(task.Type.String(), SchedulingStatus, fmt.Sprint(task.ID))
-	DistTaskGauge.WithLabelValues(task.Type.String(), SchedulingStatus).Dec()
-	DistTaskGauge.WithLabelValues(task.Type.String(), RunningStatus).Inc()
-}
-
-// UpdateMetricsForFinishTask update metrics when a task is finished
-func UpdateMetricsForFinishTask(task *proto.Task) {
-	DistTaskGauge.WithLabelValues(task.Type.String(), RunningStatus).Dec()
-	DistTaskGauge.WithLabelValues(task.Type.String(), CompletedStatus).Inc()
-}
diff --git a/pkg/metrics/grafana/tidb.json b/pkg/metrics/grafana/tidb.json
@@ -15320,6 +15320,7 @@
           "timeFrom": null,
           "timeRegions": [],
           "timeShift": null,
+          "decimals": 0,
           "title": "Task Status",
           "tooltip": {
             "shared": true,
@@ -15341,7 +15342,8 @@
               "logBase": 1,
               "max": null,
               "min": null,
-              "show": true
+              "show": true,
+              "decimals": 0
             },
             {
               "format": "short",

diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go
@@ -268,8 +268,6 @@ func RegisterMetrics() {
 	prometheus.MustRegister(PlanReplayerTaskCounter)
 	prometheus.MustRegister(PlanReplayerRegisterTaskGauge)
 
-	prometheus.MustRegister(DistTaskGauge)
-	prometheus.MustRegister(DistTaskStartTimeGauge)
 	prometheus.MustRegister(DistTaskUsedSlotsGauge)
 	prometheus.MustRegister(RunawayCheckerCounter)
 	prometheus.MustRegister(GlobalSortWriteToCloudStorageDuration)