关于golang:聊聊dkron的fsm

序

本文次要钻研一下 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 办法。

doc

• dkron