diff --git a/README.md b/README.md index 7addaca2..10ca68ad 100644 --- a/README.md +++ b/README.md @@ -76,6 +76,7 @@ go install \ go install github.com/grpc-ecosystem/protoc-gen-grpc-gateway-ts@latest go install google.golang.org/protobuf/cmd/protoc-gen-go@latest go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest +go install github.com/favadi/protoc-go-inject-tag@latest go install github.com/bufbuild/buf/cmd/buf@v1.27.2 ``` ## Building diff --git a/gen/go/v1/operations.pb.go b/gen/go/v1/operations.pb.go index 8552ad93..dc49dfca 100644 --- a/gen/go/v1/operations.pb.go +++ b/gen/go/v1/operations.pb.go @@ -177,13 +177,22 @@ type Operation struct { sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - Id int64 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"` - RepoId string `protobuf:"bytes,2,opt,name=repo_id,json=repoId,proto3" json:"repo_id,omitempty"` // repo id if associated with a repo (always true) - PlanId string `protobuf:"bytes,3,opt,name=plan_id,json=planId,proto3" json:"plan_id,omitempty"` // plan id if associated with a plan (always true) + // @gotags: `storm:"id,increment"` + Id int64 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty" storm:"id,increment"` + // repo id if associated with a repo (always true) + // @gotags: `storm:"index"` + RepoId string `protobuf:"bytes,2,opt,name=repo_id,json=repoId,proto3" json:"repo_id,omitempty" storm:"index"` + // plan id if associated with a plan (always true) + // @gotags: `storm:"index"` + PlanId string `protobuf:"bytes,3,opt,name=plan_id,json=planId,proto3" json:"plan_id,omitempty" storm:"index"` + // normalized snapshot id (first 8 chars) if associated with a snapshot + // @gotags: `storm:"index"` + SnapshotId string `protobuf:"bytes,8,opt,name=snapshot_id,json=snapshotId,proto3" json:"snapshot_id,omitempty" storm:"index"` Status OperationStatus `protobuf:"varint,4,opt,name=status,proto3,enum=v1.OperationStatus" json:"status,omitempty"` UnixTimeStartMs int64 `protobuf:"varint,5,opt,name=unix_time_start_ms,json=unixTimeStartMs,proto3" json:"unix_time_start_ms,omitempty"` - UnixTimeEndMs int64 `protobuf:"varint,6,opt,name=unix_time_end_ms,json=unixTimeEndMs,proto3" json:"unix_time_end_ms,omitempty"` - DisplayMessage string `protobuf:"bytes,7,opt,name=display_message,json=displayMessage,proto3" json:"display_message,omitempty"` // human readable context message (if any) + // operation ended time in unix ms + UnixTimeEndMs int64 `protobuf:"varint,6,opt,name=unix_time_end_ms,json=unixTimeEndMs,proto3" json:"unix_time_end_ms,omitempty"` + DisplayMessage string `protobuf:"bytes,7,opt,name=display_message,json=displayMessage,proto3" json:"display_message,omitempty"` // human readable context message (if any) // Types that are assignable to Op: // // *Operation_OperationBackup @@ -244,6 +253,13 @@ func (x *Operation) GetPlanId() string { return "" } +func (x *Operation) GetSnapshotId() string { + if x != nil { + return x.SnapshotId + } + return "" +} + func (x *Operation) GetStatus() OperationStatus { if x != nil { return x.Status @@ -469,65 +485,67 @@ var file_v1_operations_proto_rawDesc = []byte{ 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x4c, 0x69, 0x73, 0x74, 0x12, 0x2d, 0x0a, 0x0a, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x76, 0x31, 0x2e, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x0a, - 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x22, 0x99, 0x03, 0x0a, 0x09, 0x4f, + 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x22, 0xba, 0x03, 0x0a, 0x09, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x03, 0x52, 0x02, 0x69, 0x64, 0x12, 0x17, 0x0a, 0x07, 0x72, 0x65, 0x70, 0x6f, 0x5f, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x72, 0x65, 0x70, 0x6f, 0x49, 0x64, 0x12, 0x17, 0x0a, 0x07, 0x70, 0x6c, 0x61, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, - 0x28, 0x09, 0x52, 0x06, 0x70, 0x6c, 0x61, 0x6e, 0x49, 0x64, 0x12, 0x2b, 0x0a, 0x06, 0x73, 0x74, - 0x61, 0x74, 0x75, 0x73, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x13, 0x2e, 0x76, 0x31, 0x2e, - 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, - 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x2b, 0x0a, 0x12, 0x75, 0x6e, 0x69, 0x78, 0x5f, - 0x74, 0x69, 0x6d, 0x65, 0x5f, 0x73, 0x74, 0x61, 0x72, 0x74, 0x5f, 0x6d, 0x73, 0x18, 0x05, 0x20, - 0x01, 0x28, 0x03, 0x52, 0x0f, 0x75, 0x6e, 0x69, 0x78, 0x54, 0x69, 0x6d, 0x65, 0x53, 0x74, 0x61, - 0x72, 0x74, 0x4d, 0x73, 0x12, 0x27, 0x0a, 0x10, 0x75, 0x6e, 0x69, 0x78, 0x5f, 0x74, 0x69, 0x6d, - 0x65, 0x5f, 0x65, 0x6e, 0x64, 0x5f, 0x6d, 0x73, 0x18, 0x06, 0x20, 0x01, 0x28, 0x03, 0x52, 0x0d, - 0x75, 0x6e, 0x69, 0x78, 0x54, 0x69, 0x6d, 0x65, 0x45, 0x6e, 0x64, 0x4d, 0x73, 0x12, 0x27, 0x0a, - 0x0f, 0x64, 0x69, 0x73, 0x70, 0x6c, 0x61, 0x79, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, - 0x18, 0x07, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0e, 0x64, 0x69, 0x73, 0x70, 0x6c, 0x61, 0x79, 0x4d, - 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x40, 0x0a, 0x10, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, - 0x69, 0x6f, 0x6e, 0x5f, 0x62, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x18, 0x64, 0x20, 0x01, 0x28, 0x0b, - 0x32, 0x13, 0x2e, 0x76, 0x31, 0x2e, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x42, - 0x61, 0x63, 0x6b, 0x75, 0x70, 0x48, 0x00, 0x52, 0x0f, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, - 0x6f, 0x6e, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x12, 0x56, 0x0a, 0x18, 0x6f, 0x70, 0x65, 0x72, - 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x6e, 0x64, 0x65, 0x78, 0x5f, 0x73, 0x6e, 0x61, 0x70, - 0x73, 0x68, 0x6f, 0x74, 0x18, 0x65, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x76, 0x31, 0x2e, - 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x6e, 0x64, 0x65, 0x78, 0x53, 0x6e, - 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x48, 0x00, 0x52, 0x16, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, - 0x69, 0x6f, 0x6e, 0x49, 0x6e, 0x64, 0x65, 0x78, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, - 0x42, 0x04, 0x0a, 0x02, 0x6f, 0x70, 0x22, 0x69, 0x0a, 0x0e, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, - 0x69, 0x6f, 0x6e, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x12, 0x2a, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, - 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x16, 0x2e, 0x76, 0x31, 0x2e, 0x4f, 0x70, 0x65, 0x72, - 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x54, 0x79, 0x70, 0x65, 0x52, 0x04, - 0x74, 0x79, 0x70, 0x65, 0x12, 0x2b, 0x0a, 0x09, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, - 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x76, 0x31, 0x2e, 0x4f, 0x70, 0x65, - 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, - 0x6e, 0x22, 0x4b, 0x0a, 0x0f, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x42, 0x61, - 0x63, 0x6b, 0x75, 0x70, 0x12, 0x38, 0x0a, 0x0b, 0x6c, 0x61, 0x73, 0x74, 0x5f, 0x73, 0x74, 0x61, - 0x74, 0x75, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x17, 0x2e, 0x76, 0x31, 0x2e, 0x42, - 0x61, 0x63, 0x6b, 0x75, 0x70, 0x50, 0x72, 0x6f, 0x67, 0x72, 0x65, 0x73, 0x73, 0x45, 0x6e, 0x74, - 0x72, 0x79, 0x52, 0x0a, 0x6c, 0x61, 0x73, 0x74, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x22, 0x48, - 0x0a, 0x16, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x6e, 0x64, 0x65, 0x78, - 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x12, 0x2e, 0x0a, 0x08, 0x73, 0x6e, 0x61, 0x70, - 0x73, 0x68, 0x6f, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x76, 0x31, 0x2e, - 0x52, 0x65, 0x73, 0x74, 0x69, 0x63, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x08, - 0x73, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x2a, 0x4d, 0x0a, 0x12, 0x4f, 0x70, 0x65, 0x72, - 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x54, 0x79, 0x70, 0x65, 0x12, 0x11, - 0x0a, 0x0d, 0x45, 0x56, 0x45, 0x4e, 0x54, 0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, - 0x00, 0x12, 0x11, 0x0a, 0x0d, 0x45, 0x56, 0x45, 0x4e, 0x54, 0x5f, 0x43, 0x52, 0x45, 0x41, 0x54, - 0x45, 0x44, 0x10, 0x01, 0x12, 0x11, 0x0a, 0x0d, 0x45, 0x56, 0x45, 0x4e, 0x54, 0x5f, 0x55, 0x50, - 0x44, 0x41, 0x54, 0x45, 0x44, 0x10, 0x02, 0x2a, 0x76, 0x0a, 0x0f, 0x4f, 0x70, 0x65, 0x72, 0x61, - 0x74, 0x69, 0x6f, 0x6e, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x54, - 0x41, 0x54, 0x55, 0x53, 0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x12, - 0x0a, 0x0e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x50, 0x45, 0x4e, 0x44, 0x49, 0x4e, 0x47, - 0x10, 0x01, 0x12, 0x15, 0x0a, 0x11, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x49, 0x4e, 0x50, - 0x52, 0x4f, 0x47, 0x52, 0x45, 0x53, 0x53, 0x10, 0x02, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x54, 0x41, - 0x54, 0x55, 0x53, 0x5f, 0x53, 0x55, 0x43, 0x43, 0x45, 0x53, 0x53, 0x10, 0x03, 0x12, 0x10, 0x0a, - 0x0c, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x45, 0x52, 0x52, 0x4f, 0x52, 0x10, 0x04, 0x42, - 0x2e, 0x5a, 0x2c, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x67, 0x61, - 0x72, 0x65, 0x74, 0x68, 0x67, 0x65, 0x6f, 0x72, 0x67, 0x65, 0x2f, 0x72, 0x65, 0x73, 0x74, 0x69, - 0x63, 0x75, 0x69, 0x2f, 0x67, 0x6f, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x76, 0x31, 0x62, - 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x28, 0x09, 0x52, 0x06, 0x70, 0x6c, 0x61, 0x6e, 0x49, 0x64, 0x12, 0x1f, 0x0a, 0x0b, 0x73, 0x6e, + 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x08, 0x20, 0x01, 0x28, 0x09, 0x52, + 0x0a, 0x73, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x49, 0x64, 0x12, 0x2b, 0x0a, 0x06, 0x73, + 0x74, 0x61, 0x74, 0x75, 0x73, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x13, 0x2e, 0x76, 0x31, + 0x2e, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, + 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x2b, 0x0a, 0x12, 0x75, 0x6e, 0x69, 0x78, + 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x5f, 0x73, 0x74, 0x61, 0x72, 0x74, 0x5f, 0x6d, 0x73, 0x18, 0x05, + 0x20, 0x01, 0x28, 0x03, 0x52, 0x0f, 0x75, 0x6e, 0x69, 0x78, 0x54, 0x69, 0x6d, 0x65, 0x53, 0x74, + 0x61, 0x72, 0x74, 0x4d, 0x73, 0x12, 0x27, 0x0a, 0x10, 0x75, 0x6e, 0x69, 0x78, 0x5f, 0x74, 0x69, + 0x6d, 0x65, 0x5f, 0x65, 0x6e, 0x64, 0x5f, 0x6d, 0x73, 0x18, 0x06, 0x20, 0x01, 0x28, 0x03, 0x52, + 0x0d, 0x75, 0x6e, 0x69, 0x78, 0x54, 0x69, 0x6d, 0x65, 0x45, 0x6e, 0x64, 0x4d, 0x73, 0x12, 0x27, + 0x0a, 0x0f, 0x64, 0x69, 0x73, 0x70, 0x6c, 0x61, 0x79, 0x5f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, + 0x65, 0x18, 0x07, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0e, 0x64, 0x69, 0x73, 0x70, 0x6c, 0x61, 0x79, + 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x40, 0x0a, 0x10, 0x6f, 0x70, 0x65, 0x72, 0x61, + 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x62, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x18, 0x64, 0x20, 0x01, 0x28, + 0x0b, 0x32, 0x13, 0x2e, 0x76, 0x31, 0x2e, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, + 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x48, 0x00, 0x52, 0x0f, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, + 0x69, 0x6f, 0x6e, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x12, 0x56, 0x0a, 0x18, 0x6f, 0x70, 0x65, + 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x6e, 0x64, 0x65, 0x78, 0x5f, 0x73, 0x6e, 0x61, + 0x70, 0x73, 0x68, 0x6f, 0x74, 0x18, 0x65, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x76, 0x31, + 0x2e, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x6e, 0x64, 0x65, 0x78, 0x53, + 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x48, 0x00, 0x52, 0x16, 0x6f, 0x70, 0x65, 0x72, 0x61, + 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x6e, 0x64, 0x65, 0x78, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, + 0x74, 0x42, 0x04, 0x0a, 0x02, 0x6f, 0x70, 0x22, 0x69, 0x0a, 0x0e, 0x4f, 0x70, 0x65, 0x72, 0x61, + 0x74, 0x69, 0x6f, 0x6e, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x12, 0x2a, 0x0a, 0x04, 0x74, 0x79, 0x70, + 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x16, 0x2e, 0x76, 0x31, 0x2e, 0x4f, 0x70, 0x65, + 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x54, 0x79, 0x70, 0x65, 0x52, + 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x2b, 0x0a, 0x09, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, + 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x76, 0x31, 0x2e, 0x4f, 0x70, + 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, + 0x6f, 0x6e, 0x22, 0x4b, 0x0a, 0x0f, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x42, + 0x61, 0x63, 0x6b, 0x75, 0x70, 0x12, 0x38, 0x0a, 0x0b, 0x6c, 0x61, 0x73, 0x74, 0x5f, 0x73, 0x74, + 0x61, 0x74, 0x75, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x17, 0x2e, 0x76, 0x31, 0x2e, + 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x50, 0x72, 0x6f, 0x67, 0x72, 0x65, 0x73, 0x73, 0x45, 0x6e, + 0x74, 0x72, 0x79, 0x52, 0x0a, 0x6c, 0x61, 0x73, 0x74, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x22, + 0x48, 0x0a, 0x16, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x6e, 0x64, 0x65, + 0x78, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x12, 0x2e, 0x0a, 0x08, 0x73, 0x6e, 0x61, + 0x70, 0x73, 0x68, 0x6f, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x76, 0x31, + 0x2e, 0x52, 0x65, 0x73, 0x74, 0x69, 0x63, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, + 0x08, 0x73, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x2a, 0x4d, 0x0a, 0x12, 0x4f, 0x70, 0x65, + 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x54, 0x79, 0x70, 0x65, 0x12, + 0x11, 0x0a, 0x0d, 0x45, 0x56, 0x45, 0x4e, 0x54, 0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, + 0x10, 0x00, 0x12, 0x11, 0x0a, 0x0d, 0x45, 0x56, 0x45, 0x4e, 0x54, 0x5f, 0x43, 0x52, 0x45, 0x41, + 0x54, 0x45, 0x44, 0x10, 0x01, 0x12, 0x11, 0x0a, 0x0d, 0x45, 0x56, 0x45, 0x4e, 0x54, 0x5f, 0x55, + 0x50, 0x44, 0x41, 0x54, 0x45, 0x44, 0x10, 0x02, 0x2a, 0x76, 0x0a, 0x0f, 0x4f, 0x70, 0x65, 0x72, + 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x12, 0x0a, 0x0e, 0x53, + 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, + 0x12, 0x0a, 0x0e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x50, 0x45, 0x4e, 0x44, 0x49, 0x4e, + 0x47, 0x10, 0x01, 0x12, 0x15, 0x0a, 0x11, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x49, 0x4e, + 0x50, 0x52, 0x4f, 0x47, 0x52, 0x45, 0x53, 0x53, 0x10, 0x02, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x54, + 0x41, 0x54, 0x55, 0x53, 0x5f, 0x53, 0x55, 0x43, 0x43, 0x45, 0x53, 0x53, 0x10, 0x03, 0x12, 0x10, + 0x0a, 0x0c, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x45, 0x52, 0x52, 0x4f, 0x52, 0x10, 0x04, + 0x42, 0x2e, 0x5a, 0x2c, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x67, + 0x61, 0x72, 0x65, 0x74, 0x68, 0x67, 0x65, 0x6f, 0x72, 0x67, 0x65, 0x2f, 0x72, 0x65, 0x73, 0x74, + 0x69, 0x63, 0x75, 0x69, 0x2f, 0x67, 0x6f, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x76, 0x31, + 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( diff --git a/go.mod b/go.mod index 918163d6..9827816c 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module github.com/garethgeorge/resticui go 1.21.3 require ( + github.com/asdine/storm/v3 v3.2.1 github.com/gitploy-io/cronexpr v0.2.2 github.com/google/renameio v1.0.1 github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.1 @@ -11,7 +12,7 @@ require ( github.com/mattn/go-colorable v0.1.13 go.etcd.io/bbolt v1.3.8 go.uber.org/zap v1.26.0 - google.golang.org/genproto/googleapis/api v0.0.0-20231106174013-bbf56f31fb17 + google.golang.org/genproto/googleapis/api v0.0.0-20231120223509-83a465c0220f google.golang.org/grpc v1.59.0 google.golang.org/protobuf v1.31.0 ) @@ -24,6 +25,6 @@ require ( golang.org/x/net v0.18.0 // indirect golang.org/x/sys v0.14.0 // indirect golang.org/x/text v0.14.0 // indirect - google.golang.org/genproto v0.0.0-20231106174013-bbf56f31fb17 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20231106174013-bbf56f31fb17 // indirect + google.golang.org/genproto v0.0.0-20231120223509-83a465c0220f // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20231120223509-83a465c0220f // indirect ) diff --git a/go.sum b/go.sum index 43ab565c..ef3a4e4e 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +github.com/asdine/storm/v3 v3.2.1 h1:I5AqhkPK6nBZ/qJXySdI7ot5BlXSZ7qvDY1zAn5ZJac= +github.com/asdine/storm/v3 v3.2.1/go.mod h1:LEpXwGt4pIqrE/XcTvCnZHT5MgZCV6Ub9q7yQzOFWr0= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/gitploy-io/cronexpr v0.2.2 h1:Au+wK6FqmOLAF7AkW6q4gnrNXTe3rEW97XFZ4chy0xs= @@ -47,10 +49,16 @@ golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= google.golang.org/genproto v0.0.0-20231106174013-bbf56f31fb17 h1:wpZ8pe2x1Q3f2KyT5f8oP/fa9rHAKgFPr/HZdNuS+PQ= google.golang.org/genproto v0.0.0-20231106174013-bbf56f31fb17/go.mod h1:J7XzRzVy1+IPwWHZUzoD0IccYZIrXILAQpc+Qy9CMhY= +google.golang.org/genproto v0.0.0-20231120223509-83a465c0220f h1:Vn+VyHU5guc9KjB5KrjI2q0wCOWEOIh0OEsleqakHJg= +google.golang.org/genproto v0.0.0-20231120223509-83a465c0220f/go.mod h1:nWSwAFPb+qfNJXsoeO3Io7zf4tMSfN8EA8RlDA04GhY= google.golang.org/genproto/googleapis/api v0.0.0-20231106174013-bbf56f31fb17 h1:JpwMPBpFN3uKhdaekDpiNlImDdkUAyiJ6ez/uxGaUSo= google.golang.org/genproto/googleapis/api v0.0.0-20231106174013-bbf56f31fb17/go.mod h1:0xJLfVdJqpAPl8tDg1ujOCGzx6LFLttXT5NhllGOXY4= +google.golang.org/genproto/googleapis/api v0.0.0-20231120223509-83a465c0220f h1:2yNACc1O40tTnrsbk9Cv6oxiW8pxI/pXj0wRtdlYmgY= +google.golang.org/genproto/googleapis/api v0.0.0-20231120223509-83a465c0220f/go.mod h1:Uy9bTZJqmfrw2rIBxgGLnamc78euZULUBrLZ9XTITKI= google.golang.org/genproto/googleapis/rpc v0.0.0-20231106174013-bbf56f31fb17 h1:Jyp0Hsi0bmHXG6k9eATXoYtjd6e2UzZ1SCn/wIupY14= google.golang.org/genproto/googleapis/rpc v0.0.0-20231106174013-bbf56f31fb17/go.mod h1:oQ5rr10WTTMvP4A36n8JpR1OrO1BEiV4f78CneXZxkA= +google.golang.org/genproto/googleapis/rpc v0.0.0-20231120223509-83a465c0220f h1:ultW7fxlIvee4HYrtnaRPon9HpEgFk5zYpmfMgtKB5I= +google.golang.org/genproto/googleapis/rpc v0.0.0-20231120223509-83a465c0220f/go.mod h1:L9KNLi232K1/xB6f7AlSX692koaRnKaWSR0stBki0Yc= google.golang.org/grpc v1.59.0 h1:Z5Iec2pjwb+LEOqzpB2MR12/eKFhDPhuqW91O+4bwUk= google.golang.org/grpc v1.59.0/go.mod h1:aUPDwccQo6OTjy7Hct4AfBPD1GptF4fyUjIkQ9YtF98= google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= diff --git a/internal/api/server.go b/internal/api/server.go index 9027daa7..06930399 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -10,7 +10,7 @@ import ( "github.com/garethgeorge/resticui/gen/go/types" v1 "github.com/garethgeorge/resticui/gen/go/v1" "github.com/garethgeorge/resticui/internal/config" - "github.com/garethgeorge/resticui/internal/database/oplog" + "github.com/garethgeorge/resticui/internal/oplog" "github.com/garethgeorge/resticui/internal/orchestrator" "github.com/garethgeorge/resticui/pkg/restic" "go.uber.org/zap" diff --git a/internal/database/indexutil/indexutil.go b/internal/database/indexutil/indexutil.go deleted file mode 100644 index 4cfdd0da..00000000 --- a/internal/database/indexutil/indexutil.go +++ /dev/null @@ -1,57 +0,0 @@ -package indexutil - -import ( - "bytes" - - "github.com/garethgeorge/resticui/internal/database/serializationutil" - bolt "go.etcd.io/bbolt" -) - -// IndexByteValue indexes a value and recordId tuple creating multimap from value to lists of associated recordIds. -func IndexByteValue(b *bolt.Bucket, value []byte, recordId int64) error { - key := serializationutil.BytesToKey(value) - key = append(key, serializationutil.Itob(recordId)...) - return b.Put(key, []byte{}) -} - -// IndexSearchByteValue searches the index given a value and returns an iterator over the associated recordIds. -func IndexSearchByteValue(b *bolt.Bucket, value []byte) *IndexSearchIterator { - return newSearchIterator(b, serializationutil.BytesToKey(value)) -} - -type IndexSearchIterator struct { - c *bolt.Cursor - k []byte - prefix []byte -} - -func newSearchIterator(b *bolt.Bucket, prefix []byte) *IndexSearchIterator { - c := b.Cursor() - k, _ := c.Seek(prefix) - return &IndexSearchIterator{ - c: c, - k: k, - prefix: prefix, - } -} - -func (i *IndexSearchIterator) Next() (int64, bool) { - if i.k == nil || !bytes.HasPrefix(i.k, i.prefix) { - return 0, false - } - id, err := serializationutil.Btoi(i.k[len(i.prefix):]) - if err != nil { - // this sholud never happen, if it does it indicates database corruption. - return 0, false - } - i.k, _ = i.c.Next() - return id, true -} - -func (i *IndexSearchIterator) ToSlice() []int64 { - var ids []int64 - for id, ok := i.Next(); ok; id, ok = i.Next() { - ids = append(ids, id) - } - return ids -} \ No newline at end of file diff --git a/internal/database/indexutil/indexutil_test.go b/internal/database/indexutil/indexutil_test.go deleted file mode 100644 index d04eb072..00000000 --- a/internal/database/indexutil/indexutil_test.go +++ /dev/null @@ -1,45 +0,0 @@ -package indexutil - -import ( - "fmt" - "testing" - - "go.etcd.io/bbolt" -) - -func TestIndexing(t *testing.T) { - db, err := bbolt.Open(t.TempDir() + "/test.boltdb", 0600, nil) - if err != nil { - t.Fatalf("error opening database: %s", err) - } - - if err := db.Update(func(tx *bbolt.Tx) error { - b, err := tx.CreateBucket([]byte("test")) - if err != nil { - return fmt.Errorf("error creating bucket: %s", err) - } - for id := 0; id < 100; id += 1 { - if err := IndexByteValue(b, []byte("document"), int64(id)); err != nil { - return err - } - } - return nil - }); err != nil { - t.Fatalf("db.Update error: %v", err) - } - - if err := db.View(func(tx *bbolt.Tx) error { - b := tx.Bucket([]byte("test")) - ids := IndexSearchByteValue(b, []byte("document")).ToSlice() - if len(ids) != 100 { - t.Errorf("want 100 ids, got %d", len(ids)) - } - ids = IndexSearchByteValue(b, []byte("other")).ToSlice() - if len(ids) != 0 { - t.Errorf("want 0 ids, got %d", len(ids)) - } - return nil - }); err != nil { - t.Fatalf("db.View error: %v", err) - } -} diff --git a/internal/database/oplog/oplog.go b/internal/database/oplog/oplog.go deleted file mode 100644 index 1459750c..00000000 --- a/internal/database/oplog/oplog.go +++ /dev/null @@ -1,386 +0,0 @@ -package oplog - -import ( - "errors" - "fmt" - "os" - "path" - "sync" - "time" - - v1 "github.com/garethgeorge/resticui/gen/go/v1" - "github.com/garethgeorge/resticui/internal/database/indexutil" - "github.com/garethgeorge/resticui/internal/database/serializationutil" - bolt "go.etcd.io/bbolt" - "go.uber.org/zap" - "google.golang.org/protobuf/proto" -) - -type EventType int - -const ( - EventTypeUnknown = EventType(iota) - EventTypeOpCreated = EventType(iota) - EventTypeOpUpdated = EventType(iota) -) - -var ( - SystemBucket = []byte("oplog.system") // system stores metadata - OpLogBucket = []byte("oplog.log") // oplog stores the operations themselves - RepoIndexBucket = []byte("oplog.repo_idx") // repo_index tracks IDs of operations affecting a given repo - PlanIndexBucket = []byte("oplog.plan_idx") // plan_index tracks IDs of operations affecting a given plan - IndexedSnapshotsSetBucket = []byte("oplog.indexed_snapshots") // indexed_snapshots is a set of snapshot IDs that have been indexed -) - -// OpLog represents a log of operations performed. -// Operations are indexed by repo and plan. -type OpLog struct { - db *bolt.DB - - subscribersMu sync.RWMutex - subscribers []*func(EventType, *v1.Operation) -} - -func NewOpLog(databasePath string) (*OpLog, error) { - if err := os.MkdirAll(path.Dir(databasePath), 0700); err != nil { - return nil, fmt.Errorf("error creating database directory: %s", err) - } - - db, err := bolt.Open(databasePath, 0600, &bolt.Options{Timeout: 1 * time.Second}) - if err != nil { - return nil, fmt.Errorf("error opening database: %s", err) - } - - if err := db.Update(func(tx *bolt.Tx) error { - // Create the buckets if they don't exist - for _, bucket := range [][]byte{ - SystemBucket, OpLogBucket, RepoIndexBucket, PlanIndexBucket, IndexedSnapshotsSetBucket, - } { - if _, err := tx.CreateBucketIfNotExists(bucket); err != nil { - return fmt.Errorf("creating bucket %s: %s", string(bucket), err) - } - } - - // Validate the operation log on startup. - sysBucket := tx.Bucket(SystemBucket) - opLogBucket := tx.Bucket(OpLogBucket) - c := opLogBucket.Cursor() - if lastValidated := sysBucket.Get([]byte("last_validated")); lastValidated != nil { - c.Seek(lastValidated) - } - for k, v := c.First(); k != nil; k, v = c.Next() { - op := &v1.Operation{} - if err := proto.Unmarshal(v, op); err != nil { - zap.L().Error("error unmarshalling operation, there may be corruption in the oplog", zap.Error(err)) - continue - } - if op.Status == v1.OperationStatus_STATUS_INPROGRESS { - op.Status = v1.OperationStatus_STATUS_ERROR - op.DisplayMessage = "Operation timeout." - bytes, err := proto.Marshal(op) - if err != nil { - return fmt.Errorf("marshalling operation: %w", err) - } - if err := opLogBucket.Put(k, bytes); err != nil { - return fmt.Errorf("putting operation into bucket: %w", err) - } - } - } - if lastValidated, _ := c.Last(); lastValidated != nil { - if err := sysBucket.Put([]byte("last_validated"), lastValidated); err != nil { - return fmt.Errorf("checkpointing last_validated key: %w", err) - } - } - - return nil - }); err != nil { - return nil, err - } - - return &OpLog{db: db}, nil -} - -func (o *OpLog) Close() error { - return o.db.Close() -} - -// Add adds a generic operation to the operation log. -func (o *OpLog) Add(op *v1.Operation) error { - if op.Id != 0 { - return errors.New("operation already has an ID, OpLog.Add is expected to set the ID") - } - - err := o.db.Update(func(tx *bolt.Tx) error { - err := o.addOperationHelper(tx, op) - if err != nil { - return err - } - return nil - }) - if err == nil { - o.notifyHelper(EventTypeOpCreated, op) - } - return err -} - -func (o *OpLog) BulkAdd(ops []*v1.Operation) error { - err := o.db.Update(func(tx *bolt.Tx) error { - for _, op := range ops { - if err := o.addOperationHelper(tx, op); err != nil { - return err - } - } - return nil - }) - if err == nil { - for _, op := range ops { - o.notifyHelper(EventTypeOpCreated, op) - } - } - return err -} - -func (o *OpLog) addOperationHelper(tx *bolt.Tx, op *v1.Operation) error { - b := tx.Bucket(OpLogBucket) - - id, err := b.NextSequence() - if err != nil { - return fmt.Errorf("error getting next sequence: %w", err) - } - - op.Id = int64(id) - - bytes, err := proto.Marshal(op) - if err != nil { - return fmt.Errorf("error marshalling operation: %w", err) - } - - if err := b.Put(serializationutil.Itob(op.Id), bytes); err != nil { - return fmt.Errorf("error putting operation into bucket: %w", err) - } - - // Update always universal indices - if op.RepoId != "" { - if err := indexutil.IndexByteValue(tx.Bucket(RepoIndexBucket), []byte(op.RepoId), op.Id); err != nil { - return fmt.Errorf("error adding operation to repo index: %w", err) - } - } - if op.PlanId != "" { - if err := indexutil.IndexByteValue(tx.Bucket(PlanIndexBucket), []byte(op.PlanId), op.Id); err != nil { - return fmt.Errorf("error adding operation to repo index: %w", err) - } - } - - // Update operation type dependent indices. - switch wrappedOp := op.Op.(type) { - case *v1.Operation_OperationBackup: - // Nothing extra to be done. - case *v1.Operation_OperationIndexSnapshot: - if wrappedOp.OperationIndexSnapshot == nil || wrappedOp.OperationIndexSnapshot.Snapshot == nil { - return errors.New("op.OperationIndexSnapshot or op.OperationIndexSnapshot.Snapshot is nil") - } - snapshotId := serializationutil.NormalizeSnapshotId(wrappedOp.OperationIndexSnapshot.Snapshot.Id) - key := serializationutil.BytesToKey([]byte(snapshotId)) - if err := tx.Bucket(IndexedSnapshotsSetBucket).Put(key, serializationutil.Itob(op.Id)); err != nil { - return fmt.Errorf("error adding OperationIndexSnapshot to indexed snapshots set: %w", err) - } - default: - return fmt.Errorf("unknown operation type: %T", wrappedOp) - } - - return nil -} - -func (o *OpLog) HasIndexedSnapshot(snapshotId string) (int64, error) { - var id int64 - if err := o.db.View(func(tx *bolt.Tx) error { - snapshotId := serializationutil.NormalizeSnapshotId(snapshotId) - key := serializationutil.BytesToKey([]byte(snapshotId)) - idBytes := tx.Bucket(IndexedSnapshotsSetBucket).Get(key) - if idBytes == nil { - id = -1 - } else { - var err error - id, err = serializationutil.Btoi(idBytes) - if err != nil { - return fmt.Errorf("database corrupt, couldn't convert ID bytes to int: %w", err) - } - } - return nil - }); err != nil { - return 0, err - } - return id, nil -} - -func (o *OpLog) Update(op *v1.Operation) error { - if op.Id == 0 { - return errors.New("operation does not have an ID, OpLog.Update expects operation with an ID") - } - - err := o.db.Update(func(tx *bolt.Tx) error { - b := tx.Bucket(OpLogBucket) - - if b.Get(serializationutil.Itob(op.Id)) == nil { - return fmt.Errorf("operation with ID %d does not exist", op.Id) - } - - bytes, err := proto.Marshal(op) - if err != nil { - return fmt.Errorf("error marshalling operation: %w", err) - } - - if err := b.Put(serializationutil.Itob(op.Id), bytes); err != nil { - return fmt.Errorf("error putting operation into bucket: %w", err) - } - - return nil - }) - if err == nil { - o.notifyHelper(EventTypeOpUpdated, op) - } - return err -} - -func (o *OpLog) notifyHelper(eventType EventType, op *v1.Operation) { - o.subscribersMu.RLock() - defer o.subscribersMu.RUnlock() - for _, sub := range o.subscribers { - (*sub)(eventType, op) - } -} - -func (o *OpLog) getHelper(b *bolt.Bucket, id int64) (*v1.Operation, error) { - bytes := b.Get(serializationutil.Itob(id)) - if bytes == nil { - return nil, fmt.Errorf("operation with ID %d does not exist", id) - } - - var op v1.Operation - if err := proto.Unmarshal(bytes, &op); err != nil { - return nil, fmt.Errorf("error unmarshalling operation: %w", err) - } - - return &op, nil -} - -func (o *OpLog) Get(id int64) (*v1.Operation, error) { - var op *v1.Operation - if err := o.db.View(func(tx *bolt.Tx) error { - var err error - op, err = o.getHelper(tx.Bucket(OpLogBucket), id) - return err - }); err != nil { - return nil, err - } - return op, nil -} - -func (o *OpLog) GetByRepo(repoId string, filter Filter) ([]*v1.Operation, error) { - var ops []*v1.Operation - if err := o.db.View(func(tx *bolt.Tx) error { - ids := indexutil.IndexSearchByteValue(tx.Bucket(RepoIndexBucket), []byte(repoId)).ToSlice() - ids = filter(ids) - - b := tx.Bucket(OpLogBucket) - for _, id := range ids { - op, err := o.getHelper(b, id) - if err != nil { - return err - } - ops = append(ops, op) - } - - return nil - }); err != nil { - return nil, err - } - return ops, nil -} - -func (o *OpLog) GetByPlan(planId string, filter Filter) ([]*v1.Operation, error) { - var ops []*v1.Operation - if err := o.db.View(func(tx *bolt.Tx) error { - ids := indexutil.IndexSearchByteValue(tx.Bucket(PlanIndexBucket), []byte(planId)).ToSlice() - ids = filter(ids) - - b := tx.Bucket(OpLogBucket) - for _, id := range ids { - op, err := o.getHelper(b, id) - if err != nil { - return err - } - ops = append(ops, op) - } - - return nil - }); err != nil { - return nil, err - } - return ops, nil -} - -func (o *OpLog) GetAll(filter Filter) ([]*v1.Operation, error) { - var ops []*v1.Operation - if err := o.db.View(func(tx *bolt.Tx) error { - c := tx.Bucket(OpLogBucket).Cursor() - for k, v := c.First(); k != nil; k, v = c.Next() { - op := &v1.Operation{} - if err := proto.Unmarshal(v, op); err != nil { - return fmt.Errorf("error unmarshalling operation: %w", err) - } - ops = append(ops, op) - } - return nil - }); err != nil { - return nil, err - } - return ops, nil -} - -func (o *OpLog) Subscribe(callback *func(EventType, *v1.Operation)) { - o.subscribersMu.Lock() - defer o.subscribersMu.Unlock() - o.subscribers = append(o.subscribers, callback) -} - -func (o *OpLog) Unsubscribe(callback *func(EventType, *v1.Operation)) { - o.subscribersMu.Lock() - defer o.subscribersMu.Unlock() - subs := o.subscribers - for i, c := range subs { - if c == callback { - subs[i] = subs[len(subs)-1] - o.subscribers = subs[:len(o.subscribers)-1] - } - } -} - -type Filter func([]int64) []int64 - -func FilterKeepAll() Filter { - return func(ids []int64) []int64 { - return ids - } -} - -func FilterLastN(n int64) Filter { - return func(ids []int64) []int64 { - if len(ids) > int(n) { - ids = ids[len(ids)-int(n):] - } - return ids - } -} - -func FilterLimitOffset(limit, offset int64) Filter { - return func(ids []int64) []int64 { - if len(ids) > int(offset) { - ids = ids[offset:] - } - if len(ids) > int(limit) { - ids = ids[:limit] - } - return ids - } -} diff --git a/internal/database/serializationutil/serializationutil.go b/internal/database/serializationutil/serializationutil.go deleted file mode 100644 index 62b5db90..00000000 --- a/internal/database/serializationutil/serializationutil.go +++ /dev/null @@ -1,53 +0,0 @@ -package serializationutil - -import ( - "encoding/binary" - "errors" -) - -var ErrInvalidLength = errors.New("invalid length") - -func Itob(v int64) []byte { - b := make([]byte, 8) - binary.BigEndian.PutUint64(b, uint64(v)) - return b -} - -func Btoi(b []byte) (int64, error) { - if len(b) != 8 { - return 0, ErrInvalidLength - } - return int64(binary.BigEndian.Uint64(b)), nil -} - -func Stob(v string) []byte { - var b []byte - b = append(b, Itob(int64(len(v)))...) - b = append(b, []byte(v)...) - return b -} - -func Btos(b []byte) (string, int64, error) { - if len(b) < 8 { - return "", 0, ErrInvalidLength - } - length, _ := Btoi(b[:8]) - if int64(len(b)) < 8+length { - return "", 0, ErrInvalidLength - } - return string(b[8:8+length]), 8+length, nil -} - -func BytesToKey(b []byte) []byte { - var key []byte - key = append(key, Itob(int64(len(b)))...) - key = append(key, b...) - return key -} - -func NormalizeSnapshotId(id string) string { - if len(id) < 8 { - return id - } - return id[:8] -} \ No newline at end of file diff --git a/internal/database/serializationutil/serializationutil_test.go b/internal/database/serializationutil/serializationutil_test.go deleted file mode 100644 index 524a768e..00000000 --- a/internal/database/serializationutil/serializationutil_test.go +++ /dev/null @@ -1,23 +0,0 @@ -package serializationutil - -import "testing" - -func TestItoa(t *testing.T) { - nums := []int64{0, 1, 2, 3, 4, 1 << 32, int64(1) << 62} - for _, num := range nums { - b := Itob(num) - if v, _ := Btoi(b); v != num { - t.Errorf("itob/btoi failed for %d", num) - } - } -} - -func TestStob(t *testing.T) { - strs := []string{"", "a", "ab", "abc", "abcd", "abcde", "abcdef"} - for _, str := range strs { - b := Stob(str) - if val, _, _ := Btos(b); val != str { - t.Errorf("stob/btos failed for %s", str) - } - } -} diff --git a/internal/oplog/oplog.go b/internal/oplog/oplog.go new file mode 100644 index 00000000..dcdc5a44 --- /dev/null +++ b/internal/oplog/oplog.go @@ -0,0 +1,244 @@ +package oplog + +import ( + "errors" + "fmt" + "os" + "path" + "sync" + "time" + + storm "github.com/asdine/storm/v3" + "github.com/asdine/storm/v3/codec/protobuf" + "github.com/asdine/storm/v3/index" + v1 "github.com/garethgeorge/resticui/gen/go/v1" + bolt "go.etcd.io/bbolt" + "go.uber.org/zap" +) + +type EventType int + +const ( + EventTypeUnknown = EventType(iota) + EventTypeOpCreated = EventType(iota) + EventTypeOpUpdated = EventType(iota) +) + +const ( + OpLogSchemaVersion = 1 +) + +var ( + OpLogSysBucket = "oplog.sys" + OpLogBucket = "oplog.log" + schemaVersionKey = "schema_version" +) + +// OpLog represents a log of operations performed. +// Operations are indexed by repo and plan. +type OpLog struct { + db *bolt.DB + sdb *storm.DB + + subscribersMu sync.RWMutex + subscribers []*func(EventType, *v1.Operation) +} + +func NewOpLog(databasePath string) (*OpLog, error) { + if err := os.MkdirAll(path.Dir(databasePath), 0700); err != nil { + return nil, fmt.Errorf("creating database directory: %s", err) + } + + db, err := bolt.Open(databasePath, 0600, &bolt.Options{Timeout: 1 * time.Second}) + if err != nil { + return nil, fmt.Errorf("opening database: %s", err) + } + + sdb, err := storm.Open("", storm.UseDB(db), storm.Codec(protobuf.Codec)) + if err != nil { + return nil, fmt.Errorf("opening storm: %s", err) + } + + var schemaVersion int + if err := sdb.Get(OpLogSysBucket, schemaVersionKey, &schemaVersion); err != nil { + if errors.Is(err, storm.ErrNotFound) { + schemaVersion = -1 + } else { + return nil, fmt.Errorf("getting schema version: %s", err) + } + } + if schemaVersion != OpLogSchemaVersion { + zap.L().Info("oplog schema version mismatch, reindexing") + sdb.From(OpLogBucket).ReIndex(&v1.Operation{}) + if err := sdb.Set(OpLogSysBucket, schemaVersionKey, OpLogSchemaVersion); err != nil { + return nil, fmt.Errorf("setting schema version: %s", err) + } + } + + return &OpLog{ + db: db, + sdb: sdb, + }, nil +} + +func (o *OpLog) Close() error { + return o.db.Close() +} + +// Add adds a generic operation to the operation log. +func (o *OpLog) Add(op *v1.Operation) error { + if op.Id != 0 { + return errors.New("operation already has an ID, OpLog.Add is expected to set the ID") + } + op.SnapshotId = NormalizeSnapshotId(op.SnapshotId) + + if err := o.sdb.From(OpLogBucket).Save(op); err != nil { + return fmt.Errorf("saving operation: %w", err) + } + + o.notifyHelper(EventTypeOpCreated, op) + return nil +} + +func (o *OpLog) BulkAdd(ops []*v1.Operation) error { + tx, err := o.sdb.From(OpLogBucket).Begin(true) + if err != nil { + return fmt.Errorf("beginning transaction: %w", err) + } + defer tx.Rollback() + + for _, op := range ops { + if op.Id != 0 { + return errors.New("operation already has an ID, OpLog.BulkAdd is expected to set the ID") + } + + if err := tx.Save(op); err != nil { + return fmt.Errorf("saving operation: %w", err) + } + } + + if err := tx.Commit(); err != nil { + return fmt.Errorf("committing transaction: %w", err) + } + + for _, op := range ops { + o.notifyHelper(EventTypeOpCreated, op) + } + + return nil +} + +func (o *OpLog) Update(op *v1.Operation) error { + if err := o.sdb.From(OpLogBucket).Update(op); err != nil { + return fmt.Errorf("updating operation %v: %w", op.Id, err) + } + o.notifyHelper(EventTypeOpUpdated, op) + return nil +} + +func (o *OpLog) notifyHelper(eventType EventType, op *v1.Operation) { + o.subscribersMu.RLock() + defer o.subscribersMu.RUnlock() + for _, sub := range o.subscribers { + (*sub)(eventType, op) + } +} + +func (o *OpLog) Get(id int64) (*v1.Operation, error) { + var op v1.Operation + if err := o.sdb.From(OpLogBucket).One("Id", id, &op); err != nil { + return nil, fmt.Errorf("getting operation: %w", err) + } + return &op, nil +} + +func (o *OpLog) GetByRepo(repoId string, filter Filter) ([]*v1.Operation, error) { + var ops []v1.Operation + if err := o.sdb.From(OpLogBucket).Find("RepoId", repoId, &ops); err != nil { + if errors.Is(err, storm.ErrNotFound) { + return nil, nil + } + return nil, fmt.Errorf("getting operations by repo %q: %w", repoId, err) + } + return opsToRefs(ops), nil +} + +func (o *OpLog) GetByPlan(planId string, filter Filter) ([]*v1.Operation, error) { + var ops []v1.Operation + if err := o.sdb.From(OpLogBucket).Find("PlanId", planId, &ops, filter); err != nil { + if errors.Is(err, storm.ErrNotFound) { + return nil, nil + } + return nil, fmt.Errorf("getting operations by plan %q: %w", planId, err) + } + return opsToRefs(ops), nil +} + +func (o *OpLog) GetBySnapshotId(snapshotId string) ([]*v1.Operation, error) { + snapshotId = NormalizeSnapshotId(snapshotId) + var ops []v1.Operation + if err := o.sdb.From(OpLogBucket).Find("SnapshotId", snapshotId, &ops); err != nil { + if errors.Is(err, storm.ErrNotFound) { + return nil, nil + } + return nil, fmt.Errorf("getting operations by snapshot %q: %w", snapshotId, err) + } + return opsToRefs(ops), nil +} +func (o *OpLog) GetAll(filter Filter) ([]*v1.Operation, error) { + ops := []v1.Operation{} + if err := o.sdb.From(OpLogBucket).All(ops); err != nil { + return nil, fmt.Errorf("getting all operations: %w", err) + } + return opsToRefs(ops), nil +} + +func (o *OpLog) Subscribe(callback *func(EventType, *v1.Operation)) { + o.subscribersMu.Lock() + defer o.subscribersMu.Unlock() + o.subscribers = append(o.subscribers, callback) +} + +func (o *OpLog) Unsubscribe(callback *func(EventType, *v1.Operation)) { + o.subscribersMu.Lock() + defer o.subscribersMu.Unlock() + subs := o.subscribers + for i, c := range subs { + if c == callback { + subs[i] = subs[len(subs)-1] + o.subscribers = subs[:len(o.subscribers)-1] + } + } +} + +type Filter func(*index.Options) + +func FilterKeepAll() Filter { + return func(*index.Options) {} +} + +func FilterLastN(n int64) Filter { + return storm.Limit(int(n)) +} + +func FilterLimitOffset(limit, offset int64) Filter { + return func(opts *index.Options) { + storm.Skip(int(offset))(opts) + storm.Limit(int(limit))(opts) + } +} + +func NormalizeSnapshotId(id string) string { + if len(id) <= 8 { + return id + } + return id[:8] +} + +func opsToRefs(ops []v1.Operation) []*v1.Operation { + refs := make([]*v1.Operation, len(ops)) + for i := range ops { + refs[i] = &ops[i] + } + return refs +} diff --git a/internal/database/oplog/oplog_test.go b/internal/oplog/oplog_test.go similarity index 71% rename from internal/database/oplog/oplog_test.go rename to internal/oplog/oplog_test.go index 60d60d6e..4792790f 100644 --- a/internal/database/oplog/oplog_test.go +++ b/internal/oplog/oplog_test.go @@ -7,6 +7,56 @@ import ( v1 "github.com/garethgeorge/resticui/gen/go/v1" ) +type MyStruct struct { + ID string `storm:"id"` + Field string `storm:"index"` +} + +func TestDatabase(t *testing.T) { + log, err := NewOpLog(t.TempDir() + "/test.boltdb") + if err != nil { + t.Fatalf("error creating oplog: %s", err) + } + + t.Run("can store struct", func(t *testing.T) { + log.sdb.From("struct").Save(&MyStruct{ID: "1", Field: "test"}) + + var s MyStruct + if err := log.sdb.From("struct").One("Field", "test", &s); err != nil { + t.Fatalf("error getting struct: %s", err) + } + t.Logf("Got struct: %+v", s) + if s.ID != "1" { + t.Errorf("want ID 1, got %s", s.ID) + } + if s.Field != "test" { + t.Errorf("want field test, got %s", s.Field) + } + }) + + t.Run("can store proto", func(t *testing.T) { + log.sdb.From("proto").Save(&v1.Operation{Id: 1, PlanId: "foo", RepoId: "bar"}) + + var op v1.Operation + if err := log.sdb.From("proto").One("PlanId", "foo", &op); err != nil { + t.Fatalf("error getting operation: %s", err) + } + t.Logf("Got operation: %+v", op) + if op.Id != 1 { + t.Errorf("want ID 1, got %d", op.Id) + } + + var ops []v1.Operation + if err := log.sdb.From("proto").Find("RepoId", "bar", &ops); err != nil { + t.Fatalf("error getting operations: %s", err) + } + t.Logf("Got operations: %+v", ops) + if len(ops) != 1 { + t.Errorf("want 1 operation, got %d", len(ops)) + } + }) +} + func TestCreate(t *testing.T) { // t.Parallel() log, err := NewOpLog(t.TempDir() + "/test.boltdb") @@ -32,18 +82,10 @@ func TestAddOperation(t *testing.T) { op *v1.Operation wantErr bool }{ - { - name: "no operation", - op: &v1.Operation{ - Id: 0, - }, - wantErr: true, - }, { name: "basic backup operation", op: &v1.Operation{ Id: 0, - Op: &v1.Operation_OperationBackup{}, }, wantErr: false, }, @@ -61,21 +103,10 @@ func TestAddOperation(t *testing.T) { }, wantErr: false, }, - { - name: "basic snapshot operation with no snapshot", - op: &v1.Operation{ - Id: 0, - Op: &v1.Operation_OperationIndexSnapshot{ - OperationIndexSnapshot: &v1.OperationIndexSnapshot{}, - }, - }, - wantErr: true, - }, { name: "operation with ID", op: &v1.Operation{ Id: 1, - Op: &v1.Operation_OperationBackup{}, }, wantErr: true, }, @@ -84,7 +115,6 @@ func TestAddOperation(t *testing.T) { op: &v1.Operation{ Id: 0, RepoId: "testrepo", - Op: &v1.Operation_OperationBackup{}, }, }, { @@ -92,7 +122,6 @@ func TestAddOperation(t *testing.T) { op: &v1.Operation{ Id: 0, PlanId: "testplan", - Op: &v1.Operation_OperationBackup{}, }, }, } @@ -106,13 +135,20 @@ func TestAddOperation(t *testing.T) { if tc.op.Id == 0 { t.Errorf("Add() did not set op ID") } + + got, err := log.Get(tc.op.Id) + if err != nil { + t.Errorf("Get() error = %v", err) + } + if got.Id != tc.op.Id { + t.Errorf("Get() got = %+v, want %+v", got, tc.op) + } } }) } } func TestListOperation(t *testing.T) { - // t.Parallel() log, err := NewOpLog(t.TempDir() + "/test.boltdb") if err != nil { t.Fatalf("error creating oplog: %s", err) @@ -125,19 +161,16 @@ func TestListOperation(t *testing.T) { PlanId: "plan1", RepoId: "repo1", DisplayMessage: "op1", - Op: &v1.Operation_OperationBackup{}, }, { PlanId: "plan1", RepoId: "repo2", DisplayMessage: "op2", - Op: &v1.Operation_OperationBackup{}, }, { PlanId: "plan2", RepoId: "repo2", DisplayMessage: "op3", - Op: &v1.Operation_OperationBackup{}, }, } @@ -183,7 +216,7 @@ func TestListOperation(t *testing.T) { for _, tc := range tests { tc := tc t.Run(tc.name, func(t *testing.T) { - // t.Parallel() + t.Parallel() var ops []*v1.Operation var err error if tc.byPlan { @@ -205,7 +238,7 @@ func TestListOperation(t *testing.T) { } func TestBigIO(t *testing.T) { - // t.Parallel() + t.Skip() log, err := NewOpLog(t.TempDir() + "/test.boltdb") if err != nil { t.Fatalf("error creating oplog: %s", err) @@ -248,8 +281,9 @@ func TestIndexSnapshot(t *testing.T) { t.Cleanup(func() { log.Close() }) op := &v1.Operation{ - PlanId: "plan1", - RepoId: "repo1", + PlanId: "plan1", + RepoId: "repo1", + SnapshotId: NormalizeSnapshotId("abcdefghijklmnop"), Op: &v1.Operation_OperationIndexSnapshot{ OperationIndexSnapshot: &v1.OperationIndexSnapshot{ Snapshot: &v1.ResticSnapshot{ @@ -262,27 +296,30 @@ func TestIndexSnapshot(t *testing.T) { t.Fatalf("error adding operation: %s", err) } - id, err := log.HasIndexedSnapshot("abcdefgh") + ops, err := log.GetBySnapshotId("abcdefgh") if err != nil { t.Fatalf("error checking for snapshot: %s", err) } - if id != op.Id { - t.Fatalf("want id %d, got %d", op.Id, id) + if len(ops) != 1 { + t.Fatalf("want 1 operation, got %d", len(ops)) + } + if ops[0].Id != op.Id { + t.Fatalf("want id %d, got %d", op.Id, ops[0].Id) } - id, err = log.HasIndexedSnapshot("notfound") + ops, err = log.GetBySnapshotId("notfound") if err != nil { t.Fatalf("error checking for snapshot: %s", err) } - if id != -1 { - t.Fatalf("want id -1, got %d", id) + if len(ops) != 0 { + t.Fatalf("want 0 operations, got %d", len(ops)) } } func collectMessages(ops []*v1.Operation) []string { var messages []string - for _, op := range ops { - messages = append(messages, op.DisplayMessage) + for idx, _ := range ops { + messages = append(messages, ops[idx].DisplayMessage) } return messages } diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go index 3aa1d617..2f8eb506 100644 --- a/internal/orchestrator/orchestrator.go +++ b/internal/orchestrator/orchestrator.go @@ -9,7 +9,7 @@ import ( v1 "github.com/garethgeorge/resticui/gen/go/v1" "github.com/garethgeorge/resticui/internal/config" - "github.com/garethgeorge/resticui/internal/database/oplog" + "github.com/garethgeorge/resticui/internal/oplog" "github.com/garethgeorge/resticui/pkg/restic" "go.uber.org/zap" "google.golang.org/protobuf/proto" @@ -21,14 +21,13 @@ var ErrPlanNotFound = errors.New("plan not found") // Orchestrator is responsible for managing repos and backups. type Orchestrator struct { - mu sync.Mutex - config *v1.Config - oplog *oplog.OpLog + mu sync.Mutex + config *v1.Config + oplog *oplog.OpLog repoPool *resticRepoPool - configUpdates chan *v1.Config // configUpdates chan makes config changes available to Run() - externTasks chan Task // externTasks is a channel that externally added tasks can be added to, they will be consumed by Run() + externTasks chan Task // externTasks is a channel that externally added tasks can be added to, they will be consumed by Run() } func NewOrchestrator(configProvider config.ConfigStore, oplog *oplog.OpLog) (*Orchestrator, error) { @@ -38,9 +37,9 @@ func NewOrchestrator(configProvider config.ConfigStore, oplog *oplog.OpLog) (*Or } return &Orchestrator{ - config: cfg, - oplog: oplog, - repoPool: newResticRepoPool(&config.MemoryStore{Config: cfg}), + config: cfg, + oplog: oplog, + repoPool: newResticRepoPool(&config.MemoryStore{Config: cfg}), externTasks: make(chan Task, 2), }, nil } @@ -51,7 +50,7 @@ func (o *Orchestrator) ApplyConfig(cfg *v1.Config) error { o.config = cfg zap.L().Debug("Applying config to orchestrator", zap.Any("config", cfg)) - + // Update the config provided to the repo pool. if err := o.repoPool.configProvider.Update(cfg); err != nil { return fmt.Errorf("failed to update repo pool config: %w", err) @@ -70,7 +69,7 @@ func (o *Orchestrator) GetRepo(repoId string) (repo *RepoOrchestrator, err error defer o.mu.Unlock() r, err := o.repoPool.GetRepo(repoId) - if err != nil { + if err != nil { return nil, fmt.Errorf("failed to get repo %q: %w", repoId, err) } return r, nil @@ -119,7 +118,7 @@ func (o *Orchestrator) Run(mainCtx context.Context) error { func (o *Orchestrator) runVersion(mainCtx context.Context, config *v1.Config) bool { var lock sync.Mutex ctx, cancel := context.WithCancel(mainCtx) - + var wg sync.WaitGroup var execTask func(t Task) @@ -149,7 +148,7 @@ func (o *Orchestrator) runVersion(mainCtx context.Context, config *v1.Config) bo lock.Lock() defer lock.Unlock() zap.L().Info("running task", zap.String("task", t.Name())) - + // Task execution runs with mainCtx meaning config changes do not interrupt it, but cancelling the orchestration loop will. if err := t.Run(mainCtx); err != nil { zap.L().Error("task failed", zap.String("task", t.Name()), zap.Error(err)) @@ -159,7 +158,7 @@ func (o *Orchestrator) runVersion(mainCtx context.Context, config *v1.Config) bo if ctx.Err() != nil { zap.L().Debug("not attempting to reschedule task, orchestrator context is cancelled.", zap.String("task", t.Name())) - return + return } execTask(t) @@ -202,15 +201,14 @@ func (o *Orchestrator) EnqueueTask(t Task) { // resticRepoPool caches restic repos. type resticRepoPool struct { - mu sync.Mutex - repos map[string]*RepoOrchestrator + mu sync.Mutex + repos map[string]*RepoOrchestrator configProvider config.ConfigStore } - func newResticRepoPool(configProvider config.ConfigStore) *resticRepoPool { return &resticRepoPool{ - repos: make(map[string]*RepoOrchestrator), + repos: make(map[string]*RepoOrchestrator), configProvider: configProvider, } } @@ -244,7 +242,7 @@ func (rp *resticRepoPool) GetRepo(repoId string) (repo *RepoOrchestrator, err er if ok && proto.Equal(repo.repoConfig, repoProto) { return repo, nil } - delete(rp.repos, repoId); + delete(rp.repos, repoId) var opts []restic.GenericOption opts = append(opts, restic.WithPropagatedEnvVars(restic.EnvToPropagate...)) diff --git a/internal/orchestrator/tasks.go b/internal/orchestrator/tasks.go index 3c693401..409b4afd 100644 --- a/internal/orchestrator/tasks.go +++ b/internal/orchestrator/tasks.go @@ -6,16 +6,15 @@ import ( "time" v1 "github.com/garethgeorge/resticui/gen/go/v1" - "github.com/garethgeorge/resticui/internal/database/oplog" + "github.com/garethgeorge/resticui/internal/oplog" "github.com/garethgeorge/resticui/pkg/restic" "github.com/gitploy-io/cronexpr" "github.com/hashicorp/go-multierror" "go.uber.org/zap" ) - type Task interface { - Name() string // huamn readable name for this task. + Name() string // huamn readable name for this task. Next(now time.Time) *time.Time // when this task would like to be run. Run(ctx context.Context) error // run the task. } @@ -23,8 +22,8 @@ type Task interface { // BackupTask is a scheduled backup operation. type ScheduledBackupTask struct { orchestrator *Orchestrator // owning orchestrator - plan *v1.Plan - schedule *cronexpr.Schedule + plan *v1.Plan + schedule *cronexpr.Schedule } var _ Task = &ScheduledBackupTask{} @@ -37,8 +36,8 @@ func NewScheduledBackupTask(orchestrator *Orchestrator, plan *v1.Plan) (*Schedul return &ScheduledBackupTask{ orchestrator: orchestrator, - plan: plan, - schedule: sched, + plan: plan, + schedule: sched, }, nil } @@ -58,15 +57,15 @@ func (t *ScheduledBackupTask) Run(ctx context.Context) error { // OnetimeBackupTask is a single backup operation. type OnetimeBackupTask struct { orchestrator *Orchestrator - plan *v1.Plan - time *time.Time + plan *v1.Plan + time *time.Time } func NewOneofBackupTask(orchestrator *Orchestrator, plan *v1.Plan, at time.Time) *OnetimeBackupTask { return &OnetimeBackupTask{ orchestrator: orchestrator, - plan: plan, - time: &at, + plan: plan, + time: &at, } } @@ -75,9 +74,9 @@ func (t *OnetimeBackupTask) Name() string { } func (t *OnetimeBackupTask) Next(now time.Time) *time.Time { - ret := t.time + ret := t.time t.time = nil - return ret + return ret } func (t *OnetimeBackupTask) Run(ctx context.Context) error { @@ -91,11 +90,11 @@ func backupHelper(ctx context.Context, orchestrator *Orchestrator, plan *v1.Plan } op := &v1.Operation{ - PlanId: plan.Id, - RepoId: plan.Repo, + PlanId: plan.Id, + RepoId: plan.Repo, UnixTimeStartMs: curTimeMillis(), - Status: v1.OperationStatus_STATUS_INPROGRESS, - Op: backupOp, + Status: v1.OperationStatus_STATUS_INPROGRESS, + Op: backupOp, } startTime := time.Now() @@ -109,7 +108,7 @@ func backupHelper(ctx context.Context, orchestrator *Orchestrator, plan *v1.Plan lastSent := time.Now() // debounce progress updates, these can endup being very frequent. summary, err := repo.Backup(ctx, plan, func(entry *restic.BackupProgressEntry) { - if time.Since(lastSent) < 200 * time.Millisecond { + if time.Since(lastSent) < 200*time.Millisecond { return } lastSent = time.Now() @@ -124,10 +123,9 @@ func backupHelper(ctx context.Context, orchestrator *Orchestrator, plan *v1.Plan return fmt.Errorf("repo.Backup for repo %q: %w", plan.Repo, err) } + // Set final properties (i.e. SnapshotId and LastStatus) + op.SnapshotId = oplog.NormalizeSnapshotId(summary.SnapshotId) backupOp.OperationBackup.LastStatus = summary.ToProto() - if err := orchestrator.oplog.Update(op); err != nil { - return fmt.Errorf("update oplog with summary for backup: %v", err) - } zap.L().Info("backup complete", zap.String("plan", plan.Id), zap.Duration("duration", time.Since(startTime))) return nil @@ -160,22 +158,32 @@ func indexSnapshotsHelper(ctx context.Context, orchestrator *Orchestrator, plan opTime := curTimeMillis() var indexOps []*v1.Operation for _, snapshot := range snapshots { - opid, err := orchestrator.oplog.HasIndexedSnapshot(snapshot.Id) + ops, err := orchestrator.oplog.GetBySnapshotId(snapshot.Id) if err != nil { return fmt.Errorf("HasIndexSnapshot for snapshot %q: %w", snapshot.Id, err) } - if opid >= 0 { + // check if an indexing operation exists for this snapshot + indexOpFound := false + for _, op := range ops { + if _, ok := op.Op.(*v1.Operation_OperationIndexSnapshot); ok { + indexOpFound = true + break + } + } + if indexOpFound { alreadyIndexed += 1 continue } - + + // otherwise add one. indexOps = append(indexOps, &v1.Operation{ - RepoId: plan.Repo, - PlanId: plan.Id, + RepoId: plan.Repo, + PlanId: plan.Id, + SnapshotId: oplog.NormalizeSnapshotId(snapshot.Id), UnixTimeStartMs: opTime, - UnixTimeEndMs: opTime, - Status: v1.OperationStatus_STATUS_SUCCESS, + UnixTimeEndMs: opTime, + Status: v1.OperationStatus_STATUS_SUCCESS, Op: &v1.Operation_OperationIndexSnapshot{ OperationIndexSnapshot: &v1.OperationIndexSnapshot{ Snapshot: snapshot.ToProto(), @@ -188,11 +196,11 @@ func indexSnapshotsHelper(ctx context.Context, orchestrator *Orchestrator, plan return fmt.Errorf("BulkAdd snapshot operations: %w", err) } - zap.L().Debug("Indexed snapshots", - zap.String("plan", plan.Id), - zap.Duration("duration", time.Since(startTime)), - zap.Int("alreadyIndexed", alreadyIndexed), - zap.Int("newlyAdded", len(snapshots) - alreadyIndexed), + zap.L().Debug("Indexed snapshots", + zap.String("plan", plan.Id), + zap.Duration("duration", time.Since(startTime)), + zap.Int("alreadyIndexed", alreadyIndexed), + zap.Int("newlyAdded", len(snapshots)-alreadyIndexed), ) return err @@ -209,7 +217,7 @@ func WithOperation(oplog *oplog.OpLog, op *v1.Operation, do func() error) error } err := do() if err != nil { - op.Status = v1.OperationStatus_STATUS_ERROR + op.Status = v1.OperationStatus_STATUS_ERROR op.DisplayMessage = err.Error() } op.UnixTimeEndMs = curTimeMillis() @@ -224,5 +232,5 @@ func WithOperation(oplog *oplog.OpLog, op *v1.Operation, do func() error) error func curTimeMillis() int64 { t := time.Now() - return t.Unix() * 1000 + int64(t.Nanosecond() / 1000000) -} \ No newline at end of file + return t.Unix()*1000 + int64(t.Nanosecond()/1000000) +} diff --git a/proto/buf.gen.yaml b/proto/buf.gen.yaml index 2b42b990..facd1bb1 100644 --- a/proto/buf.gen.yaml +++ b/proto/buf.gen.yaml @@ -16,4 +16,4 @@ plugins: - plugin: grpc-gateway-ts out: ../webui/gen/ts opt: - - paths=source_relative \ No newline at end of file + - paths=source_relative diff --git a/proto/build.sh b/proto/build.sh index 0676734d..38782744 100755 --- a/proto/build.sh +++ b/proto/build.sh @@ -1,2 +1,3 @@ #! /bin/bash buf generate +find ../gen/go -name "*.pb.go" -exec protoc-go-inject-tag -input="{}" \; \ No newline at end of file diff --git a/proto/v1/operations.proto b/proto/v1/operations.proto index 7cc36d67..abb735f6 100644 --- a/proto/v1/operations.proto +++ b/proto/v1/operations.proto @@ -11,11 +11,20 @@ message OperationList { } message Operation { + // @gotags: `storm:"id,increment"` int64 id = 1; - string repo_id = 2; // repo id if associated with a repo (always true) - string plan_id = 3; // plan id if associated with a plan (always true) + // repo id if associated with a repo (always true) + // @gotags: `storm:"index"` + string repo_id = 2; + // plan id if associated with a plan (always true) + // @gotags: `storm:"index"` + string plan_id = 3; + // normalized snapshot id (first 8 chars) if associated with a snapshot + // @gotags: `storm:"index"` + string snapshot_id = 8; OperationStatus status = 4; int64 unix_time_start_ms = 5; + // operation ended time in unix ms int64 unix_time_end_ms = 6; string display_message = 7; // human readable context message (if any) diff --git a/resticui.go b/resticui.go index 3f93a801..82dc5887 100644 --- a/resticui.go +++ b/resticui.go @@ -14,7 +14,7 @@ import ( "github.com/garethgeorge/resticui/internal/api" "github.com/garethgeorge/resticui/internal/config" - "github.com/garethgeorge/resticui/internal/database/oplog" + "github.com/garethgeorge/resticui/internal/oplog" "github.com/garethgeorge/resticui/internal/orchestrator" static "github.com/garethgeorge/resticui/webui" "github.com/mattn/go-colorable" @@ -33,18 +33,17 @@ func main() { ctx := context.Background() ctx, cancel := context.WithCancel(ctx) go onterm(cancel) - + if _, err := config.Default.Get(); err != nil { zap.S().Fatalf("Error loading config: %v", err) } - + var wg sync.WaitGroup // Configure the HTTP mux mux := http.NewServeMux() mux.Handle("/", http.FileServer(http.FS(&SubdirFilesystem{FS: static.FS, subdir: "dist"}))) - // Create and serve API server oplogFile := path.Join(dataPath(), "oplog.boltdb") oplog, err := oplog.NewOpLog(oplogFile) @@ -115,9 +114,9 @@ func init() { c.EncodeLevel = zapcore.CapitalColorLevelEncoder c.EncodeTime = zapcore.ISO8601TimeEncoder l := zap.New(zapcore.NewCore( - zapcore.NewConsoleEncoder(c), - zapcore.AddSync(colorable.NewColorableStdout()), - zapcore.DebugLevel, + zapcore.NewConsoleEncoder(c), + zapcore.AddSync(colorable.NewColorableStdout()), + zapcore.DebugLevel, )) zap.ReplaceGlobals(l) } @@ -142,7 +141,7 @@ func (s *SubdirFilesystem) Open(name string) (fs.File, error) { return s.FS.Open(path.Join(s.subdir, name)) } -func (s *SubdirFilesystem) ReadDir(name string) ([]fs.DirEntry, error) { +func (s *SubdirFilesystem) ReadDir(name string) ([]fs.DirEntry, error) { readDirFS := s.FS.(fs.ReadDirFS) if readDirFS == nil { return nil, &fs.PathError{Op: "readdir", Path: name, Err: errors.New("not implemented")} @@ -156,4 +155,4 @@ func dataPath() string { datahome = path.Join(os.Getenv("HOME") + "/.local/share") } return path.Join(datahome, "resticui") -} \ No newline at end of file +} diff --git a/webui/gen/ts/v1/operations.pb.ts b/webui/gen/ts/v1/operations.pb.ts index 7b1711f0..474b5cbf 100644 --- a/webui/gen/ts/v1/operations.pb.ts +++ b/webui/gen/ts/v1/operations.pb.ts @@ -38,6 +38,7 @@ type BaseOperation = { id?: string repoId?: string planId?: string + snapshotId?: string status?: OperationStatus unixTimeStartMs?: string unixTimeEndMs?: string