Merge branch 'opensearch-project:main' into main

Author: amdhing

.github/workflows/data-prepper-trace-analytics-raw-span-e2e-tests.yml

@@ -14,7 +14,7 @@ jobs:
     strategy:
       matrix:
         java: [11, 17, 21, docker]
-        otelVersion: ['0.9.0-alpha', '0.16.0-alpha']
+        otelVersion: ['0.16.0-alpha', '1.3.2-alpha']
       fail-fast: false
 
     runs-on: ubuntu-latest

data-prepper-api/build.gradle

@@ -14,6 +14,7 @@ dependencies {
     implementation 'com.fasterxml.jackson.datatype:jackson-datatype-jdk8'
     implementation libs.parquet.common
     implementation libs.commons.lang3
+    implementation 'jakarta.validation:jakarta.validation-api:3.0.2'
     testImplementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-yaml'
     testImplementation project(':data-prepper-test-common')
     testImplementation 'org.skyscreamer:jsonassert:1.5.3'

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/annotations/ValidRegex.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.model.annotations;
+
+import jakarta.validation.Constraint;
+import jakarta.validation.Payload;
+import org.opensearch.dataprepper.model.validation.RegexValueValidator;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Constraint(validatedBy = RegexValueValidator.class)
+@Target({ElementType.FIELD, ElementType.PARAMETER})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface ValidRegex {
+    String message() default "Invalid regular expression pattern";
+
+    Class<?>[] groups() default {};
+
+    Class<? extends Payload>[] payload() default {};
+}

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/codec/JsonDecoder.java

@@ -8,6 +8,7 @@
 import com.fasterxml.jackson.core.JsonFactory;
 import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.core.JsonToken;
+import com.fasterxml.jackson.core.StreamReadConstraints;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 import org.opensearch.dataprepper.model.event.Event;
@@ -30,10 +31,15 @@ public class JsonDecoder implements ByteDecoder {
     private Collection<String> includeKeys;
     private Collection<String> includeKeysMetadata;
 
-    public JsonDecoder(String keyName, Collection<String> includeKeys, Collection<String> includeKeysMetadata) {
+    public JsonDecoder(String keyName, Collection<String> includeKeys, Collection<String> includeKeysMetadata, Integer maxEventLength) {
         this.keyName = keyName;
         this.includeKeys = includeKeys;
         this.includeKeysMetadata = includeKeysMetadata;
+        if (maxEventLength != null) {
+        jsonFactory.setStreamReadConstraints(StreamReadConstraints.builder()
+                .maxStringLength(maxEventLength)
+                .build());
+        }
     }
 
     public JsonDecoder() {

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/codec/OutputCodec.java

@@ -15,6 +15,7 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
+import java.io.ByteArrayOutputStream;
 import java.util.Map;
 
 public interface OutputCodec {
@@ -28,7 +29,7 @@ public interface OutputCodec {
      * @param outputStream  outputStream param for wrapping
      * @param event         Event to auto-generate schema
      * @param context       Extra Context used in Codec.
-     * @throws IOException throws IOException when invalid input is received or not able to create wrapping
+     * @throws IOException  throws IOException when invalid input is received or not able to create wrapping
      */
     void start(OutputStream outputStream, Event event, OutputCodecContext context) throws IOException;
 
@@ -51,6 +52,21 @@ public interface OutputCodec {
      */
     void complete(OutputStream outputStream) throws IOException;
 
+    /**
+     * this method get called from {@link Sink} to estimate size of event in {@link OutputStream}
+     *
+     * @param event        event Record event
+     * @return long        size of the serialized event
+     * @throws IOException throws IOException when invalid input is received or not able to create wrapping
+     */
+    default long getEstimatedSize(Event event, OutputCodecContext codecContext) throws IOException {
+        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+        start(outputStream, event, codecContext);
+        writeEvent(event, outputStream);
+        complete(outputStream);
+        return outputStream.toByteArray().length;
+    }
+
     /**
      * used to get extension of file
      *

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/event/Event.java

@@ -247,6 +247,11 @@ public interface Event extends Serializable {
      */
     EventHandle getEventHandle();
 
+    default void putIfAbsent(final String key, final Class clazz, final Object value) {
+        if (get(key, clazz) == null)
+            put(key, value);
+    }
+
     JsonStringBuilder jsonBuilder();
 
     abstract class JsonStringBuilder {

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/failures/DlqObject.java

@@ -7,9 +7,9 @@
 
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
-import org.opensearch.dataprepper.model.event.EventHandle;
 import org.apache.commons.lang3.StringUtils;
 import org.opensearch.dataprepper.model.event.EventHandle;
+import org.opensearch.dataprepper.model.configuration.PluginSetting;
 
 import java.time.Instant;
 import java.time.ZoneId;
@@ -122,6 +122,16 @@ public String toString() {
             '}';
     }
 
+    public static DlqObject createDlqObject(PluginSetting pluginSetting, EventHandle eventHandle, Object failedData) {
+        return DlqObject.builder()
+                .withEventHandle(eventHandle)
+                .withFailedData(failedData)
+                .withPluginName(pluginSetting.getName())
+                .withPipelineName(pluginSetting.getPipelineName())
+                .withPluginId(pluginSetting.getName())
+                .build();
+    }
+
     public static Builder builder() {
         return new Builder();
     }

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/log/JacksonOtelLog.java

@@ -24,7 +24,9 @@
  */
 public class JacksonOtelLog extends JacksonEvent implements OpenTelemetryLog {
 
-    protected static final String OBSERVED_TIME_KEY = "observedTime";
+    protected static final String OBSERVED_TIME_KEY = "observedTimestamp";
+    protected static final String SCOPE_KEY = "instrumentationScope";
+    protected static final String RESOURCE_KEY = "resource";
     protected static final String TIME_KEY = "time";
     protected static final String SERVICE_NAME_KEY = "serviceName";
     protected static final String ATTRIBUTES_KEY = "attributes";
@@ -33,13 +35,18 @@ public class JacksonOtelLog extends JacksonEvent implements OpenTelemetryLog {
     protected static final String BODY_KEY = "body";
     protected static final String SPAN_ID_KEY = "spanId";
     protected static final String TRACE_ID_KEY = "traceId";
+    protected static final String SEVERITY_KEY = "severity";
     protected static final String SEVERITY_NUMBER_KEY = "severityNumber";
     protected static final String SEVERITY_TEXT_KEY = "severityText";
     protected static final String DROPPED_ATTRIBUTES_COUNT_KEY = "droppedAttributesCount";
 
+    protected void checkAndSetDefaultValues() {
+        putIfAbsent(ATTRIBUTES_KEY, Map.class, new HashMap<>());
+    }
 
     protected JacksonOtelLog(final JacksonOtelLog.Builder builder) {
         super(builder);
+        checkAndSetDefaultValues();
 
         checkArgument(this.getMetadata().getEventType().equals("LOG"), "eventType must be of type Log");
     }
@@ -94,6 +101,16 @@ public String getSeverityText() {
         return this.get(SEVERITY_TEXT_KEY, String.class);
     }
 
+    @Override
+    public Map<String, Object> getScope() {
+        return this.get(SCOPE_KEY, Map.class);
+    }
+
+    @Override
+    public Map<String, Object> getResource() {
+        return this.get(RESOURCE_KEY, Map.class);
+    }
+
     @Override
     public Integer getDroppedAttributesCount() {
         return this.get(DROPPED_ATTRIBUTES_COUNT_KEY, Integer.class);
@@ -221,6 +238,30 @@ public Builder withSchemaUrl(final String schemaUrl) {
             return getThis();
         }
 
+        /**
+         * Sets the scope of the log event
+         *
+         * @param scope scope to be set
+         * @return the builder
+         * @since 2.11
+         */
+        public Builder withScope(final Map<String, Object> scope) {
+            data.put(SCOPE_KEY, scope);
+            return getThis();
+        }
+
+        /**
+         * Sets the resource of the log event
+         *
+         * @param resource resource to be set
+         * @return the builder
+         * @since 2.11
+         */
+        public Builder withResource(final Map<String, Object> resource) {
+            data.put(RESOURCE_KEY, resource);
+            return getThis();
+        }
+
         /**
          * Sets the flags that are associated with this log event
          *
@@ -305,21 +346,20 @@ public Builder withDroppedAttributesCount(final Integer droppedAttributesCount)
             return getThis();
         }
 
+        protected void populateEvent() {
+            this.withEventType(EventType.LOG.toString());
+            this.withData(data);
+        }
+
         /**
          * Returns a newly created {@link JacksonOtelLog}.
          *
          * @return a log
          * @since 2.1
          */
         public JacksonOtelLog build() {
-            this.withEventType(EventType.LOG.toString());
-            this.withData(data);
-            checkAndSetDefaultValues();
+            populateEvent();
             return new JacksonOtelLog(this);
         }
-
-        private void checkAndSetDefaultValues() {
-            data.computeIfAbsent(ATTRIBUTES_KEY, k -> new HashMap<>());
-        }
     }
 }

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/log/JacksonStandardOTelLog.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.model.log;
+
+/**
+ * A Jackson Standard implementation for {@link OpenTelemetryLog}.
+ *
+ * @since 2.11
+ */
+public class JacksonStandardOTelLog extends JacksonOtelLog {
+
+    JacksonStandardOTelLog(final JacksonOtelLog.Builder builder) {
+        super(builder);
+    }
+
+    public static JacksonStandardOTelLog.Builder builder() {
+        return new JacksonStandardOTelLog.Builder();
+    }
+
+    @Override
+    public String toJsonString() {
+        return getJsonNode().toString();
+    }
+
+    public static class Builder extends JacksonOtelLog.Builder {
+
+        @Override
+        public JacksonOtelLog build() {
+            populateEvent();
+            return new JacksonStandardOTelLog(this);
+        }
+
+    }
+}

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/log/OpenTelemetryLog.java

@@ -112,4 +112,21 @@ public interface OpenTelemetryLog extends Log {
      * @since 2.1
      */
     Object getBody();
+
+    /**
+     * Gets the scope of this log event.
+     *
+     * @return the scope
+     * @since 2.11
+     */
+    Map<String, Object> getScope();
+
+    /**
+     * Gets the resource of this log event.
+     *
+     * @return the resource
+     * @since 2.11
+     */
+    Map<String, Object> getResource();
+
 }