Tweak line parsing to not create heap memory

pkg/extractor/extractor.go

@@ -72,7 +72,7 @@ func indexToSlices(s string, indexMatches []int) []string {
 }
 
 // async safe
-func (s *Extractor) processLineSync(line BString) *Match {
+func (s *Extractor) processLineSync(line BString) (Match, bool) {
 	lineNum := atomic.AddUint64(&s.readLines, 1)
 	matches := s.regex.FindSubmatchIndex(line)
 
@@ -88,23 +88,23 @@ func (s *Extractor) processLineSync(line BString) *Match {
 
 			if len(extractedKey) > 0 {
 				matchNum := atomic.AddUint64(&s.matchedLines, 1)
-				return &Match{
+				return Match{
 					bLine:       line,
 					Line:        lineStringPtr,
 					Groups:      slices,
 					Indices:     matches,
 					Extracted:   extractedKey,
 					LineNumber:  lineNum,
 					MatchNumber: matchNum,
-				}
+				}, true
 			}
 
 			atomic.AddUint64(&s.ignoredLines, 1)
 		} else {
 			atomic.AddUint64(&s.ignoredLines, 1)
 		}
 	}
-	return nil
+	return Match{}, false
 }
 
 // New an extractor from an input channel
@@ -135,13 +135,12 @@ func New(inputBatch <-chan []BString, config *Config) (*Extractor, error) {
 
 				var matchBatch []Match
 				for _, s := range batch {
-					match := extractor.processLineSync(s)
-					if match != nil {
+					if match, ok := extractor.processLineSync(s); ok {
 						if matchBatch == nil {
 							// Initialize to expected cap (only if we have any matches)
 							matchBatch = make([]Match, 0, len(batch))
 						}
-						matchBatch = append(matchBatch, *match)
+						matchBatch = append(matchBatch, match)
 					}
 				}
 				if len(matchBatch) > 0 {

profiling.go

@@ -18,7 +18,7 @@ func startProfiler(basename string) {
 		idx := 0
 		for {
 			select {
-			case <-time.After(100 * time.Millisecond):
+			case <-time.After(500 * time.Millisecond):
 				filename := fmt.Sprintf("%s_%03d.prof", basename, idx)
 				if f, err := os.Create(filename); err == nil {
 					pprof.WriteHeapProfile(f)