diff options
Diffstat (limited to 'internal/io')
26 files changed, 1143 insertions, 772 deletions
diff --git a/internal/io/dlog/dlog.go b/internal/io/dlog/dlog.go new file mode 100644 index 0000000..5e0c3a1 --- /dev/null +++ b/internal/io/dlog/dlog.go @@ -0,0 +1,272 @@ +package dlog + +import ( + "context" + "fmt" + "io/ioutil" + "os" + "path/filepath" + "runtime" + "strconv" + "strings" + "sync" + "time" + + "github.com/mimecast/dtail/internal/color/brush" + "github.com/mimecast/dtail/internal/config" + "github.com/mimecast/dtail/internal/io/dlog/loggers" + "github.com/mimecast/dtail/internal/io/pool" + "github.com/mimecast/dtail/internal/protocol" + "github.com/mimecast/dtail/internal/source" +) + +// Client is the log handler for the client packages. +var Client *DLog + +// Server is the log handler for the server packages. +var Server *DLog + +// Common is the log handler for all other packages. +var Common *DLog + +var mutex sync.Mutex +var started bool + +// Start logger(s). +func Start(ctx context.Context, wg *sync.WaitGroup, sourceProcess source.Source) { + mutex.Lock() + defer mutex.Unlock() + + if started { + Common.FatalPanic("Logger already started") + } + + Client = new(sourceProcess, source.Client) + Server = new(sourceProcess, source.Server) + Common = Client + if sourceProcess == source.Server { + Common = Server + } + + var wg2 sync.WaitGroup + wg2.Add(2) + go Client.start(ctx, &wg2) + go Server.start(ctx, &wg2) + + go rotation(ctx) + go func() { + wg2.Wait() + wg.Done() + }() + + started = true +} + +// DLog is the DTail logger. +type DLog struct { + logger loggers.Logger + // Is this a DTail server or client process logging? + sourceProcess source.Source + // Is this a DTail server or client package logging? In serverless mode + // the client can also execute code from the server package. + sourcePackage source.Source + // Max log level to log. + maxLevel level + // Current hostname. + hostname string +} + +// new creates a new DTail logger. +func new(sourceProcess, sourcePackage source.Source) *DLog { + hostname, err := os.Hostname() + if err != nil { + panic(err) + } + logRotation := loggers.NewStrategy(config.Common.LogRotation) + loggerName := config.Common.Logger + level := newLevel(config.Common.LogLevel) + + return &DLog{ + logger: loggers.Factory(sourceProcess.String(), loggerName, logRotation), + sourceProcess: sourceProcess, + sourcePackage: sourcePackage, + maxLevel: level, + hostname: hostname, + } +} + +func (d *DLog) start(ctx context.Context, wg *sync.WaitGroup) { + defer wg.Done() + var wg2 sync.WaitGroup + wg2.Add(1) + d.logger.Start(ctx, &wg2) + <-ctx.Done() + wg2.Wait() +} + +func (d *DLog) log(level level, args []interface{}) string { + if d.maxLevel < level { + return "" + } + sb := pool.BuilderBuffer.Get().(*strings.Builder) + defer pool.RecycleBuilderBuffer(sb) + now := time.Now() + + switch d.sourceProcess { + case source.Client: + sb.WriteString(d.sourcePackage.String()) + sb.WriteString(protocol.FieldDelimiter) + sb.WriteString(d.hostname) + sb.WriteString(protocol.FieldDelimiter) + sb.WriteString(level.String()) + default: + sb.WriteString(level.String()) + sb.WriteString(protocol.FieldDelimiter) + sb.WriteString(now.Format("20060102-150405")) + } + sb.WriteString(protocol.FieldDelimiter) + d.writeArgStrings(sb, args) + + message := sb.String() + if !config.Client.TermColorsEnable || !d.logger.SupportsColors() { + d.logger.Log(now, message) + return message + } + + d.logger.LogWithColors(now, message, brush.Colorfy(message)) + return message +} + +func (d *DLog) writeArgStrings(sb *strings.Builder, args []interface{}) { + for i, arg := range args { + if i > 0 { + sb.WriteString(protocol.FieldDelimiter) + } + switch v := arg.(type) { + case string: + sb.WriteString(v) + case error: + sb.WriteString(v.Error()) + default: + sb.WriteString(fmt.Sprintf("%v", v)) + } + } +} + +// FatalPanic terminates the process with a fatal error. +func (d *DLog) FatalPanic(args ...interface{}) { + d.log(Fatal, args) + d.Flush() + + var sb strings.Builder + d.writeArgStrings(&sb, args) + panic(sb.String()) +} + +// Fatal logs a fatal error. +func (d *DLog) Fatal(args ...interface{}) string { + return d.log(Fatal, args) +} + +// Error logging. +func (d *DLog) Error(args ...interface{}) string { + return d.log(Error, args) +} + +// Warn logs a warning message. +func (d *DLog) Warn(args ...interface{}) string { + return d.log(Warn, args) +} + +// Info logging. +func (d *DLog) Info(args ...interface{}) string { + return d.log(Info, args) +} + +// Verbose logging. +func (d *DLog) Verbose(args ...interface{}) string { + return d.log(Verbose, args) +} + +// Debug logging. +func (d *DLog) Debug(args ...interface{}) string { + return d.log(Debug, args) +} + +// Trace logging. +func (d *DLog) Trace(args ...interface{}) string { + _, file, line, _ := runtime.Caller(1) + args = append(args, fmt.Sprintf("at %s:%d", file, line)) + return d.log(Trace, args) +} + +// Devel used for development purpose only logging (e.g. "print" debugging). +func (d *DLog) Devel(args ...interface{}) string { + _, file, line, _ := runtime.Caller(1) + args = append(args, fmt.Sprintf("at %s:%d", file, line)) + return d.log(Devel, args) +} + +// Raw message logging. +func (d *DLog) Raw(message string) string { + if !config.Client.TermColorsEnable || !d.logger.SupportsColors() { + d.logger.Log(time.Now(), message) + return message + } + d.logger.LogWithColors(time.Now(), message, brush.Colorfy(message)) + return message +} + +// Mapreduce logging. +func (d *DLog) Mapreduce(table string, data map[string]interface{}) string { + args := make([]interface{}, len(data)+1) + + if d.sourceProcess == source.Server { + // level|date-time|process|caller|cpus|goroutines|cgocalls|loadavg|uptime|MAPREDUCE:TABLE|key=value|... + + var loadAvg string + if loadAvgBytes, err := ioutil.ReadFile("/proc/loadavg"); err == nil { + tmp := string(loadAvgBytes) + s := strings.SplitN(tmp, " ", 2) + loadAvg = s[0] + } + + var uptime string + if uptimeBytes, err := ioutil.ReadFile("/proc/uptime"); err == nil { + tmp := string(uptimeBytes) + s := strings.SplitN(tmp, ".", 2) + i, _ := strconv.ParseInt(s[0], 10, 64) + t := time.Duration(i) * time.Second + uptime = fmt.Sprintf("%v", t) + } + + _, file, line, _ := runtime.Caller(1) + args[0] = fmt.Sprintf("%d|%s:%d|%d|%d|%d|%s|%s|MAPREDUCE:%s", + os.Getpid(), + filepath.Base(file), line, + runtime.NumCPU(), + runtime.NumGoroutine(), + runtime.NumCgoCall(), + loadAvg, + uptime, + strings.ToUpper(table)) + } else { + args[0] = fmt.Sprintf("STATS:%s", strings.ToUpper(table)) + } + + i := 1 + for k, v := range data { + args[i] = fmt.Sprintf("%s=%v", k, v) + i++ + } + return d.log(Info, args) +} + +// Flush the log buffers. +func (d *DLog) Flush() { d.logger.Flush() } + +// Pause the logging. +func (d *DLog) Pause() { d.logger.Pause() } + +// Resume the logging. +func (d *DLog) Resume() { d.logger.Resume() } diff --git a/internal/io/dlog/level.go b/internal/io/dlog/level.go new file mode 100644 index 0000000..05d9ed9 --- /dev/null +++ b/internal/io/dlog/level.go @@ -0,0 +1,84 @@ +package dlog + +import ( + "fmt" + "strings" +) + +type level int + +// Available log levels. +const ( + None level = iota + Fatal level = iota + Error level = iota + Warn level = iota + Info level = iota + Default level = iota + Verbose level = iota + Debug level = iota + Devel level = iota + Trace level = iota + All level = iota +) + +var allLevels = []level{Fatal, Error, Warn, Info, Default, Verbose, Debug, + Devel, Trace, All} + +func newLevel(l string) level { + switch strings.ToLower(l) { + case "none": + return None + case "fatal": + return Fatal + case "error": + return Error + case "warn": + return Warn + case "info": + return Info + case "": + fallthrough + case "default": + return Default + case "verbose": + return Verbose + case "debug": + return Debug + case "devel": + return Devel + case "trace": + return Trace + case "all": + return All + } + panic(fmt.Sprintf("Unknown log level %s, must be one of: %v", l, allLevels)) +} + +func (l level) String() string { + switch l { + case None: + return "NONE" + case Fatal: + return "FATAL" + case Error: + return "ERROR" + case Warn: + return "WARN" + case Info: + return "INFO" + case Default: + return "DEFAULT" + case Verbose: + return "VERBOSE" + case Debug: + return "DEBUG" + case Devel: + return "DEVEL" + case Trace: + return "TRACE" + case All: + return "ALL" + } + panic("Unknown log level " + fmt.Sprintf("%d", l)) +} diff --git a/internal/io/dlog/loggers/factory.go b/internal/io/dlog/loggers/factory.go new file mode 100644 index 0000000..a5cc7cf --- /dev/null +++ b/internal/io/dlog/loggers/factory.go @@ -0,0 +1,54 @@ +package loggers + +import ( + "fmt" + "strings" + "sync" +) + +var factoryMap map[string]Logger +var factoryMutex sync.Mutex + +// Factory is there to retrieve a logger based on various settings. +func Factory(sourceName, loggerName string, logRotation Strategy) Logger { + factoryMutex.Lock() + defer factoryMutex.Unlock() + + id := fmt.Sprintf("sourceName:%s,fileBase:%s,loggerName:%s", sourceName, + logRotation.FileBase, loggerName) + if factoryMap == nil { + factoryMap = make(map[string]Logger) + } + + singleton, ok := factoryMap[id] + if !ok { + switch strings.ToLower(loggerName) { + case "none": + singleton = none{} + case "stdout": + singleton = newStdout() + factoryMap[id] = singleton + case "file": + singleton = newFile(logRotation) + factoryMap[id] = singleton + case "fout": + singleton = newFout(logRotation) + factoryMap[id] = singleton + default: + panic(fmt.Sprintf("Unsupported logger type '%s'", loggerName)) + } + } + return singleton +} + +// FactoryRotate invokes a log rotation of all loggers. +func FactoryRotate() { + factoryMutex.Lock() + defer factoryMutex.Unlock() + if factoryMap == nil { + return + } + for _, logger := range factoryMap { + logger.Rotate() + } +} diff --git a/internal/io/dlog/loggers/file.go b/internal/io/dlog/loggers/file.go new file mode 100644 index 0000000..94824fe --- /dev/null +++ b/internal/io/dlog/loggers/file.go @@ -0,0 +1,165 @@ +package loggers + +import ( + "bufio" + "context" + "fmt" + "os" + "runtime" + "sync" + "time" + + "github.com/mimecast/dtail/internal/config" +) + +type fileWriter struct{} + +type fileMessageBuf struct { + now time.Time + message string +} + +type file struct { + bufferCh chan *fileMessageBuf + pauseCh chan struct{} + resumeCh chan struct{} + rotateCh chan struct{} + flushCh chan struct{} + fd *os.File + writer *bufio.Writer + mutex sync.Mutex + started bool + lastFileName string + strategy Strategy +} + +func newFile(strategy Strategy) *file { + return &file{ + bufferCh: make(chan *fileMessageBuf, runtime.NumCPU()*100), + pauseCh: make(chan struct{}), + resumeCh: make(chan struct{}), + rotateCh: make(chan struct{}), + flushCh: make(chan struct{}), + strategy: strategy, + } +} + +func (f *file) Start(ctx context.Context, wg *sync.WaitGroup) { + f.mutex.Lock() + defer func() { + f.started = true + f.mutex.Unlock() + }() + + if f.started { + // Logger already started from another Goroutine. + wg.Done() + return + } + + pause := func(ctx context.Context) { + select { + case <-f.resumeCh: + return + case <-ctx.Done(): + return + } + } + + go func() { + defer wg.Done() + for { + select { + case m := <-f.bufferCh: + f.write(m) + case <-f.pauseCh: + pause(ctx) + case <-f.flushCh: + f.flush() + case <-ctx.Done(): + f.flush() + f.fd.Close() + return + } + } + }() +} + +func (f *file) Log(now time.Time, message string) { + f.bufferCh <- &fileMessageBuf{now, message} +} + +func (f *file) LogWithColors(now time.Time, message, coloredMessage string) { + panic("Colors not supported in file logger") +} + +func (f *file) Pause() { f.pauseCh <- struct{}{} } +func (f *file) Resume() { f.resumeCh <- struct{}{} } +func (f *file) Flush() { f.flushCh <- struct{}{} } + +func (f *file) Rotate() { f.rotateCh <- struct{}{} } +func (*file) SupportsColors() bool { return false } + +func (f *file) write(m *fileMessageBuf) { + select { + case <-f.rotateCh: + // Force re-opening the outfile next time in getWriter. + f.lastFileName = "" + default: + } + + var writer *bufio.Writer + if f.strategy.Rotation == DailyRotation { + writer = f.getWriter(m.now.Format("20060102")) + } else { + writer = f.getWriter(f.strategy.FileBase) + } + + writer.WriteString(m.message) + writer.WriteByte('\n') +} + +func (f *file) getWriter(name string) *bufio.Writer { + if f.lastFileName == name { + return f.writer + } + if _, err := os.Stat(config.Common.LogDir); os.IsNotExist(err) { + if err = os.MkdirAll(config.Common.LogDir, 0755); err != nil { + panic(err) + } + } + + logFile := fmt.Sprintf("%s/%s.log", config.Common.LogDir, name) + newFd, err := os.OpenFile(logFile, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0644) + if err != nil { + panic(err) + } + + // Close old writer. + if f.fd != nil { + f.writer.Flush() + f.fd.Close() + } + // Set new writer. + f.fd = newFd + f.writer = bufio.NewWriterSize(f.fd, 1) + f.lastFileName = name + + return f.writer +} + +func (f *file) flush() { + defer func() { + if f.writer != nil { + f.writer.Flush() + } + }() + for { + select { + case m := <-f.bufferCh: + f.write(m) + default: + return + } + } +} diff --git a/internal/io/dlog/loggers/fout.go b/internal/io/dlog/loggers/fout.go new file mode 100644 index 0000000..60c318d --- /dev/null +++ b/internal/io/dlog/loggers/fout.go @@ -0,0 +1,46 @@ +package loggers + +import ( + "context" + "sync" + "time" +) + +type fout struct { + file *file + stdout *stdout +} + +// Logs to both, a file and stdout +func newFout(strategy Strategy) *fout { + return &fout{file: newFile(strategy), stdout: newStdout()} +} + +func (f *fout) Start(ctx context.Context, wg *sync.WaitGroup) { + go func() { + defer wg.Done() + + var wg2 sync.WaitGroup + wg2.Add(2) + f.file.Start(ctx, &wg2) + f.stdout.Start(ctx, &wg2) + wg2.Wait() + }() +} + +func (f *fout) Log(now time.Time, message string) { + f.stdout.Log(now, message) + f.file.Log(now, message) +} + +func (f *fout) LogWithColors(now time.Time, message, coloredMessage string) { + f.stdout.LogWithColors(now, "", coloredMessage) + f.file.Log(now, message) +} + +func (f *fout) Flush() { f.stdout.Flush(); f.file.Flush() } +func (f *fout) Pause() { f.stdout.Pause(); f.file.Pause() } +func (f *fout) Resume() { f.stdout.Resume(); f.file.Resume() } +func (f *fout) Rotate() { f.file.Rotate() } + +func (fout) SupportsColors() bool { return true } diff --git a/internal/io/dlog/loggers/logger.go b/internal/io/dlog/loggers/logger.go new file mode 100644 index 0000000..d4e85de --- /dev/null +++ b/internal/io/dlog/loggers/logger.go @@ -0,0 +1,19 @@ +package loggers + +import ( + "context" + "sync" + "time" +) + +// Logger is there to plug in your own log implementation. +type Logger interface { + Log(now time.Time, message string) + LogWithColors(now time.Time, message, messageWithColors string) + Start(ctx context.Context, wg *sync.WaitGroup) + Flush() + Pause() + Resume() + Rotate() + SupportsColors() bool +} diff --git a/internal/io/dlog/loggers/none.go b/internal/io/dlog/loggers/none.go new file mode 100644 index 0000000..270027f --- /dev/null +++ b/internal/io/dlog/loggers/none.go @@ -0,0 +1,21 @@ +package loggers + +import ( + "context" + "sync" + "time" +) + +// don't log anything +type none struct{} + +func (none) Start(ctx context.Context, wg *sync.WaitGroup) { wg.Done() } +func (none) Log(now time.Time, message string) {} + +func (none) LogWithColors(now time.Time, message, coloredMessage string) {} + +func (none) Flush() {} +func (none) Pause() {} +func (none) Resume() {} +func (none) Rotate() {} +func (none) SupportsColors() bool { return false } diff --git a/internal/io/dlog/loggers/stdout.go b/internal/io/dlog/loggers/stdout.go new file mode 100644 index 0000000..05485c6 --- /dev/null +++ b/internal/io/dlog/loggers/stdout.go @@ -0,0 +1,54 @@ +package loggers + +import ( + "context" + "fmt" + "sync" + "time" +) + +type stdout struct { + pauseCh chan struct{} + resumeCh chan struct{} + mutex sync.Mutex +} + +func newStdout() *stdout { + return &stdout{ + pauseCh: make(chan struct{}), + resumeCh: make(chan struct{}), + } +} + +func (s *stdout) Start(ctx context.Context, wg *sync.WaitGroup) { + wg.Done() +} + +func (s *stdout) Log(now time.Time, message string) { + s.log(message) +} + +func (s *stdout) LogWithColors(now time.Time, message, coloredMessage string) { + s.log(coloredMessage) +} + +func (s *stdout) log(message string) { + s.mutex.Lock() + defer s.mutex.Unlock() + + select { + case <-s.pauseCh: + // Pause until resumed. + <-s.resumeCh + default: + } + + fmt.Println(message) +} + +func (s *stdout) Pause() { s.pauseCh <- struct{}{} } +func (s *stdout) Resume() { s.resumeCh <- struct{}{} } +func (s *stdout) Flush() {} +func (s *stdout) Rotate() {} + +func (stdout) SupportsColors() bool { return true } diff --git a/internal/io/dlog/loggers/strategy.go b/internal/io/dlog/loggers/strategy.go new file mode 100644 index 0000000..48e7d44 --- /dev/null +++ b/internal/io/dlog/loggers/strategy.go @@ -0,0 +1,35 @@ +package loggers + +import ( + "os" + "path/filepath" + "strings" +) + +// Rotation is the actual strategy used for log rotation.. +type Rotation int + +const ( + // DailyRotation tells DTail to rotate its logs on a daily basis or on SIGHUP. + DailyRotation Rotation = iota + // SignalRotation tells DTail to rotate its logs only on SIGHUP. + SignalRotation Rotation = iota +) + +// Strategy is a pair of the rotation and the file base. +type Strategy struct { + // Rotation is the actual rotation strategy used. + Rotation Rotation + // FileBase can be a name (e.g. "dserver", "dmap") when signal rotation is used. + FileBase string +} + +// NewStrategy returns the stratey based on its name. +func NewStrategy(name string) Strategy { + switch strings.ToLower(name) { + case "daily": + return Strategy{DailyRotation, ""} + default: + return Strategy{SignalRotation, filepath.Base(os.Args[0])} + } +} diff --git a/internal/io/dlog/rotation.go b/internal/io/dlog/rotation.go new file mode 100644 index 0000000..15ce1fd --- /dev/null +++ b/internal/io/dlog/rotation.go @@ -0,0 +1,27 @@ +package dlog + +import ( + "context" + "os" + "os/signal" + "syscall" + + "github.com/mimecast/dtail/internal/io/dlog/loggers" +) + +func rotation(ctx context.Context) { + rotateCh := make(chan os.Signal, 1) + signal.Notify(rotateCh, syscall.SIGHUP) + go func() { + for { + select { + case <-rotateCh: + Common.Debug("Invoking log rotation") + loggers.FactoryRotate() + return + case <-ctx.Done(): + return + } + } + }() +} diff --git a/internal/io/fs/catfile.go b/internal/io/fs/catfile.go index 7f387bc..01c15ba 100644 --- a/internal/io/fs/catfile.go +++ b/internal/io/fs/catfile.go @@ -6,7 +6,9 @@ type CatFile struct { } // NewCatFile returns a new file catter. -func NewCatFile(filePath string, globID string, serverMessages chan<- string, limiter chan struct{}) CatFile { +func NewCatFile(filePath string, globID string, serverMessages chan<- string, + limiter chan struct{}) CatFile { + return CatFile{ readFile: readFile{ filePath: filePath, diff --git a/internal/io/fs/filereader.go b/internal/io/fs/filereader.go index efd410e..b05fd39 100644 --- a/internal/io/fs/filereader.go +++ b/internal/io/fs/filereader.go @@ -8,9 +8,11 @@ import ( "github.com/mimecast/dtail/internal/regex" ) -// FileReader is the interface used on the dtail server to read/cat/grep/mapr... a file. +// FileReader is the interface used on the dtail server to read/cat/grep/mapr... +// a file. type FileReader interface { - Start(ctx context.Context, lContext lcontext.LContext, lines chan<- line.Line, re regex.Regex) error + Start(ctx context.Context, ltx lcontext.LContext, lines chan<- line.Line, + re regex.Regex) error FilePath() string Retry() bool } diff --git a/internal/io/fs/filter.go b/internal/io/fs/filter.go deleted file mode 100644 index c4f605e..0000000 --- a/internal/io/fs/filter.go +++ /dev/null @@ -1,167 +0,0 @@ -package fs - -import ( - "context" - - "github.com/mimecast/dtail/internal/io/line" - "github.com/mimecast/dtail/internal/lcontext" - "github.com/mimecast/dtail/internal/regex" -) - -func (f readFile) filter(ctx context.Context, rawLines <-chan []byte, lines chan<- line.Line, re regex.Regex, lContext lcontext.LContext) { - // Do we have any kind of local context settings? If so then run the more complex - // filterWithLContext method. - if lContext.Has() { - // We can not skip transmitting any lines to the client with a local - // grep context specified. - f.canSkipLines = false - f.filterWithLContext(ctx, rawLines, lines, re, lContext) - return - } - - f.filterWithoutLContext(ctx, rawLines, lines, re) -} - -// Filter log lines matching a given regular expression, however with local grep context. -func (f readFile) filterWithLContext(ctx context.Context, rawLines <-chan []byte, lines chan<- line.Line, re regex.Regex, lContext lcontext.LContext) { - // Scenario 1: Finish once maxCount hits found - maxCount := lContext.MaxCount - processMaxCount := maxCount > 0 - maxReached := false - - // Scenario 2: Print prev. N lines when current line matches. - before := lContext.BeforeContext - processBefore := before > 0 - var beforeBuf chan []byte - if processBefore { - beforeBuf = make(chan []byte, before) - } - - // Screnario 3: Print next N lines when current line matches. - after := 0 - processAfter := lContext.AfterContext > 0 - - for rawLine := range rawLines { - // logger.Debug("rawLine", string(rawLine)) - f.updatePosition() - - if !re.Match(rawLine) { - f.updateLineNotMatched() - - if processAfter && after > 0 { - after-- - myLine := line.Line{Content: rawLine, SourceID: f.globID, Count: f.totalLineCount(), TransmittedPerc: 100} - select { - case lines <- myLine: - case <-ctx.Done(): - return - } - - } else if processBefore { - // Keep last num BeforeContext raw messages. - select { - case beforeBuf <- rawLine: - default: - <-beforeBuf - beforeBuf <- rawLine - } - } - continue - } - - f.updateLineMatched() - - if processAfter { - if maxReached { - return - } - after = lContext.AfterContext - } - - if processBefore { - i := uint64(len(beforeBuf)) - for { - select { - case myRawLine := <-beforeBuf: - myLine := line.Line{Content: myRawLine, SourceID: f.globID, Count: f.totalLineCount() - i, TransmittedPerc: 100} - i-- - select { - case lines <- myLine: - case <-ctx.Done(): - return - } - default: - // beforeBuf is now empty. - } - if len(beforeBuf) == 0 { - break - } - } - } - - line := line.Line{Content: rawLine, SourceID: f.globID, Count: f.totalLineCount(), TransmittedPerc: 100} - - select { - case lines <- line: - if processMaxCount { - maxCount-- - if maxCount == 0 { - if !processAfter || after == 0 { - return - } - // Unfortunatley we have to continue filter, as there might be more lines to print - maxReached = true - } - } - case <-ctx.Done(): - return - } - } -} - -// Filter log lines matching a given regular expression, there is no local grep context specified. -func (f readFile) filterWithoutLContext(ctx context.Context, rawLines <-chan []byte, lines chan<- line.Line, re regex.Regex) { - for { - select { - case rawLine, ok := <-rawLines: - f.updatePosition() - if !ok { - return - } - - if f.lineUntransmittable(rawLine, len(lines), cap(lines), re) { - continue - } - - line := line.Line{Content: rawLine, SourceID: f.globID, Count: f.totalLineCount(), TransmittedPerc: f.transmittedPerc()} - - select { - case lines <- line: - continue - case <-ctx.Done(): - return - } - } - } -} - -func (f readFile) lineUntransmittable(rawLine []byte, length, capacity int, re regex.Regex) bool { - if !re.Match(rawLine) { - f.updateLineNotMatched() - f.updateLineNotTransmitted() - // Regex dosn't match, so not interested in it. - return true - } - f.updateLineMatched() - - // Can we actually send more messages, channel capacity reached? - if f.canSkipLines && length >= capacity { - f.updateLineNotTransmitted() - // Matching, not transmittable - return true - } - f.updateLineTransmitted() - - // Matching, transmittable - return false -} diff --git a/internal/io/fs/permissions/permission.go b/internal/io/fs/permissions/permission.go index cc5dd9b..d621c09 100644 --- a/internal/io/fs/permissions/permission.go +++ b/internal/io/fs/permissions/permission.go @@ -3,12 +3,12 @@ package permissions import ( - "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/dlog" ) // ToRead is to check whether user has read permissions to a given file. func ToRead(user, filePath string) (bool, error) { // Only implemented fo |