circolog/cmd/circologd/main.go

package main

import (
	"flag"
	"fmt"
	"log"
	"net"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"

	"git.lattuga.net/boyska/circolog"
	"github.com/coreos/go-systemd/daemon"
	syslog "gopkg.in/mcuadros/go-syslog.v2"
)

func cleanSocket(socket string) {
	if err := os.Remove(socket); err != nil {
		fmt.Fprintln(os.Stderr, "Error cleaning", socket, ":", err)
	}
}

func main() {
	var err error
	syslogSocketPath := flag.String("syslogd-socket", "", "The socket to listen to syslog addresses")
	// dumpSocketPath := flag.String("dump-socket", "/run/buffer.sock", "The socket that user will connect to in order to receive logs")
	bufsize := flag.Int("buffer-size", 1000, "Number of messages to keep")
	syslogAddr := flag.String("syslog-addr", "127.0.0.1:9514", "Address:port where to listen for syslog messages")
	queryAddr := flag.String("query-addr", "127.0.0.1:9080", "Address:port where to bind the query service")
	querySocket := flag.String("query-socket", "", "Path to a unix domain socket for the HTTP server; recommended for security reasons!")
	ctlSocket := flag.String("ctl-socket", "/tmp/circologd-ctl.sock", "Path to a unix domain socket for the control server; leave empty to disable")
	verbose := flag.Bool("verbose", false, "Print more output executing the daemon")
	debug := flag.Bool("debug", false, "Print debugging info executing the daemon")
	flag.Parse()

	interrupt := make(chan os.Signal, 1)
	signal.Notify(interrupt, syscall.SIGINT, syscall.SIGUSR1, syscall.SIGUSR2, syscall.SIGTERM)

	hub := circolog.NewHub(*bufsize)
	handler := syslog.NewChannelHandler(hub.LogMessages)
	go hub.Run()

	server := syslog.NewServer()
	server.SetFormat(syslog.RFC5424)
	server.SetHandler(handler)
	if *syslogSocketPath != "" {
		if err = server.ListenUnixgram(*syslogSocketPath); err != nil {
			fmt.Fprintln(os.Stderr, "argh", err)
			os.Exit(1)
		}
		defer cleanSocket(*syslogSocketPath)
		fmt.Printf("Binding socket `%s` [syslog]\n", *syslogSocketPath)
	} else {
		fmt.Printf("Binding address `%s` [syslog]\n", *syslogAddr)
		if err = server.ListenUDP(*syslogAddr); err != nil {
			fmt.Fprintln(os.Stderr, "argh", err)
			os.Exit(1)
		}
	}
	if err = server.Boot(); err != nil {
		fmt.Fprintln(os.Stderr, "argh", err)
		os.Exit(1)
	}

	httpQueryServer := http.Server{Handler: setupHTTP(hub)}
	if *querySocket != "" {
		fmt.Printf("Binding address `%s` [http]\n", *querySocket)
		unixListener, err := net.Listen("unix", *querySocket)
		if err != nil {
			fmt.Fprintln(os.Stderr, "Error binding HTTP unix domain socket", err)
			return
		}
		defer cleanSocket(*querySocket)
		go func() {
			if err := httpQueryServer.Serve(unixListener); err != nil && err != http.ErrServerClosed {
				fmt.Fprintln(os.Stderr, "error binding", *querySocket, ":", err)
			}
		}()
	} else {
		httpQueryServer.Addr = *queryAddr
		fmt.Printf("Binding address `%s` [http]\n", *queryAddr)
		go func() {
			err := httpQueryServer.ListenAndServe()
			if err != nil && err != http.ErrServerClosed {
				fmt.Fprintln(os.Stderr, "error binding", *queryAddr, ":", err)
			}
		}()
	}

	httpCtlServer := http.Server{Handler: setupHTTPCtl(hub, *verbose, *debug)}
	if *ctlSocket != "" {
		fmt.Printf("Binding address `%s` [http]\n", *ctlSocket)
		unixListener, err := net.Listen("unix", *ctlSocket)
		if err != nil {
			fmt.Fprintln(os.Stderr, "Error binding HTTP unix domain socket", err)
			return
		}
		defer cleanSocket(*ctlSocket)
		go func() {
			if err := httpCtlServer.Serve(unixListener); err != nil && err != http.ErrServerClosed {
				fmt.Fprintln(os.Stderr, "error binding:", err)
			}
		}()
	}

	daemon.SdNotify(false, daemon.SdNotifyReady)

	var wdTick <-chan time.Time
	if watchdogTime, err := daemon.SdWatchdogEnabled(false); err == nil && watchdogTime != 0 {
		fmt.Println("systemd watchdog enabled")
		wdTick = time.Tick(watchdogTime / 2) // much less than systemd default of 30s; TODO: make it configurable
	}
	for {
		select {
		case <-wdTick:
			daemon.SdNotify(false, daemon.SdNotifyWatchdog)
		case sig := <-interrupt:
			if sig == syscall.SIGUSR1 {
				response := make(chan circolog.CommandResponse)
				hub.Commands <- circolog.HubFullCommand{Command: circolog.CommandPauseToggle, Response: response}
				resp := <-response
				if resp.Value.(bool) {
					log.Println("resumed")
				} else {
					log.Println("paused")
				}
			}
			if sig == syscall.SIGUSR2 {
				response := make(chan circolog.CommandResponse)
				hub.Commands <- circolog.HubFullCommand{Command: circolog.CommandClear, Response: response}
				resp := <-response
				if resp.Value.(bool) {
					log.Println("buffer cleaned")
				} else {
					log.Println("buffer NOT cleaned")
				}
			}
			if sig == syscall.SIGTERM || sig == syscall.SIGINT {
				log.Println("Quitting because of signal", sig)
				daemon.SdNotify(false, daemon.SdNotifyStopping)
				server.Kill()
				if err := httpQueryServer.Shutdown(nil); err != nil {
					fmt.Fprintln(os.Stderr, "Error closing http server:", err)
				}
				if err := httpCtlServer.Shutdown(nil); err != nil {
					fmt.Fprintln(os.Stderr, "Error closing control server:", err)
				}
				return
			}
		}
	}
}