Propagate telemetry.resource to the internal logs

.chloggen/fix-resource-attributes-internal-console-logger.yaml

@@ -0,0 +1,29 @@
+# Use this changelog template to create an entry for release notes.
+
+ # One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+ change_type: bug_fix
+
+ # The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+ component: "internal telemetry"
+
+ # A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+ note: "Add resource attributes from telemetry.resource to the logger"
+
+ # One or more tracking issues or pull requests related to the change
+ issues: [12582]
+
+ # (Optional) One or more lines of additional information to render under the primary note.
+ # These lines will be padded with 2 spaces and then inserted directly into the document.
+ # Use pipe (|) for multiline entries.
+ subtext: |
+   Resource attributes from telemetry.resource were not added to the internal
+   console logs.
+
+   Now, they are added to the logger as part of the "resource" field.
+
+ # Optional: The change log or logs in which this entry should be included.
+ # e.g. '[user]' or '[user, api]'
+ # Include 'user' if the change is relevant to end users.
+ # Include 'api' if there is a change to a library API.
+ # Default: '[user]'
+ change_logs: []

service/telemetry/logger.go

@@ -39,6 +39,22 @@ func newLogger(set Settings, cfg Config) (*zap.Logger, log.LoggerProvider, error
 		return nil, nil, err
 	}
 
+	// The attributes in cfg.Resource are added as resource attributes for logs exported through the LoggerProvider instantiated below.
+	// To make sure they are also exposed in logs written to stdout, we add them as fields to the Zap core created above using WrapCore.
+	// We do NOT add them to the logger using With, because that would apply to all logs, even ones exported through the core that wraps
+	// the LoggerProvider, meaning that the attributes would be exported twice.
+	logger = logger.WithOptions(zap.WrapCore(func(c zapcore.Core) zapcore.Core {
+		fields := []zap.Field{}
+		for k, v := range cfg.Resource {
+			if v != nil {
+				f := zap.Stringp(k, v)
+				fields = append(fields, f)
+			}
+		}
+		r := zap.Dict("resource", fields...)
+		return c.With([]zapcore.Field{r})
+	}))
+
 	var lp log.LoggerProvider
 
 	logger = logger.WithOptions(zap.WrapCore(func(core zapcore.Core) zapcore.Core {

service/telemetry/logger_test.go

@@ -6,15 +6,28 @@ package telemetry // import "go.opentelemetry.io/collector/service/telemetry"
 import (
 	"context"
 	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
 	"reflect"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	config "go.opentelemetry.io/contrib/otelconf/v0.3.0"
+	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
+	"go.uber.org/zap/zaptest/observer"
+
+	"go.opentelemetry.io/collector/pdata/plog/plogotlp"
+	semconv "go.opentelemetry.io/collector/semconv/v1.18.0"
 )
 
+type shutdownable interface {
+	Shutdown(context.Context) error
+}
+
 func TestNewLogger(t *testing.T) {
 	tests := []struct {
 		name         string
@@ -102,9 +115,6 @@ func TestNewLogger(t *testing.T) {
 				require.NoError(t, err)
 				gotType := reflect.TypeOf(l.Core()).String()
 				require.Equal(t, wantCoreType, gotType)
-				type shutdownable interface {
-					Shutdown(context.Context) error
-				}
 				if prov, ok := lp.(shutdownable); ok {
 					require.NoError(t, prov.Shutdown(context.Background()))
 				}
@@ -114,3 +124,149 @@ func TestNewLogger(t *testing.T) {
 		testCoreType(t, tt.wantCoreType)
 	}
 }
+
+func TestNewLoggerWithResource(t *testing.T) {
+	observerCore, observedLogs := observer.New(zap.InfoLevel)
+
+	set := Settings{
+		ZapOptions: []zap.Option{
+			zap.WrapCore(func(core zapcore.Core) zapcore.Core {
+				// Combine original core and observer core to capture everything
+				return zapcore.NewTee(core, observerCore)
+			}),
+		},
+	}
+
+	cfg := Config{
+		Logs: LogsConfig{
+			Level:    zapcore.InfoLevel,
+			Encoding: "json",
+		},
+		Resource: map[string]*string{
+			"myfield": ptr("myvalue"),
+		},
+	}
+
+	mylogger, _, _ := newLogger(set, cfg)
+
+	mylogger.Info("Test log message")
+	require.Len(t, observedLogs.All(), 1)
+
+	entry := observedLogs.All()[0]
+	assert.Equal(t, "resource", entry.Context[0].Key)
+	dict := entry.Context[0].Interface.(zapcore.ObjectMarshaler)
+	enc := zapcore.NewMapObjectEncoder()
+	require.NoError(t, dict.MarshalLogObject(enc))
+	require.Equal(t, "myvalue", enc.Fields["myfield"])
+}
+
+func TestOTLPLogExport(t *testing.T) {
+	version := "1.2.3"
+	service := "test-service"
+	testAttribute := "test-attribute"
+	testValue := "test-value"
+	receivedLogs := 0
+	totalLogs := 10
+
+	// Create a backend to receive the logs and assert the content
+	srv := createBackend("/v1/logs", func(writer http.ResponseWriter, request *http.Request) {
+		body, err := io.ReadAll(request.Body)
+		assert.NoError(t, err)
+		defer request.Body.Close()
+
+		// Unmarshal the protobuf body into logs
+		req := plogotlp.NewExportRequest()
+		err = req.UnmarshalProto(body)
+		assert.NoError(t, err)
+
+		logs := req.Logs()
+		rl := logs.ResourceLogs().At(0)
+
+		resourceAttrs := rl.Resource().Attributes().AsRaw()
+		assert.Equal(t, resourceAttrs[semconv.AttributeServiceName], service)
+		assert.Equal(t, resourceAttrs[semconv.AttributeServiceVersion], version)
+		assert.Equal(t, resourceAttrs[testAttribute], testValue)
+
+		// Check that the resource attributes are not duplicated in the log records
+		sl := rl.ScopeLogs().At(0)
+		logRecord := sl.LogRecords().At(0)
+		attrs := logRecord.Attributes().AsRaw()
+		assert.NotContains(t, attrs, semconv.AttributeServiceName)
+		assert.NotContains(t, attrs, semconv.AttributeServiceVersion)
+		assert.NotContains(t, attrs, testAttribute)
+
+		receivedLogs++
+
+		writer.WriteHeader(http.StatusOK)
+	})
+	defer srv.Close()
+
+	processors := []config.LogRecordProcessor{
+		{
+			Simple: &config.SimpleLogRecordProcessor{
+				Exporter: config.LogRecordExporter{
+					OTLP: &config.OTLP{
+						Endpoint: ptr(srv.URL),
+						Protocol: ptr("http/protobuf"),
+						Insecure: ptr(true),
+					},
+				},
+			},
+		},
+	}
+
+	cfg := Config{
+		Logs: LogsConfig{
+			Level:       zapcore.DebugLevel,
+			Development: true,
+			Encoding:    "json",
+			Processors:  processors,
+		},
+	}
+
+	sdk, _ := config.NewSDK(
+		config.WithOpenTelemetryConfiguration(
+			config.OpenTelemetryConfiguration{
+				LoggerProvider: &config.LoggerProvider{
+					Processors: processors,
+				},
+				Resource: &config.Resource{
+					SchemaUrl: ptr(""),
+					Attributes: []config.AttributeNameValue{
+						{Name: semconv.AttributeServiceName, Value: service},
+						{Name: semconv.AttributeServiceVersion, Value: version},
+						{Name: testAttribute, Value: testValue},
+					},
+				},
+			},
+		),
+	)
+
+	l, lp, err := newLogger(Settings{SDK: &sdk}, cfg)
+	require.NoError(t, err)
+	require.NotNil(t, l)
+	require.NotNil(t, lp)
+
+	defer func() {
+		if prov, ok := lp.(shutdownable); ok {
+			require.NoError(t, prov.Shutdown(context.Background()))
+		}
+	}()
+
+	// Reset counter for each test case
+	receivedLogs = 0
+
+	// Generate some logs to send to the backend
+	for i := 0; i < totalLogs; i++ {
+		l.Info("Test log message")
+	}
+
+	// Ensure the correct number of logs were received
+	require.Equal(t, totalLogs, receivedLogs)
+}
+
+func createBackend(endpoint string, handler func(writer http.ResponseWriter, request *http.Request)) *httptest.Server {
+	mux := http.NewServeMux()
+	mux.HandleFunc(endpoint, handler)
+	return httptest.NewServer(mux)
+}