arthur and config testing and fixes
This commit is contained in:
parent
8d572a0c3f
commit
69fbcfb13d
|
@ -20,6 +20,7 @@ This folder contains the code for merlin (which does the actual syncing) and art
|
|||
- [ ] wget
|
||||
|
||||
### TODO
|
||||
- [ ] ensure that the proper permissions (file mode, group, user) are used
|
||||
- [ ] detect if an rsync process is stuck (\*\*)
|
||||
- [ ] place each rsync process in a separate cgroup (\*\*\*)
|
||||
|
||||
|
|
|
@ -1,13 +1,13 @@
|
|||
package arthur
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
"text/tabwriter"
|
||||
"time"
|
||||
|
@ -17,68 +17,83 @@ import (
|
|||
"git.csclub.uwaterloo.ca/public/merlin/sync"
|
||||
)
|
||||
|
||||
// get repo by name function in config
|
||||
// Reads and parses the message sent over the accepted connection
|
||||
func GetCommand(conn net.Conn) (command, repoName string) {
|
||||
command = ""
|
||||
repoName = ""
|
||||
|
||||
func GetAndRunCommand(conn net.Conn) (newSync bool) {
|
||||
newSync = false
|
||||
defer conn.Close()
|
||||
|
||||
var buf bytes.Buffer
|
||||
_, err := io.Copy(&buf, conn)
|
||||
buf, err := ioutil.ReadAll(conn)
|
||||
if err != nil {
|
||||
logger.ErrLog(err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
command := buf.String()
|
||||
args := strings.Split(command, ":")
|
||||
respondAndLogErr := func(msg string) {
|
||||
logger.OutLog(msg)
|
||||
conn.Write([]byte(msg))
|
||||
args := strings.Split(string(buf), ":")
|
||||
if len(args) >= 1 {
|
||||
command = args[0]
|
||||
}
|
||||
|
||||
if args[0] == "status" {
|
||||
status := tabwriter.NewWriter(conn, 5, 5, 5, ' ', 0)
|
||||
fmt.Fprintf(status, "Repository\tLast Synced\tNext Expected Sync\n")
|
||||
|
||||
// for time formating see https://pkg.go.dev/time#pkg-constants
|
||||
for name, repo := range config.RepoMap {
|
||||
fmt.Fprintf(status, "%s\t%s\t%s\n",
|
||||
name,
|
||||
time.Unix(repo.State.LastAttemptStartTime, 0).Format(time.RFC1123),
|
||||
time.Unix(repo.State.LastAttemptRunTime+int64(repo.Frequency), 0).Format(time.RFC1123),
|
||||
)
|
||||
}
|
||||
|
||||
status.Flush()
|
||||
} else if args[0] == "sync" {
|
||||
if len(args) != 2 {
|
||||
respondAndLogErr("Could not parse sync command, forced sync fails.")
|
||||
return
|
||||
}
|
||||
|
||||
if repo, inMap := config.RepoMap[args[1]]; inMap {
|
||||
logger.OutLog("Attempting to force sync of " + repo.Name)
|
||||
if sync.SyncIfPossible(repo) {
|
||||
conn.Write([]byte("Forced sync for " + repo.Name))
|
||||
newSync = true
|
||||
} else {
|
||||
respondAndLogErr("Cannot force sync: " + repo.Name + ", already syncing.")
|
||||
}
|
||||
} else {
|
||||
respondAndLogErr(args[1] + " is not tracked so cannot sync")
|
||||
}
|
||||
} else {
|
||||
respondAndLogErr("Received unrecognized command: " + command)
|
||||
if len(args) >= 2 {
|
||||
repoName = args[1]
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func UnixSockListener(connChan chan net.Conn, stopLisChan chan struct{}) {
|
||||
func SendAndLog(conn net.Conn, msg string) {
|
||||
logger.OutLog(msg)
|
||||
conn.Write([]byte(msg))
|
||||
}
|
||||
|
||||
func SendStatus(conn net.Conn) {
|
||||
status := tabwriter.NewWriter(conn, 5, 5, 5, ' ', 0)
|
||||
fmt.Fprintf(status, "Repository\tLast Synced\tNext Expected Sync\tRunning\n")
|
||||
|
||||
keys := make([]string, 0)
|
||||
for name, _ := range config.RepoMap {
|
||||
keys = append(keys, name)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
|
||||
// for other ways to format the time see: https://pkg.go.dev/time#pkg-constants
|
||||
for _, name := range keys {
|
||||
repo := config.RepoMap[name]
|
||||
lastSync := repo.State.LastAttemptStartTime
|
||||
nextSync := lastSync + int64(repo.Frequency)
|
||||
|
||||
fmt.Fprintf(status, "%s\t%s\t%s\t%t\n",
|
||||
name,
|
||||
time.Unix(lastSync, 0).Format(time.RFC1123),
|
||||
time.Unix(nextSync, 0).Format(time.RFC1123),
|
||||
repo.State.IsRunning,
|
||||
)
|
||||
}
|
||||
|
||||
status.Flush()
|
||||
}
|
||||
|
||||
// Attempt to force the sync of the repo
|
||||
func ForceSync(conn net.Conn, repoName string) (newSync bool) {
|
||||
newSync = false
|
||||
|
||||
// TODO: send repoName and every key in RepoMap to lowercase
|
||||
if repo, isInMap := config.RepoMap[repoName]; isInMap {
|
||||
logger.OutLog("Attempting to force sync of " + repoName)
|
||||
if sync.SyncIfPossible(repo) {
|
||||
conn.Write([]byte("Forced sync for " + repoName))
|
||||
newSync = true
|
||||
} else {
|
||||
SendAndLog(conn, "Could not force sync: "+repoName+" is already syncing.")
|
||||
}
|
||||
} else {
|
||||
SendAndLog(conn, repoName+" is not tracked so cannot sync")
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func StartListener(connChan chan net.Conn, stopLisChan chan struct{}) {
|
||||
sockpath := config.Conf.SockPath
|
||||
// must remove cfg.SockPath otherwise get "bind: address already in use"
|
||||
// must remove old cfg.SockPath otherwise get "bind: address already in use"
|
||||
if filepath.Ext(sockpath) != ".sock" {
|
||||
panic(fmt.Errorf("Socket file must end with .sock"))
|
||||
panic(fmt.Errorf("socket file must end with .sock"))
|
||||
} else if _, err := os.Stat(sockpath); err == nil {
|
||||
if err := os.Remove(sockpath); err != nil {
|
||||
panic(err)
|
||||
|
@ -95,10 +110,10 @@ func UnixSockListener(connChan chan net.Conn, stopLisChan chan struct{}) {
|
|||
|
||||
go func() {
|
||||
for {
|
||||
// will exit when ear is closed
|
||||
// Attempting to accept on a closed net.Listener will return a non-temporary error
|
||||
conn, err := ear.Accept()
|
||||
if err != nil {
|
||||
if ne, ok := err.(net.Error); ok && ne.Temporary() {
|
||||
if netErr, isTemp := err.(net.Error); isTemp && netErr.Temporary() {
|
||||
logger.ErrLog("Accepted socket error: " + err.Error())
|
||||
continue
|
||||
}
|
||||
|
@ -109,6 +124,7 @@ func UnixSockListener(connChan chan net.Conn, stopLisChan chan struct{}) {
|
|||
}
|
||||
}()
|
||||
|
||||
// TODO: check handling of multiple SIGHUP
|
||||
<-stopLisChan
|
||||
ear.Close()
|
||||
}
|
||||
|
|
|
@ -1 +1,283 @@
|
|||
package arthur
|
||||
|
||||
import (
|
||||
"io/ioutil"
|
||||
"net"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.csclub.uwaterloo.ca/public/merlin/config"
|
||||
"git.csclub.uwaterloo.ca/public/merlin/logger"
|
||||
)
|
||||
|
||||
func TestStatusCommand(t *testing.T) {
|
||||
r, w := net.Pipe()
|
||||
|
||||
go func() {
|
||||
// will only finish write when EOF is sent
|
||||
// only way to send EOF is to close
|
||||
w.Write([]byte("status"))
|
||||
w.Close()
|
||||
}()
|
||||
command, repoName := GetCommand(r)
|
||||
if command != "status" {
|
||||
t.Errorf("command for status should be \"status\", got " + command)
|
||||
} else if repoName != "" {
|
||||
t.Errorf("status should return an empty string for the repoName, got " + repoName)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSyncCommand(t *testing.T) {
|
||||
r, w := net.Pipe()
|
||||
|
||||
go func() {
|
||||
w.Write([]byte("sync:ubuntu"))
|
||||
w.Close()
|
||||
}()
|
||||
command, repoName := GetCommand(r)
|
||||
r.Close()
|
||||
if command != "sync" {
|
||||
t.Errorf("command for sync:ubuntu should be \"sync\", got " + command)
|
||||
} else if repoName != "ubuntu" {
|
||||
t.Errorf("name of repo for sync:ubuntu should be \"ubuntu\", got " + repoName)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSendStatus(t *testing.T) {
|
||||
saveRepoMap := config.RepoMap
|
||||
defer func() {
|
||||
config.RepoMap = saveRepoMap
|
||||
}()
|
||||
|
||||
repoMap := make(map[string]*config.Repo)
|
||||
repoMap["eeeee"] = &config.Repo{
|
||||
Frequency: 30 * 86400,
|
||||
State: config.RepoState{
|
||||
IsRunning: true,
|
||||
LastAttemptStartTime: 1600000000,
|
||||
},
|
||||
}
|
||||
repoMap["alinux"] = &config.Repo{
|
||||
Frequency: 7*86400 + 3,
|
||||
State: config.RepoState{
|
||||
IsRunning: true,
|
||||
LastAttemptStartTime: 1620000000,
|
||||
},
|
||||
}
|
||||
repoMap["lnux"] = &config.Repo{
|
||||
Frequency: 86400,
|
||||
State: config.RepoState{
|
||||
IsRunning: false,
|
||||
LastAttemptStartTime: 1640000000,
|
||||
},
|
||||
}
|
||||
config.RepoMap = repoMap
|
||||
|
||||
r, w := net.Pipe()
|
||||
go func() {
|
||||
SendStatus(w)
|
||||
w.Close()
|
||||
}()
|
||||
msg, err := ioutil.ReadAll(r)
|
||||
r.Close()
|
||||
if err != nil {
|
||||
t.Errorf(err.Error())
|
||||
}
|
||||
expected := `Repository Last Synced Next Expected Sync Running
|
||||
alinux Sun, 02 May 2021 20:00:00 EDT Sun, 09 May 2021 20:00:03 EDT true
|
||||
eeeee Sun, 13 Sep 2020 08:26:40 EDT Tue, 13 Oct 2020 08:26:40 EDT true
|
||||
lnux Mon, 20 Dec 2021 06:33:20 EST Tue, 21 Dec 2021 06:33:20 EST false
|
||||
`
|
||||
if expected != string(msg) {
|
||||
t.Errorf("Expected:\n" + expected + "\nGot:\n" + string(msg))
|
||||
}
|
||||
}
|
||||
|
||||
func TestForceSync(t *testing.T) {
|
||||
saveRepos := config.Repos
|
||||
saveRepoMap := config.RepoMap
|
||||
doneChan := make(chan config.SyncResult)
|
||||
defer func() {
|
||||
config.Repos = saveRepos
|
||||
config.RepoMap = saveRepoMap
|
||||
close(doneChan)
|
||||
}()
|
||||
|
||||
// Part 1: run a dummy sync
|
||||
repo := config.Repo{
|
||||
Name: "nux",
|
||||
SyncType: "csc-sync-dummy",
|
||||
Frequency: 7 * 86400,
|
||||
MaxTime: 30,
|
||||
Logger: logger.NewLogger("nux", "/tmp/merlin_force_sync_test_logs"),
|
||||
StateFile: "/tmp/merlin_force_sync_test_state",
|
||||
DoneChan: doneChan,
|
||||
State: config.RepoState{
|
||||
IsRunning: false,
|
||||
LastAttemptStartTime: 0,
|
||||
LastAttemptRunTime: 0,
|
||||
LastAttemptExit: config.NOT_RUN_YET,
|
||||
},
|
||||
}
|
||||
config.Repos = nil
|
||||
config.Repos = append(config.Repos, &repo)
|
||||
config.RepoMap = make(map[string]*config.Repo)
|
||||
config.RepoMap["nux"] = &repo
|
||||
|
||||
r, w := net.Pipe()
|
||||
go func() {
|
||||
if !ForceSync(w, "nux") {
|
||||
t.Errorf("Sync for nux did not start")
|
||||
}
|
||||
w.Close()
|
||||
}()
|
||||
msg, err := ioutil.ReadAll(r)
|
||||
r.Close()
|
||||
if err != nil {
|
||||
t.Errorf(err.Error())
|
||||
}
|
||||
expected := "Forced sync for nux"
|
||||
if expected != string(msg) {
|
||||
t.Errorf("Expected:\n" + expected + "\nGot:\n" + string(msg))
|
||||
}
|
||||
|
||||
select {
|
||||
case result := <-doneChan:
|
||||
if result.Exit != config.SUCCESS {
|
||||
t.Errorf("Sync should exit with SUCCESS, got %d", result.Exit)
|
||||
}
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Errorf("Dummy sync should be done in 1 second, waited 3 seconds")
|
||||
}
|
||||
|
||||
// Part 2: attempt the same thing but with repo.State.IsRunning = true
|
||||
r, w = net.Pipe()
|
||||
go func() {
|
||||
if ForceSync(w, "nux") {
|
||||
t.Errorf("Sync for nux should not have started")
|
||||
}
|
||||
w.Close()
|
||||
}()
|
||||
msg, err = ioutil.ReadAll(r)
|
||||
r.Close()
|
||||
if err != nil {
|
||||
t.Errorf(err.Error())
|
||||
}
|
||||
expected = "Could not force sync: nux is already syncing."
|
||||
if expected != string(msg) {
|
||||
t.Errorf("Expected:\n" + expected + "\nGot:\n" + string(msg))
|
||||
}
|
||||
|
||||
select {
|
||||
case <-doneChan:
|
||||
t.Errorf("Sync for nux should not have been started")
|
||||
case <-time.After(2 * time.Second):
|
||||
}
|
||||
|
||||
// Part 3: attempt a force sync with a repo that does not exist
|
||||
r, w = net.Pipe()
|
||||
go func() {
|
||||
if ForceSync(w, "nixx") {
|
||||
t.Errorf("Sync for nixx should not have started")
|
||||
}
|
||||
w.Close()
|
||||
}()
|
||||
msg, err = ioutil.ReadAll(r)
|
||||
r.Close()
|
||||
if err != nil {
|
||||
t.Errorf(err.Error())
|
||||
}
|
||||
expected = "nixx is not tracked so cannot sync"
|
||||
if expected != string(msg) {
|
||||
t.Errorf("Expected:\n" + expected + "\nGot:\n" + string(msg))
|
||||
}
|
||||
}
|
||||
|
||||
func TestStartListener(t *testing.T) {
|
||||
saveConf := config.Conf
|
||||
connChan := make(chan net.Conn)
|
||||
stopLisChan := make(chan struct{})
|
||||
wait := make(chan struct{})
|
||||
defer func() {
|
||||
config.Conf = saveConf
|
||||
close(connChan)
|
||||
close(stopLisChan)
|
||||
}()
|
||||
config.Conf = config.Config{
|
||||
SockPath: "/tmp/merlin_listener_test.sock",
|
||||
}
|
||||
|
||||
// Test 1: check that closing/sending something to stopLisChan will stop the listener
|
||||
// and that a new listener can be created after stopping the old one
|
||||
go func() {
|
||||
StartListener(connChan, stopLisChan)
|
||||
wait <- struct{}{}
|
||||
}()
|
||||
stopLisChan <- struct{}{}
|
||||
select {
|
||||
case <-wait:
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Errorf("StartListener should stop when struct{}{} is sent to stopLisChan")
|
||||
}
|
||||
|
||||
go func() {
|
||||
StartListener(connChan, stopLisChan)
|
||||
wait <- struct{}{}
|
||||
}()
|
||||
close(stopLisChan)
|
||||
select {
|
||||
case <-wait:
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Errorf("StartListener should stop when stopLisChan is closed")
|
||||
}
|
||||
close(wait)
|
||||
|
||||
// Test 2: check that connections can be made to the unix socket
|
||||
// this test does not appear to be very stable (I think there is a race condition somewhere)
|
||||
stopLisChan = make(chan struct{})
|
||||
go StartListener(connChan, stopLisChan)
|
||||
waitForMsg := func(expected string) {
|
||||
select {
|
||||
case conn := <-connChan:
|
||||
msg, err := ioutil.ReadAll(conn)
|
||||
if err != nil {
|
||||
t.Errorf(err.Error())
|
||||
} else if expected != string(msg) {
|
||||
t.Errorf("Message expected was " + expected + " got " + string(msg))
|
||||
}
|
||||
conn.Close()
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Errorf("StartListener should stop when struct{}{} is sent to stopLisChan")
|
||||
}
|
||||
}
|
||||
sendMsg := func(msg string) {
|
||||
<-time.After(500 * time.Millisecond)
|
||||
send, err := net.Dial("unix", "/tmp/merlin_listener_test.sock")
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
_, err = send.Write([]byte(msg))
|
||||
if err != nil {
|
||||
t.Errorf(err.Error())
|
||||
}
|
||||
send.Close()
|
||||
}
|
||||
go func() {
|
||||
waitForMsg("status")
|
||||
}()
|
||||
sendMsg("status")
|
||||
|
||||
go func() {
|
||||
waitForMsg("sync:uuunix")
|
||||
}()
|
||||
sendMsg("sync:uuunix")
|
||||
|
||||
go func() {
|
||||
waitForMsg("$UPz2L2yWsE^UY8iG9JX@^dBb@5yb*")
|
||||
}()
|
||||
sendMsg("$UPz2L2yWsE^UY8iG9JX@^dBb@5yb*")
|
||||
|
||||
// unsure why I can't put this in the defer
|
||||
os.Remove("/tmp/merlin_listener_test.sock")
|
||||
}
|
||||
|
|
|
@ -2,6 +2,7 @@ package config
|
|||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
||||
"gopkg.in/ini.v1"
|
||||
|
||||
|
@ -18,17 +19,18 @@ const (
|
|||
TEN_MINUTELY = 600
|
||||
FIVE_MINUTELY = 300
|
||||
|
||||
CONFIG_PATH = "merlin-config.ini"
|
||||
|
||||
DEFAULT_MAX_JOBS = 6
|
||||
DEFAULT_MAX_TIME = DAILY / 4
|
||||
DEFAULT_DOWNLOAD_DIR = "/mirror/root"
|
||||
DEFAULT_PASSWORD_DIR = "/home/mirror/passwords"
|
||||
DEFAULT_STATE_DIR = "/home/mirror/merlin/states"
|
||||
DEFAULT_LOG_DIR = "/home/mirror/merlin/logs"
|
||||
DEFAULT_RSYNC_LOG_DIR = "/home/mirror/merlin/logs-rsync"
|
||||
DEFAULT_ZFSSYNC_LOG_DIR = "/home/mirror/merlin/logs-zfssync"
|
||||
DEFAULT_SOCK_PATH = "/run/merlin.sock"
|
||||
// could change this into a default_config
|
||||
DEFAULT_MAX_JOBS = 6
|
||||
DEFAULT_MAX_TIME = DAILY / 4
|
||||
DEFAULT_SYNC_TYPE = "csc-sync-standard"
|
||||
DEFAULT_FREQUENCY_STRING = "by-hourly"
|
||||
DEFAULT_DOWNLOAD_DIR = "/mirror/root"
|
||||
DEFAULT_PASSWORD_DIR = "/home/mirror/passwords"
|
||||
DEFAULT_STATE_DIR = "/home/mirror/merlin/states"
|
||||
DEFAULT_LOG_DIR = "/home/mirror/merlin/logs"
|
||||
DEFAULT_RSYNC_LOG_DIR = "/home/mirror/merlin/logs-rsync"
|
||||
DEFAULT_ZFSSYNC_LOG_DIR = "/home/mirror/merlin/logs-zfssync"
|
||||
DEFAULT_SOCK_PATH = "/run/merlin.sock"
|
||||
)
|
||||
|
||||
var frequencies = map[string]int{
|
||||
|
@ -62,11 +64,11 @@ type Config struct {
|
|||
IPv4Address string `ini:"ipv4_address"`
|
||||
IPv6Address string `ini:"ipv6_address"`
|
||||
// the default sync type
|
||||
SyncType string `ini:"default_sync_type"`
|
||||
// the default frequency string for the repos
|
||||
FrequencyStr string `ini:"default_frequency"`
|
||||
// the default MaxTime for each repo
|
||||
MaxTime int `ini:"default_max_time"`
|
||||
DefaultSyncType string `ini:"default_sync_type"`
|
||||
// the default frequency string
|
||||
DefaultFrequencyStr string `ini:"default_frequency"`
|
||||
// the default MaxTime
|
||||
DefaultMaxTime int `ini:"default_max_time"`
|
||||
// the directory where rsync should download files
|
||||
DownloadDir string `ini:"download_dir"`
|
||||
// the directory where rsync passwords are stored
|
||||
|
@ -74,15 +76,16 @@ type Config struct {
|
|||
// the directory where the state of each repo sync is saved
|
||||
StateDir string `ini:"states_dir"`
|
||||
// the directory where merlin will store the general logs for each repo
|
||||
LoggerDir string `ini:"log_dir"`
|
||||
RepoLogDir string `ini:"repo_logs_dir"`
|
||||
// the directory to store the rsync logs for each repo
|
||||
RsyncLogDir string `ini:"rsync_log_dir"`
|
||||
RsyncLogDir string `ini:"rsync_logs_dir"`
|
||||
// the directory to store the zfssync logs for each repo
|
||||
ZfssyncLogDir string `ini:"zfssync_log_dir"`
|
||||
ZfssyncLogDir string `ini:"zfssync_logs_dir"`
|
||||
// the Unix socket path which arthur will use to communicate with us
|
||||
SockPath string `ini:"sock_path"`
|
||||
}
|
||||
|
||||
// make it more clear when full path should be used vs when just the file name is needed
|
||||
type Repo struct {
|
||||
// the name of this repo
|
||||
Name string `ini:"-"`
|
||||
|
@ -106,18 +109,20 @@ type Repo struct {
|
|||
RsyncUser string `ini:"rsync_user"`
|
||||
// the file storing the password for rsync (optional)
|
||||
PasswordFile string `ini:"password_file"`
|
||||
// the file for general logging of this repo
|
||||
LoggerFile string `ini:"log_file"`
|
||||
// the file storing the repo sync state (used to override default)
|
||||
StateFile string `ini:"state_file"`
|
||||
// the full file path for general logging of this repo (used to override default)
|
||||
RepoLogFile string `ini:"repo_log_file"`
|
||||
// a reference to the general logger
|
||||
Logger *logger.Logger `ini:"-"`
|
||||
// the file for logging this repo's rsync
|
||||
// the full file path for logging this repo's rsync (used to override default)
|
||||
RsyncLogFile string `ini:"rsync_log_file"`
|
||||
// the file for logging this repo's zfssync
|
||||
// the full file path for logging this repo's zfssync (used to override default)
|
||||
ZfssyncLogFile string `ini:"zfssync_log_file"`
|
||||
// the repo will write its name and status in a Result struct to DoneChan
|
||||
// when it has finished a job (shared by all repos)
|
||||
DoneChan chan<- SyncResult `ini:"-"`
|
||||
// the repo should stop syncing if StopChan is closed (shared by all repos)
|
||||
// repos should stop syncing if StopChan is closed (shared by all repos)
|
||||
StopChan chan struct{} `ini:"-"`
|
||||
// a struct that stores the repo's status
|
||||
State RepoState `ini:"-"`
|
||||
|
@ -145,32 +150,34 @@ var (
|
|||
// GetConfig reads the config from a JSON file, initializes default values,
|
||||
// and initializes the non-configurable fields of each repo.
|
||||
// It returns a Config.
|
||||
func LoadConfig(doneChan chan SyncResult, stopChan chan struct{}) {
|
||||
func LoadConfig(configPath string, doneChan chan SyncResult, stopChan chan struct{}) {
|
||||
// set default values then load config from file
|
||||
newConf := Config{
|
||||
MaxJobs: DEFAULT_MAX_JOBS,
|
||||
MaxTime: DEFAULT_MAX_TIME,
|
||||
PasswordDir: DEFAULT_PASSWORD_DIR,
|
||||
DownloadDir: DEFAULT_DOWNLOAD_DIR,
|
||||
StateDir: DEFAULT_STATE_DIR,
|
||||
LoggerDir: DEFAULT_LOG_DIR,
|
||||
RsyncLogDir: DEFAULT_RSYNC_LOG_DIR,
|
||||
ZfssyncLogDir: DEFAULT_ZFSSYNC_LOG_DIR,
|
||||
SockPath: DEFAULT_SOCK_PATH,
|
||||
MaxJobs: DEFAULT_MAX_JOBS,
|
||||
DefaultSyncType: DEFAULT_SYNC_TYPE,
|
||||
DefaultFrequencyStr: DEFAULT_FREQUENCY_STRING,
|
||||
DefaultMaxTime: DEFAULT_MAX_TIME,
|
||||
PasswordDir: DEFAULT_PASSWORD_DIR,
|
||||
DownloadDir: DEFAULT_DOWNLOAD_DIR,
|
||||
StateDir: DEFAULT_STATE_DIR,
|
||||
RepoLogDir: DEFAULT_LOG_DIR,
|
||||
RsyncLogDir: DEFAULT_RSYNC_LOG_DIR,
|
||||
ZfssyncLogDir: DEFAULT_ZFSSYNC_LOG_DIR,
|
||||
SockPath: DEFAULT_SOCK_PATH,
|
||||
}
|
||||
iniInfo, err := ini.Load(CONFIG_PATH)
|
||||
iniInfo, err := ini.Load(configPath)
|
||||
panicIfErr(err)
|
||||
err = iniInfo.MapTo(&newConf)
|
||||
panicIfErr(err)
|
||||
|
||||
// check config for major errors
|
||||
for _, dir := range []string{Conf.StateDir, Conf.LoggerDir, Conf.RsyncLogDir, Conf.ZfssyncLogDir} {
|
||||
for _, dir := range []string{newConf.StateDir, newConf.RepoLogDir, newConf.RsyncLogDir, newConf.ZfssyncLogDir} {
|
||||
err := os.MkdirAll(dir, 0755)
|
||||
panicIfErr(err)
|
||||
}
|
||||
if Conf.IPv4Address == "" {
|
||||
if newConf.IPv4Address == "" {
|
||||
panic("Missing IPv4 address from config")
|
||||
} else if Conf.IPv6Address == "" {
|
||||
} else if newConf.IPv6Address == "" {
|
||||
panic("Missing IPv6 address from config")
|
||||
}
|
||||
|
||||
|
@ -182,26 +189,32 @@ func LoadConfig(doneChan chan SyncResult, stopChan chan struct{}) {
|
|||
}
|
||||
|
||||
// set the default values for the repo then load from file
|
||||
// TODO: check if local_dir and repoName are always the same value
|
||||
// TODO: check to ensure that every Repo.Name is unique (may already be done by ini)
|
||||
repo := Repo{
|
||||
Name: repoName,
|
||||
SyncType: Conf.SyncType,
|
||||
FrequencyStr: Conf.FrequencyStr,
|
||||
MaxTime: Conf.MaxTime,
|
||||
LoggerFile: Conf.LoggerDir + "/" + repoName + ".log",
|
||||
RsyncLogFile: Conf.RsyncLogDir + "/" + repoName + "-rsync.log",
|
||||
ZfssyncLogFile: Conf.ZfssyncLogDir + "/" + repoName + "-zfssync.log",
|
||||
SyncType: newConf.DefaultSyncType,
|
||||
FrequencyStr: newConf.DefaultFrequencyStr,
|
||||
MaxTime: newConf.DefaultMaxTime,
|
||||
StateFile: filepath.Join(newConf.StateDir, repoName),
|
||||
RepoLogFile: filepath.Join(newConf.RepoLogDir, repoName) + ".log",
|
||||
RsyncLogFile: filepath.Join(newConf.RsyncLogDir, repoName) + "-rsync.log",
|
||||
ZfssyncLogFile: filepath.Join(newConf.ZfssyncLogDir, repoName) + "-zfssync.log",
|
||||
DoneChan: doneChan,
|
||||
StopChan: stopChan,
|
||||
}
|
||||
err := section.MapTo(&repo)
|
||||
panicIfErr(err)
|
||||
|
||||
// TODO: ensure that the parent dirs to the file also exist when touching
|
||||
// or just remove the ability to override
|
||||
touchFiles(
|
||||
repo.LoggerFile,
|
||||
repo.StateFile,
|
||||
repo.RepoLogFile,
|
||||
repo.RsyncLogFile,
|
||||
repo.ZfssyncLogFile,
|
||||
)
|
||||
repo.Logger = logger.NewLogger(repo.Name, repo.LoggerFile)
|
||||
repo.Logger = logger.NewLogger(repo.Name, repo.RepoLogFile)
|
||||
repo.Frequency = frequencies[repo.FrequencyStr]
|
||||
if repo.SyncType == "" {
|
||||
panic("Missing sync type from " + repo.Name)
|
||||
|
@ -215,16 +228,9 @@ func LoadConfig(doneChan chan SyncResult, stopChan chan struct{}) {
|
|||
LastAttemptRunTime: 0,
|
||||
LastAttemptExit: NOT_RUN_YET,
|
||||
}
|
||||
|
||||
// create the state file if it does not exist, otherwise load it from existing file
|
||||
repoStateFile := Conf.StateDir + "/" + repo.Name
|
||||
if _, err := os.Stat(repoStateFile); err != nil {
|
||||
touchFile(repoStateFile)
|
||||
repo.SaveState()
|
||||
} else {
|
||||
err := ini.MapTo(&repo.State, repoStateFile)
|
||||
panicIfErr(err)
|
||||
}
|
||||
err = ini.MapTo(&repo.State, repo.StateFile)
|
||||
panicIfErr(err)
|
||||
repo.SaveState()
|
||||
|
||||
newRepos = append(newRepos, &repo)
|
||||
}
|
||||
|
@ -235,6 +241,7 @@ func LoadConfig(doneChan chan SyncResult, stopChan chan struct{}) {
|
|||
|
||||
Conf = newConf
|
||||
Repos = newRepos
|
||||
RepoMap = make(map[string]*Repo)
|
||||
for _, repo := range Repos {
|
||||
RepoMap[repo.Name] = repo
|
||||
}
|
||||
|
@ -242,17 +249,17 @@ func LoadConfig(doneChan chan SyncResult, stopChan chan struct{}) {
|
|||
|
||||
// save the current state of the repo to a file
|
||||
func (repo *Repo) SaveState() {
|
||||
repo.Logger.Debug("Saving state")
|
||||
// repo.Logger.Debug("Saving state")
|
||||
state_cfg := ini.Empty()
|
||||
if err := ini.ReflectFrom(state_cfg, &repo.State); err != nil {
|
||||
repo.Logger.Error(err.Error())
|
||||
}
|
||||
file, err := os.OpenFile(Conf.StateDir+"/"+repo.Name, os.O_RDWR|os.O_CREATE, 0644)
|
||||
file, err := os.OpenFile(repo.StateFile, os.O_RDWR|os.O_CREATE, 0644)
|
||||
if err != nil {
|
||||
repo.Logger.Error(err.Error())
|
||||
}
|
||||
if _, err := state_cfg.WriteTo(file); err != nil {
|
||||
repo.Logger.Error(err.Error())
|
||||
}
|
||||
repo.Logger.Debug("Saved state")
|
||||
// repo.Logger.Debug("Saved state")
|
||||
}
|
||||
|
|
|
@ -1 +1,122 @@
|
|||
package config
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"os"
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"github.com/davecgh/go-spew/spew"
|
||||
)
|
||||
|
||||
func TestTouchFiles(t *testing.T) {
|
||||
files := []string{
|
||||
"/tmp/merlin_touch_test_1",
|
||||
"/tmp/merlin_touch_test_2",
|
||||
"/tmp/merlin_touch_test_3",
|
||||
}
|
||||
touchFiles(files[0], files[1], files[2])
|
||||
for _, file := range files {
|
||||
if _, err := os.Stat(file); err != nil {
|
||||
t.Errorf(err.Error())
|
||||
} else if err := os.Remove(file); err != nil {
|
||||
t.Errorf(err.Error())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestPanicIfErr(t *testing.T) {
|
||||
panicIfErr(nil)
|
||||
defer func() { recover() }()
|
||||
panicIfErr(errors.New("AAAAAAAAAA"))
|
||||
t.Errorf("panicIfErr should have panicked")
|
||||
}
|
||||
|
||||
func TestLoadConfig(t *testing.T) {
|
||||
doneChan := make(chan SyncResult)
|
||||
stopChan := make(chan struct{})
|
||||
LoadConfig("config_test.ini", doneChan, stopChan)
|
||||
// TODO: Fill out parts not part of the ini or state file
|
||||
expectedConfig := Config{
|
||||
MaxJobs: 6,
|
||||
IPv4Address: "129.97.134.129",
|
||||
IPv6Address: "2620:101:f000:4901:c5c::129",
|
||||
DefaultSyncType: "csc-sync-standard",
|
||||
DefaultFrequencyStr: "daily",
|
||||
DefaultMaxTime: 1000,
|
||||
DownloadDir: "/tmp/test-mirror",
|
||||
PasswordDir: "/home/mirror/passwords",
|
||||
StateDir: "test_files",
|
||||
RepoLogDir: "test_files/logs",
|
||||
RsyncLogDir: "test_files/rsync",
|
||||
ZfssyncLogDir: "test_files/zfssync",
|
||||
SockPath: "test_files/test.sock",
|
||||
}
|
||||
expectedRepo1 := Repo{
|
||||
Name: "eelinux",
|
||||
SyncType: "csc-sync-nonstandard",
|
||||
FrequencyStr: "tri-hourly",
|
||||
Frequency: 10800,
|
||||
MaxTime: 2000,
|
||||
LocalDir: "eelinux",
|
||||
RsyncHost: "rsync.releases.eelinux.ca",
|
||||
RsyncDir: "releases",
|
||||
StateFile: "test_files/eeelinux",
|
||||
RepoLogFile: "test_files/logs/eelinux.log",
|
||||
Logger: Repos[0].Logger,
|
||||
RsyncLogFile: "test_files/rsync/eelinux.log",
|
||||
ZfssyncLogFile: "test_files/zfssync/eelinux.log",
|
||||
DoneChan: doneChan,
|
||||
StopChan: stopChan,
|
||||
State: RepoState{
|
||||
IsRunning: false,
|
||||
LastAttemptStartTime: 1600000000,
|
||||
LastAttemptRunTime: 100,
|
||||
LastAttemptExit: 1,
|
||||
},
|
||||
}
|
||||
expectedRepo2 := Repo{
|
||||
Name: "yoland",
|
||||
SyncType: "csc-sync-standard",
|
||||
FrequencyStr: "daily",
|
||||
Frequency: 86400,
|
||||
MaxTime: 1000,
|
||||
LocalDir: "yoland-releases",
|
||||
RsyncHost: "rsync.releases.yoland.io",
|
||||
RsyncDir: "releases",
|
||||
StateFile: "test_files/yoland",
|
||||
RepoLogFile: "test_files/logs/yoland.log",
|
||||
Logger: Repos[1].Logger,
|
||||
RsyncLogFile: "test_files/rsync/yoland-rsync.log",
|
||||
ZfssyncLogFile: "test_files/zfssync/yoland-zfssync.log",
|
||||
DoneChan: doneChan,
|
||||
StopChan: stopChan,
|
||||
State: RepoState{
|
||||
IsRunning: false,
|
||||
LastAttemptStartTime: 0,
|
||||
LastAttemptRunTime: 0,
|
||||
LastAttemptExit: 0,
|
||||
},
|
||||
}
|
||||
|
||||
if !reflect.DeepEqual(expectedConfig, Conf) {
|
||||
t.Errorf("Config loaded does not match expected config")
|
||||
spew.Dump(expectedConfig)
|
||||
spew.Dump(Conf)
|
||||
}
|
||||
if !reflect.DeepEqual(expectedRepo1, *Repos[0]) {
|
||||
t.Errorf("The eelinux repo loaded does not match the exected repo config")
|
||||
spew.Dump(expectedRepo1)
|
||||
spew.Dump(*Repos[0])
|
||||
}
|
||||
if !reflect.DeepEqual(expectedRepo2, *Repos[1]) {
|
||||
t.Errorf("The yoland repo loaded does not match the exected repo config")
|
||||
spew.Dump(expectedRepo2)
|
||||
spew.Dump(*Repos[1])
|
||||
}
|
||||
|
||||
os.Remove("test_files/yoland")
|
||||
os.RemoveAll("test_files/logs")
|
||||
os.RemoveAll("test_files/rsync")
|
||||
os.RemoveAll("test_files/zfssync")
|
||||
}
|
||||
|
|
|
@ -0,0 +1,39 @@
|
|||
; default values are commented out
|
||||
|
||||
; max_jobs = 6
|
||||
ipv4_address = 129.97.134.129
|
||||
ipv6_address = 2620:101:f000:4901:c5c::129
|
||||
; default_sync_type = csc-sync-standard
|
||||
default_frequency = daily
|
||||
default_max_time = 1000
|
||||
download_dir = /tmp/test-mirror
|
||||
; password_dir = /home/mirror/passwords
|
||||
states_dir = test_files
|
||||
repo_logs_dir = test_files/logs
|
||||
rsync_logs_dir = test_files/rsync
|
||||
zfssync_logs_dir = test_files/zfssync
|
||||
sock_path = test_files/test.sock
|
||||
|
||||
[eelinux]
|
||||
sync_type = csc-sync-nonstandard
|
||||
frequency = tri-hourly
|
||||
max_time = 2000
|
||||
local_dir = eelinux
|
||||
rsync_host = rsync.releases.eelinux.ca
|
||||
rsync_dir = releases
|
||||
state_file = test_files/eeelinux
|
||||
repo_log_file = test_files/logs/eelinux.log
|
||||
rsync_log_file = test_files/rsync/eelinux.log
|
||||
zfssync_log_file = test_files/zfssync/eelinux.log
|
||||
|
||||
[yoland]
|
||||
; sync_type = csc-sync-standard
|
||||
; frequency = daily
|
||||
; max_time = 1000
|
||||
local_dir = yoland-releases
|
||||
rsync_host = rsync.releases.yoland.io
|
||||
rsync_dir = releases
|
||||
; state_file = test_files/yoland
|
||||
; repo_log_file = test_files/logs/yoland.log
|
||||
; rsync_log_file = test_files/rsync/yoland-rsync.log
|
||||
; zfssync_log_file = test_files/zfssync/yoland-zfssync.log
|
|
@ -0,0 +1,5 @@
|
|||
is_running = false
|
||||
last_attempt_time = 1600000000
|
||||
last_attempt_runtime = 100
|
||||
last_attempt_exit = 1
|
||||
|
|
@ -21,14 +21,13 @@ func touchFile(file string) {
|
|||
f.Close()
|
||||
} else if fi.IsDir() {
|
||||
panic(fmt.Errorf("%s is a directory", file))
|
||||
} else if os.Geteuid() != 1001 {
|
||||
// UID 1001 is the hardcoded uid for mirror
|
||||
err := os.Chown(file, 1001, os.Getegid())
|
||||
panicIfErr(err)
|
||||
} else if fi.Mode().Perm() != 0644 {
|
||||
err := os.Chmod(file, 0644)
|
||||
panicIfErr(err)
|
||||
// } else if os.Geteuid() != 1001 {
|
||||
// // mirror is UID 1001
|
||||
// err := os.Chown(file, 1001, os.Getegid())
|
||||
// panicIfErr(err)
|
||||
}
|
||||
err = os.Chmod(file, 0644)
|
||||
panicIfErr(err)
|
||||
}
|
||||
|
||||
func touchFiles(files ...string) {
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
module git.csclub.uwaterloo.ca/public/merlin
|
||||
|
||||
require (
|
||||
github.com/davecgh/go-spew v1.1.0
|
||||
github.com/stretchr/testify v1.7.0 // indirect
|
||||
golang.org/x/sys v0.0.0-20210915083310-ed5796bab164
|
||||
gopkg.in/ini.v1 v1.63.2
|
||||
|
|
|
@ -67,7 +67,9 @@ func (logger *Logger) Debug(v ...interface{}) {
|
|||
}
|
||||
|
||||
func (logger *Logger) Info(v ...interface{}) {
|
||||
OutLog(append([]interface{}{"[" + logger.name + "]"}, v...))
|
||||
// src := []interface{}{logger.name + ":"}
|
||||
// args := append(src, v...)
|
||||
OutLog(v...)
|
||||
logger.Log(INFO, v...)
|
||||
}
|
||||
|
||||
|
|
|
@ -16,6 +16,9 @@ import (
|
|||
"git.csclub.uwaterloo.ca/public/merlin/sync"
|
||||
)
|
||||
|
||||
// get config path from command args
|
||||
var CONFIG_PATH = "merlin-config.ini"
|
||||
|
||||
func main() {
|
||||
// check that merlin is run as mirror user
|
||||
// check that mirror user has pid of 1001
|
||||
|
@ -40,11 +43,11 @@ func main() {
|
|||
repoIdx := 0
|
||||
|
||||
loadConfig := func() {
|
||||
config.LoadConfig(doneChan, stopChan)
|
||||
config.LoadConfig(CONFIG_PATH, doneChan, stopChan)
|
||||
logger.OutLog("Loaded config:\n" + fmt.Sprintf("%+v\n", config.Conf))
|
||||
|
||||
repoIdx = 0
|
||||
go arthur.UnixSockListener(connChan, stopLisChan)
|
||||
go arthur.StartListener(connChan, stopLisChan)
|
||||
}
|
||||
// We use a round-robin strategy. It's not the most efficient, but it's simple
|
||||
// (read: easy to understand) and guarantees each repo will eventually get a chance to run.
|
||||
|
@ -66,7 +69,8 @@ func main() {
|
|||
|
||||
loadConfig()
|
||||
// ensure that IsRunning is false otherwise repo will never sync
|
||||
for _, repo := range config.RepoMap {
|
||||
// (only on startup can we assume that repos were not previously syncing)
|
||||
for _, repo := range config.Repos {
|
||||
repo.State.IsRunning = false
|
||||
}
|
||||
runAsManyAsPossible()
|
||||
|
@ -82,18 +86,28 @@ runLoop:
|
|||
case <-reloadSig:
|
||||
stopLisChan <- struct{}{}
|
||||
loadConfig()
|
||||
// ensure that SyncCompleted can handle it if reloading config
|
||||
// removes a repo that was already syncing
|
||||
|
||||
case done := <-doneChan:
|
||||
sync.SyncCompleted(config.RepoMap[done.Name], done.Exit)
|
||||
numJobsRunning--
|
||||
|
||||
case conn := <-connChan:
|
||||
// TODO: may want to split this into GetCommand and something else
|
||||
// to make it more clear tha GetAndRunCommand returns true if
|
||||
// it starts a sync
|
||||
if arthur.GetAndRunCommand(conn) {
|
||||
numJobsRunning--
|
||||
command, repoName := arthur.GetCommand(conn)
|
||||
switch command {
|
||||
case "status":
|
||||
arthur.SendStatus(conn)
|
||||
case "sync":
|
||||
if arthur.ForceSync(conn, repoName) {
|
||||
numJobsRunning++
|
||||
}
|
||||
default:
|
||||
arthur.SendAndLog(conn, "Received unrecognized command: "+command)
|
||||
}
|
||||
// None of the arthur functions close the connection so you will need to
|
||||
// close it manually for the message to be sent
|
||||
conn.Close()
|
||||
|
||||
case <-time.After(1 * time.Minute):
|
||||
}
|
||||
|
|
|
@ -2,8 +2,7 @@ package sync
|
|||
|
||||
import (
|
||||
"fmt"
|
||||
"math/rand"
|
||||
"strconv"
|
||||
"os"
|
||||
|
||||
"git.csclub.uwaterloo.ca/public/merlin/config"
|
||||
)
|
||||
|
@ -170,17 +169,14 @@ func cscSyncStandardIPV6(repo *config.Repo) []string {
|
|||
return args
|
||||
}
|
||||
|
||||
// for testing, to be removed later
|
||||
func cscSyncDummy(repo *config.Repo) []string {
|
||||
|
||||
sleepDur := strconv.FormatInt(rand.Int63n(10)+5, 10)
|
||||
args := []string{"sleep", sleepDur}
|
||||
args := []string{"sleep", "1"}
|
||||
|
||||
return args
|
||||
}
|
||||
|
||||
// executes a particular sync job depending on repo.SyncType.
|
||||
func getSyncCommand(repo *config.Repo) []string {
|
||||
func getSyncCommand(repo *config.Repo) (args []string) {
|
||||
/*
|
||||
# scripts used by merlin.py
|
||||
csc-sync-debian
|
||||
|
@ -207,40 +203,55 @@ func getSyncCommand(repo *config.Repo) []string {
|
|||
ubuntu-releases-sync
|
||||
*/
|
||||
switch repo.SyncType {
|
||||
|
||||
case "csc-sync-apache":
|
||||
return cscSyncApache(repo)
|
||||
args = cscSyncApache(repo)
|
||||
case "csc-sync-archlinux":
|
||||
return cscSyncArchLinux(repo)
|
||||
args = cscSyncArchLinux(repo)
|
||||
case "csc-sync-badperms":
|
||||
return cscSyncBadPerms(repo)
|
||||
args = cscSyncBadPerms(repo)
|
||||
case "csc-sync-cdimage":
|
||||
return cscSyncCDImage(repo)
|
||||
args = cscSyncCDImage(repo)
|
||||
// case "csc-sync-ceph":
|
||||
// return cscSyncCeph(repo)
|
||||
// args = cscSyncCeph(repo)
|
||||
case "csc-sync-chmod":
|
||||
return cscSyncChmod(repo)
|
||||
args = cscSyncChmod(repo)
|
||||
case "csc-sync-debian":
|
||||
return cscSyncDebian(repo)
|
||||
args = cscSyncDebian(repo)
|
||||
case "csc-sync-debian-cd":
|
||||
return cscSyncDebianCD(repo)
|
||||
args = cscSyncDebianCD(repo)
|
||||
case "csc-sync-gentoo":
|
||||
return cscSyncGentoo(repo)
|
||||
args = cscSyncGentoo(repo)
|
||||
// case "csc-sync-s3":
|
||||
// return cscSyncS3(repo)
|
||||
// args = cscSyncS3(repo)
|
||||
case "csc-sync-ssh":
|
||||
return cscSyncSSH(repo)
|
||||
args = cscSyncSSH(repo)
|
||||
case "csc-sync-standard":
|
||||
return cscSyncStandard(repo)
|
||||
args = cscSyncStandard(repo)
|
||||
case "csc-sync-standard-ipv6":
|
||||
return cscSyncStandardIPV6(repo)
|
||||
args = cscSyncStandardIPV6(repo)
|
||||
// case "csc-sync-wget":
|
||||
// return cscSyncWget(repo)
|
||||
// args = cscSyncWget(repo)
|
||||
case "csc-sync-dummy":
|
||||
return cscSyncDummy(repo)
|
||||
default:
|
||||
repo.Logger.Error("Unrecognized sync type", "'"+repo.SyncType+"'")
|
||||
}
|
||||
|
||||
return []string{}
|
||||
localDir := buildDownloadDir(repo)
|
||||
err := os.MkdirAll(localDir, 0775)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("could not create directory %s: %w", localDir, err)
|
||||
// I'm not sure if logger can handle error so just use the string?
|
||||
repo.Logger.Error(err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
if repo.PasswordFile != "" {
|
||||
filename := config.Conf.PasswordDir + "/" + repo.PasswordFile
|
||||
args = append(args, "--password-file", filename)
|
||||
}
|
||||
|
||||
args = append(args, buildRsyncHost(repo), localDir)
|
||||
|
||||
return
|
||||
}
|
||||
|
|
|
@ -52,6 +52,8 @@ func SyncCompleted(repo *config.Repo, exit int) {
|
|||
}
|
||||
repo.SaveState()
|
||||
repo.Logger.Info(fmt.Sprintf("Sync "+exitStr+" after running for %d seconds, will run again in %d seconds", syncTook, nextSync))
|
||||
|
||||
// TODO: make it possible for this SyncCompleted to be run without zfsSync being run
|
||||
if exit == config.SUCCESS {
|
||||
// it is possible that the zfssync from the last repo sync is still running is that fine?
|
||||
go zfsSync(repo)
|
||||
|
|
|
@ -58,6 +58,7 @@ func spawnSyncProcess(repo *config.Repo, args []string) (ch <-chan *exec.Cmd) {
|
|||
case <-cmdDoneChan:
|
||||
if !cmd.ProcessState.Success() {
|
||||
repo.Logger.Warning("Process ended with status code", cmd.ProcessState.ExitCode())
|
||||
// repo.Logger.Info(strings.Join(args, " "))
|
||||
}
|
||||
|
||||
case <-repo.StopChan:
|
||||
|
@ -81,22 +82,7 @@ func startSync(repo *config.Repo) {
|
|||
}
|
||||
}()
|
||||
|
||||
localDir := buildDownloadDir(repo)
|
||||
err := os.MkdirAll(localDir, 0775)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("Could not create directory %s: %w", localDir, err)
|
||||
// I'm not sure if logger can handle error so just use the string?
|
||||
repo.Logger.Error(err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
args := getSyncCommand(repo)
|
||||
if repo.PasswordFile != "" {
|
||||
filename := config.Conf.PasswordDir + "/" + repo.PasswordFile
|
||||
args = append(args, "--password-file", filename)
|
||||
}
|
||||
args = append(args, buildRsyncHost(repo), localDir)
|
||||
|
||||
ch := spawnSyncProcess(repo, args)
|
||||
if ch == nil {
|
||||
// SpawnProcess will have already logged error
|
||||
|
|
Loading…
Reference in New Issue