ttl: add more metrics for TTL

metrics/metrics.go

@@ -218,6 +218,7 @@ func RegisterMetrics() {
  prometheus.MustRegister(TTLJobStatus)
  prometheus.MustRegister(TTLTaskStatus)
  prometheus.MustRegister(TTLPhaseTime)
+ prometheus.MustRegister(TTLInsertRowsCount)
 
  prometheus.MustRegister(EMACPUUsageGauge)
  prometheus.MustRegister(PoolConcurrencyCounter)

metrics/ttl.go

@@ -58,4 +58,12 @@ var (
  Name: "ttl_phase_time",
  Help: "The time spent in each phase",
  }, []string{LblType, LblPhase})
+
+ TTLInsertRowsCount = prometheus.NewCounter(
+ prometheus.CounterOpts{
+ Namespace: "tidb",
+ Subsystem: "server",
+ Name: "ttl_insert_rows",
+ Help: "The count of TTL rows inserted",
+ })
 )

session/BUILD.bazel

@@ -136,6 +136,7 @@ go_test(
  "//expression",
  "//kv",
  "//meta",
+ "//metrics",
  "//parser/ast",
  "//parser/auth",
  "//parser/model",
@@ -162,6 +163,8 @@ go_test(
  "//util/sqlexec",
  "@com_github_pingcap_failpoint//:failpoint",
  "@com_github_pingcap_log//:log",
+ "@com_github_prometheus_client_golang//prometheus",
+ "@com_github_prometheus_client_model//go",
  "@com_github_stretchr_testify//require",
  "@com_github_tikv_client_go_v2//testutils",
  "@com_github_tikv_client_go_v2//tikv",

session/session.go

@@ -1091,6 +1091,9 @@ func (s *session) CommitTxn(ctx context.Context) error {
  s.sessionVars.StmtCtx.MergeExecDetails(nil, commitDetail)
  }
 
+ // record the TTLInsertRows in the metric
+ metrics.TTLInsertRowsCount.Add(float64(s.sessionVars.TxnCtx.InsertTTLRowsCount))
+
  failpoint.Inject("keepHistory", func(val failpoint.Value) {
  if val.(bool) {
  failpoint.Return(err)

session/session_test.go

@@ -21,10 +21,13 @@ import (
  "testing"
 
  "github.com/pingcap/tidb/ddl"
+ "github.com/pingcap/tidb/metrics"
  "github.com/pingcap/tidb/session"
  "github.com/pingcap/tidb/tablecodec"
  "github.com/pingcap/tidb/testkit"
  "github.com/pingcap/tidb/testkit/external"
+ "github.com/prometheus/client_golang/prometheus"
+ dto "github.com/prometheus/client_model/go"
  "github.com/stretchr/testify/require"
 )
 
@@ -89,3 +92,49 @@ func TestMetaTableRegion(t *testing.T) {
 
  require.NotEqual(t, ddlBackfillTableRegionID, ddlBackfillHistoryTableRegionID)
 }
+
+func MustReadCounter(t *testing.T, m prometheus.Counter) float64 {
+ pb := &dto.Metric{}
+ require.NoError(t, m.Write(pb))
+ return pb.GetCounter().GetValue()
+}
+
+func TestRecordTTLRows(t *testing.T) {
+ store := testkit.CreateMockStore(t)
+ tk := testkit.NewTestKit(t, store)
+
+ tk.MustExec("use test")
+ tk.MustExec("create table t(created_at datetime) TTL = created_at + INTERVAL 1 DAY")
+ // simple insert should be recorded
+ tk.MustExec("insert into t values (NOW())")
+ require.Equal(t, 1.0, MustReadCounter(t, metrics.TTLInsertRowsCount))
+
+ // insert in a explicit transaction should be recorded
+ tk.MustExec("begin")
+ tk.MustExec("insert into t values (NOW())")
+ tk.MustExec("commit")
+ require.Equal(t, 2.0, MustReadCounter(t, metrics.TTLInsertRowsCount))
+
+ // insert multiple rows should be the same
+ tk.MustExec("begin")
+ tk.MustExec("insert into t values (NOW())")
+ tk.MustExec("insert into t values (NOW())")
+ tk.MustExec("commit")
+ require.Equal(t, 4.0, MustReadCounter(t, metrics.TTLInsertRowsCount))
+
+ // rollback will remove all recorded TTL rows
+ tk.MustExec("begin")
+ tk.MustExec("insert into t values (NOW())")
+ tk.MustExec("insert into t values (NOW())")
+ tk.MustExec("rollback")
+ require.Equal(t, 6.0, MustReadCounter(t, metrics.TTLInsertRowsCount))
+
+ // savepoint will save the recorded TTL rows
+ tk.MustExec("begin")
+ tk.MustExec("insert into t values (NOW())")
+ tk.MustExec("savepoint insert1")
+ tk.MustExec("insert into t values (NOW())")
+ tk.MustExec("rollback to insert1")
+ tk.MustExec("commit")
+ require.Equal(t, 7.0, MustReadCounter(t, metrics.TTLInsertRowsCount))
+}

sessionctx/variable/session.go

@@ -187,6 +187,9 @@ type TxnCtxNeedToRestore struct {
 
  // CachedTables is not nil if the transaction write on cached table.
  CachedTables map[int64]interface{}
+
+ // InsertTTLRowsCount counts how many rows are inserted in this statement
+ InsertTTLRowsCount int
 }
 
 // TxnCtxNoNeedToRestore stores transaction variables which do not need to restored when rolling back to a savepoint.
@@ -377,6 +380,7 @@ func (tc *TransactionContext) GetCurrentSavepoint() TxnCtxNeedToRestore {
  TableDeltaMap: tableDeltaMap,
  pessimisticLockCache: pessimisticLockCache,
  CachedTables: cachedTables,
+ InsertTTLRowsCount: tc.InsertTTLRowsCount,
  }
 }
 
@@ -385,6 +389,7 @@ func (tc *TransactionContext) RestoreBySavepoint(savepoint TxnCtxNeedToRestore)
  tc.TableDeltaMap = savepoint.TableDeltaMap
  tc.pessimisticLockCache = savepoint.pessimisticLockCache
  tc.CachedTables = savepoint.CachedTables
+ tc.InsertTTLRowsCount = savepoint.InsertTTLRowsCount
 }
 
 // AddSavepoint adds a new savepoint.

table/tables/tables.go

@@ -924,6 +924,9 @@ func (t *TableCommon) AddRecord(sctx sessionctx.Context, r []types.Datum, opts .
  return nil, err
  }
  }
+ if shouldIncreaseTTLMetricCount(t.meta) {
+ sctx.GetSessionVars().TxnCtx.InsertTTLRowsCount += 1
+ }
  if sessVars.TxnCtx == nil {
  return recordID, nil
  }
@@ -1592,6 +1595,10 @@ func shouldWriteBinlog(ctx sessionctx.Context, tblInfo *model.TableInfo) bool {
  return !ctx.GetSessionVars().InRestrictedSQL
 }
 
+func shouldIncreaseTTLMetricCount(tblInfo *model.TableInfo) bool {
+ return tblInfo.TTLInfo != nil
+}
+
 func (t *TableCommon) getMutation(ctx sessionctx.Context) *binlog.TableMutation {
  return ctx.StmtGetMutation(t.tableID)
 }

ttl/metrics/metrics.go

@@ -49,7 +49,8 @@ var (
  RunningJobsCnt = metrics.TTLJobStatus.With(prometheus.Labels{metrics.LblType: "running"})
  CancellingJobsCnt = metrics.TTLJobStatus.With(prometheus.Labels{metrics.LblType: "cancelling"})
 
- RunningTaskCnt = metrics.TTLTaskStatus.With(prometheus.Labels{metrics.LblType: "running"})
+ ScanningTaskCnt = metrics.TTLTaskStatus.With(prometheus.Labels{metrics.LblType: "scanning"})
+ DeletingTaskCnt = metrics.TTLTaskStatus.With(prometheus.Labels{metrics.LblType: "deleting"})
 )
 
 func initWorkerPhases(workerType string) map[string]prometheus.Counter {

ttl/ttlworker/task_manager.go

@@ -497,7 +497,17 @@ func (m *taskManager) checkInvalidTask(se session.Session) {
 }
 
 func (m *taskManager) reportMetrics() {
- metrics.RunningTaskCnt.Set(float64(len(m.runningTasks)))
+ scanningTaskCnt := 0
+ deletingTaskCnt := 0
+ for _, task := range m.runningTasks {
+ if task.result != nil {
+ scanningTaskCnt += 1
+ } else {
+ deletingTaskCnt += 1
+ }
+ }
+ metrics.ScanningTaskCnt.Set(float64(scanningTaskCnt))
+ metrics.DeletingTaskCnt.Set(float64(deletingTaskCnt))
 }
 
 type runningScanTask struct {

ttl/ttlworker/task_manager_integration_test.go

@@ -227,7 +227,7 @@ func TestTaskMetrics(t *testing.T) {
 
  m.ReportMetrics()
  out := &dto.Metric{}
- require.NoError(t, metrics.RunningTaskCnt.Write(out))
+ require.NoError(t, metrics.DeletingTaskCnt.Write(out))
  require.Equal(t, float64(1), out.GetGauge().GetValue())
 }