corrected pr comments + enhanced code and logic

cmd/cmd.go

@@ -298,6 +298,13 @@ var GlobalFlags = []cli.Flag{
 		Usage:       "If turned on, the crawler will send back URLs that hit a rate limit to crawl HQ.",
 		Destination: &config.App.Flags.HQRateLimitingSendBack,
 	},
+	// Logging flags
+	&cli.StringFlag{
+		Name:        "log-file-output-dir",
+		Usage:       "Directory to write log files to.",
+		Value:       "jobs",
+		Destination: &config.App.Flags.LogFileOutputDir,
+	},
 	&cli.StringFlag{
 		Name:        "es-url",
 		Usage:       "comma-separated ElasticSearch URL to use for indexing crawl logs.",
@@ -315,8 +322,8 @@ var GlobalFlags = []cli.Flag{
 	},
 	&cli.StringFlag{
 		Name:        "es-index-prefix",
-		Usage:       "ElasticSearch index prefix to use for indexing crawl logs. Default is : `zeno-`",
-		Value:       "zeno-",
+		Usage:       "ElasticSearch index prefix to use for indexing crawl logs. Default is : `zeno`, without `-`",
+		Value:       "zeno",
 		Destination: &config.App.Flags.ElasticSearchIndexPrefix,
 	},
 	&cli.StringSliceFlag{

cmd/utils.go

@@ -37,8 +37,12 @@ func InitCrawlWithCMD(flags config.Flags) *crawl.Crawl {
 		}
 	}
 
+	logFileOutput := &log.Logfile{
+		Dir:    strings.TrimRight(flags.LogFileOutputDir, "/"),
+		Prefix: "zeno",
+	}
 	customLogger, err := log.New(log.Config{
-		FileOutput:               "zeno.log",
+		FileOutput:               logFileOutput,
 		FileLevel:                slog.LevelDebug,
 		StdoutLevel:              slog.LevelInfo,
 		RotateLogFile:            true,
@@ -51,16 +55,6 @@ func InitCrawlWithCMD(flags config.Flags) *crawl.Crawl {
 	}
 	c.Log = customLogger
 
-	go func() {
-		errChan := c.Log.Errors()
-		for {
-			select {
-			case err := <-errChan:
-				fmt.Fprintf(os.Stderr, "Logging error: %v\n", err)
-			}
-		}
-	}()
-
 	// Statistics counters
 	c.CrawledSeeds = new(ratecounter.Counter)
 	c.CrawledAssets = new(ratecounter.Counter)
@@ -71,6 +65,7 @@ func InitCrawlWithCMD(flags config.Flags) *crawl.Crawl {
 
 	// Frontier
 	c.Frontier = new(frontier.Frontier)
+	c.Frontier.Log = c.Log
 
 	// If the job name isn't specified, we generate a random name
 	if flags.Job == "" {

config/config.go

@@ -62,12 +62,12 @@ type Flags struct {
 	DisableAssetsCapture bool
 	CertValidation       bool
 
-	CloudflareStream         bool
 	ElasticSearchURLs        string
 	ElasticSearchUsername    string
 	ElasticSearchPassword    string
 	ElasticSearchIndexPrefix string
 	ExcludedStrings          cli.StringSlice
+	LogFileOutputDir         string
 }
 
 type Application struct {

internal/pkg/crawl/crawl.go

@@ -201,7 +201,7 @@ func (c *Crawl) Start() (err error) {
 
 	c.Client, err = warc.NewWARCWritingHTTPClient(HTTPClientSettings)
 	if err != nil {
-		logrus.Fatalf("Unable to init WARC writing HTTP client: %s", err)
+		c.Log.Fatal("Unable to init WARC writing HTTP client", "error", err)
 	}
 
 	go func() {
@@ -211,7 +211,7 @@ func (c *Crawl) Start() (err error) {
 	}()
 
 	c.Client.Timeout = time.Duration(c.HTTPTimeout) * time.Second
-	logrus.Infof("HTTP client timeout set to %d seconds", c.HTTPTimeout)
+	c.Log.Info("HTTP client timeout set", "timeout", c.HTTPTimeout)
 
 	if c.Proxy != "" {
 		proxyHTTPClientSettings := HTTPClientSettings
@@ -229,7 +229,7 @@ func (c *Crawl) Start() (err error) {
 		}()
 	}
 
-	logrus.Info("WARC writer initialized")
+	c.Log.Info("WARC writer initialized")
 
 	// Process responsible for slowing or pausing the crawl
 	// when the WARC writing queue gets too big
@@ -280,13 +280,13 @@ func (c *Crawl) Start() (err error) {
 		go c.HQWebsocket()
 	} else {
 		// Push the seed list to the queue
-		logrus.Info("Pushing seeds in the local queue..")
+		c.Log.Info("Pushing seeds in the local queue..")
 		for _, item := range c.SeedList {
 			item := item
 			c.Frontier.PushChan <- &item
 		}
 		c.SeedList = nil
-		logrus.Info("All seeds are now in queue, crawling will start")
+		c.Log.Info("All seeds are now in queue, crawling will start")
 	}
 
 	// Start the background process that will catch when there

internal/pkg/crawl/finish.go

@@ -99,7 +99,8 @@ func (crawl *Crawl) finish() {
 	crawl.Log.Warn("Finished!")
 
 	crawl.Log.Warn("Shutting down the logger, bai bai")
-	crawl.Log.Stop()
+	crawl.Log.StopRotation()
+	crawl.Log.StopErrorLog()
 
 	os.Exit(0)
 }

internal/pkg/frontier/frontier.go

@@ -5,6 +5,7 @@ import (
 	"sync"
 
 	"github.com/beeker1121/goque"
+	"github.com/internetarchive/Zeno/internal/pkg/log"
 	"github.com/internetarchive/Zeno/internal/pkg/utils"
 	"github.com/paulbellamy/ratecounter"
 	"github.com/philippgille/gokv/leveldb"
@@ -42,6 +43,7 @@ type Frontier struct {
 	UseSeencheck bool
 	Seencheck    *Seencheck
 	LoggingChan  chan *FrontierLogMessage
+	Log          *log.Logger
 }
 
 type FrontierLogMessage struct {
@@ -70,7 +72,7 @@ func (f *Frontier) Init(jobPath string, loggingChan chan *FrontierLogMessage, wo
 	f.QueueCount = new(ratecounter.Counter)
 	f.QueueCount.Incr(int64(f.Queue.Length()))
 
-	logrus.Info("persistent queue initialized")
+	f.Log.Info("persistent queue initialized")
 
 	// Initialize the seencheck
 	f.UseSeencheck = useSeencheck
@@ -82,7 +84,7 @@ func (f *Frontier) Init(jobPath string, loggingChan chan *FrontierLogMessage, wo
 			return err
 		}
 
-		logrus.Info("seencheck initialized")
+		f.Log.Info("seencheck initialized")
 	}
 
 	f.FinishingQueueReader = new(utils.TAtomBool)

internal/pkg/log/elasticsearch.go

@@ -28,6 +28,7 @@ type ElasticsearchHandler struct {
 	level  slog.Level
 	attrs  []slog.Attr
 	groups []string
+	config *ElasticsearchConfig
 }
 
 // Handle is responsible for passing the log record to all underlying handlers.
@@ -146,3 +147,31 @@ func (h *ElasticsearchHandler) createIndex() error {
 
 	return nil
 }
+
+// Rotate implements the rotation for the Elasticsearch handler.
+// It updates the index name to use the current date and creates the new index if it doesn't exist.
+func (h *ElasticsearchHandler) Rotate() error {
+	newIndex := fmt.Sprintf("%s-%s", h.config.IndexPrefix, time.Now().Format("2006.01.02"))
+
+	// If the index name hasn't changed, no need to rotate
+	if newIndex == h.index {
+		return nil
+	}
+
+	// Update the index name
+	h.index = newIndex
+
+	// Create the new index
+	err := h.createIndex()
+	if err != nil {
+		return fmt.Errorf("failed to create new Elasticsearch index during rotation: %w", err)
+	}
+
+	return nil
+}
+
+// NextRotation calculates the next rotation time, which is the start of the next day
+func (h *ElasticsearchHandler) NextRotation() time.Time {
+	now := time.Now()
+	return time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()).Add(24 * time.Hour)
+}

internal/pkg/log/file.go

@@ -0,0 +1,52 @@
+package log
+
+import (
+	"fmt"
+	"log/slog"
+	"os"
+	"time"
+)
+
+type fileHandler struct {
+	slog.Handler
+	filename         string
+	file             *os.File
+	rotationInterval time.Duration
+	lastRotation     time.Time
+	level            slog.Level
+	logfile          *Logfile
+}
+
+type Logfile struct {
+	Dir    string
+	Prefix string
+}
+
+func (h *fileHandler) Rotate() error {
+	if h.file != nil {
+		h.file.Close()
+	}
+
+	h.filename = h.logfile.Filename()
+
+	file, err := os.OpenFile(h.filename, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
+	if err != nil {
+		return fmt.Errorf("failed to open new log file: %w", err)
+	}
+
+	h.file = file
+	h.Handler = slog.NewJSONHandler(file, &slog.HandlerOptions{
+		Level: h.level,
+	})
+
+	h.lastRotation = time.Now()
+	return nil
+}
+
+func (h *fileHandler) NextRotation() time.Time {
+	return h.lastRotation.Add(h.rotationInterval)
+}
+
+func (s *Logfile) Filename() string {
+	return fmt.Sprintf("%s/%s.%s.log", s.Dir, s.Prefix, time.Now().Format("2006.01.02-15h"))
+}