Statistics for reserved connections

go/vt/vttablet/tabletserver/query_executor.go

@@ -68,7 +68,6 @@ var sequenceFields = []*querypb.Field{
 
 // Execute performs a non-streaming query execution.
 func (qre *QueryExecutor) Execute() (reply *sqltypes.Result, err error) {
- qre.logStats.TransactionID = qre.connID
  planName := qre.plan.PlanID.String()
  qre.logStats.PlanType = planName
  defer func(start time.Time) {

go/vt/vttablet/tabletserver/querylogz.go

@@ -52,6 +52,7 @@ var (
  <th>RowsAffected</th>
  <th>Response Size</th>
  <th>Transaction ID</th>
+ <th>Reserved ID</th>
  <th>Error</th>
  </tr>
  </thead>
@@ -80,6 +81,7 @@ var (
  <td>{{.RowsAffected}}</td>
  <td>{{.SizeOfResponse}}</td>
  <td>{{.TransactionID}}</td>
+ <td>{{.ReservedID}}</td>
  <td>{{.ErrorStr}}</td>
  </tr>
  `))

go/vt/vttablet/tabletserver/querylogz_test.go

@@ -55,6 +55,7 @@ func TestQuerylogzHandler(t *testing.T) {
  logStats.MysqlResponseTime = 1 * time.Millisecond
  logStats.WaitingForConnection = 10 * time.Nanosecond
  logStats.TransactionID = 131
+ logStats.ReservedID = 313
  logStats.Ctx = callerid.NewContext(
  context.Background(),
  callerid.NewEffectiveCallerID("effective-caller", "component", "subcomponent"),
@@ -80,6 +81,7 @@ func TestQuerylogzHandler(t *testing.T) {
  `<td>1000</td>`,
  `<td>0</td>`,
  `<td>131</td>`,
+ `<td>313</td>`,
  `<td></td>`,
  }
  logStats.EndTime = logStats.StartTime.Add(1 * time.Millisecond)
@@ -110,6 +112,7 @@ func TestQuerylogzHandler(t *testing.T) {
  `<td>1000</td>`,
  `<td>0</td>`,
  `<td>131</td>`,
+ `<td>313</td>`,
  `<td></td>`,
  }
  logStats.EndTime = logStats.StartTime.Add(20 * time.Millisecond)
@@ -140,6 +143,7 @@ func TestQuerylogzHandler(t *testing.T) {
  `<td>1000</td>`,
  `<td>0</td>`,
  `<td>131</td>`,
+ `<td>313</td>`,
  `<td></td>`,
  }
  logStats.EndTime = logStats.StartTime.Add(500 * time.Millisecond)

go/vt/vttablet/tabletserver/stateful_connection.go

@@ -18,6 +18,13 @@ package tabletserver
 
 import (
  "fmt"
+ "time"
+
+ "vitess.io/vitess/go/vt/log"
+
+ "vitess.io/vitess/go/vt/callerid"
+ querypb "vitess.io/vitess/go/vt/proto/query"
+ "vitess.io/vitess/go/vt/servenv"
 
  "vitess.io/vitess/go/mysql"
  "vitess.io/vitess/go/sqltypes"
@@ -41,10 +48,19 @@ type StatefulConnection struct {
  ConnID tx.ConnID
  env tabletenv.Env
  txProps *tx.Properties
+ reservedProps *Properties
  tainted bool
  enforceTimeout bool
 }
 
+//Properties contains meta information about the connection
+type Properties struct {
+ EffectiveCaller *vtrpcpb.CallerID
+ ImmediateCaller *querypb.VTGateCallerID
+ StartTime time.Time
+ Stats *servenv.TimingsWrapper
+}
+
 // Close closes the underlying connection. When the connection is Unblocked, it will be Released
 func (sc *StatefulConnection) Close() {
  if sc.dbConn != nil {
@@ -124,6 +140,7 @@ func (sc *StatefulConnection) Releasef(reasonFormat string, a ...interface{}) {
  sc.pool.unregister(sc.ConnID, fmt.Sprintf(reasonFormat, a...))
  sc.dbConn.Recycle()
  sc.dbConn = nil
+ sc.logReservedConn()
 }
 
 //Renew the existing connection with new connection id.
@@ -171,15 +188,71 @@ func (sc *StatefulConnection) Stats() *tabletenv.Stats {
 }
 
 //Taint taints the existing connection.
-func (sc *StatefulConnection) Taint() {
+func (sc *StatefulConnection) Taint(ctx context.Context, stats *servenv.TimingsWrapper) error {
+ if sc.dbConn == nil {
+ return vterrors.New(vtrpcpb.Code_FAILED_PRECONDITION, "connection is closed")
+ }
+ if sc.tainted {
+ return vterrors.New(vtrpcpb.Code_FAILED_PRECONDITION, "connection is already reserved")
+ }
+ immediateCaller := callerid.ImmediateCallerIDFromContext(ctx)
+ effectiveCaller := callerid.EffectiveCallerIDFromContext(ctx)
+
  sc.tainted = true
- // if we don't have an active dbConn, we can silently ignore this request
- if sc.dbConn != nil {
- sc.dbConn.Taint()
+ sc.reservedProps = &Properties{
+ EffectiveCaller: effectiveCaller,
+ ImmediateCaller: immediateCaller,
+ StartTime: time.Now(),
+ Stats: stats,
  }
+ sc.dbConn.Taint()
+ sc.Stats().UserActiveReservedCount.Add(sc.getUsername(), 1)
+ return nil
 }
 
 //IsTainted tells us whether this connection is tainted
 func (sc *StatefulConnection) IsTainted() bool {
  return sc.tainted
 }
+
+//LogTransaction logs transaction related stats
+func (sc *StatefulConnection) LogTransaction(reason tx.ReleaseReason) {
+ if sc.txProps == nil {
+ return //Nothing to log as no transaction exists on this connection.
+ }
+ sc.txProps.Conclusion = reason.Name()
+ sc.txProps.EndTime = time.Now()
+
+ username := callerid.GetPrincipal(sc.txProps.EffectiveCaller)
+ if username == "" {
+ username = callerid.GetUsername(sc.txProps.ImmediateCaller)
+ }
+ duration := sc.txProps.EndTime.Sub(sc.txProps.StartTime)
+ sc.Stats().UserTransactionCount.Add([]string{username, reason.Name()}, 1)
+ sc.Stats().UserTransactionTimesNs.Add([]string{username, reason.Name()}, int64(duration))
+ sc.txProps.Stats.Add(reason.Name(), duration)
+ if sc.txProps.LogToFile {
+ log.Infof("Logged transaction: %s", sc.String())
+ }
+ tabletenv.TxLogger.Send(sc)
+}
+
+//logReservedConn logs reserved connection related stats.
+func (sc *StatefulConnection) logReservedConn() {
+ if sc.reservedProps == nil {
+ return //Nothing to log as this connection is not reserved.
+ }
+ duration := time.Since(sc.reservedProps.StartTime)
+ username := sc.getUsername()
+ sc.Stats().UserActiveReservedCount.Add(username, -1)
+ sc.Stats().UserReservedCount.Add(username, 1)
+ sc.Stats().UserReservedTimesNs.Add(username, int64(duration))
+}
+
+func (sc *StatefulConnection) getUsername() string {
+ username := callerid.GetPrincipal(sc.reservedProps.EffectiveCaller)
+ if username != "" {
+ return username
+ }
+ return callerid.GetUsername(sc.reservedProps.ImmediateCaller)
+}

go/vt/vttablet/tabletserver/tabletenv/logstats.go

@@ -59,6 +59,7 @@ type LogStats struct {
  QuerySources byte
  Rows [][]sqltypes.Value
  TransactionID int64
+ ReservedID int64
  Error error
 }
 

go/vt/vttablet/tabletserver/tabletenv/stats.go

@@ -43,6 +43,10 @@ type Stats struct {
  TableaclAllowed *stats.CountersWithMultiLabels // Number of allows
  TableaclDenied *stats.CountersWithMultiLabels // Number of denials
  TableaclPseudoDenied *stats.CountersWithMultiLabels // Number of pseudo denials
+
+ UserActiveReservedCount *stats.CountersWithSingleLabel // Per CallerID active reserved connection counts
+ UserReservedCount *stats.CountersWithSingleLabel // Per CallerID reserved connection counts
+ UserReservedTimesNs *stats.CountersWithSingleLabel // Per CallerID reserved connection duration
 }
 
 // NewStats instantiates a new set of stats scoped by exporter.
@@ -51,7 +55,7 @@ func NewStats(exporter *servenv.Exporter) *Stats {
  MySQLTimings: exporter.NewTimings("Mysql", "MySQl query time", "operation"),
  QueryTimings: exporter.NewTimings("Queries", "MySQL query timings", "plan_type"),
  WaitTimings: exporter.NewTimings("Waits", "Wait operations", "type"),
- KillCounters: exporter.NewCountersWithSingleLabel("Kills", "Number of connections being killed", "query_type", "Transactions", "Queries"),
+ KillCounters: exporter.NewCountersWithSingleLabel("Kills", "Number of connections being killed", "query_type", "Transactions", "Queries", "ReservedConnection"),
  ErrorCounters: exporter.NewCountersWithSingleLabel(
  "Errors",
  "Critical errors",
@@ -85,6 +89,10 @@ func NewStats(exporter *servenv.Exporter) *Stats {
  TableaclAllowed: exporter.NewCountersWithMultiLabels("TableACLAllowed", "ACL acceptances", []string{"TableName", "TableGroup", "PlanID", "Username"}),
  TableaclDenied: exporter.NewCountersWithMultiLabels("TableACLDenied", "ACL denials", []string{"TableName", "TableGroup", "PlanID", "Username"}),
  TableaclPseudoDenied: exporter.NewCountersWithMultiLabels("TableACLPseudoDenied", "ACL pseudodenials", []string{"TableName", "TableGroup", "PlanID", "Username"}),
+
+ UserActiveReservedCount: exporter.NewCountersWithSingleLabel("UserActiveReservedCount", "active reserved connection for each CallerID", "CallerID"),
+ UserReservedCount: exporter.NewCountersWithSingleLabel("UserReservedCount", "reserved connection received for each CallerID", "CallerID"),
+ UserReservedTimesNs: exporter.NewCountersWithSingleLabel("UserReservedTimesNs", "Total reserved connection latency for each CallerID", "CallerID"),
  }
  stats.QPSRates = exporter.NewRates("QPS", stats.QueryTimings, 15*60/5, 5*time.Second)
  return stats

go/vt/vttablet/tabletserver/tabletserver.go

@@ -763,6 +763,7 @@ func (tsv *TabletServer) begin(ctx context.Context, target *querypb.Target, rese
  var beginSQL string
  transactionID, beginSQL, err = tsv.te.Begin(ctx, reservedID, options)
  logStats.TransactionID = transactionID
+ logStats.ReservedID = reservedID
 
  // Record the actual statements that were executed in the logStats.
  // If nothing was actually executed, don't count the operation in
@@ -792,6 +793,10 @@ func (tsv *TabletServer) Commit(ctx context.Context, target *querypb.Target, tra
 
  var commitSQL string
  newReservedID, commitSQL, err = tsv.te.Commit(ctx, transactionID)
+ if newReservedID > 0 {
+ // commit executed on old reserved id.
+ logStats.ReservedID = transactionID
+ }
 
  // If nothing was actually executed, don't count the operation in
  // the tablet metrics, and clear out the logStats Method so that
@@ -817,6 +822,10 @@ func (tsv *TabletServer) Rollback(ctx context.Context, target *querypb.Target, t
  defer tsv.stats.QueryTimings.Record("ROLLBACK", time.Now())
  logStats.TransactionID = transactionID
  newReservedID, err = tsv.te.Rollback(ctx, transactionID)
+ if newReservedID > 0 {
+ // rollback executed on old reserved id.
+ logStats.ReservedID = transactionID
+ }
  return err
  },
  )
@@ -993,7 +1002,8 @@ func (tsv *TabletServer) Execute(ctx context.Context, target *querypb.Target, sq
  if transactionID != 0 {
  connID = transactionID
  }
-
+ logStats.ReservedID = reservedID
+ logStats.TransactionID = transactionID
  qre := &QueryExecutor{
  query: query,
  marginComments: comments,
@@ -1389,12 +1399,13 @@ func (tsv *TabletServer) ReserveBeginExecute(ctx context.Context, target *queryp
  "ReserveBegin", "begin", bindVariables,
  target, options, false, /* allowOnShutdown */
  func(ctx context.Context, logStats *tabletenv.LogStats) error {
- defer tsv.stats.QueryTimings.Record("ReserveBegin", time.Now())
+ defer tsv.stats.QueryTimings.Record("RESERVE", time.Now())
  connID, err = tsv.te.ReserveBegin(ctx, options, preQueries)
  if err != nil {
  return err
  }
  logStats.TransactionID = connID
+ logStats.ReservedID = connID
  return nil
  },
  )
@@ -1417,12 +1428,13 @@ func (tsv *TabletServer) ReserveExecute(ctx context.Context, target *querypb.Tar
  "Reserve", "", bindVariables,
  target, options, false, /* allowOnShutdown */
  func(ctx context.Context, logStats *tabletenv.LogStats) error {
- defer tsv.stats.QueryTimings.Record("Reserve", time.Now())
+ defer tsv.stats.QueryTimings.Record("RESERVE", time.Now())
  connID, err = tsv.te.Reserve(ctx, options, transactionID, preQueries)
  if err != nil {
  return err
  }
- logStats.TransactionID = connID
+ logStats.TransactionID = transactionID
+ logStats.ReservedID = connID
  return nil
  },
  )
@@ -1446,13 +1458,13 @@ func (tsv *TabletServer) Release(ctx context.Context, target *querypb.Target, tr
  target, nil, true, /* allowOnShutdown */
  func(ctx context.Context, logStats *tabletenv.LogStats) error {
  defer tsv.stats.QueryTimings.Record("RELEASE", time.Now())
+ logStats.TransactionID = transactionID
+ logStats.ReservedID = reservedID
  if reservedID != 0 {
  //Release to close the underlying connection.
- logStats.TransactionID = reservedID
- return tsv.te.Release(ctx, reservedID)
+ return tsv.te.Release(reservedID)
  }
  // Rollback to cleanup the transaction before returning to the pool.
- logStats.TransactionID = transactionID
  _, err := tsv.te.Rollback(ctx, transactionID)
  return err
  },