Ignore build/

This reverts commit 07f4246c80, reversing
changes made to 6da968177b.

.dockerignore

@@ -0,0 +1 @@
+/Dockerfile

.gitignore

@@ -0,0 +1,2 @@
+.*.vim
+/build

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "vendor/gopkg.in/mcuadros/go-syslog.v2"]
+	path = vendor/gopkg.in/mcuadros/go-syslog.v2
+	url = https://github.com/boyska/go-syslog

Dockerfile

@@ -0,0 +1,13 @@
+FROM golang:1.12
+
+ENV GOOS=linux
+ENV CGO_ENABLED=0
+
+RUN mkdir -p $GOPATH/src/git.lattuga.net/boyska/circolog/
+COPY . $GOPATH/src/git.lattuga.net/boyska/circolog/
+RUN go get git.lattuga.net/boyska/circolog/...
+
+VOLUME [$GOPATH"/bin"]
+
+ENTRYPOINT ["/go/bin/circologd"]
+CMD []

README.md

@@ -72,10 +72,10 @@ Pausing might be the easiest way to make circologd only run "when needed".
 
 When circologd resumes, no previous message is lost.
 
-
-To pause circologd with signals , send a `USR1` signal to the main pid. To "resume", send a `USR1` again.
-
-To pause with HTTP, send a `POST /pause/toggle` to your circologd control socket.
+To pause/unpause:
+ * `circologctl pause`
+ * `pkill -USR1 circologd`
+ * `POST /pause/toggle` to your circologd control socket
 
 ### Clear
 
@@ -83,3 +83,36 @@ When you clear the circologd's buffer, it will discard every message it has, but
 messages.
 
 You can do that with `POST /logs/clear`
+
+### Filter
+
+circologd can drop irrelevant messages using filters. A filter is a sql-like expression (for the exact syntax
+you can see [the doc for the underlying library](https://github.com/araddon/qlbridge/blob/master/FilterQL.md),
+qlbridge), but just imitating sql where clauses can be enough!
+
+`circologctl filter message NOT LIKE '%usb%'` will discard everything related to usb.
+
+The filter will be applied to incoming messages, so messages mentioning usb will not be saved in memory at all.
+
+You can put zero or one filters at a time. That is, you can not stack more filters... but FilterQL syntax
+supports `AND` operators, so this is not an issue.
+
+To remove filtering (thus accepting every message) run `circologctl filter`
+
+NOTE: `circolog-tail` supports filters with exactly the same syntax, but they are two different kinds of
+filtering: one is server-side, the other is client-side. When you filter server-side with `circologctl
+filter`, circologd will refuse messages not matching the filter. If you only filter with `circolog-tail`, the
+message you are filtering out will still consume space in memory (and will be available to other clients).
+
+Filtering brings big dependencies, which will add some 5-6 megabytes to circolog binaries. If you want to
+avoid it, install with `go install -tags nofilter git.lattuga.net/boyska/circolog/...` and your binaries will
+be a bit smaller.
+
+## Develop
+
+To tinker with circolog, there is also a `Dockerfile`. Simply:
+
+```
+$ docker build -t circolog .
+$ docker run -t circolog
+```

cmd/circolog-tail/main.go

@@ -11,23 +11,71 @@ import (
 	"strconv"
 	"time"
 
+	"git.lattuga.net/boyska/circolog/data"
+	"git.lattuga.net/boyska/circolog/filtering"
 	"git.lattuga.net/boyska/circolog/formatter"
 	"github.com/gorilla/websocket"
-	"gopkg.in/mcuadros/go-syslog.v2/format"
+	isatty "github.com/mattn/go-isatty"
+	"github.com/mgutz/ansi"
 	"gopkg.in/mgo.v2/bson"
 )
 
+type BoolAuto uint
+
+const (
+	BoolAuto_NO   BoolAuto = iota
+	BoolAuto_YES  BoolAuto = iota
+	BoolAuto_AUTO BoolAuto = iota
+)
+
+func (b *BoolAuto) String() string {
+	switch *b {
+	case BoolAuto_NO:
+		return "no"
+	case BoolAuto_YES:
+		return "always"
+	case BoolAuto_AUTO:
+		return "auto"
+	}
+	return ""
+}
+func (b *BoolAuto) Set(s string) error {
+	switch s {
+	case "auto":
+		*b = BoolAuto_AUTO
+	case "always":
+		*b = BoolAuto_YES
+	case "no":
+		*b = BoolAuto_NO
+	default:
+		return fmt.Errorf("Invalid value %s", s)
+	}
+	return nil
+}
+
 func main() {
-	addr := flag.String("addr", "localhost:9080", "http service address")
-	querySocket := flag.String("socket", "", "Path to a unix domain socket for the HTTP server")
+	queryAddr := flag.String("addr", "", "http service address")
+	querySocket := flag.String("socket", "/tmp/circologd-query.sock", "Path to a unix domain socket for the HTTP server")
 	backlogLimit := flag.Int("n", -1, "Limit the backlog length, defaults to no limit (-1)")
+	var format formatter.Format
+	format = formatter.FormatSyslog
+	flag.Var(&format, "fmt", "Output format [syslog|json]")
+	var filter filtering.ExprValue
+	flag.Var(&filter, "where", "sql-like query to filter logs")
+	// TODO: change to color-mode=auto/no/always
+	hasColor := BoolAuto_AUTO
+	flag.Var(&hasColor, "color", "dis/enable colors; yes/no/auto")
 	flag.Parse()
 
+	if hasColor == BoolAuto_NO || (!isatty.IsTerminal(os.Stdout.Fd()) && hasColor != BoolAuto_YES) {
+		ansi.DisableColors(true)
+	}
+
 	interrupt := make(chan os.Signal, 1)
 	signal.Notify(interrupt, os.Interrupt)
 	var d *websocket.Dialer
 	u := url.URL{Scheme: "ws",
-		Host: *addr, // ignored in case of -socket; see the Dialer below
+		Host: *queryAddr, // ignored in case of -socket; see the Dialer below
 		Path: "/ws",
 	}
 	q := u.Query()
@@ -36,7 +84,7 @@ func main() {
 		q.Set("l", strconv.Itoa(*backlogLimit))
 	}
 	u.RawQuery = q.Encode()
-	if *querySocket != "" {
+	if *queryAddr == "" {
 		d = &websocket.Dialer{
 			NetDial: func(network, addr string) (net.Conn, error) {
 				return net.Dial("unix", *querySocket)
@@ -47,7 +95,7 @@ func main() {
 		log.Printf("connecting to %s", *querySocket)
 	} else {
 		d = websocket.DefaultDialer
-		log.Printf("connecting to %s", *addr)
+		log.Printf("connecting to %s", *queryAddr)
 	}
 
 	c, _, err := d.Dial(u.String(), nil)
@@ -67,15 +115,20 @@ func main() {
 				log.Println("close:", err)
 				return
 			}
-			var parsed format.LogParts
+			var parsed data.Message
 			if err := bson.Unmarshal(serialized, &parsed); err != nil {
-				log.Println("invalid YAML", err)
+				log.Println("invalid BSON", err)
 				continue
 			}
-			if err := formatter.WriteFormatted(os.Stdout, formatter.FormatSyslog, parsed); err != nil {
+			if !filter.Validate(parsed) {
+				continue
+			}
+			if err := formatter.WriteFormatted(os.Stdout, format, parsed); err != nil {
 				log.Println("error printing", err)
 			}
-			fmt.Println()
+			if format == formatter.FormatSyslog { // oops
+				fmt.Println()
+			}
 		}
 	}()
 
@@ -96,7 +149,7 @@ func main() {
 			select {
 			case <-done:
 				log.Println("Successfully close")
-			case <-time.After(1 * time.Second):
+			case <-time.After(5 * time.Second):
 				log.Println("Forced close")
 			}
 			return

cmd/circologctl/main.go

@@ -0,0 +1,184 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"os"
+	"strings"
+	"time"
+
+	"git.lattuga.net/boyska/circolog"
+)
+
+var globalOpts struct {
+	ctlSock string
+	verbose bool
+	debug   bool
+}
+
+var ctl http.Client
+
+type commandFunc func([]string) error
+
+var cmdMap map[string]commandFunc
+
+func init() {
+	cmdMap = map[string]commandFunc{
+		// TODO: implement set and get of config at runtime
+		//"set":     setCmd,
+		//"get":     getCmd,
+		"status":  statusCmd,
+		"pause":   pauseCmd,
+		"filter":  filterCmd,
+		"reload":  reloadCmd,
+		"restart": restartCmd,
+		"help":    helpCmd,
+	}
+}
+
+//func setCmd(ctlSock string, args []string) error {}
+
+//func getCmd(ctlSock string, args []string) error {}
+
+func statusCmd(args []string) error {
+	flagset := flag.NewFlagSet(args[0], flag.ExitOnError)
+	outFormat := flagset.String("format", "plain", "Which format to use as output for this command (json, pretty, plain)")
+	flagset.Parse(args[1:])
+	resp, err := ctl.Get("http://unix/status")
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	respBytes, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return err
+	}
+	respJSON := make(map[string]circolog.StatusResponse)
+	err = json.Unmarshal(respBytes, &respJSON)
+	if err != nil {
+		return err
+	}
+	switch *outFormat {
+	case "json":
+		fmt.Printf("%s", string(respBytes))
+	case "pretty":
+		prettyJSON, err := json.MarshalIndent(respJSON, "", "  ")
+		if err != nil {
+			return err
+		}
+		fmt.Printf("%s\n", prettyJSON)
+	case "plain":
+		fmt.Printf("Buffer Size:   %d\n", respJSON["status"].Size)
+		fmt.Printf("Server Status: %s\n", respJSON["status"].Status())
+		fmt.Printf("Filter String: %s\n", respJSON["status"].Filter)
+	}
+	return nil
+}
+
+func pauseCmd(args []string) error {
+	var dontChangeAgain time.Duration
+	flagset := flag.NewFlagSet(args[0], flag.ExitOnError)
+	waitTime := flagset.Duration("wait-time", dontChangeAgain, "How long to wait before untoggling the state, defaults to never")
+	flagset.Parse(args[1:])
+	postBody := make(map[string][]string)
+	if *waitTime != dontChangeAgain {
+		postBody["waitTime"] = []string{fmt.Sprintf("%s", *waitTime)}
+	}
+	if globalOpts.debug {
+		fmt.Println("[DEBUG] postBody:", postBody)
+	}
+	resp, err := ctl.PostForm("http://unix/pause/toggle", postBody)
+	if globalOpts.verbose {
+		defer resp.Body.Close()
+		bodyBytes, err := ioutil.ReadAll(resp.Body)
+		if err != nil {
+			return err
+		}
+		fmt.Println(string(bodyBytes))
+	}
+	return err
+}
+
+func filterCmd(args []string) error {
+	filter := strings.Join(args[1:], " ")
+	postBody := make(map[string][]string)
+	postBody["where"] = []string{filter}
+	if globalOpts.debug {
+		fmt.Println("[DEBUG] postBody:", postBody)
+	}
+	resp, err := ctl.PostForm("http://unix/filter", postBody)
+	if resp.StatusCode != 200 || globalOpts.verbose {
+		defer resp.Body.Close()
+		bodyBytes, err := ioutil.ReadAll(resp.Body)
+		if err != nil {
+			return err
+		}
+		fmt.Println(string(bodyBytes))
+	}
+	return err
+}
+
+func reloadCmd(args []string) error {
+	return nil
+}
+
+func restartCmd(args []string) error {
+	return nil
+}
+
+func helpCmd(args []string) error {
+	usage(os.Stdout)
+	os.Exit(0)
+	return nil
+}
+
+func usage(w io.Writer) {
+	fmt.Fprintf(w, "USAGE: %s [globalOpts] [SUBCOMMAND] [opts]\n", os.Args[0])
+	fmt.Fprintf(w, "\nSUBCOMMANDS:\n\n")
+	for command := range cmdMap {
+		fmt.Fprintf(w, "\t%s\n", command)
+	}
+}
+
+func parseAndRun(args []string) {
+	cmdName := args[0]
+	cmdToRun, ok := cmdMap[cmdName]
+	if !ok {
+		fmt.Fprintf(os.Stderr, "Unknown subcommand: %s\n", cmdName)
+		usage(os.Stderr)
+		os.Exit(2)
+	}
+	// from here: https://gist.github.com/teknoraver/5ffacb8757330715bcbcc90e6d46ac74
+	ctl = http.Client{
+		Transport: &http.Transport{
+			DialContext: func(_ context.Context, _, _ string) (net.Conn, error) {
+				return net.Dial("unix", globalOpts.ctlSock)
+			},
+		},
+	}
+	err := cmdToRun(args)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error:\n%s\n", err)
+		os.Exit(1)
+	}
+}
+
+func main() {
+	flag.StringVar(&globalOpts.ctlSock, "ctl-socket", "/tmp/circologd-ctl.sock",
+		"Path to a unix domain socket for the control server; leave empty to disable")
+	flag.BoolVar(&globalOpts.verbose, "verbose", false, "Print more output")
+	flag.BoolVar(&globalOpts.debug, "debug", false, "Print debugging info")
+	flag.Parse()
+	args := flag.Args()
+	if len(args) == 0 {
+		usage(os.Stderr)
+		os.Exit(-1)
+	}
+	parseAndRun(args)
+}

cmd/circologd/activation.go

@@ -0,0 +1,39 @@
+package main
+
+import (
+	"net"
+
+	"github.com/coreos/go-systemd/activation"
+)
+
+func Listeners() ([]net.Listener, error) {
+	files := activation.Files(false)
+	listeners := make([]net.Listener, len(files))
+
+	for i, f := range files {
+		if pc, err := net.FileListener(f); err == nil {
+			listeners[i] = pc
+			f.Close()
+		}
+	}
+	return listeners, nil
+}
+
+// PacketConns returns a slice containing a net.PacketConn for each matching socket type
+// passed to this process.
+//
+// The order of the file descriptors is preserved in the returned slice.
+// Nil values are used to fill any gaps. For example if systemd were to return file descriptors
+// corresponding with "udp, tcp, udp", then the slice would contain {net.PacketConn, nil, net.PacketConn}
+func PacketConns() ([]net.PacketConn, error) {
+	files := activation.Files(false)
+	conns := make([]net.PacketConn, len(files))
+
+	for i, f := range files {
+		if pc, err := net.FilePacketConn(f); err == nil {
+			conns[i] = pc
+			f.Close()
+		}
+	}
+	return conns, nil
+}

cmd/circologd/http_ctl.go

@@ -2,23 +2,69 @@ package main
 
 import (
 	"encoding/json"
+	"fmt"
+	"log"
 	"net/http"
+	"os"
+	"time"
 
+	fractal "git.lattuga.net/blallo/gotools/formatting"
 	"git.lattuga.net/boyska/circolog"
 	"github.com/gorilla/mux"
 )
 
-func setupHTTPCtl(hub circolog.Hub) *mux.Router {
-	mux := mux.NewRouter()
-	mux.HandleFunc("/pause/toggle", togglePause(hub)).Methods("POST")
-	mux.HandleFunc("/logs/clear", clearQueue(hub)).Methods("POST")
-	return mux
+func setupHTTPCtl(hub circolog.Hub, verbose, debug bool) *mux.Router {
+	m := mux.NewRouter()
+	m.HandleFunc("/pause/toggle", togglePause(hub, verbose, debug)).Methods("POST")
+	m.HandleFunc("/filter", setFilter(hub, verbose, debug)).Methods("POST")
+	m.HandleFunc("/status", getStatus(hub, verbose, debug)).Methods("GET")
+	m.HandleFunc("/logs/clear", clearQueue(hub, verbose)).Methods("POST")
+	m.HandleFunc("/help", printHelp(verbose)).Methods("GET")
+	m.HandleFunc("/echo", echo(verbose)).Methods("GET")
+	return m
 }
 
-func togglePause(hub circolog.Hub) http.HandlerFunc {
+func getStatus(hub circolog.Hub, verbose, debug bool) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
-		hub.Commands <- circolog.HubFullCommand{Command: circolog.CommandPauseToggle}
-		resp := <-hub.Responses
+		response := make(chan circolog.CommandResponse)
+		hub.Commands <- circolog.HubFullCommand{
+			Command:  circolog.CommandStatus,
+			Response: response,
+		}
+		resp := <-response
+		w.Header().Set("content-type", "application/json")
+		enc := json.NewEncoder(w)
+		enc.Encode(map[string]interface{}{"status": resp.Value})
+	}
+}
+
+func togglePause(hub circolog.Hub, verbose, debug bool) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		if verbose {
+			log.Printf("[%s] %s - toggled pause", r.Method, r.RemoteAddr)
+		}
+		r.ParseForm()
+		waitTimePar := r.FormValue("waitTime")
+		var waitTime time.Duration
+		var err error
+		if waitTimePar != "" {
+			waitTime, err = time.ParseDuration(waitTimePar)
+			if err != nil {
+				w.WriteHeader(400)
+				fmt.Fprintln(w, "waitTime not understood:", waitTimePar)
+				return
+			}
+		}
+		if debug {
+			fmt.Println("[DEBUG] waitTime:", waitTime)
+		}
+		response := make(chan circolog.CommandResponse)
+		hub.Commands <- circolog.HubFullCommand{
+			Command:    circolog.CommandPauseToggle,
+			Response:   response,
+			Parameters: map[string]interface{}{"waitTime": waitTime},
+		}
+		resp := <-response
 		active := resp.Value.(bool)
 		w.Header().Set("content-type", "application/json")
 		enc := json.NewEncoder(w)
@@ -26,13 +72,70 @@ func togglePause(hub circolog.Hub) http.HandlerFunc {
 	}
 }
 
-func clearQueue(hub circolog.Hub) http.HandlerFunc {
+func setFilter(hub circolog.Hub, verbose, debug bool) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
-		hub.Commands <- circolog.HubFullCommand{Command: circolog.CommandClear}
-		resp := <-hub.Responses
+		r.ParseForm()
+		where := r.FormValue("where")
+		response := make(chan circolog.CommandResponse)
+		hub.Commands <- circolog.HubFullCommand{
+			Command:    circolog.CommandNewFilter,
+			Response:   response,
+			Parameters: map[string]interface{}{"where": where},
+		}
+		resp := <-response
+		if !resp.Value.(map[string]interface{})["success"].(bool) {
+			w.WriteHeader(400)
+		}
+		w.Header().Set("content-type", "application/json")
+		enc := json.NewEncoder(w)
+		enc.Encode(resp.Value)
+	}
+}
+
+func clearQueue(hub circolog.Hub, verbose bool) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		if verbose {
+			log.Printf("[%s] %s - cleared queue", r.Method, r.RemoteAddr)
+		}
+		response := make(chan circolog.CommandResponse)
+		hub.Commands <- circolog.HubFullCommand{Command: circolog.CommandClear, Response: response}
+		resp := <-response
 		success := resp.Value.(bool)
 		w.Header().Set("content-type", "application/json")
 		enc := json.NewEncoder(w)
 		enc.Encode(map[string]interface{}{"success": success})
 	}
 }
+
+func printHelp(verbose bool) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		if verbose {
+			log.Printf("[%s] %s - asked for help", r.Method, r.RemoteAddr)
+		}
+		w.Header().Set("content-type", "application/json")
+		enc := json.NewEncoder(w)
+		var pathsWithDocs = map[string]string{
+			"/pause/toggle": "Toggle the server from pause state (not listening)",
+			"/logs/clear":   "Wipe the buffer from all the messages",
+			"/status":       "Get info on the status of the server",
+			"/help":         "This help",
+			"/echo":         "Answers to heartbeat",
+		}
+		fractal.EncodeJSON(pathsWithDocs, enc)
+		if verbose {
+			errEnc := json.NewEncoder(os.Stderr)
+			fractal.EncodeJSON(pathsWithDocs, errEnc)
+		}
+	}
+}
+
+func echo(verbose bool) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		log.Println("Echo")
+		if verbose {
+			log.Printf("[%s] %s - asked for echo", r.Method, r.RemoteAddr)
+		}
+		w.Header().Set("content-type", "text/plain")
+		fmt.Fprintln(w, "I am on!")
+	}
+}

cmd/circologd/http_log.go

@@ -9,9 +9,9 @@ import (
 	"time"
 
 	"git.lattuga.net/boyska/circolog"
+	"git.lattuga.net/boyska/circolog/data"
 	"git.lattuga.net/boyska/circolog/formatter"
 	"github.com/gorilla/websocket"
-	"gopkg.in/mcuadros/go-syslog.v2/format"
 )
 
 func setupHTTP(hub circolog.Hub) *http.ServeMux {
@@ -100,7 +100,7 @@ func getHTTPHandler(hub circolog.Hub) http.HandlerFunc {
 		opts.Nofollow = true
 
 		client := circolog.Client{
-			Messages: make(chan format.LogParts, 20),
+			Messages: make(chan data.Message, 20),
 			Options:  opts,
 		}
 		hub.Register <- client
@@ -141,7 +141,7 @@ func getWSHandler(hub circolog.Hub) http.HandlerFunc {
 			return
 		}
 		client := circolog.Client{
-			Messages: make(chan format.LogParts, 20),
+			Messages: make(chan data.Message, 20),
 			Options:  opts,
 		}
 		hub.Register <- client
@@ -153,6 +153,15 @@ func getWSHandler(hub circolog.Hub) http.HandlerFunc {
 				hub.Unregister <- c
 				conn.Close()
 			}()
+			go func() {
+				for {
+					_, _, err := conn.ReadMessage()
+					if err != nil {
+						conn.Close()
+						return
+					}
+				}
+			}()
 			for {
 				select {
 				case message, ok := <-c.Messages:

cmd/circologd/main.go

@@ -9,50 +9,78 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	"git.lattuga.net/boyska/circolog"
+	"git.lattuga.net/boyska/circolog/formatter"
+	"github.com/coreos/go-systemd/daemon"
 	syslog "gopkg.in/mcuadros/go-syslog.v2"
 )
 
+var socketsToRemove []string
+
 func cleanSocket(socket string) {
 	if err := os.Remove(socket); err != nil {
 		fmt.Fprintln(os.Stderr, "Error cleaning", socket, ":", err)
 	}
 }
 
+func removeAtExit(socket string) {
+	socketsToRemove = append(socketsToRemove, socket)
+}
+
 func main() {
 	var err error
-	syslogSocketPath := flag.String("syslogd-socket", "", "The socket to listen to syslog addresses")
+	var syslogSocket SyslogSocket
+	var logFmt formatter.SyslogRFC
+	logFmt.Format = syslog.Automatic
+	flag.Var(&syslogSocket, "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!")
+	queryAddr := flag.String("query-addr", "", "Address:port where to bind the query service")
+	querySocket := flag.String("query-socket", "/tmp/circologd-query.sock", "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")
+	flag.Var(&logFmt, "log-fmt", "Log messages format. If not set, defaults to automatic choice. Allowed values: rfc3164, rfc5424, auto.")
+	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.SIGTERM)
+	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.SetFormat(logFmt.Format)
+	fmt.Printf("Syslog format set to: %s\n", logFmt.String())
 	server.SetHandler(handler)
-	if *syslogSocketPath != "" {
-		if err = server.ListenUnixgram(*syslogSocketPath); err != nil {
-			fmt.Fprintln(os.Stderr, "argh", err)
-			os.Exit(1)
+	if syslogSocket.isSocketActivated {
+		fmt.Printf("Binding to socket `%s` [syslog]\n", syslogSocket.String())
+		if syslogSocket.Listener != nil {
+			fmt.Println("(stream)")
+			server.Listen(syslogSocket.Listener)
+		} else {
+			fmt.Println("(datagram)", syslogSocket.Conn)
+			server.ListenDgram(syslogSocket.Conn)
 		}
-		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)
+		syslogSocketPath := syslogSocket.Path
+		if syslogSocketPath != "" {
+			if err = server.ListenUnixgram(syslogSocketPath); err != nil {
+				fmt.Fprintln(os.Stderr, "argh", err)
+				os.Exit(1)
+			}
+			fmt.Printf("Binding socket `%s` [syslog]\n", syslogSocketPath)
+			removeAtExit(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 {
@@ -61,14 +89,14 @@ func main() {
 	}
 
 	httpQueryServer := http.Server{Handler: setupHTTP(hub)}
-	if *querySocket != "" {
+	if *queryAddr == "" {
 		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)
+		removeAtExit(*querySocket)
 		go func() {
 			if err := httpQueryServer.Serve(unixListener); err != nil && err != http.ErrServerClosed {
 				fmt.Fprintln(os.Stderr, "error binding", *querySocket, ":", err)
@@ -85,7 +113,7 @@ func main() {
 		}()
 	}
 
-	httpCtlServer := http.Server{Handler: setupHTTPCtl(hub)}
+	httpCtlServer := http.Server{Handler: setupHTTPCtl(hub, *verbose, *debug)}
 	if *ctlSocket != "" {
 		fmt.Printf("Binding address `%s` [http]\n", *ctlSocket)
 		unixListener, err := net.Listen("unix", *ctlSocket)
@@ -93,7 +121,7 @@ func main() {
 			fmt.Fprintln(os.Stderr, "Error binding HTTP unix domain socket", err)
 			return
 		}
-		defer cleanSocket(*ctlSocket)
+		removeAtExit(*ctlSocket)
 		go func() {
 			if err := httpCtlServer.Serve(unixListener); err != nil && err != http.ErrServerClosed {
 				fmt.Fprintln(os.Stderr, "error binding:", err)
@@ -101,22 +129,41 @@ func main() {
 		}()
 	}
 
-	// TODO: now we are ready
+	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 {
-				hub.Commands <- circolog.HubFullCommand{Command: circolog.CommandPauseToggle}
-				resp := <-hub.Responses
+				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)
@@ -124,6 +171,9 @@ func main() {
 				if err := httpCtlServer.Shutdown(nil); err != nil {
 					fmt.Fprintln(os.Stderr, "Error closing control server:", err)
 				}
+				for _, socket := range socketsToRemove {
+					cleanSocket(socket)
+				}
 				return
 			}
 		}

cmd/circologd/sockets.go

@@ -0,0 +1,50 @@
+package main
+
+import (
+	"net"
+)
+
+// SyslogSocket is a struct eventually containing a net.Listener
+// ready with messages, and a Path in case the Listener is not present.
+type SyslogSocket struct {
+	Listener          net.Listener
+	Conn              net.PacketConn
+	Path              string
+	isSocketActivated bool
+}
+
+// Set from command-line
+func (s *SyslogSocket) Set(v string) error {
+	err := s.getActivationSocket()
+	if err == nil && (s.Conn != nil || s.Listener != nil) {
+		s.isSocketActivated = true
+	}
+	if !s.isSocketActivated {
+		s.Path = v
+	}
+	return nil
+}
+
+func (s *SyslogSocket) String() string {
+	if s.isSocketActivated {
+		return "systemd-provided"
+	}
+	return s.Path
+}
+
+func (s *SyslogSocket) getActivationSocket() error {
+	conns, err := PacketConns()
+	if err == nil && len(conns) > 0 && conns[0] != nil {
+		s.Conn = conns[0]
+		return nil
+	}
+	listeners, err := Listeners()
+	if err != nil {
+		return err
+	}
+	if len(listeners) == 0 {
+		return nil
+	}
+	s.Listener = listeners[0]
+	return nil
+}

data/data.go

@@ -0,0 +1,32 @@
+package data
+
+import "gopkg.in/mcuadros/go-syslog.v2/format"
+
+// Message is currently an alias for format.Logparts, but this is only temporary; sooner or later, a real struct will be used
+// The advantage of having an explicit Message is to clear out what data we are sending to circolog "readers"
+// This is not necessarily (and not in practice) the same structure that we receive from logging programs
+type Message format.LogParts
+
+// LogEntryToMessage converts messages received from writers to the format we promise to readers
+func LogEntryToMessage(orig format.LogParts) Message {
+	m := Message{}
+	if orig["version"] == 1 { // RFC5424
+		m["prog"] = orig["app_name"]
+		m["client"] = orig["client"]
+		m["host"] = orig["hostname"]
+		m["proc_id"] = orig["proc_id"]
+		m["msg"] = orig["message"]
+		m["facility"] = orig["facility"]
+		m["time"] = orig["timestamp"]
+		m["sev"] = orig["severity"]
+	} else { //RFC3164
+		m["prog"] = orig["tag"]
+		m["client"] = orig["client"]
+		m["host"] = orig["hostname"]
+		m["msg"] = orig["content"]
+		m["sev"] = orig["severity"]
+		m["time"] = orig["timestamp"]
+		m["proc_id"] = "-"
+	}
+	return m
+}

data/data_test.go

@@ -0,0 +1,58 @@
+package data
+
+import (
+	"testing"
+	"time"
+
+	"gopkg.in/mcuadros/go-syslog.v2/format"
+)
+
+var timeNow = time.Now()
+
+var returnValues = map[string]interface{}{
+	"prog":   "test_app",
+	"client": "test_client",
+	"host":   "my_machine",
+	"msg":    "test message",
+	"sev":    3,
+	"time":   timeNow,
+}
+
+var msgRFC5424 = format.LogParts{
+	"version":   1,
+	"app_name":  returnValues["prog"],
+	"client":    returnValues["client"],
+	"hostname":  returnValues["host"],
+	"proc_id":   "spam_process",
+	"message":   returnValues["msg"],
+	"facility":  "hell",
+	"timestamp": returnValues["time"],
+	"severity":  returnValues["sev"],
+}
+
+var msgRFC3164 = format.LogParts{
+	"tag":       returnValues["prog"],
+	"client":    returnValues["client"],
+	"hostname":  returnValues["host"],
+	"content":   returnValues["msg"],
+	"severity":  returnValues["sev"],
+	"timestamp": returnValues["time"],
+	"proc_id":   "spam_process",
+}
+
+var testMessages = []format.LogParts{
+	msgRFC5424,
+	msgRFC3164,
+}
+
+func TestLogEntryToMessage(t *testing.T) {
+	for _, msg := range testMessages {
+		parsedMsg := LogEntryToMessage(msg)
+
+		for key, value := range returnValues {
+			if data, ok := parsedMsg[key]; !ok || data != value {
+				t.Errorf("Missing/wrong key: %s\nmsg: %s\nparsed msg: %s\n", key, value, data)
+			}
+		}
+	}
+}

docs/.gitignore

@@ -0,0 +1 @@
+site

docs/docs/hacking.md

@@ -0,0 +1,20 @@
+## Why not on github.com
+
+We love collaboration between people, and software development can be
+a wonderful playground. Nevertheless, we are not comfortable with the
+social-neworkish nature of github. Also, the choice of a platform is something
+that is meaningful in itself. Fortunately `go` allows to pull packages from any
+git server that is publicly available. Therefore do not be afraid of go-getting
+from git.lattuga.net.
+
+lattuga.net is a self-managed server, run with clear principles of not-for-profit and antifascist nature.
+
+We think using situated, decentralized tools and networks is fundamental to get a technological landscape that
+doesn't lead to oppression.
+
+### So how to collaborate
+
+Pull request is not the only paradigm for collaboration :)
+
+If you want to collaborate to the software, either drop a mail to
+<indirizzo@scamuffo.com>. We are also reachable at [IRC irc.mufhd0.net]

docs/docs/index.md

@@ -0,0 +1,61 @@
+# Circolog
+
+Circolog proposes a different approach to the problem of having useful logs.
+
+Circolog is a syslog server which doesn't write to disk. Ever. It doesn't
+consume tons of RAM like elasticsearch/logstash, nor does it give you plots and
+a very long history. It is however a very useful tool when you want to minimize
+disk writes. We wrote it thinking about user privacy: your logs can be just as
+sensitive as your database if you log too much!
+
+[Spiegone su casi d'uso, buttare i log, debuggare rapidamente magari
+applicazioni verbose, eccetera]
+
+## Why it is cool in 2 minutes
+
+- It keeps your users safer.
+
+- Read logs conveniently: filter with a proper (and easy) query language!
+  `grep` is powerful, but sometimes you want something more expressive:
+
+```
+circolog-tail -where 'prog=="apache" and msg LIKE "%memory%"'
+circolog-tail -where '(prog=="apache" OR prog LIKE "php%") AND msg LIKE "%memory%"'
+circolog-tail -where 'prog=="mysql" OR sev >= warning'
+```
+
+- Colors: highlight severity and visually group related message
+
+    [screenshot]
+
+    It's not only about being nice, we swear! Coloring logs also means reaching the
+    most important entries easily, and grouping related entries together. Output
+    logs in the format you prefer _now_. Depending on what you're doing, log format
+    might be useful... or distracting. For example, how many times have you used
+    the `hostname` part of it? With the common disk-based logging, you need to
+    choose once and for all how your logs will be saved.
+
+
+- Hackable: we think that `circolog-tail` is pretty cool, but you definitely can
+  reuse simpler tools to get logs and filter them the way you prefer. Clients
+  can read logs using plain HTTP (or websocket). Most of the cool features of
+  `circolog-tail` are actually implemented server-side, so you can use filters
+  (or other options) with any client.
+
+
+- Fast, secure by default (?), easy to deploy. Those are features that you
+  should expect, not be surprised of! Circolog can easily process thousands
+  of log entries per seconds, has sane defaults and can be deployed as a single
+  binary.
+
+
+## Security considerations
+
+While we try our best not to introduce vulnerabilities, this software is not
+meant to be exposed on the wider internet. Beware of binding it on something
+different from `localhost`.
+
+Even without being exposed, care must be given to socket permissions: don't let
+unprivileged users read your logs! We suggest that you use a dedicated
+user/group to run circolog, and make root part of that group.
+

docs/docs/install.md

@@ -0,0 +1,70 @@
+## Installation and Configuration
+
+
+If you trust us, just get the compiled binaries and copy them to
+`/usr/local/{s,}bin/` (but you shouldn't: [TODO: compiling -> altra pagina]).
+
+Running circolog is pretty easy: just run `circologd` by hand or in your
+favourite service manager: here are some contrib scripts for SysV, systemd,
+supervisor... but consider them as hints, don't just copy them!
+
+[TODO: add scripts]
+
+### System Integration
+
+While circologd can be your only syslog daemon, you might want to couple it
+with another syslog to have the best of both worlds. For example, you could use
+rsyslog (or syslog-ng) to write important (ie: `priority >= notice`) log to
+persistent storage, letting circolog handle short-term but heavily detailed
+logs.
+
+Or you could make circolog get messages from `journald`. While possibilities are
+endless, we tried to document some common setups.
+
+#### Use Case: circologd as the only syslog
+
+[mica lo so se funziona, credo di no, ma proviamo] bind it to `/dev/log`.
+
+#### Use Case: circologd receiving messages from syslog-ng
+
+Put this in `/etc/syslog-ng/conf.d/circolog.conf`
+
+```
+destination d_circolog {
+	unix-dgram("/run/circolog/syslog.sock"
+			flags(syslog-protocol)
+		   );
+};
+log { source(s_src); destination(d_circolog); };
+```
+
+This doesn't tell syslog-ng not to log to disk.
+
+Change `/etc/syslog-ng.conf` according to your need.
+
+#### Use Case: circologd receiving messages from rsyslog
+
+rsyslogd can easily be configured to send every message to circolog:
+
+```
+$ModLoad omuxsock
+$OMUxSockSocket /tmp/circolog.sock
+*.* :omuxsock:
+```
+
+If you want to force messages of level `debug` and `info` not to be logged to disk, you can put this snippet
+at the **top** of the rules
+
+```
+$ModLoad omuxsock
+$OMUxSockSocket /tmp/circolog.sock
+*.* :omuxsock:
+*.=debug;*.=info stop
+```
+
+#### Use Case: circologd receiving messages from journald
+
+[TODO: il codice va ancora scritto: bisogna bindare /run/systemd/journal/syslog
+con formato != rfc5424]
+
+

docs/docs/query.md

@@ -0,0 +1,27 @@
+Query language
+==============
+
+Circolog uses a sql-inspired query language. If you know SQL, then you can use
+"where clauses" in circolog. If you don't know SQL, don't worry: the language
+is easy enough for you to learn the most basic queries without worrying too
+much.
+
+You can only filter the rows, you can't sort them or group them in any way.
+
+Reference
+---------
+
+Available fields:
+
+ - `msg`: the string with the main information
+ - `prog`: also known as "program" sometimes
+ - `facility`: an integer describing auth, daemon, user, etc.
+ - `host`: the hostname where the entry originated
+ - `time`: date in format `2019-01-07T15:28:58+01:00`
+ - `sev`: an integer describing severity
+
+
+Examples
+--------
+
+TODO

docs/docs/systemd.md

@@ -0,0 +1,94 @@
+## A simple start
+
+The bare minimum you need to get circologd on a systemd-based system is this unit.
+Other options with more features or more security are provided below
+
+    [Unit]
+    Description=In-memory logging
+
+    [Service]
+    User=root
+    Group=adm
+    ExecStart=/usr/local/sbin/circologd -syslogd-socket /run/circolog/syslog.sock  -buffer-size 2000 -query-socket /run/circolog/query.sock
+
+    [Install]
+    WantedBy=multi-user.target
+
+
+## A better unit
+
+This is another unit, which has several security features, such as `DynamicUser`, filesystem restrictions, and
+more.
+
+    [Unit]
+    Description=In-memory logging
+
+    [Service]
+    DynamicUser=true
+    Group=adm
+    RuntimeDirectory=circolog
+    # this is important: circologd will respect umask, so if you want to have files that are not world-readable, you must set it
+    RuntimeDirectoryMode=0750
+    UMask=0026
+    ProtectSystem=full
+    ExecStart=/usr/local/sbin/circologd -syslogd-socket /run/circolog/syslog.sock  -buffer-size 2000 -query-socket /run/circolog/query.sock
+    # security restrictions; useful, but not needed
+    PrivateTmp=true
+    PrivateNetwork=true
+    NoNewPrivileges=true
+    Restrictnamespaces=true
+
+    #optional: watchdog
+    WatchdogSec=30
+
+    [Install]
+    WantedBy=multi-user.target
+
+## Journald
+
+None of those are integrated with journald, however. The simplest way to integrate with journald is the
+following.
+
+First of all, ensure `ForwardToSyslog=yes` in `/etc/systemd/journald.conf`.
+Then, you need to run circologd as root and bind it [to a special
+address](https://www.freedesktop.org/software/systemd/man/journald.conf.html#Forwarding%20to%20traditional%20syslog%20daemons).
+Ok, you don't strictly _need_ to run it as root, but that's the easiest way to run it.
+Here is a working unit for this:
+
+    [Unit]
+    Description=In-memory logging
+
+    [Service]
+    User=root
+    Group=adm
+    ExecStart=/usr/local/sbin/circologd -syslogd-socket /run/systemd/journal/syslog -buffer-size 2000 -query-socket /run/circolog/query.sock
+
+    [Install]
+    WantedBy=multi-user.target
+
+### journald with socket activation
+
+To run circologd as non-root, while listening on a root-owned socket (`/run/systemd/journal/syslog`) use
+socket activation. Create a unit in `/etc/systemd/system/circolog.service`:
+
+    [Unit]
+    Description=In-memory logging
+
+    [Service]
+    User=nobody
+    Group=nogroup
+    ExecStart=/usr/local/sbin/circologd -syslogd-socket "" -buffer-size 2000 -query-socket /run/circolog/query.sock
+
+    [Install]
+    WantedBy=multi-user.target
+
+Then symlink the `syslog.service` unit to the newly created one:
+
+    ln -sf /etc/systemd/system/circolog.service /etc/systemd/system/syslog.service
+
+and restart the service:
+
+    systemctl daemon-reload
+    systemctl restart syslog.service
+
+Now circolog is activated and receives messages from `journald`.

docs/mkdocs.yml

@@ -0,0 +1,9 @@
+site_name: Circolog
+nav:
+    - Home: index.md
+    - Install: install.md
+    - Queries: query.md
+    - Hacking: hacking.md
+    - Systemd: systemd.md
+repo_url: https://git.lattuga.net/boyska/circolog
+repo_name: 'Repository'

filtering/filter.go

@@ -0,0 +1,58 @@
+// +build !nofilter
+
+package filtering
+
+import (
+	"fmt"
+	"os"
+
+	"git.lattuga.net/boyska/circolog/data"
+	"github.com/araddon/qlbridge/datasource"
+	"github.com/araddon/qlbridge/expr"
+	"github.com/araddon/qlbridge/value"
+	"github.com/araddon/qlbridge/vm"
+)
+
+type ExprValue struct {
+	node       expr.Node
+	expression string
+}
+
+func (e *ExprValue) String() string {
+	if e.node != nil {
+		return e.node.String()
+	} else {
+		return "<Empty Expression>"
+	}
+}
+func (e *ExprValue) Set(value string) error {
+	if value == "" {
+		e.node = nil
+		e.expression = value
+		return nil
+	}
+	ast, err := expr.ParseExpression(value)
+	if err != nil {
+		return err
+	}
+	e.node = ast
+	e.expression = value
+	return nil
+}
+
+// Validate answers the question whether to include a log line or not.
+func (e *ExprValue) Validate(line data.Message) bool {
+	if e.node == nil {
+		return true
+	}
+	context := datasource.NewContextSimpleNative(line)
+	val, ok := vm.Eval(context, e.node)
+	if !ok || val == nil { // errors when evaluating
+		return false
+	}
+	if bv, isBool := val.(value.BoolValue); isBool {
+		return bv.Val()
+	}
+	fmt.Fprintln(os.Stderr, "WARNING: The 'where' expression doesn't return a boolean")
+	return false
+}

filtering/filter_fake.go

@@ -0,0 +1,17 @@
+// +build nofilter
+
+package filtering
+
+type ExprValue struct {
+}
+
+func (e *ExprValue) String() string {
+	return "<filtering disabled>"
+}
+
+func (e *ExprValue) Set(value string) error {
+	return nil
+}
+func (e *ExprValue) Validate(line map[string]interface{}) bool {
+	return true
+}

formatter/format.go

@@ -3,16 +3,18 @@ package formatter
 import (
 	"encoding/json"
 	"fmt"
+	"hash/fnv"
 	"io"
 	"text/template"
 	"time"
 
-	"gopkg.in/mcuadros/go-syslog.v2/format"
+	"git.lattuga.net/boyska/circolog/data"
+	"github.com/mgutz/ansi"
 	"gopkg.in/mgo.v2/bson"
 )
 
 // Formatter is an interface, so that multiple implementations can exist
-type Formatter func(format.LogParts) string
+type Formatter func(data.Message) string
 
 var tmplFuncs template.FuncMap
 var syslogTmpl *template.Template
@@ -25,12 +27,40 @@ func init() {
 		"rfc822": func(dt time.Time) string {
 			return dt.Format(time.RFC822)
 		},
+		"sevName": func(s int) string {
+			names := []string{"emerg ", "alert ", "crit  ", "err   ", "warn  ", "notice", "info  ", "dbg   "}
+			switch {
+			case s < 2: // emerg..alert
+				return ansi.Color(names[s], "red+b")
+			case s < 4: // emerg..err
+				return ansi.Color(names[s], "red")
+			case s < 6: // warn..notice
+				return ansi.Color(names[s], "white+b")
+			case s >= len(names):
+				return "???"
+			default:
+				return names[s]
+			}
+		},
+		"color": func(color, text string) string {
+			return ansi.Color(text, color) // slow; should use colorfunc
+		},
+		"red": ansi.ColorFunc("red+b"),
+		"autoColor": func(s string) string {
+			// from https://weechat.org/blog/post/2011/08/28/Beautify-your-WeeChat
+			palette := []string{"31", "35", "38", "40", "49", "63", "70", "80", "92", "99", "112", "126", "130", "138", "142", "148", "167", "169", "174", "176", "178", "184", "186", "210", "212", "215", "247"}
+			hash := fnv.New32()
+			hash.Write([]byte(s))
+			picked := palette[int(hash.Sum32())%len(palette)]
+			return ansi.Color(s, picked)
+		},
 	}
 	syslogTmpl = template.Must(template.New("syslog").Funcs(tmplFuncs).Parse(
-		"{{rfc822 (index . \"timestamp\")}} {{index . \"hostname\"}} " +
-			"{{index . \"app_name\"}}" +
+		"{{color \"yellow\" (rfc822 (index . \"time\")) }} {{index . \"host\"}} " +
+			"{{index . \"prog\" | autoColor}}" +
 			"{{ if (ne (index . \"proc_id\") \"-\")}}[{{index . \"proc_id\"}}]{{end}}: " +
-			"{{index . \"message\"}}",
+			"{{ sevName (index . \"sev\") }} " +
+			"{{index . \"msg\"}}",
 	))
 }
 
@@ -57,7 +87,7 @@ func (rf Format) String() string {
 	return ""
 }
 
-func (rf Format) WriteFormatted(w io.Writer, msg format.LogParts) error {
+func (rf Format) WriteFormatted(w io.Writer, msg data.Message) error {
 	return WriteFormatted(w, rf, msg)
 }
 
@@ -80,7 +110,7 @@ const (
 	FormatBSON   = iota
 )
 
-func WriteFormatted(w io.Writer, f Format, msg format.LogParts) error {
+func WriteFormatted(w io.Writer, f Format, msg data.Message) error {
 	switch f {
 	case FormatSyslog:
 		return syslogTmpl.Execute(w, msg)

formatter/rfc.go

@@ -0,0 +1,49 @@
+package formatter
+
+import (
+	"errors"
+
+	syslog "gopkg.in/mcuadros/go-syslog.v2"
+	"gopkg.in/mcuadros/go-syslog.v2/format"
+)
+
+// SyslogRFC is the formatter that the server should use
+type SyslogRFC struct{ format.Format }
+
+func (rfc *SyslogRFC) Set(v string) error {
+	newval, err := parseRFCValue(v)
+	if err != nil {
+		return err
+	}
+	rfc.Format = newval
+	return nil
+}
+
+func (rfc *SyslogRFC) String() string {
+	switch {
+	case rfc.Format == syslog.Automatic:
+		return "auto"
+	case rfc.Format == syslog.RFC3164:
+		return "rfc3164"
+	case rfc.Format == syslog.RFC5424:
+		return "rfc5424"
+	}
+	return ""
+}
+
+func parseRFCValue(v string) (format.Format, error) {
+	switch {
+	case v == "rfc3164":
+		return syslog.RFC3164, nil
+	case v == "rfc5424":
+		return syslog.RFC5424, nil
+	case v == "auto":
+		return syslog.Automatic, nil
+	default:
+		return nil, ErrRFCNotSupported
+	}
+}
+
+// ErrRFCNotSupported is raised if the supplied rfc string is
+// not recognized.
+var ErrRFCNotSupported = errors.New("RFC not known")

hub.go

@@ -2,18 +2,23 @@ package circolog
 
 import (
 	"container/ring"
+	"fmt"
+	"os"
 	"time"
 
+	"git.lattuga.net/boyska/circolog/data"
+	"git.lattuga.net/boyska/circolog/filtering"
 	"gopkg.in/mcuadros/go-syslog.v2/format"
 )
 
 // Client represent a client connected via websocket. Its most important field is the messages channel, where
 // new messages are sent.
 type Client struct {
-	Messages chan format.LogParts // only hub should write/close this
+	Messages chan data.Message // only hub should write/close this
 	Options  ClientOptions
 }
 
+// ClientOptions is a struct containing connection options for every reader
 type ClientOptions struct {
 	BacklogLength int  // how many past messages the client wants to receive upon connection
 	Nofollow      bool // if Nofollow is true, the hub will not keep this client permanently. Rather, it will send every message to "Messages" and close the channel. Use this if you want to get the messages one-shot
@@ -31,22 +36,41 @@ type HubCommand int
 const (
 	CommandClear       = iota
 	CommandPauseToggle = iota
+	CommandStatus      = iota
+	CommandNewFilter   = iota
 )
 
 // An HubFullCommand is a Command, complete with arguments
 type HubFullCommand struct {
-	Command HubCommand
+	Command    HubCommand
+	Parameters map[string]interface{}
+	Response   chan CommandResponse
 }
+
 type CommandResponse struct {
 	Value interface{}
 }
 
+// StatusResponse is an implementation of a CommandResponse
+type StatusResponse struct {
+	Size      int    `json:"size"`
+	IsRunning bool   `json:"running"`
+	Filter    string `json:"filter"`
+}
+
+// Status return "paused/unpaused" based on isRunning value
+func (r StatusResponse) Status() string {
+	if r.IsRunning {
+		return "unpaused"
+	}
+	return "paused"
+}
+
 type Hub struct {
 	Register    chan Client
 	Unregister  chan Client
 	LogMessages chan format.LogParts
 	Commands    chan HubFullCommand
-	Responses   chan CommandResponse
 
 	clients map[Client]bool
 	circbuf *ring.Ring
@@ -59,7 +83,6 @@ func NewHub(ringBufSize int) Hub {
 		Unregister:  make(chan Client),
 		LogMessages: make(chan format.LogParts),
 		Commands:    make(chan HubFullCommand),
-		Responses:   make(chan CommandResponse),
 		circbuf:     ring.New(ringBufSize),
 	}
 }
@@ -79,7 +102,7 @@ func (h *Hub) register(cl Client) {
 			item := buf.Value
 			if item != nil {
 				select { // send with short timeout
-				case cl.Messages <- item.(format.LogParts):
+				case cl.Messages <- item.(data.Message):
 					break
 				case <-time.After(500 * time.Millisecond):
 					close(cl.Messages)
@@ -99,6 +122,7 @@ func (h *Hub) register(cl Client) {
 // Run is hub main loop; keeps everything going
 func (h *Hub) Run() {
 	active := true
+	var filter filtering.ExprValue
 	for {
 		select {
 		case cl := <-h.Register:
@@ -110,12 +134,13 @@ func (h *Hub) Run() {
 				delete(h.clients, cl)
 			}
 		case msg := <-h.LogMessages:
-			if active == true {
-				h.circbuf.Value = msg
+			newmsg := data.LogEntryToMessage(msg)
+			if active == true && filter.Validate(newmsg) {
+				h.circbuf.Value = newmsg
 				h.circbuf = h.circbuf.Next()
 				for client := range h.clients {
 					select { // send without blocking
-					case client.Messages <- msg:
+					case client.Messages <- newmsg:
 						break
 					default:
 						break
@@ -123,19 +148,54 @@ func (h *Hub) Run() {
 				}
 			}
 		case cmd := <-h.Commands:
-			if cmd.Command == CommandClear {
+			switch cmd.Command {
+			case CommandClear:
 				h.clear()
-				h.Responses <- CommandResponse{Value: true}
-			}
-			if cmd.Command == CommandPauseToggle {
-				active = !active
-				h.Responses <- CommandResponse{Value: active}
+				cmd.Response <- CommandResponse{Value: true}
+			case CommandPauseToggle:
+				togglePause(cmd.Parameters["waitTime"].(time.Duration), &active)
+				if active {
+					fmt.Print("un")
+				}
+				fmt.Println("paused")
+				cmd.Response <- CommandResponse{Value: active}
+			case CommandStatus:
+				var resp = StatusResponse{
+					Size:      h.circbuf.Len(),
+					IsRunning: active,
+					Filter:    filter.String(),
+				}
+				cmd.Response <- CommandResponse{Value: resp}
+			case CommandNewFilter:
+				if err := filter.Set(cmd.Parameters["where"].(string)); err != nil {
+					cmd.Response <- CommandResponse{Value: map[string]interface{}{
+						"success": false,
+						"error":   err.Error(),
+					}}
+				} else {
+					cmd.Response <- CommandResponse{Value: map[string]interface{}{
+						"success": true,
+						"error":   "",
+					}}
+				}
+
 			}
 		}
 	}
 }
 
-// Clear removes every all elements from the buffer
+func togglePause(waitTime time.Duration, status *bool) {
+	if waitTime != 0 {
+		go func() {
+			time.Sleep(waitTime)
+			fmt.Fprintln(os.Stderr, "toggling again")
+			togglePause(0, status)
+		}()
+	}
+	*status = !*status
+}
+
+// Clear removes all elements from the buffer
 func (h *Hub) clear() {
 	buf := h.circbuf
 	for i := 0; i < buf.Len(); i++ {

hub_test.go

@@ -3,11 +3,12 @@ package circolog
 import (
 	"testing"
 
+	"git.lattuga.net/boyska/circolog/data"
 	"gopkg.in/mcuadros/go-syslog.v2/format"
 )
 
 func msg(s string) format.LogParts {
-	return format.LogParts{"text": s}
+	return format.LogParts{"content": s}
 }
 
 func hubCount(h Hub) int {
@@ -19,10 +20,10 @@ var DefaultClient ClientOptions = ClientOptions{Nofollow: true, BacklogLength: -
 func hubToArrayOpt(h Hub, opt ClientOptions) []string {
 	r := make([]string, 0)
 	cl := Client{Options: opt}
-	cl.Messages = make(chan format.LogParts)
+	cl.Messages = make(chan data.Message)
 	h.Register <- cl
 	for m := range cl.Messages {
-		r = append(r, m["text"].(string))
+		r = append(r, m["msg"].(string))
 	}
 	return r
 }

release.sh

@@ -0,0 +1,20 @@
+#!/bin/bash
+
+set -u
+
+for goosarch in $(go tool dist list | grep -vw -e aix -e js/wasm -e plan9 -e solaris -e android -e nacl)
+do
+    mkdir -p "build/$goosarch"
+    goos=$(cut -d/ -f 1 <<<$goosarch)
+    goarch=$(cut -d/ -f 2 <<<$goosarch)
+    for cmd in cmd/*; do
+        GOOS=${goos} GOARCH=${goarch} go build -o "build/$goos/$goarch/$(basename $cmd)" ./$cmd
+    done
+done
+
+find build/ -type f|cut -d/ -f 1-3|uniq|while read -r dir; do
+    find $dir/ -type f -executable | xargs sha1sum > $dir/SHA1SUMS.txt
+    # TODO: touch to last commit date maybe
+    find build -exec touch -d @1234567890 {} \;
+    zip -q -X -j -r "circolog-$(git describe --tags --always)-$(cut -d/ -f 2-3 <<<"$dir"|tr / -)" "$dir"
+done

vendor/gopkg.in/mcuadros/go-syslog.v2

@@ -0,0 +1 @@
+Subproject commit 166aad3f993ce4a67bf486e62d637c834c8a8fe6