main.go

/* See the file "LICENSE.txt" for the full license governing this code. */

// Functions according to sub commands given on command line
// Main snapshot creation and purging loops

package main

import (
	"errors"
	"flag"
	"fmt"
	"github.com/daviddengcn/go-colortext"
	"io"
	"io/ioutil"
	"log"
	"os"
	"os/signal"
	"path/filepath"
	"syscall"
	"time"
)

const initialWait = time.Second * 30

var config *Config
var logger *log.Logger

func debugf(format string, args ...interface{}) {
	if os.Getenv("SNAPRD_DEBUG") == "1" {
		logger.Output(2, "<DEBUG> "+fmt.Sprintf(format, args...))
	}
}

// lastGoodTicker is the clock for the create loop. It takes the last
// created snapshot on its input channel and outputs it on the output channel,
// but only after an appropriate waiting time. To start things off, the first
// lastGood snapshot has to be read from disk.
func lastGoodTicker(in, out chan *snapshot, cl clock) {
	var gap, wait time.Duration
	var sn *snapshot
	sn = lastGoodFromDisk(cl)
	if sn != nil {
		debugf("lastgood from disk: %s\n", sn.String())
	}
	// kick off the loop
	go func() {
		in <- sn
		return
	}()
	for {
		sn := <-in
		if sn != nil {
			gap = cl.Now().Sub(sn.startTime)
			debugf("gap: %s", gap)
			wait = schedules[config.Schedule][0] - gap
			if wait > 0 {
				sigc := make(chan os.Signal, 1)
				signal.Notify(sigc, syscall.SIGUSR2)
				log.Println("wait", wait, "before next snapshot")
				select {
				case <-sigc:
					log.Println("Snapshot forced by signal, skipping wait time.")
				case <-time.After(wait):
					debugf("Awoken at %s\n", cl.Now())
				}
			}
		}
		out <- sn
	}
}

// subcmdRun is the main, long-running routine and starts off a couple of
// helper goroutines.
func subcmdRun() (ferr error) {
	pl := newPidLocker(filepath.Join(config.repository, ".pid"))
	err := pl.Lock()
	if err != nil {
		ferr = err
		return
	}
	defer pl.Unlock()
	if !config.NoWait {
		sigc := make(chan os.Signal, 1)
		signal.Notify(sigc, syscall.SIGINT, syscall.SIGTERM)
		log.Printf("waiting %s before making snapshots\n", initialWait)
		select {
		case <-sigc:
			return errors.New("-> Early exit")
		case <-time.After(initialWait):
		}
	}
	createExit := make(chan bool)
	createExitDone := make(chan error)
	// The obsoleteQueue should not be larger than the absolute number of
	// expected snapshots. However, there is no way (yet) to calculate that
	// number.
	obsoleteQueue := make(chan *snapshot, 10000)
	lastGoodIn := make(chan *snapshot)
	lastGoodOut := make(chan *snapshot)
	// Empty type for the channel: we don't care about what is inside, only
	// about the fact that there is something inside
	freeSpaceCheck := make(chan struct{})

	cl := new(realClock)
	go lastGoodTicker(lastGoodIn, lastGoodOut, cl)

	// Snapshot creation loop
	go func() {
		var lastGood *snapshot
		var createError error
	CREATE_LOOP:
		for {
			debugf("start of create loop")
			select {
			case <-createExit:
				debugf("gracefully exiting snapshot creation goroutine")
				lastGoodOut = nil
				break CREATE_LOOP
			case lastGood = <-lastGoodOut:
				sn, err := createSnapshot(lastGood)
				if err != nil || sn == nil {
					debugf("snapshot creation finally failed (%s), the partial transfer will hopefully be reused", err)
					createError = err
					go func() {
						// need to stop the lastGoodTicker here because it could
						// happen that it will be faster and the create loop would
						// run again instead of exiting
						lastGoodOut = nil
						debugf("subcmdRun: sending createExit")
						createExit <- true
						debugf("subcmdRun: createExit sent")
						return
					}()
					time.Sleep(time.Second)
				}
				lastGoodIn <- sn
				debugf("pruning")
				prune(obsoleteQueue, cl)
				// If we purge automatically all the expired snapshots,
				// there's nothing to remove to free space.
				if config.NoPurge {
					debugf("checking space constraints")
					freeSpaceCheck <- struct{}{}
				}
			}
		}
		createExitDone <- createError
	}()
	debugf("started snapshot creation goroutine")

	// Usually the purger gets its input only from prune(). But there could be
	// snapshots left behind from a previously failed snaprd run, so we fill
	// the obsoleteQueue once at the beginning.
	for _, sn := range findDangling(cl) {
		obsoleteQueue <- sn
	}

	// Purger loop
	go func() {
		for {
			if sn := <-obsoleteQueue; !config.NoPurge {
				sn.purge()
			}
		}
	}()
	debugf("started purge goroutine")

	// If we are going to automatically purge all expired snapshots, we
	// needn't even starting the gofunc
	if config.NoPurge {
		// Free space claiming function
		go func() {
			for {
				// Wait until we are ordered to do something
				<-freeSpaceCheck
				// Get all obsolete snapshots
				// This returns a sorted list
				snapshots, err := findSnapshots(cl)
				if err != nil {
					log.Println(err)
					return
				}
				if len(snapshots) < 2 {
					log.Println("less than 2 snapshots found, not pruning")
					return
				}
				obsolete := snapshots.state(stateObsolete, none)
				// We only delete as long as we need *AND* we have something to delete
				for !checkFreeSpace(config.repository, config.MinPercSpace, config.MinGiBSpace) && len(obsolete) > 0 {
					// If there is not enough space, purge the oldest snapshot
					last := len(obsolete) - 1
					obsolete[last].purge()
					// We remove it from the list, it's quicker than recalculating the list.
					obsolete = obsolete[:last]
				}
			}
		}()
	}

	// Global signal handling
	sigc := make(chan os.Signal, 1)
	signal.Notify(sigc, syscall.SIGINT, syscall.SIGTERM, syscall.SIGUSR1)
	select {
	case sig := <-sigc:
		debugf("Got signal %s", sig)
		switch sig {
		case syscall.SIGINT, syscall.SIGTERM:
			log.Println("-> Immediate exit")
		case syscall.SIGUSR1:
			log.Println("-> Graceful exit")
			createExit <- true
			ferr = <-createExitDone
		}
	// ferr will hold the error that happened in the CREATE_LOOP
	case ferr = <-createExitDone:
		log.Println("-> Rsync exit")
	}
	return
}

// subcmdList give the user an overview of what's in the repository.
func subcmdList(cl clock) {
	intervals := schedules[config.Schedule]
	if cl == nil {
		cl = new(realClock)
	}
	snapshots, err := findSnapshots(cl)
	if err != nil {
		log.Println(err)
	}
	for n := len(intervals) - 2; n >= 0; n-- {
		debugf("listing interval %d", n)
		if config.showAll {
			snapshots = snapshots.state(any, none)
		} else {
			snapshots = snapshots.state(stateComplete, none)
		}
		snapshots := snapshots.interval(intervals, n, cl)
		debugf("snapshots in interval %d: %s", n, snapshots)
		if n < len(intervals)-2 {
			ct.Foreground(ct.Yellow, false)
			fmt.Printf("### From %s ago, %d/%d\n", intervals.offset(n+1), len(snapshots), intervals.goal(n))
			ct.ResetColor()
		} else {
			ct.Foreground(ct.Yellow, false)
			if config.MaxKeep != 0 {
				fmt.Printf("### From past, %d/%d\n", len(snapshots), config.MaxKeep)
			} else if config.MinPercSpace != 0 {
				fmt.Printf("### From past, %d/(keep %.1f%% free)\n", len(snapshots), config.MinPercSpace)
			} else if config.MinGiBSpace != 0 {
				fmt.Printf("### From past, %d/(keep %dGiB free)\n", len(snapshots), config.MinGiBSpace)
			} else {
				fmt.Printf("### From past, %d/∞\n", len(snapshots))
			}
			ct.ResetColor()
		}
		for i, sn := range snapshots {
			stime := sn.startTime.Format("2006-01-02 Monday 15:04:05")
			var dur, dist time.Duration
			if i < len(snapshots)-1 {
				dist = snapshots[i+1].startTime.Sub(sn.startTime)
			}
			if sn.endTime.After(sn.startTime) {
				dur = sn.endTime.Sub(sn.startTime)
			}
			if config.verbose {
				fmt.Printf("%d %s (%s, %s/%s, %s) \"%s\"\n", n, stime, dur, intervals[n], dist, sn.state, sn.Name())
			} else {
				fmt.Printf("%s (%s, %s)\n", stime, dur, intervals[n])
			}
		}
	}
}

func mainExitCode(logIO io.Writer) int {
	logger = log.New(logIO, "", log.Ldate|log.Ltime|log.Lshortfile)
	log.SetOutput(logIO)
	var err error
	if config, err = loadConfig(); err != nil || config == nil {
		if err == flag.ErrHelp {
			return 0
		}
		log.Println(err)
		return 1
	}
	if config.NoLogDate {
		log.SetFlags(logger.Flags() - log.Ldate - log.Ltime)
		logger.SetFlags(logger.Flags() - log.Ldate - log.Ltime)
	}
	switch subcmd {
	case "run":
		log.Printf("%s %s started with pid %d\n", myName, version, os.Getpid())
		log.Printf("### Repository: %s, Origin: %s, Schedule: %s\n", config.repository, config.Origin, config.Schedule)
		err = subcmdRun()
		if err != nil {
			log.Println(err)
			return 2
		}
	case "list":
		if config.noColor {
			ct.Writer = ioutil.Discard
		}
		ct.Foreground(ct.Green, false)
		fmt.Printf("### Repository: %s, Origin: %s, Schedule: %s\n", config.repository, config.Origin, config.Schedule)
		ct.ResetColor()
		subcmdList(nil)
	case "scheds":
		schedules.list()
	}
	return 0
}

func main() {
	rio := newRingIO(os.Stderr, 25, 100)
	exitCode := mainExitCode(rio)
	// do not send a notification when error code is 0 or 1 (error in flag handling)
	// because in the case 1 we can not access the config yet.
	if exitCode > 1 && config.Notify != "" {
		FailureMail(exitCode, rio)
	}
	os.Exit(exitCode)
}