module/apmlogrus: introduce logrus integration

Introduce a new package which provides integrations for logrus:

 - Hook, which implements logrus.Hook, to reports error messages.
 - TraceContext, which extracts logrus.Fields from transactions
   and spans in the provided context.

CHANGELOG.md

@@ -12,6 +12,7 @@
  - Add Kubernetes pod info to metadata (#342)
  - module/apmsql: don't report driver.ErrBadConn, context.Canceled (#346, #348)
  - Added ErrorLogRecord.Error field, for associating an error value with a log record (#380)
+ - module/apmlogrus: introduce logrus exception-tracking hook, and log correlation (#381)
 
 ## [v1.0.0](https://github.com/elastic/apm-agent-go/releases/tag/v1.0.0)
 

apmtest/discard.go

@@ -0,0 +1,23 @@
+package apmtest
+
+import (
+	"log"
+
+	"go.elastic.co/apm"
+	"go.elastic.co/apm/transport/transporttest"
+)
+
+// DiscardTracer is an apm.Tracer that discards all events.
+//
+// This tracer may be used by multiple tests, and so should
+// not be modified or closed.
+var DiscardTracer *apm.Tracer
+
+func init() {
+	tracer, err := apm.NewTracer("", "")
+	if err != nil {
+		log.Fatal(err)
+	}
+	tracer.Transport = transporttest.Discard
+	DiscardTracer = tracer
+}

module/apmlogrus/example_test.go

@@ -0,0 +1,47 @@
+package apmlogrus_test
+
+import (
+	"context"
+
+	"github.com/sirupsen/logrus"
+
+	"go.elastic.co/apm"
+	"go.elastic.co/apm/module/apmlogrus"
+)
+
+func ExampleHook() {
+	logger := logrus.New()
+
+	// Report "error", "panic", and "fatal" log messages
+	// to Elastic APM using apm.DefaultTracer.
+	logger.AddHook(&apmlogrus.Hook{})
+
+	// Report "error", "panic", and "fatal" log messages
+	// to Elastic APM using a specific tracer.
+	var tracer *apm.Tracer
+	logger.AddHook(&apmlogrus.Hook{
+		Tracer: tracer,
+	})
+
+	// Report only "panic" log messages to Elastic APM
+	// using apm.DefaultTracer.
+	logger.AddHook(&apmlogrus.Hook{
+		LogLevels: []logrus.Level{logrus.PanicLevel},
+	})
+}
+
+func ExampleTraceContext() {
+	logger := logrus.New()
+
+	tx := apm.DefaultTracer.StartTransaction("name", "type")
+	defer tx.End()
+
+	ctx := apm.ContextWithTransaction(context.Background(), tx)
+	span, ctx := apm.StartSpan(ctx, "name", "type")
+	defer span.End()
+
+	// apmlogrus.TraceContext returns fields including the trace ID,
+	// transaction ID, and span ID, for the transaction and span in
+	// the given context.
+	logger.WithFields(apmlogrus.TraceContext(ctx)).Fatal("ohhh, what a world")
+}

module/apmlogrus/fields.go

@@ -0,0 +1,38 @@
+package apmlogrus
+
+import (
+	"context"
+
+	"github.com/sirupsen/logrus"
+
+	"go.elastic.co/apm"
+)
+
+const (
+	// FieldKeyTraceID is the field key for the trace ID.
+	FieldKeyTraceID = "trace.id"
+
+	// FieldKeyTransactionID is the field key for the transaction ID.
+	FieldKeyTransactionID = "transaction.id"
+
+	// FieldKeySpanID is the field key for the span ID.
+	FieldKeySpanID = "span.id"
+)
+
+// TraceContext returns a logrus.Fields containing the trace
+// context of the transaction and span contained in ctx, if any.
+func TraceContext(ctx context.Context) logrus.Fields {
+	tx := apm.TransactionFromContext(ctx)
+	if tx == nil {
+		return nil
+	}
+	traceContext := tx.TraceContext()
+	fields := logrus.Fields{
+		FieldKeyTraceID:       traceContext.Trace,
+		FieldKeyTransactionID: traceContext.Span,
+	}
+	if span := apm.SpanFromContext(ctx); span != nil {
+		fields[FieldKeySpanID] = span.TraceContext().Span
+	}
+	return fields
+}

module/apmlogrus/fields_test.go

@@ -0,0 +1,84 @@
+package apmlogrus_test
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"go.elastic.co/apm"
+	"go.elastic.co/apm/apmtest"
+	"go.elastic.co/apm/module/apmlogrus"
+)
+
+func TestTraceContext(t *testing.T) {
+	var buf bytes.Buffer
+	logger := newLogger(&buf)
+
+	tx, spans, _ := apmtest.WithTransaction(func(ctx context.Context) {
+		span, ctx := apm.StartSpan(ctx, "name", "type")
+		defer span.End()
+		logger.WithTime(time.Unix(0, 0).UTC()).WithFields(apmlogrus.TraceContext(ctx)).Debug("beep")
+	})
+	require.Len(t, spans, 1)
+
+	assert.Equal(t,
+		fmt.Sprintf(
+			`{"level":"debug","msg":"beep","span.id":"%x","time":"1970-01-01T00:00:00Z","trace.id":"%x","transaction.id":"%x"}`+"\n",
+			spans[0].ID[:], tx.TraceID[:], tx.ID[:],
+		),
+		buf.String(),
+	)
+}
+
+func TestTraceContextTextFormatter(t *testing.T) {
+	var buf bytes.Buffer
+	logger := newLogger(&buf)
+	logger.Formatter = &logrus.TextFormatter{}
+
+	tx, spans, _ := apmtest.WithTransaction(func(ctx context.Context) {
+		span, ctx := apm.StartSpan(ctx, "name", "type")
+		defer span.End()
+		logger.WithTime(time.Unix(0, 0).UTC()).WithFields(apmlogrus.TraceContext(ctx)).Debug("beep")
+	})
+	require.Len(t, spans, 1)
+
+	assert.Equal(t,
+		fmt.Sprintf(
+			"time=\"1970-01-01T00:00:00Z\" level=debug msg=beep span.id=%x trace.id=%x transaction.id=%x\n",
+			spans[0].ID[:], tx.TraceID[:], tx.ID[:],
+		),
+		buf.String(),
+	)
+}
+
+func TestTraceContextNoSpan(t *testing.T) {
+	var buf bytes.Buffer
+	logger := newLogger(&buf)
+	tx, _, _ := apmtest.WithTransaction(func(ctx context.Context) {
+		logger.WithTime(time.Unix(0, 0).UTC()).WithFields(apmlogrus.TraceContext(ctx)).Debug("beep")
+	})
+
+	assert.Equal(t,
+		fmt.Sprintf(
+			`{"level":"debug","msg":"beep","time":"1970-01-01T00:00:00Z","trace.id":"%x","transaction.id":"%x"}`+"\n",
+			tx.TraceID[:], tx.ID[:],
+		),
+		buf.String(),
+	)
+}
+
+func TestTraceContextEmpty(t *testing.T) {
+	var buf bytes.Buffer
+	logger := newLogger(&buf)
+
+	// apmlogrus.TraceContext will return nil if the context does not contain a transaction.
+	ctx := context.Background()
+	logger.WithTime(time.Unix(0, 0).UTC()).WithFields(apmlogrus.TraceContext(ctx)).Debug("beep")
+	assert.Equal(t, `{"level":"debug","msg":"beep","time":"1970-01-01T00:00:00Z"}`+"\n", buf.String())
+}

module/apmlogrus/hook.go

@@ -0,0 +1,115 @@
+package apmlogrus
+
+import (
+	"context"
+	"time"
+
+	"github.com/sirupsen/logrus"
+
+	"go.elastic.co/apm"
+	"go.elastic.co/apm/stacktrace"
+)
+
+var (
+	// DefaultLogLevels is the log levels for which errors are reported by Hook, if Hook.LogLevels is not set.
+	DefaultLogLevels = []logrus.Level{
+		logrus.PanicLevel,
+		logrus.FatalLevel,
+		logrus.ErrorLevel,
+	}
+)
+
+const (
+	// DefaultFatalFlushTimeout is the default value for Hook.FatalFlushTimeout.
+	DefaultFatalFlushTimeout = 5 * time.Second
+)
+
+func init() {
+	stacktrace.RegisterLibraryPackage("github.com/sirupsen/logrus")
+}
+
+// Hook implements logrus.Hook, reporting log records as errors
+// to the APM Server. If TraceContext is used to add trace IDs
+// to the log records, the errors reported will be associated
+// with them.
+type Hook struct {
+	// Tracer is the apm.Tracer to use for reporting errors.
+	// If Tracer is nil, then apm.DefaultTracer will be used.
+	Tracer *apm.Tracer
+
+	// LogLevels holds the log levels to report as errors.
+	// If LogLevels is nil, then the DefaultLogLevels will
+	// be used.
+	LogLevels []logrus.Level
+
+	// FatalFlushTimeout is the amount of time to wait while
+	// flushing a fatal log message to the APM Server before
+	// the process is exited. If this is 0, then
+	// DefaultFatalFlushTimeout will be used. If the timeout
+	// is a negative value, then no flushing will be performed.
+	FatalFlushTimeout time.Duration
+}
+
+func (h *Hook) tracer() *apm.Tracer {
+	tracer := h.Tracer
+	if tracer == nil {
+		tracer = apm.DefaultTracer
+	}
+	return tracer
+}
+
+// Levels returns h.LogLevels, satisfying the logrus.Hook interface.
+func (h *Hook) Levels() []logrus.Level {
+	levels := h.LogLevels
+	if levels == nil {
+		levels = DefaultLogLevels
+	}
+	return levels
+}
+
+// Fire reports the log entry as an error to the APM Server.
+func (h *Hook) Fire(entry *logrus.Entry) error {
+	tracer := h.tracer()
+	if !tracer.Active() {
+		return nil
+	}
+
+	err, _ := entry.Data[logrus.ErrorKey].(error)
+	errlog := tracer.NewErrorLog(apm.ErrorLogRecord{
+		Message: entry.Message,
+		Level:   entry.Level.String(),
+		Error:   err,
+	})
+	errlog.Handled = true
+	errlog.Timestamp = entry.Time
+	errlog.SetStacktrace(1)
+
+	// Extract trace context added with apmlogrus.TraceContext,
+	// and include it in the reported error.
+	if traceID, ok := entry.Data[FieldKeyTraceID].(apm.TraceID); ok {
+		errlog.TraceID = traceID
+	}
+	if transactionID, ok := entry.Data[FieldKeyTransactionID].(apm.SpanID); ok {
+		errlog.TransactionID = transactionID
+		errlog.ParentID = transactionID
+	}
+	if spanID, ok := entry.Data[FieldKeySpanID].(apm.SpanID); ok {
+		errlog.ParentID = spanID
+	}
+
+	errlog.Send()
+	if entry.Level == logrus.FatalLevel {
+		// In its default configuration, logrus will exit the process
+		// following a fatal log message, so we flush the tracer.
+		flushTimeout := h.FatalFlushTimeout
+		if flushTimeout == 0 {
+			flushTimeout = DefaultFatalFlushTimeout
+		}
+		if flushTimeout >= 0 {
+			ctx, cancel := context.WithTimeout(context.Background(), flushTimeout)
+			defer cancel()
+			tracer.Flush(ctx.Done())
+		}
+	}
+	return nil
+}