luxolus
/
open-nomad
2
0
Fork 0
Code Issues 1 Pull Requests Packages Releases Wiki Activity

Made the syslog server use a buffered chan

This commit is contained in:
Diptanu Choudhury 2016-02-22 16:10:23 -08:00
parent 91e8800b6b
commit 3211e67961
1 changed files with 25 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
client/driver/logging/universal_collector.go
View File

@ -4,6 +4,7 @@ package logging
import ( import (
"fmt" "fmt"
"io"
"log" "log"
"log/syslog" "log/syslog"
"net" "net"
@ -60,6 +61,7 @@ type SyslogCollector struct {
lro *FileRotator lro *FileRotator
lre *FileRotator lre *FileRotator
server *SyslogServer server *SyslogServer
syslogChan chan *SyslogMessage
taskDir string taskDir string
logger *log.Logger logger *log.Logger
@ -67,7 +69,7 @@ type SyslogCollector struct {
// NewSyslogCollector returns an implementation of the SyslogCollector // NewSyslogCollector returns an implementation of the SyslogCollector
func NewSyslogCollector(logger *log.Logger) *SyslogCollector { func NewSyslogCollector(logger *log.Logger) *SyslogCollector {
return &SyslogCollector{logger: logger} return &SyslogCollector{logger: logger, syslogChan: make(chan *SyslogMessage, 2048)}
} }
// LaunchCollector launches a new syslog server and starts writing log lines to // LaunchCollector launches a new syslog server and starts writing log lines to
@ -84,10 +86,8 @@ func (s *SyslogCollector) LaunchCollector(ctx *LogCollectorContext) (*SyslogColl
return nil, err return nil, err
} }
channel := make(chan *SyslogMessage) s.server = NewSyslogServer(l, s.syslogChan, s.logger)
syslogServer := NewSyslogServer(l, channel, s.logger) go s.server.Start()
s.server = syslogServer
go syslogServer.Start()
logFileSize := int64(ctx.LogConfig.MaxFileSizeMB * 1024 * 1024) logFileSize := int64(ctx.LogConfig.MaxFileSizeMB * 1024 * 1024)
path := filepath.Join(s.taskDir, allocdir.TaskLocal) path := filepath.Join(s.taskDir, allocdir.TaskLocal)
@ -106,8 +106,12 @@ func (s *SyslogCollector) LaunchCollector(ctx *LogCollectorContext) (*SyslogColl
} }
s.lre = lre s.lre = lre
go func(channel chan *SyslogMessage) { go s.collectLogs(lre, lro)
for logParts := range channel { return &SyslogCollectorState{Addr: l.Addr().String()}, nil
}
func (s *SyslogCollector) collectLogs(we io.Writer, wo io.Writer) {
for logParts := range s.syslogChan {
// If the severity of the log line is err then we write to stderr // If the severity of the log line is err then we write to stderr
// otherwise all messages go to stdout // otherwise all messages go to stdout
if logParts.Severity == syslog.LOG_ERR { if logParts.Severity == syslog.LOG_ERR {
@ -118,13 +122,12 @@ func (s *SyslogCollector) LaunchCollector(ctx *LogCollectorContext) (*SyslogColl
s.lro.Write([]byte{'\n'}) s.lro.Write([]byte{'\n'})
} }
} }
}(channel)
return &SyslogCollectorState{Addr: l.Addr().String()}, nil
} }
// Exit kills the syslog server // Exit kills the syslog server
func (s *SyslogCollector) Exit() error { func (s *SyslogCollector) Exit() error {
s.server.Shutdown() s.server.Shutdown()
close(s.syslogChan)
return nil return nil
} }