Created new urgent priority threadpool for remote cluster state uploads

CHANGELOG.md

@@ -14,6 +14,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Provide service accounts tokens to extensions ([#9618](https://github.com/opensearch-project/OpenSearch/pull/9618))
 - [Admission control] Add enhancements to FS stats to include read/write time, queue size and IO time ([#10541](https://github.com/opensearch-project/OpenSearch/pull/10541))
 - [Admission control] Add Resource usage collector service and resource usage tracker  ([#9890](https://github.com/opensearch-project/OpenSearch/pull/9890))
+- Change file names for remote cluster state ([#10557](https://github.com/opensearch-project/OpenSearch/pull/10557))
 
 ### Dependencies
 - Bump `log4j-core` from 2.18.0 to 2.19.0
@@ -105,12 +106,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add the means to extract the contextual properties from HttpChannel, TcpCChannel and TrasportChannel without excessive typecasting ([#10562](https://github.com/opensearch-project/OpenSearch/pull/10562))
 - [Remote Store] Add Remote Store backpressure rejection stats to `_nodes/stats` ([#10524](https://github.com/opensearch-project/OpenSearch/pull/10524))
 - [BUG] Fix java.lang.SecurityException in repository-gcs plugin ([#10642](https://github.com/opensearch-project/OpenSearch/pull/10642))
+- Add telemetry tracer/metric enable flag and integ test. ([#10395](https://github.com/opensearch-project/OpenSearch/pull/10395))
 
 ### Deprecated
 
 ### Removed
 
 ### Fixed
+- Fix failure in dissect ingest processor parsing empty brackets ([#9225](https://github.com/opensearch-project/OpenSearch/pull/9255))
 - Fix class_cast_exception when passing int to _version and other metadata fields in ingest simulate API ([#10101](https://github.com/opensearch-project/OpenSearch/pull/10101))
 - Fix Segment Replication ShardLockObtainFailedException bug during index corruption ([10370](https://github.com/opensearch-project/OpenSearch/pull/10370))
 - Fix some test methods in SimulatePipelineRequestParsingTests never run and fix test failure ([#10496](https://github.com/opensearch-project/OpenSearch/pull/10496))

libs/dissect/src/main/java/org/opensearch/dissect/DissectParser.java

@@ -231,7 +231,10 @@ public Map<String, String> parse(String inputString) {
             int lookAheadMatches;
             // start walking the input string byte by byte, look ahead for matches where needed
             // if a match is found jump forward to the end of the match
-            for (; i < input.length; i++) {
+            while (i < input.length) {
+                // start is only used to record the value of i
+                int start = i;
+
                 lookAheadMatches = 0;
                 // potential match between delimiter and input string
                 if (delimiter.length > 0 && input[i] == delimiter[0]) {
@@ -283,8 +286,14 @@ public Map<String, String> parse(String inputString) {
                         delimiter = dissectPair.getDelimiter().getBytes(StandardCharsets.UTF_8);
                         // i is always one byte after the last found delimiter, aka the start of the next value
                         valueStart = i;
+                    } else {
+                        i++;
                     }
+                } else {
+                    i++;
                 }
+                // i should change anyway
+                assert (i != start);
             }
             // the last key, grab the rest of the input (unless consecutive delimiters already grabbed the last key)
             // and there is no trailing delimiter

libs/telemetry/src/main/java/org/opensearch/telemetry/metrics/MetricsTelemetry.java

@@ -10,14 +10,12 @@
 
 import org.opensearch.common.annotation.ExperimentalApi;
 
-import java.io.Closeable;
-
 /**
  * Interface for metrics telemetry providers
   *
  * @opensearch.experimental
  */
 @ExperimentalApi
-public interface MetricsTelemetry extends MetricsRegistry, Closeable {
+public interface MetricsTelemetry extends MetricsRegistry {
 
 }

libs/telemetry/src/main/java/org/opensearch/telemetry/tracing/DefaultTracer.java

@@ -85,6 +85,11 @@ public SpanScope withSpanInScope(Span span) {
         return DefaultSpanScope.create(span, tracerContextStorage).attach();
     }
 
+    @Override
+    public boolean isRecording() {
+        return true;
+    }
+
     private Span createSpan(SpanCreationContext spanCreationContext, Span parentSpan) {
         return tracingTelemetry.createSpan(spanCreationContext, parentSpan);
     }

libs/telemetry/src/main/java/org/opensearch/telemetry/tracing/Tracer.java

@@ -53,4 +53,10 @@ public interface Tracer extends HttpTracer, Closeable {
      */
     SpanScope withSpanInScope(Span span);
 
+    /**
+     * Tells if the traces are being recorded or not
+     * @return boolean
+     */
+    boolean isRecording();
+
 }

libs/telemetry/src/main/java/org/opensearch/telemetry/tracing/noop/NoopTracer.java

@@ -54,6 +54,11 @@ public SpanScope withSpanInScope(Span span) {
         return SpanScope.NO_OP;
     }
 
+    @Override
+    public boolean isRecording() {
+        return false;
+    }
+
     @Override
     public void close() {
 

libs/telemetry/src/test/java/org/opensearch/telemetry/tracing/DefaultTracerTests.java

@@ -62,6 +62,7 @@ public void testCreateSpan() {
 
         String spanName = defaultTracer.getCurrentSpan().getSpan().getSpanName();
         assertEquals("span_name", spanName);
+        assertTrue(defaultTracer.isRecording());
     }
 
     @SuppressWarnings("unchecked")

modules/ingest-common/src/test/java/org/opensearch/ingest/common/DissectProcessorTests.java

@@ -155,4 +155,28 @@ public void testNullValueWithOutIgnoreMissing() {
         IngestDocument ingestDocument = new IngestDocument(originalIngestDocument);
         expectThrows(IllegalArgumentException.class, () -> processor.execute(ingestDocument));
     }
+
+    public void testMatchEmptyBrackets() {
+        IngestDocument ingestDocument = new IngestDocument(
+            "_index",
+            "_id",
+            null,
+            null,
+            null,
+            Collections.singletonMap("message", "[foo],[bar],[]")
+        );
+        DissectProcessor dissectProcessor = new DissectProcessor("", null, "message", "[%{a}],[%{b}],[%{c}]", "", true);
+        dissectProcessor.execute(ingestDocument);
+        assertEquals("foo", ingestDocument.getFieldValue("a", String.class));
+        assertEquals("bar", ingestDocument.getFieldValue("b", String.class));
+        assertEquals("", ingestDocument.getFieldValue("c", String.class));
+
+        ingestDocument = new IngestDocument("_index", "_id", null, null, null, Collections.singletonMap("message", "{}{}{}{baz}"));
+        dissectProcessor = new DissectProcessor("", null, "message", "{%{a}}{%{b}}{%{c}}{%{d}}", "", true);
+        dissectProcessor.execute(ingestDocument);
+        assertEquals("", ingestDocument.getFieldValue("a", String.class));
+        assertEquals("", ingestDocument.getFieldValue("b", String.class));
+        assertEquals("", ingestDocument.getFieldValue("c", String.class));
+        assertEquals("baz", ingestDocument.getFieldValue("d", String.class));
+    }
 }

modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/200_dissect_processor.yml

@@ -84,3 +84,38 @@ teardown:
             }
           ]
         }
+
+---
+"Test dissect processor can match empty brackets":
+  - do:
+      ingest.put_pipeline:
+        id: "my_pipeline"
+        body:  >
+          {
+            "description": "_description",
+            "processors": [
+              {
+                "dissect" : {
+                  "field" : "message",
+                  "pattern" : "[%{a}][%{b}][%{c}]"
+                }
+              }
+            ]
+          }
+  - match: { acknowledged: true }
+
+  - do:
+      index:
+        index: test
+        id: 1
+        pipeline: "my_pipeline"
+        body: {message: "[foo][bar][]"}
+
+  - do:
+      get:
+        index: test
+        id: 1
+  - match: { _source.message: "[foo][bar][]" }
+  - match: { _source.a: "foo" }
+  - match: { _source.b: "bar" }
+  - match: { _source.c: "" }

plugins/telemetry-otel/src/internalClusterTest/java/org/opensearch/telemetry/tracing/IntegrationTestOTelTelemetryPlugin.java → plugins/telemetry-otel/src/internalClusterTest/java/org/opensearch/telemetry/IntegrationTestOTelTelemetryPlugin.java

@@ -6,12 +6,9 @@
  * compatible open source license.
  */
 
-package org.opensearch.telemetry.tracing;
+package org.opensearch.telemetry;
 
 import org.opensearch.common.settings.Settings;
-import org.opensearch.telemetry.OTelTelemetryPlugin;
-import org.opensearch.telemetry.Telemetry;
-import org.opensearch.telemetry.TelemetrySettings;
 
 import java.util.Optional;
 

plugins/telemetry-otel/src/internalClusterTest/java/org/opensearch/telemetry/metrics/InMemorySingletonMetricsExporter.java

@@ -0,0 +1,65 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.telemetry.metrics;
+
+import java.util.Collection;
+import java.util.List;
+
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.metrics.InstrumentType;
+import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
+import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.metrics.export.MetricExporter;
+import io.opentelemetry.sdk.testing.exporter.InMemoryMetricExporter;
+
+public class InMemorySingletonMetricsExporter implements MetricExporter {
+
+    public static final InMemorySingletonMetricsExporter INSTANCE = new InMemorySingletonMetricsExporter(InMemoryMetricExporter.create());
+
+    private static InMemoryMetricExporter delegate;
+
+    public static InMemorySingletonMetricsExporter create() {
+        return INSTANCE;
+    }
+
+    private InMemorySingletonMetricsExporter(InMemoryMetricExporter delegate) {
+        InMemorySingletonMetricsExporter.delegate = delegate;
+    }
+
+    @Override
+    public CompletableResultCode export(Collection<MetricData> metrics) {
+        return delegate.export(metrics);
+    }
+
+    @Override
+    public CompletableResultCode flush() {
+        return delegate.flush();
+    }
+
+    @Override
+    public CompletableResultCode shutdown() {
+        return delegate.shutdown();
+    }
+
+    public List<MetricData> getFinishedMetricItems() {
+        return delegate.getFinishedMetricItems();
+    }
+
+    /**
+     * Clears the state.
+     */
+    public void reset() {
+        delegate.reset();
+    }
+
+    @Override
+    public AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) {
+        return delegate.getAggregationTemporality(instrumentType);
+    }
+}

plugins/telemetry-otel/src/internalClusterTest/java/org/opensearch/telemetry/metrics/TelemetryMetricsDisabledSanityIT.java

@@ -0,0 +1,62 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.telemetry.metrics;
+
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.unit.TimeValue;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.telemetry.IntegrationTestOTelTelemetryPlugin;
+import org.opensearch.telemetry.OTelTelemetrySettings;
+import org.opensearch.telemetry.TelemetrySettings;
+import org.opensearch.telemetry.metrics.noop.NoopCounter;
+import org.opensearch.telemetry.metrics.noop.NoopMetricsRegistry;
+import org.opensearch.test.OpenSearchIntegTestCase;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, minNumDataNodes = 1)
+public class TelemetryMetricsDisabledSanityIT extends OpenSearchIntegTestCase {
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder()
+            .put(super.nodeSettings(nodeOrdinal))
+            .put(TelemetrySettings.METRICS_FEATURE_ENABLED_SETTING.getKey(), false)
+            .put(
+                OTelTelemetrySettings.OTEL_METRICS_EXPORTER_CLASS_SETTING.getKey(),
+                "org.opensearch.telemetry.metrics.InMemorySingletonMetricsExporter"
+            )
+            .put(TelemetrySettings.METRICS_PUBLISH_INTERVAL_SETTING.getKey(), TimeValue.timeValueSeconds(1))
+            .build();
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Arrays.asList(IntegrationTestOTelTelemetryPlugin.class);
+    }
+
+    @Override
+    protected boolean addMockTelemetryPlugin() {
+        return false;
+    }
+
+    public void testSanityChecksWhenMetricsDisabled() throws Exception {
+        MetricsRegistry metricsRegistry = internalCluster().getInstance(MetricsRegistry.class);
+
+        Counter counter = metricsRegistry.createCounter("test-counter", "test", "1");
+        counter.add(1.0);
+
+        Thread.sleep(2000);
+
+        assertTrue(metricsRegistry instanceof NoopMetricsRegistry);
+        assertTrue(counter instanceof NoopCounter);
+    }
+
+}