open-nomad/client/driver/logging/syslog_server.go

75 lines
1.4 KiB
Go
Raw Normal View History

2016-02-19 21:08:25 +00:00
package logging
2016-02-17 22:48:25 +00:00
import (
"bufio"
"log"
"net"
)
2016-02-19 22:20:21 +00:00
// SyslogServer is a server which listens to syslog messages and parses them
2016-02-17 22:48:25 +00:00
type SyslogServer struct {
listener net.Listener
messages chan *SyslogMessage
parser *DockerLogParser
doneCh chan interface{}
logger *log.Logger
}
2016-02-19 22:20:21 +00:00
// NewSyslogServer creates a new syslog server
2016-02-17 22:48:25 +00:00
func NewSyslogServer(l net.Listener, messages chan *SyslogMessage, logger *log.Logger) *SyslogServer {
parser := NewDockerLogParser(logger)
return &SyslogServer{
listener: l,
messages: messages,
parser: parser,
logger: logger,
doneCh: make(chan interface{}),
}
}
2016-02-19 22:20:21 +00:00
// Start starts accepting syslog connections
2016-02-17 22:48:25 +00:00
func (s *SyslogServer) Start() {
for {
select {
case <-s.doneCh:
2016-02-19 22:20:21 +00:00
s.listener.Close()
2016-02-17 22:48:25 +00:00
return
default:
connection, err := s.listener.Accept()
if err != nil {
s.logger.Printf("[ERROR] logcollector.server: error in accepting connection: %v", err)
continue
}
2016-02-19 22:20:21 +00:00
go s.read(connection)
2016-02-17 22:48:25 +00:00
}
}
}
2016-02-19 22:20:21 +00:00
// read reads the bytes from a connection
func (s *SyslogServer) read(connection net.Conn) {
2016-02-17 22:48:25 +00:00
defer connection.Close()
scanner := bufio.NewScanner(bufio.NewReader(connection))
for {
select {
case <-s.doneCh:
2016-02-20 05:58:44 +00:00
return
2016-02-17 22:48:25 +00:00
default:
}
if scanner.Scan() {
b := scanner.Bytes()
msg := s.parser.Parse(b)
s.messages <- msg
} else {
2016-02-20 05:58:44 +00:00
return
2016-02-17 22:48:25 +00:00
}
}
}
2016-02-19 22:20:21 +00:00
// Shutdown shutsdown the syslog server
2016-02-17 22:48:25 +00:00
func (s *SyslogServer) Shutdown() {
close(s.doneCh)
2016-02-23 17:46:10 +00:00
close(s.messages)
2016-02-17 22:48:25 +00:00
}