DSET-3998 feat: export Logs resource info based on export_resource_info_on_event configuration (open-telemetry#23250)

* DSET-3998 - export Logs resource info based on     export_resource_info_on_event configuration

* DSET-3998 - simplify

* DSET-3998 - improve docs

* Fix log exporter to set AddEvents Event timestamp (ts) field to event
observed timetamp in case LogRecord doesn't contain timestamp.

Even though ObservedTimestamp should always be present, we fall back to
current time in case it's not.

In addition to that, remove duplicate and redundant "timestamp"
attribute from the AddEvents event.

* Add additional assertions.

* Remove dummy debug logs.

* Create export-logs-resource-info-based-configuration

* address PR notes - fix changelog gen

* fix docs typo

* fix changelog file suffix

---------

Co-authored-by: Tomaz Muraus <[email protected]>
Co-authored-by: Tomaz Muraus <[email protected]>

Author: zdaratom-s1

.chloggen/export-logs-resource-info-based-configuration.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'enhancement'
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: datasetexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Allow include Logs resource info export to DataSet based on new export_resource_info_on_event configuration. Fix timestamp handling."
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [20660]
+
+# (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:

exporter/datasetexporter/README.md

@@ -33,6 +33,8 @@ If you do not want to specify `api_key` in the file, you can use the [builtin fu
 - `traces`:
   - `aggregate` (default = false): Count the number of spans and errors belonging to a trace.
   - `max_wait` (default = 5s): The maximum waiting for all spans from single trace to arrive; ignored if `aggregate` is false.
+- `logs`:
+  - `export_resource_info_on_event` (default = false): Include resource info to DataSet Event while exporting Logs. This is especially useful when reducing DataSet billable log volume.
 - `retry_on_failure`: See [retry_on_failure](https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/README.md)
 - `sending_queue`: See [sending_queue](https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/README.md)
 - `timeout`: See [timeout](https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/README.md)

exporter/datasetexporter/config.go

@@ -32,6 +32,22 @@ func newDefaultTracesSettings() TracesSettings {
 	}
 }
 
+const logsExportResourceInfoDefault = false
+
+type LogsSettings struct {
+	// ExportResourceInfo is optional flag to signal that the resource info is being exported to DataSet while exporting Logs.
+	// This is especially useful when reducing DataSet billable log volume.
+	// Default value: false.
+	ExportResourceInfo bool `mapstructure:"export_resource_info_on_event"`
+}
+
+// newDefaultLogsSettings returns the default settings for LogsSettings.
+func newDefaultLogsSettings() LogsSettings {
+	return LogsSettings{
+		ExportResourceInfo: logsExportResourceInfoDefault,
+	}
+}
+
 const bufferMaxLifetime = 5 * time.Second
 const bufferRetryInitialInterval = 5 * time.Second
 const bufferRetryMaxInterval = 30 * time.Second
@@ -61,6 +77,7 @@ type Config struct {
 	APIKey                         configopaque.String `mapstructure:"api_key"`
 	BufferSettings                 `mapstructure:"buffer"`
 	TracesSettings                 `mapstructure:"traces"`
+	LogsSettings                   `mapstructure:"logs"`
 	exporterhelper.RetrySettings   `mapstructure:"retry_on_failure"`
 	exporterhelper.QueueSettings   `mapstructure:"sending_queue"`
 	exporterhelper.TimeoutSettings `mapstructure:"timeout"`
@@ -96,7 +113,8 @@ func (c *Config) String() string {
 	s += fmt.Sprintf("%s: %+v; ", "TracesSettings", c.TracesSettings)
 	s += fmt.Sprintf("%s: %+v; ", "RetrySettings", c.RetrySettings)
 	s += fmt.Sprintf("%s: %+v; ", "QueueSettings", c.QueueSettings)
-	s += fmt.Sprintf("%s: %+v", "TimeoutSettings", c.TimeoutSettings)
+	s += fmt.Sprintf("%s: %+v; ", "TimeoutSettings", c.TimeoutSettings)
+	s += fmt.Sprintf("%s: %+v", "LogsSettings", c.LogsSettings)
 
 	return s
 }
@@ -123,11 +141,13 @@ func (c *Config) convert() (*ExporterConfig, error) {
 				},
 			},
 			tracesSettings: c.TracesSettings,
+			logsSettings:   c.LogsSettings,
 		},
 		nil
 }
 
 type ExporterConfig struct {
 	datasetConfig  *datasetConfig.DataSetConfig
 	tracesSettings TracesSettings
+	logsSettings   LogsSettings
 }

exporter/datasetexporter/config_test.go

@@ -43,6 +43,7 @@ func TestConfigUseDefaults(t *testing.T) {
 	assert.Equal(t, "secret", string(config.APIKey))
 	assert.Equal(t, bufferMaxLifetime, config.MaxLifetime)
 	assert.Equal(t, tracesMaxWait, config.TracesSettings.MaxWait)
+	assert.Equal(t, logsExportResourceInfoDefault, config.LogsSettings.ExportResourceInfo)
 }
 
 func TestConfigValidate(t *testing.T) {
@@ -114,7 +115,22 @@ func TestConfigString(t *testing.T) {
 	}
 
 	assert.Equal(t,
-		"DatasetURL: https://example.com; BufferSettings: {MaxLifetime:123ns GroupBy:[field1 field2] RetryInitialInterval:0s RetryMaxInterval:0s RetryMaxElapsedTime:0s}; TracesSettings: {Aggregate:true MaxWait:45s}; RetrySettings: {Enabled:true InitialInterval:5s RandomizationFactor:0.5 Multiplier:1.5 MaxInterval:30s MaxElapsedTime:5m0s}; QueueSettings: {Enabled:true NumConsumers:10 QueueSize:1000 StorageID:<nil>}; TimeoutSettings: {Timeout:5s}",
+		"DatasetURL: https://example.com; BufferSettings: {MaxLifetime:123ns GroupBy:[field1 field2] RetryInitialInterval:0s RetryMaxInterval:0s RetryMaxElapsedTime:0s}; TracesSettings: {Aggregate:true MaxWait:45s}; RetrySettings: {Enabled:true InitialInterval:5s RandomizationFactor:0.5 Multiplier:1.5 MaxInterval:30s MaxElapsedTime:5m0s}; QueueSettings: {Enabled:true NumConsumers:10 QueueSize:1000 StorageID:<nil>}; TimeoutSettings: {Timeout:5s}; LogsSettings: {ExportResourceInfo:false}",
 		config.String(),
 	)
 }
+
+func TestConfigUseProvidedExportResourceInfoValue(t *testing.T) {
+	f := NewFactory()
+	config := f.CreateDefaultConfig().(*Config)
+	configMap := confmap.NewFromStringMap(map[string]interface{}{
+		"dataset_url": "https://example.com",
+		"api_key":     "secret",
+		"logs": map[string]any{
+			"export_resource_info_on_event": true,
+		},
+	})
+	err := config.Unmarshal(configMap)
+	assert.Nil(t, err)
+	assert.Equal(t, true, config.LogsSettings.ExportResourceInfo)
+}

exporter/datasetexporter/datasetexporter.go

@@ -24,6 +24,7 @@ type DatasetExporter struct {
 	logger      *zap.Logger
 	session     string
 	spanTracker *spanTracker
+	exporterCfg *ExporterConfig
 }
 
 func newDatasetExporter(entity string, config *Config, logger *zap.Logger) (*DatasetExporter, error) {
@@ -60,6 +61,7 @@ func newDatasetExporter(entity string, config *Config, logger *zap.Logger) (*Dat
 		session:     uuid.New().String(),
 		logger:      logger,
 		spanTracker: tracker,
+		exporterCfg: exporterCfg,
 	}, nil
 }
 

exporter/datasetexporter/factory.go

@@ -27,6 +27,7 @@ func createDefaultConfig() component.Config {
 	return &Config{
 		BufferSettings:  newDefaultBufferSettings(),
 		TracesSettings:  newDefaultTracesSettings(),
+		LogsSettings:    newDefaultLogsSettings(),
 		RetrySettings:   exporterhelper.NewDefaultRetrySettings(),
 		QueueSettings:   exporterhelper.NewDefaultQueueSettings(),
 		TimeoutSettings: exporterhelper.NewDefaultTimeoutSettings(),

exporter/datasetexporter/factory_test.go

@@ -49,6 +49,7 @@ func TestLoadConfig(t *testing.T) {
 				APIKey:          "key-minimal",
 				BufferSettings:  newDefaultBufferSettings(),
 				TracesSettings:  newDefaultTracesSettings(),
+				LogsSettings:    newDefaultLogsSettings(),
 				RetrySettings:   exporterhelper.NewDefaultRetrySettings(),
 				QueueSettings:   exporterhelper.NewDefaultQueueSettings(),
 				TimeoutSettings: exporterhelper.NewDefaultTimeoutSettings(),
@@ -67,6 +68,7 @@ func TestLoadConfig(t *testing.T) {
 					RetryMaxElapsedTime:  bufferRetryMaxElapsedTime,
 				},
 				TracesSettings:  newDefaultTracesSettings(),
+				LogsSettings:    newDefaultLogsSettings(),
 				RetrySettings:   exporterhelper.NewDefaultRetrySettings(),
 				QueueSettings:   exporterhelper.NewDefaultQueueSettings(),
 				TimeoutSettings: exporterhelper.NewDefaultTimeoutSettings(),
@@ -87,6 +89,9 @@ func TestLoadConfig(t *testing.T) {
 				TracesSettings: TracesSettings{
 					MaxWait: 3 * time.Second,
 				},
+				LogsSettings: LogsSettings{
+					ExportResourceInfo: true,
+				},
 				RetrySettings: exporterhelper.RetrySettings{
 					Enabled:             true,
 					InitialInterval:     11 * time.Nanosecond,
@@ -133,7 +138,7 @@ func createExporterTests() []CreateTest {
 		{
 			name:          "broken",
 			config:        &Config{},
-			expectedError: fmt.Errorf("cannot get DataSetExpoter: cannot convert config: DatasetURL: ; BufferSettings: {MaxLifetime:0s GroupBy:[] RetryInitialInterval:0s RetryMaxInterval:0s RetryMaxElapsedTime:0s}; TracesSettings: {Aggregate:false MaxWait:0s}; RetrySettings: {Enabled:false InitialInterval:0s RandomizationFactor:0 Multiplier:0 MaxInterval:0s MaxElapsedTime:0s}; QueueSettings: {Enabled:false NumConsumers:0 QueueSize:0 StorageID:<nil>}; TimeoutSettings: {Timeout:0s}; config is not valid: api_key is required"),
+			expectedError: fmt.Errorf("cannot get DataSetExpoter: cannot convert config: DatasetURL: ; BufferSettings: {MaxLifetime:0s GroupBy:[] RetryInitialInterval:0s RetryMaxInterval:0s RetryMaxElapsedTime:0s}; TracesSettings: {Aggregate:false MaxWait:0s}; RetrySettings: {Enabled:false InitialInterval:0s RandomizationFactor:0 Multiplier:0 MaxInterval:0s MaxElapsedTime:0s}; QueueSettings: {Enabled:false NumConsumers:0 QueueSize:0 StorageID:<nil>}; TimeoutSettings: {Timeout:0s}; LogsSettings: {ExportResourceInfo:false}; config is not valid: api_key is required"),
 		},
 		{
 			name: "valid",

exporter/datasetexporter/logs_exporter.go

@@ -17,6 +17,8 @@ import (
 	"go.opentelemetry.io/collector/pdata/plog"
 )
 
+var now = time.Now
+
 func createLogsExporter(ctx context.Context, set exporter.CreateSettings, config component.Config) (exporter.Logs, error) {
 	cfg := castConfig(config)
 	e, err := newDatasetExporter("logs", cfg, set.Logger)
@@ -63,17 +65,22 @@ func buildBody(attrs map[string]interface{}, value pcommon.Value) string {
 	return message
 }
 
-func buildEventFromLog(log plog.LogRecord, resource pcommon.Resource, scope pcommon.InstrumentationScope) *add_events.EventBundle {
+func buildEventFromLog(
+	log plog.LogRecord,
+	resource pcommon.Resource,
+	scope pcommon.InstrumentationScope,
+	settings LogsSettings,
+) *add_events.EventBundle {
 	attrs := make(map[string]interface{})
 	event := add_events.Event{}
 
+	observedTs := log.ObservedTimestamp().AsTime()
 	if sevNum := log.SeverityNumber(); sevNum > 0 {
 		attrs["severity.number"] = sevNum
 		event.Sev = int(sevNum)
 	}
 
 	if timestamp := log.Timestamp().AsTime(); !timestamp.Equal(time.Unix(0, 0)) {
-		attrs["timestamp"] = timestamp.String()
 		event.Ts = strconv.FormatInt(timestamp.UnixNano(), 10)
 	}
 
@@ -86,8 +93,8 @@ func buildEventFromLog(log plog.LogRecord, resource pcommon.Resource, scope pcom
 	if dropped := log.DroppedAttributesCount(); dropped > 0 {
 		attrs["dropped_attributes_count"] = dropped
 	}
-	if observed := log.ObservedTimestamp().AsTime(); !observed.Equal(time.Unix(0, 0)) {
-		attrs["observed.timestamp"] = observed.String()
+	if !observedTs.Equal(time.Unix(0, 0)) {
+		attrs["observed.timestamp"] = observedTs.String()
 	}
 	if sevText := log.SeverityText(); sevText != "" {
 		attrs["severity.text"] = sevText
@@ -100,11 +107,27 @@ func buildEventFromLog(log plog.LogRecord, resource pcommon.Resource, scope pcom
 		attrs["trace_id"] = trace
 	}
 
+	// Event needs to always have timestamp set otherwise it will get set to unix epoch start time
+	if event.Ts == "" {
+		// ObservedTimestamp should always be set, but in case if it's not, we fall back to
+		// current time
+		// TODO: We should probably also do a rate limited log message here since this
+		// could indicate an issue with the current setup in case most events don't contain
+		// a timestamp.
+		if !observedTs.Equal(time.Unix(0, 0)) {
+			event.Ts = strconv.FormatInt(observedTs.UnixNano(), 10)
+		} else {
+			event.Ts = strconv.FormatInt(now().UnixNano(), 10)
+		}
+	}
+
 	updateWithPrefixedValues(attrs, "attributes.", ".", log.Attributes().AsRaw(), 0)
 	attrs["flags"] = log.Flags()
 	attrs["flag.is_sampled"] = log.Flags().IsSampled()
 
-	updateWithPrefixedValues(attrs, "resource.attributes.", ".", resource.Attributes().AsRaw(), 0)
+	if settings.ExportResourceInfo {
+		updateWithPrefixedValues(attrs, "resource.attributes.", ".", resource.Attributes().AsRaw(), 0)
+	}
 	attrs["scope.name"] = scope.Name()
 	updateWithPrefixedValues(attrs, "scope.attributes.", ".", scope.Attributes().AsRaw(), 0)
 
@@ -130,7 +153,7 @@ func (e *DatasetExporter) consumeLogs(_ context.Context, ld plog.Logs) error {
 			logRecords := scopeLogs.At(j).LogRecords()
 			for k := 0; k < logRecords.Len(); k++ {
 				logRecord := logRecords.At(k)
-				events = append(events, buildEventFromLog(logRecord, resource, scope))
+				events = append(events, buildEventFromLog(logRecord, resource, scope, e.exporterCfg.logsSettings))
 			}
 		}
 	}