loggingexporter: Add Log Support (#1298)

This makes the loggingexporter able to output in a log pipeline.

exporter/loggingexporter/factory.go

@@ -107,3 +107,21 @@ func (f *Factory) CreateMetricsExporter(_ context.Context, _ component.ExporterC
 	}
 	return lexp, nil
 }
+
+// CreateLogExporter creates a log exporter based on this config.
+func (f *Factory) CreateLogExporter(_ context.Context, _ component.ExporterCreateParams, config configmodels.Exporter) (component.LogExporter, error) {
+	cfg := config.(*Config)
+
+	exporterLogger, err := f.createLogger(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	lexp, err := NewLogExporter(config, cfg.LogLevel, exporterLogger)
+	if err != nil {
+		return nil, err
+	}
+	return lexp, nil
+}
+
+var _ component.LogExporterFactory = (*Factory)(nil)

exporter/loggingexporter/factory_test.go

@@ -49,3 +49,12 @@ func TestCreateTraceExporter(t *testing.T) {
 	assert.NoError(t, err)
 	assert.NotNil(t, te)
 }
+
+func TestCreateLogExporter(t *testing.T) {
+	factory := &Factory{}
+	cfg := factory.CreateDefaultConfig()
+
+	te, err := factory.CreateLogExporter(context.Background(), component.ExporterCreateParams{Logger: zap.NewNop()}, cfg)
+	assert.NoError(t, err)
+	assert.NotNil(t, te)
+}

exporter/loggingexporter/logging_exporter.go

@@ -29,6 +29,7 @@ import (
 	"go.opentelemetry.io/collector/consumer/pdata"
 	"go.opentelemetry.io/collector/consumer/pdatautil"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.opentelemetry.io/collector/internal/data"
 )
 
 type logDataBuffer struct {
@@ -211,6 +212,14 @@ func (b *logDataBuffer) logDataPointLabels(labels pdata.StringMap) {
 	b.logStringMap("Data point labels", labels)
 }
 
+func (b *logDataBuffer) logLogRecord(lr pdata.LogRecord) {
+	b.logEntry("Timestamp: %d", lr.Timestamp())
+	b.logEntry("Severity: %s", lr.SeverityText())
+	b.logEntry("ShortName: %s", lr.ShortName())
+	b.logEntry("Body: %s", lr.Body())
+	b.logAttributeMap("Attributes", lr.Attributes())
+}
+
 func attributeValueToString(av pdata.AttributeValue) string {
 	switch av.Type() {
 	case pdata.AttributeValueSTRING:
@@ -402,3 +411,57 @@ func loggerSync(logger *zap.Logger) func(context.Context) error {
 		return err
 	}
 }
+
+// NewLogExporter creates an exporter.LogExporter that just drops the
+// received data and logs debugging messages.
+func NewLogExporter(config configmodels.Exporter, level string, logger *zap.Logger) (component.LogExporter, error) {
+	s := &loggingExporter{
+		debug:  level == "debug",
+		logger: logger,
+	}
+
+	return exporterhelper.NewLogsExporter(
+		config,
+		s.pushLogData,
+		exporterhelper.WithShutdown(loggerSync(logger)),
+	)
+}
+
+func (s *loggingExporter) pushLogData(
+	_ context.Context,
+	ld data.Logs,
+) (int, error) {
+	s.logger.Info("LogExporter", zap.Int("#logs", ld.LogRecordCount()))
+
+	if !s.debug {
+		return 0, nil
+	}
+
+	buf := logDataBuffer{}
+	rms := ld.ResourceLogs()
+	for i := 0; i < rms.Len(); i++ {
+		buf.logEntry("ResourceLog #%d", i)
+		rm := rms.At(i)
+		if rm.IsNil() {
+			buf.logEntry("* Nil ResourceLog")
+			continue
+		}
+		if !rm.Resource().IsNil() {
+			buf.logAttributeMap("Resource labels", rm.Resource().Attributes())
+		}
+		lrs := rm.Logs()
+		for j := 0; j < lrs.Len(); j++ {
+			buf.logEntry("LogRecord #%d", j)
+			lr := lrs.At(j)
+			if lr.IsNil() {
+				buf.logEntry("* Nil LogRecord")
+				continue
+			}
+			buf.logLogRecord(lr)
+		}
+	}
+
+	s.logger.Debug(buf.str.String())
+
+	return 0, nil
+}

exporter/loggingexporter/logging_exporter_test.go

@@ -57,3 +57,17 @@ func TestLoggingMetricsExporterNoErrors(t *testing.T) {
 
 	assert.NoError(t, lme.Shutdown(context.Background()))
 }
+
+func TestLoggingLogExporterNoErrors(t *testing.T) {
+	lle, err := NewLogExporter(&configmodels.ExporterSettings{}, "debug", zap.NewNop())
+	require.NotNil(t, lle)
+	assert.NoError(t, err)
+
+	assert.NoError(t, lle.ConsumeLogs(context.Background(), testdata.GenerateLogDataEmpty()))
+	assert.NoError(t, lle.ConsumeLogs(context.Background(), testdata.GenerateLogDataOneEmptyResourceLogs()))
+	assert.NoError(t, lle.ConsumeLogs(context.Background(), testdata.GenerateLogDataOneEmptyOneNilResourceLogs()))
+	assert.NoError(t, lle.ConsumeLogs(context.Background(), testdata.GenerateLogDataNoLogRecords()))
+	assert.NoError(t, lle.ConsumeLogs(context.Background(), testdata.GenerateLogDataOneEmptyLogs()))
+
+	assert.NoError(t, lle.Shutdown(context.Background()))
+}