mirror of
https://github.com/DominicBreuker/pspy.git
synced 2025-12-21 11:44:51 +00:00
restructure inotify package and add some tests
This commit is contained in:
Dominic Breuker
@@ -2,15 +2,10 @@ package pspy
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"github.com/dominicbreuker/pspy/internal/config"
|
||||
"github.com/dominicbreuker/pspy/internal/inotify"
|
||||
"github.com/dominicbreuker/pspy/internal/process"
|
||||
"github.com/dominicbreuker/pspy/internal/walker"
|
||||
)
|
||||
|
||||
type Logger interface {
|
||||
@@ -20,7 +15,7 @@ type Logger interface {
|
||||
}
|
||||
|
||||
type InotifyWatcher interface {
|
||||
Setup(rdirs, dirs []string) (chan struct{}, chan string, error)
|
||||
Setup(rdirs, dirs []string, errCh chan error) (chan struct{}, chan string, error)
|
||||
}
|
||||
|
||||
type ProcfsScanner interface {
|
||||
@@ -30,7 +25,11 @@ type ProcfsScanner interface {
|
||||
func Start(cfg config.Config, logger Logger, inotify InotifyWatcher, pscan ProcfsScanner, sigCh chan os.Signal) (chan struct{}, error) {
|
||||
logger.Infof("Config: %+v\n", cfg)
|
||||
|
||||
triggerCh, fsEventCh, err := inotify.Setup(cfg.RDirs, cfg.Dirs)
|
||||
// log all errors
|
||||
errCh := make(chan error, 10)
|
||||
go logErrors(errCh, logger)
|
||||
|
||||
triggerCh, fsEventCh, err := inotify.Setup(cfg.RDirs, cfg.Dirs, errCh)
|
||||
if err != nil {
|
||||
logger.Errorf("Can't set up inotify watchers: %v\n", err)
|
||||
return nil, errors.New("inotify error")
|
||||
@@ -41,8 +40,10 @@ func Start(cfg config.Config, logger Logger, inotify InotifyWatcher, pscan Procf
|
||||
return nil, errors.New("procfs scanner error")
|
||||
}
|
||||
|
||||
exit := make(chan struct{})
|
||||
// ignore all file system events created on startup
|
||||
drainChanFor(fsEventCh, 1*time.Second)
|
||||
|
||||
exit := make(chan struct{})
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
@@ -63,72 +64,89 @@ func Start(cfg config.Config, logger Logger, inotify InotifyWatcher, pscan Procf
|
||||
return exit, nil
|
||||
}
|
||||
|
||||
const MaxInt = int(^uint(0) >> 1)
|
||||
|
||||
func Watch(rdirs, dirs []string, logPS, logFS bool) {
|
||||
maxWatchers, err := inotify.WatcherLimit()
|
||||
if err != nil {
|
||||
log.Printf("Can't get inotify watcher limit...: %v\n", err)
|
||||
func logErrors(errCh chan error, logger Logger) {
|
||||
for {
|
||||
err := <-errCh
|
||||
logger.Errorf("ERROR: %v\n", err)
|
||||
}
|
||||
log.Printf("Inotify watcher limit: %d (/proc/sys/fs/inotify/max_user_watches)\n", maxWatchers)
|
||||
|
||||
ping := make(chan struct{})
|
||||
in, err := inotify.NewInotify(ping, logFS)
|
||||
if err != nil {
|
||||
log.Fatalf("Can't init inotify: %v", err)
|
||||
}
|
||||
defer in.Close()
|
||||
|
||||
for _, dir := range rdirs {
|
||||
addWatchers(dir, MaxInt, in, maxWatchers)
|
||||
}
|
||||
for _, dir := range dirs {
|
||||
addWatchers(dir, 0, in, maxWatchers)
|
||||
}
|
||||
|
||||
log.Printf("Inotify watchers set up: %s - watching now\n", in)
|
||||
|
||||
procList := process.NewProcList()
|
||||
|
||||
ticker := time.NewTicker(100 * time.Millisecond)
|
||||
}
|
||||
|
||||
func drainChanFor(c chan string, d time.Duration) {
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
refresh(in, procList, logPS)
|
||||
case <-ping:
|
||||
refresh(in, procList, logPS)
|
||||
case <-c:
|
||||
case <-time.After(d):
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func addWatchers(dir string, depth int, in *inotify.Inotify, maxWatchers int) {
|
||||
dirCh, errCh, doneCh := walker.Walk(dir, depth)
|
||||
loop:
|
||||
for {
|
||||
if in.NumWatchers() >= maxWatchers {
|
||||
close(doneCh)
|
||||
break loop
|
||||
}
|
||||
select {
|
||||
case dir, ok := <-dirCh:
|
||||
if !ok {
|
||||
break loop
|
||||
}
|
||||
if err := in.Watch(dir); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "Can't create watcher: %v\n", err)
|
||||
}
|
||||
case err := <-errCh:
|
||||
fmt.Fprintf(os.Stderr, "Error walking filesystem: %v\n", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
// const MaxInt = int(^uint(0) >> 1)
|
||||
|
||||
func refresh(in *inotify.Inotify, pl *process.ProcList, print bool) {
|
||||
in.Pause()
|
||||
if err := pl.Refresh(print); err != nil {
|
||||
log.Printf("ERROR refreshing process list: %v", err)
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
in.UnPause()
|
||||
}
|
||||
// func Watch(rdirs, dirs []string, logPS, logFS bool) {
|
||||
// maxWatchers, err := inotify.WatcherLimit()
|
||||
// if err != nil {
|
||||
// log.Printf("Can't get inotify watcher limit...: %v\n", err)
|
||||
// }
|
||||
// log.Printf("Inotify watcher limit: %d (/proc/sys/fs/inotify/max_user_watches)\n", maxWatchers)
|
||||
|
||||
// ping := make(chan struct{})
|
||||
// in, err := inotify.NewInotify(ping, logFS)
|
||||
// if err != nil {
|
||||
// log.Fatalf("Can't init inotify: %v", err)
|
||||
// }
|
||||
// defer in.Close()
|
||||
|
||||
// for _, dir := range rdirs {
|
||||
// addWatchers(dir, MaxInt, in, maxWatchers)
|
||||
// }
|
||||
// for _, dir := range dirs {
|
||||
// addWatchers(dir, 0, in, maxWatchers)
|
||||
// }
|
||||
|
||||
// log.Printf("Inotify watchers set up: %s - watching now\n", in)
|
||||
|
||||
// procList := process.NewProcList()
|
||||
|
||||
// ticker := time.NewTicker(100 * time.Millisecond)
|
||||
|
||||
// for {
|
||||
// select {
|
||||
// case <-ticker.C:
|
||||
// refresh(in, procList, logPS)
|
||||
// case <-ping:
|
||||
// refresh(in, procList, logPS)
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
|
||||
// func addWatchers(dir string, depth int, in *inotify.Inotify, maxWatchers int) {
|
||||
// dirCh, errCh, doneCh := walker.Walk(dir, depth)
|
||||
// loop:
|
||||
// for {
|
||||
// if in.NumWatchers() >= maxWatchers {
|
||||
// close(doneCh)
|
||||
// break loop
|
||||
// }
|
||||
// select {
|
||||
// case dir, ok := <-dirCh:
|
||||
// if !ok {
|
||||
// break loop
|
||||
// }
|
||||
// if err := in.Watch(dir); err != nil {
|
||||
// fmt.Fprintf(os.Stderr, "Can't create watcher: %v\n", err)
|
||||
// }
|
||||
// case err := <-errCh:
|
||||
// fmt.Fprintf(os.Stderr, "Error walking filesystem: %v\n", err)
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
|
||||
// func refresh(in *inotify.Inotify, pl *process.ProcList, print bool) {
|
||||
// in.Pause()
|
||||
// if err := pl.Refresh(print); err != nil {
|
||||
// log.Printf("ERROR refreshing process list: %v", err)
|
||||
// }
|
||||
// time.Sleep(5 * time.Millisecond)
|
||||
// in.UnPause()
|
||||
// }
|
||||
|
||||
@@ -92,6 +92,7 @@ type mockInotifyWatcher struct {
|
||||
triggerCh chan struct{}
|
||||
fsEventCh chan string
|
||||
setupErr error
|
||||
closed bool
|
||||
}
|
||||
|
||||
func newMockInotifyWatcher(setupErr error) *mockInotifyWatcher {
|
||||
@@ -99,16 +100,21 @@ func newMockInotifyWatcher(setupErr error) *mockInotifyWatcher {
|
||||
triggerCh: make(chan struct{}),
|
||||
fsEventCh: make(chan string),
|
||||
setupErr: setupErr,
|
||||
closed: false,
|
||||
}
|
||||
}
|
||||
|
||||
func (i *mockInotifyWatcher) Setup(rdirs, dirs []string) (chan struct{}, chan string, error) {
|
||||
func (i *mockInotifyWatcher) Setup(rdirs, dirs []string, errCh chan error) (chan struct{}, chan string, error) {
|
||||
if i.setupErr != nil {
|
||||
return nil, nil, i.setupErr
|
||||
}
|
||||
return i.triggerCh, i.fsEventCh, nil
|
||||
}
|
||||
|
||||
func (i mockInotifyWatcher) Close() {
|
||||
i.closed = true
|
||||
}
|
||||
|
||||
// ProcfsScanner
|
||||
|
||||
type mockProcfsScanner struct {
|
||||
|
||||
Reference in New Issue
Block a user