lee/backrest
1
0
Fork 0
mirror of https://github.com/garethgeorge/backrest.git synced 2026-08-26 10:56:33 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

feat: implement backup scheduling in orchestrator

Browse Source
This commit is contained in:
garethgeorge
2023-11-14 21:52:06 -08:00
parent dd9142c0e9
commit eadb1a8201
19 changed files with 434 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/resticui/resticui.go
+18 -3
View File
@@ -47,8 +47,22 @@ func main() {
}
defer oplog.Close()
orchestrator, err := orchestrator.NewOrchestrator(config.Default, oplog)
if err != nil {
zap.S().Fatalf("Error creating orchestrator: %v", err)
}
// Start orchestration loop.
go func() {
err := orchestrator.Run(ctx)
if err != nil && !errors.Is(err, context.Canceled) {
zap.S().Fatal("Orchestrator loop exited with error: ", zap.Error(err))
cancel() // cancel the context when the orchestrator exits (e.g. on fatal error)
}
}()
apiServer := api.NewServer(
orchestrator.NewOrchestrator(config.Default), // TODO: eliminate default config
orchestrator, // TODO: eliminate default config
oplog,
)
@@ -78,9 +92,9 @@ func main() {
server.Shutdown(context.Background())
}()
if err := server.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
zap.S().Error("Error starting server", zap.Error(err))
zap.L().Error("Error starting server", zap.Error(err))
}
zap.S().Info("HTTP gateway shutdown")
zap.L().Info("HTTP gateway shutdown")
cancel() // cancel the context when the HTTP server exits (e.g. on fatal error)
}()
@@ -92,6 +106,7 @@ func init() {
if os.Getenv("DEBUG") != "" {
c := zap.NewDevelopmentEncoderConfig()
c.EncodeLevel = zapcore.CapitalColorLevelEncoder
c.EncodeTime = zapcore.ISO8601TimeEncoder
l := zap.New(zapcore.NewCore(
zapcore.NewConsoleEncoder(c),
zapcore.AddSync(colorable.NewColorableStdout()),
gen/go/v1/operations.pb.go
+43 -42
View File
@@ -139,7 +139,7 @@ type Operation struct {
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_Backup
// *Operation_OperationBackup
Op isOperation_Op `protobuf_oneof:"op"`
}
@@ -231,9 +231,9 @@ func (m *Operation) GetOp() isOperation_Op {
return nil
}
func (x *Operation) GetBackup() *OperationBackup {
if x, ok := x.GetOp().(*Operation_Backup); ok {
return x.Backup
func (x *Operation) GetOperationBackup() *OperationBackup {
if x, ok := x.GetOp().(*Operation_OperationBackup); ok {
return x.OperationBackup
}
return nil
}
@@ -242,11 +242,11 @@ type isOperation_Op interface {
isOperation_Op()
}
type Operation_Backup struct {
Backup *OperationBackup `protobuf:"bytes,100,opt,name=backup,proto3,oneof"`
type Operation_OperationBackup struct {
OperationBackup *OperationBackup `protobuf:"bytes,100,opt,name=operation_backup,json=operationBackup,proto3,oneof"`
}
func (*Operation_Backup) isOperation_Op() {}
func (*Operation_OperationBackup) isOperation_Op() {}
// OperationEvent is used in the wireformat to stream operation changes to clients
type OperationEvent struct {
@@ -356,7 +356,7 @@ var File_v1_operations_proto protoreflect.FileDescriptor
var file_v1_operations_proto_rawDesc = []byte{
0x0a, 0x13, 0x76, 0x31, 0x2f, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x2e,
0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x02, 0x76, 0x31, 0x1a, 0x0f, 0x76, 0x31, 0x2f, 0x72, 0x65,
0x73, 0x74, 0x69, 0x63, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xae, 0x02, 0x0a, 0x09, 0x4f,
0x73, 0x74, 0x69, 0x63, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xc1, 0x02, 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,
@@ -372,38 +372,39 @@ var file_v1_operations_proto_rawDesc = []byte{
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, 0x2d, 0x0a, 0x06, 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, 0x06, 0x62,
0x61, 0x63, 0x6b, 0x75, 0x70, 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, 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,
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, 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, 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 (
@@ -430,7 +431,7 @@ var file_v1_operations_proto_goTypes = []interface{}{
}
var file_v1_operations_proto_depIdxs = []int32{
1, // 0: v1.Operation.status:type_name -> v1.OperationStatus
4, // 1: v1.Operation.backup:type_name -> v1.OperationBackup
4, // 1: v1.Operation.operation_backup:type_name -> v1.OperationBackup
0, // 2: v1.OperationEvent.type:type_name -> v1.OperationEventType
2, // 3: v1.OperationEvent.operation:type_name -> v1.Operation
5, // 4: v1.OperationBackup.last_status:type_name -> v1.BackupProgressEntry
@@ -486,7 +487,7 @@ func file_v1_operations_proto_init() {
}
}
file_v1_operations_proto_msgTypes[0].OneofWrappers = []interface{}{
(*Operation_Backup)(nil),
(*Operation_OperationBackup)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
gen/go/v1/restic.pb.go
+15 -15
View File
@@ -340,19 +340,19 @@ type BackupProgressSummary struct {
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
FilesNew int64 `protobuf:"varint,1,opt,name=files_new,json=filesNew,proto3" json:"files_new,omitempty"`
FilesChanged int64 `protobuf:"varint,2,opt,name=files_changed,json=filesChanged,proto3" json:"files_changed,omitempty"`
FilesUnmodified int64 `protobuf:"varint,3,opt,name=files_unmodified,json=filesUnmodified,proto3" json:"files_unmodified,omitempty"`
DirsNew int64 `protobuf:"varint,4,opt,name=dirs_new,json=dirsNew,proto3" json:"dirs_new,omitempty"`
DirsChanged int64 `protobuf:"varint,5,opt,name=dirs_changed,json=dirsChanged,proto3" json:"dirs_changed,omitempty"`
DirsUnmodified int64 `protobuf:"varint,6,opt,name=dirs_unmodified,json=dirsUnmodified,proto3" json:"dirs_unmodified,omitempty"`
DataBlobs int64 `protobuf:"varint,7,opt,name=data_blobs,json=dataBlobs,proto3" json:"data_blobs,omitempty"`
TreeBlobs int64 `protobuf:"varint,8,opt,name=tree_blobs,json=treeBlobs,proto3" json:"tree_blobs,omitempty"`
DataAdded int64 `protobuf:"varint,9,opt,name=data_added,json=dataAdded,proto3" json:"data_added,omitempty"`
TotalFilesProcessed int64 `protobuf:"varint,10,opt,name=total_files_processed,json=totalFilesProcessed,proto3" json:"total_files_processed,omitempty"`
TotalBytesProcessed int64 `protobuf:"varint,11,opt,name=total_bytes_processed,json=totalBytesProcessed,proto3" json:"total_bytes_processed,omitempty"`
TotalDuration int64 `protobuf:"varint,12,opt,name=total_duration,json=totalDuration,proto3" json:"total_duration,omitempty"`
SnapshotId string `protobuf:"bytes,13,opt,name=snapshot_id,json=snapshotId,proto3" json:"snapshot_id,omitempty"`
FilesNew int64 `protobuf:"varint,1,opt,name=files_new,json=filesNew,proto3" json:"files_new,omitempty"`
FilesChanged int64 `protobuf:"varint,2,opt,name=files_changed,json=filesChanged,proto3" json:"files_changed,omitempty"`
FilesUnmodified int64 `protobuf:"varint,3,opt,name=files_unmodified,json=filesUnmodified,proto3" json:"files_unmodified,omitempty"`
DirsNew int64 `protobuf:"varint,4,opt,name=dirs_new,json=dirsNew,proto3" json:"dirs_new,omitempty"`
DirsChanged int64 `protobuf:"varint,5,opt,name=dirs_changed,json=dirsChanged,proto3" json:"dirs_changed,omitempty"`
DirsUnmodified int64 `protobuf:"varint,6,opt,name=dirs_unmodified,json=dirsUnmodified,proto3" json:"dirs_unmodified,omitempty"`
DataBlobs int64 `protobuf:"varint,7,opt,name=data_blobs,json=dataBlobs,proto3" json:"data_blobs,omitempty"`
TreeBlobs int64 `protobuf:"varint,8,opt,name=tree_blobs,json=treeBlobs,proto3" json:"tree_blobs,omitempty"`
DataAdded int64 `protobuf:"varint,9,opt,name=data_added,json=dataAdded,proto3" json:"data_added,omitempty"`
TotalFilesProcessed int64 `protobuf:"varint,10,opt,name=total_files_processed,json=totalFilesProcessed,proto3" json:"total_files_processed,omitempty"`
TotalBytesProcessed int64 `protobuf:"varint,11,opt,name=total_bytes_processed,json=totalBytesProcessed,proto3" json:"total_bytes_processed,omitempty"`
TotalDuration float64 `protobuf:"fixed64,12,opt,name=total_duration,json=totalDuration,proto3" json:"total_duration,omitempty"`
SnapshotId string `protobuf:"bytes,13,opt,name=snapshot_id,json=snapshotId,proto3" json:"snapshot_id,omitempty"`
}
func (x *BackupProgressSummary) Reset() {
@@ -464,7 +464,7 @@ func (x *BackupProgressSummary) GetTotalBytesProcessed() int64 {
return 0
}
func (x *BackupProgressSummary) GetTotalDuration() int64 {
func (x *BackupProgressSummary) GetTotalDuration() float64 {
if x != nil {
return x.TotalDuration
}
@@ -549,7 +549,7 @@ var file_v1_restic_proto_rawDesc = []byte{
0x5f, 0x70, 0x72, 0x6f, 0x63, 0x65, 0x73, 0x73, 0x65, 0x64, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x03,
0x52, 0x13, 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x42, 0x79, 0x74, 0x65, 0x73, 0x50, 0x72, 0x6f, 0x63,
0x65, 0x73, 0x73, 0x65, 0x64, 0x12, 0x25, 0x0a, 0x0e, 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x5f, 0x64,
0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x0c, 0x20, 0x01, 0x28, 0x03, 0x52, 0x0d, 0x74,
0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x0c, 0x20, 0x01, 0x28, 0x01, 0x52, 0x0d, 0x74,
0x6f, 0x74, 0x61, 0x6c, 0x44, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1f, 0x0a, 0x0b,
0x73, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x0d, 0x20, 0x01, 0x28,
0x09, 0x52, 0x0a, 0x73, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x49, 0x64, 0x42, 0x2e, 0x5a,
go.mod
+1
View File
@@ -30,6 +30,7 @@ require (
github.com/golang/snappy v0.0.4 // indirect
github.com/google/flatbuffers v23.5.26+incompatible // indirect
github.com/google/orderedcode v0.0.1 // indirect
github.com/gorhill/cronexpr v0.0.0-20180427100037-88b0669f7d75 // indirect
github.com/grpc-ecosystem/grpc-gateway v1.16.0 // indirect
github.com/hashicorp/errwrap v1.1.0 // indirect
github.com/klauspost/compress v1.17.2 // indirect
go.sum
+2
View File
@@ -84,6 +84,8 @@ github.com/google/orderedcode v0.0.1/go.mod h1:iVyU4/qPKHY5h/wSd6rZZCDcLJNxiWO6d
github.com/google/renameio v1.0.1 h1:Lh/jXZmvZxb0BBeSY5VKEfidcbcbenKjZFzM/q0fSeU=
github.com/google/renameio v1.0.1/go.mod h1:t/HQoYBZSsWSNK35C6CO/TpPLDVWvxOHboWUAweKUpk=
github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorhill/cronexpr v0.0.0-20180427100037-88b0669f7d75 h1:f0n1xnMSmBLzVfsMMvriDyA75NB/oBgILX2GcHXIQzY=
github.com/gorhill/cronexpr v0.0.0-20180427100037-88b0669f7d75/go.mod h1:g2644b03hfBX9Ov0ZBDgXXens4rxSxmqFBbhvKv2yVA=
github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.1 h1:HcUWd006luQPljE73d5sk+/VgYPGUReEVz2y1/qylwY=
github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.1/go.mod h1:w9Y7gY31krpLmrVU5ZPG9H7l9fZuRu5/3R3S3FMtVQ4=
github.com/grpc-ecosystem/grpc-gateway v1.16.0 h1:gmcG1KaJ57LophUzW0Hy8NmPhnMZb4M0+kPpLofRdBo=
internal/api/server.go
+4 -19
View File
@@ -5,8 +5,6 @@ import (
"errors"
"fmt"
"os"
"sync"
"sync/atomic"
"github.com/garethgeorge/resticui/gen/go/types"
v1 "github.com/garethgeorge/resticui/gen/go/v1"
@@ -23,17 +21,12 @@ type Server struct {
*v1.UnimplementedResticUIServer
orchestrator *orchestrator.Orchestrator
oplog *oplog.OpLog
reqId atomic.Uint64
eventChannelsMu sync.Mutex
eventChannels map[uint64]chan *v1.Event
}
var _ v1.ResticUIServer = &Server{}
func NewServer(orchestrator *orchestrator.Orchestrator, oplog *oplog.OpLog) *Server {
s := &Server{
eventChannels: make(map[uint64]chan *v1.Event),
orchestrator: orchestrator,
}
@@ -84,11 +77,13 @@ func (s *Server) AddRepo(ctx context.Context, repo *v1.Repo) (*v1.Config, error)
return nil, fmt.Errorf("failed to init repo: %w", err)
}
zap.S().Debug("Updating config")
zap.L().Debug("Updating config")
if err := config.Default.Update(c); err != nil {
return nil, fmt.Errorf("failed to update config: %w", err)
}
s.orchestrator.ApplyConfig(c)
return c, nil
}
@@ -139,7 +134,7 @@ func (s *Server) GetOperationEvents(_ *emptypb.Empty, stream v1.ResticUI_GetOper
case oplog.EventTypeOpUpdated:
eventTypeMapped = v1.OperationEventType_EVENT_UPDATED
default:
zap.S().Error("Unknown event type", zap.Int("eventType", int(eventType)))
zap.L().Error("Unknown event type", zap.Int("eventType", int(eventType)))
eventTypeMapped = v1.OperationEventType_EVENT_UNKNOWN
}
@@ -179,13 +174,3 @@ func (s *Server) PathAutocomplete(ctx context.Context, path *types.StringValue)
return &types.StringList{Values: paths}, nil
}
// PublishEvent publishes an event to all GetEvents streams. It is effectively a multicast.
func (s *Server) PublishEvent(event *v1.Event) {
zap.S().Debug("Publishing event", zap.Any("event", event))
s.eventChannelsMu.Lock()
defer s.eventChannelsMu.Unlock()
for _, ch := range s.eventChannels {
ch <- event
}
}
internal/config/memstore.go
+27
View File
@@ -0,0 +1,27 @@
package config
import (
"sync"
v1 "github.com/garethgeorge/resticui/gen/go/v1"
)
type MemoryStore struct {
mu sync.Mutex
Config *v1.Config
}
var _ ConfigStore = &MemoryStore{}
func (c *MemoryStore) Get() (*v1.Config, error) {
c.mu.Lock()
defer c.mu.Unlock()
return c.Config, nil
}
func (c *MemoryStore) Update(config *v1.Config) error {
c.mu.Lock()
defer c.mu.Unlock()
c.Config = config
return nil
}
internal/oplog/oplog.go
+11 -6
View File
@@ -4,6 +4,8 @@ import (
"bytes"
"errors"
"fmt"
"os"
"path"
"sync"
"time"
@@ -26,11 +28,12 @@ var (
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
SnapshotIdBucket = []byte("oplog.snapshot_id") // index by snapshot ID.
)
// OpLog represents a log of operations performed.
// TODO: implement trim support for old operations.
// Operations are indexed by repo and plan.
type OpLog struct {
db *bolt.DB
@@ -38,8 +41,12 @@ type OpLog struct {
subscribers []*func(EventType, *v1.Operation)
}
func NewOpLog(databaseDir string) (*OpLog, error) {
db, err := bolt.Open(databaseDir, 0600, &bolt.Options{Timeout: 1 * time.Second})
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)
}
@@ -257,9 +264,7 @@ func (o *OpLog) readOpsFromIndexBucket(tx *bolt.Tx, bucket []byte, indexId strin
var ops []int64
c := b.Cursor()
var prefix []byte
prefix = append(prefix, itob(int64(len(indexId)))...)
prefix = append(prefix, []byte(indexId)...)
prefix := stob(indexId)
for k, _ := c.Seek(prefix); k != nil && bytes.HasPrefix(k, prefix); k, _ = c.Next() {
ops = append(ops, btoi(k[len(prefix):]))
}
internal/oplog/serializationutil.go
+8 -1
View File
@@ -10,4 +10,11 @@ func itob(v int64) []byte {
func btoi(b []byte) int64 {
return int64(binary.BigEndian.Uint64(b))
}
}
func stob(v string) []byte {
var b []byte
b = append(b, itob(int64(len(v)))...)
b = append(b, []byte(v)...)
return b
}
internal/oplog/serializationutil_test.go
+13
View File
@@ -0,0 +1,13 @@
package oplog
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 btoi(b) != num {
t.Errorf("itob/btoi failed for %d", num)
}
}
}
internal/orchestrator/orchestrator.go
+135 -11
View File
@@ -1,13 +1,17 @@
package orchestrator
import (
"context"
"errors"
"fmt"
"sync"
"time"
v1 "github.com/garethgeorge/resticui/gen/go/v1"
"github.com/garethgeorge/resticui/internal/config"
"github.com/garethgeorge/resticui/internal/oplog"
"github.com/garethgeorge/resticui/pkg/restic"
"go.uber.org/zap"
"google.golang.org/protobuf/proto"
)
@@ -17,18 +21,52 @@ var ErrPlanNotFound = errors.New("plan not found")
// Orchestrator is responsible for managing repos and backups.
type Orchestrator struct {
configProvider config.ConfigStore
mu sync.Mutex
config *v1.Config
oplog *oplog.OpLog
repoPool *resticRepoPool
// configUpdates chan makes config changes available to Run()
configUpdates chan *v1.Config
}
func NewOrchestrator(configProvider config.ConfigStore) *Orchestrator {
return &Orchestrator{
configProvider: configProvider,
repoPool: newResticRepoPool(configProvider),
func NewOrchestrator(configProvider config.ConfigStore, oplog *oplog.OpLog) (*Orchestrator, error) {
cfg, err := configProvider.Get()
if err != nil {
return nil, fmt.Errorf("failed to get config: %w", err)
}
return &Orchestrator{
config: cfg,
oplog: oplog,
repoPool: newResticRepoPool(&config.MemoryStore{Config: cfg}),
}, nil
}
func (o *Orchestrator) ApplyConfig(cfg *v1.Config) error {
o.mu.Lock()
defer o.mu.Unlock()
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)
}
if o.configUpdates != nil {
// orchestrator loop is running, notify it of the config change.
o.configUpdates <- cfg
}
return nil
}
func (o *Orchestrator) GetRepo(repoId string) (repo *RepoOrchestrator, err error) {
o.mu.Lock()
defer o.mu.Unlock()
r, err := o.repoPool.GetRepo(repoId)
if err != nil {
return nil, fmt.Errorf("failed to get repo %q: %w", repoId, err)
@@ -37,16 +75,14 @@ func (o *Orchestrator) GetRepo(repoId string) (repo *RepoOrchestrator, err error
}
func (o *Orchestrator) GetPlan(planId string) (*v1.Plan, error) {
cfg, err := o.configProvider.Get()
if err != nil {
return nil, fmt.Errorf("failed to get config: %w", err)
}
o.mu.Lock()
defer o.mu.Unlock()
if cfg.Plans == nil {
if o.config.Plans == nil {
return nil, ErrPlanNotFound
}
for _, p := range cfg.Plans {
for _, p := range o.config.Plans {
if p.Id == planId {
return p, nil
}
@@ -55,6 +91,93 @@ func (o *Orchestrator) GetPlan(planId string) (*v1.Plan, error) {
return nil, ErrPlanNotFound
}
// Run is the main orchestration loop. Cancel the context to stop the loop.
func (o *Orchestrator) Run(mainCtx context.Context) error {
zap.L().Info("Starting orchestrator loop")
o.mu.Lock()
o.configUpdates = make(chan *v1.Config)
o.mu.Unlock()
for {
lock := sync.Mutex{}
ctx, cancel := context.WithCancel(mainCtx)
var wg sync.WaitGroup
var execTask func(t task, runAt time.Time)
execTask = func(t task, runAt time.Time) {
curTime := time.Now()
timer := time.NewTimer(runAt.Sub(curTime))
zap.L().Debug("Scheduling task", zap.String("task", t.Name()), zap.String("runAt", runAt.Format(time.RFC3339)))
wg.Add(1)
go func() {
defer wg.Done()
select {
case <-ctx.Done():
timer.Stop()
zap.L().Debug("Not running task, orchestrator context is cancelled.", zap.String("task", t.Name()))
return
case <-timer.C:
lock.Lock()
defer lock.Unlock()
zap.L().Debug("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))
} else {
zap.L().Debug("Task finished", zap.String("task", t.Name()))
}
if ctx.Err() != nil {
zap.L().Debug("Not attempting to reschedule task, orchestrator context is cancelled.", zap.String("task", t.Name()))
return
}
next := t.Next(time.Now())
if next != nil {
execTask(t, *next)
} else {
zap.L().Debug("Task has no next run, not rescheduling.", zap.String("task", t.Name()))
}
}
}()
}
// Schedule all backup tasks.
for _, plan := range o.config.Plans {
t, err := newBackupTask(o, plan)
if err != nil {
zap.L().Error("Failed to create backup task for plan", zap.String("plan", plan.Id), zap.Error(err))
}
next := t.Next(time.Now())
if next != nil {
execTask(t, *next)
} else {
zap.L().Debug("Task has no next run, not scheduling.", zap.String("task", t.Name()))
}
}
// wait for either an error or the context to be cancelled, then wait for all tasks.
select {
case <-ctx.Done():
cancel()
wg.Wait()
return nil
case <-o.configUpdates:
zap.L().Info("Orchestrator received config change, waiting for in-progress operations")
cancel()
wg.Wait()
zap.L().Info("Restarting orchestrator loop")
continue
}
}
}
// resticRepoPool caches restic repos.
type resticRepoPool struct {
mu sync.Mutex
@@ -102,6 +225,7 @@ func (rp *resticRepoPool) GetRepo(repoId string) (repo *RepoOrchestrator, err er
delete(rp.repos, repoId);
var opts []restic.GenericOption
opts = append(opts, restic.WithPropagatedEnvVars(restic.EnvToPropagate...))
if len(repoProto.GetEnv()) > 0 {
opts = append(opts, restic.WithEnv(repoProto.GetEnv()...))
}
internal/orchestrator/repo_test.go
+2 -2
View File
@@ -61,12 +61,12 @@ func TestSnapshotParenting(t *testing.T) {
}
plans := []*v1.Plan{
&v1.Plan{
{
Id: "test",
Repo: "test",
Paths: []string{testData},
},
&v1.Plan{
{
Id: "test2",
Repo: "test",
Paths: []string{testData},
internal/orchestrator/tasks.go
+109
View File
@@ -0,0 +1,109 @@
package orchestrator
import (
"context"
"fmt"
"time"
v1 "github.com/garethgeorge/resticui/gen/go/v1"
"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.
Next(now time.Time) *time.Time // when this task would like to be run.
Run(ctx context.Context) error // run the task.
}
type backupTask struct {
orchestrator *Orchestrator // owning orchestrator
plan *v1.Plan
schedule *cronexpr.Schedule
}
var _ task = &backupTask{}
func newBackupTask(orchestrator *Orchestrator, plan *v1.Plan) (*backupTask, error) {
sched, err := cronexpr.Parse(plan.Cron)
if err != nil {
return nil, fmt.Errorf("failed to parse schedule %q: %w", plan.Cron, err)
}
return &backupTask{
orchestrator: orchestrator,
plan: plan,
schedule: sched,
}, nil
}
func (t *backupTask) Name() string {
return fmt.Sprintf("execute backup plan %v", t.plan.Id)
}
func (t *backupTask) Next(now time.Time) *time.Time {
next := t.schedule.Next(now)
return &next
}
func (t *backupTask) Run(ctx context.Context) error {
backupOp := &v1.Operation_OperationBackup{
OperationBackup: &v1.OperationBackup{},
}
op := &v1.Operation{
PlanId: t.plan.Id,
RepoId: t.plan.Repo,
UnixTimeStartMs: time.Now().Unix(),
Status: v1.OperationStatus_STATUS_INPROGRESS,
Op: backupOp,
}
return WithOperation(t.orchestrator.oplog, op, func() error {
zap.L().Info("Starting backup", zap.String("plan", t.plan.Id))
repo, err := t.orchestrator.GetRepo(t.plan.Repo)
if err != nil {
return fmt.Errorf("failed to get repo %q: %w", t.plan.Repo, err)
}
if _, err := repo.Backup(ctx, t.plan, func(entry *restic.BackupProgressEntry) {
backupOp.OperationBackup.LastStatus = entry.ToProto()
if err := t.orchestrator.oplog.Update(op); err != nil {
zap.S().Errorf("failed to update oplog with progress for backup: %v", err)
}
zap.L().Info("Backup progress", zap.Float64("progress", entry.PercentDone))
}); err != nil {
return fmt.Errorf("failed to backup repo %q: %w", t.plan.Repo, err)
}
return nil
})
}
// WithOperation is a utility that creates an operation to track the function's execution.
// timestamps are automatically added and the status is automatically updated if an error occurs.
func WithOperation(oplog *oplog.OpLog, op *v1.Operation, do func() error) error {
if err := oplog.Add(op); err != nil {
return fmt.Errorf("failed to add operation to oplog: %w", err)
}
if op.Status == v1.OperationStatus_STATUS_UNKNOWN {
op.Status = v1.OperationStatus_STATUS_INPROGRESS
}
err := do()
if err != nil {
op.Status = v1.OperationStatus_STATUS_ERROR
op.DisplayMessage = err.Error()
}
op.UnixTimeEndMs = time.Now().Unix()
if op.Status == v1.OperationStatus_STATUS_INPROGRESS {
op.Status = v1.OperationStatus_STATUS_SUCCESS
}
if e := oplog.Update(op); err != nil {
return multierror.Append(err, fmt.Errorf("failed to update operation in oplog: %w", e))
}
return err
}
pkg/restic/outputs.go
+41
View File
@@ -5,6 +5,7 @@ import (
"encoding/json"
"fmt"
"io"
"log"
"os/exec"
"slices"
"time"
@@ -80,6 +81,46 @@ type BackupProgressEntry struct {
BytesDone int `json:"bytes_done"`
}
func (b *BackupProgressEntry) ToProto() *v1.BackupProgressEntry {
switch b.MessageType {
case "summary":
return &v1.BackupProgressEntry{
Entry: &v1.BackupProgressEntry_Summary{
Summary: &v1.BackupProgressSummary{
FilesNew: int64(b.FilesNew),
FilesChanged: int64(b.FilesChanged),
FilesUnmodified: int64(b.FilesUnmodified),
DirsNew: int64(b.DirsNew),
DirsChanged: int64(b.DirsChanged),
DirsUnmodified: int64(b.DirsUnmodified),
DataBlobs: int64(b.DataBlobs),
TreeBlobs: int64(b.TreeBlobs),
DataAdded: int64(b.DataAdded),
TotalFilesProcessed: int64(b.TotalFilesProcessed),
TotalBytesProcessed: int64(b.TotalBytesProcessed),
TotalDuration: float64(b.TotalDuration),
SnapshotId: b.SnapshotId,
},
},
}
case "status":
return &v1.BackupProgressEntry{
Entry: &v1.BackupProgressEntry_Status{
Status: &v1.BackupProgressStatusEntry{
PercentDone: b.PercentDone,
TotalFiles: int64(b.TotalFiles),
FilesDone: int64(b.FilesDone),
TotalBytes: int64(b.TotalBytes),
BytesDone: int64(b.BytesDone),
},
},
}
default:
log.Fatalf("unknown message type: %s", b.MessageType)
return nil
}
}
// readBackupProgressEntrys returns the summary event or an error if the command failed.
func readBackupProgressEntries(cmd *exec.Cmd, output io.Reader, callback func(event *BackupProgressEntry)) (*BackupProgressEntry, error) {
scanner := bufio.NewScanner(output)
proto/v1/operations.proto
+1 -2
View File
@@ -16,7 +16,7 @@ message Operation {
string display_message = 7; // human readable context message (if any)
oneof op {
OperationBackup backup = 100;
OperationBackup operation_backup = 100;
}
}
@@ -44,4 +44,3 @@ enum OperationStatus {
message OperationBackup {
BackupProgressEntry last_status = 3;
}
proto/v1/restic.proto
+1 -1
View File
@@ -51,6 +51,6 @@ message BackupProgressSummary {
int64 data_added = 9;
int64 total_files_processed = 10;
int64 total_bytes_processed = 11;
int64 total_duration = 12;
double total_duration = 12;
string snapshot_id = 13;
}
webui/gen/ts/v1/operations.pb.ts
+1 -1
View File
@@ -41,7 +41,7 @@ type BaseOperation = {
}
export type Operation = BaseOperation
& OneOf<{ backup: OperationBackup }>
& OneOf<{ operationBackup: OperationBackup }>
export type OperationEvent = {
type?: OperationEventType
webui/gen/ts/v1/restic.pb.ts
+1 -1
View File
@@ -54,6 +54,6 @@ export type BackupProgressSummary = {
dataAdded?: string
totalFilesProcessed?: string
totalBytesProcessed?: string
totalDuration?: string
totalDuration?: number
snapshotId?: string
}
webui/src/lib/patterns.ts
+1 -1
View File
@@ -1 +1 @@
export const nameRegex = /^[a-zA-Z0-9_]+$/;
export const nameRegex = /^[a-zA-Z0-9_\-]+$/;