etcdserver/server.go
func (s *EtcdServer) applyEntryNormal(e *raftpb.Entry) {
	shouldApplyV3 := false
	if e.Index > s.consistIndex.ConsistentIndex() {
		// set the consistent index of current executing entry
		s.consistIndex.setConsistentIndex(e.Index)
		shouldApplyV3 = true
	}

	// raft state machine may generate noop entry when leader confirmation.
	// skip it in advance to avoid some potential bug in the future
	if len(e.Data) == 0 {
		select {
		case s.forceVersionC <- struct{}{}:
		default:
		}
		// promote lessor when the local member is leader and finished
		// applying all entries from the last term.
		if s.isLeader() {
			s.lessor.Promote(s.Cfg.electionTimeout())
		}
		return
	}

	var raftReq pb.InternalRaftRequest
	if !pbutil.MaybeUnmarshal(&raftReq, e.Data) { // backward compatible
		var r pb.Request
		rp := &r
		pbutil.MustUnmarshal(rp, e.Data)
		s.w.Trigger(r.ID, s.applyV2Request((*RequestV2)(rp)))
		return
	}
	if raftReq.V2 != nil {
		req := (*RequestV2)(raftReq.V2)
		s.w.Trigger(req.ID, s.applyV2Request(req))
		return
	}

	// do not re-apply applied entries.
	if !shouldApplyV3 {
		return
	}

	id := raftReq.ID
	if id == 0 {
		id = raftReq.Header.ID
	}

	var ar *applyResult
	needResult := s.w.IsRegistered(id)
	if needResult || !noSideEffect(&raftReq) {
		if !needResult && raftReq.Txn != nil {
			removeNeedlessRangeReqs(raftReq.Txn)
		}
		ar = s.applyV3.Apply(&raftReq)
	}

	if ar == nil {
		return
	}

	if ar.err != ErrNoSpace || len(s.alarmStore.Get(pb.AlarmType_NOSPACE)) > 0 {
		s.w.Trigger(id, ar)
		return
	}

	if lg := s.getLogger(); lg != nil {
		lg.Warn(
			"message exceeded backend quota; raising alarm",
			zap.Int64("quota-size-bytes", s.Cfg.QuotaBackendBytes),
			zap.String("quota-size", humanize.Bytes(uint64(s.Cfg.QuotaBackendBytes))),
			zap.Error(ar.err),
		)
	} else {
		plog.Errorf("applying raft message exceeded backend quota")
	}

	s.goAttach(func() {
		a := &pb.AlarmRequest{
			MemberID: uint64(s.ID()),
			Action:   pb.AlarmRequest_ACTIVATE,
			Alarm:    pb.AlarmType_NOSPACE,
		}
		s.raftRequest(s.ctx, pb.InternalRaftRequest{Alarm: a})
		s.w.Trigger(id, ar)
	})
}
func (s *EtcdServer) applyEntryNormal(e *raftpb.Entry) {
	shouldApplyV3 := false
	if e.Index > s.consistIndex.ConsistentIndex() {
		// set the consistent index of current executing entry
		s.consistIndex.setConsistentIndex(e.Index)
		shouldApplyV3 = true
	}

	// raft state machine may generate noop entry when leader confirmation.
	// skip it in advance to avoid some potential bug in the future
	if len(e.Data) == 0 {
		select {
		case s.forceVersionC <- struct{}{}:
		default:
		}
		// promote lessor when the local member is leader and finished
		// applying all entries from the last term.
		if s.isLeader() {
			s.lessor.Promote(s.Cfg.electionTimeout())
		}
		return
	}

	var raftReq pb.InternalRaftRequest
	if !pbutil.MaybeUnmarshal(&raftReq, e.Data) { // backward compatible
		var r pb.Request
		rp := &r
		pbutil.MustUnmarshal(rp, e.Data)
		s.w.Trigger(r.ID, s.applyV2Request((*RequestV2)(rp)))
		return
	}
	if raftReq.V2 != nil {
		req := (*RequestV2)(raftReq.V2)
		s.w.Trigger(req.ID, s.applyV2Request(req))
		return
	}

	// do not re-apply applied entries.
	if !shouldApplyV3 {
		return
	}

	id := raftReq.ID
	if id == 0 {
		id = raftReq.Header.ID
	}

	var ar *applyResult
	needResult := s.w.IsRegistered(id)
	if needResult || !noSideEffect(&raftReq) {
		if !needResult && raftReq.Txn != nil {
			removeNeedlessRangeReqs(raftReq.Txn)
		}
		ar = s.applyV3.Apply(&raftReq)
	}

	if ar == nil {
		return
	}

	if ar.err != ErrNoSpace || len(s.alarmStore.Get(pb.AlarmType_NOSPACE)) > 0 {
		s.w.Trigger(id, ar)
		return
	}

	if lg := s.getLogger(); lg != nil {
		lg.Warn(
			"message exceeded backend quota; raising alarm",
			zap.Int64("quota-size-bytes", s.Cfg.QuotaBackendBytes),
			zap.String("quota-size", humanize.Bytes(uint64(s.Cfg.QuotaBackendBytes))),
			zap.Error(ar.err),
		)
	} else {
		plog.Errorf("applying raft message exceeded backend quota")
	}

	s.goAttach(func() {
		a := &pb.AlarmRequest{
			MemberID: uint64(s.ID()),
			Action:   pb.AlarmRequest_ACTIVATE,
			Alarm:    pb.AlarmType_NOSPACE,
		}
		s.raftRequest(s.ctx, pb.InternalRaftRequest{Alarm: a})
		s.w.Trigger(id, ar)
	})
}
func ( s * EtcdServer ) applyEntryNormal ( e * raftpb . Entry ) { shouldApplyV3 := false if e . Index > s . consistIndex . ConsistentIndex ( ) { // set the consistent index of current executing entry s . consistIndex . setConsistentIndex ( e . Index ) shouldApplyV3 = true } // raft state machine may generate noop entry when leader confirmation. // skip it in advance to avoid some potential bug in the future if len ( e . Data ) == 0 { select { case s . forceVersionC <- struct { } { } : default : } // promote lessor when the local member is leader and finished // applying all entries from the last term. if s . isLeader ( ) { s . lessor . Promote ( s . Cfg . electionTimeout ( ) ) } return } var raftReq pb . InternalRaftRequest if ! pbutil . MaybeUnmarshal ( & raftReq , e . Data ) { // backward compatible var r pb . Request rp := & r pbutil . MustUnmarshal ( rp , e . Data ) s . w . Trigger ( r . ID , s . applyV2Request ( ( * RequestV2 ) ( rp ) ) ) return } if raftReq . V2 != nil { req := ( * RequestV2 ) ( raftReq . V2 ) s . w . Trigger ( req . ID , s . applyV2Request ( req ) ) return } // do not re-apply applied entries. if ! shouldApplyV3 { return } id := raftReq . ID if id == 0 { id = raftReq . Header . ID } var ar * applyResult needResult := s . w . IsRegistered ( id ) if needResult || ! noSideEffect ( & raftReq ) { if ! needResult && raftReq . Txn != nil { removeNeedlessRangeReqs ( raftReq . Txn ) } ar = s . applyV3 . Apply ( & raftReq ) } if ar == nil { return } if ar . err != ErrNoSpace || len ( s . alarmStore . Get ( pb . AlarmType_NOSPACE ) ) > 0 { s . w . Trigger ( id , ar ) return } if lg := s . getLogger ( ) ; lg != nil { lg . Warn ( " " , zap . Int64 ( " " , s . Cfg . QuotaBackendBytes ) , zap . String ( " " , humanize . Bytes ( uint64 ( s . Cfg . QuotaBackendBytes ) ) ) , zap . Error ( ar . err ) , ) } else { plog . Errorf ( " " ) } s . goAttach ( func ( ) { a := & pb . AlarmRequest { MemberID : uint64 ( s . ID ( ) ) , Action : pb . AlarmRequest_ACTIVATE , Alarm : pb . AlarmType_NOSPACE , } s . raftRequest ( s . ctx , pb . InternalRaftRequest { Alarm : a } ) s . w . Trigger ( id , ar ) } ) }
// applyEntryNormal apples an EntryNormal type raftpb request to the EtcdServer
applyEntryNormal apples an EntryNormal type raftpb request to the EtcdServer