forked from ipfs-cluster/ipfs-cluster
/
state.go
170 lines (145 loc) · 4.41 KB
/
state.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
package main
import (
"bytes"
"encoding/json"
"errors"
"io"
"io/ioutil"
ipfscluster "github.com/ipfs/ipfs-cluster"
"github.com/ipfs/ipfs-cluster/api"
"github.com/ipfs/ipfs-cluster/consensus/raft"
"github.com/ipfs/ipfs-cluster/pstoremgr"
"github.com/ipfs/ipfs-cluster/state/mapstate"
)
var errNoSnapshot = errors.New("no snapshot found")
func upgrade() error {
newState, current, err := restoreStateFromDisk()
if err != nil {
return err
}
if current {
logger.Warning("Skipping migration of up-to-date state")
return nil
}
cfgMgr, cfgs := makeConfigs()
err = cfgMgr.LoadJSONFromFile(configPath)
if err != nil {
return err
}
pm := pstoremgr.New(nil, cfgs.clusterCfg.GetPeerstorePath())
raftPeers := append(ipfscluster.PeersFromMultiaddrs(pm.LoadPeerstore()), cfgs.clusterCfg.ID)
return raft.SnapshotSave(cfgs.consensusCfg, newState, raftPeers)
}
func export(w io.Writer) error {
stateToExport, _, err := restoreStateFromDisk()
if err != nil {
return err
}
return exportState(stateToExport, w)
}
// restoreStateFromDisk returns a mapstate containing the latest
// snapshot, a flag set to true when the state format has the
// current version and an error
func restoreStateFromDisk() (*mapstate.MapState, bool, error) {
cfgMgr, cfgs := makeConfigs()
err := cfgMgr.LoadJSONFromFile(configPath)
if err != nil {
return nil, false, err
}
r, snapExists, err := raft.LastStateRaw(cfgs.consensusCfg)
if !snapExists {
err = errNoSnapshot
}
if err != nil {
return nil, false, err
}
stateFromSnap := mapstate.NewMapState()
// duplicate reader to both check version and migrate
var buf bytes.Buffer
r2 := io.TeeReader(r, &buf)
raw, err := ioutil.ReadAll(r2)
if err != nil {
return nil, false, err
}
err = stateFromSnap.Unmarshal(raw)
if err != nil {
return nil, false, err
}
if stateFromSnap.GetVersion() == mapstate.Version {
return stateFromSnap, true, nil
}
err = stateFromSnap.Migrate(&buf)
if err != nil {
return nil, false, err
}
return stateFromSnap, false, nil
}
func stateImport(r io.Reader) error {
cfgMgr, cfgs := makeConfigs()
err := cfgMgr.LoadJSONFromFile(configPath)
if err != nil {
return err
}
pinSerials := make([]api.PinSerial, 0)
dec := json.NewDecoder(r)
err = dec.Decode(&pinSerials)
if err != nil {
return err
}
stateToImport := mapstate.NewMapState()
for _, pS := range pinSerials {
err = stateToImport.Add(pS.ToPin())
if err != nil {
return err
}
}
pm := pstoremgr.New(nil, cfgs.clusterCfg.GetPeerstorePath())
raftPeers := append(ipfscluster.PeersFromMultiaddrs(pm.LoadPeerstore()), cfgs.clusterCfg.ID)
return raft.SnapshotSave(cfgs.consensusCfg, stateToImport, raftPeers)
}
func validateVersion(cfg *ipfscluster.Config, cCfg *raft.Config) error {
state := mapstate.NewMapState()
r, snapExists, err := raft.LastStateRaw(cCfg)
if !snapExists && err != nil {
logger.Error("error before reading latest snapshot.")
} else if snapExists && err != nil {
logger.Error("error after reading last snapshot. Snapshot potentially corrupt.")
} else if snapExists && err == nil {
raw, err2 := ioutil.ReadAll(r)
if err2 != nil {
return err2
}
err2 = state.Unmarshal(raw)
if err2 != nil {
logger.Error("error unmarshalling snapshot. Snapshot potentially corrupt.")
return err2
}
if state.GetVersion() != mapstate.Version {
logger.Error("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
logger.Error("Out of date ipfs-cluster state is saved.")
logger.Error("To migrate to the new version, run ipfs-cluster-service state upgrade.")
logger.Error("To launch a node without this state, rename the consensus data directory.")
logger.Error("Hint, the default is .ipfs-cluster/ipfs-cluster-data.")
logger.Error("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
err = errors.New("outdated state version stored")
}
} // !snapExists && err == nil // no existing state, no check needed
return err
}
// ExportState saves a json representation of a state
func exportState(state *mapstate.MapState, w io.Writer) error {
// Serialize pins
pins := state.List()
pinSerials := make([]api.PinSerial, len(pins), len(pins))
for i, pin := range pins {
pinSerials[i] = pin.ToSerial()
}
// Write json to output file
enc := json.NewEncoder(w)
enc.SetIndent("", " ")
return enc.Encode(pinSerials)
}
// CleanupState cleans the state
func cleanupState(cCfg *raft.Config) error {
return raft.CleanupRaft(cCfg.GetDataFolder(), cCfg.BackupsRotate)
}