snap: init commit

snap/snap.pb.go

@@ -0,0 +1,203 @@
+// Code generated by protoc-gen-gogo.
+// source: snap.proto
+// DO NOT EDIT!
+
+/*
+	Package snap is a generated protocol buffer package.
+
+	It is generated from these files:
+		snap.proto
+
+	It has these top-level messages:
+		Snapshot
+*/
+package snap
+
+import proto "code.google.com/p/gogoprotobuf/proto"
+import json "encoding/json"
+import math "math"
+
+// discarding unused import gogoproto "code.google.com/p/gogoprotobuf/gogoproto/gogo.pb"
+
+import io "io"
+import code_google_com_p_gogoprotobuf_proto "code.google.com/p/gogoprotobuf/proto"
+
+// Reference proto, json, and math imports to suppress error if they are not otherwise used.
+var _ = proto.Marshal
+var _ = &json.SyntaxError{}
+var _ = math.Inf
+
+type Snapshot struct {
+	Crc              uint32 `protobuf:"varint,1,req,name=crc" json:"crc"`
+	Data             []byte `protobuf:"bytes,2,opt,name=data" json:"data,omitempty"`
+	XXX_unrecognized []byte `json:"-"`
+}
+
+func (m *Snapshot) Reset()         { *m = Snapshot{} }
+func (m *Snapshot) String() string { return proto.CompactTextString(m) }
+func (*Snapshot) ProtoMessage()    {}
+
+func init() {
+}
+func (m *Snapshot) Unmarshal(data []byte) error {
+	l := len(data)
+	index := 0
+	for index < l {
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if index >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := data[index]
+			index++
+			wire |= (uint64(b) & 0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		switch fieldNum {
+		case 1:
+			if wireType != 0 {
+				return code_google_com_p_gogoprotobuf_proto.ErrWrongType
+			}
+			for shift := uint(0); ; shift += 7 {
+				if index >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := data[index]
+				index++
+				m.Crc |= (uint32(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 2:
+			if wireType != 2 {
+				return code_google_com_p_gogoprotobuf_proto.ErrWrongType
+			}
+			var byteLen int
+			for shift := uint(0); ; shift += 7 {
+				if index >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := data[index]
+				index++
+				byteLen |= (int(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			postIndex := index + byteLen
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Data = append(m.Data, data[index:postIndex]...)
+			index = postIndex
+		default:
+			var sizeOfWire int
+			for {
+				sizeOfWire++
+				wire >>= 7
+				if wire == 0 {
+					break
+				}
+			}
+			index -= sizeOfWire
+			skippy, err := code_google_com_p_gogoprotobuf_proto.Skip(data[index:])
+			if err != nil {
+				return err
+			}
+			if (index + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, data[index:index+skippy]...)
+			index += skippy
+		}
+	}
+	return nil
+}
+func (m *Snapshot) Size() (n int) {
+	var l int
+	_ = l
+	n += 1 + sovSnap(uint64(m.Crc))
+	if m.Data != nil {
+		l = len(m.Data)
+		n += 1 + l + sovSnap(uint64(l))
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func sovSnap(x uint64) (n int) {
+	for {
+		n++
+		x >>= 7
+		if x == 0 {
+			break
+		}
+	}
+	return n
+}
+func sozSnap(x uint64) (n int) {
+	return sovSnap(uint64((x << 1) ^ uint64((int64(x) >> 63))))
+}
+func (m *Snapshot) Marshal() (data []byte, err error) {
+	size := m.Size()
+	data = make([]byte, size)
+	n, err := m.MarshalTo(data)
+	if err != nil {
+		return nil, err
+	}
+	return data[:n], nil
+}
+
+func (m *Snapshot) MarshalTo(data []byte) (n int, err error) {
+	var i int
+	_ = i
+	var l int
+	_ = l
+	data[i] = 0x8
+	i++
+	i = encodeVarintSnap(data, i, uint64(m.Crc))
+	if m.Data != nil {
+		data[i] = 0x12
+		i++
+		i = encodeVarintSnap(data, i, uint64(len(m.Data)))
+		i += copy(data[i:], m.Data)
+	}
+	if m.XXX_unrecognized != nil {
+		i += copy(data[i:], m.XXX_unrecognized)
+	}
+	return i, nil
+}
+func encodeFixed64Snap(data []byte, offset int, v uint64) int {
+	data[offset] = uint8(v)
+	data[offset+1] = uint8(v >> 8)
+	data[offset+2] = uint8(v >> 16)
+	data[offset+3] = uint8(v >> 24)
+	data[offset+4] = uint8(v >> 32)
+	data[offset+5] = uint8(v >> 40)
+	data[offset+6] = uint8(v >> 48)
+	data[offset+7] = uint8(v >> 56)
+	return offset + 8
+}
+func encodeFixed32Snap(data []byte, offset int, v uint32) int {
+	data[offset] = uint8(v)
+	data[offset+1] = uint8(v >> 8)
+	data[offset+2] = uint8(v >> 16)
+	data[offset+3] = uint8(v >> 24)
+	return offset + 4
+}
+func encodeVarintSnap(data []byte, offset int, v uint64) int {
+	for v >= 1<<7 {
+		data[offset] = uint8(v&0x7f | 0x80)
+		v >>= 7
+		offset++
+	}
+	data[offset] = uint8(v)
+	return offset + 1
+}

snap/snap.proto

@@ -0,0 +1,13 @@
+package snap;
+
+import "code.google.com/p/gogoprotobuf/gogoproto/gogo.proto";
+
+option (gogoproto.marshaler_all) = true;
+option (gogoproto.sizer_all) = true;
+option (gogoproto.unmarshaler_all) = true;
+option (gogoproto.goproto_getters_all) = false;
+
+message snapshot {
+	required uint32 crc  = 1 [(gogoproto.nullable) = false];
+	optional bytes data  = 2;
+}

snap/snapshotter.go

@@ -0,0 +1,92 @@
+package snap
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"hash/crc32"
+	"io/ioutil"
+	"log"
+	"os"
+	"path"
+	"sort"
+
+	"github.com/coreos/etcd/raft"
+)
+
+var (
+	ErrNoSnapshot  = errors.New("snap: no available snapshot")
+	ErrCRCMismatch = errors.New("snap: crc mismatch")
+	crcTable       = crc32.MakeTable(crc32.Castagnoli)
+)
+
+type Snapshotter struct {
+	dir string
+}
+
+func New(dir string) *Snapshotter {
+	return &Snapshotter{
+		dir: dir,
+	}
+}
+
+func (s *Snapshotter) Save(snapshot *raft.Snapshot) error {
+	n := fmt.Sprintf("%016x%016x%016x.snap", snapshot.ClusterId, snapshot.Term, snapshot.Index)
+	// TODO(xiangli): make raft.Snapshot a protobuf type
+	b, err := json.Marshal(snapshot)
+	if err != nil {
+		panic(err)
+	}
+	crc := crc32.Update(0, crcTable, b)
+	snap := Snapshot{Crc: crc, Data: b}
+	d, err := snap.Marshal()
+	if err != nil {
+		return err
+	}
+	return ioutil.WriteFile(path.Join(s.dir, n), d, 0666)
+}
+
+func (s *Snapshotter) Load() (*raft.Snapshot, error) {
+	dir, err := os.Open(s.dir)
+	if err != nil {
+		return nil, err
+	}
+	defer dir.Close()
+	names, err := dir.Readdirnames(-1)
+	if err != nil {
+		return nil, err
+	}
+	if len(names) == 0 {
+		return nil, ErrNoSnapshot
+	}
+	sort.Sort(sort.Reverse(sort.StringSlice(names)))
+
+	var snapshot raft.Snapshot
+	var snap Snapshot
+	var b []byte
+	for _, name := range names {
+		b, err = ioutil.ReadFile(path.Join(s.dir, name))
+		if err != nil {
+			log.Printf("Snapshotter cannot read file %v: %v", name, err)
+			continue
+		}
+		if err = snap.Unmarshal(b); err != nil {
+			log.Printf("Corruptted snapshot file %v: %v", name, err)
+			continue
+		}
+		crc := crc32.Update(0, crcTable, snap.Data)
+		if crc != snap.Crc {
+			log.Printf("Corruptted snapshot file %v: crc mismatch", name)
+			err = ErrCRCMismatch
+			continue
+		}
+		if err = json.Unmarshal(snap.Data, &snapshot); err != nil {
+			log.Printf("Corruptted snapshot file %v: %v", name, err)
+		}
+		break
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &snapshot, nil
+}

snap/snapshotter_test.go

@@ -0,0 +1,123 @@
+package snap
+
+import (
+	"fmt"
+	"hash/crc32"
+	"io/ioutil"
+	"os"
+	"path"
+	"reflect"
+	"testing"
+
+	"github.com/coreos/etcd/raft"
+)
+
+var testSnap = &raft.Snapshot{
+	ClusterId: 0xBEEF,
+	Data:      []byte("some snapshot"),
+	Nodes:     []int64{1, 2, 3},
+	Index:     1,
+	Term:      1,
+}
+
+func TestSaveAndLoad(t *testing.T) {
+	dir := path.Join(os.TempDir(), "snapshot")
+	os.Mkdir(dir, 0700)
+	defer os.RemoveAll(dir)
+	ss := New(dir)
+	err := ss.Save(testSnap)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	g, err := ss.Load()
+	if err != nil {
+		t.Errorf("err = %v, want nil", err)
+	}
+	if !reflect.DeepEqual(g, testSnap) {
+		t.Errorf("snap = %#v, want %#v", g, testSnap)
+	}
+}
+
+func TestBadCRC(t *testing.T) {
+	dir := path.Join(os.TempDir(), "snapshot")
+	os.Mkdir(dir, 0700)
+	defer os.RemoveAll(dir)
+	ss := New(dir)
+	err := ss.Save(testSnap)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { crcTable = crc32.MakeTable(crc32.Castagnoli) }()
+	// switch to use another crc table
+	// fake a crc mismatch
+	crcTable = crc32.MakeTable(crc32.Koopman)
+
+	_, err = ss.Load()
+	if err == nil || err != ErrCRCMismatch {
+		t.Errorf("err = %v, want %v", err, ErrCRCMismatch)
+	}
+}
+
+func TestFailback(t *testing.T) {
+	dir := path.Join(os.TempDir(), "snapshot")
+	os.Mkdir(dir, 0700)
+	defer os.RemoveAll(dir)
+
+	large := fmt.Sprintf("%016x%016x%016x.snap", 0xFFFF, 0xFFFF, 0xFFFF)
+	err := ioutil.WriteFile(path.Join(dir, large), []byte("bad data"), 0666)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ss := New(dir)
+	err = ss.Save(testSnap)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	g, err := ss.Load()
+	if err != nil {
+		t.Errorf("err = %v, want nil", err)
+	}
+	if !reflect.DeepEqual(g, testSnap) {
+		t.Errorf("snap = %#v, want %#v", g, testSnap)
+	}
+}
+
+func TestLoadNewestSnap(t *testing.T) {
+	dir := path.Join(os.TempDir(), "snapshot")
+	os.Mkdir(dir, 0700)
+	defer os.RemoveAll(dir)
+	ss := New(dir)
+	err := ss.Save(testSnap)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	newSnap := *testSnap
+	newSnap.Index = 5
+	err = ss.Save(&newSnap)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	g, err := ss.Load()
+	if err != nil {
+		t.Errorf("err = %v, want nil", err)
+	}
+	if !reflect.DeepEqual(g, &newSnap) {
+		t.Errorf("snap = %#v, want %#v", g, &newSnap)
+	}
+}
+
+func TestNoSnapshot(t *testing.T) {
+	dir := path.Join(os.TempDir(), "snapshot")
+	os.Mkdir(dir, 0700)
+	defer os.RemoveAll(dir)
+	ss := New(dir)
+	_, err := ss.Load()
+	if err == nil || err != ErrNoSnapshot {
+		t.Errorf("err = %v, want %v", err, ErrNoSnapshot)
+	}
+}