job type selection and web scraper job logging (#54)

* web scraper logging

* var assignment fix

Author: alvin-reyes

internal/jobs/webscraper.go

@@ -41,28 +41,46 @@ func NewWebScraper(jc types.JobConfiguration, statsCollector *stats.StatsCollect
 }
 
 func (ws *WebScraper) ExecuteJob(j types.Job) (types.JobResult, error) {
+	logrus.Info("Starting ExecuteJob for web scraper")
+
+	// Step 1: Unmarshal arguments
 	args := &WebScraperArgs{}
-	j.Arguments.Unmarshal(args)
+	logrus.Info("Unmarshaling job arguments")
+	if err := j.Arguments.Unmarshal(args); err != nil {
+		logrus.Errorf("Failed to unmarshal job arguments: %v", err)
+		return types.JobResult{Error: fmt.Sprintf("Invalid arguments: %v", err)}, err
+	}
+	logrus.Infof("Job arguments unmarshaled successfully: %+v", args)
 
+	// Step 2: Validate URL against blacklist
+	logrus.Info("Validating URL against blacklist")
 	for _, u := range ws.configuration.Blacklist {
+		logrus.Debugf("Checking if URL contains blacklisted term: %s", u)
 		if strings.Contains(args.URL, u) {
+			logrus.Warnf("URL %s is blacklisted due to term: %s", args.URL, u)
 			ws.stats.Add(stats.WebInvalid, 1)
 			logrus.Errorf("Blacklisted URL: %s", args.URL)
 			return types.JobResult{
 				Error: fmt.Sprintf("URL blacklisted: %s", args.URL),
 			}, nil
 		}
 	}
+	logrus.Infof("URL %s passed blacklist validation", args.URL)
 
+	// Step 3: Perform web scraping
+	logrus.Infof("Initiating web scraping for URL: %s with depth: %d", args.URL, args.Depth)
 	result, err := scrapeWeb([]string{args.URL}, args.Depth)
 	if err != nil {
+		logrus.Errorf("Web scraping failed for URL %s: %v", args.URL, err)
 		ws.stats.Add(stats.WebErrors, 1)
 		return types.JobResult{Error: err.Error()}, err
 	}
+	logrus.Infof("Web scraping succeeded for URL %s: %v", args.URL, result)
 
-	// Do the web scraping here
-	// For now, just return the URL
+	// Step 4: Process result and return
+	logrus.Info("Updating statistics for successful web scraping")
 	ws.stats.Add(stats.WebSuccess, 1)
+	logrus.Infof("Returning web scraping result for URL %s", args.URL)
 	return types.JobResult{
 		Data: result,
 	}, nil
@@ -107,64 +125,82 @@ type CollectedData struct {
 //		logrus.WithField("result", string(res)).Info("Scraping completed")
 //	}()
 func scrapeWeb(uri []string, depth int) ([]byte, error) {
+	logrus.Infof("Starting scrapeWeb with parameters: URIs=%v, Depth=%d", uri, depth)
 	// Set default depth to 1 if 0 is provided
 	if depth <= 0 {
+		logrus.Infof("Invalid depth (%d) provided, setting default depth to 1", depth)
 		depth = 1
 	}
 
+	logrus.Info("Initializing CollectedData struct")
 	var collectedData CollectedData
 
+	logrus.Info("Creating new Colly collector")
 	c := colly.NewCollector(
 		colly.Async(true), // Enable asynchronous requests
 		colly.AllowURLRevisit(),
 		colly.IgnoreRobotsTxt(),
 		colly.MaxDepth(depth),
 	)
+	logrus.Info("Colly collector created successfully")
 
 	// Adjust the parallelism and delay based on your needs and server capacity
+	logrus.Info("Setting scraping limits with parallelism and delay")
 	limitRule := colly.LimitRule{
 		DomainGlob:  "*",
 		Parallelism: 4,                      // Increased parallelism
 		Delay:       500 * time.Millisecond, // Reduced delay
 	}
+	logrus.Info("Applying scraping limits to the collector")
 	if err := c.Limit(&limitRule); err != nil {
 		logrus.Errorf("[-] Unable to set scraper limit. Using default. Error: %v", err)
 	}
 
 	// Increase the timeout slightly if necessary
-	c.SetRequestTimeout(240 * time.Second) // Increased to 4 minutes
+	logrus.Info("Setting request timeout to 240 seconds")
+	c.SetRequestTimeout(240 * time.Second)
 
 	// Initialize a backoff strategy
+	logrus.Info("Initializing exponential backoff strategy")
 	backoffStrategy := backoff.NewExponentialBackOff()
 
+	logrus.Info("Registering OnError callback to handle request errors")
 	c.OnError(func(r *colly.Response, err error) {
+		logrus.Errorf("Error occurred during request to URL: %s. StatusCode: %d, Error: %v", r.Request.URL, r.StatusCode, err)
 		if r.StatusCode == http.StatusTooManyRequests {
 			// Parse the Retry-After header (in seconds)
 			retryAfter, convErr := strconv.Atoi(r.Headers.Get("Retry-After"))
 			if convErr != nil {
 				// If not in seconds, it might be a date. Handle accordingly.
-				logrus.Debugf("[-] Retry-After: %s", r.Headers.Get("Retry-After"))
+				logrus.Warnf("Retry-After header is present but unrecognized format: %s", r.Headers.Get("Retry-After"))
 			}
 			// Calculate the next delay
 			nextDelay := backoffStrategy.NextBackOff()
 			if retryAfter > 0 {
 				nextDelay = time.Duration(retryAfter) * time.Second
 			}
-			logrus.Warnf("[-] Rate limited. Retrying after %v", nextDelay)
+			logrus.Warnf("Rate limited for URL: %s. Retrying after %v", r.Request.URL, nextDelay)
 			time.Sleep(nextDelay)
 			// Retry the request
+			logrus.Info("Retrying the request")
 			_ = r.Request.Retry()
+
 		} else {
+			logrus.Errorf("Request failed for URL: %s with error: %v", r.Request.URL, err)
 			logrus.Errorf("[-] Request URL: %s failed with error: %v", r.Request.URL, err)
 		}
 	})
 
+	logrus.Info("Registering OnHTML callback for h1, h2 elements (titles)")
 	c.OnHTML("h1, h2", func(e *colly.HTMLElement) {
+		logrus.Infof("Title (h1/h2) found: %s", e.Text)
 		// Directly append a new Section to collectedData.Sections
 		collectedData.Sections = append(collectedData.Sections, Section{Title: e.Text})
 	})
 
+	logrus.Info("Registering OnHTML callback for paragraph elements")
 	c.OnHTML("p", func(e *colly.HTMLElement) {
+		logrus.Infof("Paragraph detected: %s", e.Text)
 		// Check if there are any sections to append paragraphs to
 		if len(collectedData.Sections) > 0 {
 			// Get a reference to the last section
@@ -185,15 +221,19 @@ func scrapeWeb(uri []string, depth int) ([]byte, error) {
 		}
 	})
 
+	logrus.Info("Registering OnHTML callback for image elements")
 	c.OnHTML("img", func(e *colly.HTMLElement) {
+		logrus.Infof("Image detected with source URL: %s", e.Attr("src"))
 		imageURL := e.Request.AbsoluteURL(e.Attr("src"))
 		if len(collectedData.Sections) > 0 {
 			lastSection := &collectedData.Sections[len(collectedData.Sections)-1]
 			lastSection.Images = append(lastSection.Images, imageURL)
 		}
 	})
 
+	logrus.Info("Registering OnHTML callback for anchor elements")
 	c.OnHTML("a", func(e *colly.HTMLElement) {
+		logrus.Infof("Link detected: %s", e.Attr("href"))
 		pageURL := e.Request.AbsoluteURL(e.Attr("href"))
 		// Check if the URL protocol is supported (http or https)
 		if strings.HasPrefix(pageURL, "http://") || strings.HasPrefix(pageURL, "https://") {
@@ -202,16 +242,24 @@ func scrapeWeb(uri []string, depth int) ([]byte, error) {
 		}
 	})
 
+	logrus.Infof("Starting to visit URLs: %v", uri)
 	for _, u := range uri {
 		err := c.Visit(u)
+		logrus.Infof("Visiting URL: %s", u)
+		err = c.Visit(u)
 		if err != nil {
+			logrus.Errorf("Failed to visit URL: %s. Error: %v", u, err)
 			return nil, err
 		}
 	}
 
 	// Wait for all requests to finish
+	logrus.Info("Waiting for all requests to complete")
 	c.Wait()
 
+	logrus.Info("Scraping completed, marshaling collected data into JSON format")
 	j, _ := json.Marshal(collectedData)
+
+	logrus.Infof("Scraping successful. Returning data for URIs: %v", uri)
 	return j, nil
 }

internal/jobserver/jobserver.go

@@ -2,6 +2,7 @@ package jobserver
 
 import (
 	"context"
+	"github.com/sirupsen/logrus"
 	"sync"
 
 	"github.com/google/uuid"
@@ -28,35 +29,61 @@ type jobWorkerEntry struct {
 }
 
 func NewJobServer(workers int, jc types.JobConfiguration) *JobServer {
+	logrus.Info("Initializing JobServer...")
+
+	// Validate and set worker count
 	if workers <= 0 {
+		logrus.Infof("Invalid worker count (%d), defaulting to 1 worker.", workers)
 		workers = 1
+	} else {
+		logrus.Infof("Setting worker count to %d.", workers)
 	}
 
+	// Retrieve and set buffer size for stats collector
 	bufSize, ok := jc["stats_buf_size"].(uint)
 	if !ok {
+		logrus.Info("stats_buf_size not provided or invalid in JobConfiguration. Defaulting to 128.")
 		bufSize = 128
+	} else {
+		logrus.Infof("Using stats_buf_size: %d.", bufSize)
 	}
 
 	// Start stats collector
+	logrus.Info("Starting stats collector...")
 	s := stats.StartCollector(bufSize, jc)
+	logrus.Info("Stats collector started successfully.")
 
 	// Set worker ID in stats collector if available
 	if workerID, ok := jc["worker_id"].(string); ok && workerID != "" {
+		logrus.Infof("Setting worker ID to '%s' in stats collector.", workerID)
 		s.SetWorkerID(workerID)
+	} else {
+		logrus.Info("No worker ID found in JobConfiguration.")
+	}
+
+	// Initialize job workers
+	logrus.Info("Setting up job workers...")
+	jobworkers := make(map[string]*jobWorkerEntry)
+
+	jobworkers[jobs.WebScraperType] = &jobWorkerEntry{
+		w: jobs.NewWebScraper(jc, s),
 	}
+	logrus.Infof("Initialized job worker for: %s", jobs.WebScraperType)
 
-	jobworkers := map[string]*jobWorkerEntry{
-		jobs.WebScraperType: {
-			w: jobs.NewWebScraper(jc, s),
-		},
-		jobs.TwitterScraperType: {
-			w: jobs.NewTwitterScraper(jc, s),
-		},
-		jobs.TelemetryJobType: {
-			w: jobs.NewTelemetryJob(jc, s),
-		},
+	jobworkers[jobs.TwitterScraperType] = &jobWorkerEntry{
+		w: jobs.NewTwitterScraper(jc, s),
 	}
+	logrus.Infof("Initialized job worker for: %s", jobs.TwitterScraperType)
+
+	jobworkers[jobs.TelemetryJobType] = &jobWorkerEntry{
+		w: jobs.NewTelemetryJob(jc, s),
+	}
+	logrus.Infof("Initialized job worker for: %s", jobs.TelemetryJobType)
+
+	logrus.Info("Job workers setup completed.")
 
+	// Return the JobServer instance
+	logrus.Info("JobServer initialization complete.")
 	return &JobServer{
 		jobChan:          make(chan types.Job),
 		results:          make(map[string]types.JobResult),