b6dd1191b2
Stream snapshot to FSM when restoring from archive The `RestoreFromArchive` helper decompresses the snapshot archive to a temporary file before reading it into the FSM. For large snapshots this performs a lot of disk IO. Stream decompress the snapshot as we read it, without first writing to a temporary file. Add bexpr filters to the `RestoreFromArchive` helper. The operator can pass these as `-filter` arguments to `nomad operator snapshot state` (and other commands in the future) to include only desired data when reading the snapshot.
50 lines
1 KiB
Go
50 lines
1 KiB
Go
package raftutil
|
|
|
|
import (
|
|
"fmt"
|
|
"io"
|
|
|
|
"github.com/hashicorp/go-hclog"
|
|
"github.com/hashicorp/raft"
|
|
|
|
"github.com/hashicorp/nomad/helper/snapshot"
|
|
"github.com/hashicorp/nomad/nomad"
|
|
"github.com/hashicorp/nomad/nomad/state"
|
|
)
|
|
|
|
func RestoreFromArchive(archive io.Reader, filter *nomad.FSMFilter) (*state.StateStore, *raft.SnapshotMeta, error) {
|
|
logger := hclog.L()
|
|
|
|
fsm, err := dummyFSM(logger)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to create FSM: %w", err)
|
|
}
|
|
|
|
// r is closed by RestoreFiltered, w is closed by CopySnapshot
|
|
r, w := io.Pipe()
|
|
|
|
errCh := make(chan error)
|
|
metaCh := make(chan *raft.SnapshotMeta)
|
|
|
|
go func() {
|
|
meta, err := snapshot.CopySnapshot(archive, w)
|
|
if err != nil {
|
|
errCh <- fmt.Errorf("failed to read snapshot: %w", err)
|
|
} else {
|
|
metaCh <- meta
|
|
}
|
|
}()
|
|
|
|
err = fsm.RestoreWithFilter(r, filter)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to restore from snapshot: %w", err)
|
|
}
|
|
|
|
select {
|
|
case err := <-errCh:
|
|
return nil, nil, err
|
|
case meta := <-metaCh:
|
|
return fsm.State(), meta, nil
|
|
}
|
|
}
|