aboutsummaryrefslogtreecommitdiff
path: root/vendor/github.com/nxadm/tail/watch
diff options
context:
space:
mode:
Diffstat (limited to 'vendor/github.com/nxadm/tail/watch')
-rw-r--r--vendor/github.com/nxadm/tail/watch/filechanges.go36
-rw-r--r--vendor/github.com/nxadm/tail/watch/inotify.go135
-rw-r--r--vendor/github.com/nxadm/tail/watch/inotify_tracker.go248
-rw-r--r--vendor/github.com/nxadm/tail/watch/polling.go118
-rw-r--r--vendor/github.com/nxadm/tail/watch/watch.go20
5 files changed, 557 insertions, 0 deletions
diff --git a/vendor/github.com/nxadm/tail/watch/filechanges.go b/vendor/github.com/nxadm/tail/watch/filechanges.go
new file mode 100644
index 000000000..f80aead9a
--- /dev/null
+++ b/vendor/github.com/nxadm/tail/watch/filechanges.go
@@ -0,0 +1,36 @@
+package watch
+
+type FileChanges struct {
+ Modified chan bool // Channel to get notified of modifications
+ Truncated chan bool // Channel to get notified of truncations
+ Deleted chan bool // Channel to get notified of deletions/renames
+}
+
+func NewFileChanges() *FileChanges {
+ return &FileChanges{
+ make(chan bool, 1), make(chan bool, 1), make(chan bool, 1)}
+}
+
+func (fc *FileChanges) NotifyModified() {
+ sendOnlyIfEmpty(fc.Modified)
+}
+
+func (fc *FileChanges) NotifyTruncated() {
+ sendOnlyIfEmpty(fc.Truncated)
+}
+
+func (fc *FileChanges) NotifyDeleted() {
+ sendOnlyIfEmpty(fc.Deleted)
+}
+
+// sendOnlyIfEmpty sends on a bool channel only if the channel has no
+// backlog to be read by other goroutines. This concurrency pattern
+// can be used to notify other goroutines if and only if they are
+// looking for it (i.e., subsequent notifications can be compressed
+// into one).
+func sendOnlyIfEmpty(ch chan bool) {
+ select {
+ case ch <- true:
+ default:
+ }
+}
diff --git a/vendor/github.com/nxadm/tail/watch/inotify.go b/vendor/github.com/nxadm/tail/watch/inotify.go
new file mode 100644
index 000000000..439921810
--- /dev/null
+++ b/vendor/github.com/nxadm/tail/watch/inotify.go
@@ -0,0 +1,135 @@
+// Copyright (c) 2015 HPE Software Inc. All rights reserved.
+// Copyright (c) 2013 ActiveState Software Inc. All rights reserved.
+
+package watch
+
+import (
+ "fmt"
+ "os"
+ "path/filepath"
+
+ "github.com/nxadm/tail/util"
+
+ "github.com/fsnotify/fsnotify"
+ "gopkg.in/tomb.v1"
+)
+
+// InotifyFileWatcher uses inotify to monitor file changes.
+type InotifyFileWatcher struct {
+ Filename string
+ Size int64
+}
+
+func NewInotifyFileWatcher(filename string) *InotifyFileWatcher {
+ fw := &InotifyFileWatcher{filepath.Clean(filename), 0}
+ return fw
+}
+
+func (fw *InotifyFileWatcher) BlockUntilExists(t *tomb.Tomb) error {
+ err := WatchCreate(fw.Filename)
+ if err != nil {
+ return err
+ }
+ defer RemoveWatchCreate(fw.Filename)
+
+ // Do a real check now as the file might have been created before
+ // calling `WatchFlags` above.
+ if _, err = os.Stat(fw.Filename); !os.IsNotExist(err) {
+ // file exists, or stat returned an error.
+ return err
+ }
+
+ events := Events(fw.Filename)
+
+ for {
+ select {
+ case evt, ok := <-events:
+ if !ok {
+ return fmt.Errorf("inotify watcher has been closed")
+ }
+ evtName, err := filepath.Abs(evt.Name)
+ if err != nil {
+ return err
+ }
+ fwFilename, err := filepath.Abs(fw.Filename)
+ if err != nil {
+ return err
+ }
+ if evtName == fwFilename {
+ return nil
+ }
+ case <-t.Dying():
+ return tomb.ErrDying
+ }
+ }
+ panic("unreachable")
+}
+
+func (fw *InotifyFileWatcher) ChangeEvents(t *tomb.Tomb, pos int64) (*FileChanges, error) {
+ err := Watch(fw.Filename)
+ if err != nil {
+ return nil, err
+ }
+
+ changes := NewFileChanges()
+ fw.Size = pos
+
+ go func() {
+
+ events := Events(fw.Filename)
+
+ for {
+ prevSize := fw.Size
+
+ var evt fsnotify.Event
+ var ok bool
+
+ select {
+ case evt, ok = <-events:
+ if !ok {
+ RemoveWatch(fw.Filename)
+ return
+ }
+ case <-t.Dying():
+ RemoveWatch(fw.Filename)
+ return
+ }
+
+ switch {
+ case evt.Op&fsnotify.Remove == fsnotify.Remove:
+ fallthrough
+
+ case evt.Op&fsnotify.Rename == fsnotify.Rename:
+ RemoveWatch(fw.Filename)
+ changes.NotifyDeleted()
+ return
+
+ //With an open fd, unlink(fd) - inotify returns IN_ATTRIB (==fsnotify.Chmod)
+ case evt.Op&fsnotify.Chmod == fsnotify.Chmod:
+ fallthrough
+
+ case evt.Op&fsnotify.Write == fsnotify.Write:
+ fi, err := os.Stat(fw.Filename)
+ if err != nil {
+ if os.IsNotExist(err) {
+ RemoveWatch(fw.Filename)
+ changes.NotifyDeleted()
+ return
+ }
+ // XXX: report this error back to the user
+ util.Fatal("Failed to stat file %v: %v", fw.Filename, err)
+ }
+ fw.Size = fi.Size()
+
+ if prevSize > 0 && prevSize > fw.Size {
+ changes.NotifyTruncated()
+ } else {
+ changes.NotifyModified()
+ }
+ prevSize = fw.Size
+ }
+ }
+ }()
+
+ return changes, nil
+}
diff --git a/vendor/github.com/nxadm/tail/watch/inotify_tracker.go b/vendor/github.com/nxadm/tail/watch/inotify_tracker.go
new file mode 100644
index 000000000..a94bcd4cb
--- /dev/null
+++ b/vendor/github.com/nxadm/tail/watch/inotify_tracker.go
@@ -0,0 +1,248 @@
+// Copyright (c) 2015 HPE Software Inc. All rights reserved.
+// Copyright (c) 2013 ActiveState Software Inc. All rights reserved.
+
+package watch
+
+import (
+ "log"
+ "os"
+ "path/filepath"
+ "sync"
+ "syscall"
+
+ "github.com/nxadm/tail/util"
+
+ "github.com/fsnotify/fsnotify"
+)
+
+type InotifyTracker struct {
+ mux sync.Mutex
+ watcher *fsnotify.Watcher
+ chans map[string]chan fsnotify.Event
+ done map[string]chan bool
+ watchNums map[string]int
+ watch chan *watchInfo
+ remove chan *watchInfo
+ error chan error
+}
+
+type watchInfo struct {
+ op fsnotify.Op
+ fname string
+}
+
+func (this *watchInfo) isCreate() bool {
+ return this.op == fsnotify.Create
+}
+
+var (
+ // globally shared InotifyTracker; ensures only one fsnotify.Watcher is used
+ shared *InotifyTracker
+
+ // these are used to ensure the shared InotifyTracker is run exactly once
+ once = sync.Once{}
+ goRun = func() {
+ shared = &InotifyTracker{
+ mux: sync.Mutex{},
+ chans: make(map[string]chan fsnotify.Event),
+ done: make(map[string]chan bool),
+ watchNums: make(map[string]int),
+ watch: make(chan *watchInfo),
+ remove: make(chan *watchInfo),
+ error: make(chan error),
+ }
+ go shared.run()
+ }
+
+ logger = log.New(os.Stderr, "", log.LstdFlags)
+)
+
+// Watch signals the run goroutine to begin watching the input filename
+func Watch(fname string) error {
+ return watch(&watchInfo{
+ fname: fname,
+ })
+}
+
+// Watch create signals the run goroutine to begin watching the input filename
+// if call the WatchCreate function, don't call the Cleanup, call the RemoveWatchCreate
+func WatchCreate(fname string) error {
+ return watch(&watchInfo{
+ op: fsnotify.Create,
+ fname: fname,
+ })
+}
+
+func watch(winfo *watchInfo) error {
+ // start running the shared InotifyTracker if not already running
+ once.Do(goRun)
+
+ winfo.fname = filepath.Clean(winfo.fname)
+ shared.watch <- winfo
+ return <-shared.error
+}
+
+// RemoveWatch signals the run goroutine to remove the watch for the input filename
+func RemoveWatch(fname string) error {
+ return remove(&watchInfo{
+ fname: fname,
+ })
+}
+
+// RemoveWatch create signals the run goroutine to remove the watch for the input filename
+func RemoveWatchCreate(fname string) error {
+ return remove(&watchInfo{
+ op: fsnotify.Create,
+ fname: fname,
+ })
+}
+
+func remove(winfo *watchInfo) error {
+ // start running the shared InotifyTracker if not already running
+ once.Do(goRun)
+
+ winfo.fname = filepath.Clean(winfo.fname)
+ shared.mux.Lock()
+ done := shared.done[winfo.fname]
+ if done != nil {
+ delete(shared.done, winfo.fname)
+ close(done)
+ }
+ shared.mux.Unlock()
+
+ shared.remove <- winfo
+ return <-shared.error
+}
+
+// Events returns a channel to which FileEvents corresponding to the input filename
+// will be sent. This channel will be closed when removeWatch is called on this
+// filename.
+func Events(fname string) <-chan fsnotify.Event {
+ shared.mux.Lock()
+ defer shared.mux.Unlock()
+
+ return shared.chans[fname]
+}
+
+// Cleanup removes the watch for the input filename if necessary.
+func Cleanup(fname string) error {
+ return RemoveWatch(fname)
+}
+
+// watchFlags calls fsnotify.WatchFlags for the input filename and flags, creating
+// a new Watcher if the previous Watcher was closed.
+func (shared *InotifyTracker) addWatch(winfo *watchInfo) error {
+ shared.mux.Lock()
+ defer shared.mux.Unlock()
+
+ if shared.chans[winfo.fname] == nil {
+ shared.chans[winfo.fname] = make(chan fsnotify.Event)
+ }
+ if shared.done[winfo.fname] == nil {
+ shared.done[winfo.fname] = make(chan bool)
+ }
+
+ fname := winfo.fname
+ if winfo.isCreate() {
+ // Watch for new files to be created in the parent directory.
+ fname = filepath.Dir(fname)
+ }
+
+ var err error
+ // already in inotify watch
+ if shared.watchNums[fname] == 0 {
+ err = shared.watcher.Add(fname)
+ }
+ if err == nil {
+ shared.watchNums[fname]++
+ }
+ return err
+}
+
+// removeWatch calls fsnotify.RemoveWatch for the input filename and closes the
+// corresponding events channel.
+func (shared *InotifyTracker) removeWatch(winfo *watchInfo) error {
+ shared.mux.Lock()
+
+ ch := shared.chans[winfo.fname]
+ if ch != nil {
+ delete(shared.chans, winfo.fname)
+ close(ch)
+ }
+
+ fname := winfo.fname
+ if winfo.isCreate() {
+ // Watch for new files to be created in the parent directory.
+ fname = filepath.Dir(fname)
+ }
+ shared.watchNums[fname]--
+ watchNum := shared.watchNums[fname]
+ if watchNum == 0 {
+ delete(shared.watchNums, fname)
+ }
+ shared.mux.Unlock()
+
+ var err error
+ // If we were the last ones to watch this file, unsubscribe from inotify.
+ // This needs to happen after releasing the lock because fsnotify waits
+ // synchronously for the kernel to acknowledge the removal of the watch
+ // for this file, which causes us to deadlock if we still held the lock.
+ if watchNum == 0 {
+ err = shared.watcher.Remove(fname)
+ }
+
+ return err
+}
+
+// sendEvent sends the input event to the appropriate Tail.
+func (shared *InotifyTracker) sendEvent(event fsnotify.Event) {
+ name := filepath.Clean(event.Name)
+
+ shared.mux.Lock()
+ ch := shared.chans[name]
+ done := shared.done[name]
+ shared.mux.Unlock()
+
+ if ch != nil && done != nil {
+ select {
+ case ch <- event:
+ case <-done:
+ }
+ }
+}
+
+// run starts the goroutine in which the shared struct reads events from its
+// Watcher's Event channel and sends the events to the appropriate Tail.
+func (shared *InotifyTracker) run() {
+ watcher, err := fsnotify.NewWatcher()
+ if err != nil {
+ util.Fatal("failed to create Watcher")
+ }
+ shared.watcher = watcher
+
+ for {
+ select {
+ case winfo := <-shared.watch:
+ shared.error <- shared.addWatch(winfo)
+
+ case winfo := <-shared.remove:
+ shared.error <- shared.removeWatch(winfo)
+
+ case event, open := <-shared.watcher.Events:
+ if !open {
+ return
+ }
+ shared.sendEvent(event)
+
+ case err, open := <-shared.watcher.Errors:
+ if !open {
+ return
+ } else if err != nil {
+ sysErr, ok := err.(*os.SyscallError)
+ if !ok || sysErr.Err != syscall.EINTR {
+ logger.Printf("Error in Watcher Error channel: %s", err)
+ }
+ }
+ }
+ }
+}
diff --git a/vendor/github.com/nxadm/tail/watch/polling.go b/vendor/github.com/nxadm/tail/watch/polling.go
new file mode 100644
index 000000000..fb1706908
--- /dev/null
+++ b/vendor/github.com/nxadm/tail/watch/polling.go
@@ -0,0 +1,118 @@
+// Copyright (c) 2015 HPE Software Inc. All rights reserved.
+// Copyright (c) 2013 ActiveState Software Inc. All rights reserved.
+
+package watch
+
+import (
+ "os"
+ "runtime"
+ "time"
+
+ "github.com/nxadm/tail/util"
+ "gopkg.in/tomb.v1"
+)
+
+// PollingFileWatcher polls the file for changes.
+type PollingFileWatcher struct {
+ Filename string
+ Size int64
+}
+
+func NewPollingFileWatcher(filename string) *PollingFileWatcher {
+ fw := &PollingFileWatcher{filename, 0}
+ return fw
+}
+
+var POLL_DURATION time.Duration
+
+func (fw *PollingFileWatcher) BlockUntilExists(t *tomb.Tomb) error {
+ for {
+ if _, err := os.Stat(fw.Filename); err == nil {
+ return nil
+ } else if !os.IsNotExist(err) {
+ return err
+ }
+ select {
+ case <-time.After(POLL_DURATION):
+ continue
+ case <-t.Dying():
+ return tomb.ErrDying
+ }
+ }
+ panic("unreachable")
+}
+
+func (fw *PollingFileWatcher) ChangeEvents(t *tomb.Tomb, pos int64) (*FileChanges, error) {
+ origFi, err := os.Stat(fw.Filename)
+ if err != nil {
+ return nil, err
+ }
+
+ changes := NewFileChanges()
+ var prevModTime time.Time
+
+ // XXX: use tomb.Tomb to cleanly manage these goroutines. replace
+ // the fatal (below) with tomb's Kill.
+
+ fw.Size = pos
+
+ go func() {
+ prevSize := fw.Size
+ for {
+ select {
+ case <-t.Dying():
+ return
+ default:
+ }
+
+ time.Sleep(POLL_DURATION)
+ fi, err := os.Stat(fw.Filename)
+ if err != nil {
+ // Windows cannot delete a file if a handle is still open (tail keeps one open)
+ // so it gives access denied to anything trying to read it until all handles are released.
+ if os.IsNotExist(err) || (runtime.GOOS == "windows" && os.IsPermission(err)) {
+ // File does not exist (has been deleted).
+ changes.NotifyDeleted()
+ return
+ }
+
+ // XXX: report this error back to the user
+ util.Fatal("Failed to stat file %v: %v", fw.Filename, err)
+ }
+
+ // File got moved/renamed?
+ if !os.SameFile(origFi, fi) {
+ changes.NotifyDeleted()
+ return
+ }
+
+ // File got truncated?
+ fw.Size = fi.Size()
+ if prevSize > 0 && prevSize > fw.Size {
+ changes.NotifyTruncated()
+ prevSize = fw.Size
+ continue
+ }
+ // File got bigger?
+ if prevSize > 0 && prevSize < fw.Size {
+ changes.NotifyModified()
+ prevSize = fw.Size
+ continue
+ }
+ prevSize = fw.Size
+
+ // File was appended to (changed)?
+ modTime := fi.ModTime()
+ if modTime != prevModTime {
+ prevModTime = modTime
+ changes.NotifyModified()
+ }
+ }
+ }()
+
+ return changes, nil
+}
+
+func init() {
+ POLL_DURATION = 250 * time.Millisecond
+}
diff --git a/vendor/github.com/nxadm/tail/watch/watch.go b/vendor/github.com/nxadm/tail/watch/watch.go
new file mode 100644
index 000000000..2e1783ef0
--- /dev/null
+++ b/vendor/github.com/nxadm/tail/watch/watch.go
@@ -0,0 +1,20 @@
+// Copyright (c) 2015 HPE Software Inc. All rights reserved.
+// Copyright (c) 2013 ActiveState Software Inc. All rights reserved.
+
+package watch
+
+import "gopkg.in/tomb.v1"
+
+// FileWatcher monitors file-level events.
+type FileWatcher interface {
+ // BlockUntilExists blocks until the file comes into existence.
+ BlockUntilExists(*tomb.Tomb) error
+
+ // ChangeEvents reports on changes to a file, be it modification,
+ // deletion, renames or truncations. Returned FileChanges group of
+ // channels will be closed, thus become unusable, after a deletion
+ // or truncation event.
+ // In order to properly report truncations, ChangeEvents requires
+ // the caller to pass their current offset in the file.
+ ChangeEvents(*tomb.Tomb, int64) (*FileChanges, error)
+}