try some more tsting

internal/fswatcher/event.go

@@ -0,0 +1,75 @@
+package fswatcher
+
+import (
+	"bytes"
+	"fmt"
+	"strconv"
+	"unsafe"
+
+	"github.com/dominicbreuker/pspy/internal/fswatcher/inotify"
+	"golang.org/x/sys/unix"
+)
+
+var InotifyEvents = map[uint32]string{
+	unix.IN_ACCESS:                          "ACCESS",
+	unix.IN_ATTRIB:                          "ATTRIB",
+	unix.IN_CLOSE_NOWRITE:                   "CLOSE_NOWRITE",
+	unix.IN_CLOSE_WRITE:                     "CLOSE_WRITE",
+	unix.IN_CREATE:                          "CREATE",
+	unix.IN_DELETE:                          "DELETE",
+	unix.IN_DELETE_SELF:                     "DELETE_SELF",
+	unix.IN_MODIFY:                          "MODIFY",
+	unix.IN_MOVED_FROM:                      "MOVED_FROM",
+	unix.IN_MOVED_TO:                        "MOVED_TO",
+	unix.IN_MOVE_SELF:                       "MOVE_SELF",
+	unix.IN_OPEN:                            "OPEN",
+	(unix.IN_ACCESS | unix.IN_ISDIR):        "ACCESS DIR",
+	(unix.IN_ATTRIB | unix.IN_ISDIR):        "ATTRIB DIR",
+	(unix.IN_CLOSE_NOWRITE | unix.IN_ISDIR): "CLOSE_NOWRITE DIR",
+	(unix.IN_CLOSE_WRITE | unix.IN_ISDIR):   "CLOSE_WRITE DIR",
+	(unix.IN_CREATE | unix.IN_ISDIR):        "CREATE DIR",
+	(unix.IN_DELETE | unix.IN_ISDIR):        "DELETE DIR",
+	(unix.IN_DELETE_SELF | unix.IN_ISDIR):   "DELETE_SELF DIR",
+	(unix.IN_MODIFY | unix.IN_ISDIR):        "MODIFY DIR",
+	(unix.IN_MOVED_FROM | unix.IN_ISDIR):    "MOVED_FROM DIR",
+	(unix.IN_MOVE_SELF | unix.IN_ISDIR):     "MODE_SELF DIR",
+	(unix.IN_OPEN | unix.IN_ISDIR):          "OPEN DIR",
+}
+
+func parseEvents(i *inotify.Inotify, dataCh chan []byte, eventCh chan string, errCh chan error) {
+	for buf := range dataCh {
+		n := len(buf)
+		if n < unix.SizeofInotifyEvent {
+			errCh <- fmt.Errorf("Inotify event parser: incomplete read: n=%d", n)
+			continue
+		}
+
+		var ptr uint32
+		var name string
+		for ptr <= uint32(n-unix.SizeofInotifyEvent) {
+			sys := (*unix.InotifyEvent)(unsafe.Pointer(&buf[ptr]))
+			ptr += unix.SizeofInotifyEvent
+
+			watcher, ok := i.Watchers[int(sys.Wd)]
+			if !ok {
+				errCh <- fmt.Errorf("Inotify event parser: unknown watcher ID: %d", sys.Wd)
+				continue
+			}
+			name = watcher.Dir + "/"
+			if sys.Len > 0 && len(buf) >= int(ptr+sys.Len) {
+				name += string(bytes.TrimRight(buf[ptr:ptr+sys.Len], "\x00"))
+				ptr += sys.Len
+			}
+
+			eventCh <- formatEvent(name, sys.Mask)
+		}
+	}
+}
+
+func formatEvent(name string, mask uint32) string {
+	op, ok := InotifyEvents[mask]
+	if !ok {
+		op = strconv.FormatInt(int64(mask), 2)
+	}
+	return fmt.Sprintf("%20s | %s", op, name)
+}

internal/fswatcher/inotify.go

@@ -0,0 +1,58 @@
+package fswatcher
+
+import (
+	"fmt"
+
+	"golang.org/x/sys/unix"
+)
+
+type Inotify struct {
+	fd       int
+	watchers map[int]*watcher
+}
+
+func NewInotify() (*Inotify, error) {
+	fd, errno := unix.InotifyInit1(unix.IN_CLOEXEC)
+	if fd == -1 {
+		return nil, fmt.Errorf("Can't init inotify: %d", errno)
+	}
+
+	i := &Inotify{
+		fd:       fd,
+		watchers: make(map[int]*watcher),
+	}
+
+	return i, nil
+}
+
+func (i *Inotify) Watch(dir string) error {
+	w, err := newWatcher(i.fd, dir)
+	if err != nil {
+		return fmt.Errorf("creating watcher: %v", err)
+	}
+	i.watchers[w.wd] = w
+	return nil
+}
+
+func (i *Inotify) Close() error {
+	if err := unix.Close(i.fd); err != nil {
+		return fmt.Errorf("closing inotify file descriptor: %v", err)
+	}
+	return nil
+}
+
+func (i *Inotify) NumWatchers() int {
+	return len(i.watchers)
+}
+
+func (i *Inotify) String() string {
+	if len(i.watchers) < 20 {
+		dirs := make([]string, 0)
+		for _, w := range i.watchers {
+			dirs = append(dirs, w.dir)
+		}
+		return fmt.Sprintf("Watching: %v", dirs)
+	} else {
+		return fmt.Sprintf("Watching %d directories", len(i.watchers))
+	}
+}

internal/fswatcher/inotify/inotify.go

@@ -0,0 +1,67 @@
+package inotify
+
+import (
+	"fmt"
+)
+
+type InotifySyscalls interface {
+	Init() (int, error)
+	AddWatch(int, string) (int, error)
+	Close(int) error
+}
+
+type Inotify struct {
+	FD       int
+	Watchers map[int]*Watcher
+	sys      InotifySyscalls
+}
+
+func NewInotify(isys InotifySyscalls) (*Inotify, error) {
+	fd, err := isys.Init()
+	if err != nil {
+		return nil, fmt.Errorf("initializing inotify: %v", err)
+	}
+
+	i := &Inotify{
+		FD:       fd,
+		Watchers: make(map[int]*Watcher),
+		sys:      isys,
+	}
+
+	return i, nil
+}
+
+func (i *Inotify) Watch(dir string) error {
+	wd, err := i.sys.AddWatch(i.FD, dir)
+	if err != nil {
+		return fmt.Errorf("adding watcher on %s: %v", dir, err)
+	}
+	i.Watchers[wd] = &Watcher{
+		WD:  wd,
+		Dir: dir,
+	}
+	return nil
+}
+
+func (i *Inotify) Close() error {
+	if err := i.sys.Close(i.FD); err != nil {
+		return fmt.Errorf("closing inotify file descriptor: %v", err)
+	}
+	return nil
+}
+
+func (i *Inotify) NumWatchers() int {
+	return len(i.Watchers)
+}
+
+func (i *Inotify) String() string {
+	if len(i.Watchers) < 20 {
+		dirs := make([]string, 0)
+		for _, w := range i.Watchers {
+			dirs = append(dirs, w.Dir)
+		}
+		return fmt.Sprintf("Watching: %v", dirs)
+	} else {
+		return fmt.Sprintf("Watching %d directories", len(i.Watchers))
+	}
+}

internal/fswatcher/inotify/inotify_test.go

@@ -0,0 +1,49 @@
+package inotify
+
+import (
+	"errors"
+	"testing"
+)
+
+func TestNewInotify(t *testing.T) {
+	mis := &MockInotifySyscalls{fd: 1}
+
+	i, err := NewInotify(mis)
+	if err != nil {
+		t.Fatalf("Unexpected error")
+	}
+	if i.FD != mis.fd {
+		t.Fatalf("Did not set FD of inotify object")
+	}
+}
+
+func TestNewInotifyError(t *testing.T) {
+	mis := &MockInotifySyscalls{fd: -1}
+
+	_, err := NewInotify(mis)
+	if err == nil || err.Error() != "initializing inotify: syscall error" {
+		t.Fatalf("Expected syscall error but did not get: %v", err)
+	}
+}
+
+// mock
+
+type MockInotifySyscalls struct {
+	fd int
+}
+
+func (mis *MockInotifySyscalls) Init() (int, error) {
+	if mis.fd >= 0 {
+		return mis.fd, nil
+	} else {
+		return -1, errors.New("syscall error")
+	}
+}
+
+func (mis *MockInotifySyscalls) AddWatch(fd int, dir string) (int, error) {
+	return 2, nil
+}
+
+func (mis *MockInotifySyscalls) Close(fd int) error {
+	return nil
+}

internal/fswatcher/inotify/sys/syscalls.go

@@ -0,0 +1,36 @@
+// +build linux
+
+package sys
+
+import (
+	"fmt"
+
+	"golang.org/x/sys/unix"
+)
+
+const events = unix.IN_ALL_EVENTS
+
+type InotifySyscallsUNIX struct{}
+
+func (isu *InotifySyscallsUNIX) Init() (int, error) {
+	fd, errno := unix.InotifyInit1(unix.IN_CLOEXEC)
+	if fd < 0 {
+		return fd, fmt.Errorf("errno: %d", errno)
+	}
+	return fd, nil
+}
+
+func (isu *InotifySyscallsUNIX) AddWatch(fd int, dir string) (int, error) {
+	wd, errno := unix.InotifyAddWatch(fd, dir, events)
+	if wd < 0 {
+		return wd, fmt.Errorf("errno: %d", errno)
+	}
+	return wd, nil
+}
+
+func (isu *InotifySyscallsUNIX) Close(fd int) error {
+	if err := unix.Close(fd); err != nil {
+		return err
+	}
+	return nil
+}

internal/fswatcher/inotify/sys/syscalls_test.go

@@ -0,0 +1,45 @@
+// +build linux
+
+package sys
+
+import (
+	"testing"
+)
+
+func TestSyscalls(t *testing.T) {
+	is := &InotifySyscallsUNIX{}
+
+	fd, err := is.Init()
+	if err != nil {
+		t.Fatalf("Unexpected error for inotify init: %v", err)
+	}
+
+	_, err = is.AddWatch(fd, "testdata")
+	if err != nil {
+		t.Fatalf("Unexpected error adding watch to dir 'testdata': %v", err)
+	}
+
+	err = is.Close(fd)
+	if err != nil {
+		t.Fatalf("Unexpected error closing inotify: %v", err)
+	}
+}
+
+func TestSyscallsError(t *testing.T) {
+	is := &InotifySyscallsUNIX{}
+
+	fd, err := is.Init()
+	if err != nil {
+		t.Fatalf("Unexpected error for inotify init: %v", err)
+	}
+
+	_, err = is.AddWatch(fd, "non-existing-dir")
+	if err == nil || err.Error() != "errno: 2" {
+		t.Fatalf("Expected errno 2 for non-existing-dir but got: %v", err)
+	}
+
+	err = is.Close(fd)
+	if err != nil {
+		t.Fatalf("Unexpected error closing inotify: %v", err)
+	}
+}

internal/fswatcher/inotify/watcher.go

@@ -0,0 +1,30 @@
+package inotify
+
+import (
+	"fmt"
+	"io/ioutil"
+	"strconv"
+	"strings"
+)
+
+const maximumWatchersFile = "/proc/sys/fs/inotify/max_user_watches"
+
+type Watcher struct {
+	WD  int
+	Dir string
+}
+
+func GetLimit() (int, error) {
+	b, err := ioutil.ReadFile(maximumWatchersFile)
+	if err != nil {
+		return 0, fmt.Errorf("reading from %s: %v", maximumWatchersFile, err)
+	}
+
+	s := strings.TrimSpace(string(b))
+	m, err := strconv.Atoi(s)
+	if err != nil {
+		return 0, fmt.Errorf("converting to integer: %v", err)
+	}
+
+	return m, nil
+}

internal/fswatcher/observer.go

@@ -0,0 +1,24 @@
+package fswatcher
+
+import (
+	"fmt"
+
+	"github.com/dominicbreuker/pspy/internal/fswatcher/inotify"
+	"golang.org/x/sys/unix"
+)
+
+func Observe(i *inotify.Inotify, triggerCh chan struct{}, dataCh chan []byte, errCh chan error) {
+	buf := make([]byte, 5*unix.SizeofInotifyEvent)
+
+	for {
+		n, errno := unix.Read(i.FD, buf)
+		if n == -1 {
+			errCh <- fmt.Errorf("reading from inotify fd: errno: %d", errno)
+			return
+		}
+		triggerCh <- struct{}{}
+		bufCopy := make([]byte, n)
+		copy(bufCopy, buf)
+		dataCh <- bufCopy
+	}
+}

internal/fswatcher/setup.go

@@ -0,0 +1,73 @@
+package fswatcher
+
+import (
+	"fmt"
+
+	"github.com/dominicbreuker/pspy/internal/fswatcher/inotify"
+	isys "github.com/dominicbreuker/pspy/internal/fswatcher/inotify/sys"
+	"github.com/dominicbreuker/pspy/internal/fswatcher/walker"
+)
+
+type InotifyWatcher struct {
+	i *inotify.Inotify
+}
+
+func (iw *InotifyWatcher) Close() {
+	iw.i.Close()
+}
+
+func NewInotifyWatcher() (*InotifyWatcher, error) {
+	i, err := inotify.NewInotify(&isys.InotifySyscallsUNIX{})
+	if err != nil {
+		return nil, fmt.Errorf("setting up inotify: %v", err)
+	}
+	return &InotifyWatcher{
+		i: i,
+	}, nil
+}
+
+func (iw *InotifyWatcher) Setup(rdirs, dirs []string, errCh chan error) (chan struct{}, chan string, error) {
+	maxWatchers, err := getLimit()
+	if err != nil {
+		errCh <- fmt.Errorf("Can't get inotify watcher limit...: %v\n", err)
+		maxWatchers = -1
+	}
+
+	for _, dir := range rdirs {
+		addWatchers(dir, -1, iw.i, maxWatchers, errCh)
+	}
+	for _, dir := range dirs {
+		addWatchers(dir, 0, iw.i, maxWatchers, errCh)
+	}
+
+	triggerCh := make(chan struct{})
+	dataCh := make(chan []byte)
+	go Observe(iw.i, triggerCh, dataCh, errCh)
+
+	eventCh := make(chan string)
+	go parseEvents(iw.i, dataCh, eventCh, errCh)
+
+	return triggerCh, eventCh, nil
+}
+
+func addWatchers(dir string, depth int, i *inotify.Inotify, maxWatchers int, errCh chan error) {
+	dirCh, walkErrCh, doneCh := walker.Walk(dir, depth)
+loop:
+	for {
+		if maxWatchers > 0 && i.NumWatchers() >= maxWatchers {
+			close(doneCh)
+			break loop
+		}
+		select {
+		case err := <-walkErrCh:
+			errCh <- fmt.Errorf("adding inotift watchers: %v", err)
+		case dir, ok := <-dirCh:
+			if !ok {
+				break loop
+			}
+			if err := i.Watch(dir); err != nil {
+				errCh <- fmt.Errorf("Can't create watcher: %v", err)
+			}
+		}
+	}
+}

internal/fswatcher/walker/walker.go

@@ -0,0 +1,53 @@
+package walker
+
+import (
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+)
+
+const maxInt = int(^uint(0) >> 1)
+
+func Walk(root string, depth int) (dirCh chan string, errCh chan error, doneCh chan struct{}) {
+	if depth < 0 {
+		depth = maxInt
+	}
+	dirCh = make(chan string)
+	errCh = make(chan error)
+	doneCh = make(chan struct{})
+
+	go func() {
+		defer close(dirCh)
+		descent(root, depth-1, dirCh, errCh, doneCh)
+	}()
+	return dirCh, errCh, doneCh
+}
+
+func descent(dir string, depth int, dirCh chan string, errCh chan error, doneCh chan struct{}) {
+	_, err := os.Stat(dir)
+	if err != nil {
+		errCh <- fmt.Errorf("visiting %s: %v", dir, err)
+		return
+	}
+	select {
+	case dirCh <- dir:
+	case <-doneCh:
+		return
+	}
+	if depth < 0 {
+		return
+	}
+
+	ls, err := ioutil.ReadDir(dir)
+	if err != nil {
+		errCh <- fmt.Errorf("opening dir %s: %v", dir, err)
+	}
+
+	for _, e := range ls {
+		if e.IsDir() {
+			newDir := filepath.Join(dir, e.Name())
+			descent(newDir, depth-1, dirCh, errCh, doneCh)
+		}
+	}
+}

internal/fswatcher/walker/walker_test.go

@@ -0,0 +1,84 @@
+package walker
+
+import (
+	"reflect"
+	"strings"
+	"testing"
+)
+
+func TestWalk(t *testing.T) {
+	tests := []struct {
+		root   string
+		depth  int
+		errCh  chan error
+		result []string
+		errs   []string
+	}{
+		{root: "testdata", depth: 999, errCh: newErrCh(), result: []string{
+			"testdata",
+			"testdata/subdir",
+			"testdata/subdir/subsubdir",
+		}, errs: make([]string, 0)},
+		{root: "testdata", depth: -1, errCh: newErrCh(), result: []string{
+			"testdata",
+			"testdata/subdir",
+			"testdata/subdir/subsubdir",
+		}, errs: []string{}},
+		{root: "testdata", depth: 1, errCh: newErrCh(), result: []string{
+			"testdata",
+			"testdata/subdir",
+		}, errs: []string{}},
+		{root: "testdata", depth: 0, errCh: newErrCh(), result: []string{
+			"testdata",
+		}, errs: []string{}},
+		{root: "testdata/subdir", depth: 1, errCh: newErrCh(), result: []string{
+			"testdata/subdir",
+			"testdata/subdir/subsubdir",
+		}, errs: []string{}},
+		{root: "testdata/non-existing-dir", depth: 1, errCh: newErrCh(), result: []string{}, errs: []string{"visiting testdata/non-existing-dir"}},
+	}
+
+	for i, tt := range tests {
+		dirCh, errCh, doneCh := Walk(tt.root, tt.depth)
+		dirs, errs := getAllDirsAndErrors(dirCh, errCh)
+
+		if !reflect.DeepEqual(dirs, tt.result) {
+			t.Fatalf("[%d] Wrong number of dirs found: %+v", i, dirs)
+		}
+		if !reflect.DeepEqual(errs, tt.errs) {
+			t.Fatalf("[%d] Wrong number of errs found: %+v vs %+v", i, errs, tt.errs)
+		}
+		close(doneCh)
+	}
+
+}
+
+func getAllDirsAndErrors(dirCh chan string, errCh chan error) ([]string, []string) {
+	dirs := make([]string, 0)
+	errs := make([]string, 0)
+
+	doneDirsCh := make(chan struct{})
+	go func() {
+		defer close(doneDirsCh)
+		defer close(errCh)
+		for d := range dirCh {
+			dirs = append(dirs, d)
+		}
+	}()
+
+	doneErrsCh := make(chan struct{})
+	go func() {
+		defer close(doneErrsCh)
+		for err := range errCh {
+			tokens := strings.SplitN(err.Error(), ":", 2)
+			errs = append(errs, tokens[0])
+		}
+	}()
+	<-doneDirsCh
+	<-doneErrsCh
+	return dirs, errs
+}
+
+func newErrCh() chan error {
+	return make(chan error)
+}

internal/fswatcher/watcher.go

@@ -0,0 +1,44 @@
+package fswatcher
+
+import (
+	"fmt"
+	"io/ioutil"
+	"strconv"
+	"strings"
+
+	"golang.org/x/sys/unix"
+)
+
+const events = unix.IN_ALL_EVENTS
+const MaximumWatchersFile = "/proc/sys/fs/inotify/max_user_watches"
+
+type watcher struct {
+	wd  int
+	dir string
+}
+
+func newWatcher(fd int, dir string) (*watcher, error) {
+	wd, errno := unix.InotifyAddWatch(fd, dir, events)
+	if wd == -1 {
+		return nil, fmt.Errorf("adding watcher on %s: %d", dir, errno)
+	}
+	return &watcher{
+		wd:  wd,
+		dir: dir,
+	}, nil
+}
+
+func getLimit() (int, error) {
+	b, err := ioutil.ReadFile(MaximumWatchersFile)
+	if err != nil {
+		return 0, fmt.Errorf("reading from %s: %v", MaximumWatchersFile, err)
+	}
+
+	s := strings.TrimSpace(string(b))
+	m, err := strconv.Atoi(s)
+	if err != nil {
+		return 0, fmt.Errorf("converting to integer: %v", err)
+	}
+
+	return m, nil
+}