sql: track CPU time on gateway node

pkg/roachpb/app_stats.go

@@ -156,6 +156,7 @@ func (s *StatementStatistics) Add(other *StatementStatistics) {
 	s.Nodes = util.CombineUniqueInt64(s.Nodes, other.Nodes)
 	s.PlanGists = util.CombineUniqueString(s.PlanGists, other.PlanGists)
 	s.IndexRecommendations = other.IndexRecommendations
+	s.CpuTime.Add(other.CpuTime, s.Count, other.Count)
 
 	s.ExecStats.Add(other.ExecStats)
 
@@ -189,7 +190,8 @@ func (s *StatementStatistics) AlmostEqual(other *StatementStatistics, eps float6
 		s.SensitiveInfo.Equal(other.SensitiveInfo) &&
 		s.BytesRead.AlmostEqual(other.BytesRead, eps) &&
 		s.RowsRead.AlmostEqual(other.RowsRead, eps) &&
-		s.RowsWritten.AlmostEqual(other.RowsWritten, eps)
+		s.RowsWritten.AlmostEqual(other.RowsWritten, eps) &&
+		s.CpuTime.AlmostEqual(other.CpuTime, eps)
 	// s.ExecStats are deliberately ignored since they are subject to sampling
 	// probability and are not fully deterministic (e.g. the number of network
 	// messages depends on the range cache state).

pkg/roachpb/app_stats.proto

@@ -113,6 +113,9 @@ message StatementStatistics {
   // index_recommendations is the list of index recommendations generated for the statement fingerprint.
   repeated string index_recommendations = 27;
 
+  // cpu_time is the amount of CPU time spent (in seconds) by this query.
+  optional NumericStat cpu_time = 28 [(gogoproto.nullable) = false];
+
   // Note: be sure to update `sql/app_stats.go` when adding/removing fields here!
 
   reserved 13, 14, 17, 18, 19, 20;

pkg/sql/BUILD.bazel

@@ -472,6 +472,7 @@ go_library(
         "//pkg/util/errorutil/unimplemented",
         "//pkg/util/fsm",
         "//pkg/util/grpcutil",
+        "//pkg/util/grunning",
         "//pkg/util/hlc",
         "//pkg/util/humanizeutil",
         "//pkg/util/interval",

pkg/sql/conn_executor.go

@@ -62,6 +62,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/envutil"
 	"github.com/cockroachdb/cockroach/pkg/util/errorutil"
 	"github.com/cockroachdb/cockroach/pkg/util/fsm"
+	"github.com/cockroachdb/cockroach/pkg/util/grunning"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/log/logcrash"
@@ -1878,12 +1879,16 @@ func (ex *connExecutor) execCmd() error {
 			ex.machine.CurState(), pos, cmd)
 	}
 
+	nanos := grunning.Time().Nanoseconds()
+	receivedCPUTime := time.Unix(0, nanos)
+
 	var ev fsm.Event
 	var payload fsm.EventPayload
 	var res ResultBase
 	switch tcmd := cmd.(type) {
 	case ExecStmt:
 		ex.phaseTimes.SetSessionPhaseTime(sessionphase.SessionQueryReceived, tcmd.TimeReceived)
+		ex.phaseTimes.SetSessionPhaseTime(sessionphase.SessionQueryReceivedCPUTime, receivedCPUTime)
 		ex.phaseTimes.SetSessionPhaseTime(sessionphase.SessionStartParse, tcmd.ParseStart)
 		ex.phaseTimes.SetSessionPhaseTime(sessionphase.SessionEndParse, tcmd.ParseEnd)
 
@@ -1937,6 +1942,7 @@ func (ex *connExecutor) execCmd() error {
 		// ExecPortal is handled like ExecStmt, except that the placeholder info
 		// is taken from the portal.
 		ex.phaseTimes.SetSessionPhaseTime(sessionphase.SessionQueryReceived, tcmd.TimeReceived)
+		ex.phaseTimes.SetSessionPhaseTime(sessionphase.SessionQueryReceivedCPUTime, receivedCPUTime)
 		// When parsing has been done earlier, via a separate parse
 		// message, it is not any more part of the statistics collected
 		// for this execution. In that case, we simply report that

pkg/sql/conn_executor_exec.go

@@ -53,6 +53,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/contextutil"
 	"github.com/cockroachdb/cockroach/pkg/util/duration"
 	"github.com/cockroachdb/cockroach/pkg/util/fsm"
+	"github.com/cockroachdb/cockroach/pkg/util/grunning"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/log/eventpb"
@@ -1196,6 +1197,9 @@ func (ex *connExecutor) dispatchToExecutionEngine(
 	}
 	ex.sessionTracing.TraceExecEnd(ctx, res.Err(), res.RowsAffected())
 	ex.statsCollector.PhaseTimes().SetSessionPhaseTime(sessionphase.PlannerEndExecStmt, timeutil.Now())
+	cpuNanos := grunning.Time().Nanoseconds()
+	fmt.Println("planner end exec stmt nanos", cpuNanos)
+	ex.statsCollector.PhaseTimes().SetSessionPhaseTime(sessionphase.PlannerEndExecStmtCpuTime, time.Unix(0, cpuNanos))
 
 	ex.extraTxnState.rowsRead += stats.rowsRead
 	ex.extraTxnState.bytesRead += stats.bytesRead
@@ -1955,6 +1959,7 @@ var showQueryStatsFns = map[tree.Name]func(*sessionphase.Times) time.Duration{
 	// Since the last query has already finished, it is safe to retrieve its
 	// total service latency.
 	"service_latency":          func(phaseTimes *sessionphase.Times) time.Duration { return phaseTimes.GetServiceLatencyTotal() },
+	"cpu_time":                 func(phaseTimes *sessionphase.Times) time.Duration { return phaseTimes.GetServiceCPUTimeTotal() },
 	"post_commit_jobs_latency": func(phaseTimes *sessionphase.Times) time.Duration { return phaseTimes.GetPostCommitJobsLatency() },
 }
 

pkg/sql/crdb_internal.go

@@ -1094,6 +1094,8 @@ CREATE TABLE crdb_internal.node_statement_statistics (
   service_lat_var     FLOAT NOT NULL,
   overhead_lat_avg    FLOAT NOT NULL,
   overhead_lat_var    FLOAT NOT NULL,
+  cpu_time_avg        FLOAT NOT NULL,
+  cpu_time_var        FLOAT NOT NULL,
   bytes_read_avg      FLOAT NOT NULL,
   bytes_read_var      FLOAT NOT NULL,
   rows_read_avg       FLOAT NOT NULL,
@@ -1197,6 +1199,8 @@ CREATE TABLE crdb_internal.node_statement_statistics (
 				tree.NewDFloat(tree.DFloat(stats.Stats.ServiceLat.GetVariance(stats.Stats.Count))),  // service_lat_var
 				tree.NewDFloat(tree.DFloat(stats.Stats.OverheadLat.Mean)),                           // overhead_lat_avg
 				tree.NewDFloat(tree.DFloat(stats.Stats.OverheadLat.GetVariance(stats.Stats.Count))), // overhead_lat_var
+				tree.NewDFloat(tree.DFloat(stats.Stats.CpuTime.Mean)),                               // cpu_time_avg
+				tree.NewDFloat(tree.DFloat(stats.Stats.CpuTime.GetVariance(stats.Stats.Count))),     // cpu_time_var
 				tree.NewDFloat(tree.DFloat(stats.Stats.BytesRead.Mean)),                             // bytes_read_avg
 				tree.NewDFloat(tree.DFloat(stats.Stats.BytesRead.GetVariance(stats.Stats.Count))),   // bytes_read_var
 				tree.NewDFloat(tree.DFloat(stats.Stats.RowsRead.Mean)),                              // rows_read_avg

pkg/sql/executor_statement_metrics.go

@@ -131,6 +131,7 @@ func (ex *connExecutor) recordStatementSummary(
 	// We want to exclude any overhead to reduce possible confusion.
 	svcLatRaw := phaseTimes.GetServiceLatencyNoOverhead()
 	svcLat := svcLatRaw.Seconds()
+	cpuTime := phaseTimes.GetServiceCPUTimeTotal().Seconds()
 
 	// processing latency: contributing towards SQL results.
 	processingLat := parseLat + planLat + runLat
@@ -192,6 +193,7 @@ func (ex *connExecutor) recordStatementSummary(
 		RunLatency:           runLat,
 		ServiceLatency:       svcLat,
 		OverheadLatency:      execOverhead,
+		ServiceCPUTime:       cpuTime,
 		BytesRead:            stats.bytesRead,
 		RowsRead:             stats.rowsRead,
 		RowsWritten:          stats.rowsWritten,

pkg/sql/pgwire/BUILD.bazel

@@ -54,6 +54,7 @@ go_library(
         "//pkg/util/duration",
         "//pkg/util/envutil",
         "//pkg/util/errorutil/unimplemented",
+        "//pkg/util/grunning",
         "//pkg/util/humanizeutil",
         "//pkg/util/ipaddr",
         "//pkg/util/json",

pkg/sql/sessionphase/session_phase.go

@@ -33,6 +33,8 @@ const (
 
 	// Executor phases.
 
+	SessionQueryReceivedCPUTime
+
 	// SessionQueryReceived is the SessionPhase when a query is received.
 	SessionQueryReceived
 
@@ -54,6 +56,8 @@ const (
 	// PlannerEndExecStmt is the SessionPhase when execution ends.
 	PlannerEndExecStmt
 
+	PlannerEndExecStmtCpuTime
+
 	// SessionQueryServiced is the SessionPhase when a query is serviced.
 	// Note: we compute this even for empty queries or  "special" statements that
 	// have no execution, like SHOW TRANSACTION STATUS.
@@ -166,6 +170,13 @@ func (t *Times) GetServiceLatencyTotal() time.Duration {
 	return t.times[SessionQueryServiced].Sub(t.times[SessionQueryReceived])
 }
 
+// GetServiceCPUTimeTotal returns the total CPU time of serving a query
+// including any overhead like internal retries.
+// NOTE: SessionQueryServicedCPUTime phase must have been set.
+func (t *Times) GetServiceCPUTimeTotal() time.Duration {
+	return t.times[PlannerEndExecStmtCpuTime].Sub(t.times[SessionQueryReceivedCPUTime])
+}
+
 // GetRunLatency returns the time between a query execution starting and
 // ending.
 func (t *Times) GetRunLatency() time.Duration {

pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil/json_impl.go

@@ -286,6 +286,7 @@ func (s *innerStmtStats) jsonFields() jsonFields {
 		{"runLat", (*numericStats)(&s.RunLat)},
 		{"svcLat", (*numericStats)(&s.ServiceLat)},
 		{"ovhLat", (*numericStats)(&s.OverheadLat)},
+		{"cpuTime", (*numericStats)(&s.CpuTime)},
 		{"bytesRead", (*numericStats)(&s.BytesRead)},
 		{"rowsRead", (*numericStats)(&s.RowsRead)},
 		{"rowsWritten", (*numericStats)(&s.RowsWritten)},

pkg/sql/sqlstats/ssmemstorage/ss_mem_writer.go

@@ -129,6 +129,7 @@ func (s *Container) RecordStatement(
 	stats.mu.data.RunLat.Record(stats.mu.data.Count, value.RunLatency)
 	stats.mu.data.ServiceLat.Record(stats.mu.data.Count, value.ServiceLatency)
 	stats.mu.data.OverheadLat.Record(stats.mu.data.Count, value.OverheadLatency)
+	stats.mu.data.CpuTime.Record(stats.mu.data.Count, value.ServiceCPUTime)
 	stats.mu.data.BytesRead.Record(stats.mu.data.Count, float64(value.BytesRead))
 	stats.mu.data.RowsRead.Record(stats.mu.data.Count, float64(value.RowsRead))
 	stats.mu.data.RowsWritten.Record(stats.mu.data.Count, float64(value.RowsWritten))

pkg/sql/sqlstats/ssprovider.go

@@ -201,6 +201,7 @@ type RecordedStmtStats struct {
 	RunLatency           float64
 	ServiceLatency       float64
 	OverheadLatency      float64
+	ServiceCPUTime       float64
 	BytesRead            int64
 	RowsRead             int64
 	RowsWritten          int64