From b38f97d97d88bfb69097e149d112536d18be94a6 Mon Sep 17 00:00:00 2001 From: Martin Geno Date: Thu, 10 Aug 2017 20:11:35 +0200 Subject: [PATCH] syslog reciever complete: sry filter notifier --- cmd/logmania/main.go | 26 +++--- lib/config.go | 6 +- lib/syslog/main.go | 1 + log/client/main.go | 4 +- log/client/{main_test.go => main_deadtest.go} | 0 logmania_example.conf | 6 +- notify/all/internal.go | 4 +- notify/config.go | 83 +++++++++++++++++++ notify/console/main.go | 11 ++- notify/main.go | 2 +- notify/xmpp/internal.go | 8 +- notify/xmpp/main.go | 18 ++-- receive/syslog/internal.go | 34 ++++---- receive/syslog/main.go | 62 +++++++------- 14 files changed, 193 insertions(+), 72 deletions(-) create mode 100644 lib/syslog/main.go rename log/client/{main_test.go => main_deadtest.go} (100%) create mode 100644 notify/config.go diff --git a/cmd/logmania/main.go b/cmd/logmania/main.go index 76b443b..4a3d5fe 100644 --- a/cmd/logmania/main.go +++ b/cmd/logmania/main.go @@ -24,32 +24,38 @@ import ( ) var ( - configPath string - config *lib.Config - notifier notify.Notifier - receiver receive.Receiver - logChannel chan *log.Entry + configPath string + config *lib.Config + notifyConfig *notify.NotifyState + notifier notify.Notifier + receiver receive.Receiver + logChannel chan *log.Entry ) func main() { flag.StringVar(&configPath, "config", "logmania.conf", "config file") flag.Parse() - log.Info("starting logmania") - config, err := lib.ReadConfig(configPath) if config == nil || err != nil { log.Panicf("Could not load '%s' for configuration.", configPath) } - notifier = allNotify.Init(&config.Notify) + notifyConfig := notify.ReadStateFile(config.Notify.StateFile) + go notifyConfig.Saver(config.Notify.StateFile) + + notifier = allNotify.Init(&config.Notify, notifyConfig) log.Save = notifier.Send + logChannel = make(chan *log.Entry) + go func() { for a := range logChannel { - notifier.Send(a) + log.Save(a) } }() + log.Info("starting logmania") + receiver = allReceiver.Init(&config.Receive, logChannel) go receiver.Listen() @@ -91,5 +97,5 @@ func reload() { go receiver.Listen() notifier.Close() - notifier = allNotify.Init(&config.Notify) + notifier = allNotify.Init(&config.Notify, notifyConfig) } diff --git a/lib/config.go b/lib/config.go index a1d8930..b8351b3 100644 --- a/lib/config.go +++ b/lib/config.go @@ -16,7 +16,8 @@ type Config struct { } type NotifyConfig struct { - XMPP struct { + StateFile string `toml:"state_file"` + XMPP struct { Host string `toml:"host"` Username string `toml:"username"` Password string `toml:"password"` @@ -33,7 +34,8 @@ type NotifyConfig struct { type ReceiveConfig struct { Syslog struct { - Bind string `toml:"bind"` + Type string `toml:"type"` + Address string `toml:"address"` } `toml:"syslog"` } diff --git a/lib/syslog/main.go b/lib/syslog/main.go new file mode 100644 index 0000000..7f88add --- /dev/null +++ b/lib/syslog/main.go @@ -0,0 +1 @@ +package syslog diff --git a/log/client/main.go b/log/client/main.go index 6a6e2ab..406925c 100644 --- a/log/client/main.go +++ b/log/client/main.go @@ -1,6 +1,7 @@ -// logger to bind at github.com/genofire/logmania/log.AddLogger to send log entries to logmania server package client +/* logger to bind at github.com/genofire/logmania/log.AddLogger to send log entries to logmania server + import ( "fmt" @@ -80,3 +81,4 @@ func Init(url, token string, AboveLevel log.LogLevel) *Logger { log.AddLogger(LOGGER_NAME, CurrentLogger) return CurrentLogger } +*/ diff --git a/log/client/main_test.go b/log/client/main_deadtest.go similarity index 100% rename from log/client/main_test.go rename to log/client/main_deadtest.go diff --git a/logmania_example.conf b/logmania_example.conf index daa4269..c9c8d43 100644 --- a/logmania_example.conf +++ b/logmania_example.conf @@ -1,2 +1,6 @@ +[notify] +state_file = "/tmp/logmania.state.json" + [receive.syslog] -bind = ":10001" +type = "udp" +address = ":10001" diff --git a/notify/all/internal.go b/notify/all/internal.go index f5a6d64..b00f69c 100644 --- a/notify/all/internal.go +++ b/notify/all/internal.go @@ -12,10 +12,10 @@ type Notifier struct { channelNotify chan *log.Entry } -func Init(config *lib.NotifyConfig) notify.Notifier { +func Init(config *lib.NotifyConfig, state *notify.NotifyState) notify.Notifier { var list []notify.Notifier for _, init := range notify.NotifyRegister { - notify := init(config) + notify := init(config, state) if notify == nil { continue diff --git a/notify/config.go b/notify/config.go new file mode 100644 index 0000000..65134d9 --- /dev/null +++ b/notify/config.go @@ -0,0 +1,83 @@ +package notify + +import ( + "encoding/json" + "os" + "regexp" + "time" + + "github.com/genofire/logmania/log" +) + +type NotifyState struct { + Hostname map[string]string `json:"hostname"` + HostTo map[string][]string `json:"host_to"` + MaxPrioIn map[string]log.LogLevel `json:"maxLevel"` + RegexIn map[string][]string `json:"regexIn"` + regexIn map[string][]*regexp.Regexp `json:"-"` +} + +func (state *NotifyState) SendTo(e *log.Entry) []string { + if to, ok := state.HostTo[e.Hostname]; ok { + var toList []string + for _, toEntry := range to { + if lvl := state.MaxPrioIn[toEntry]; e.Level > lvl { + continue + } + toList = append(toList, toEntry) + } + e.Hostname = state.Hostname[e.Hostname] + return toList + } + return nil +} + +func ReadStateFile(path string) *NotifyState { + var state *NotifyState + if f, err := os.Open(path); err == nil { // transform data to legacy meshviewer + if err = json.NewDecoder(f).Decode(state); err == nil { + log.Info("loaded", len(state.HostTo), "nodes") + state.regexIn = make(map[string][]*regexp.Regexp) + return state + } else { + log.Error("failed to unmarshal nodes:", err) + } + } else { + log.Error("failed to open state notify file: ", path, ":", err) + } + return &NotifyState{ + Hostname: make(map[string]string), + HostTo: make(map[string][]string), + MaxPrioIn: make(map[string]log.LogLevel), + RegexIn: make(map[string][]string), + regexIn: make(map[string][]*regexp.Regexp), + } +} + +func (state *NotifyState) Saver(path string) { + c := time.Tick(time.Minute) + + for range c { + state.SaveJSON(path) + } +} + +// SaveJSON to path +func (state *NotifyState) SaveJSON(outputFile string) { + tmpFile := outputFile + ".tmp" + + f, err := os.OpenFile(tmpFile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644) + if err != nil { + log.Panic(err) + } + + err = json.NewEncoder(f).Encode(state) + if err != nil { + log.Panic(err) + } + + f.Close() + if err := os.Rename(tmpFile, outputFile); err != nil { + log.Panic(err) + } +} diff --git a/notify/console/main.go b/notify/console/main.go index 2b83089..818a749 100644 --- a/notify/console/main.go +++ b/notify/console/main.go @@ -25,7 +25,7 @@ type Notifier struct { ShowTime bool } -func Init(config *lib.NotifyConfig) notify.Notifier { +func Init(config *lib.NotifyConfig, state *notify.NotifyState) notify.Notifier { return &Notifier{ TimeFormat: "2006-01-02 15:04:05", ShowTime: true, @@ -35,12 +35,17 @@ func Init(config *lib.NotifyConfig) notify.Notifier { // handle a log entry (print it on the terminal with color) func (n *Notifier) Send(e *log.Entry) { v := []interface{}{} - format := "[%s] %s" + format := "[%s]" if n.ShowTime { - format = "%s [%s] %s" + format = "%s [%s]" v = append(v, color.LightBlue(time.Now().Format(n.TimeFormat))) } + if e.Hostname != "" { + format = fmt.Sprintf("%s [%%s]", format) + v = append(v, color.Purple(e.Hostname)) + } + format = fmt.Sprintf("%s %%s", format) lvl := e.Level.String() switch e.Level { case log.DebugLevel: diff --git a/notify/main.go b/notify/main.go index ac23265..e85cbda 100644 --- a/notify/main.go +++ b/notify/main.go @@ -12,7 +12,7 @@ type Notifier interface { Close() } -type NotifyInit func(*lib.NotifyConfig) Notifier +type NotifyInit func(*lib.NotifyConfig, *NotifyState) Notifier func AddNotifier(n NotifyInit) { NotifyRegister = append(NotifyRegister, n) diff --git a/notify/xmpp/internal.go b/notify/xmpp/internal.go index efa6814..aaab0ed 100644 --- a/notify/xmpp/internal.go +++ b/notify/xmpp/internal.go @@ -1,7 +1,11 @@ package xmpp -import "github.com/genofire/logmania/log" +import ( + "fmt" + + "github.com/genofire/logmania/log" +) func formatEntry(e *log.Entry) string { - return e.Text + return fmt.Sprintf("[%s] [%s] %s", e.Hostname, e.Level, e.Text) } diff --git a/notify/xmpp/main.go b/notify/xmpp/main.go index 9626777..a2a60ce 100644 --- a/notify/xmpp/main.go +++ b/notify/xmpp/main.go @@ -10,9 +10,10 @@ import ( type Notifier struct { notify.Notifier client *xmpp.Client + state *notify.NotifyState } -func Init(config *lib.NotifyConfig) notify.Notifier { +func Init(config *lib.NotifyConfig, state *notify.NotifyState) notify.Notifier { options := xmpp.Options{ Host: config.XMPP.Host, User: config.XMPP.Username, @@ -27,16 +28,17 @@ func Init(config *lib.NotifyConfig) notify.Notifier { if err != nil { return nil } - return &Notifier{client: client} + return &Notifier{client: client, state: state} } func (n *Notifier) Send(e *log.Entry) { - /*users := - for _, user := range users { - if user.NotifyXMPP && log.LogLevel(e.Level) >= user.NotifyAfterLoglevel { - n.client.SendHtml(xmpp.Chat{Remote: user.XMPP, Type: "chat", Text: formatEntry(e)}) - } - }*/ + to := n.state.SendTo(e) + if to == nil { + return + } + for _, to := range to { + n.client.SendHtml(xmpp.Chat{Remote: to, Type: "chat", Text: formatEntry(e)}) + } } func (n *Notifier) Close() {} diff --git a/receive/syslog/internal.go b/receive/syslog/internal.go index 0a483e9..68c4be2 100644 --- a/receive/syslog/internal.go +++ b/receive/syslog/internal.go @@ -1,8 +1,13 @@ package syslog -import "github.com/genofire/logmania/log" +import ( + "regexp" + "strconv" -var SyslogPriorityMap = map[uint]log.LogLevel{ + "github.com/genofire/logmania/log" +) + +var SyslogPriorityMap = map[int]log.LogLevel{ 0: log.PanicLevel, 1: log.PanicLevel, 2: log.PanicLevel, @@ -13,23 +18,24 @@ var SyslogPriorityMap = map[uint]log.LogLevel{ 7: log.DebugLevel, } -func toLogEntry(logParts map[string]interface{}) *log.Entry { - severityID := uint(logParts["severity"].(int)) - level := SyslogPriorityMap[severityID] +func toLogEntry(msg []byte, from string) *log.Entry { + re := regexp.MustCompile("<([0-9]*)>(.*)") + match := re.FindStringSubmatch(string(msg)) - if _, ok := logParts["content"]; ok { + if len(match) <= 1 { return &log.Entry{ - Level: level, - Hostname: logParts["hostname"].(string), - Service: logParts["tag"].(string), - Text: logParts["content"].(string), + Level: log.DebugLevel, + Text: string(msg), + Hostname: from, } } + v, _ := strconv.Atoi(match[1]) + prio := v % 8 + text := match[2] return &log.Entry{ - Level: level, - Hostname: logParts["hostname"].(string), - Service: logParts["app_name"].(string), - Text: logParts["message"].(string), + Level: SyslogPriorityMap[prio], + Text: text, + Hostname: from, } } diff --git a/receive/syslog/main.go b/receive/syslog/main.go index 19bdde2..feb7382 100644 --- a/receive/syslog/main.go +++ b/receive/syslog/main.go @@ -1,7 +1,7 @@ package syslog import ( - "gopkg.in/mcuadros/go-syslog.v2" + "net" "github.com/genofire/logmania/lib" "github.com/genofire/logmania/log" @@ -9,43 +9,49 @@ import ( ) type Receiver struct { - channel syslog.LogPartsChannel - exportChannel chan *log.Entry - server *syslog.Server receive.Receiver + exportChannel chan *log.Entry + serverSocket *net.UDPConn } func Init(config *lib.ReceiveConfig, exportChannel chan *log.Entry) receive.Receiver { - channel := make(syslog.LogPartsChannel) - handler := syslog.NewChannelHandler(channel) + addr, err := net.ResolveUDPAddr(config.Syslog.Type, config.Syslog.Address) + ln, err := net.ListenUDP(config.Syslog.Type, addr) - server := syslog.NewServer() - server.SetFormat(syslog.RFC5424) - server.SetHandler(handler) - server.ListenUDP(config.Syslog.Bind) - - log.Info("syslog binded to: ", config.Syslog.Bind) - - return &Receiver{ - channel: channel, - server: server, + if err != nil { + log.Error("syslog init ", err) + return nil + } + recv := &Receiver{ + serverSocket: ln, exportChannel: exportChannel, } + + log.Info("syslog init") + + return recv +} + +const maxDataGramSize = 8192 + +func (rc *Receiver) Listen() { + log.Info("syslog listen") + for { + buf := make([]byte, maxDataGramSize) + n, src, err := rc.serverSocket.ReadFromUDP(buf) + if err != nil { + log.Warn("failed to accept connection", err) + continue + } + + raw := make([]byte, n) + copy(raw, buf) + rc.exportChannel <- toLogEntry(raw, src.IP.String()) + } } -func (rc *Receiver) Listen() { - rc.server.Boot() - log.Info("boot syslog") - go func(channel syslog.LogPartsChannel) { - for logParts := range channel { - rc.exportChannel <- toLogEntry(logParts) - } - }(rc.channel) -} - func (rc *Receiver) Close() { - rc.server.Kill() - rc.server.Wait() + rc.serverSocket.Close() } func init() {