Version 0.2.1

CHANGELOG.md

@@ -1,10 +1,16 @@
 ## Changelog
 
-0.1.0
+0.2.1
 -----
-* First initial beta release.
+* Added `GetSyslogMsgDetails` function to compute the syslog facility and severity when used as syslog receiver (when `enable_syslog_format_only` set to `true`)
+* UDP connection is now properly closed when process is stoped.
+* Fixed issue where 0 byte length message was being processed upon closing of the UDP connection.
 
 0.2.0
 -----
 * Added configuration option `json_document_type_schema` which gives ability to optionally enable and enforce a JSON schema validation for JSON format messages.
 * Added option `enable_syslog_format_only`, which gives the ability to run process in a mode that only accepts syslog type messages.  This could be used as a logging replacement for processes that log to local syslog address.
+
+0.1.0
+-----
+* First initial beta release.

beater/udplogbeat.go

@@ -3,8 +3,6 @@ package beater
 import (
 	"fmt"
 	"net"
-	"os"
-	"os/signal"
 	"strings"
 	"time"
 
@@ -24,9 +22,10 @@ type Udplogbeat struct {
 	config             config.Config
 	client             publisher.Client
 	jsonDocumentSchema map[string]gojsonschema.JSONLoader
+	conn               *net.UDPConn
 }
 
-// Creates beater
+// New creates beater
 func New(b *beat.Beat, cfg *common.Config) (beat.Beater, error) {
 	config := config.DefaultConfig
 	if err := cfg.Unpack(&config); err != nil {
@@ -53,14 +52,6 @@ func New(b *beat.Beat, cfg *common.Config) (beat.Beater, error) {
 
 	bt.config.Addr = fmt.Sprintf("127.0.0.1:%d", bt.config.Port)
 
-	go func() {
-		c := make(chan os.Signal, 1)
-		signal.Notify(c, os.Interrupt)
-		<-c
-		logp.Info("Caught interrupt signal, terminating udplogbeat.")
-		os.Exit(0)
-	}()
-
 	return bt, nil
 }
 
@@ -72,6 +63,7 @@ func (bt *Udplogbeat) Run(b *beat.Beat) error {
 
 	addr, err := net.ResolveUDPAddr("udp", bt.config.Addr)
 	l, err := net.ListenUDP(addr.Network(), addr)
+	bt.conn = l
 
 	logp.Info("Listening on %s (UDP)", bt.config.Addr)
 
@@ -94,7 +86,11 @@ func (bt *Udplogbeat) Run(b *beat.Beat) error {
 		now = common.Time(time.Now())
 
 		// Events should be in the format of: [FORMAT]:[ES_TYPE]:[EVENT_DATA]
-		logSize, _, err := l.ReadFrom(udpBuf)
+		logSize, _, err := bt.conn.ReadFrom(udpBuf)
+
+		if logSize == 0 {
+			continue
+		}
 
 		if err != nil {
 			e, ok := err.(net.Error)
@@ -124,8 +120,6 @@ func (bt *Udplogbeat) Run(b *beat.Beat) error {
 			logp.Info("Size, Format, ES Type: %d bytes, %s, %s", logSize, logFormat, logType)
 		}
 
-		//logp.Info("Data: %s...", logData[0:40])
-
 		event = common.MapStr{}
 
 		if logFormat == "json" {
@@ -157,7 +151,16 @@ func (bt *Udplogbeat) Run(b *beat.Beat) error {
 				event["tags"] = []string{"_udplogbeat_jspf"}
 			}
 		} else {
-			event["message"] = logData
+			if bt.config.EnableSyslogFormatOnly {
+				msg, facility, severity, err := udploglib.GetSyslogMsgDetails(logData)
+				if err == nil {
+					event["facility"] = facility
+					event["severity"] = severity
+					event["message"] = msg
+				}
+			} else {
+				event["message"] = logData
+			}
 		}
 
 	SendFailedMsg:
@@ -166,13 +169,15 @@ func (bt *Udplogbeat) Run(b *beat.Beat) error {
 		event["counter"] = counter
 
 		bt.client.PublishEvent(event)
-		//logp.Info("Event sent")
 		counter++
 
 	}
 }
 
 func (bt *Udplogbeat) Stop() {
+	if err := bt.conn.Close(); err != nil {
+		logp.Err("Could not close UDP connection before terminating: %v", err)
+	}
 	bt.client.Close()
 	close(bt.done)
 }

sample_clients/logger.py

@@ -3,7 +3,7 @@
 import json
 
 '''
-
+Sample logging client that writes to the local instance of udplogbeat listening on the configured port
 '''
 
 class Logger:

udploglib/utils.go

@@ -1,12 +1,92 @@
 package udploglib
 
 import (
-	//"errors"
+	"errors"
 	"fmt"
+	"math"
+	"regexp"
+	"strconv"
 	"strings"
-	//"github.com/xeipuuv/gojsonschema"
 )
 
+const (
+	SyslogFacilityKernel = iota
+	SyslogFacilityUser
+	SyslogFacilityMail
+	SyslogFacilitySystemDaemons
+	SyslogFacilitySecurityAuth
+	SyslogFacilityInternalSyslogd
+	SyslogFacilityLinePrinter
+	SyslogFacilityNetworkNews
+	SyslogFacilityUUCP
+	SyslogFacilityClockDaemon
+	SyslogFacilitySecurityAuth2
+	SyslogFacilityFTP
+	SyslogFacilityNTP
+	SyslogFacilityLogAudit
+	SyslogFacilityLogAlert
+	SyslogFacilityClockDaemon2
+	SyslogFacilityLocal0
+	SyslogFacilityLocal1
+	SyslogFacilityLocal2
+	SyslogFacilityLocal3
+	SyslogFacilityLocal4
+	SyslogFacilityLocal5
+	SyslogFacilityLocal6
+	SyslogFacilityLocal7
+)
+
+const (
+	SyslogSeverityEmergency = iota
+	SyslogSeverityAlert
+	SyslogSeverityCritical
+	SyslogSeverityError
+	SyslogSeverityWarning
+	SyslogSeverityNotice
+	SyslogSeverityInformational
+	SyslogSeverityDebug
+)
+
+// SyslogFacilityString is a map containing the textual equivalence of a given facility number
+var SyslogFacilityString = map[int]string{
+	SyslogFacilityKernel:          "kernel",
+	SyslogFacilityUser:            "user",
+	SyslogFacilityMail:            "mail",
+	SyslogFacilitySystemDaemons:   "system daemons",
+	SyslogFacilitySecurityAuth:    "security/auth",
+	SyslogFacilityInternalSyslogd: "internal syslogd",
+	SyslogFacilityLinePrinter:     "line printer",
+	SyslogFacilityNetworkNews:     "network news",
+	SyslogFacilityUUCP:            "uucp",
+	SyslogFacilityClockDaemon:     "clock daemon",
+	SyslogFacilitySecurityAuth2:   "security/auth",
+	SyslogFacilityFTP:             "ftp",
+	SyslogFacilityNTP:             "ntp",
+	SyslogFacilityLogAudit:        "log audit",
+	SyslogFacilityLogAlert:        "log alert",
+	SyslogFacilityClockDaemon2:    "clock daemon",
+	SyslogFacilityLocal0:          "local0",
+	SyslogFacilityLocal1:          "local1",
+	SyslogFacilityLocal2:          "local2",
+	SyslogFacilityLocal3:          "local3",
+	SyslogFacilityLocal4:          "local4",
+	SyslogFacilityLocal5:          "local5",
+	SyslogFacilityLocal6:          "local6",
+	SyslogFacilityLocal7:          "local7",
+}
+
+// SyslogSeverityString is a map containing the textual equivalence of a given severity number
+var SyslogSeverityString = map[int]string{
+	SyslogSeverityEmergency:     "emergency",
+	SyslogSeverityAlert:         "alert",
+	SyslogSeverityCritical:      "critical",
+	SyslogSeverityError:         "error",
+	SyslogSeverityWarning:       "warning",
+	SyslogSeverityNotice:        "notice",
+	SyslogSeverityInformational: "informational",
+	SyslogSeverityDebug:         "debug",
+}
+
 // GetLogItem returns the log entry format, elasticsearch type, message and error (if any)
 func GetLogItem(buf []byte) ([]string, error) {
 
@@ -26,3 +106,28 @@ func GetLogItem(buf []byte) ([]string, error) {
 
 	return []string{strings.TrimSpace(parts[0]), strings.TrimSpace(parts[1]), strings.TrimSpace(parts[2])}, nil
 }
+
+/*
+   The Priority value is calculated by first multiplying the Facility
+   number by 8 and then adding the numerical value of the Severity.
+
+	 Source: https://tools.ietf.org/html/rfc5424 [Page 10]
+*/
+
+// GetSyslogMsgDetails returns the facility and severity of a valid syslog message
+func GetSyslogMsgDetails(syslogMsg string) (string, int, int, error) {
+
+	re := regexp.MustCompile(`^<([0-9]{1,3})>(.*)`)
+	matches := re.FindStringSubmatch(syslogMsg)
+	if len(matches) < 3 {
+		return "", 0, 0, errors.New("Could not extract syslog priority from message")
+	}
+	priorityNum, err := strconv.Atoi(matches[1])
+	if err != nil {
+		return "", 0, 0, nil
+	}
+	severity := int(math.Mod(float64(priorityNum), 8.0))
+	facility := (priorityNum - severity) / 8
+	return matches[2], facility, severity, nil
+
+}