diff --git a/README.md b/README.md index 10ca68ad..7addaca2 100644 --- a/README.md +++ b/README.md @@ -76,7 +76,6 @@ 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 dc49dfca..8552ad93 100644 --- a/gen/go/v1/operations.pb.go +++ b/gen/go/v1/operations.pb.go @@ -177,22 +177,13 @@ type Operation struct { sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - // @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"` + 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) 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"` - // 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) + 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 @@ -253,13 +244,6 @@ 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 @@ -485,67 +469,65 @@ 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, 0xba, 0x03, 0x0a, 0x09, 0x4f, + 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x22, 0x99, 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, 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, + 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, } var ( diff --git a/go.mod b/go.mod index 9827816c..918163d6 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,6 @@ 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 @@ -12,7 +11,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-20231120223509-83a465c0220f + google.golang.org/genproto/googleapis/api v0.0.0-20231106174013-bbf56f31fb17 google.golang.org/grpc v1.59.0 google.golang.org/protobuf v1.31.0 ) @@ -25,6 +24,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-20231120223509-83a465c0220f // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20231120223509-83a465c0220f // indirect + google.golang.org/genproto v0.0.0-20231106174013-bbf56f31fb17 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20231106174013-bbf56f31fb17 // indirect ) diff --git a/go.sum b/go.sum index ef3a4e4e..43ab565c 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,3 @@ -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= @@ -49,16 +47,10 @@ 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 06930399..9027daa7 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/oplog" + "github.com/garethgeorge/resticui/internal/database/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 new file mode 100644 index 00000000..4cfdd0da --- /dev/null +++ b/internal/database/indexutil/indexutil.go @@ -0,0 +1,57 @@ +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 new file mode 100644 index 00000000..d04eb072 --- /dev/null +++ b/internal/database/indexutil/indexutil_test.go @@ -0,0 +1,45 @@ +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 new file mode 100644 index 00000000..1459750c --- /dev/null +++ b/internal/database/oplog/oplog.go @@ -0,0 +1,386 @@ +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/oplog/oplog_test.go b/internal/database/oplog/oplog_test.go similarity index 71% rename from internal/oplog/oplog_test.go rename to internal/database/oplog/oplog_test.go index 4792790f..60d60d6e 100644 --- a/internal/oplog/oplog_test.go +++ b/internal/database/oplog/oplog_test.go @@ -7,56 +7,6 @@ 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") @@ -82,10 +32,18 @@ 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, }, @@ -103,10 +61,21 @@ 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, }, @@ -115,6 +84,7 @@ func TestAddOperation(t *testing.T) { op: &v1.Operation{ Id: 0, RepoId: "testrepo", + Op: &v1.Operation_OperationBackup{}, }, }, { @@ -122,6 +92,7 @@ func TestAddOperation(t *testing.T) { op: &v1.Operation{ Id: 0, PlanId: "testplan", + Op: &v1.Operation_OperationBackup{}, }, }, } @@ -135,20 +106,13 @@ 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) @@ -161,16 +125,19 @@ 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{}, }, } @@ -216,7 +183,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 { @@ -238,7 +205,7 @@ func TestListOperation(t *testing.T) { } func TestBigIO(t *testing.T) { - t.Skip() + // t.Parallel() log, err := NewOpLog(t.TempDir() + "/test.boltdb") if err != nil { t.Fatalf("error creating oplog: %s", err) @@ -281,9 +248,8 @@ func TestIndexSnapshot(t *testing.T) { t.Cleanup(func() { log.Close() }) op := &v1.Operation{ - PlanId: "plan1", - RepoId: "repo1", - SnapshotId: NormalizeSnapshotId("abcdefghijklmnop"), + PlanId: "plan1", + RepoId: "repo1", Op: &v1.Operation_OperationIndexSnapshot{ OperationIndexSnapshot: &v1.OperationIndexSnapshot{ Snapshot: &v1.ResticSnapshot{ @@ -296,30 +262,27 @@ func TestIndexSnapshot(t *testing.T) { t.Fatalf("error adding operation: %s", err) } - ops, err := log.GetBySnapshotId("abcdefgh") + id, err := log.HasIndexedSnapshot("abcdefgh") if err != nil { t.Fatalf("error checking for snapshot: %s", err) } - 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) + if id != op.Id { + t.Fatalf("want id %d, got %d", op.Id, id) } - ops, err = log.GetBySnapshotId("notfound") + id, err = log.HasIndexedSnapshot("notfound") if err != nil { t.Fatalf("error checking for snapshot: %s", err) } - if len(ops) != 0 { - t.Fatalf("want 0 operations, got %d", len(ops)) + if id != -1 { + t.Fatalf("want id -1, got %d", id) } } func collectMessages(ops []*v1.Operation) []string { var messages []string - for idx, _ := range ops { - messages = append(messages, ops[idx].DisplayMessage) + for _, op := range ops { + messages = append(messages, op.DisplayMessage) } return messages } diff --git a/internal/database/serializationutil/serializationutil.go b/internal/database/serializationutil/serializationutil.go new file mode 100644 index 00000000..62b5db90 --- /dev/null +++ b/internal/database/serializationutil/serializationutil.go @@ -0,0 +1,53 @@ +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 new file mode 100644 index 00000000..524a768e --- /dev/null +++ b/internal/database/serializationutil/serializationutil_test.go @@ -0,0 +1,23 @@ +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 deleted file mode 100644 index dcdc5a44..00000000 --- a/internal/oplog/oplog.go +++ /dev/null @@ -1,244 +0,0 @@ -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/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go index 2f8eb506..3aa1d617 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/oplog" + "github.com/garethgeorge/resticui/internal/database/oplog" "github.com/garethgeorge/resticui/pkg/restic" "go.uber.org/zap" "google.golang.org/protobuf/proto" @@ -21,13 +21,14 @@ 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) { @@ -37,9 +38,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 } @@ -50,7 +51,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) @@ -69,7 +70,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 @@ -118,7 +119,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) @@ -148,7 +149,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)) @@ -158,7 +159,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) @@ -201,14 +202,15 @@ 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, } } @@ -242,7 +244,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 409b4afd..3c693401 100644 --- a/internal/orchestrator/tasks.go +++ b/internal/orchestrator/tasks.go @@ -6,15 +6,16 @@ import ( "time" v1 "github.com/garethgeorge/resticui/gen/go/v1" - "github.com/garethgeorge/resticui/internal/oplog" + "github.com/garethgeorge/resticui/internal/database/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. } @@ -22,8 +23,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{} @@ -36,8 +37,8 @@ func NewScheduledBackupTask(orchestrator *Orchestrator, plan *v1.Plan) (*Schedul return &ScheduledBackupTask{ orchestrator: orchestrator, - plan: plan, - schedule: sched, + plan: plan, + schedule: sched, }, nil } @@ -57,15 +58,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, } } @@ -74,9 +75,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 { @@ -90,11 +91,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() @@ -108,7 +109,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() @@ -123,9 +124,10 @@ 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 @@ -158,32 +160,22 @@ func indexSnapshotsHelper(ctx context.Context, orchestrator *Orchestrator, plan opTime := curTimeMillis() var indexOps []*v1.Operation for _, snapshot := range snapshots { - ops, err := orchestrator.oplog.GetBySnapshotId(snapshot.Id) + opid, err := orchestrator.oplog.HasIndexedSnapshot(snapshot.Id) if err != nil { return fmt.Errorf("HasIndexSnapshot for snapshot %q: %w", snapshot.Id, err) } - // 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 { + if opid >= 0 { alreadyIndexed += 1 continue } - - // otherwise add one. + indexOps = append(indexOps, &v1.Operation{ - RepoId: plan.Repo, - PlanId: plan.Id, - SnapshotId: oplog.NormalizeSnapshotId(snapshot.Id), + RepoId: plan.Repo, + PlanId: plan.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(), @@ -196,11 +188,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 @@ -217,7 +209,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() @@ -232,5 +224,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) -} + return t.Unix() * 1000 + int64(t.Nanosecond() / 1000000) +} \ No newline at end of file diff --git a/proto/buf.gen.yaml b/proto/buf.gen.yaml index facd1bb1..2b42b990 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 + - paths=source_relative \ No newline at end of file diff --git a/proto/build.sh b/proto/build.sh index 38782744..0676734d 100755 --- a/proto/build.sh +++ b/proto/build.sh @@ -1,3 +1,2 @@ #! /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 abb735f6..7cc36d67 100644 --- a/proto/v1/operations.proto +++ b/proto/v1/operations.proto @@ -11,20 +11,11 @@ message OperationList { } message Operation { - // @gotags: `storm:"id,increment"` int64 id = 1; - // 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; + 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) 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 82dc5887..3f93a801 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/oplog" + "github.com/garethgeorge/resticui/internal/database/oplog" "github.com/garethgeorge/resticui/internal/orchestrator" static "github.com/garethgeorge/resticui/webui" "github.com/mattn/go-colorable" @@ -33,17 +33,18 @@ 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) @@ -114,9 +115,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) } @@ -141,7 +142,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")} @@ -155,4 +156,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 474b5cbf..7b1711f0 100644 --- a/webui/gen/ts/v1/operations.pb.ts +++ b/webui/gen/ts/v1/operations.pb.ts @@ -38,7 +38,6 @@ type BaseOperation = { id?: string repoId?: string planId?: string - snapshotId?: string status?: OperationStatus unixTimeStartMs?: string unixTimeEndMs?: string