Add LogRecord observed timestamp field

api/logs/src/main/java/io/opentelemetry/api/logs/DefaultLogger.java

@@ -40,6 +40,16 @@ public LogRecordBuilder setTimestamp(Instant instant) {
       return this;
     }
 
+    @Override
+    public LogRecordBuilder setObservedTimestamp(long timestamp, TimeUnit unit) {
+      return this;
+    }
+
+    @Override
+    public LogRecordBuilder setObservedTimestamp(Instant instant) {
+      return this;
+    }
+
     @Override
     public LogRecordBuilder setContext(Context context) {
       return this;

api/logs/src/main/java/io/opentelemetry/api/logs/LogRecordBuilder.java

@@ -35,6 +35,26 @@ public interface LogRecordBuilder {
    */
   LogRecordBuilder setTimestamp(Instant instant);
 
+  /**
+   * Set the epoch {@code observedTimestamp}, using the timestamp and unit.
+   *
+   * <p>The {@code observedTimestamp} is the time at which the log record was observed. If unset, it
+   * will be set to the {@code timestamp}. {@code observedTimestamp} may be different from {@code
+   * timestamp} if logs are being processed asynchronously (e.g. from a file or on a different
+   * thread).
+   */
+  LogRecordBuilder setObservedTimestamp(long timestamp, TimeUnit unit);
+
+  /**
+   * Set the {@code observedTimestamp}, using the instant.
+   *
+   * <p>The {@code observedTimestamp} is the time at which the log record was observed. If unset, it
+   * will be set to the {@code timestamp}. {@code observedTimestamp} may be different from {@code
+   * timestamp} if logs are being processed asynchronously (e.g. from a file or on a different
+   * thread).
+   */
+  LogRecordBuilder setObservedTimestamp(Instant instant);
+
   /** Set the context. */
   LogRecordBuilder setContext(Context context);
 

api/logs/src/test/java/io/opentelemetry/api/logs/DefaultLoggerTest.java

@@ -24,6 +24,8 @@ void buildAndEmit() {
                     .logRecordBuilder()
                     .setTimestamp(100, TimeUnit.SECONDS)
                     .setTimestamp(Instant.now())
+                    .setObservedTimestamp(100, TimeUnit.SECONDS)
+                    .setObservedTimestamp(Instant.now())
                     .setContext(Context.root())
                     .setSeverity(Severity.DEBUG)
                     .setSeverityText("debug")

exporters/otlp/testing-internal/src/main/java/io/opentelemetry/exporter/otlp/testing/internal/FakeTelemetryUtil.java

@@ -91,6 +91,7 @@ public static LogRecordData generateFakeLogRecordData() {
         .setSeverity(Severity.INFO)
         .setSeverityText(Severity.INFO.name())
         .setTimestamp(Instant.now())
+        .setObservedTimestamp(Instant.now().plusNanos(100))
         .build();
   }
 

sdk/logs-testing/src/main/java/io/opentelemetry/sdk/testing/assertj/LogRecordDataAssert.java

@@ -71,6 +71,20 @@ public LogRecordDataAssert hasTimestamp(long timestampEpochNanos) {
     return this;
   }
 
+  /** Asserts the log has the given epoch {@code observedTimestamp}. */
+  public LogRecordDataAssert hasObservedTimestamp(long observedEpochNanos) {
+    isNotNull();
+    if (actual.getObservedTimestampEpochNanos() != observedEpochNanos) {
+      failWithActualExpectedAndMessage(
+          actual.getObservedTimestampEpochNanos(),
+          observedEpochNanos,
+          "Expected log to have observed timestamp <%s> nanos but was <%s>",
+          observedEpochNanos,
+          actual.getObservedTimestampEpochNanos());
+    }
+    return this;
+  }
+
   /** Asserts the log has the given span context. */
   public LogRecordDataAssert hasSpanContext(SpanContext spanContext) {
     isNotNull();

sdk/logs-testing/src/main/java/io/opentelemetry/sdk/testing/logs/TestLogRecordData.java

@@ -28,6 +28,7 @@ public static Builder builder() {
         .setResource(Resource.empty())
         .setInstrumentationScopeInfo(InstrumentationScopeInfo.empty())
         .setTimestamp(0, TimeUnit.NANOSECONDS)
+        .setObservedTimestamp(0, TimeUnit.NANOSECONDS)
         .setSpanContext(SpanContext.getInvalid())
         .setSeverity(Severity.UNDEFINED_SEVERITY_NUMBER)
         .setBody("")
@@ -81,6 +82,32 @@ public Builder setTimestamp(long timestamp, TimeUnit unit) {
      */
     abstract Builder setTimestampEpochNanos(long epochNanos);
 
+    /**
+     * Set the {@code observedTimestamp}, using the instant.
+     *
+     * <p>The {@code observedTimestamp} is the time at which the log record was observed.
+     */
+    public Builder setObservedTimestamp(Instant instant) {
+      return setObservedTimestampEpochNanos(
+          TimeUnit.SECONDS.toNanos(instant.getEpochSecond()) + instant.getNano());
+    }
+
+    /**
+     * Set the epoch {@code observedTimestamp}, using the timestamp and unit.
+     *
+     * <p>The {@code observedTimestamp} is the time at which the log record was observed.
+     */
+    public Builder setObservedTimestamp(long timestamp, TimeUnit unit) {
+      return setObservedTimestampEpochNanos(unit.toNanos(timestamp));
+    }
+
+    /**
+     * Set the epoch {@code observedTimestamp}.
+     *
+     * <p>The {@code observedTimestamp} is the time at which the log record was observed.
+     */
+    abstract Builder setObservedTimestampEpochNanos(long epochNanos);
+
     /** Set the span context. */
     public abstract Builder setSpanContext(SpanContext spanContext);
 

sdk/logs-testing/src/test/java/io/opentelemetry/sdk/testing/assertj/LogAssertionsTest.java

@@ -49,6 +49,7 @@ public class LogAssertionsTest {
           .setResource(RESOURCE)
           .setInstrumentationScopeInfo(INSTRUMENTATION_SCOPE_INFO)
           .setTimestamp(100, TimeUnit.NANOSECONDS)
+          .setObservedTimestamp(200, TimeUnit.NANOSECONDS)
           .setSpanContext(
               SpanContext.create(
                   TRACE_ID, SPAN_ID, TraceFlags.getDefault(), TraceState.getDefault()))
@@ -65,6 +66,7 @@ void passing() {
         .hasResource(RESOURCE)
         .hasInstrumentationScope(INSTRUMENTATION_SCOPE_INFO)
         .hasTimestamp(100)
+        .hasObservedTimestamp(200)
         .hasSpanContext(
             SpanContext.create(TRACE_ID, SPAN_ID, TraceFlags.getDefault(), TraceState.getDefault()))
         .hasSeverity(Severity.INFO)
@@ -133,6 +135,7 @@ void failure() {
     assertThatThrownBy(
         () -> assertThat(LOG_DATA).hasInstrumentationScope(InstrumentationScopeInfo.empty()));
     assertThatThrownBy(() -> assertThat(LOG_DATA).hasTimestamp(200));
+    assertThatThrownBy(() -> assertThat(LOG_DATA).hasObservedTimestamp(100));
     assertThatThrownBy(
         () ->
             assertThat(LOG_DATA)

sdk/logs/src/main/java/io/opentelemetry/sdk/logs/SdkLogRecordBuilder.java

@@ -25,6 +25,7 @@ final class SdkLogRecordBuilder implements LogRecordBuilder {
 
   private final InstrumentationScopeInfo instrumentationScopeInfo;
   private long timestampEpochNanos;
+  private long observedTimestampEpochNanos;
   @Nullable private Context context;
   private Severity severity = Severity.UNDEFINED_SEVERITY_NUMBER;
   @Nullable private String severityText;
@@ -51,6 +52,19 @@ public SdkLogRecordBuilder setTimestamp(Instant instant) {
     return this;
   }
 
+  @Override
+  public LogRecordBuilder setObservedTimestamp(long timestamp, TimeUnit unit) {
+    this.observedTimestampEpochNanos = unit.toNanos(timestamp);
+    return this;
+  }
+
+  @Override
+  public LogRecordBuilder setObservedTimestamp(Instant instant) {
+    this.observedTimestampEpochNanos =
+        TimeUnit.SECONDS.toNanos(instant.getEpochSecond()) + instant.getNano();
+    return this;
+  }
+
   @Override
   public SdkLogRecordBuilder setContext(Context context) {
     this.context = context;
@@ -95,6 +109,10 @@ public void emit() {
       return;
     }
     Context context = this.context == null ? Context.current() : this.context;
+    long observedTimestampEpochNanos =
+        this.observedTimestampEpochNanos == 0
+            ? this.loggerSharedState.getClock().now()
+            : this.observedTimestampEpochNanos;
     loggerSharedState
         .getLogRecordProcessor()
         .onEmit(
@@ -103,7 +121,8 @@ public void emit() {
                 loggerSharedState.getLogLimits(),
                 loggerSharedState.getResource(),
                 instrumentationScopeInfo,
-                this.timestampEpochNanos,
+                timestampEpochNanos,
+                observedTimestampEpochNanos,
                 Span.fromContext(context).getSpanContext(),
                 severity,
                 severityText,

sdk/logs/src/main/java/io/opentelemetry/sdk/logs/SdkLogRecordData.java

@@ -27,6 +27,7 @@ static SdkLogRecordData create(
       Resource resource,
       InstrumentationScopeInfo instrumentationScopeInfo,
       long epochNanos,
+      long observedEpochNanos,
       SpanContext spanContext,
       Severity severity,
       @Nullable String severityText,
@@ -37,6 +38,7 @@ static SdkLogRecordData create(
         resource,
         instrumentationScopeInfo,
         epochNanos,
+        observedEpochNanos,
         spanContext,
         severity,
         severityText,

sdk/logs/src/main/java/io/opentelemetry/sdk/logs/SdkReadWriteLogRecord.java

@@ -24,7 +24,8 @@ class SdkReadWriteLogRecord implements ReadWriteLogRecord {
   private final LogLimits logLimits;
   private final Resource resource;
   private final InstrumentationScopeInfo instrumentationScopeInfo;
-  private final long epochNanos;
+  private final long timestampEpochNanos;
+  private final long observedTimestampEpochNanos;
   private final SpanContext spanContext;
   private final Severity severity;
   @Nullable private final String severityText;
@@ -39,7 +40,8 @@ private SdkReadWriteLogRecord(
       LogLimits logLimits,
       Resource resource,
       InstrumentationScopeInfo instrumentationScopeInfo,
-      long epochNanos,
+      long timestampEpochNanos,
+      long observedTimestampEpochNanos,
       SpanContext spanContext,
       Severity severity,
       @Nullable String severityText,
@@ -48,7 +50,8 @@ private SdkReadWriteLogRecord(
     this.logLimits = logLimits;
     this.resource = resource;
     this.instrumentationScopeInfo = instrumentationScopeInfo;
-    this.epochNanos = epochNanos;
+    this.timestampEpochNanos = timestampEpochNanos;
+    this.observedTimestampEpochNanos = observedTimestampEpochNanos;
     this.spanContext = spanContext;
     this.severity = severity;
     this.severityText = severityText;
@@ -61,7 +64,8 @@ static SdkReadWriteLogRecord create(
       LogLimits logLimits,
       Resource resource,
       InstrumentationScopeInfo instrumentationScopeInfo,
-      long epochNanos,
+      long timestampEpochNanos,
+      long observedTimestampEpochNanos,
       SpanContext spanContext,
       Severity severity,
       @Nullable String severityText,
@@ -71,7 +75,8 @@ static SdkReadWriteLogRecord create(
         logLimits,
         resource,
         instrumentationScopeInfo,
-        epochNanos,
+        timestampEpochNanos,
+        observedTimestampEpochNanos,
         spanContext,
         severity,
         severityText,
@@ -110,7 +115,8 @@ public LogRecordData toLogRecordData() {
       return SdkLogRecordData.create(
           resource,
           instrumentationScopeInfo,
-          epochNanos,
+          timestampEpochNanos,
+          observedTimestampEpochNanos,
           spanContext,
           severity,
           severityText,

sdk/logs/src/main/java/io/opentelemetry/sdk/logs/data/LogRecordData.java

@@ -31,6 +31,9 @@ public interface LogRecordData {
   /** Returns the timestamp at which the log record occurred, in epoch nanos. */
   long getTimestampEpochNanos();
 
+  /** Returns the timestamp at which the log record was observed, in epoch nanos. */
+  long getObservedTimestampEpochNanos();
+
   /** Return the span context for this log, or {@link SpanContext#getInvalid()} if unset. */
   SpanContext getSpanContext();
 

sdk/logs/src/test/java/io/opentelemetry/sdk/logs/SdkLogRecordBuilderTest.java

@@ -16,6 +16,7 @@
 import io.opentelemetry.api.trace.TraceFlags;
 import io.opentelemetry.api.trace.TraceState;
 import io.opentelemetry.context.Context;
+import io.opentelemetry.sdk.common.Clock;
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
 import io.opentelemetry.sdk.logs.data.Body;
 import io.opentelemetry.sdk.resources.Resource;
@@ -38,6 +39,7 @@ class SdkLogRecordBuilderTest {
   private static final InstrumentationScopeInfo SCOPE_INFO = InstrumentationScopeInfo.empty();
 
   @Mock LoggerSharedState loggerSharedState;
+  @Mock Clock clock;
 
   private final AtomicReference<ReadWriteLogRecord> emittedLog = new AtomicReference<>();
   private SdkLogRecordBuilder builder;
@@ -48,13 +50,15 @@ void setup() {
     when(loggerSharedState.getLogRecordProcessor())
         .thenReturn((context, logRecord) -> emittedLog.set(logRecord));
     when(loggerSharedState.getResource()).thenReturn(RESOURCE);
+    when(loggerSharedState.getClock()).thenReturn(clock);
 
     builder = new SdkLogRecordBuilder(loggerSharedState, SCOPE_INFO);
   }
 
   @Test
   void emit_AllFields() {
     Instant timestamp = Instant.now();
+    Instant observedTimestamp = Instant.now().plusNanos(100);
 
     String bodyStr = "body";
     String sevText = "sevText";
@@ -69,6 +73,8 @@ void emit_AllFields() {
     builder.setBody(bodyStr);
     builder.setTimestamp(123, TimeUnit.SECONDS);
     builder.setTimestamp(timestamp);
+    builder.setObservedTimestamp(456, TimeUnit.SECONDS);
+    builder.setObservedTimestamp(observedTimestamp);
     builder.setAttribute(null, null);
     builder.setAttribute(AttributeKey.stringKey("k1"), "v1");
     builder.setAllAttributes(Attributes.builder().put("k2", "v2").put("k3", "v3").build());
@@ -81,6 +87,9 @@ void emit_AllFields() {
         .hasInstrumentationScope(SCOPE_INFO)
         .hasBody(bodyStr)
         .hasTimestamp(TimeUnit.SECONDS.toNanos(timestamp.getEpochSecond()) + timestamp.getNano())
+        .hasObservedTimestamp(
+            TimeUnit.SECONDS.toNanos(observedTimestamp.getEpochSecond())
+                + observedTimestamp.getNano())
         .hasAttributes(Attributes.builder().put("k1", "v1").put("k2", "v2").put("k3", "v3").build())
         .hasSpanContext(spanContext)
         .hasSeverity(severity)
@@ -89,13 +98,16 @@ void emit_AllFields() {
 
   @Test
   void emit_NoFields() {
+    when(clock.now()).thenReturn(10L);
+
     builder.emit();
 
     assertThat(emittedLog.get().toLogRecordData())
         .hasResource(RESOURCE)
         .hasInstrumentationScope(SCOPE_INFO)
         .hasBody(Body.empty().asString())
         .hasTimestamp(0L)
+        .hasObservedTimestamp(10L)
         .hasAttributes(Attributes.empty())
         .hasSpanContext(SpanContext.getInvalid())
         .hasSeverity(Severity.UNDEFINED_SEVERITY_NUMBER);