Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Created new urgent priority threadpool for remote cluster state uploads #1

Closed
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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))
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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]) {
Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 {

}
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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();

}
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,11 @@ public SpanScope withSpanInScope(Span span) {
return SpanScope.NO_OP;
}

@Override
public boolean isRecording() {
return false;
}

@Override
public void close() {

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@ public void testCreateSpan() {

String spanName = defaultTracer.getCurrentSpan().getSpan().getSpanName();
assertEquals("span_name", spanName);
assertTrue(defaultTracer.isRecording());
}

@SuppressWarnings("unchecked")
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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));
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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: "" }
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down
Original file line number Diff line number Diff line change
@@ -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);
}
}
Original file line number Diff line number Diff line change
@@ -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);
}

}
Loading