spooler.go

/*
 * Copyright (c) 2019 Karim Kanso. All Rights Reserved.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

// Extract ids events from a logging directory. Scans directory for
// new files and adds them to the processing list.
package spooler

import (
	"fmt"
	"io/ioutil"
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/fsnotify/fsnotify"

	"github.com/kazkansouh/u2text/parser"
	"github.com/kazkansouh/u2text/parser/u2"
)

// Error codes produced by Spooler function
type ErrorCode int

// Sequence of error constants produced by Spooler function
const (
	// Unable to read information about file that has been created
	// in spooler directory
	E_WatcherStat ErrorCode = iota
	// Error passed down from fsnotify
	E_Watcher
	// Unable to list files in the spooler directory
	E_WatcherSpoolerDirectoryList
	// Error passed down from parser
	E_Parse
)

// Maximum number of files to enqueue
const FileNameBufferLen = 512

// Errors generated by spooler function, type primarily used for test
// code
type SpoolError interface {
	error
	Code() ErrorCode
	NextError() error
}

type spoolError struct {
	message string
	ErrorCode
	error
}

func (e *spoolError) Error() string {
	if e.error != nil {
		return e.message + ": " + e.error.Error()
	}
	return e.message
}

func (e *spoolError) Code() ErrorCode {
	return e.ErrorCode
}

func (e *spoolError) NextError() error {
	return e.error
}

// File and offset within the file that has been processed
// already. Assumes events in files which have in lexicographical
// order less than the indicated file have been processed.
type Marker struct {
	File   string
	Offset int64
}

// wrap return values for the Spooler.process function to pass over
// channel
type processReturn struct {
	*Marker
	error
}

// Stubbed interface of fsnotify.Watcher, to support testing
type Watcher interface {
	Add(name string) error
	Close() error
	EventC() <-chan fsnotify.Event
	ErrorC() <-chan error
}

// Wrap fsnotify.Watcher struct in local structure so its possible to
// expose the Events and Errors channels via functions.
type FileWatcher fsnotify.Watcher

// return fsnotify.Watcher.Events channel
func (w *FileWatcher) EventC() <-chan fsnotify.Event { return w.Events }

// return fsnotify.Watcher.Errors channel
func (w *FileWatcher) ErrorC() <-chan error { return w.Errors }

// lift fsnotify.Watcher.Add
func (w *FileWatcher) Add(name string) error { return (*fsnotify.Watcher)(w).Add(name) }

// lift fsnotify.Watcher.Close
func (w *FileWatcher) Close() error { return (*fsnotify.Watcher)(w).Close() }

// function references for tests
var (
	newWatcher = func() (Watcher, error) {
		w, err := fsnotify.NewWatcher()
		return (*FileWatcher)(w), err
	}

	readDir = ioutil.ReadDir

	stat = os.Stat
)

type Spooler interface {
	// Asynchronously run the spooler. Starts a new goroutine.
	//
	// Set marker to the start location (i.e. file and offset
	// within file to start parsing).
	//
	// All errors are reported via ErrorC.  The channels will be
	// closed when finished.
	//
	// The marker channel will return exactly 1 result when the
	// process finishes which can be used for resuming spooling
	// from last processed record.
	Start(marker *Marker) (<-chan *u2.Record, <-chan *Marker)

	// Run the spooler. The read records are returned in the
	// results channel.
	//
	// SyncStart blocks until both the main process loop and the
	// file watcher have been terminated (either by error or call
	// to Stop).
	SyncStart(marker *Marker, results chan<- *u2.Record) (*Marker, error)

	// Stops processing events and release resources.
	//
	// When graceful is true, stops watching for new files and
	// then processes all events available (i.e. it runs in batch
	// mode). When false, it will sharply end the processing
	// mid-file.
	//
	// Returns false when it was not possible to processes the
	// request.
	Stop(graceful bool) bool

	// Used by Start() to return errors
	ErrorC() <-chan error
}

// Stores all state information needed by spooler
type spooler struct {
	basename        string
	logdir          string
	processshutdown chan bool
	shutdown        chan bool
	watcher         Watcher
	watcherdone     chan error
	processdone     chan processReturn
	todo            chan string
	done            Marker
	stopWatcher     sync.Once
	errors          chan error

	// channel for signalling to tests that the watcher is running
	watcherReady chan struct{}
}

// Create a new instance of a spooler for a given logdir. Searches for
// files which start with basename in logidr.
func NewSpooler(logdir, basename string) Spooler {
	return &spooler{
		basename:        basename,
		logdir:          logdir,
		processshutdown: make(chan bool, 0),
		shutdown:        make(chan bool, 0),
		watcherdone:     make(chan error, 0),
		processdone:     make(chan processReturn, 0),
		todo:            make(chan string, FileNameBufferLen),
		errors:          make(chan error, 0),
		watcherReady:    make(chan struct{}, 0),
	}
}

// watch directory for new files
func (s *spooler) watch() {
	var result_err error = nil
	defer func() {
		s.watcherdone <- result_err
	}()

	// Scan spool directory for files that match basename
	files, err := readDir(s.logdir)
	if err != nil {
		result_err = &spoolError{
			message:   fmt.Sprintf("Unable to list directory: %q", s.logdir),
			ErrorCode: E_WatcherSpoolerDirectoryList,
			error:     err,
		}
		return
	}
	for _, file := range files {
		if !file.IsDir() && strings.HasPrefix(file.Name(), s.basename) {
			s.todo <- file.Name()
		}
	}

	// signal to tests that spooler directory has been scanned (so
	// new files will be picked up by watcher)
	close(s.watcherReady)

	// separate out error channel into goroutine to reduce
	// non-determinism in code flow
	anError := make(chan error, 1)
	go func() {
		err, ok := <-s.watcher.ErrorC()
		if ok {
			anError <- err
		}
	}()

	// watch for new files
	for {
		select {
		case event, ok := <-s.watcher.EventC():
			if !ok {
				return
			}
			if event.Op&fsnotify.Create == fsnotify.Create {
				fi, err := stat(event.Name)
				if err != nil {
					result_err = &spoolError{
						message:   fmt.Sprintf("Unable to stat file: %q", event.Name),
						ErrorCode: E_WatcherStat,
						error:     err,
					}
					return
				}
				if !fi.IsDir() && strings.HasPrefix(fi.Name(), s.basename) {
					s.todo <- fi.Name()
				}
			}
		case err := <-anError:
			result_err = &spoolError{
				ErrorCode: E_Watcher,
				error:     err,
			}
			return
		}
	}
}

// main event loop, manages which file to parse
func (s *spooler) process(result chan<- *u2.Record) {
	var err error = nil
	defer func() {
		if s.done.File != "" {
			s.processdone <- processReturn{&s.done, err}
		} else {
			s.processdone <- processReturn{nil, err}
		}
	}()
	var clientresult <-chan *u2.Record
	var clienterrors <-chan error
	var clientshutdown chan *u2.Unit
	var clientshutdown_pending chan *u2.Unit

	timer := time.NewTimer(0)

	// caller has requested shutdown at end of records
	batch := false
	// caller has requested showdown now
	brutal := false
	// a new file is pending to be read
	var newpending string = ""
	// reference to s.processshutdown, initially nil to prevent
	// early shutdown
	var processshutdown_local <-chan bool = nil
	// timer to control how long to disable shutdown feature in
	// case no processing has started, e.g. to allow for spooler
	// directory to be listed and started reading
	processshutdown_timer := time.NewTimer(time.Millisecond * 750)

	start := func(file string, offset int64) {
		clientshutdown = make(chan *u2.Unit, 1)
		clientshutdown_pending = nil
		clientresult, clienterrors = parser.ParseU2(filepath.Join(s.logdir, file), offset, clientshutdown)
		if batch {
			clientshutdown_pending = clientshutdown
		}
		if processshutdown_local == nil {
			processshutdown_local = s.processshutdown
			processshutdown_timer.Stop()
		}
	}
loop:
	for {
		if newpending == "" && !brutal {
			// continue used in following to quickly
			// populate newpending with next file to
			// process
			select {
			case file := <-s.todo:
				// skip processed files
				if file < s.done.File {
					log.Println("WARNING: Skipping log file: ", file)
					continue
				}
				if file == s.done.File {
					if clientresult == nil {
						// continue processing
						start(file, s.done.Offset)
					}
					continue
				} else {
					if clientresult == nil {
						// new file
						s.done.File = file
						s.done.Offset = 0
						start(file, 0)
						continue

					} else {
						// already parsing file, request graceful shutdown
						log.Println("Newfile pending: " + file)
						if !batch {
							// already notified in batch mode
							clientshutdown_pending = clientshutdown
						}
						newpending = file
					}
				}
			default:
			}
		}

		select {
		case <-processshutdown_timer.C:
			processshutdown_local = s.processshutdown
		case graceful := <-processshutdown_local:
			// if no processing has started yet
			if clientresult == nil {
				return
			}
			if graceful {
				batch = true
				log.Println("Shutdown requested (graceful)")
				clientshutdown_pending = clientshutdown
			} else {
				brutal = true
				log.Println("Shutdown requested (brutal)")
				newpending = ""
				if clientshutdown != nil {
					close(clientshutdown)
					clientshutdown = nil
					clientshutdown_pending = nil
				}
			}
		case perr := <-clienterrors:
			err = &spoolError{
				ErrorCode: E_Parse,
				error:     perr,
			}
			// client will now exit, no need to send shutdown request
			brutal = true
			newpending = ""
			log.Println("ERROR: Unable to parse unified2 file:", err.Error())
		case clientshutdown_pending <- u2.U:
			clientshutdown_pending = nil
		default:
		}

		// clean up timer state
		if !timer.Stop() {
			select {
			case <-timer.C:
			default:
			}
		}

		timer.Reset(time.Millisecond * 100)

	clientresult:
		select {
		case record, ok := <-clientresult:
			if !ok {
				if !brutal {
					// reached eof
					log.Println("Finsihed processing:", s.done.File)
				}

				if newpending != "" {
					// client ended, start on next file
					s.done.File = newpending
					s.done.Offset = 0
					start(newpending, 0)
					newpending = ""
				} else {
					break loop
				}
				break clientresult
			}
			s.done.Offset = record.Offset
			result <- record
		case <-timer.C:
		}

	}
}

// Stops processing events and release resources.
//
// When graceful is true, stops watching for new files and then
// processes all events available (i.e. it runs in batch mode). When
// false, it will sharply end the processing mid-file.
//
// Returns false when it was not possible to processes the request
// (blocks for upto 125 milliseconds).
func (s *spooler) Stop(graceful bool) bool {
	select {
	case s.shutdown <- graceful:
		return true
	case <-time.NewTimer(125 * time.Millisecond).C:
		// 125 is chosen as its > 100 ms used by the main
		// loop, to ensure that a shutdown is not ignored
		return false
	}
}

// Keep processing shutdown requests until stopped is written to
func (s *spooler) shutdownHandler(stopped chan struct{}) {
	for {
		select {
		case graceful := <-s.shutdown:
			select {
			case s.processshutdown <- graceful:
				s.stopWatcher.Do(func() { s.watcher.Close() })
			case <-stopped:
				return
			}
		case <-stopped:
			return
		}
	}
}

// Run the spooler. The read records are returned in the results
// channel.
//
// SyncStart blocks until both the main process loop and the file
// watcher have been terminated (either by error or call to Stop).
func (s *spooler) SyncStart(marker *Marker, results chan<- *u2.Record) (*Marker, error) {
	defer close(results)
	var err error
	if s.watcher, err = newWatcher(); err != nil {

		return marker, &spoolError{
			message:   "Unable to start watcher",
			ErrorCode: E_Watcher,
			error:     err,
		}
	}

	// start watching log directory before scanning to ensure
	// there is no gap in detecting added files
	if err = s.watcher.Add(s.logdir); err != nil {
		return marker, &spoolError{
			message:   fmt.Sprintf("Unable to add directory %q to watcher", s.logdir),
			ErrorCode: E_Watcher,
			error:     err,
		}
	}

	// save start point
	if marker != nil {
		s.done = *marker
	}

	// start the main processing loop
	go s.process(results)

	// add any new files that arrive to queue
	go s.watch()

	// process shutdown requests
	stop := make(chan struct{}, 0)
	go s.shutdownHandler(stop)

	// block until processes have yielded results using
	// sync.WaitGroup
	var res processReturn
	var werr error
	var wg sync.WaitGroup
	wg.Add(2)

	processdone := make(chan struct{}, 0)
	go func() {
		defer wg.Done()
		defer close(processdone)

		// wait for main process to finish, then stop watcher
		res = <-s.processdone
		s.stopWatcher.Do(func() { s.watcher.Close() })
	}()

	go func() {
		defer wg.Done()
		werr = <-s.watcherdone

		// in case of error, send a brutal shutdown event
		if werr != nil {
			// stop main loop
			select {
			case s.shutdown <- false:
			case <-processdone:
			}
		}
	}()

	wg.Wait()

	// stop shutdown handler
	stop <- struct{}{}

	// parser errors take precedence
	if res.error != nil {
		return res.Marker, res.error
	}
	return res.Marker, werr

}

// Asynchronously run the SyncStart function. Set marker to the start
// location (i.e. file and offset within file to start parsing).
//
// All errors are reported via Spooler.ErrorC().  The channels will be
// closed when finished.
//
// The marker channel will return exactly 1 result when the process
// finishes which can be used for resuming spooling from last
// processed record.
//
// Note, the error channel must be first read from until its closed,
// then the marker channel will have a result provided.
func (s *spooler) Start(marker *Marker) (<-chan *u2.Record, <-chan *Marker) {

	results := make(chan *u2.Record, 10)
	newmarker := make(chan *Marker, 0)

	go func() {
		done, err := s.SyncStart(marker, results)

		if err != nil {
			s.errors <- err
		}
		close(s.errors)

		newmarker <- done
		close(newmarker)
	}()

	return results, newmarker
}

// Return reference to s.errors channel
func (s *spooler) ErrorC() <-chan error {
	return s.errors
}