WIP: Allow for per package error handling

feeds/feed.go

@@ -16,7 +16,7 @@ type UnsupportedOptionError struct {
 }
 
 type ScheduledFeed interface {
-	Latest(cutoff time.Time) ([]*Package, error)
+	Latest(cutoff time.Time) ([]*Package, []error)
 	GetFeedOptions() FeedOptions
 	GetName() string
 }

feeds/pypi/pypi.go

@@ -8,8 +8,6 @@ import (
 	"strings"
 	"time"
 
-	log "github.com/sirupsen/logrus"
-
 	"github.com/ossf/package-feeds/events"
 	"github.com/ossf/package-feeds/feeds"
 	"github.com/ossf/package-feeds/utils"
@@ -167,41 +165,39 @@ func New(feedOptions feeds.FeedOptions, eventHandler *events.Handler) (*Feed, er
 	}, nil
 }
 
-func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, error) {
+func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
 	pkgs := []*feeds.Package{}
 	var pypiPackages []*Package
-	var err error
 	var errs []error
+	var err error
 
 	if feed.packages == nil {
 		// Firehose fetch all packages.
 		// If this fails then we need to return, as it's the only source of
 		// data.
 		pypiPackages, err = fetchPackages(feed.baseURL)
 		if err != nil {
-			return nil, err
+			return nil, append(errs, err)
 		}
 	} else {
 		// Fetch specific packages individually from configured packages list.
 		pypiPackages, errs = fetchCriticalPackages(feed.baseURL, *feed.packages)
-		for _, err := range errs {
-			// TODO: This could be an entry point for a 'lossy' type of event.
-			log.Errorf(err.Error())
-		}
 		if len(pypiPackages) == 0 {
 			// If none of the packages were successfully polled for, return early.
-			return nil, feeds.ErrNoPackagesPolled
+			return nil, append(errs, feeds.ErrNoPackagesPolled)
 		}
 	}
 
 	for _, pkg := range pypiPackages {
 		pkgName, err := pkg.Name()
 		if err != nil {
-			return nil, err
+			errs = append(errs, err)
+			continue
 		}
 		pkgVersion, err := pkg.Version()
 		if err != nil {
-			return nil, err
+			errs = append(errs, err)
+			continue
 		}
 		pkg := feeds.NewPackage(pkg.CreatedDate.Time, pkgName, pkgVersion, FeedName)
 		pkgs = append(pkgs, pkg)
@@ -213,7 +209,7 @@ func (feed Feed) Latest(cutoff time.Time) ([]*feeds.Package, error) {
 	}
 
 	pkgs = feeds.ApplyCutoff(pkgs, cutoff)
-	return pkgs, nil
+	return pkgs, errs
 }
 
 func (feed Feed) GetPackageList() *[]string {

feeds/pypi/pypi_test.go

@@ -26,8 +26,8 @@ func TestPypiLatest(t *testing.T) {
 	feed.baseURL = srv.URL
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	pkgs, err := feed.Latest(cutoff)
-	if err != nil {
+	pkgs, errs := feed.Latest(cutoff)
+	if len(errs) != 0 {
 		t.Fatalf("feed.Latest returned error: %v", err)
 	}
 
@@ -73,8 +73,8 @@ func TestPypiCriticalLatest(t *testing.T) {
 	feed.baseURL = srv.URL
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	pkgs, err := feed.Latest(cutoff)
-	if err != nil {
+	pkgs, errs := feed.Latest(cutoff)
+	if len(errs) != 0 {
 		t.Fatalf("Failed to call Latest() with err: %v", err)
 	}
 
@@ -126,11 +126,11 @@ func TestPypiAllNotFound(t *testing.T) {
 	feed.baseURL = srv.URL
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	_, err = feed.Latest(cutoff)
-	if err == nil {
-		t.Fatalf("feed.Latest() was successful when an error was expected")
+	_, errs := feed.Latest(cutoff)
+	if len(errs) == 0 {
+		t.Fatalf("feed.Latest() was successful when errors were expected")
 	}
-	if !errors.Is(err, feeds.ErrNoPackagesPolled) {
+	if !errors.Is(errs[len(errs)-1], feeds.ErrNoPackagesPolled) {
 		t.Fatalf("feed.Latest() returned an error which did not match the expected error")
 	}
 }
@@ -157,9 +157,9 @@ func TestPypiPartialNotFound(t *testing.T) {
 	feed.baseURL = srv.URL
 
 	cutoff := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
-	pkgs, err := feed.Latest(cutoff)
-	if err != nil {
-		t.Fatalf("feed.Latest() was unsuccessful when an error wasn't expected")
+	pkgs, errs := feed.Latest(cutoff)
+	if len(errs) != 1 {
+		t.Fatalf("feed.Latest() returned no errors when an error was expected")
 	}
 	if len(pkgs) != 2 {
 		t.Fatalf("Latest() produced %v packages instead of the expected %v", len(pkgs), 2)

feeds/scheduler/feed_group.go

@@ -38,10 +38,9 @@ func (fg *FeedGroup) Run() {
 }
 
 func (fg *FeedGroup) PollAndPublish() (int, error) {
-	pkgs, errs := fg.poll()
-	if len(errs) > 0 {
-		return 0, errs[0]
-	} else if len(pkgs) == 0 {
+	// poll() logs errs, so no need to handle.
+	pkgs, _ := fg.poll()
+	if len(pkgs) == 0 {
 		return 0, nil
 	}
 
@@ -64,7 +63,7 @@ func (fg *FeedGroup) poll() ([]*feeds.Package, []error) {
 				name: feed.GetName(),
 				feed: feed,
 			}
-			result.packages, result.err = feed.Latest(fg.lastPoll)
+			result.packages, result.errs = feed.Latest(fg.lastPoll)
 			results <- result
 		}(feed)
 	}
@@ -74,10 +73,9 @@ func (fg *FeedGroup) poll() ([]*feeds.Package, []error) {
 		result := <-results
 
 		logger := log.WithField("feed", result.name)
-		if result.err != nil {
-			logger.WithError(result.err).Error("Error fetching packages")
-			errs = append(errs, result.err)
-			continue
+		for _, err := range result.errs {
+			logger.WithError(err).Error("Error fetching packages")
+			errs = append(errs, err)
 		}
 		for _, pkg := range result.packages {
 			log.WithFields(log.Fields{

feeds/scheduler/feed_group_test.go

@@ -54,7 +54,7 @@ func TestFeedGroupPollWithErr(t *testing.T) {
 
 	mockFeeds := []feeds.ScheduledFeed{
 		mockFeed{
-			err: errPackage,
+			errs: []error{errPackage},
 		},
 		mockFeed{
 			packages: []*feeds.Package{

feeds/scheduler/mocks.go

@@ -9,7 +9,7 @@ import (
 
 type mockFeed struct {
 	packages []*feeds.Package
-	err      error
+	errs     []error
 	options  feeds.FeedOptions
 }
 
@@ -21,8 +21,8 @@ func (feed mockFeed) GetFeedOptions() feeds.FeedOptions {
 	return feed.options
 }
 
-func (feed mockFeed) Latest(cutoff time.Time) ([]*feeds.Package, error) {
-	return feed.packages, feed.err
+func (feed mockFeed) Latest(cutoff time.Time) ([]*feeds.Package, []error) {
+	return feed.packages, feed.errs
 }
 
 type mockPublisher struct {

feeds/scheduler/scheduler.go

@@ -33,7 +33,7 @@ type pollResult struct {
 	name     string
 	feed     feeds.ScheduledFeed
 	packages []*feeds.Package
-	err      error
+	errs     []error
 }
 
 // Runs several services for the operation of scheduler, this call is blocking until application exit