AutoCommit commits to database after run a txn command
(cmd TxnCommand)
| 173 | |
| 174 | // AutoCommit commits to database after run a txn command |
| 175 | func AutoCommit(cmd TxnCommand) Command { |
| 176 | return func(ctx *Context) { |
| 177 | retry.Ensure(ctx, func() error { |
| 178 | mt := metrics.GetMetrics() |
| 179 | start := time.Now() |
| 180 | txn, err := ctx.Client.DB.Begin() |
| 181 | key := "" |
| 182 | if len(ctx.Args) > 0 { |
| 183 | key = ctx.Args[0] |
| 184 | if len(ctx.Args) > 1 { |
| 185 | mt.CommandArgsNumHistogramVec.WithLabelValues(ctx.Client.Namespace, ctx.Name).Observe(float64(len(ctx.Args)-1)) |
| 186 | } |
| 187 | } |
| 188 | cost := time.Since(start).Seconds() |
| 189 | zap.L().Debug("transation begin", zap.String("name", ctx.Name), zap.String("key", key), zap.Int64("cost(us)", int64(cost*1000000))) |
| 190 | mt.TxnBeginHistogramVec.WithLabelValues(ctx.Client.Namespace, ctx.Name).Observe(cost) |
| 191 | if err != nil { |
| 192 | mt.TxnFailuresCounterVec.WithLabelValues(ctx.Client.Namespace, ctx.Name).Inc() |
| 193 | resp.ReplyError(ctx.Out, "ERR "+err.Error()) |
| 194 | zap.L().Error("txn begin failed", |
| 195 | zap.Int64("clientid", ctx.Client.ID), |
| 196 | zap.String("command", ctx.Name), |
| 197 | zap.String("traceid", ctx.TraceID), |
| 198 | zap.Error(err)) |
| 199 | return err |
| 200 | } |
| 201 | |
| 202 | start = time.Now() |
| 203 | onCommit, err := cmd(ctx, txn) |
| 204 | cost = time.Since(start).Seconds() |
| 205 | zap.L().Debug("command done", zap.String("name", ctx.Name), zap.String("key", key), zap.Int64("cost(us)", int64(cost*1000000))) |
| 206 | mt.CommandFuncDoneHistogramVec.WithLabelValues(ctx.Client.Namespace, ctx.Name).Observe(cost) |
| 207 | if err != nil { |
| 208 | mt.TxnFailuresCounterVec.WithLabelValues(ctx.Client.Namespace, ctx.Name).Inc() |
| 209 | resp.ReplyError(ctx.Out, err.Error()) |
| 210 | txn.Rollback() |
| 211 | zap.L().Error("command process failed", |
| 212 | zap.Int64("clientid", ctx.Client.ID), |
| 213 | zap.String("command", ctx.Name), |
| 214 | zap.String("traceid", ctx.TraceID), |
| 215 | zap.Error(err)) |
| 216 | return err |
| 217 | } |
| 218 | |
| 219 | start = time.Now() |
| 220 | mtFunc := func() { |
| 221 | cost = time.Since(start).Seconds() |
| 222 | mt.TxnCommitHistogramVec.WithLabelValues(ctx.Client.Namespace, ctx.Name).Observe(cost) |
| 223 | } |
| 224 | if err := txn.Commit(ctx); err != nil { |
| 225 | txn.Rollback() |
| 226 | mt.TxnFailuresCounterVec.WithLabelValues(ctx.Client.Namespace, ctx.Name).Inc() |
| 227 | if db.IsRetryableError(err) { |
| 228 | mt.TxnRetriesCounterVec.WithLabelValues(ctx.Client.Namespace, ctx.Name).Inc() |
| 229 | mt.TxnConflictsCounterVec.WithLabelValues(ctx.Client.Namespace, ctx.Name).Inc() |
| 230 | mtFunc() |
| 231 | zap.L().Error("txn commit retry", |
| 232 | zap.Int64("clientid", ctx.Client.ID), |
no test coverage detected