Make logging even more verbose when computing item signals

#40
brawer · May 16, 2024 · b5537d7 · b5537d7
1 parent b13f38b
commit b5537d7
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 3 deletions.
diff --git a/cmd/qrank-builder/itemsignals.go b/cmd/qrank-builder/itemsignals.go
@@ -198,9 +198,10 @@ func buildItemSignals(ctx context.Context, pageviews []string, sites *map[string
 		joiner := itemSignalsJoiner{out: sigChan}
 		var linesMerged int64
 		for merger.Advance() {
-			if err := joiner.Process(merger.Line()); err != nil {
+			line := merger.Line()
+			if err := joiner.Process(line); err != nil {
 				joiner.Close()
-				logger.Printf("ItemSignalsJoiner.Process() failed: %v", err)
+				logger.Printf(`ItemSignalsJoiner.Process("%s") failed: %v`, line, err)
 				return err
 			}
 			linesMerged += 1
@@ -219,7 +220,10 @@ func buildItemSignals(ctx context.Context, pageviews []string, sites *map[string
 		for {
 			select {
 			case <-groupCtx.Done():
-				return groupCtx.Err()
+				err := groupCtx.Err()
+				logger.Printf("BuildItemSignals(): sorting canceled, groupCtx.Err()=%v", err)
+				return err
+
 			case s, more := <-outChan:
 				if !more {
 					err := writer.Close()
@@ -236,6 +240,7 @@ func buildItemSignals(ctx context.Context, pageviews []string, sites *map[string
 		}
 	})
 	if err := group.Wait(); err != nil {
+		logger.Printf("BuildItemSignals(): group.Wait() failed, err==%v", err)
 		return time.Time{}, err
 	}
 	if err := <-errChan; err != nil {

diff --git a/cmd/qrank-builder/pagesignals.go b/cmd/qrank-builder/pagesignals.go
@@ -41,6 +41,7 @@ func buildPageSignals(ctx context.Context, dumps string, sites *map[string]WikiS
 			for {
 				select {
 				case <-groupCtx.Done():
+					logger.Printf("BuildPageSignals(): canceled, groupCtx.Err()=%v", groupCtx.Err())
 					return groupCtx.Err()
 
 				case t, more := <-tasks:
@@ -125,19 +126,23 @@ func buildSitePageSignals(site WikiSite, ctx context.Context, dumps string, s3 S
 		for {
 			select {
 			case <-groupCtx.Done():
+				logger.Printf("BuildSitePageSignals(): canceled, groupCtx.Err()=%v", groupCtx.Err())
 				return groupCtx.Err()
+
 			case line, more := <-outChan:
 				if !more {
 					return merger.Close()
 				}
 				err := merger.Process(line)
 				if err != nil {
+					logger.Printf(`BuildSitePageSignals(): merger.Process("%s") failed, err=%v`, line, err)
 					return err
 				}
 			}
 		}
 	})
 	if err := group.Wait(); err != nil {
+		logger.Printf(`BuildSitePageSignals(): group.Wait() failed, err=%v`, err)
 		return err
 	}
 	if err := <-errChan; err != nil {
@@ -360,6 +365,7 @@ func NewPageSignalsScanner(sites *map[string]WikiSite, s3 S3) *pageSignalsScanne
 func (s *pageSignalsScanner) Scan() bool {
 	s.curLine.Truncate(0)
 	if s.err != nil {
+		logger.Printf("PageSignalsScanner.Scan(): early exit due to err=%v", s.err)
 		return false
 	}
 	for s.curDomain < len(s.domains) {
@@ -401,6 +407,7 @@ func (s *pageSignalsScanner) Scan() bool {
 		s.scanner = bufio.NewScanner(s.decompressor)
 	}
 
+	logger.Printf("PageSignalsScanner.Scan(): cleaning up")
 	if s.decompressor != nil {
 		s.decompressor.Close()
 		s.decompressor = nil