Skip to content

Commit

Permalink
DSET-3998 feat: export Logs resource info based on export_resource_in…
Browse files Browse the repository at this point in the history
…fo_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 <tomazm@sentinelone.com>
Co-authored-by: Tomaz Muraus <126863902+tomaz-s1@users.noreply.github.com>
  • Loading branch information
3 people authored Jun 14, 2023
1 parent 5a582d9 commit 19e2a2d
Show file tree
Hide file tree
Showing 10 changed files with 224 additions and 40 deletions.
20 changes: 20 additions & 0 deletions .chloggen/export-logs-resource-info-based-configuration.yaml
Original file line number Diff line number Diff line change
@@ -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:
2 changes: 2 additions & 0 deletions exporter/datasetexporter/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
22 changes: 21 additions & 1 deletion exporter/datasetexporter/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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"`
Expand Down Expand Up @@ -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
}
Expand All @@ -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
}
18 changes: 17 additions & 1 deletion exporter/datasetexporter/config_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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)
}
2 changes: 2 additions & 0 deletions exporter/datasetexporter/datasetexporter.go
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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
}

Expand Down
1 change: 1 addition & 0 deletions exporter/datasetexporter/factory.go
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand Down
7 changes: 6 additions & 1 deletion exporter/datasetexporter/factory_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand All @@ -67,6 +68,7 @@ func TestLoadConfig(t *testing.T) {
RetryMaxElapsedTime: bufferRetryMaxElapsedTime,
},
TracesSettings: newDefaultTracesSettings(),
LogsSettings: newDefaultLogsSettings(),
RetrySettings: exporterhelper.NewDefaultRetrySettings(),
QueueSettings: exporterhelper.NewDefaultQueueSettings(),
TimeoutSettings: exporterhelper.NewDefaultTimeoutSettings(),
Expand All @@ -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,
Expand Down Expand Up @@ -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",
Expand Down
35 changes: 29 additions & 6 deletions exporter/datasetexporter/logs_exporter.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)
}

Expand All @@ -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
Expand All @@ -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)

Expand All @@ -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))
}
}
}
Expand Down
Loading

0 comments on commit 19e2a2d

Please sign in to comment.