688dfe3138
This is the OSS portion of enterprise PR 2242. This PR introduces a server-local implementation of the proxycfg.ServiceList interface, backed by streaming events and a local materializer.
335 lines
11 KiB
Go
335 lines
11 KiB
Go
package fsm
|
|
|
|
import (
|
|
"fmt"
|
|
"io"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/hashicorp/go-hclog"
|
|
"github.com/hashicorp/go-raftchunking"
|
|
"github.com/hashicorp/raft"
|
|
|
|
"github.com/hashicorp/consul-net-rpc/go-msgpack/codec"
|
|
|
|
"github.com/hashicorp/consul/agent/consul/state"
|
|
"github.com/hashicorp/consul/agent/consul/stream"
|
|
"github.com/hashicorp/consul/agent/structs"
|
|
"github.com/hashicorp/consul/logging"
|
|
)
|
|
|
|
// command is a command method on the FSM.
|
|
type command func(buf []byte, index uint64) interface{}
|
|
|
|
// unboundCommand is a command method on the FSM, not yet bound to an FSM
|
|
// instance.
|
|
type unboundCommand func(c *FSM, buf []byte, index uint64) interface{}
|
|
|
|
// commands is a map from message type to unbound command.
|
|
var commands map[structs.MessageType]unboundCommand
|
|
|
|
// registerCommand registers a new command with the FSM, which should be done
|
|
// at package init() time.
|
|
func registerCommand(msg structs.MessageType, fn unboundCommand) {
|
|
if commands == nil {
|
|
commands = make(map[structs.MessageType]unboundCommand)
|
|
}
|
|
if commands[msg] != nil {
|
|
panic(fmt.Errorf("Message %d is already registered", msg))
|
|
}
|
|
commands[msg] = fn
|
|
}
|
|
|
|
// FSM implements a finite state machine that is used
|
|
// along with Raft to provide strong consistency. We implement
|
|
// this outside the Server to avoid exposing this outside the package.
|
|
type FSM struct {
|
|
deps Deps
|
|
logger hclog.Logger
|
|
chunker *raftchunking.ChunkingFSM
|
|
|
|
// apply is built off the commands global and is used to route apply
|
|
// operations to their appropriate handlers.
|
|
apply map[structs.MessageType]command
|
|
|
|
// stateLock is only used to protect outside callers to State() from
|
|
// racing with Restore(), which is called by Raft (it puts in a totally
|
|
// new state store). Everything internal here is synchronized by the
|
|
// Raft side, so doesn't need to lock this.
|
|
stateLock sync.RWMutex
|
|
state *state.Store
|
|
|
|
publisher *stream.EventPublisher
|
|
}
|
|
|
|
// New is used to construct a new FSM with a blank state.
|
|
//
|
|
// Deprecated: use NewFromDeps.
|
|
func New(gc *state.TombstoneGC, logger hclog.Logger) (*FSM, error) {
|
|
newStateStore := func() *state.Store {
|
|
return state.NewStateStore(gc)
|
|
}
|
|
return NewFromDeps(Deps{Logger: logger, NewStateStore: newStateStore}), nil
|
|
}
|
|
|
|
// Deps are dependencies used to construct the FSM.
|
|
type Deps struct {
|
|
// Logger used to emit log messages
|
|
Logger hclog.Logger
|
|
// NewStateStore returns a state.Store which the FSM will use to make changes
|
|
// to the state.
|
|
// NewStateStore will be called once when the FSM is created and again any
|
|
// time Restore() is called.
|
|
NewStateStore func() *state.Store
|
|
|
|
Publisher *stream.EventPublisher
|
|
}
|
|
|
|
// NewFromDeps creates a new FSM from its dependencies.
|
|
func NewFromDeps(deps Deps) *FSM {
|
|
if deps.Logger == nil {
|
|
deps.Logger = hclog.New(&hclog.LoggerOptions{})
|
|
}
|
|
|
|
fsm := &FSM{
|
|
deps: deps,
|
|
logger: deps.Logger.Named(logging.FSM),
|
|
apply: make(map[structs.MessageType]command),
|
|
state: deps.NewStateStore(),
|
|
}
|
|
|
|
// Build out the apply dispatch table based on the registered commands.
|
|
for msg, fn := range commands {
|
|
thisFn := fn
|
|
fsm.apply[msg] = func(buf []byte, index uint64) interface{} {
|
|
return thisFn(fsm, buf, index)
|
|
}
|
|
}
|
|
|
|
fsm.chunker = raftchunking.NewChunkingFSM(fsm, nil)
|
|
|
|
// register the streaming snapshot handlers if an event publisher was provided.
|
|
fsm.registerStreamSnapshotHandlers()
|
|
|
|
return fsm
|
|
}
|
|
|
|
func (c *FSM) ChunkingFSM() *raftchunking.ChunkingFSM {
|
|
return c.chunker
|
|
}
|
|
|
|
// State is used to return a handle to the current state
|
|
func (c *FSM) State() *state.Store {
|
|
c.stateLock.RLock()
|
|
defer c.stateLock.RUnlock()
|
|
return c.state
|
|
}
|
|
|
|
func (c *FSM) Apply(log *raft.Log) interface{} {
|
|
buf := log.Data
|
|
msgType := structs.MessageType(buf[0])
|
|
|
|
// Check if this message type should be ignored when unknown. This is
|
|
// used so that new commands can be added with developer control if older
|
|
// versions can safely ignore the command, or if they should crash.
|
|
ignoreUnknown := false
|
|
if msgType&structs.IgnoreUnknownTypeFlag == structs.IgnoreUnknownTypeFlag {
|
|
msgType &= ^structs.IgnoreUnknownTypeFlag
|
|
ignoreUnknown = true
|
|
}
|
|
|
|
// Apply based on the dispatch table, if possible.
|
|
if fn := c.apply[msgType]; fn != nil {
|
|
return fn(buf[1:], log.Index)
|
|
}
|
|
|
|
// Otherwise, see if it's safe to ignore. If not, we have to panic so
|
|
// that we crash and our state doesn't diverge.
|
|
if ignoreUnknown {
|
|
c.logger.Warn("ignoring unknown message type, upgrade to newer version", "type", msgType)
|
|
return nil
|
|
}
|
|
panic(fmt.Errorf("failed to apply request: %#v", buf))
|
|
}
|
|
|
|
func (c *FSM) Snapshot() (raft.FSMSnapshot, error) {
|
|
defer func(start time.Time) {
|
|
c.logger.Info("snapshot created", "duration", time.Since(start).String())
|
|
}(time.Now())
|
|
|
|
chunkState, err := c.chunker.CurrentState()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return &snapshot{
|
|
state: c.state.Snapshot(),
|
|
chunkState: chunkState,
|
|
}, nil
|
|
}
|
|
|
|
// Restore streams in the snapshot and replaces the current state store with a
|
|
// new one based on the snapshot if all goes OK during the restore.
|
|
func (c *FSM) Restore(old io.ReadCloser) error {
|
|
defer old.Close()
|
|
|
|
stateNew := c.deps.NewStateStore()
|
|
|
|
// Set up a new restore transaction
|
|
restore := stateNew.Restore()
|
|
defer restore.Abort()
|
|
|
|
handler := func(header *SnapshotHeader, msg structs.MessageType, dec *codec.Decoder) error {
|
|
switch {
|
|
case msg == structs.ChunkingStateType:
|
|
chunkState := &raftchunking.State{
|
|
ChunkMap: make(raftchunking.ChunkMap),
|
|
}
|
|
if err := dec.Decode(chunkState); err != nil {
|
|
return err
|
|
}
|
|
if err := c.chunker.RestoreState(chunkState); err != nil {
|
|
return err
|
|
}
|
|
case restorers[msg] != nil:
|
|
fn := restorers[msg]
|
|
if err := fn(header, restore, dec); err != nil {
|
|
return err
|
|
}
|
|
default:
|
|
return fmt.Errorf("Unrecognized msg type %d", msg)
|
|
}
|
|
return nil
|
|
}
|
|
if err := ReadSnapshot(old, handler); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := restore.Commit(); err != nil {
|
|
return err
|
|
}
|
|
|
|
// External code might be calling State(), so we need to synchronize
|
|
// here to make sure we swap in the new state store atomically.
|
|
c.stateLock.Lock()
|
|
stateOld := c.state
|
|
c.state = stateNew
|
|
|
|
// Tell the EventPublisher to cycle anything watching these topics. Replacement
|
|
// of the state store means that indexes could have gone backwards and data changed.
|
|
//
|
|
// This needs to happen while holding the state lock to ensure its not racey. If we
|
|
// did this outside of the locked section closer to where we abandon the old store
|
|
// then there would be a possibility for new streams to be opened that would get
|
|
// a snapshot from the cache sourced from old data but would be receiving events
|
|
// for new data. To prevent that inconsistency we refresh the topics while holding
|
|
// the lock which ensures that any subscriptions to topics for FSM generated events
|
|
if c.deps.Publisher != nil {
|
|
c.deps.Publisher.RefreshTopic(state.EventTopicServiceHealth)
|
|
c.deps.Publisher.RefreshTopic(state.EventTopicServiceHealthConnect)
|
|
c.deps.Publisher.RefreshTopic(state.EventTopicCARoots)
|
|
}
|
|
c.stateLock.Unlock()
|
|
|
|
// Signal that the old state store has been abandoned. This is required
|
|
// because we don't operate on it any more, we just throw it away, so
|
|
// blocking queries won't see any changes and need to be woken up.
|
|
stateOld.Abandon()
|
|
|
|
return nil
|
|
}
|
|
|
|
// ReadSnapshot decodes each message type and utilizes the handler function to
|
|
// process each message type individually
|
|
func ReadSnapshot(r io.Reader, handler func(header *SnapshotHeader, msg structs.MessageType, dec *codec.Decoder) error) error {
|
|
// Create a decoder
|
|
dec := codec.NewDecoder(r, structs.MsgpackHandle)
|
|
|
|
// Read in the header
|
|
var header SnapshotHeader
|
|
if err := dec.Decode(&header); err != nil {
|
|
return err
|
|
}
|
|
|
|
// Populate the new state
|
|
msgType := make([]byte, 1)
|
|
for {
|
|
// Read the message type
|
|
_, err := r.Read(msgType)
|
|
if err == io.EOF {
|
|
return nil
|
|
} else if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Decode
|
|
msg := structs.MessageType(msgType[0])
|
|
|
|
if err := handler(&header, msg, dec); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
func (c *FSM) registerStreamSnapshotHandlers() {
|
|
if c.deps.Publisher == nil {
|
|
return
|
|
}
|
|
|
|
err := c.deps.Publisher.RegisterHandler(state.EventTopicServiceHealth, func(req stream.SubscribeRequest, buf stream.SnapshotAppender) (uint64, error) {
|
|
return c.State().ServiceHealthSnapshot(req, buf)
|
|
}, false)
|
|
if err != nil {
|
|
panic(fmt.Errorf("fatal error encountered registering streaming snapshot handlers: %w", err))
|
|
}
|
|
|
|
err = c.deps.Publisher.RegisterHandler(state.EventTopicServiceHealthConnect, func(req stream.SubscribeRequest, buf stream.SnapshotAppender) (uint64, error) {
|
|
return c.State().ServiceHealthSnapshot(req, buf)
|
|
}, false)
|
|
if err != nil {
|
|
panic(fmt.Errorf("fatal error encountered registering streaming snapshot handlers: %w", err))
|
|
}
|
|
|
|
err = c.deps.Publisher.RegisterHandler(state.EventTopicCARoots, func(req stream.SubscribeRequest, buf stream.SnapshotAppender) (uint64, error) {
|
|
return c.State().CARootsSnapshot(req, buf)
|
|
}, false)
|
|
if err != nil {
|
|
panic(fmt.Errorf("fatal error encountered registering streaming snapshot handlers: %w", err))
|
|
}
|
|
|
|
err = c.deps.Publisher.RegisterHandler(state.EventTopicMeshConfig, func(req stream.SubscribeRequest, buf stream.SnapshotAppender) (uint64, error) {
|
|
return c.State().MeshConfigSnapshot(req, buf)
|
|
}, true)
|
|
if err != nil {
|
|
panic(fmt.Errorf("fatal error encountered registering streaming snapshot handlers: %w", err))
|
|
}
|
|
|
|
err = c.deps.Publisher.RegisterHandler(state.EventTopicServiceResolver, func(req stream.SubscribeRequest, buf stream.SnapshotAppender) (uint64, error) {
|
|
return c.State().ServiceResolverSnapshot(req, buf)
|
|
}, true)
|
|
if err != nil {
|
|
panic(fmt.Errorf("fatal error encountered registering streaming snapshot handlers: %w", err))
|
|
}
|
|
|
|
err = c.deps.Publisher.RegisterHandler(state.EventTopicIngressGateway, func(req stream.SubscribeRequest, buf stream.SnapshotAppender) (uint64, error) {
|
|
return c.State().IngressGatewaySnapshot(req, buf)
|
|
}, true)
|
|
if err != nil {
|
|
panic(fmt.Errorf("fatal error encountered registering streaming snapshot handlers: %w", err))
|
|
}
|
|
|
|
err = c.deps.Publisher.RegisterHandler(state.EventTopicServiceIntentions, func(req stream.SubscribeRequest, buf stream.SnapshotAppender) (uint64, error) {
|
|
return c.State().ServiceIntentionsSnapshot(req, buf)
|
|
}, true)
|
|
if err != nil {
|
|
panic(fmt.Errorf("fatal error encountered registering streaming snapshot handlers: %w", err))
|
|
}
|
|
|
|
err = c.deps.Publisher.RegisterHandler(state.EventTopicServiceList, func(req stream.SubscribeRequest, buf stream.SnapshotAppender) (uint64, error) {
|
|
return c.State().ServiceListSnapshot(req, buf)
|
|
}, true)
|
|
if err != nil {
|
|
panic(fmt.Errorf("fatal error encountered registering streaming snapshot handlers: %w", err))
|
|
}
|
|
}
|