1// Copyright (c) 2019 FOSS contributors of https://github.com/nxadm/tail
  2// Copyright (c) 2015 HPE Software Inc. All rights reserved.
  3// Copyright (c) 2013 ActiveState Software Inc. All rights reserved.
  4
  5//nxadm/tail provides a Go library that emulates the features of the BSD `tail`
  6//program. The library comes with full support for truncation/move detection as
  7//it is designed to work with log rotation tools. The library works on all
  8//operating systems supported by Go, including POSIX systems like Linux and
  9//*BSD, and MS Windows. Go 1.9 is the oldest compiler release supported.
 10package tail
 11
 12import (
 13	"bufio"
 14	"errors"
 15	"fmt"
 16	"io"
 17	"io/ioutil"
 18	"log"
 19	"os"
 20	"strings"
 21	"sync"
 22	"time"
 23
 24	"github.com/nxadm/tail/ratelimiter"
 25	"github.com/nxadm/tail/util"
 26	"github.com/nxadm/tail/watch"
 27	"gopkg.in/tomb.v1"
 28)
 29
 30var (
 31	// ErrStop is returned when the tail of a file has been marked to be stopped.
 32	ErrStop = errors.New("tail should now stop")
 33)
 34
 35type Line struct {
 36	Text     string    // The contents of the file
 37	Num      int       // The line number
 38	SeekInfo SeekInfo  // SeekInfo
 39	Time     time.Time // Present time
 40	Err      error     // Error from tail
 41}
 42
 43// Deprecated: this function is no longer used internally and it has little of no
 44// use in the API. As such, it will be removed from the API in a future major
 45// release.
 46//
 47// NewLine returns a * pointer to a Line struct.
 48func NewLine(text string, lineNum int) *Line {
 49	return &Line{text, lineNum, SeekInfo{}, time.Now(), nil}
 50}
 51
 52// SeekInfo represents arguments to io.Seek. See: https://golang.org/pkg/io/#SectionReader.Seek
 53type SeekInfo struct {
 54	Offset int64
 55	Whence int
 56}
 57
 58type logger interface {
 59	Fatal(v ...interface{})
 60	Fatalf(format string, v ...interface{})
 61	Fatalln(v ...interface{})
 62	Panic(v ...interface{})
 63	Panicf(format string, v ...interface{})
 64	Panicln(v ...interface{})
 65	Print(v ...interface{})
 66	Printf(format string, v ...interface{})
 67	Println(v ...interface{})
 68}
 69
 70// Config is used to specify how a file must be tailed.
 71type Config struct {
 72	// File-specifc
 73	Location  *SeekInfo // Tail from this location. If nil, start at the beginning of the file
 74	ReOpen    bool      // Reopen recreated files (tail -F)
 75	MustExist bool      // Fail early if the file does not exist
 76	Poll      bool      // Poll for file changes instead of using the default inotify
 77	Pipe      bool      // The file is a named pipe (mkfifo)
 78
 79	// Generic IO
 80	Follow      bool // Continue looking for new lines (tail -f)
 81	MaxLineSize int  // If non-zero, split longer lines into multiple lines
 82
 83	// Optionally, use a ratelimiter (e.g. created by the ratelimiter/NewLeakyBucket function)
 84	RateLimiter *ratelimiter.LeakyBucket
 85
 86	// Optionally use a Logger. When nil, the Logger is set to tail.DefaultLogger.
 87	// To disable logging, set it to tail.DiscardingLogger
 88	Logger logger
 89}
 90
 91type Tail struct {
 92	Filename string     // The filename
 93	Lines    chan *Line // A consumable channel of *Line
 94	Config              // Tail.Configuration
 95
 96	file    *os.File
 97	reader  *bufio.Reader
 98	lineNum int
 99
100	watcher watch.FileWatcher
101	changes *watch.FileChanges
102
103	tomb.Tomb // provides: Done, Kill, Dying
104
105	lk sync.Mutex
106}
107
108var (
109	// DefaultLogger logs to os.Stderr and it is used when Config.Logger == nil
110	DefaultLogger = log.New(os.Stderr, "", log.LstdFlags)
111	// DiscardingLogger can be used to disable logging output
112	DiscardingLogger = log.New(ioutil.Discard, "", 0)
113)
114
115// TailFile begins tailing the file. And returns a pointer to a Tail struct
116// and an error. An output stream is made available via the Tail.Lines
117// channel (e.g. to be looped and printed). To handle errors during tailing,
118// after finishing reading from the Lines channel, invoke the `Wait` or `Err`
119// method on the returned *Tail.
120func TailFile(filename string, config Config) (*Tail, error) {
121	if config.ReOpen && !config.Follow {
122		util.Fatal("cannot set ReOpen without Follow.")
123	}
124
125	t := &Tail{
126		Filename: filename,
127		Lines:    make(chan *Line),
128		Config:   config,
129	}
130
131	// when Logger was not specified in config, use default logger
132	if t.Logger == nil {
133		t.Logger = DefaultLogger
134	}
135
136	if t.Poll {
137		t.watcher = watch.NewPollingFileWatcher(filename)
138	} else {
139		t.watcher = watch.NewInotifyFileWatcher(filename)
140	}
141
142	if t.MustExist {
143		var err error
144		t.file, err = OpenFile(t.Filename)
145		if err != nil {
146			return nil, err
147		}
148	}
149
150	go t.tailFileSync()
151
152	return t, nil
153}
154
155// Tell returns the file's current position, like stdio's ftell() and an error.
156// Beware that this value may not be completely accurate because one line from
157// the chan(tail.Lines) may have been read already.
158func (tail *Tail) Tell() (offset int64, err error) {
159	if tail.file == nil {
160		return
161	}
162	offset, err = tail.file.Seek(0, io.SeekCurrent)
163	if err != nil {
164		return
165	}
166
167	tail.lk.Lock()
168	defer tail.lk.Unlock()
169	if tail.reader == nil {
170		return
171	}
172
173	offset -= int64(tail.reader.Buffered())
174	return
175}
176
177// Stop stops the tailing activity.
178func (tail *Tail) Stop() error {
179	tail.Kill(nil)
180	return tail.Wait()
181}
182
183// StopAtEOF stops tailing as soon as the end of the file is reached. The function
184// returns an error,
185func (tail *Tail) StopAtEOF() error {
186	tail.Kill(errStopAtEOF)
187	return tail.Wait()
188}
189
190var errStopAtEOF = errors.New("tail: stop at eof")
191
192func (tail *Tail) close() {
193	close(tail.Lines)
194	tail.closeFile()
195}
196
197func (tail *Tail) closeFile() {
198	if tail.file != nil {
199		tail.file.Close()
200		tail.file = nil
201	}
202}
203
204func (tail *Tail) reopen() error {
205	tail.closeFile()
206	tail.lineNum = 0
207	for {
208		var err error
209		tail.file, err = OpenFile(tail.Filename)
210		if err != nil {
211			if os.IsNotExist(err) {
212				tail.Logger.Printf("Waiting for %s to appear...", tail.Filename)
213				if err := tail.watcher.BlockUntilExists(&tail.Tomb); err != nil {
214					if err == tomb.ErrDying {
215						return err
216					}
217					return fmt.Errorf("Failed to detect creation of %s: %s", tail.Filename, err)
218				}
219				continue
220			}
221			return fmt.Errorf("Unable to open file %s: %s", tail.Filename, err)
222		}
223		break
224	}
225	return nil
226}
227
228func (tail *Tail) readLine() (string, error) {
229	tail.lk.Lock()
230	line, err := tail.reader.ReadString('\n')
231	tail.lk.Unlock()
232	if err != nil {
233		// Note ReadString "returns the data read before the error" in
234		// case of an error, including EOF, so we return it as is. The
235		// caller is expected to process it if err is EOF.
236		return line, err
237	}
238
239	line = strings.TrimRight(line, "\n")
240
241	return line, err
242}
243
244func (tail *Tail) tailFileSync() {
245	defer tail.Done()
246	defer tail.close()
247
248	if !tail.MustExist {
249		// deferred first open.
250		err := tail.reopen()
251		if err != nil {
252			if err != tomb.ErrDying {
253				tail.Kill(err)
254			}
255			return
256		}
257	}
258
259	// Seek to requested location on first open of the file.
260	if tail.Location != nil {
261		_, err := tail.file.Seek(tail.Location.Offset, tail.Location.Whence)
262		if err != nil {
263			tail.Killf("Seek error on %s: %s", tail.Filename, err)
264			return
265		}
266	}
267
268	tail.openReader()
269
270	// Read line by line.
271	for {
272		// do not seek in named pipes
273		if !tail.Pipe {
274			// grab the position in case we need to back up in the event of a half-line
275			if _, err := tail.Tell(); err != nil {
276				tail.Kill(err)
277				return
278			}
279		}
280
281		line, err := tail.readLine()
282
283		// Process `line` even if err is EOF.
284		if err == nil {
285			cooloff := !tail.sendLine(line)
286			if cooloff {
287				// Wait a second before seeking till the end of
288				// file when rate limit is reached.
289				msg := ("Too much log activity; waiting a second before resuming tailing")
290				offset, _ := tail.Tell()
291				tail.Lines <- &Line{msg, tail.lineNum, SeekInfo{Offset: offset}, time.Now(), errors.New(msg)}
292				select {
293				case <-time.After(time.Second):
294				case <-tail.Dying():
295					return
296				}
297				if err := tail.seekEnd(); err != nil {
298					tail.Kill(err)
299					return
300				}
301			}
302		} else if err == io.EOF {
303			if !tail.Follow {
304				if line != "" {
305					tail.sendLine(line)
306				}
307				return
308			}
309
310			if tail.Follow && line != "" {
311				tail.sendLine(line)
312				if err := tail.seekEnd(); err != nil {
313					tail.Kill(err)
314					return
315				}
316			}
317
318			// When EOF is reached, wait for more data to become
319			// available. Wait strategy is based on the `tail.watcher`
320			// implementation (inotify or polling).
321			err := tail.waitForChanges()
322			if err != nil {
323				if err != ErrStop {
324					tail.Kill(err)
325				}
326				return
327			}
328		} else {
329			// non-EOF error
330			tail.Killf("Error reading %s: %s", tail.Filename, err)
331			return
332		}
333
334		select {
335		case <-tail.Dying():
336			if tail.Err() == errStopAtEOF {
337				continue
338			}
339			return
340		default:
341		}
342	}
343}
344
345// waitForChanges waits until the file has been appended, deleted,
346// moved or truncated. When moved or deleted - the file will be
347// reopened if ReOpen is true. Truncated files are always reopened.
348func (tail *Tail) waitForChanges() error {
349	if tail.changes == nil {
350		pos, err := tail.file.Seek(0, io.SeekCurrent)
351		if err != nil {
352			return err
353		}
354		tail.changes, err = tail.watcher.ChangeEvents(&tail.Tomb, pos)
355		if err != nil {
356			return err
357		}
358	}
359
360	select {
361	case <-tail.changes.Modified:
362		return nil
363	case <-tail.changes.Deleted:
364		tail.changes = nil
365		if tail.ReOpen {
366			// XXX: we must not log from a library.
367			tail.Logger.Printf("Re-opening moved/deleted file %s ...", tail.Filename)
368			if err := tail.reopen(); err != nil {
369				return err
370			}
371			tail.Logger.Printf("Successfully reopened %s", tail.Filename)
372			tail.openReader()
373			return nil
374		}
375		tail.Logger.Printf("Stopping tail as file no longer exists: %s", tail.Filename)
376		return ErrStop
377	case <-tail.changes.Truncated:
378		// Always reopen truncated files (Follow is true)
379		tail.Logger.Printf("Re-opening truncated file %s ...", tail.Filename)
380		if err := tail.reopen(); err != nil {
381			return err
382		}
383		tail.Logger.Printf("Successfully reopened truncated %s", tail.Filename)
384		tail.openReader()
385		return nil
386	case <-tail.Dying():
387		return ErrStop
388	}
389}
390
391func (tail *Tail) openReader() {
392	tail.lk.Lock()
393	if tail.MaxLineSize > 0 {
394		// add 2 to account for newline characters
395		tail.reader = bufio.NewReaderSize(tail.file, tail.MaxLineSize+2)
396	} else {
397		tail.reader = bufio.NewReader(tail.file)
398	}
399	tail.lk.Unlock()
400}
401
402func (tail *Tail) seekEnd() error {
403	return tail.seekTo(SeekInfo{Offset: 0, Whence: io.SeekEnd})
404}
405
406func (tail *Tail) seekTo(pos SeekInfo) error {
407	_, err := tail.file.Seek(pos.Offset, pos.Whence)
408	if err != nil {
409		return fmt.Errorf("Seek error on %s: %s", tail.Filename, err)
410	}
411	// Reset the read buffer whenever the file is re-seek'ed
412	tail.reader.Reset(tail.file)
413	return nil
414}
415
416// sendLine sends the line(s) to Lines channel, splitting longer lines
417// if necessary. Return false if rate limit is reached.
418func (tail *Tail) sendLine(line string) bool {
419	now := time.Now()
420	lines := []string{line}
421
422	// Split longer lines
423	if tail.MaxLineSize > 0 && len(line) > tail.MaxLineSize {
424		lines = util.PartitionString(line, tail.MaxLineSize)
425	}
426
427	for _, line := range lines {
428		tail.lineNum++
429		offset, _ := tail.Tell()
430		select {
431		case tail.Lines <- &Line{line, tail.lineNum, SeekInfo{Offset: offset}, now, nil}:
432		case <-tail.Dying():
433			return true
434		}
435	}
436
437	if tail.Config.RateLimiter != nil {
438		ok := tail.Config.RateLimiter.Pour(uint16(len(lines)))
439		if !ok {
440			tail.Logger.Printf("Leaky bucket full (%v); entering 1s cooloff period.",
441				tail.Filename)
442			return false
443		}
444	}
445
446	return true
447}
448
449// Cleanup removes inotify watches added by the tail package. This function is
450// meant to be invoked from a process's exit handler. Linux kernel may not
451// automatically remove inotify watches after the process exits.
452// If you plan to re-read a file, don't call Cleanup in between.
453func (tail *Tail) Cleanup() {
454	watch.Cleanup(tail.Filename)
455}