func (t *Tailer) PollLogStreams() error {
	t.logstreamsMu.Lock()
	defer t.logstreamsMu.Unlock()
	for name, l := range t.logstreams {
		if l.IsComplete() {
			glog.Infof("%s is complete", name)
			delete(t.logstreams, name)
			logCount.Add(-1)
			continue
		}
	}
	return nil
}
PollLogStreams looks at the existing paths and checks if they're already
complete, removing it from the map if so.
package tailer
import (
	"context"
	"errors"
	"expvar"
	"fmt"
	"net/url"
	"os"
	"path/filepath"
	"regexp"
	"sync"
	"time"
	"github.com/golang/glog"
	"github.com/google/mtail/internal/logline"
	"github.com/google/mtail/internal/tailer/logstream"
	"github.com/google/mtail/internal/waker"
)
var logCount = expvar.NewInt("log_count")
type Tailer struct {
	ctx   context.Context
	wg    sync.WaitGroup  
	lines chan<- *logline.LogLine
	globPatternsMu     sync.RWMutex         
	globPatterns       map[string]struct{}  
	ignoreRegexPattern *regexp.Regexp
	socketPaths []string
	oneShot bool
	pollMu sync.Mutex  
	logstreamPollWaker waker.Waker                     
	logstreamsMu       sync.RWMutex                    
	logstreams         map[string]logstream.LogStream  
	initDone chan struct{}
}
type Option interface {
	apply(*Tailer) error
}
type niladicOption struct {
	applyfunc func(*Tailer) error
}
func (n *niladicOption) apply(t *Tailer) error {
	return n.applyfunc(t)
}
var OneShot = &niladicOption{func(t *Tailer) error { t.oneShot = true; return nil }}
type LogPatterns []string
func (opt LogPatterns) apply(t *Tailer) error {
	for _, p := range opt {
		if err := t.AddPattern(p); err != nil {
			return err
		}
	}
	return nil
}
type IgnoreRegex string
func (opt IgnoreRegex) apply(t *Tailer) error {
	return t.SetIgnorePattern(string(opt))
}
func StaleLogGcWaker(w waker.Waker) Option {
	return &staleLogGcWaker{w}
}
type staleLogGcWaker struct {
	waker.Waker
}
func (opt staleLogGcWaker) apply(t *Tailer) error {
	t.StartGcLoop(opt.Waker)
	return nil
}
func LogPatternPollWaker(w waker.Waker) Option {
	return &logPatternPollWaker{w}
}
type logPatternPollWaker struct {
	waker.Waker
}
func (opt logPatternPollWaker) apply(t *Tailer) error {
	t.StartLogPatternPollLoop(opt.Waker)
	return nil
}
func LogstreamPollWaker(w waker.Waker) Option {
	return &logstreamPollWaker{w}
}
type logstreamPollWaker struct {
	waker.Waker
}
func (opt logstreamPollWaker) apply(t *Tailer) error {
	t.logstreamPollWaker = opt.Waker
	return nil
}
var ErrNoLinesChannel = errors.New("Tailer needs a lines channel")
func New(ctx context.Context, wg *sync.WaitGroup, lines chan<- *logline.LogLine, options ...Option) (*Tailer, error) {
	if lines == nil {
		return nil, ErrNoLinesChannel
	}
	t := &Tailer{
		ctx:          ctx,
		lines:        lines,
		initDone:     make(chan struct{}),
		globPatterns: make(map[string]struct{}),
		logstreams:   make(map[string]logstream.LogStream),
	}
	defer close(t.initDone)
	if err := t.SetOption(options...); err != nil {
		return nil, err
	}
	if len(t.globPatterns) == 0 && len(t.socketPaths) == 0 {
		glog.Info("No patterns or sockets to tail, tailer done.")
		close(t.lines)
		return t, nil
	}
	for _, pattern := range t.socketPaths {
		if err := t.TailPath(pattern); err != nil {
			return nil, err
		}
	}
	if err := t.PollLogPatterns(); err != nil {
		return nil, err
	}
	wg.Add(1)
	go func() {
		defer wg.Done()
		<-t.initDone
		if !t.oneShot {
			<-t.ctx.Done()
		}
		t.wg.Wait()
		close(t.lines)
	}()
	return t, nil
}
var ErrNilOption = errors.New("nil option supplied")
func (t *Tailer) SetOption(options ...Option) error {
	for _, option := range options {
		if option == nil {
			return ErrNilOption
		}
		if err := option.apply(t); err != nil {
			return err
		}
	}
	return nil
}
var ErrUnsupportedURLScheme = errors.New("unsupported URL scheme")
func (t *Tailer) AddPattern(pattern string) error {
	u, err := url.Parse(pattern)
	if err != nil {
		return err
	}
	switch u.Scheme {
	default:
		return fmt.Errorf("%w: %q in path pattern %q", ErrUnsupportedURLScheme, u.Scheme, pattern)
	case "unix", "unixgram", "tcp", "udp":
		glog.V(2).Infof("AddPattern: socket %q", pattern)
		t.socketPaths = append(t.socketPaths, pattern)
		return nil
	case "", "file":
	}
	absPath, err := filepath.Abs(u.Path)
	if err != nil {
		glog.V(2).Infof("Couldn't canonicalize path %q: %s", u.Path, err)
		return err
	}
	glog.V(2).Infof("AddPattern: file %q", absPath)
	t.globPatternsMu.Lock()
	t.globPatterns[absPath] = struct{}{}
	t.globPatternsMu.Unlock()
	return nil
}
func (t *Tailer) Ignore(pathname string) (bool, error) {
	absPath, err := filepath.Abs(pathname)
	if err != nil {
		return false, err
	}
	fi, err := os.Stat(absPath)
	if err != nil {
		return false, err
	}
	if fi.Mode().IsDir() {
		glog.V(2).Infof("ignore path %q because it is a folder", pathname)
		return true, nil
	}
	return t.ignoreRegexPattern != nil && t.ignoreRegexPattern.MatchString(fi.Name()), nil
}
func (t *Tailer) SetIgnorePattern(pattern string) error {
	if len(pattern) == 0 {
		return nil
	}
	glog.V(2).Infof("Set filename ignore regex pattern %q", pattern)
	ignoreRegexPattern, err := regexp.Compile(pattern)
	if err != nil {
		glog.V(2).Infof("Couldn't compile regex %q: %s", pattern, err)
		fmt.Printf("error: %v\n", err)
		return err
	}
	t.ignoreRegexPattern = ignoreRegexPattern
	return nil
}
func (t *Tailer) TailPath(pathname string) error {
	t.logstreamsMu.Lock()
	defer t.logstreamsMu.Unlock()
	if l, ok := t.logstreams[pathname]; ok {
		if !l.IsComplete() {
			glog.V(2).Infof("already got a logstream on %q", pathname)
			return nil
		}
		logCount.Add(-1)  
		glog.V(2).Infof("Existing logstream is finished, creating a new one.")
	}
	l, err := logstream.New(t.ctx, &t.wg, t.logstreamPollWaker, pathname, t.lines, t.oneShot)
	if err != nil {
		return err
	}
	if t.oneShot {
		glog.V(2).Infof("Starting oneshot read at startup of %q", pathname)
		l.Stop()
	}
	t.logstreams[pathname] = l
	glog.Infof("Tailing %s", pathname)
	logCount.Add(1)
	return nil
}
func (t *Tailer) Gc() error {
	t.logstreamsMu.Lock()
	defer t.logstreamsMu.Unlock()
	for _, v := range t.logstreams {
		if time.Since(v.LastReadTime()) > (time.Hour * 24) {
			v.Stop()
		}
	}
	return nil
}
func (t *Tailer) StartGcLoop(waker waker.Waker) {
	if waker == nil {
		glog.Info("Log handle expiration disabled")
		return
	}
	t.wg.Add(1)
	go func() {
		defer t.wg.Done()
		<-t.initDone
		if t.oneShot {
			glog.Info("No gc loop in oneshot mode.")
			return
		}
		for {
			select {
			case <-t.ctx.Done():
				return
			case <-waker.Wake():
				if err := t.Gc(); err != nil {
					glog.Info(err)
				}
			}
		}
	}()
}
func (t *Tailer) StartLogPatternPollLoop(waker waker.Waker) {
	if waker == nil {
		glog.Info("Log pattern polling disabled")
		return
	}
	t.wg.Add(1)
	go func() {
		defer t.wg.Done()
		<-t.initDone
		if t.oneShot {
			glog.Info("No polling loop in oneshot mode.")
			return
		}
		for {
			select {
			case <-t.ctx.Done():
				return
			case <-waker.Wake():
				if err := t.Poll(); err != nil {
					glog.Info(err)
				}
			}
		}
	}()
}
func (t *Tailer) PollLogPatterns() error {
	t.globPatternsMu.RLock()
	defer t.globPatternsMu.RUnlock()
	for pattern := range t.globPatterns {
		matches, err := filepath.Glob(pattern)
		if err != nil {
			return err
		}
		glog.V(1).Infof("glob matches: %v", matches)
		for _, pathname := range matches {
			ignore, err := t.Ignore(pathname)
			if err != nil {
				return err
			}
			if ignore {
				continue
			}
			absPath, err := filepath.Abs(pathname)
			if err != nil {
				return err
			}
			glog.V(2).Infof("watched path is %q", absPath)
			if err := t.TailPath(absPath); err != nil {
				glog.Info(err)
			}
		}
	}
	return nil
}