luxolus
/
open-consul
2
0
Fork 0
Code Issues 1 Pull Requests Packages Releases Wiki Activity

Enhance the output of consul snapshot inspect (#8787)

This commit is contained in:
s-christoff 2020-10-09 14:57:29 -05:00 committed by GitHub
parent 1d42dd0cdb
commit a62705101f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 406 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.changelog/8787.txt Normal file
View File

@ -0,0 +1,3 @@
```release-note:feature
cli: update `snapshot inspect` command to provide more detailed snapshot data
```

62
agent/consul/fsm/fsm.go
View File

@ -159,28 +159,7 @@ func (c *FSM) Restore(old io.ReadCloser) error {
restore := stateNew.Restore() restore := stateNew.Restore()
defer restore.Abort() defer restore.Abort()
// Create a decoder handler := func(header *SnapshotHeader, msg structs.MessageType, dec *codec.Decoder) error {
dec := codec.NewDecoder(old, 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 := old.Read(msgType)
if err == io.EOF {
break
} else if err != nil {
return err
}
// Decode
msg := structs.MessageType(msgType[0])
switch { switch {
case msg == structs.ChunkingStateType: case msg == structs.ChunkingStateType:
chunkState := &raftchunking.State{ chunkState := &raftchunking.State{
@ -194,13 +173,18 @@ func (c *FSM) Restore(old io.ReadCloser) error {
} }
case restorers[msg] != nil: case restorers[msg] != nil:
fn := restorers[msg] fn := restorers[msg]
if err := fn(&header, restore, dec); err != nil { if err := fn(header, restore, dec); err != nil {
return err return err
} }
default: default:
return fmt.Errorf("Unrecognized msg type %d", msg) 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 { if err := restore.Commit(); err != nil {
return err return err
} }
@ -218,3 +202,35 @@ func (c *FSM) Restore(old io.ReadCloser) error {
stateOld.Abandon() stateOld.Abandon()
return nil 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
}
}
}

8
agent/consul/fsm/snapshot.go
View File

@ -20,8 +20,8 @@ type snapshot struct {
chunkState *raftchunking.State chunkState *raftchunking.State
} }
// snapshotHeader is the first entry in our snapshot // SnapshotHeader is the first entry in our snapshot
type snapshotHeader struct { type SnapshotHeader struct {
// LastIndex is the last index that affects the data. // LastIndex is the last index that affects the data.
// This is used when we do the restore for watchers. // This is used when we do the restore for watchers.
LastIndex uint64 LastIndex uint64
@ -40,7 +40,7 @@ func registerPersister(fn persister) {
} }
// restorer is a function used to load back a snapshot of the FSM state. // restorer is a function used to load back a snapshot of the FSM state.
type restorer func(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error type restorer func(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error
// restorers is a map of restore functions by message type. // restorers is a map of restore functions by message type.
var restorers map[structs.MessageType]restorer var restorers map[structs.MessageType]restorer
@ -62,7 +62,7 @@ func (s *snapshot) Persist(sink raft.SnapshotSink) error {
defer metrics.MeasureSince([]string{"fsm", "persist"}, time.Now()) defer metrics.MeasureSince([]string{"fsm", "persist"}, time.Now())
// Write the header // Write the header
header := snapshotHeader{ header := SnapshotHeader{
LastIndex: s.state.LastIndex(), LastIndex: s.state.LastIndex(),
} }
encoder := codec.NewEncoder(sink, structs.MsgpackHandle) encoder := codec.NewEncoder(sink, structs.MsgpackHandle)

44
agent/consul/fsm/snapshot_oss.go
View File

@ -506,7 +506,7 @@ func (s *snapshot) persistIndex(sink raft.SnapshotSink, encoder *codec.Encoder)
return nil return nil
} }
func restoreRegistration(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreRegistration(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.RegisterRequest var req structs.RegisterRequest
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -517,7 +517,7 @@ func restoreRegistration(header *snapshotHeader, restore *state.Restore, decoder
return nil return nil
} }
func restoreKV(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreKV(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.DirEntry var req structs.DirEntry
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -528,7 +528,7 @@ func restoreKV(header *snapshotHeader, restore *state.Restore, decoder *codec.De
return nil return nil
} }
func restoreTombstone(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreTombstone(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.DirEntry var req structs.DirEntry
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -547,7 +547,7 @@ func restoreTombstone(header *snapshotHeader, restore *state.Restore, decoder *c
return nil return nil
} }
func restoreSession(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreSession(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.Session var req structs.Session
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -558,7 +558,7 @@ func restoreSession(header *snapshotHeader, restore *state.Restore, decoder *cod
return nil return nil
} }
func restoreACL(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreACL(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.ACL var req structs.ACL
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -571,7 +571,7 @@ func restoreACL(header *snapshotHeader, restore *state.Restore, decoder *codec.D
} }
// DEPRECATED (ACL-Legacy-Compat) - remove once v1 acl compat is removed // DEPRECATED (ACL-Legacy-Compat) - remove once v1 acl compat is removed
func restoreACLBootstrap(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreACLBootstrap(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.ACLBootstrap var req structs.ACLBootstrap
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -582,7 +582,7 @@ func restoreACLBootstrap(header *snapshotHeader, restore *state.Restore, decoder
return restore.IndexRestore(&state.IndexEntry{Key: "acl-token-bootstrap", Value: req.ModifyIndex}) return restore.IndexRestore(&state.IndexEntry{Key: "acl-token-bootstrap", Value: req.ModifyIndex})
} }
func restoreCoordinates(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreCoordinates(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.Coordinates var req structs.Coordinates
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -593,7 +593,7 @@ func restoreCoordinates(header *snapshotHeader, restore *state.Restore, decoder
return nil return nil
} }
func restorePreparedQuery(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restorePreparedQuery(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.PreparedQuery var req structs.PreparedQuery
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -604,7 +604,7 @@ func restorePreparedQuery(header *snapshotHeader, restore *state.Restore, decode
return nil return nil
} }
func restoreAutopilot(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreAutopilot(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req autopilot.Config var req autopilot.Config
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -615,7 +615,7 @@ func restoreAutopilot(header *snapshotHeader, restore *state.Restore, decoder *c
return nil return nil
} }
func restoreLegacyIntention(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreLegacyIntention(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.Intention var req structs.Intention
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -627,7 +627,7 @@ func restoreLegacyIntention(header *snapshotHeader, restore *state.Restore, deco
return nil return nil
} }
func restoreConnectCA(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreConnectCA(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.CARoot var req structs.CARoot
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -638,7 +638,7 @@ func restoreConnectCA(header *snapshotHeader, restore *state.Restore, decoder *c
return nil return nil
} }
func restoreConnectCAProviderState(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreConnectCAProviderState(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.CAConsulProviderState var req structs.CAConsulProviderState
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -649,7 +649,7 @@ func restoreConnectCAProviderState(header *snapshotHeader, restore *state.Restor
return nil return nil
} }
func restoreConnectCAConfig(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreConnectCAConfig(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.CAConfiguration var req structs.CAConfiguration
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -660,7 +660,7 @@ func restoreConnectCAConfig(header *snapshotHeader, restore *state.Restore, deco
return nil return nil
} }
func restoreIndex(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreIndex(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req state.IndexEntry var req state.IndexEntry
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -668,7 +668,7 @@ func restoreIndex(header *snapshotHeader, restore *state.Restore, decoder *codec
return restore.IndexRestore(&req) return restore.IndexRestore(&req)
} }
func restoreToken(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreToken(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.ACLToken var req structs.ACLToken
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -688,7 +688,7 @@ func restoreToken(header *snapshotHeader, restore *state.Restore, decoder *codec
return restore.ACLToken(&req) return restore.ACLToken(&req)
} }
func restorePolicy(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restorePolicy(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.ACLPolicy var req structs.ACLPolicy
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -696,7 +696,7 @@ func restorePolicy(header *snapshotHeader, restore *state.Restore, decoder *code
return restore.ACLPolicy(&req) return restore.ACLPolicy(&req)
} }
func restoreConfigEntry(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreConfigEntry(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.ConfigEntryRequest var req structs.ConfigEntryRequest
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -704,7 +704,7 @@ func restoreConfigEntry(header *snapshotHeader, restore *state.Restore, decoder
return restore.ConfigEntry(req.Entry) return restore.ConfigEntry(req.Entry)
} }
func restoreRole(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreRole(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.ACLRole var req structs.ACLRole
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -712,7 +712,7 @@ func restoreRole(header *snapshotHeader, restore *state.Restore, decoder *codec.
return restore.ACLRole(&req) return restore.ACLRole(&req)
} }
func restoreBindingRule(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreBindingRule(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.ACLBindingRule var req structs.ACLBindingRule
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -720,7 +720,7 @@ func restoreBindingRule(header *snapshotHeader, restore *state.Restore, decoder
return restore.ACLBindingRule(&req) return restore.ACLBindingRule(&req)
} }
func restoreAuthMethod(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreAuthMethod(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.ACLAuthMethod var req structs.ACLAuthMethod
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -728,7 +728,7 @@ func restoreAuthMethod(header *snapshotHeader, restore *state.Restore, decoder *
return restore.ACLAuthMethod(&req) return restore.ACLAuthMethod(&req)
} }
func restoreFederationState(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreFederationState(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.FederationStateRequest var req structs.FederationStateRequest
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err
@ -736,7 +736,7 @@ func restoreFederationState(header *snapshotHeader, restore *state.Restore, deco
return restore.FederationState(req.State) return restore.FederationState(req.State)
} }
func restoreSystemMetadata(header *snapshotHeader, restore *state.Restore, decoder *codec.Decoder) error { func restoreSystemMetadata(header *SnapshotHeader, restore *state.Restore, decoder *codec.Decoder) error {
var req structs.SystemMetadataEntry var req structs.SystemMetadataEntry
if err := decoder.Decode(&req); err != nil { if err := decoder.Decode(&req); err != nil {
return err return err

55
agent/structs/structs.go
View File

@ -72,6 +72,46 @@ const (
SystemMetadataRequestType = 31 SystemMetadataRequestType = 31
) )
// if a new request type is added above it must be
// added to the map below
// requestTypeStrings is used for snapshot enhance
// any new request types added must be placed here
var requestTypeStrings = map[MessageType]string{
RegisterRequestType: "Register",
DeregisterRequestType: "Deregister",
KVSRequestType: "KVS",
SessionRequestType: "Session",
ACLRequestType: "ACL", // DEPRECATED (ACL-Legacy-Compat)
TombstoneRequestType: "Tombstone",
CoordinateBatchUpdateType: "CoordinateBatchUpdate",
PreparedQueryRequestType: "PreparedQuery",
TxnRequestType: "Txn",
AutopilotRequestType: "Autopilot",
AreaRequestType: "Area",
ACLBootstrapRequestType: "ACLBootstrap",
IntentionRequestType: "Intention",
ConnectCARequestType: "ConnectCA",
ConnectCAProviderStateType: "ConnectCAProviderState",
ConnectCAConfigType: "ConnectCAConfig", // FSM snapshots only.
IndexRequestType: "Index", // FSM snapshots only.
ACLTokenSetRequestType: "ACLToken",
ACLTokenDeleteRequestType: "ACLTokenDelete",
ACLPolicySetRequestType: "ACLPolicy",
ACLPolicyDeleteRequestType: "ACLPolicyDelete",
ConnectCALeafRequestType: "ConnectCALeaf",
ConfigEntryRequestType: "ConfigEntry",
ACLRoleSetRequestType: "ACLRole",
ACLRoleDeleteRequestType: "ACLRoleDelete",
ACLBindingRuleSetRequestType: "ACLBindingRule",
ACLBindingRuleDeleteRequestType: "ACLBindingRuleDelete",
ACLAuthMethodSetRequestType: "ACLAuthMethod",
ACLAuthMethodDeleteRequestType: "ACLAuthMethodDelete",
ChunkingStateType: "ChunkingState",
FederationStateRequestType: "FederationState",
SystemMetadataRequestType: "SystemMetadata",
}
const ( const (
// IgnoreUnknownTypeFlag is set along with a MessageType // IgnoreUnknownTypeFlag is set along with a MessageType
// to indicate that the message type can be safely ignored // to indicate that the message type can be safely ignored
@ -2442,6 +2482,21 @@ func (r *KeyringResponses) New() interface{} {
return new(KeyringResponses) return new(KeyringResponses)
} }
// String converts message type int to string
func (m MessageType) String() string {
s, ok := requestTypeStrings[m]
if ok {
return s
}
s, ok = enterpriseRequestType(m)
if ok {
return s
}
return "Unknown(" + strconv.Itoa(int(m)) + ")"
}
// UpstreamDownstream pairs come from individual proxy registrations, which can be updated independently. // UpstreamDownstream pairs come from individual proxy registrations, which can be updated independently.
type UpstreamDownstream struct { type UpstreamDownstream struct {
Upstream ServiceName Upstream ServiceName

4
agent/structs/structs_oss.go
View File

@ -131,6 +131,10 @@ func (_ *HealthCheck) Validate() error {
return nil return nil
} }
func enterpriseRequestType(m MessageType) (string, bool) {
return "", false
}
// CheckIDs returns the IDs for all checks associated with a session, regardless of type // CheckIDs returns the IDs for all checks associated with a session, regardless of type
func (s *Session) CheckIDs() []types.CheckID { func (s *Session) CheckIDs() []types.CheckID {
// Merge all check IDs into a single slice, since they will be handled the same way // Merge all check IDs into a single slice, since they will be handled the same way

187
command/snapshot/inspect/snapshot_inspect.go
View File

@ -4,11 +4,20 @@ import (
"bytes" "bytes"
"flag" "flag"
"fmt" "fmt"
"io"
"os" "os"
"sort"
"strconv"
"strings"
"text/tabwriter" "text/tabwriter"
"github.com/hashicorp/consul/agent/consul/fsm"
"github.com/hashicorp/consul/agent/structs"
"github.com/hashicorp/consul/command/flags" "github.com/hashicorp/consul/command/flags"
"github.com/hashicorp/consul/snapshot" "github.com/hashicorp/consul/snapshot"
"github.com/hashicorp/go-hclog"
"github.com/hashicorp/go-msgpack/codec"
"github.com/hashicorp/raft"
"github.com/mitchellh/cli" "github.com/mitchellh/cli"
) )
@ -31,12 +40,13 @@ func (c *cmd) init() {
func (c *cmd) Run(args []string) int { func (c *cmd) Run(args []string) int {
if err := c.flags.Parse(args); err != nil { if err := c.flags.Parse(args); err != nil {
c.UI.Error(err.Error())
return 1 return 1
} }
var file string var file string
args = c.flags.Args() args = c.flags.Args()
switch len(args) { switch len(args) {
case 0: case 0:
c.UI.Error("Missing FILE argument") c.UI.Error("Missing FILE argument")
@ -56,12 +66,45 @@ func (c *cmd) Run(args []string) int {
} }
defer f.Close() defer f.Close()
meta, err := snapshot.Verify(f) readFile, meta, err := snapshot.Read(hclog.New(nil), f)
if err != nil { if err != nil {
c.UI.Error(fmt.Sprintf("Error verifying snapshot: %s", err)) c.UI.Error(fmt.Sprintf("Error reading snapshot: %s", err))
}
defer func() {
if err := readFile.Close(); err != nil {
c.UI.Error(fmt.Sprintf("Failed to close temp snapshot: %v", err))
}
if err := os.Remove(readFile.Name()); err != nil {
c.UI.Error(fmt.Sprintf("Failed to clean up temp snapshot: %v", err))
}
}()
stats, totalSize, err := enhance(readFile)
if err != nil {
c.UI.Error(fmt.Sprintf("Error extracting snapshot data: %s", err))
return 1 return 1
} }
// Outputs the original style of inspect information
legacy, err := c.legacyStats(meta)
if err != nil {
c.UI.Error(fmt.Sprintf("Error outputting snapshot data: %s", err))
}
c.UI.Info(legacy.String())
// Outputs the more detailed snapshot information
enhanced, err := c.readStats(stats, totalSize)
if err != nil {
c.UI.Error(fmt.Sprintf("Error outputting enhanced snapshot data: %s", err))
return 1
}
c.UI.Info(enhanced.String())
return 0
}
// legacyStats outputs the expected stats from the original snapshot
// inspect command
func (c *cmd) legacyStats(meta *raft.SnapshotMeta) (bytes.Buffer, error) {
var b bytes.Buffer var b bytes.Buffer
tw := tabwriter.NewWriter(&b, 0, 2, 6, ' ', 0) tw := tabwriter.NewWriter(&b, 0, 2, 6, ' ', 0)
fmt.Fprintf(tw, "ID\t%s\n", meta.ID) fmt.Fprintf(tw, "ID\t%s\n", meta.ID)
@ -69,14 +112,142 @@ func (c *cmd) Run(args []string) int {
fmt.Fprintf(tw, "Index\t%d\n", meta.Index) fmt.Fprintf(tw, "Index\t%d\n", meta.Index)
fmt.Fprintf(tw, "Term\t%d\n", meta.Term) fmt.Fprintf(tw, "Term\t%d\n", meta.Term)
fmt.Fprintf(tw, "Version\t%d\n", meta.Version) fmt.Fprintf(tw, "Version\t%d\n", meta.Version)
if err = tw.Flush(); err != nil { if err := tw.Flush(); err != nil {
c.UI.Error(fmt.Sprintf("Error rendering snapshot info: %s", err)) return b, err
return 1 }
return b, nil
}
type typeStats struct {
Name string
Sum int
Count int
}
// countingReader helps keep track of the bytes we have read
// when reading snapshots
type countingReader struct {
wrappedReader io.Reader
read int
}
func (r *countingReader) Read(p []byte) (n int, err error) {
n, err = r.wrappedReader.Read(p)
if err == nil {
r.read += n
}
return n, err
}
// enhance utilizes ReadSnapshot to populate the struct with
// all of the snapshot's itemized data
func enhance(file io.Reader) (map[structs.MessageType]typeStats, int, error) {
stats := make(map[structs.MessageType]typeStats)
cr := &countingReader{wrappedReader: file}
totalSize := 0
handler := func(header *fsm.SnapshotHeader, msg structs.MessageType, dec *codec.Decoder) error {
name := structs.MessageType.String(msg)
s := stats[msg]
if s.Name == "" {
s.Name = name
}
var val interface{}
err := dec.Decode(&val)
if err != nil {
return fmt.Errorf("failed to decode msg type %v, error %v", name, err)
} }
c.UI.Info(b.String()) size := cr.read - totalSize
s.Sum += size
s.Count++
totalSize = cr.read
stats[msg] = s
return nil
}
if err := fsm.ReadSnapshot(cr, handler); err != nil {
return nil, 0, err
}
return stats, totalSize, nil
return 0 }
// readStats takes the information generated from enhance and creates human
// readable output from it
func (c *cmd) readStats(stats map[structs.MessageType]typeStats, totalSize int) (bytes.Buffer, error) {
// Output stats in size-order
ss := make([]typeStats, 0, len(stats))
for _, s := range stats {
ss = append(ss, s)
}
// Sort the stat slice
sort.Slice(ss, func(i, j int) bool { return ss[i].Sum > ss[j].Sum })
var b bytes.Buffer
tw := tabwriter.NewWriter(&b, 8, 8, 6, ' ', 0)
fmt.Fprintln(tw, "\n Type\tCount\tSize\t")
fmt.Fprintf(tw, " %s\t%s\t%s\t", "----", "----", "----")
// For each different type generate new output
for _, s := range ss {
fmt.Fprintf(tw, "\n %s\t%d\t%s\t", s.Name, s.Count, ByteSize(uint64(s.Sum)))
}
fmt.Fprintf(tw, "\n %s\t%s\t%s\t", "----", "----", "----")
fmt.Fprintf(tw, "\n Total\t\t%s\t", ByteSize(uint64(totalSize)))
if err := tw.Flush(); err != nil {
c.UI.Error(fmt.Sprintf("Error rendering snapshot info: %s", err))
return b, err
}
return b, nil
}
// ByteSize returns a human-readable byte string of the form 10MB, 12.5KB, and so forth. The following units are available:
// TB: Terabyte
// GB: Gigabyte
// MB: Megabyte
// KB: Kilobyte
// B: Byte
// The unit that results in the smallest number greater than or equal to 1 is always chosen.
// From https://github.com/cloudfoundry/bytefmt/blob/master/bytes.go
const (
BYTE = 1 << (10 * iota)
KILOBYTE
MEGABYTE
GIGABYTE
TERABYTE
)
func ByteSize(bytes uint64) string {
unit := ""
value := float64(bytes)
switch {
case bytes >= TERABYTE:
unit = "TB"
value = value / TERABYTE
case bytes >= GIGABYTE:
unit = "GB"
value = value / GIGABYTE
case bytes >= MEGABYTE:
unit = "MB"
value = value / MEGABYTE
case bytes >= KILOBYTE:
unit = "KB"
value = value / KILOBYTE
case bytes >= BYTE:
unit = "B"
case bytes == 0:
return "0"
}
result := strconv.FormatFloat(value, 'f', 1, 64)
result = strings.TrimSuffix(result, ".0")
return result + unit
} }
func (c *cmd) Synopsis() string { func (c *cmd) Synopsis() string {

69
command/snapshot/inspect/snapshot_inspect_test.go
View File

@ -1,17 +1,36 @@
package inspect package inspect
import ( import (
"io" "flag"
"os" "io/ioutil"
"path/filepath" "path/filepath"
"strings" "strings"
"testing" "testing"
"github.com/hashicorp/consul/agent"
"github.com/hashicorp/consul/sdk/testutil"
"github.com/mitchellh/cli" "github.com/mitchellh/cli"
"github.com/stretchr/testify/require"
) )
// update allows golden files to be updated based on the current output.
var update = flag.Bool("update", false, "update golden files")
// golden reads and optionally writes the expected data to the golden file,
// returning the contents as a string.
func golden(t *testing.T, name, got string) string {
t.Helper()
golden := filepath.Join("testdata", name+".golden")
if *update && got != "" {
err := ioutil.WriteFile(golden, []byte(got), 0644)
require.NoError(t, err)
}
expected, err := ioutil.ReadFile(golden)
require.NoError(t, err)
return string(expected)
}
func TestSnapshotInspectCommand_noTabs(t *testing.T) { func TestSnapshotInspectCommand_noTabs(t *testing.T) {
t.Parallel() t.Parallel()
if strings.ContainsRune(New(cli.NewMockUi()).Help(), '\t') { if strings.ContainsRune(New(cli.NewMockUi()).Help(), '\t') {
@ -60,53 +79,19 @@ func TestSnapshotInspectCommand_Validation(t *testing.T) {
} }
func TestSnapshotInspectCommand(t *testing.T) { func TestSnapshotInspectCommand(t *testing.T) {
t.Parallel()
a := agent.NewTestAgent(t, ``)
defer a.Shutdown()
client := a.Client()
dir := testutil.TempDir(t, "snapshot") filepath := "./testdata/backup.snap"
file := filepath.Join(dir, "backup.tgz")
// Save a snapshot of the current Consul state
f, err := os.Create(file)
if err != nil {
t.Fatalf("err: %v", err)
}
snap, _, err := client.Snapshot().Save(nil)
if err != nil {
f.Close()
t.Fatalf("err: %v", err)
}
if _, err := io.Copy(f, snap); err != nil {
f.Close()
t.Fatalf("err: %v", err)
}
if err := f.Close(); err != nil {
t.Fatalf("err: %v", err)
}
// Inspect the snapshot // Inspect the snapshot
ui := cli.NewMockUi() ui := cli.NewMockUi()
c := New(ui) c := New(ui)
args := []string{file} args := []string{filepath}
code := c.Run(args) code := c.Run(args)
if code != 0 { if code != 0 {
t.Fatalf("bad: %d. %#v", code, ui.ErrorWriter.String()) t.Fatalf("bad: %d. %#v", code, ui.ErrorWriter.String())
} }
output := ui.OutputWriter.String() want := golden(t, t.Name(), ui.OutputWriter.String())
for _, key := range []string{ require.Equal(t, want, ui.OutputWriter.String())
"ID",
"Size",
"Index",
"Term",
"Version",
} {
if !strings.Contains(output, key) {
t.Fatalf("bad %#v, missing %q", output, key)
}
}
} }

20
command/snapshot/inspect/testdata/TestSnapshotInspectCommand.golden vendored Normal file
View File

@ -0,0 +1,20 @@
ID 2-13-1602222343947
Size 5141
Index 13
Term 2
Version 1
Type Count Size
---- ---- ----
Register 3 1.7KB
ConnectCA 1 1.2KB
ConnectCAProviderState 1 1.1KB
Index 12 344B
Autopilot 1 199B
ConnectCAConfig 1 197B
FederationState 1 139B
SystemMetadata 1 68B
ChunkingState 1 12B
---- ---- ----
Total 5KB

BIN
command/snapshot/inspect/testdata/backup.snap (Stored with Git LFS) vendored Normal file
View File

Binary file not shown.

56
snapshot/snapshot.go
View File

@ -160,13 +160,12 @@ func concludeGzipRead(decomp *gzip.Reader) error {
return nil return nil
} }
// Restore takes the snapshot from the reader and attempts to apply it to the // Read a snapshot into a temporary file. The caller is responsible for removing the file.
// given Raft instance. func Read(logger hclog.Logger, in io.Reader) (*os.File, *raft.SnapshotMeta, error) {
func Restore(logger hclog.Logger, in io.Reader, r *raft.Raft) error {
// Wrap the reader in a gzip decompressor. // Wrap the reader in a gzip decompressor.
decomp, err := gzip.NewReader(in) decomp, err := gzip.NewReader(in)
if err != nil { if err != nil {
return fmt.Errorf("failed to decompress snapshot: %v", err) return nil, nil, fmt.Errorf("failed to decompress snapshot: %v", err)
} }
defer func() { defer func() {
if err := decomp.Close(); err != nil { if err := decomp.Close(); err != nil {
@ -178,9 +177,37 @@ func Restore(logger hclog.Logger, in io.Reader, r *raft.Raft) error {
// we can avoid buffering in memory. // we can avoid buffering in memory.
snap, err := ioutil.TempFile("", "snapshot") snap, err := ioutil.TempFile("", "snapshot")
if err != nil { if err != nil {
return fmt.Errorf("failed to create temp snapshot file: %v", err) return nil, nil, fmt.Errorf("failed to create temp snapshot file: %v", err)
} }
// Read the archive.
var metadata raft.SnapshotMeta
if err := read(decomp, &metadata, snap); err != nil {
return nil, nil, fmt.Errorf("failed to read snapshot file: %v", err)
}
if err := concludeGzipRead(decomp); err != nil {
return nil, nil, err
}
// Sync and rewind the file so it's ready to be read again.
if err := snap.Sync(); err != nil {
return nil, nil, fmt.Errorf("failed to sync temp snapshot: %v", err)
}
if _, err := snap.Seek(0, 0); err != nil {
return nil, nil, fmt.Errorf("failed to rewind temp snapshot: %v", err)
}
return snap, &metadata, nil
}
// Restore takes the snapshot from the reader and attempts to apply it to the
// given Raft instance.
func Restore(logger hclog.Logger, in io.Reader, r *raft.Raft) error {
snap, metadata, err := Read(logger, in)
defer func() { defer func() {
if snap == nil {
return
}
if err := snap.Close(); err != nil { if err := snap.Close(); err != nil {
logger.Error("Failed to close temp snapshot", "error", err) logger.Error("Failed to close temp snapshot", "error", err)
} }
@ -188,27 +215,12 @@ func Restore(logger hclog.Logger, in io.Reader, r *raft.Raft) error {
logger.Error("Failed to clean up temp snapshot", "error", err) logger.Error("Failed to clean up temp snapshot", "error", err)
} }
}() }()
if err != nil {
// Read the archive.
var metadata raft.SnapshotMeta
if err := read(decomp, &metadata, snap); err != nil {
return fmt.Errorf("failed to read snapshot file: %v", err)
}
if err := concludeGzipRead(decomp); err != nil {
return err return err
} }
// Sync and rewind the file so it's ready to be read again.
if err := snap.Sync(); err != nil {
return fmt.Errorf("failed to sync temp snapshot: %v", err)
}
if _, err := snap.Seek(0, 0); err != nil {
return fmt.Errorf("failed to rewind temp snapshot: %v", err)
}
// Feed the snapshot into Raft. // Feed the snapshot into Raft.
if err := r.Restore(&metadata, snap, 0); err != nil { if err := r.Restore(metadata, snap, 0); err != nil {
return fmt.Errorf("Raft error when restoring snapshot: %v", err) return fmt.Errorf("Raft error when restoring snapshot: %v", err)
} }

14
website/pages/commands/snapshot/inspect.mdx
View File

@ -41,6 +41,20 @@ Size 667
Index 5 Index 5
Term 2 Term 2
Version 1 Version 1
Type Count Size
---- ---- ----
Register 3 1.7KB
ConnectCA 1 1.2KB
ConnectCAProviderState 1 1.1KB
Index 12 344B
AutopilotRequest 1 199B
ConnectCAConfig 1 197B
FederationState 1 139B
SystemMetadata 1 68B
ChunkingState 1 12B
---- ---- ----
Total 5KB
``` ```
Please see the [HTTP API](/api/snapshot) documentation for Please see the [HTTP API](/api/snapshot) documentation for