ingest: Render captive core logs as Horizon logs

ingest/ledgerbackend/captive_core_backend.go

@@ -139,6 +139,20 @@ func NewCaptive(executablePath, configPath, networkPassphrase string, historyURL
 	}, nil
 }
 
+func (c *CaptiveStellarCore) SetLogger(logger *log.Entry) {
+	// If the caller decides to set a custom logger, we recreate the factory to
+	// pass this along to all subsequently-created instances.
+	previousFactory := c.stellarCoreRunnerFactory
+	c.stellarCoreRunnerFactory = func(configPath string) (stellarCoreRunnerInterface, error) {
+		core, err := previousFactory(configPath)
+		if err != nil {
+			return core, err
+		}
+		core.setLogger(logger)
+		return core, nil
+	}
+}
+
 func (c *CaptiveStellarCore) getLatestCheckpointSequence() (uint32, error) {
 	has, err := c.archive.GetRootHAS()
 	if err != nil {

ingest/ledgerbackend/captive_core_backend_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/stellar/go/historyarchive"
 	"github.com/stellar/go/network"
 	"github.com/stellar/go/support/errors"
+	"github.com/stellar/go/support/log"
 	"github.com/stellar/go/xdr"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -47,6 +48,8 @@ func (m *stellarCoreRunnerMock) close() error {
 	return a.Error(0)
 }
 
+func (m *stellarCoreRunnerMock) setLogger(*log.Entry) {}
+
 func buildLedgerCloseMeta(sequence uint32) xdr.LedgerCloseMeta {
 	opResults := []xdr.OperationResult{}
 	opMeta := []xdr.OperationMeta{}

ingest/ledgerbackend/stellar_core_runner.go

@@ -15,13 +15,15 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
+	"github.com/stellar/go/support/log"
 )
 
 type stellarCoreRunnerInterface interface {
 	catchup(from, to uint32) error
 	runFrom(from uint32, hash string) error
 	getMetaPipe() io.Reader
 	getProcessExitChan() <-chan error
+	setLogger(*log.Entry)
 	close() error
 }
 
@@ -42,6 +44,9 @@ type stellarCoreRunner struct {
 	metaPipe    io.Reader
 	tempDir     string
 	nonce       string
+
+	// Optionally, logging can be done to something other than stdout.
+	Log *log.Entry
 }
 
 func newStellarCoreRunner(executablePath, configPath, networkPassphrase string, historyURLs []string) (*stellarCoreRunner, error) {
@@ -106,9 +111,10 @@ func (r *stellarCoreRunner) getConfFileName() string {
 	return filepath.Join(r.tempDir, "stellar-core.conf")
 }
 
-func (*stellarCoreRunner) getLogLineWriter() io.Writer {
-	r, w := io.Pipe()
-	br := bufio.NewReader(r)
+func (r *stellarCoreRunner) getLogLineWriter() io.Writer {
+	rd, wr := io.Pipe()
+	br := bufio.NewReader(rd)
+
 	// Strip timestamps from log lines from captive stellar-core. We emit our own.
 	dateRx := regexp.MustCompile(`^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3} `)
 	go func() {
@@ -118,10 +124,45 @@ func (*stellarCoreRunner) getLogLineWriter() io.Writer {
 				break
 			}
 			line = dateRx.ReplaceAllString(line, "")
-			fmt.Print(line)
+
+			// If there's a logger, we attempt to extract metadata about the log
+			// entry, then redirect it to the logger. Otherwise, we just use stdout.
+			if r.Log == nil {
+				fmt.Print(line)
+				continue
+			}
+
+			levelRx := regexp.MustCompile(`\[(\w+) ([A-Z]+)\]`)
+			indices := levelRx.FindStringSubmatchIndex(line)
+			if len(indices) >= 6 {
+				// Identify the indices that match our regex subexpressions
+				categoryIdx := indices[2:4]
+				levelIdx := indices[4:6]
+
+				// Extract the substrings from the log entry
+				category := line[categoryIdx[0]:categoryIdx[1]]
+				level := line[levelIdx[0]:levelIdx[1]]
+
+				line = line[indices[1]+1:] // dump the matched part of the line
+
+				levelMapping := map[string]func(string, ...interface{}){
+					"FATAL":   r.Log.Errorf,
+					"ERROR":   r.Log.Errorf,
+					"WARNING": r.Log.Warnf,
+					"INFO":    r.Log.Infof,
+				}
+
+				if writer, ok := levelMapping[strings.ToUpper(level)]; ok {
+					writer("%s: %s", category, line)
+				} else {
+					r.Log.Infof(line)
+				}
+			} else {
+				r.Log.Infof(line)
+			}
 		}
 	}()
-	return w
+	return wr
 }
 
 // Makes the temp directory and writes the config file to it; called by the
@@ -221,6 +262,14 @@ func (r *stellarCoreRunner) getProcessExitChan() <-chan error {
 	return r.processExit
 }
 
+func (r *stellarCoreRunner) setLogger(logger *log.Entry) {
+	if logger != nil {
+		r.Log = logger.WithField("subservice", "stellar-core")
+	} else {
+		r.Log = nil
+	}
+}
+
 func (r *stellarCoreRunner) close() error {
 	var err1, err2 error
 

services/horizon/internal/ingest/main.go

@@ -171,8 +171,8 @@ func NewSystem(config Config) (System, error) {
 				return nil, errors.Wrap(err, "error creating captive core backend")
 			}
 		} else {
-			//
-			ledgerBackend, err = ledgerbackend.NewCaptive(
+			var captiveCoreBackend *ledgerbackend.CaptiveStellarCore
+			captiveCoreBackend, err = ledgerbackend.NewCaptive(
 				config.StellarCoreBinaryPath,
 				config.StellarCoreConfigPath,
 				config.NetworkPassphrase,
@@ -182,6 +182,8 @@ func NewSystem(config Config) (System, error) {
 				cancel()
 				return nil, errors.Wrap(err, "error creating captive core backend")
 			}
+			captiveCoreBackend.SetLogger(log)
+			ledgerBackend = captiveCoreBackend
 		}
 	} else {
 		coreSession := config.CoreSession.Clone()