Skip to content

Commit dac4878

Browse files
LantaoJinLantaoJin
committed
Trigger GC only in dedicated coordinator
Signed-off-by: Lantao Jin <ltjin@amazon.com>
1 parent 18639ad commit dac4878

13 files changed

Lines changed: 83 additions & 27 deletions

File tree

‎integ-test/src/test/java/org/opensearch/sql/calcite/standalone/CalcitePPLIntegTestCase.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,10 @@
1919
import java.util.concurrent.atomic.AtomicReference;
2020
import lombok.RequiredArgsConstructor;
2121
import org.json.JSONObject;
22+
import org.mockito.Mock;
2223
import org.opensearch.client.RestClient;
2324
import org.opensearch.client.RestHighLevelClient;
25+
import org.opensearch.cluster.service.ClusterService;
2426
import org.opensearch.common.inject.AbstractModule;
2527
import org.opensearch.common.inject.Injector;
2628
import org.opensearch.common.inject.ModulesBuilder;
@@ -78,6 +80,8 @@
7880
public abstract class CalcitePPLIntegTestCase extends PPLIntegTestCase {
7981
protected PPLService pplService;
8082

83+
@Mock private ClusterService clusterService;
84+
8185
@Override
8286
public void init() throws IOException {
8387
RestHighLevelClient restClient =
@@ -86,7 +90,7 @@ public void init() throws IOException {
8690
DataSourceService dataSourceService =
8791
new DataSourceServiceImpl(
8892
new ImmutableSet.Builder<DataSourceFactory>()
89-
.add(new OpenSearchDataSourceFactory(client, getSettings()))
93+
.add(new OpenSearchDataSourceFactory(client, getSettings(), clusterService))
9094
.build(),
9195
getDataSourceMetadataStorage(),
9296
getDataSourceUserRoleHelper());

‎integ-test/src/test/java/org/opensearch/sql/ppl/StandaloneIT.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,9 +18,11 @@
1818
import java.util.concurrent.atomic.AtomicReference;
1919
import lombok.RequiredArgsConstructor;
2020
import org.junit.jupiter.api.Test;
21+
import org.mockito.Mock;
2122
import org.opensearch.client.Request;
2223
import org.opensearch.client.RestClient;
2324
import org.opensearch.client.RestHighLevelClient;
25+
import org.opensearch.cluster.service.ClusterService;
2426
import org.opensearch.common.inject.AbstractModule;
2527
import org.opensearch.common.inject.Injector;
2628
import org.opensearch.common.inject.ModulesBuilder;
@@ -74,6 +76,8 @@ public class StandaloneIT extends PPLIntegTestCase {
7476

7577
private PPLService pplService;
7678

79+
@Mock private ClusterService clusterService;
80+
7781
@Override
7882
public void init() throws Exception {
7983
super.init();
@@ -82,7 +86,7 @@ public void init() throws Exception {
8286
DataSourceService dataSourceService =
8387
new DataSourceServiceImpl(
8488
new ImmutableSet.Builder<DataSourceFactory>()
85-
.add(new OpenSearchDataSourceFactory(client, defaultSettings()))
89+
.add(new OpenSearchDataSourceFactory(client, defaultSettings(), clusterService))
8690
.build(),
8791
getDataSourceMetadataStorage(),
8892
getDataSourceUserRoleHelper());

‎integ-test/src/test/java/org/opensearch/sql/sql/StandalonePaginationIT.java‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,9 +21,11 @@
2121
import org.junit.Test;
2222
import org.junit.jupiter.api.DisplayNameGeneration;
2323
import org.junit.jupiter.api.DisplayNameGenerator;
24+
import org.mockito.Mock;
2425
import org.opensearch.client.Request;
2526
import org.opensearch.client.ResponseException;
2627
import org.opensearch.client.RestHighLevelClient;
28+
import org.opensearch.cluster.service.ClusterService;
2729
import org.opensearch.common.inject.Injector;
2830
import org.opensearch.common.inject.ModulesBuilder;
2931
import org.opensearch.common.unit.TimeValue;
@@ -62,6 +64,8 @@ public class StandalonePaginationIT extends SQLIntegTestCase {
6264

6365
private OpenSearchClient client;
6466

67+
@Mock private ClusterService clusterService;
68+
6569
@Override
6670
@SneakyThrows
6771
public void init() {
@@ -70,7 +74,7 @@ public void init() {
7074
DataSourceService dataSourceService =
7175
new DataSourceServiceImpl(
7276
new ImmutableSet.Builder<DataSourceFactory>()
73-
.add(new OpenSearchDataSourceFactory(client, defaultSettings()))
77+
.add(new OpenSearchDataSourceFactory(client, defaultSettings(), clusterService))
7478
.build(),
7579
getDataSourceMetadataStorage(),
7680
getDataSourceUserRoleHelper());
@@ -114,7 +118,7 @@ public void onFailure(Exception e) {
114118
}
115119

116120
// act 1, asserts in firstResponder
117-
var t = new OpenSearchIndex(client, defaultSettings(), "test");
121+
var t = new OpenSearchIndex(client, defaultSettings(), "test", clusterService);
118122
LogicalPlan p =
119123
new LogicalPaginate(
120124
1,

‎opensearch/src/main/java/org/opensearch/sql/opensearch/monitor/OpenSearchResourceMonitor.java‎

Lines changed: 23 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,9 @@
1010
import io.github.resilience4j.retry.RetryConfig;
1111
import java.util.function.Supplier;
1212
import lombok.extern.log4j.Log4j2;
13+
import org.opensearch.cluster.node.DiscoveryNode;
14+
import org.opensearch.cluster.node.DiscoveryNodes;
15+
import org.opensearch.cluster.service.ClusterService;
1316
import org.opensearch.core.common.unit.ByteSizeValue;
1417
import org.opensearch.sql.common.setting.Settings;
1518
import org.opensearch.sql.monitor.ResourceMonitor;
@@ -24,9 +27,11 @@ public class OpenSearchResourceMonitor extends ResourceMonitor {
2427
private final Settings settings;
2528
private final Retry retry;
2629
private final OpenSearchMemoryHealthy memoryMonitor;
30+
private final ClusterService clusterService;
2731

2832
/** Constructor. */
29-
public OpenSearchResourceMonitor(Settings settings, OpenSearchMemoryHealthy memoryMonitor) {
33+
public OpenSearchResourceMonitor(
34+
Settings settings, OpenSearchMemoryHealthy memoryMonitor, ClusterService clusterService) {
3035
this.settings = settings;
3136
RetryConfig config =
3237
RetryConfig.custom()
@@ -35,17 +40,31 @@ public OpenSearchResourceMonitor(Settings settings, OpenSearchMemoryHealthy memo
3540
.retryExceptions(OpenSearchMemoryHealthy.MemoryUsageExceedException.class)
3641
.ignoreExceptions(OpenSearchMemoryHealthy.MemoryUsageExceedFastFailureException.class)
3742
.build();
38-
retry = Retry.of("mem", config);
39-
retry
43+
this.retry = Retry.of("mem", config);
44+
this.retry
4045
.getEventPublisher()
4146
.onRetry(
4247
event -> {
43-
if (event.getNumberOfRetryAttempts() == 1) {
48+
if (event.getNumberOfRetryAttempts() == 2 && canTriggerGC()) {
4449
System.gc();
4550
log.warn("isMemoryHealthy() failed in first retry, triggered System.gc()");
4651
}
4752
});
4853
this.memoryMonitor = memoryMonitor;
54+
this.clusterService = clusterService;
55+
}
56+
57+
private boolean canTriggerGC() {
58+
return clusterService.isStateInitialised()
59+
&& isDedicatedCoordinator(clusterService.localNode());
60+
}
61+
62+
/** Keep the same behaviour with {@link DiscoveryNodes#getCoordinatingOnlyNodes()} */
63+
private boolean isDedicatedCoordinator(DiscoveryNode local) {
64+
return !local.isDataNode() // Not a data role
65+
&& !local.isWarmNode() // Not a warm data role
66+
&& !local.isIngestNode() // Not an ingest role
67+
&& !local.isClusterManagerNode(); // Not a manager role
4968
}
5069

5170
/**

‎opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchDataSourceFactory.java‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
package org.opensearch.sql.opensearch.storage;
77

88
import lombok.RequiredArgsConstructor;
9+
import org.opensearch.cluster.service.ClusterService;
910
import org.opensearch.sql.common.setting.Settings;
1011
import org.opensearch.sql.datasource.model.DataSource;
1112
import org.opensearch.sql.datasource.model.DataSourceMetadata;
@@ -21,6 +22,8 @@ public class OpenSearchDataSourceFactory implements DataSourceFactory {
2122

2223
private final Settings settings;
2324

25+
private final ClusterService clusterService;
26+
2427
@Override
2528
public DataSourceType getDataSourceType() {
2629
return DataSourceType.OPENSEARCH;
@@ -31,6 +34,6 @@ public DataSource createDataSource(DataSourceMetadata metadata) {
3134
return new DataSource(
3235
metadata.getName(),
3336
DataSourceType.OPENSEARCH,
34-
new OpenSearchStorageEngine(client, settings));
37+
new OpenSearchStorageEngine(client, settings, clusterService));
3538
}
3639
}

‎opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchIndex.java‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
import org.apache.calcite.plan.RelOptCluster;
1919
import org.apache.calcite.plan.RelOptTable;
2020
import org.apache.calcite.rel.RelNode;
21+
import org.opensearch.cluster.service.ClusterService;
2122
import org.opensearch.common.unit.TimeValue;
2223
import org.opensearch.sql.calcite.plan.AbstractOpenSearchTable;
2324
import org.opensearch.sql.common.setting.Settings;
@@ -98,11 +99,15 @@ public class OpenSearchIndex extends AbstractOpenSearchTable {
9899
/** The cached max result window setting of index. */
99100
private Integer cachedMaxResultWindow = null;
100101

102+
private final ClusterService clusterService;
103+
101104
/** Constructor. */
102-
public OpenSearchIndex(OpenSearchClient client, Settings settings, String indexName) {
105+
public OpenSearchIndex(
106+
OpenSearchClient client, Settings settings, String indexName, ClusterService clusterService) {
103107
this.client = client;
104108
this.settings = settings;
105109
this.indexName = new OpenSearchRequest.IndexName(indexName);
110+
this.clusterService = clusterService;
106111
}
107112

108113
@Override
@@ -276,7 +281,8 @@ public OpenSearchRequestBuilder createRequestBuilder() {
276281
}
277282

278283
public OpenSearchResourceMonitor createOpenSearchResourceMonitor() {
279-
return new OpenSearchResourceMonitor(getSettings(), new OpenSearchMemoryHealthy());
284+
return new OpenSearchResourceMonitor(
285+
getSettings(), new OpenSearchMemoryHealthy(), clusterService);
280286
}
281287

282288
public OpenSearchRequest buildRequest(OpenSearchRequestBuilder requestBuilder) {

‎opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchStorageEngine.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,8 +7,10 @@
77

88
import static org.opensearch.sql.utils.SystemIndexUtils.isSystemIndex;
99

10+
import lombok.AllArgsConstructor;
1011
import lombok.Getter;
1112
import lombok.RequiredArgsConstructor;
13+
import org.opensearch.cluster.service.ClusterService;
1214
import org.opensearch.sql.DataSourceSchemaName;
1315
import org.opensearch.sql.common.setting.Settings;
1416
import org.opensearch.sql.opensearch.client.OpenSearchClient;
@@ -18,19 +20,22 @@
1820

1921
/** OpenSearch storage engine implementation. */
2022
@RequiredArgsConstructor
23+
@AllArgsConstructor
2124
public class OpenSearchStorageEngine implements StorageEngine {
2225

2326
/** OpenSearch client connection. */
2427
@Getter private final OpenSearchClient client;
2528

2629
@Getter private final Settings settings;
2730

31+
private ClusterService clusterService;
32+
2833
@Override
2934
public Table getTable(DataSourceSchemaName dataSourceSchemaName, String name) {
3035
if (isSystemIndex(name)) {
3136
return new OpenSearchSystemIndex(client, name);
3237
} else {
33-
return new OpenSearchIndex(client, settings, name);
38+
return new OpenSearchIndex(client, settings, name, clusterService);
3439
}
3540
}
3641
}

‎opensearch/src/test/java/org/opensearch/sql/opensearch/monitor/OpenSearchResourceMonitorTest.java‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
import org.junit.jupiter.api.extension.ExtendWith;
1818
import org.mockito.Mock;
1919
import org.mockito.junit.jupiter.MockitoExtension;
20+
import org.opensearch.cluster.service.ClusterService;
2021
import org.opensearch.core.common.unit.ByteSizeValue;
2122
import org.opensearch.sql.common.setting.Settings;
2223

@@ -27,6 +28,8 @@ class OpenSearchResourceMonitorTest {
2728

2829
@Mock private OpenSearchMemoryHealthy memoryMonitor;
2930

31+
@Mock private ClusterService clusterService;
32+
3033
@BeforeEach
3134
public void setup() {
3235
when(settings.getSettingValue(Settings.Key.QUERY_MEMORY_LIMIT))
@@ -36,19 +39,17 @@ public void setup() {
3639
@Test
3740
void isHealthy() {
3841
when(memoryMonitor.isMemoryHealthy(anyLong())).thenReturn(true);
39-
4042
OpenSearchResourceMonitor resourceMonitor =
41-
new OpenSearchResourceMonitor(settings, memoryMonitor);
43+
new OpenSearchResourceMonitor(settings, memoryMonitor, clusterService);
4244
assertTrue(resourceMonitor.isHealthy());
4345
}
4446

4547
@Test
4648
void notHealthyFastFailure() {
4749
when(memoryMonitor.isMemoryHealthy(anyLong()))
4850
.thenThrow(OpenSearchMemoryHealthy.MemoryUsageExceedFastFailureException.class);
49-
5051
OpenSearchResourceMonitor resourceMonitor =
51-
new OpenSearchResourceMonitor(settings, memoryMonitor);
52+
new OpenSearchResourceMonitor(settings, memoryMonitor, clusterService);
5253
assertFalse(resourceMonitor.isHealthy());
5354
verify(memoryMonitor, times(1)).isMemoryHealthy(anyLong());
5455
}
@@ -59,20 +60,19 @@ void notHealthyWithRetry() {
5960
.thenThrow(OpenSearchMemoryHealthy.MemoryUsageExceedException.class);
6061

6162
OpenSearchResourceMonitor resourceMonitor =
62-
new OpenSearchResourceMonitor(settings, memoryMonitor);
63+
new OpenSearchResourceMonitor(settings, memoryMonitor, clusterService);
6364
assertFalse(resourceMonitor.isHealthy());
6465
verify(memoryMonitor, times(3)).isMemoryHealthy(anyLong());
6566
}
6667

6768
@Test
6869
void healthyWithRetry() {
69-
7070
when(memoryMonitor.isMemoryHealthy(anyLong()))
7171
.thenThrow(OpenSearchMemoryHealthy.MemoryUsageExceedException.class)
7272
.thenReturn(true);
7373

7474
OpenSearchResourceMonitor resourceMonitor =
75-
new OpenSearchResourceMonitor(settings, memoryMonitor);
75+
new OpenSearchResourceMonitor(settings, memoryMonitor, clusterService);
7676
assertTrue(resourceMonitor.isHealthy());
7777
verify(memoryMonitor, times(2)).isMemoryHealthy(anyLong());
7878
}

‎opensearch/src/test/java/org/opensearch/sql/opensearch/storage/OpenSearchDataSourceFactoryTest.java‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
import org.junit.jupiter.api.extension.ExtendWith;
1414
import org.mockito.Mock;
1515
import org.mockito.junit.jupiter.MockitoExtension;
16+
import org.opensearch.cluster.service.ClusterService;
1617
import org.opensearch.sql.common.setting.Settings;
1718
import org.opensearch.sql.datasource.model.DataSource;
1819
import org.opensearch.sql.datasource.model.DataSourceMetadata;
@@ -28,11 +29,13 @@ class OpenSearchDataSourceFactoryTest {
2829

2930
@Mock private DataSourceMetadata dataSourceMetadata;
3031

32+
@Mock private ClusterService clusterService;
33+
3134
private OpenSearchDataSourceFactory factory;
3235

3336
@BeforeEach
3437
public void setup() {
35-
factory = new OpenSearchDataSourceFactory(client, settings);
38+
factory = new OpenSearchDataSourceFactory(client, settings, clusterService);
3639
}
3740

3841
@Test

‎opensearch/src/test/java/org/opensearch/sql/opensearch/storage/OpenSearchIndexTest.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
import org.junit.jupiter.api.extension.ExtendWith;
4040
import org.mockito.Mock;
4141
import org.mockito.junit.jupiter.MockitoExtension;
42+
import org.opensearch.cluster.service.ClusterService;
4243
import org.opensearch.common.unit.TimeValue;
4344
import org.opensearch.sql.ast.tree.Sort;
4445
import org.opensearch.sql.common.setting.Settings;
@@ -74,11 +75,13 @@ class OpenSearchIndexTest {
7475

7576
@Mock private IndexMapping mapping;
7677

78+
@Mock private ClusterService clusterService;
79+
7780
private OpenSearchIndex index;
7881

7982
@BeforeEach
8083
void setUp() {
81-
this.index = new OpenSearchIndex(client, settings, "test");
84+
this.index = new OpenSearchIndex(client, settings, "test", clusterService);
8285
lenient().when(settings.getSettingValue(Settings.Key.FIELD_TYPE_TOLERANCE)).thenReturn(true);
8386
lenient()
8487
.when(settings.getSettingValue(Settings.Key.SQL_CURSOR_KEEP_ALIVE))
@@ -165,7 +168,7 @@ void checkCacheUsedForFieldMappings() {
165168
.thenReturn(Map.of("name", OpenSearchDataType.of(MappingType.Keyword)));
166169
when(client.getIndexMappings("test")).thenReturn(ImmutableMap.of("test", mapping));
167170

168-
OpenSearchIndex index = new OpenSearchIndex(client, settings, "test");
171+
OpenSearchIndex index = new OpenSearchIndex(client, settings, "test", clusterService);
169172
assertThat(index.getFieldTypes(), allOf(aMapWithSize(1), hasEntry("name", STRING)));
170173
assertThat(
171174
index.getFieldOpenSearchTypes(),

0 commit comments

Comments
 (0)