deblan-mirror/reaction
1
0
Fork 0
mirror of https://framagit.org/ppom/reaction synced 2024-06-02 16:42:11 +02:00
Code Issues Packages Projects Releases Wiki Activity

logger package! yay

Browse source 
fix #38
This commit is contained in:
ppom 2023-10-12 12:00:00 +02:00
parent 9d44c05a17
commit a70f1ac424
6 changed files with 138 additions and 65 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
app/client.go
View file

@ -5,11 +5,11 @@ import (
"encoding/gob"
"encoding/json"
"fmt"
"log"
"net"
"os"
"regexp"
"framagit.org/ppom/reaction/logger"
"sigs.k8s.io/yaml"
)
@ -31,19 +31,19 @@ type Response struct {
func SendAndRetrieve(data Request) Response {
conn, err := net.Dial("unix", *SocketPath)
if err != nil {
log.Fatalln("Error opening connection top daemon:", err)
logger.Fatalln("Error opening connection top daemon:", err)
}
defer conn.Close()
err = gob.NewEncoder(conn).Encode(data)
if err != nil {
log.Fatalln("Can't send message:", err)
logger.Fatalln("Can't send message:", err)
}
var response Response
err = gob.NewDecoder(conn).Decode(&response)
if err != nil {
log.Fatalln("Invalid answer from daemon:", err)
logger.Fatalln("Invalid answer from daemon:", err)
}
return response
}
@ -81,13 +81,13 @@ func (csf ClientStatusFlush) MarshalJSON() ([]byte, error) {
func usage(err string) {
fmt.Println("Usage: reactionc")
fmt.Println("Usage: reactionc flush <PATTERN>")
log.Fatalln(err)
logger.Fatalln(err)
}
func ClientShow(streamfilter, format string) {
response := SendAndRetrieve(Request{Show, streamfilter})
if response.Err != nil {
log.Fatalln("Received error from daemon:", response.Err)
logger.Fatalln("Received error from daemon:", response.Err)
os.Exit(1)
}
var text []byte
@ -98,7 +98,7 @@ func ClientShow(streamfilter, format string) {
text, err = yaml.Marshal(response.ClientStatus)
}
if err != nil {
log.Fatalln("Failed to convert daemon binary response to text format:", err)
logger.Fatalln("Failed to convert daemon binary response to text format:", err)
}
fmt.Println(string(text))
os.Exit(0)
@ -107,7 +107,7 @@ func ClientShow(streamfilter, format string) {
func ClientFlush(pattern, streamfilter, format string) {
response := SendAndRetrieve(Request{Flush, pattern})
if response.Err != nil {
log.Fatalln("Received error from daemon:", response.Err)
logger.Fatalln("Received error from daemon:", response.Err)
os.Exit(1)
}
var text []byte
@ -118,7 +118,7 @@ func ClientFlush(pattern, streamfilter, format string) {
text, err = yaml.Marshal(ClientStatusFlush(response.ClientStatus))
}
if err != nil {
log.Fatalln("Failed to convert daemon binary response to text format:", err)
logger.Fatalln("Failed to convert daemon binary response to text format:", err)
}
fmt.Println(string(text))
os.Exit(0)

42
app/daemon.go
View file

@ -2,15 +2,15 @@ package app
import (
"bufio"
"syscall"
"log"
"os"
"os/exec"
"os/signal"
"strings"
"sync"
"syscall"
"time"
"framagit.org/ppom/reaction/logger"
)
// Executes a command and channel-send its stdout
@ -21,17 +21,29 @@ func cmdStdout(commandline []string) chan *string {
cmd := exec.Command(commandline[0], commandline[1:]...)
stdout, err := cmd.StdoutPipe()
if err != nil {
log.Fatalln("FATAL couldn't open stdout on command:", err)
logger.Fatalln("couldn't open stdout on command:", err)
}
stderr, err := cmd.StderrPipe()
if err != nil {
logger.Fatalln("couldn't open stderr on command:", err)
}
if err := cmd.Start(); err != nil {
log.Fatalln("FATAL couldn't start command:", err)
logger.Fatalln("couldn't start command:", err)
}
defer stdout.Close()
defer stderr.Close()
func() {
errscan := bufio.NewScanner(stderr)
for errscan.Scan() {
line := errscan.Text()
logger.Println(logger.WARN, "stderr:", line)
}
}()
scanner := bufio.NewScanner(stdout)
for scanner.Scan() {
line := scanner.Text()
lines <- &line
logger.Println(logger.DEBUG, "stdout:", line)
}
close(lines)
}()
@ -57,7 +69,7 @@ func (f *Filter) match(line *string) string {
match := matches[regex.SubexpIndex(f.pattern.name)]
if f.pattern.notAnIgnore(&match) {
log.Printf("INFO %s.%s: match [%v]\n", f.stream.name, f.name, match)
logger.Printf(logger.INFO, "%s.%s: match [%v]\n", f.stream.name, f.name, match)
return match
}
}
@ -81,12 +93,12 @@ func (a *Action) exec(match string) {
computedCommand = append(computedCommand, strings.ReplaceAll(item, a.filter.pattern.nameWithBraces, match))
}
log.Printf("INFO %s.%s.%s: run %s\n", a.filter.stream.name, a.filter.name, a.name, computedCommand)
logger.Printf(logger.INFO, "%s.%s.%s: run %s\n", a.filter.stream.name, a.filter.name, a.name, computedCommand)
cmd := exec.Command(computedCommand[0], computedCommand[1:]...)
if ret := cmd.Run(); ret != nil {
log.Printf("ERROR %s.%s.%s: run %s, code %s\n", a.filter.stream.name, a.filter.name, a.name, computedCommand, ret)
logger.Printf(logger.ERROR, "%s.%s.%s: run %s, code %s\n", a.filter.stream.name, a.filter.name, a.name, computedCommand, ret)
}
}()
}
@ -241,7 +253,7 @@ func matchesManagerHandleMatch(pft PFT) bool {
func StreamManager(s *Stream, endedSignal chan *Stream) {
defer wgStreams.Done()
log.Printf("INFO %s: start %s\n", s.name, s.Cmd)
logger.Printf(logger.INFO, "%s: start %s\n", s.name, s.Cmd)
lines := cmdStdout(s.Cmd)
for {
@ -345,13 +357,13 @@ func Daemon(confFilename string) {
for {
select {
case finishedStream := <-endSignals:
log.Printf("ERROR %s stream finished", finishedStream.name)
logger.Printf(logger.ERROR, "%s stream finished", finishedStream.name)
nbStreamsInExecution--
if nbStreamsInExecution == 0 {
quit()
}
case <-sigs:
log.Printf("INFO Received SIGINT/SIGTERM, exiting")
logger.Printf(logger.INFO, "Received SIGINT/SIGTERM, exiting")
quit()
}
}
@ -360,19 +372,19 @@ func Daemon(confFilename string) {
func quit() {
// send stop to StreamManager·s
close(stopStreams)
log.Println("INFO Waiting for Streams to finish...")
logger.Println(logger.INFO, "Waiting for Streams to finish...")
wgStreams.Wait()
// ActionsManager calls wgActions.Done() when it has launched all pending actions
wgActions.Add(1)
// send stop to ActionsManager
close(stopActions)
// stop all actions
log.Println("INFO Waiting for Actions to finish...")
logger.Println(logger.INFO, "Waiting for Actions to finish...")
wgActions.Wait()
// delete pipe
err := os.Remove(*SocketPath)
if err != nil {
log.Println("Failed to remove socket:", err)
logger.Println(logger.ERROR, "Failed to remove socket:", err)
}
os.Exit(3)

29
app/persist.go
View file

@ -4,9 +4,10 @@ import (
"encoding/gob"
"errors"
"io"
"log"
"os"
"time"
"framagit.org/ppom/reaction/logger"
)
const (
@ -19,10 +20,10 @@ func openDB(path string) (bool, *ReadDB) {
file, err := os.Open(path)
if err != nil {
if errors.Is(err, os.ErrNotExist) {
log.Printf("WARN No DB found at %s. It's ok if this is the first time reaction is running.\n", path)
logger.Printf(logger.WARN, "No DB found at %s. It's ok if this is the first time reaction is running.\n", path)
return true, nil
}
log.Fatalln("FATAL Failed to open DB:", err)
logger.Fatalln("Failed to open DB:", err)
}
return false, &ReadDB{file, gob.NewDecoder(file)}
}
@ -30,7 +31,7 @@ func openDB(path string) (bool, *ReadDB) {
func createDB(path string) *WriteDB {
file, err := os.Create(path)
if err != nil {
log.Fatalln("FATAL Failed to create DB:", err)
logger.Fatalln("Failed to create DB:", err)
}
return &WriteDB{file, gob.NewEncoder(file)}
}
@ -53,11 +54,11 @@ func (c *Conf) manageLogs(logDB *WriteDB, flushDB *WriteDB) {
// let's say 100 000 entries ~ 10 MB
if cpt == 100_000 {
cpt = 0
log.Printf("INFO Rotating database...")
logger.Printf(logger.INFO, "Rotating database...")
logDB.file.Close()
flushDB.file.Close()
logDB, flushDB = c.RotateDB(false)
log.Printf("INFO Rotated database")
logger.Printf(logger.INFO, "Rotated database")
}
}
}
@ -78,10 +79,10 @@ func (c *Conf) RotateDB(startup bool) (*WriteDB, *WriteDB) {
}
doesntExist, flushReadDB = openDB(flushDBName)
if doesntExist {
log.Println("WARN Strange! No flushes db, opening /dev/null instead")
logger.Println(logger.WARN, "Strange! No flushes db, opening /dev/null instead")
doesntExist, flushReadDB = openDB("/dev/null")
if doesntExist {
log.Fatalln("Opening dummy /dev/null failed")
logger.Fatalln("Opening dummy /dev/null failed")
}
}
@ -91,18 +92,18 @@ func (c *Conf) RotateDB(startup bool) (*WriteDB, *WriteDB) {
err = logReadDB.file.Close()
if err != nil {
log.Fatalln("FATAL Failed to close old DB:", err)
logger.Fatalln("Failed to close old DB:", err)
}
// It should be ok to rename an open file
err = os.Rename(logDBNewName, logDBName)
if err != nil {
log.Fatalln("FATAL Failed to replace old DB with new one:", err)
logger.Fatalln("Failed to replace old DB with new one:", err)
}
err = os.Remove(flushDBName)
if err != nil && !errors.Is(err, os.ErrNotExist) {
log.Fatalln("FATAL Failed to delete old DB:", err)
logger.Fatalln("Failed to delete old DB:", err)
}
flushWriteDB = createDB(flushDBName)
@ -116,11 +117,11 @@ func rotateDB(c *Conf, logDec *gob.Decoder, flushDec *gob.Decoder, logEnc *gob.E
defer func() {
for sf, t := range discardedEntries {
if t > 0 {
log.Printf("WARN info discarded %v times from the DBs: stream/filter not found: %s.%s\n", t, sf.s, sf.f)
logger.Printf(logger.WARN, "info discarded %v times from the DBs: stream/filter not found: %s.%s\n", t, sf.s, sf.f)
}
}
if malformedEntries > 0 {
log.Printf("WARN %v malformed entries discarded from the DBs\n", malformedEntries)
logger.Printf(logger.WARN, "%v malformed entries discarded from the DBs\n", malformedEntries)
}
}()
@ -210,6 +211,6 @@ func rotateDB(c *Conf, logDec *gob.Decoder, flushDec *gob.Decoder, logEnc *gob.E
func encodeOrFatal(enc *gob.Encoder, entry LogEntry) {
err := enc.Encode(entry)
if err != nil {
log.Fatalln("FATAL Failed to write to new DB:", err)
logger.Fatalln("Failed to write to new DB:", err)
}
}

19
app/pipe.go
View file

@ -2,12 +2,13 @@ package app
import (
"encoding/gob"
"log"
"net"
"os"
"path"
"sync"
"time"
"framagit.org/ppom/reaction/logger"
)
func genClientStatus(local_actions ActionsMap, local_matches MatchesMap, local_actionsLock, local_matchesLock *sync.Mutex) ClientStatus {
@ -56,19 +57,19 @@ func genClientStatus(local_actions ActionsMap, local_matches MatchesMap, local_a
func createOpenSocket() net.Listener {
err := os.MkdirAll(path.Dir(*SocketPath), 0755)
if err != nil {
log.Fatalln("FATAL Failed to create socket directory")
logger.Fatalln("Failed to create socket directory")
}
_, err = os.Stat(*SocketPath)
if err == nil {
log.Println("WARN socket", SocketPath, "already exists: Is the daemon already running? Deleting.")
logger.Println(logger.WARN, "socket", SocketPath, "already exists: Is the daemon already running? Deleting.")
err = os.Remove(*SocketPath)
if err != nil {
log.Fatalln("FATAL Failed to remove socket:", err)
logger.Fatalln("Failed to remove socket:", err)
}
}
ln, err := net.Listen("unix", *SocketPath)
if err != nil {
log.Fatalln("FATAL Failed to create socket:", err)
logger.Fatalln("Failed to create socket:", err)
}
return ln
}
@ -80,7 +81,7 @@ func SocketManager(streams map[string]*Stream) {
for {
conn, err := ln.Accept()
if err != nil {
log.Println("ERROR Failed to open connection from cli:", err)
logger.Println(logger.ERROR, "Failed to open connection from cli:", err)
continue
}
go func(conn net.Conn) {
@ -90,7 +91,7 @@ func SocketManager(streams map[string]*Stream) {
err := gob.NewDecoder(conn).Decode(&request)
if err != nil {
log.Println("ERROR Invalid Message from cli:", err)
logger.Println(logger.ERROR, "Invalid Message from cli:", err)
return
}
@ -108,13 +109,13 @@ func SocketManager(streams map[string]*Stream) {
var lock sync.Mutex
response.ClientStatus = genClientStatus(<-actionsC.ret, <-matchesC.ret, &lock, &lock)
default:
log.Println("ERROR Invalid Message from cli: unrecognised Request type")
logger.Println(logger.ERROR, "Invalid Message from cli: unrecognised Request type")
return
}
err = gob.NewEncoder(conn).Encode(response)
if err != nil {
log.Println("ERROR Can't respond to cli:", err)
logger.Println(logger.ERROR, "Can't respond to cli:", err)
return
}
}(conn)

37
app/startup.go
View file

@ -3,12 +3,13 @@ package app
import (
"encoding/json"
"fmt"
"log"
"os"
"regexp"
"strings"
"time"
"framagit.org/ppom/reaction/logger"
"github.com/google/go-jsonnet"
)
@ -20,23 +21,23 @@ func (c *Conf) setup() {
pattern.nameWithBraces = fmt.Sprintf("<%s>", pattern.name)
if pattern.Regex == "" {
log.Fatalf("FATAL Bad configuration: pattern's regex %v is empty!", patternName)
logger.Fatalf("Bad configuration: pattern's regex %v is empty!", patternName)
}
compiled, err := regexp.Compile(fmt.Sprintf("^%v$", pattern.Regex))
if err != nil {
log.Fatalf("FATAL Bad configuration: pattern %v doesn't compile!", patternName)
logger.Fatalf("Bad configuration: pattern %v doesn't compile!", patternName)
}
c.Patterns[patternName].Regex = fmt.Sprintf("(?P<%s>%s)", patternName, pattern.Regex)
for _, ignore := range pattern.Ignore {
if !compiled.MatchString(ignore) {
log.Fatalf("FATAL Bad configuration: pattern ignore '%v' doesn't match pattern %v! It should be fixed or removed.", ignore, pattern.nameWithBraces)
logger.Fatalf("Bad configuration: pattern ignore '%v' doesn't match pattern %v! It should be fixed or removed.", ignore, pattern.nameWithBraces)
}
}
}
if len(c.Streams) == 0 {
log.Fatalln("FATAL Bad configuration: no streams configured!")
logger.Fatalln("Bad configuration: no streams configured!")
}
for streamName := range c.Streams {
@ -44,11 +45,11 @@ func (c *Conf) setup() {
stream.name = streamName
if strings.Contains(stream.name, ".") {
log.Fatalln("FATAL Bad configuration: character '.' is not allowed in stream names", stream.name)
logger.Fatalln("Bad configuration: character '.' is not allowed in stream names", stream.name)
}
if len(stream.Filters) == 0 {
log.Fatalln("FATAL Bad configuration: no filters configured in", stream.name)
logger.Fatalln("Bad configuration: no filters configured in", stream.name)
}
for filterName := range stream.Filters {
@ -57,23 +58,23 @@ func (c *Conf) setup() {
filter.name = filterName
if strings.Contains(filter.name, ".") {
log.Fatalln("FATAL Bad configuration: character '.' is not allowed in filter names", filter.name)
logger.Fatalln("Bad configuration: character '.' is not allowed in filter names", filter.name)
}
// Parse Duration
if filter.RetryPeriod == "" {
if filter.Retry > 1 {
log.Fatalln("FATAL Bad configuration: retry but no retry-duration in", stream.name, ".", filter.name)
logger.Fatalln("Bad configuration: retry but no retry-duration in", stream.name, ".", filter.name)
}
} else {
retryDuration, err := time.ParseDuration(filter.RetryPeriod)
if err != nil {
log.Fatalln("FATAL Bad configuration: Failed to parse retry time in", stream.name, ".", filter.name, ":", err)
logger.Fatalln("Bad configuration: Failed to parse retry time in", stream.name, ".", filter.name, ":", err)
}
filter.retryDuration = retryDuration
}
if len(filter.Regex) == 0 {
log.Fatalln("FATAL Bad configuration: no regexes configured in", stream.name, ".", filter.name)
logger.Fatalln("Bad configuration: no regexes configured in", stream.name, ".", filter.name)
}
// Compute Regexes
// Look for Patterns inside Regexes
@ -86,7 +87,7 @@ func (c *Conf) setup() {
} else if filter.pattern == pattern {
// no op
} else {
log.Fatalf(
logger.Fatalf(
"Bad configuration: Can't mix different patterns (%s, %s) in same filter (%s.%s)\n",
filter.pattern.name, patternName, streamName, filterName,
)
@ -101,7 +102,7 @@ func (c *Conf) setup() {
}
if len(filter.Actions) == 0 {
log.Fatalln("FATAL Bad configuration: no actions configured in", stream.name, ".", filter.name)
logger.Fatalln("Bad configuration: no actions configured in", stream.name, ".", filter.name)
}
for actionName := range filter.Actions {
@ -110,17 +111,17 @@ func (c *Conf) setup() {
action.name = actionName
if strings.Contains(action.name, ".") {
log.Fatalln("FATAL Bad configuration: character '.' is not allowed in action names", action.name)
logger.Fatalln("Bad configuration: character '.' is not allowed in action names", action.name)
}
// Parse Duration
if action.After != "" {
afterDuration, err := time.ParseDuration(action.After)
if err != nil {
log.Fatalln("FATAL Bad configuration: Failed to parse after time in ", stream.name, ".", filter.name, ".", action.name, ":", err)
logger.Fatalln("Bad configuration: Failed to parse after time in ", stream.name, ".", filter.name, ".", action.name, ":", err)
}
action.afterDuration = afterDuration
} else if action.OnExit {
log.Fatalln("FATAL Bad configuration: Cannot have `onexit: true` without an `after` directive in", stream.name, ".", filter.name, ".", action.name)
logger.Fatalln("Bad configuration: Cannot have `onexit: true` without an `after` directive in", stream.name, ".", filter.name, ".", action.name)
}
if filter.longuestActionDuration == nil || filter.longuestActionDuration.Milliseconds() < action.afterDuration.Milliseconds() {
filter.longuestActionDuration = &action.afterDuration
@ -134,7 +135,7 @@ func parseConf(filename string) *Conf {
data, err := os.Open(filename)
if err != nil {
log.Fatalln("FATAL Failed to read configuration file:", err)
logger.Fatalln("Failed to read configuration file:", err)
}
var conf Conf
@ -148,7 +149,7 @@ func parseConf(filename string) *Conf {
}
}
if err != nil {
log.Fatalln("FATAL Failed to parse configuration file:", err)
logger.Fatalln("Failed to parse configuration file:", err)
}
conf.setup()

58
logger/log.go Normal file
View file

@ -0,0 +1,58 @@
package logger
import "log"
type Level int
const (
DEBUG = Level(1)
INFO = Level(2)
WARN = Level(3)
ERROR = Level(4)
FATAL = Level(5)
)
func (l Level) String() string {
switch l {
case DEBUG:
return "DEBUG "
case INFO:
return "INFO "
case WARN:
return "WARN "
case ERROR:
return "ERROR "
case FATAL:
return "FATAL "
default:
return "????? "
}
}
var LogLevel Level = 2
func SetLogLevel(level Level) {
LogLevel = level
}
func Println(level Level, args ...any) {
if level >= LogLevel {
log.Println(level, args)
}
}
func Printf(level Level, format string, args ...any) {
if level >= LogLevel {
log.Printf(level.String()+format, args)
}
}
func Fatalln(args ...any) {
level := FATAL
log.Fatalln(level.String(), args)
}
func Fatalf(format string, args ...any) {
level := FATAL
log.Fatalf(level.String()+format, args)
}