Apply defines entry for Leader node.
(ctx context.Context, req interface{})
| 280 | |
| 281 | // Apply defines entry for Leader node. |
| 282 | func (r *Runtime) Apply(ctx context.Context, req interface{}) (result interface{}, logIndex uint64, err error) { |
| 283 | if atomic.LoadUint32(&r.started) != 1 { |
| 284 | err = kt.ErrStopped |
| 285 | return |
| 286 | } |
| 287 | |
| 288 | ctx, task := trace.NewTask(ctx, "Kayak.Apply") |
| 289 | defer task.End() |
| 290 | |
| 291 | tm := timer.NewTimer() |
| 292 | |
| 293 | defer func() { |
| 294 | log.WithField("r", logIndex). |
| 295 | WithFields(tm.ToLogFields()). |
| 296 | WithError(err). |
| 297 | Debug("kayak leader apply") |
| 298 | }() |
| 299 | |
| 300 | r.peersLock.RLock() |
| 301 | defer r.peersLock.RUnlock() |
| 302 | |
| 303 | tm.Add("peers_lock") |
| 304 | |
| 305 | if r.role != proto.Leader { |
| 306 | // not leader |
| 307 | err = kt.ErrNotLeader |
| 308 | return |
| 309 | } |
| 310 | |
| 311 | // prepare |
| 312 | prepareLog, err := r.doLeaderPrepare(ctx, tm, req) |
| 313 | |
| 314 | if prepareLog != nil { |
| 315 | defer r.markPrepareFinished(ctx, prepareLog.Index) |
| 316 | } |
| 317 | |
| 318 | if err == nil { |
| 319 | // commit |
| 320 | return r.doLeaderCommit(ctx, tm, prepareLog, req) |
| 321 | } |
| 322 | |
| 323 | // rollback |
| 324 | if prepareLog != nil { |
| 325 | r.doLeaderRollback(ctx, tm, prepareLog) |
| 326 | } |
| 327 | |
| 328 | return |
| 329 | } |
| 330 | |
| 331 | // Fetch defines entry for missing log startFetch. |
| 332 | func (r *Runtime) Fetch(ctx context.Context, index uint64) (l *kt.Log, err error) { |