From 22ca80eb7e71c10eba577e96e8356922d42ed8f3 Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Thu, 30 Aug 2018 11:51:47 +0200 Subject: [PATCH] remote snapshot destruction & replication status zfs property --- endpoint/endpoint.go | 173 ++++++++++++- replication/fsrep/fsfsm.go | 118 ++++++--- replication/fsrep/stepstate_string.go | 14 +- replication/mainfsm.go | 1 + replication/pdu/pdu.pb.go | 348 ++++++++++++++++++++++---- replication/pdu/pdu.proto | 29 +++ replication/pdu/pdu_extras.go | 21 +- zfs/replication_history.go | 25 ++ zfs/versions.go | 2 +- zfs/zfs.go | 29 ++- 10 files changed, 662 insertions(+), 98 deletions(-) create mode 100644 zfs/replication_history.go diff --git a/endpoint/endpoint.go b/endpoint/endpoint.go index fca060b..408a155 100644 --- a/endpoint/endpoint.go +++ b/endpoint/endpoint.go @@ -4,6 +4,7 @@ package endpoint import ( "bytes" "context" + "fmt" "github.com/golang/protobuf/proto" "github.com/pkg/errors" "github.com/problame/go-streamrpc" @@ -56,7 +57,7 @@ func (p *Sender) ListFilesystemVersions(ctx context.Context, fs string) ([]*pdu. } rfsvs := make([]*pdu.FilesystemVersion, len(fsvs)) for i := range fsvs { - rfsvs[i] = pdu.FilesystemVersionFromZFS(fsvs[i]) + rfsvs[i] = pdu.FilesystemVersionFromZFS(&fsvs[i]) } return rfsvs, nil } @@ -80,6 +81,62 @@ func (p *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea return &pdu.SendRes{}, stream, nil } +func (p *Sender) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error) { + dp, err := zfs.NewDatasetPath(req.Filesystem) + if err != nil { + return nil, err + } + pass, err := p.FSFilter.Filter(dp) + if err != nil { + return nil, err + } + if !pass { + return nil, replication.NewFilteredError(req.Filesystem) + } + + return doDestroySnapshots(ctx, dp, req.Snapshots) +} + +// Since replication always happens from sender to receiver, this method is only ipmlemented for the sender. +// If this method returns a *zfs.DatasetDoesNotExist as an error, it might be a good indicator +// that something is wrong with the pruning logic, which is the only consumer of this method. +func (p *Sender) SnapshotReplicationStatus(ctx context.Context, req *pdu.SnapshotReplicationStatusReq) (*pdu.SnapshotReplicationStatusRes, error) { + dp, err := zfs.NewDatasetPath(req.Filesystem) + if err != nil { + return nil, err + } + pass, err := p.FSFilter.Filter(dp) + if err != nil { + return nil, err + } + if !pass { + return nil, replication.NewFilteredError(req.Filesystem) + } + + version := zfs.FilesystemVersion{ + Type: zfs.Snapshot, + Name: req.Snapshot, //FIXME validation + } + + replicated := false + switch req.Op { + case pdu.SnapshotReplicationStatusReq_Get: + replicated, err = zfs.ZFSGetReplicatedProperty(dp, &version) + if err != nil { + return nil, err + } + case pdu.SnapshotReplicationStatusReq_SetReplicated: + err = zfs.ZFSSetReplicatedProperty(dp, &version, true) + if err != nil { + return nil, err + } + replicated = true + default: + return nil, errors.Errorf("unknown opcode %v", req.Op) + } + return &pdu.SnapshotReplicationStatusRes{Replicated: replicated}, nil +} + type FSFilter interface { Filter(path *zfs.DatasetPath) (pass bool, err error) } @@ -143,7 +200,7 @@ func (e *Receiver) ListFilesystemVersions(ctx context.Context, fs string) ([]*pd rfsvs := make([]*pdu.FilesystemVersion, len(fsvs)) for i := range fsvs { - rfsvs[i] = pdu.FilesystemVersionFromZFS(fsvs[i]) + rfsvs[i] = pdu.FilesystemVersionFromZFS(&fsvs[i]) } return rfsvs, nil @@ -215,15 +272,61 @@ func (e *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, sendStream return nil } +func (e *Receiver) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error) { + dp, err := zfs.NewDatasetPath(req.Filesystem) + if err != nil { + return nil, err + } + lp, err := e.fsmap.Map(dp) + if err != nil { + return nil, err + } + if lp == nil { + return nil, errors.New("access to filesystem denied") + } + return doDestroySnapshots(ctx, lp, req.Snapshots) +} + +func doDestroySnapshots(ctx context.Context, lp *zfs.DatasetPath, snaps []*pdu.FilesystemVersion) (*pdu.DestroySnapshotsRes, error) { + fsvs := make([]*zfs.FilesystemVersion, len(snaps)) + for i, fsv := range snaps { + if fsv.Type != pdu.FilesystemVersion_Snapshot { + return nil, fmt.Errorf("version %q is not a snapshot", fsv.Name) + } + var err error + fsvs[i], err = fsv.ZFSFilesystemVersion() + if err != nil { + return nil, err + } + } + res := &pdu.DestroySnapshotsRes{ + Results: make([]*pdu.DestroySnapshotRes, len(fsvs)), + } + for i, fsv := range fsvs { + err := zfs.ZFSDestroyFilesystemVersion(lp, fsv) + errMsg := "" + if err != nil { + errMsg = err.Error() + } + res.Results[i] = &pdu.DestroySnapshotRes{ + Snapshot: pdu.FilesystemVersionFromZFS(fsv), + Error: errMsg, + } + } + return res, nil +} + // =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-= // RPC STUBS // =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-= const ( - RPCListFilesystems = "ListFilesystems" - RPCListFilesystemVersions = "ListFilesystemVersions" - RPCReceive = "Receive" - RPCSend = "Send" + RPCListFilesystems = "ListFilesystems" + RPCListFilesystemVersions = "ListFilesystemVersions" + RPCReceive = "Receive" + RPCSend = "Send" + RPCSDestroySnapshots = "DestroySnapshots" + RPCSnapshotReplicationStatus = "SnapshotReplicationStatus" ) // Remote implements an endpoint stub that uses streamrpc as a transport. @@ -320,6 +423,26 @@ func (s Remote) Receive(ctx context.Context, r *pdu.ReceiveReq, sendStream io.Re return nil } +func (s Remote) DestroySnapshots(ctx context.Context, r *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error) { + b, err := proto.Marshal(r) + if err != nil { + return nil, err + } + rb, rs, err := s.c.RequestReply(ctx, RPCSDestroySnapshots, bytes.NewBuffer(b), nil) + if err != nil { + return nil, err + } + if rs != nil { + rs.Close() + return nil, errors.New("response contains unexpected stream") + } + var res pdu.DestroySnapshotsRes + if err := proto.Unmarshal(rb.Bytes(), &res); err != nil { + return nil, err + } + return &res, nil +} + // Handler implements the server-side streamrpc.HandlerFunc for a Remote endpoint stub. type Handler struct { ep replication.Endpoint @@ -411,6 +534,44 @@ func (a *Handler) Handle(ctx context.Context, endpoint string, reqStructured *by } return bytes.NewBuffer(b), nil, err + case RPCSDestroySnapshots: + + var req pdu.DestroySnapshotsReq + if err := proto.Unmarshal(reqStructured.Bytes(), &req); err != nil { + return nil, nil, err + } + + res, err := a.ep.DestroySnapshots(ctx, &req) + if err != nil { + return nil, nil, err + } + b, err := proto.Marshal(res) + if err != nil { + return nil, nil, err + } + return bytes.NewBuffer(b), nil, nil + + case RPCSnapshotReplicationStatus: + + sender, ok := a.ep.(replication.Sender) + if !ok { + goto Err + } + + var req pdu.SnapshotReplicationStatusReq + if err := proto.Unmarshal(reqStructured.Bytes(), &req); err != nil { + return nil, nil, err + } + res, err := sender.SnapshotReplicationStatus(ctx, &req) + if err != nil { + return nil, nil, err + } + b, err := proto.Marshal(res) + if err != nil { + return nil, nil, err + } + return bytes.NewBuffer(b), nil, nil + } Err: return nil, nil, errors.New("no handler for given endpoint") diff --git a/replication/fsrep/fsfsm.go b/replication/fsrep/fsfsm.go index 52f6c31..5957dda 100644 --- a/replication/fsrep/fsfsm.go +++ b/replication/fsrep/fsfsm.go @@ -7,7 +7,6 @@ import ( "errors" "fmt" "io" - "math/bits" "net" "sync" "time" @@ -41,6 +40,7 @@ type Sender interface { // If a non-nil io.ReadCloser is returned, it is guaranteed to be closed before // any next call to the parent github.com/zrepl/zrepl/replication.Endpoint. Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) + SnapshotReplicationStatus(ctx context.Context, r *pdu.SnapshotReplicationStatusReq) (*pdu.SnapshotReplicationStatusRes, error) } // A Sender is usually part of a github.com/zrepl/zrepl/replication.Endpoint. @@ -76,17 +76,13 @@ const ( ) func (s State) fsrsf() state { - idx := bits.TrailingZeros(uint(s)) - if idx == bits.UintSize { - panic(s) + m := map[State]state{ + Ready: stateReady, + RetryWait: stateRetryWait, + PermanentError: nil, + Completed: nil, } - m := []state{ - stateReady, - stateRetryWait, - nil, - nil, - } - return m[idx] + return m[s] } type Replication struct { @@ -115,7 +111,7 @@ func BuildReplication(fs string) *ReplicationBuilder { func (b *ReplicationBuilder) AddStep(from, to FilesystemVersion) *ReplicationBuilder { step := &ReplicationStep{ - state: StepReady, + state: StepReplicationReady, parent: b.r, from: from, to: to, @@ -147,15 +143,18 @@ func NewReplicationWithPermanentError(fs string, err error) *Replication { type StepState uint const ( - StepReady StepState = 1 << iota - StepRetry + StepReplicationReady StepState = 1 << iota + StepReplicationRetry + StepMarkReplicatedReady + StepMarkReplicatedRetry StepPermanentError StepCompleted ) type FilesystemVersion interface { SnapshotTime() time.Time - RelName() string + GetName() string // name without @ or # + RelName() string // name with @ or # } type ReplicationStep struct { @@ -223,7 +222,7 @@ func stateReady(ctx context.Context, sender Sender, receiver Receiver, u updater return s.fsrsf() } - stepState := current.do(ctx, sender, receiver) + stepState := current.doReplication(ctx, sender, receiver) return u(func(f *Replication) { switch stepState { @@ -235,7 +234,9 @@ func stateReady(ctx context.Context, sender Sender, receiver Receiver, u updater } else { f.state = Completed } - case StepRetry: + case StepReplicationRetry: + fallthrough + case StepMarkReplicatedRetry: f.retryWaitUntil = time.Now().Add(RetrySleepDuration) f.state = RetryWait case StepPermanentError: @@ -292,7 +293,22 @@ func (fsr *Replication) Report() *Report { return &rep } -func (s *ReplicationStep) do(ctx context.Context, sender Sender, receiver Receiver) StepState { +func shouldRetry(err error) bool { + switch err { + case io.EOF: + fallthrough + case io.ErrUnexpectedEOF: + fallthrough + case io.ErrClosedPipe: + return true + } + if _, ok := err.(net.Error); ok { + return true + } + return false +} + +func (s *ReplicationStep) doReplication(ctx context.Context, sender Sender, receiver Receiver) StepState { fs := s.parent.fs @@ -305,17 +321,8 @@ func (s *ReplicationStep) do(ctx context.Context, sender Sender, receiver Receiv defer s.lock.Unlock() s.err = err - switch err { - case io.EOF: - fallthrough - case io.ErrUnexpectedEOF: - fallthrough - case io.ErrClosedPipe: - s.state = StepRetry - return s.state - } - if _, ok := err.(net.Error); ok { - s.state = StepRetry + if shouldRetry(s.err) { + s.state = StepReplicationRetry return s.state } s.state = StepPermanentError @@ -326,7 +333,7 @@ func (s *ReplicationStep) do(ctx context.Context, sender Sender, receiver Receiv s.lock.Lock() defer s.lock.Unlock() s.err = nil - s.state = StepCompleted + s.state = StepMarkReplicatedReady return s.state } @@ -371,8 +378,57 @@ func (s *ReplicationStep) do(ctx context.Context, sender Sender, receiver Receiv return updateStateError(err) } log.Info("receive finished") - return updateStateCompleted() + updateStateCompleted() + + return s.doMarkReplicated(ctx, sender) + +} + +func (s *ReplicationStep) doMarkReplicated(ctx context.Context, sender Sender) StepState { + + log := getLogger(ctx). + WithField("filesystem", s.parent.fs). + WithField("step", s.String()) + + updateStateError := func(err error) StepState { + s.lock.Lock() + defer s.lock.Unlock() + + s.err = err + if shouldRetry(s.err) { + s.state = StepMarkReplicatedRetry + return s.state + } + s.state = StepPermanentError + return s.state + } + + updateStateCompleted := func() StepState { + s.lock.Lock() + defer s.lock.Unlock() + s.state = StepCompleted + return s.state + } + + log.Info("mark snapshot as replicated") + req := pdu.SnapshotReplicationStatusReq{ + Filesystem: s.parent.fs, + Snapshot: s.to.GetName(), + Op: pdu.SnapshotReplicationStatusReq_SetReplicated, + } + res, err := sender.SnapshotReplicationStatus(ctx, &req) + if err != nil { + log.WithError(err).Error("error marking snapshot as replicated") + return updateStateError(err) + } + if res.Replicated != true { + err := fmt.Errorf("sender did not report snapshot as replicated") + log.Error(err.Error()) + return updateStateError(err) + } + + return updateStateCompleted() } func (s *ReplicationStep) String() string { diff --git a/replication/fsrep/stepstate_string.go b/replication/fsrep/stepstate_string.go index 8228bfc..aeac7e1 100644 --- a/replication/fsrep/stepstate_string.go +++ b/replication/fsrep/stepstate_string.go @@ -5,13 +5,15 @@ package fsrep import "strconv" const ( - _StepState_name_0 = "StepReadyStepRetry" - _StepState_name_1 = "StepPermanentError" - _StepState_name_2 = "StepCompleted" + _StepState_name_0 = "StepReplicationReadyStepReplicationRetry" + _StepState_name_1 = "StepMarkReplicatedReady" + _StepState_name_2 = "StepMarkReplicatedRetry" + _StepState_name_3 = "StepPermanentError" + _StepState_name_4 = "StepCompleted" ) var ( - _StepState_index_0 = [...]uint8{0, 9, 18} + _StepState_index_0 = [...]uint8{0, 20, 40} ) func (i StepState) String() string { @@ -23,6 +25,10 @@ func (i StepState) String() string { return _StepState_name_1 case i == 8: return _StepState_name_2 + case i == 16: + return _StepState_name_3 + case i == 32: + return _StepState_name_4 default: return "StepState(" + strconv.FormatInt(int64(i), 10) + ")" } diff --git a/replication/mainfsm.go b/replication/mainfsm.go index 7d9bfc5..2fc5cc3 100644 --- a/replication/mainfsm.go +++ b/replication/mainfsm.go @@ -93,6 +93,7 @@ type Endpoint interface { ListFilesystems(ctx context.Context) ([]*pdu.Filesystem, error) // FIXME document FilteredError handling ListFilesystemVersions(ctx context.Context, fs string) ([]*pdu.FilesystemVersion, error) // fix depS + DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error) } type Sender interface { diff --git a/replication/pdu/pdu.pb.go b/replication/pdu/pdu.pb.go index a770a17..db361c3 100644 --- a/replication/pdu/pdu.pb.go +++ b/replication/pdu/pdu.pb.go @@ -38,7 +38,30 @@ func (x FilesystemVersion_VersionType) String() string { return proto.EnumName(FilesystemVersion_VersionType_name, int32(x)) } func (FilesystemVersion_VersionType) EnumDescriptor() ([]byte, []int) { - return fileDescriptor_pdu_25ca9e5977a7ce3f, []int{5, 0} + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{5, 0} +} + +type SnapshotReplicationStatusReq_Op int32 + +const ( + SnapshotReplicationStatusReq_Get SnapshotReplicationStatusReq_Op = 0 + SnapshotReplicationStatusReq_SetReplicated SnapshotReplicationStatusReq_Op = 1 +) + +var SnapshotReplicationStatusReq_Op_name = map[int32]string{ + 0: "Get", + 1: "SetReplicated", +} +var SnapshotReplicationStatusReq_Op_value = map[string]int32{ + "Get": 0, + "SetReplicated": 1, +} + +func (x SnapshotReplicationStatusReq_Op) String() string { + return proto.EnumName(SnapshotReplicationStatusReq_Op_name, int32(x)) +} +func (SnapshotReplicationStatusReq_Op) EnumDescriptor() ([]byte, []int) { + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{14, 0} } type ListFilesystemReq struct { @@ -51,7 +74,7 @@ func (m *ListFilesystemReq) Reset() { *m = ListFilesystemReq{} } func (m *ListFilesystemReq) String() string { return proto.CompactTextString(m) } func (*ListFilesystemReq) ProtoMessage() {} func (*ListFilesystemReq) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_25ca9e5977a7ce3f, []int{0} + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{0} } func (m *ListFilesystemReq) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_ListFilesystemReq.Unmarshal(m, b) @@ -82,7 +105,7 @@ func (m *ListFilesystemRes) Reset() { *m = ListFilesystemRes{} } func (m *ListFilesystemRes) String() string { return proto.CompactTextString(m) } func (*ListFilesystemRes) ProtoMessage() {} func (*ListFilesystemRes) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_25ca9e5977a7ce3f, []int{1} + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{1} } func (m *ListFilesystemRes) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_ListFilesystemRes.Unmarshal(m, b) @@ -121,7 +144,7 @@ func (m *Filesystem) Reset() { *m = Filesystem{} } func (m *Filesystem) String() string { return proto.CompactTextString(m) } func (*Filesystem) ProtoMessage() {} func (*Filesystem) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_25ca9e5977a7ce3f, []int{2} + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{2} } func (m *Filesystem) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_Filesystem.Unmarshal(m, b) @@ -166,7 +189,7 @@ func (m *ListFilesystemVersionsReq) Reset() { *m = ListFilesystemVersion func (m *ListFilesystemVersionsReq) String() string { return proto.CompactTextString(m) } func (*ListFilesystemVersionsReq) ProtoMessage() {} func (*ListFilesystemVersionsReq) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_25ca9e5977a7ce3f, []int{3} + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{3} } func (m *ListFilesystemVersionsReq) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_ListFilesystemVersionsReq.Unmarshal(m, b) @@ -204,7 +227,7 @@ func (m *ListFilesystemVersionsRes) Reset() { *m = ListFilesystemVersion func (m *ListFilesystemVersionsRes) String() string { return proto.CompactTextString(m) } func (*ListFilesystemVersionsRes) ProtoMessage() {} func (*ListFilesystemVersionsRes) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_25ca9e5977a7ce3f, []int{4} + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{4} } func (m *ListFilesystemVersionsRes) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_ListFilesystemVersionsRes.Unmarshal(m, b) @@ -246,7 +269,7 @@ func (m *FilesystemVersion) Reset() { *m = FilesystemVersion{} } func (m *FilesystemVersion) String() string { return proto.CompactTextString(m) } func (*FilesystemVersion) ProtoMessage() {} func (*FilesystemVersion) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_25ca9e5977a7ce3f, []int{5} + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{5} } func (m *FilesystemVersion) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_FilesystemVersion.Unmarshal(m, b) @@ -325,7 +348,7 @@ func (m *SendReq) Reset() { *m = SendReq{} } func (m *SendReq) String() string { return proto.CompactTextString(m) } func (*SendReq) ProtoMessage() {} func (*SendReq) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_25ca9e5977a7ce3f, []int{6} + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{6} } func (m *SendReq) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_SendReq.Unmarshal(m, b) @@ -399,7 +422,7 @@ func (m *Property) Reset() { *m = Property{} } func (m *Property) String() string { return proto.CompactTextString(m) } func (*Property) ProtoMessage() {} func (*Property) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_25ca9e5977a7ce3f, []int{7} + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{7} } func (m *Property) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_Property.Unmarshal(m, b) @@ -446,7 +469,7 @@ func (m *SendRes) Reset() { *m = SendRes{} } func (m *SendRes) String() string { return proto.CompactTextString(m) } func (*SendRes) ProtoMessage() {} func (*SendRes) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_25ca9e5977a7ce3f, []int{8} + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{8} } func (m *SendRes) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_SendRes.Unmarshal(m, b) @@ -493,7 +516,7 @@ func (m *ReceiveReq) Reset() { *m = ReceiveReq{} } func (m *ReceiveReq) String() string { return proto.CompactTextString(m) } func (*ReceiveReq) ProtoMessage() {} func (*ReceiveReq) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_25ca9e5977a7ce3f, []int{9} + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{9} } func (m *ReceiveReq) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_ReceiveReq.Unmarshal(m, b) @@ -537,7 +560,7 @@ func (m *ReceiveRes) Reset() { *m = ReceiveRes{} } func (m *ReceiveRes) String() string { return proto.CompactTextString(m) } func (*ReceiveRes) ProtoMessage() {} func (*ReceiveRes) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_25ca9e5977a7ce3f, []int{10} + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{10} } func (m *ReceiveRes) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_ReceiveRes.Unmarshal(m, b) @@ -557,6 +580,229 @@ func (m *ReceiveRes) XXX_DiscardUnknown() { var xxx_messageInfo_ReceiveRes proto.InternalMessageInfo +type DestroySnapshotsReq struct { + Filesystem string `protobuf:"bytes,1,opt,name=Filesystem,proto3" json:"Filesystem,omitempty"` + // Path to filesystem, snapshot or bookmark to be destroyed + Snapshots []*FilesystemVersion `protobuf:"bytes,2,rep,name=Snapshots,proto3" json:"Snapshots,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *DestroySnapshotsReq) Reset() { *m = DestroySnapshotsReq{} } +func (m *DestroySnapshotsReq) String() string { return proto.CompactTextString(m) } +func (*DestroySnapshotsReq) ProtoMessage() {} +func (*DestroySnapshotsReq) Descriptor() ([]byte, []int) { + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{11} +} +func (m *DestroySnapshotsReq) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_DestroySnapshotsReq.Unmarshal(m, b) +} +func (m *DestroySnapshotsReq) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_DestroySnapshotsReq.Marshal(b, m, deterministic) +} +func (dst *DestroySnapshotsReq) XXX_Merge(src proto.Message) { + xxx_messageInfo_DestroySnapshotsReq.Merge(dst, src) +} +func (m *DestroySnapshotsReq) XXX_Size() int { + return xxx_messageInfo_DestroySnapshotsReq.Size(m) +} +func (m *DestroySnapshotsReq) XXX_DiscardUnknown() { + xxx_messageInfo_DestroySnapshotsReq.DiscardUnknown(m) +} + +var xxx_messageInfo_DestroySnapshotsReq proto.InternalMessageInfo + +func (m *DestroySnapshotsReq) GetFilesystem() string { + if m != nil { + return m.Filesystem + } + return "" +} + +func (m *DestroySnapshotsReq) GetSnapshots() []*FilesystemVersion { + if m != nil { + return m.Snapshots + } + return nil +} + +type DestroySnapshotRes struct { + Snapshot *FilesystemVersion `protobuf:"bytes,1,opt,name=Snapshot,proto3" json:"Snapshot,omitempty"` + Error string `protobuf:"bytes,2,opt,name=Error,proto3" json:"Error,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *DestroySnapshotRes) Reset() { *m = DestroySnapshotRes{} } +func (m *DestroySnapshotRes) String() string { return proto.CompactTextString(m) } +func (*DestroySnapshotRes) ProtoMessage() {} +func (*DestroySnapshotRes) Descriptor() ([]byte, []int) { + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{12} +} +func (m *DestroySnapshotRes) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_DestroySnapshotRes.Unmarshal(m, b) +} +func (m *DestroySnapshotRes) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_DestroySnapshotRes.Marshal(b, m, deterministic) +} +func (dst *DestroySnapshotRes) XXX_Merge(src proto.Message) { + xxx_messageInfo_DestroySnapshotRes.Merge(dst, src) +} +func (m *DestroySnapshotRes) XXX_Size() int { + return xxx_messageInfo_DestroySnapshotRes.Size(m) +} +func (m *DestroySnapshotRes) XXX_DiscardUnknown() { + xxx_messageInfo_DestroySnapshotRes.DiscardUnknown(m) +} + +var xxx_messageInfo_DestroySnapshotRes proto.InternalMessageInfo + +func (m *DestroySnapshotRes) GetSnapshot() *FilesystemVersion { + if m != nil { + return m.Snapshot + } + return nil +} + +func (m *DestroySnapshotRes) GetError() string { + if m != nil { + return m.Error + } + return "" +} + +type DestroySnapshotsRes struct { + Results []*DestroySnapshotRes `protobuf:"bytes,1,rep,name=Results,proto3" json:"Results,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *DestroySnapshotsRes) Reset() { *m = DestroySnapshotsRes{} } +func (m *DestroySnapshotsRes) String() string { return proto.CompactTextString(m) } +func (*DestroySnapshotsRes) ProtoMessage() {} +func (*DestroySnapshotsRes) Descriptor() ([]byte, []int) { + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{13} +} +func (m *DestroySnapshotsRes) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_DestroySnapshotsRes.Unmarshal(m, b) +} +func (m *DestroySnapshotsRes) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_DestroySnapshotsRes.Marshal(b, m, deterministic) +} +func (dst *DestroySnapshotsRes) XXX_Merge(src proto.Message) { + xxx_messageInfo_DestroySnapshotsRes.Merge(dst, src) +} +func (m *DestroySnapshotsRes) XXX_Size() int { + return xxx_messageInfo_DestroySnapshotsRes.Size(m) +} +func (m *DestroySnapshotsRes) XXX_DiscardUnknown() { + xxx_messageInfo_DestroySnapshotsRes.DiscardUnknown(m) +} + +var xxx_messageInfo_DestroySnapshotsRes proto.InternalMessageInfo + +func (m *DestroySnapshotsRes) GetResults() []*DestroySnapshotRes { + if m != nil { + return m.Results + } + return nil +} + +type SnapshotReplicationStatusReq struct { + Filesystem string `protobuf:"bytes,1,opt,name=Filesystem,proto3" json:"Filesystem,omitempty"` + Snapshot string `protobuf:"bytes,2,opt,name=Snapshot,proto3" json:"Snapshot,omitempty"` + Op SnapshotReplicationStatusReq_Op `protobuf:"varint,3,opt,name=op,proto3,enum=pdu.SnapshotReplicationStatusReq_Op" json:"op,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *SnapshotReplicationStatusReq) Reset() { *m = SnapshotReplicationStatusReq{} } +func (m *SnapshotReplicationStatusReq) String() string { return proto.CompactTextString(m) } +func (*SnapshotReplicationStatusReq) ProtoMessage() {} +func (*SnapshotReplicationStatusReq) Descriptor() ([]byte, []int) { + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{14} +} +func (m *SnapshotReplicationStatusReq) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_SnapshotReplicationStatusReq.Unmarshal(m, b) +} +func (m *SnapshotReplicationStatusReq) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_SnapshotReplicationStatusReq.Marshal(b, m, deterministic) +} +func (dst *SnapshotReplicationStatusReq) XXX_Merge(src proto.Message) { + xxx_messageInfo_SnapshotReplicationStatusReq.Merge(dst, src) +} +func (m *SnapshotReplicationStatusReq) XXX_Size() int { + return xxx_messageInfo_SnapshotReplicationStatusReq.Size(m) +} +func (m *SnapshotReplicationStatusReq) XXX_DiscardUnknown() { + xxx_messageInfo_SnapshotReplicationStatusReq.DiscardUnknown(m) +} + +var xxx_messageInfo_SnapshotReplicationStatusReq proto.InternalMessageInfo + +func (m *SnapshotReplicationStatusReq) GetFilesystem() string { + if m != nil { + return m.Filesystem + } + return "" +} + +func (m *SnapshotReplicationStatusReq) GetSnapshot() string { + if m != nil { + return m.Snapshot + } + return "" +} + +func (m *SnapshotReplicationStatusReq) GetOp() SnapshotReplicationStatusReq_Op { + if m != nil { + return m.Op + } + return SnapshotReplicationStatusReq_Get +} + +type SnapshotReplicationStatusRes struct { + Replicated bool `protobuf:"varint,1,opt,name=Replicated,proto3" json:"Replicated,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *SnapshotReplicationStatusRes) Reset() { *m = SnapshotReplicationStatusRes{} } +func (m *SnapshotReplicationStatusRes) String() string { return proto.CompactTextString(m) } +func (*SnapshotReplicationStatusRes) ProtoMessage() {} +func (*SnapshotReplicationStatusRes) Descriptor() ([]byte, []int) { + return fileDescriptor_pdu_e1dccbd3b8cde5a3, []int{15} +} +func (m *SnapshotReplicationStatusRes) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_SnapshotReplicationStatusRes.Unmarshal(m, b) +} +func (m *SnapshotReplicationStatusRes) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_SnapshotReplicationStatusRes.Marshal(b, m, deterministic) +} +func (dst *SnapshotReplicationStatusRes) XXX_Merge(src proto.Message) { + xxx_messageInfo_SnapshotReplicationStatusRes.Merge(dst, src) +} +func (m *SnapshotReplicationStatusRes) XXX_Size() int { + return xxx_messageInfo_SnapshotReplicationStatusRes.Size(m) +} +func (m *SnapshotReplicationStatusRes) XXX_DiscardUnknown() { + xxx_messageInfo_SnapshotReplicationStatusRes.DiscardUnknown(m) +} + +var xxx_messageInfo_SnapshotReplicationStatusRes proto.InternalMessageInfo + +func (m *SnapshotReplicationStatusRes) GetReplicated() bool { + if m != nil { + return m.Replicated + } + return false +} + func init() { proto.RegisterType((*ListFilesystemReq)(nil), "pdu.ListFilesystemReq") proto.RegisterType((*ListFilesystemRes)(nil), "pdu.ListFilesystemRes") @@ -569,39 +815,55 @@ func init() { proto.RegisterType((*SendRes)(nil), "pdu.SendRes") proto.RegisterType((*ReceiveReq)(nil), "pdu.ReceiveReq") proto.RegisterType((*ReceiveRes)(nil), "pdu.ReceiveRes") + proto.RegisterType((*DestroySnapshotsReq)(nil), "pdu.DestroySnapshotsReq") + proto.RegisterType((*DestroySnapshotRes)(nil), "pdu.DestroySnapshotRes") + proto.RegisterType((*DestroySnapshotsRes)(nil), "pdu.DestroySnapshotsRes") + proto.RegisterType((*SnapshotReplicationStatusReq)(nil), "pdu.SnapshotReplicationStatusReq") + proto.RegisterType((*SnapshotReplicationStatusRes)(nil), "pdu.SnapshotReplicationStatusRes") proto.RegisterEnum("pdu.FilesystemVersion_VersionType", FilesystemVersion_VersionType_name, FilesystemVersion_VersionType_value) + proto.RegisterEnum("pdu.SnapshotReplicationStatusReq_Op", SnapshotReplicationStatusReq_Op_name, SnapshotReplicationStatusReq_Op_value) } -func init() { proto.RegisterFile("pdu.proto", fileDescriptor_pdu_25ca9e5977a7ce3f) } +func init() { proto.RegisterFile("pdu.proto", fileDescriptor_pdu_e1dccbd3b8cde5a3) } -var fileDescriptor_pdu_25ca9e5977a7ce3f = []byte{ - // 445 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x53, 0xc1, 0x6e, 0xd3, 0x40, - 0x10, 0x65, 0x13, 0xa7, 0x38, 0x93, 0xd2, 0xa6, 0x4b, 0x85, 0x0c, 0x42, 0x28, 0xda, 0x53, 0x40, - 0x22, 0x12, 0x01, 0x71, 0xe1, 0xd6, 0xa2, 0xf4, 0x82, 0xa0, 0xda, 0x9a, 0xaa, 0x57, 0x17, 0x8f, - 0x54, 0x2b, 0xb1, 0x77, 0xbb, 0x63, 0x23, 0xe5, 0x73, 0xf8, 0x2b, 0x3e, 0x07, 0x79, 0x6a, 0x27, - 0x4b, 0x0c, 0x52, 0x4e, 0x99, 0xf7, 0x66, 0x32, 0xf3, 0xe6, 0xcd, 0x1a, 0x86, 0x36, 0xad, 0x66, - 0xd6, 0x99, 0xd2, 0xc8, 0xbe, 0x4d, 0x2b, 0xf5, 0x14, 0x4e, 0xbe, 0x64, 0x54, 0x2e, 0xb2, 0x15, - 0xd2, 0x9a, 0x4a, 0xcc, 0x35, 0xde, 0xab, 0x45, 0x97, 0x24, 0xf9, 0x0e, 0x46, 0x5b, 0x82, 0x22, - 0x31, 0xe9, 0x4f, 0x47, 0xf3, 0xe3, 0x59, 0xdd, 0xcf, 0x2b, 0xf4, 0x6b, 0xd4, 0x19, 0xc0, 0x16, - 0x4a, 0x09, 0xc1, 0x65, 0x52, 0xde, 0x45, 0x62, 0x22, 0xa6, 0x43, 0xcd, 0xb1, 0x9c, 0xc0, 0x48, - 0x23, 0x55, 0x39, 0xc6, 0x66, 0x89, 0x45, 0xd4, 0xe3, 0x94, 0x4f, 0xa9, 0x4f, 0xf0, 0xfc, 0x6f, - 0x2d, 0xd7, 0xe8, 0x28, 0x33, 0x05, 0x69, 0xbc, 0x97, 0xaf, 0xfc, 0x01, 0x4d, 0x63, 0x8f, 0x51, - 0xdf, 0xfe, 0xff, 0x67, 0x92, 0x73, 0x08, 0x5b, 0xd8, 0x6c, 0xf3, 0x6c, 0x67, 0x9b, 0x26, 0xad, - 0x37, 0x75, 0xea, 0xb7, 0x80, 0x93, 0x4e, 0x5e, 0x7e, 0x84, 0x20, 0x5e, 0x5b, 0x64, 0x01, 0x47, - 0x73, 0xf5, 0xef, 0x2e, 0xb3, 0xe6, 0xb7, 0xae, 0xd4, 0x5c, 0x5f, 0x3b, 0xf2, 0x35, 0xc9, 0xb1, - 0x59, 0x9b, 0xe3, 0x9a, 0xbb, 0xa8, 0xb2, 0x34, 0xea, 0x4f, 0xc4, 0x34, 0xd0, 0x1c, 0xcb, 0x97, - 0x30, 0x3c, 0x77, 0x98, 0x94, 0x18, 0xdf, 0x5c, 0x44, 0x01, 0x27, 0xb6, 0x84, 0x7c, 0x01, 0x21, - 0x83, 0xcc, 0x14, 0xd1, 0x80, 0x3b, 0x6d, 0xb0, 0x7a, 0x0d, 0x23, 0x6f, 0xac, 0x3c, 0x84, 0xf0, - 0xaa, 0x48, 0x2c, 0xdd, 0x99, 0x72, 0xfc, 0xa8, 0x46, 0x67, 0xc6, 0x2c, 0xf3, 0xc4, 0x2d, 0xc7, - 0x42, 0xfd, 0x12, 0xf0, 0xf8, 0x0a, 0x8b, 0x74, 0x0f, 0x5f, 0x6b, 0x91, 0x0b, 0x67, 0xf2, 0x56, - 0x78, 0x1d, 0xcb, 0x23, 0xe8, 0xc5, 0x86, 0x65, 0x0f, 0x75, 0x2f, 0x36, 0xbb, 0xa7, 0x0d, 0x3a, - 0xa7, 0x65, 0xe1, 0x26, 0xb7, 0x0e, 0x89, 0x58, 0x78, 0xa8, 0x37, 0x58, 0x9e, 0xc2, 0xe0, 0x33, - 0xa6, 0x95, 0x8d, 0x0e, 0x38, 0xf1, 0x00, 0xd4, 0x07, 0x08, 0x2f, 0x9d, 0xb1, 0xe8, 0xca, 0xf5, - 0xc6, 0x3c, 0xe1, 0x99, 0x77, 0x0a, 0x83, 0xeb, 0x64, 0x55, 0xb5, 0x8e, 0x3e, 0x00, 0x75, 0xdb, - 0x2e, 0x46, 0x72, 0x0a, 0xc7, 0xdf, 0x09, 0x53, 0x5f, 0x98, 0xe0, 0x01, 0xbb, 0xb4, 0x7c, 0x0b, - 0xd0, 0x8c, 0xca, 0x90, 0xa2, 0x1e, 0xbf, 0x8f, 0x27, 0x7c, 0xd9, 0x56, 0x81, 0xf6, 0x0a, 0xd4, - 0x0d, 0x80, 0xc6, 0x1f, 0x98, 0xfd, 0xc4, 0x7d, 0xfc, 0x7b, 0x03, 0xe3, 0xf3, 0x15, 0x26, 0x6e, - 0xf7, 0xed, 0x87, 0xba, 0xc3, 0xab, 0x43, 0xaf, 0x33, 0xdd, 0x1e, 0xf0, 0xb7, 0xfb, 0xfe, 0x4f, - 0x00, 0x00, 0x00, 0xff, 0xff, 0xfd, 0xc4, 0x8d, 0xb7, 0xc8, 0x03, 0x00, 0x00, +var fileDescriptor_pdu_e1dccbd3b8cde5a3 = []byte{ + // 595 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x54, 0xdd, 0x6e, 0x12, 0x41, + 0x14, 0x76, 0x17, 0x5a, 0x96, 0x43, 0x4b, 0x61, 0xda, 0xe8, 0xda, 0x34, 0x86, 0x4c, 0xbc, 0x40, + 0x13, 0x49, 0x44, 0xe2, 0x8d, 0x89, 0x17, 0x6d, 0x05, 0x2f, 0x8c, 0x34, 0x03, 0x36, 0xbd, 0x32, + 0xd9, 0x76, 0x4f, 0xd2, 0x0d, 0x2c, 0x33, 0x9d, 0xd9, 0x35, 0xe1, 0x71, 0x7c, 0x05, 0x9f, 0xc6, + 0xc7, 0x31, 0x33, 0xec, 0xcf, 0x08, 0x2d, 0xe1, 0x8a, 0xf9, 0xce, 0xf9, 0x38, 0xe7, 0x3b, 0x7f, + 0x0b, 0x75, 0x11, 0xa6, 0x3d, 0x21, 0x79, 0xc2, 0x49, 0x45, 0x84, 0x29, 0x3d, 0x86, 0xf6, 0xb7, + 0x48, 0x25, 0xc3, 0x68, 0x8e, 0x6a, 0xa9, 0x12, 0x8c, 0x19, 0x3e, 0xd0, 0xe1, 0xa6, 0x51, 0x91, + 0xf7, 0xd0, 0x28, 0x0d, 0xca, 0x77, 0x3a, 0x95, 0x6e, 0xa3, 0x7f, 0xd4, 0xd3, 0xf1, 0x2c, 0xa2, + 0xcd, 0xa1, 0xe7, 0x00, 0x25, 0x24, 0x04, 0xaa, 0x57, 0x41, 0x72, 0xef, 0x3b, 0x1d, 0xa7, 0x5b, + 0x67, 0xe6, 0x4d, 0x3a, 0xd0, 0x60, 0xa8, 0xd2, 0x18, 0xa7, 0x7c, 0x86, 0x0b, 0xdf, 0x35, 0x2e, + 0xdb, 0x44, 0x3f, 0xc1, 0xcb, 0xff, 0xb5, 0x5c, 0xa3, 0x54, 0x11, 0x5f, 0x28, 0x86, 0x0f, 0xe4, + 0x95, 0x9d, 0x20, 0x0b, 0x6c, 0x59, 0xe8, 0xf8, 0xe9, 0x3f, 0x2b, 0xd2, 0x07, 0x2f, 0x87, 0x59, + 0x35, 0xcf, 0xd7, 0xaa, 0xc9, 0xdc, 0xac, 0xe0, 0xd1, 0xbf, 0x0e, 0xb4, 0x37, 0xfc, 0xe4, 0x23, + 0x54, 0xa7, 0x4b, 0x81, 0x46, 0x40, 0xb3, 0x4f, 0x1f, 0x8f, 0xd2, 0xcb, 0x7e, 0x35, 0x93, 0x19, + 0xbe, 0xee, 0xc8, 0xf7, 0x20, 0xc6, 0xac, 0x6c, 0xf3, 0xd6, 0xb6, 0x51, 0x1a, 0x85, 0x7e, 0xa5, + 0xe3, 0x74, 0xab, 0xcc, 0xbc, 0xc9, 0x19, 0xd4, 0x2f, 0x24, 0x06, 0x09, 0x4e, 0x6f, 0x46, 0x7e, + 0xd5, 0x38, 0x4a, 0x03, 0x39, 0x05, 0xcf, 0x80, 0x88, 0x2f, 0xfc, 0x3d, 0x13, 0xa9, 0xc0, 0xf4, + 0x0d, 0x34, 0xac, 0xb4, 0xe4, 0x00, 0xbc, 0xc9, 0x22, 0x10, 0xea, 0x9e, 0x27, 0xad, 0x67, 0x1a, + 0x9d, 0x73, 0x3e, 0x8b, 0x03, 0x39, 0x6b, 0x39, 0xf4, 0xb7, 0x03, 0xb5, 0x09, 0x2e, 0xc2, 0x1d, + 0xfa, 0xaa, 0x45, 0x0e, 0x25, 0x8f, 0x73, 0xe1, 0xfa, 0x4d, 0x9a, 0xe0, 0x4e, 0xb9, 0x91, 0x5d, + 0x67, 0xee, 0x94, 0xaf, 0x8f, 0xb6, 0xba, 0x31, 0x5a, 0x23, 0x9c, 0xc7, 0x42, 0xa2, 0x52, 0x46, + 0xb8, 0xc7, 0x0a, 0x4c, 0x4e, 0x60, 0xef, 0x12, 0xc3, 0x54, 0xf8, 0xfb, 0xc6, 0xb1, 0x02, 0x74, + 0x00, 0xde, 0x95, 0xe4, 0x02, 0x65, 0xb2, 0x2c, 0x9a, 0xe7, 0x58, 0xcd, 0x3b, 0x81, 0xbd, 0xeb, + 0x60, 0x9e, 0xe6, 0x1d, 0x5d, 0x01, 0x7a, 0x9b, 0x17, 0xa6, 0x48, 0x17, 0x8e, 0x7e, 0x28, 0x0c, + 0x6d, 0x61, 0x8e, 0x49, 0xb0, 0x6e, 0x26, 0xef, 0x00, 0xb2, 0x54, 0x11, 0x2a, 0xdf, 0x35, 0xfb, + 0x71, 0x68, 0x26, 0x9b, 0x2b, 0x60, 0x16, 0x81, 0xde, 0x00, 0x30, 0xbc, 0xc3, 0xe8, 0x17, 0xee, + 0xd2, 0xbf, 0xb7, 0xd0, 0xba, 0x98, 0x63, 0x20, 0xd7, 0x77, 0xdf, 0x63, 0x1b, 0x76, 0x7a, 0x60, + 0x45, 0x56, 0x74, 0x06, 0xc7, 0x97, 0xa8, 0x12, 0xc9, 0x97, 0xf9, 0x20, 0x77, 0x39, 0x04, 0x32, + 0x80, 0x7a, 0xc1, 0xcf, 0x8a, 0x79, 0x6a, 0xd9, 0x4b, 0x22, 0xfd, 0x09, 0x64, 0x2d, 0x59, 0x76, + 0x37, 0x39, 0x34, 0x99, 0xb6, 0xdc, 0x4d, 0xce, 0xd3, 0x83, 0xf9, 0x22, 0x25, 0x97, 0xf9, 0x60, + 0x0c, 0xa0, 0x5f, 0x1f, 0x2b, 0x46, 0x7f, 0x69, 0x6a, 0xba, 0x01, 0xf3, 0x24, 0xbf, 0xcb, 0x17, + 0x26, 0xfe, 0xa6, 0x14, 0x96, 0xf3, 0xe8, 0x1f, 0x07, 0xce, 0x4a, 0x87, 0x98, 0x47, 0x77, 0x66, + 0xff, 0x27, 0x49, 0x90, 0xa4, 0x3b, 0x35, 0xe8, 0xd4, 0x2a, 0x6a, 0xa5, 0xb1, 0x14, 0x3f, 0x00, + 0x97, 0x0b, 0xb3, 0xd9, 0xcd, 0xfe, 0x6b, 0x23, 0x65, 0x5b, 0xaa, 0xde, 0x58, 0x30, 0x97, 0x0b, + 0xda, 0x01, 0x77, 0x2c, 0x48, 0x0d, 0x2a, 0x23, 0xd4, 0xc7, 0xd6, 0x86, 0xc3, 0x09, 0x16, 0x7f, + 0xc0, 0xb0, 0xe5, 0xd0, 0xcf, 0x5b, 0x35, 0x2b, 0xad, 0xb9, 0xe4, 0x67, 0x7b, 0x6a, 0x59, 0x6e, + 0xf7, 0xcd, 0x77, 0xfc, 0xc3, 0xbf, 0x00, 0x00, 0x00, 0xff, 0xff, 0xf6, 0x1a, 0xe7, 0xf6, 0xd4, + 0x05, 0x00, 0x00, } diff --git a/replication/pdu/pdu.proto b/replication/pdu/pdu.proto index cac47b1..0d1103c 100644 --- a/replication/pdu/pdu.proto +++ b/replication/pdu/pdu.proto @@ -76,3 +76,32 @@ message ReceiveReq { } message ReceiveRes {} + +message DestroySnapshotsReq { + string Filesystem = 1; + // Path to filesystem, snapshot or bookmark to be destroyed + repeated FilesystemVersion Snapshots = 2; +} + +message DestroySnapshotRes { + FilesystemVersion Snapshot = 1; + string Error = 2; +} + +message DestroySnapshotsRes { + repeated DestroySnapshotRes Results = 1; +} + +message SnapshotReplicationStatusReq { + string Filesystem = 1; + string Snapshot = 2; + enum Op { + Get = 0; + SetReplicated = 1; + } + Op op = 3; +} + +message SnapshotReplicationStatusRes { + bool Replicated = 1; +} diff --git a/replication/pdu/pdu_extras.go b/replication/pdu/pdu_extras.go index 4d718a8..9cf52c9 100644 --- a/replication/pdu/pdu_extras.go +++ b/replication/pdu/pdu_extras.go @@ -7,7 +7,10 @@ import ( ) func (v *FilesystemVersion) RelName() string { - zv := v.ZFSFilesystemVersion() + zv, err := v.ZFSFilesystemVersion() + if err != nil { + panic(err) + } return zv.String() } @@ -22,7 +25,7 @@ func (v FilesystemVersion_VersionType) ZFSVersionType() zfs.VersionType { } } -func FilesystemVersionFromZFS(fsv zfs.FilesystemVersion) *FilesystemVersion { +func FilesystemVersionFromZFS(fsv *zfs.FilesystemVersion) *FilesystemVersion { var t FilesystemVersion_VersionType switch fsv.Type { case zfs.Bookmark: @@ -58,14 +61,10 @@ func (v *FilesystemVersion) SnapshotTime() time.Time { return t } -func (v *FilesystemVersion) ZFSFilesystemVersion() *zfs.FilesystemVersion { - ct := time.Time{} - if v.Creation != "" { - var err error - ct, err = time.Parse(time.RFC3339, v.Creation) - if err != nil { - panic(err) - } +func (v *FilesystemVersion) ZFSFilesystemVersion() (*zfs.FilesystemVersion, error) { + ct, err := v.CreationAsTime() + if err != nil { + return nil, err } return &zfs.FilesystemVersion{ Type: v.Type.ZFSVersionType(), @@ -73,5 +72,5 @@ func (v *FilesystemVersion) ZFSFilesystemVersion() *zfs.FilesystemVersion { Guid: v.Guid, CreateTXG: v.CreateTXG, Creation: ct, - } + }, nil } diff --git a/zfs/replication_history.go b/zfs/replication_history.go new file mode 100644 index 0000000..c650451 --- /dev/null +++ b/zfs/replication_history.go @@ -0,0 +1,25 @@ +package zfs + +const ReplicatedProperty = "zrepl:replicated" + +// May return *DatasetDoesNotExist as an error +func ZFSGetReplicatedProperty(fs *DatasetPath, v *FilesystemVersion) (replicated bool, err error) { + props, err := zfsGet(v.ToAbsPath(fs), []string{ReplicatedProperty}) + if err != nil { + return false, err + } + if props.Get(ReplicatedProperty) == "yes" { + return true, nil + } + return false, nil +} + +func ZFSSetReplicatedProperty(fs *DatasetPath, v *FilesystemVersion, replicated bool) error { + val := "no" + if replicated { + val = "yes" + } + props := NewZFSProperties() + props.Set(ReplicatedProperty, val) + return zfsSet(v.ToAbsPath(fs), props) +} diff --git a/zfs/versions.go b/zfs/versions.go index 4bf2d40..9deb10d 100644 --- a/zfs/versions.go +++ b/zfs/versions.go @@ -154,7 +154,7 @@ func ZFSListFilesystemVersions(fs *DatasetPath, filter FilesystemVersionFilter) return } -func ZFSDestroyFilesystemVersion(filesystem *DatasetPath, version FilesystemVersion) (err error) { +func ZFSDestroyFilesystemVersion(filesystem *DatasetPath, version *FilesystemVersion) (err error) { promTimer := prometheus.NewTimer(prom.ZFSDestroyFilesystemVersionDuration.WithLabelValues(filesystem.ToString(), version.Type.String())) defer promTimer.ObserveDuration() diff --git a/zfs/zfs.go b/zfs/zfs.go index 2b11da3..58ed6d3 100644 --- a/zfs/zfs.go +++ b/zfs/zfs.go @@ -14,6 +14,7 @@ import ( "github.com/problame/go-rwccmd" "github.com/prometheus/client_golang/prometheus" "github.com/zrepl/zrepl/util" + "regexp" ) type DatasetPath struct { @@ -397,14 +398,17 @@ func (p *ZFSProperties) appendArgs(args *[]string) (err error) { } func ZFSSet(fs *DatasetPath, props *ZFSProperties) (err error) { + return zfsSet(fs.ToString(), props) +} +func zfsSet(path string, props *ZFSProperties) (err error) { args := make([]string, 0) args = append(args, "set") err = props.appendArgs(&args) if err != nil { return err } - args = append(args, fs.ToString()) + args = append(args, path) cmd := exec.Command(ZFS_BINARY, args...) @@ -426,11 +430,32 @@ func ZFSSet(fs *DatasetPath, props *ZFSProperties) (err error) { } func ZFSGet(fs *DatasetPath, props []string) (*ZFSProperties, error) { - args := []string{"get", "-Hp", "-o", "property,value", strings.Join(props, ","), fs.ToString()} + return zfsGet(fs.ToString(), props) +} +var zfsGetDatasetDoesNotExistRegexp = regexp.MustCompile(`^cannot open '(\s+)': dataset does not exist`) + +type DatasetDoesNotExist struct { + Path string +} + +func (d *DatasetDoesNotExist) Error() string { return fmt.Sprintf("dataset %q does not exist", d.Path) } + +func zfsGet(path string, props []string) (*ZFSProperties, error) { + args := []string{"get", "-Hp", "-o", "property,value", strings.Join(props, ","), path} cmd := exec.Command(ZFS_BINARY, args...) output, err := cmd.CombinedOutput() if err != nil { + if exitErr, ok := err.(*exec.ExitError); ok { + if exitErr.Exited() { + // screen-scrape output + if sm := zfsGetDatasetDoesNotExistRegexp.FindSubmatch(output); sm != nil { + if string(sm[1]) == path { + return nil, &DatasetDoesNotExist{path} + } + } + } + } return nil, err } o := string(output)