README
¶
go-tailer
go-tailer is a Go library designed to help you tail files in a similar fashion to tail -f
Built off the wonderful work by @fstab in the grok_exporter/tailer module.
See the FOSDEM17: Implementing 'tail -f' presentation for how go-tailer works.
Usage
Basic Usage: File Tailer
Here's a basic example of how to use this library to tail log files.
path := "/var/log/messages" // a single file
path := "/usr/local/myapp/logs/" // a directory
path = "C:\\Program Files\\MyApp\\logs\\*.log" // or a file wildcard
// parse the path glob
parsedGlob, err := glob.Parse(path)
if err != nil {
panic(fmt.Sprintf("%q: failed to parse glob: %q", parsedGlob, err))
}
// startup a logrus logger
logger := logrus.New()
// startup the file tailer. RunFileTailer can take many path globs
tailer, err := fswatcher.RunFileTailer([]glob.Glob{parsedGlob}, false, true, logger)
// listen to the go channel for captured lines and do something with them
for line := range tailer.Lines() {
// line.Line contains the line contents
// line.File contains the name of the file that the line was grabbed from
DoSomethingWithLine(line.File, line.Line)
}
Polling Tailer
We recommend using the RunFileTailer, which listens for file system events to trigger tailing actions. But if that isn't working for you, you can fall back to a polling listener to periodically read the file for any new log lines.
// specify how often you want the tailer to check for updates
pollInterval := time.Duration(500 * time.Millisecond)
// startup the polling file tailer
tailer, err := fswatcher.RunPollingFileTailer([]glob.Glob{parsedGlob}, false, true, pollInterval, logger)
// listen to the go channel for captured lines and do something with them
for line := range tailer.Lines() {
DoSomethingWithLine(line.File, line.Line)
}
Other Tailers
Along with reading from files, go-tailer can read from other sources as well.
- Tail stdin (console/shell/standard input): RunStdinTailer
- Tail a Kafka stream: RunKafkaTailer
- Tail a webhook: WebhookTailer
Documentation
¶
Index ¶
- func BufferedTailer(orig fswatcher.FileTailer) fswatcher.FileTailer
- func BufferedTailerWithMetrics(orig fswatcher.FileTailer, bufferLoadMetric BufferLoadMetric, ...) fswatcher.FileTailer
- func InitWebhookTailer(inputConfig *configuration.InputConfig) fswatcher.FileTailer
- func NewLineBuffer() lineBuffer
- func RunKafkaTailer(cfg *configuration.InputConfig) fswatcher.FileTailer
- func RunStdinTailer() fswatcher.FileTailer
- func WebhookHandler() http.Handler
- func WebhookProcessBody(c *configuration.InputConfig, b []byte) []context_string
- type BufferLoadMetric
- type KafkaTailer
- type WebhookTailer
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func BufferedTailer ¶
func BufferedTailer(orig fswatcher.FileTailer) fswatcher.FileTailer
func BufferedTailerWithMetrics ¶
func BufferedTailerWithMetrics(orig fswatcher.FileTailer, bufferLoadMetric BufferLoadMetric, log logrus.FieldLogger, maxLinesInBuffer int) fswatcher.FileTailer
Wrapper around a tailer that consumes the lines channel quickly. The idea is that the original tailer can continue reading lines from the logfile, and does not need to wait until the lines are processed. The number of buffered lines are exposed as a Prometheus metric, if lines are constantly produced faster than they are consumed, we will eventually run out of memory.
--- The buffered tailer prevents the following error (this can be reproduced on Windows, where we don't keep the logfile open):
Example test actions --------------------
Sequence of actions simulated in fileTailer_test:
1) write line a 2) write line b 3) move the old logfile away and create a new logfile 4) write line c
Good case event processing --------------------------
How Events.Process() should process the file system events triggered by the actions above:
1) MODIFIED : Process() reads line a 2) MODIFIED : Process() reads line b 3) MOVED_FROM, CREATED : Process() resets the line reader and seeks the file to position 0 4) MODIFIED : Process() reads line c
Bad case event processing -------------------------
When Events.Process() receives a MODIFIED event, it does not know how many lines have been written. Therefore, it reads all new lines until EOF is reached. If line processing is slow (writing to the lines channel blocks until all grok patterns are processed), we might read 'line b' while we are still processing the first MODIFIED event:
1) MODIFIED : Process() reads 'line a' and 'line b'
Meanwhile, the test continues with steps 3 and 4, moving the logfile away, creating a new logfile, and writing 'line c'. When the tailer receives the second MODIFIED event, it learns that the file has been truncated, seeks to position 0, and reads 'line c'.
2) MODIFIED : Process() detects the truncated file, seeks to position 0, reads 'line c'
The tailer now receives MOVED_FROM, which makes it close the logfile, CREATED, which makes it open the logfile and start reading from position 0:
3) MOVED_FROM, CREATED : seek to position 0, read line c again !!!
When the last MODIFIED event is processed, there are no more changes in the file:
4) MODIFIED : no changes in file
As a result, we read 'line c' two times.
To minimize the risk, use the buffered tailer to make sure file system events are handled as quickly as possible without waiting for the grok patterns to be processed.
func InitWebhookTailer ¶
func InitWebhookTailer(inputConfig *configuration.InputConfig) fswatcher.FileTailer
func NewLineBuffer ¶
func NewLineBuffer() lineBuffer
func RunKafkaTailer ¶
func RunKafkaTailer(cfg *configuration.InputConfig) fswatcher.FileTailer
RunKafkaTailer runs the kafka tailer
func RunStdinTailer ¶
func RunStdinTailer() fswatcher.FileTailer
func WebhookHandler ¶
func WebhookProcessBody ¶
func WebhookProcessBody(c *configuration.InputConfig, b []byte) []context_string
Types ¶
type BufferLoadMetric ¶
type BufferLoadMetric interface {
Start()
Inc() // put a log line into the buffer
Dec() // take a log line from the buffer
Set(value int64) // set the current number of lines in the buffer
Stop()
}
type KafkaTailer ¶
type KafkaTailer struct {
// contains filtered or unexported fields
}
func (KafkaTailer) Close ¶
func (t KafkaTailer) Close()
func (KafkaTailer) Errors ¶
func (t KafkaTailer) Errors() chan fswatcher.Error
func (KafkaTailer) Lines ¶
func (t KafkaTailer) Lines() chan *fswatcher.Line
type WebhookTailer ¶
type WebhookTailer struct {
// contains filtered or unexported fields
}
func (*WebhookTailer) Close ¶
func (t *WebhookTailer) Close()
func (*WebhookTailer) Errors ¶
func (t *WebhookTailer) Errors() chan fswatcher.Error
func (*WebhookTailer) Lines ¶
func (t *WebhookTailer) Lines() chan *fswatcher.Line
func (WebhookTailer) ServeHTTP ¶
func (t WebhookTailer) ServeHTTP(w http.ResponseWriter, r *http.Request)
Source Files
Directories