[elasticsearchexporter] When timestamp is not set, use observedTimest…

…amp overwrites the timestamp (open-telemetry#31844)

**Description:** <Describe what has changed.>
<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->

**Link to tracking Issue:**
open-telemetry#11752

---------

Signed-off-by: Jared Tan <[email protected]>
Co-authored-by: Andrzej Stencel <[email protected]>

.chloggen/fix_ex_exporter_does_not_set_reasonable_timestamp.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# 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: elasticsearchexporter
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: When timestamp is not set, use observedTimestamp
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [11752]
+
+# (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:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# 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: [user]

exporter/elasticsearchexporter/model.go

@@ -41,7 +41,11 @@ const (
 
 func (m *encodeModel) encodeLog(resource pcommon.Resource, record plog.LogRecord, scope pcommon.InstrumentationScope) ([]byte, error) {
  var document objmodel.Document
- document.AddTimestamp("@timestamp", record.Timestamp()) // We use @timestamp in order to ensure that we can index if the default data stream logs template is used.
+ docTimeStamp := record.Timestamp()
+ if docTimeStamp.AsTime().UnixNano() == 0 {
+ docTimeStamp = record.ObservedTimestamp()
+ }
+ document.AddTimestamp("@timestamp", docTimeStamp) // We use @timestamp in order to ensure that we can index if the default data stream logs template is used.
  document.AddTraceID("TraceId", record.TraceID())
  document.AddSpanID("SpanId", record.SpanID())
  document.AddInt("TraceFlags", int64(record.Flags()))

exporter/elasticsearchexporter/model_test.go

@@ -11,6 +11,7 @@ import (
  "github.com/stretchr/testify/assert"
  "github.com/stretchr/testify/require"
  "go.opentelemetry.io/collector/pdata/pcommon"
+ "go.opentelemetry.io/collector/pdata/plog"
  "go.opentelemetry.io/collector/pdata/ptrace"
  semconv "go.opentelemetry.io/collector/semconv/v1.18.0"
 
@@ -19,6 +20,10 @@ import (
 
 var expectedSpanBody = `{"@timestamp":"2023-04-19T03:04:05.000000006Z","Attributes.service.instance.id":"23","Duration":1000000,"EndTimestamp":"2023-04-19T03:04:06.000000006Z","Events.fooEvent.evnetMockBar":"bar","Events.fooEvent.evnetMockFoo":"foo","Events.fooEvent.time":"2023-04-19T03:04:05.000000006Z","Kind":"SPAN_KIND_CLIENT","Link":"[{\"attribute\":{},\"spanID\":\"\",\"traceID\":\"01020304050607080807060504030200\"}]","Name":"client span","Resource.cloud.platform":"aws_elastic_beanstalk","Resource.cloud.provider":"aws","Resource.deployment.environment":"BETA","Resource.service.instance.id":"23","Resource.service.name":"some-service","Resource.service.version":"env-version-1234","Scope.lib-foo":"lib-bar","Scope.name":"io.opentelemetry.rabbitmq-2.7","Scope.version":"1.30.0-alpha","SpanId":"1920212223242526","TraceId":"01020304050607080807060504030201","TraceStatus":2,"TraceStatusDescription":"Test"}`
 
+var expectedLogBody = `{"@timestamp":"2023-04-19T03:04:05.000000006Z","Attributes.log-attr1":"value1","Body":"log-body","Resource.key1":"value1","Scope.name":"","Scope.version":"","SeverityNumber":0,"TraceFlags":0}`
+
+var expectedLogBodyWithEmptyTimestamp = `{"@timestamp":"1970-01-01T00:00:00.000000000Z","Attributes.log-attr1":"value1","Body":"log-body","Resource.key1":"value1","Scope.name":"","Scope.version":"","SeverityNumber":0,"TraceFlags":0}`
+
 func TestEncodeSpan(t *testing.T) {
  model := &encodeModel{dedup: true, dedot: false}
  td := mockResourceSpans()
@@ -27,6 +32,25 @@ func TestEncodeSpan(t *testing.T) {
  assert.Equal(t, expectedSpanBody, string(spanByte))
 }
 
+func TestEncodeLog(t *testing.T) {
+ t.Run("empty timestamp with observedTimestamp override", func(t *testing.T) {
+ model := &encodeModel{dedup: true, dedot: false}
+ td := mockResourceLogs()
+ td.ScopeLogs().At(0).LogRecords().At(0).SetObservedTimestamp(pcommon.NewTimestampFromTime(time.Date(2023, 4, 19, 3, 4, 5, 6, time.UTC)))
+ logByte, err := model.encodeLog(td.Resource(), td.ScopeLogs().At(0).LogRecords().At(0), td.ScopeLogs().At(0).Scope())
+ assert.NoError(t, err)
+ assert.Equal(t, expectedLogBody, string(logByte))
+ })
+
+ t.Run("both timestamp and observedTimestamp empty", func(t *testing.T) {
+ model := &encodeModel{dedup: true, dedot: false}
+ td := mockResourceLogs()
+ logByte, err := model.encodeLog(td.Resource(), td.ScopeLogs().At(0).LogRecords().At(0), td.ScopeLogs().At(0).Scope())
+ assert.NoError(t, err)
+ assert.Equal(t, expectedLogBodyWithEmptyTimestamp, string(logByte))
+ })
+}
+
 func mockResourceSpans() ptrace.Traces {
  traces := ptrace.NewTraces()
 
@@ -68,6 +92,15 @@ func mockResourceSpans() ptrace.Traces {
  return traces
 }
 
+func mockResourceLogs() plog.ResourceLogs {
+ rl := plog.NewResourceLogs()
+ rl.Resource().Attributes().PutStr("key1", "value1")
+ l := rl.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty()
+ l.Attributes().PutStr("log-attr1", "value1")
+ l.Body().SetStr("log-body")
+ return rl
+}
+
 func TestEncodeAttributes(t *testing.T) {
  t.Parallel()