序
本文主要研究一下dkron的fsm
MessageType
// MessageType is the type to encode FSM commands.
type MessageType uint8
const (
// SetJobType is the command used to store a job in the store.
SetJobType MessageType = iota
// DeleteJobType is the command used to delete a Job from the store.
DeleteJobType
// SetExecutionType is the command used to store an Execution to the store.
SetExecutionType
// DeleteExecutionsType is the command used to delete executions from the store.
DeleteExecutionsType
// ExecutionDoneType is the command to perform the logic needed once an exeuction
// is done.
ExecutionDoneType
)
复制代码
MessageType可以分为SetJobType、DeleteJobType、SetExecutionType、DeleteExecutionsType、ExecutionDoneType
dkronFSM
type dkronFSM struct {
store Storage
// proAppliers holds the set of pro only LogAppliers
proAppliers LogAppliers
}
// NewFSM is used to construct a new FSM with a blank state
func newFSM(store Storage, logAppliers LogAppliers) *dkronFSM {
return &dkronFSM{
store: store,
proAppliers: logAppliers,
}
}
// Apply applies a Raft log entry to the key-value store.
func (d *dkronFSM) Apply(l *raft.Log) interface{} {
buf := l.Data
msgType := MessageType(buf[0])
log.WithField("command", msgType).Debug("fsm: received command")
switch msgType {
case SetJobType:
return d.applySetJob(buf[1:])
case DeleteJobType:
return d.applyDeleteJob(buf[1:])
case ExecutionDoneType:
return d.applyExecutionDone(buf[1:])
case SetExecutionType:
return d.applySetExecution(buf[1:])
}
// Check enterprise only message types.
if applier, ok := d.proAppliers[msgType]; ok {
return applier(buf[1:], l.Index)
}
return nil
}
func (d *dkronFSM) applySetJob(buf []byte) interface{} {
var pj dkronpb.Job
if err := proto.Unmarshal(buf, &pj); err != nil {
return err
}
job := NewJobFromProto(&pj)
if err := d.store.SetJob(job, false); err != nil {
return err
}
return nil
}
func (d *dkronFSM) applyDeleteJob(buf []byte) interface{} {
var djr dkronpb.DeleteJobRequest
if err := proto.Unmarshal(buf, &djr); err != nil {
return err
}
job, err := d.store.DeleteJob(djr.GetJobName())
if err != nil {
return err
}
return job
}
func (d *dkronFSM) applyExecutionDone(buf []byte) interface{} {
var execDoneReq dkronpb.ExecutionDoneRequest
if err := proto.Unmarshal(buf, &execDoneReq); err != nil {
return err
}
execution := NewExecutionFromProto(execDoneReq.Execution)
log.WithField("execution", execution.Key()).
WithField("output", string(execution.Output)).
Debug("fsm: Setting execution")
_, err := d.store.SetExecutionDone(execution)
return err
}
func (d *dkronFSM) applySetExecution(buf []byte) interface{} {
var pbex dkronpb.Execution
if err := proto.Unmarshal(buf, &pbex); err != nil {
return err
}
execution := NewExecutionFromProto(&pbex)
key, err := d.store.SetExecution(execution)
if err != nil {
return err
}
return key
}
// Snapshot returns a snapshot of the key-value store. We wrap
// the things we need in dkronSnapshot and then send that over to Persist.
// Persist encodes the needed data from dkronSnapshot and transport it to
// Restore where the necessary data is replicated into the finite state machine.
// This allows the consensus algorithm to truncate the replicated log.
func (d *dkronFSM) Snapshot() (raft.FSMSnapshot, error) {
return &dkronSnapshot{store: d.store}, nil
}
// Restore stores the key-value store to a previous state.
func (d *dkronFSM) Restore(r io.ReadCloser) error {
defer r.Close()
return d.store.Restore(r)
}
// LogApplier is the definition of a function that can apply a Raft log
type LogApplier func(buf []byte, index uint64) interface{}
复制代码
dkronFSM定义了store、proAppliers属性;Apply方法将raft的log保存到KV存储中,具体分不同msgType做不同处理;最后根据msgType查找LogAppliers
小结
dkron的FSM根据不同msgType做不同处理,具体有applySetJob、applyDeleteJob、applyExecutionDone、applySetExecution方法。