watch any log files that appear in source dir

[#117078735]

Signed-off-by: Maria Shaldibina <[email protected]>

cmd/blackbox/main.go

@@ -33,40 +33,17 @@ func main() {
 		logger.Fatalf("could not load config file: %s\n", err)
 	}
 
-	members := grouper.Members{}
-
-	if len(config.Syslog.Sources) > 0 {
-		drainer, err := syslog.NewDrainer(
-			config.Syslog.Destination,
-			config.Hostname,
-		)
-		if err != nil {
-			logger.Fatalf("could not drain to syslog: %s\n", err)
-		}
-
-		members = append(members, buildTailers(config.Syslog.Sources, drainer)...)
-	}
-
-	group := grouper.NewParallel(nil, members)
+	group := grouper.NewDynamic(nil, 0, 0)
 	running := ifrit.Invoke(sigmon.New(group))
 
+	go func() {
+		drainerFactory := syslog.NewDrainerFactory(config.Syslog.Destination, config.Hostname)
+		fileWatcher := blackbox.NewFileWatcher(logger, config.Syslog.SourceDir, group.Client(), drainerFactory)
+		fileWatcher.Watch()
+	}()
+
 	err = <-running.Wait()
 	if err != nil {
 		logger.Fatalf("failed: %s", err)
 	}
 }
-
-func buildTailers(sources []blackbox.SyslogSource, drainer syslog.Drainer) grouper.Members {
-	members := make(grouper.Members, len(sources))
-
-	for i, source := range sources {
-		tailer := &blackbox.Tailer{
-			Source:  source,
-			Drainer: drainer,
-		}
-
-		members[i] = grouper.Member{source.Path, tailer}
-	}
-
-	return members
-}

config.go

@@ -40,8 +40,8 @@ type SyslogSource struct {
 }
 
 type SyslogConfig struct {
-	Destination syslog.Drain   `yaml:"destination"`
-	Sources     []SyslogSource `yaml:"sources"`
+	Destination syslog.Drain `yaml:"destination"`
+	SourceDir   string       `yaml:"source_dir"`
 }
 
 type Config struct {

file_watcher.go

@@ -0,0 +1,96 @@
+package blackbox
+
+import (
+	"io/ioutil"
+	"log"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/concourse/blackbox/syslog"
+	"github.com/tedsuo/ifrit/grouper"
+)
+
+const POLL_INTERVAL = 5 * time.Second
+
+type fileWatcher struct {
+	logger *log.Logger
+
+	sourceDir          string
+	dynamicGroupClient grouper.DynamicClient
+
+	drainerFactory syslog.DrainerFactory
+}
+
+func NewFileWatcher(
+	logger *log.Logger,
+	sourceDir string,
+	dynamicGroupClient grouper.DynamicClient,
+	drainerFactory syslog.DrainerFactory,
+) *fileWatcher {
+	return &fileWatcher{
+		logger:             logger,
+		sourceDir:          sourceDir,
+		dynamicGroupClient: dynamicGroupClient,
+		drainerFactory:     drainerFactory,
+	}
+}
+
+func (f *fileWatcher) Watch() {
+	for {
+		logDirs, err := ioutil.ReadDir(f.sourceDir)
+		if err != nil {
+			f.logger.Fatalf("could not list directories in source dir: %s\n", err)
+		}
+
+		for _, logDir := range logDirs {
+			tag := logDir.Name()
+			tagDirPath := filepath.Join(f.sourceDir, tag)
+
+			fileInfo, err := os.Stat(tagDirPath)
+			if err != nil {
+				f.logger.Fatalf("failed to determine if path is directory: %s\n", err)
+			}
+
+			if !fileInfo.IsDir() {
+				continue
+			}
+
+			logFiles, err := ioutil.ReadDir(tagDirPath)
+			if err != nil {
+				f.logger.Fatalf("could not list files in log dir %s: %s\n", tag, err)
+			}
+
+			for _, logFile := range logFiles {
+				logFileFullPath := filepath.Join(tagDirPath, logFile.Name())
+				if _, found := f.dynamicGroupClient.Get(logFileFullPath); !found {
+					f.dynamicGroupClient.Inserter() <- f.memberForFile(logFileFullPath)
+				}
+			}
+		}
+
+		time.Sleep(POLL_INTERVAL)
+	}
+}
+
+func (f *fileWatcher) memberForFile(logfilePath string) grouper.Member {
+	drainer, err := f.drainerFactory.NewDrainer()
+	if err != nil {
+		f.logger.Fatalf("could not drain to syslog: %s\n", err)
+	}
+
+	logfileDir := filepath.Dir(logfilePath)
+
+	tag, err := filepath.Rel(f.sourceDir, logfileDir)
+	if err != nil {
+		f.logger.Fatalf("could not compute tag from file path %s: %s\n", logfilePath, err)
+	}
+
+	tailer := &Tailer{
+		Path:    logfilePath,
+		Tag:     tag,
+		Drainer: drainer,
+	}
+
+	return grouper.Member{tailer.Path, tailer}
+}