acbd7fda60
This adds a new "crdt" consensus component using go-ds-crdt. This implies several refactors to fully make cluster consensus-component independent: * Delete mapstate and fully adopt dsstate (after people have migrated). * Return errors from state methods rather than ignoring them. * Add a new "datastore" modules so that we can configure datastores in the main configuration like other components. * Let the consensus components fully define the "state.State". Thus, they do not receive the state, they receive the storage where we put the state (a go-datastore). * Allow to customize how the monitor component obtains Peers() (the current peerset), including avoiding using the current peerset. At the moment the crdt consensus uses the monitoring component to define the current peerset. Therefore the monitor component cannot rely on the consensus component to produce a peerset. * Re-factor/re-implementation of "ipfs-cluster-service state" operations. Includes the dissapearance of the "migrate" one. The CRDT consensus component defines creates a crdt-datastore (with ipfs-lite) and uses it to intitialize a dssate. Thus the crdt-store is elegantly wrapped. Any modifications to the state get automatically replicated to other peers. We store all the CRDT DAG blocks in the local datastore. The consensus components only expose a ReadOnly state, as any modifications to the shared state should happen through them. DHT and PubSub facilities must now be created outside of Cluster and passed in so they can be re-used by different components.
110 lines
2.4 KiB
Go
110 lines
2.4 KiB
Go
package raft
|
|
|
|
import (
|
|
"log"
|
|
"strings"
|
|
"time"
|
|
|
|
logging "github.com/ipfs/go-log"
|
|
)
|
|
|
|
const (
|
|
debug = iota
|
|
info
|
|
warn
|
|
err
|
|
)
|
|
|
|
const repeatPoolSize = 10
|
|
const repeatReset = time.Minute
|
|
|
|
// This provides a custom logger for Raft which intercepts Raft log messages
|
|
// and rewrites us to our own logger (for "raft" facility).
|
|
type logForwarder struct {
|
|
lastMsgs map[int][]string
|
|
lastTip map[int]time.Time
|
|
}
|
|
|
|
var raftStdLogger = log.New(&logForwarder{}, "", 0)
|
|
var raftLogger = logging.Logger("raftlib")
|
|
|
|
// Write forwards to our go-log logger.
|
|
// According to https://golang.org/pkg/log/#Logger.Output
|
|
// it is called per line.
|
|
func (fw *logForwarder) Write(p []byte) (n int, e error) {
|
|
t := strings.TrimSuffix(string(p), "\n")
|
|
|
|
switch {
|
|
case strings.Contains(t, "[DEBUG]"):
|
|
if !fw.repeated(debug, t) {
|
|
fw.log(debug, strings.TrimPrefix(t, "[DEBUG] raft: "))
|
|
}
|
|
case strings.Contains(t, "[WARN]"):
|
|
if !fw.repeated(warn, t) {
|
|
fw.log(warn, strings.TrimPrefix(t, "[WARN] raft: "))
|
|
}
|
|
case strings.Contains(t, "[ERR]"):
|
|
if !fw.repeated(err, t) {
|
|
fw.log(err, strings.TrimPrefix(t, "[ERR] raft: "))
|
|
}
|
|
case strings.Contains(t, "[INFO]"):
|
|
if !fw.repeated(info, t) {
|
|
fw.log(info, strings.TrimPrefix(t, "[INFO] raft: "))
|
|
}
|
|
default:
|
|
fw.log(debug, t)
|
|
}
|
|
return len(p), nil
|
|
}
|
|
|
|
func (fw *logForwarder) repeated(t int, msg string) bool {
|
|
if fw.lastMsgs == nil {
|
|
fw.lastMsgs = make(map[int][]string)
|
|
fw.lastTip = make(map[int]time.Time)
|
|
}
|
|
|
|
// We haven't tipped about repeated log messages
|
|
// in a while, do it and forget the list
|
|
if time.Now().After(fw.lastTip[t].Add(repeatReset)) {
|
|
fw.lastTip[t] = time.Now()
|
|
fw.lastMsgs[t] = nil
|
|
fw.log(t, "NOTICE: Some RAFT log messages repeat and will only be logged once")
|
|
}
|
|
|
|
var found string
|
|
|
|
// Do we know about this message
|
|
for _, lmsg := range fw.lastMsgs[t] {
|
|
if lmsg == msg {
|
|
found = lmsg
|
|
break
|
|
}
|
|
}
|
|
|
|
if found == "" { // new message. Add to slice.
|
|
if len(fw.lastMsgs[t]) >= repeatPoolSize { // drop oldest
|
|
fw.lastMsgs[t] = fw.lastMsgs[t][1:]
|
|
}
|
|
fw.lastMsgs[t] = append(fw.lastMsgs[t], msg)
|
|
return false // not-repeated
|
|
}
|
|
|
|
// repeated, don't log
|
|
return true
|
|
}
|
|
|
|
func (fw *logForwarder) log(t int, msg string) {
|
|
switch t {
|
|
case debug:
|
|
raftLogger.Debug(msg)
|
|
case info:
|
|
raftLogger.Info(msg)
|
|
case warn:
|
|
raftLogger.Warning(msg)
|
|
case err:
|
|
raftLogger.Error(msg)
|
|
default:
|
|
raftLogger.Debug(msg)
|
|
}
|
|
}
|