Skip to content

Commit d4c2392

Browse files
tteofilitteofili
committed
Merge branch 'main' of github.com:elastic/elasticsearch into esql_namedqueries
2 parents 7e67a7b + 8085eea commit d4c2392

File tree

9 files changed

+145
-38
lines changed

9 files changed

+145
-38
lines changed

‎docs/changelog/127628.yaml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
pr: 127628
2+
summary: Ensure config reload on ..data symlink switch for CSI driver support
3+
area: Infra/Settings
4+
type: enhancement
5+
issues: []

‎server/src/internalClusterTest/java/org/elasticsearch/reservedstate/service/FileSettingsServiceIT.java‎

Lines changed: 36 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,6 @@
4444
import static org.elasticsearch.health.HealthStatus.YELLOW;
4545
import static org.elasticsearch.indices.recovery.RecoverySettings.INDICES_RECOVERY_MAX_BYTES_PER_SEC_SETTING;
4646
import static org.elasticsearch.node.Node.INITIAL_STATE_TIMEOUT_SETTING;
47-
import static org.elasticsearch.test.NodeRoles.dataNode;
4847
import static org.elasticsearch.test.NodeRoles.dataOnlyNode;
4948
import static org.elasticsearch.test.NodeRoles.masterNode;
5049
import static org.hamcrest.Matchers.allOf;
@@ -139,6 +138,11 @@ private void assertMasterNode(Client client, String node) {
139138

140139
public static void writeJSONFile(String node, String json, Logger logger, Long version) throws Exception {
141140
FileSettingsService fileSettingsService = internalCluster().getInstance(FileSettingsService.class, node);
141+
writeJSONFile(node, json, logger, version, fileSettingsService.watchedFile());
142+
}
143+
144+
public static void writeJSONFile(String node, String json, Logger logger, Long version, Path targetPath) throws Exception {
145+
FileSettingsService fileSettingsService = internalCluster().getInstance(FileSettingsService.class, node);
142146

143147
Files.createDirectories(fileSettingsService.watchedFileDir());
144148
Path tempFilePath = createTempFile();
@@ -152,8 +156,8 @@ public static void writeJSONFile(String node, String json, Logger logger, Long v
152156
do {
153157
try {
154158
// this can fail on Windows because of timing
155-
Files.move(tempFilePath, fileSettingsService.watchedFile(), StandardCopyOption.ATOMIC_MOVE);
156-
logger.info("--> after writing JSON config to node {} with path {}", node, tempFilePath);
159+
Files.move(tempFilePath, targetPath, StandardCopyOption.ATOMIC_MOVE);
160+
logger.info("--> after writing JSON config to node {} with path {}", node, targetPath);
157161
return;
158162
} catch (IOException e) {
159163
logger.info("--> retrying writing a settings file [{}]", retryCount);
@@ -503,6 +507,35 @@ public void testSettingsAppliedOnMasterReElection() throws Exception {
503507
assertClusterStateSaveOK(savedClusterState.v1(), savedClusterState.v2(), "43mb");
504508
}
505509

510+
public void testSymlinkUpdateTriggerReload() throws Exception {
511+
internalCluster().setBootstrapMasterNodeIndex(0);
512+
final String masterNode = internalCluster().startMasterOnlyNode();
513+
FileSettingsService masterFileSettingsService = internalCluster().getInstance(FileSettingsService.class, masterNode);
514+
Path baseDir = masterFileSettingsService.watchedFileDir();
515+
assertBusy(() -> assertTrue(masterFileSettingsService.watching()));
516+
517+
{
518+
var savedClusterState = setupClusterStateListener(masterNode);
519+
// Create the settings.json as a symlink to simulate k8 setup
520+
// settings.json -> ..data/settings.json
521+
// ..data -> ..TIMESTAMP_TEMP_FOLDER_1
522+
var fileDir = Files.createDirectories(baseDir.resolve("..TIMESTAMP_TEMP_FOLDER_1"));
523+
writeJSONFile(masterNode, testJSON, logger, versionCounter.incrementAndGet(), fileDir.resolve("settings.json"));
524+
var dataDir = Files.createSymbolicLink(baseDir.resolve("..data"), fileDir.getFileName());
525+
Files.createSymbolicLink(baseDir.resolve("settings.json"), dataDir.getFileName().resolve("settings.json"));
526+
assertClusterStateSaveOK(savedClusterState.v1(), savedClusterState.v2(), "50mb");
527+
}
528+
{
529+
var savedClusterState = setupClusterStateListener(masterNode);
530+
// Update ..data symlink to ..data -> ..TIMESTAMP_TEMP_FOLDER_2 to simulate kubernetes secret update
531+
var fileDir = Files.createDirectories(baseDir.resolve("..TIMESTAMP_TEMP_FOLDER_2"));
532+
writeJSONFile(masterNode, testJSON43mb, logger, versionCounter.incrementAndGet(), fileDir.resolve("settings.json"));
533+
Files.deleteIfExists(baseDir.resolve("..data"));
534+
Files.createSymbolicLink(baseDir.resolve("..data"), fileDir.getFileName());
535+
assertClusterStateSaveOK(savedClusterState.v1(), savedClusterState.v2(), "43mb");
536+
}
537+
}
538+
506539
public void testHealthIndicatorWithSingleNode() throws Exception {
507540
internalCluster().setBootstrapMasterNodeIndex(0);
508541
logger.info("--> start the node");

‎server/src/main/java/org/elasticsearch/common/file/AbstractFileWatchingService.java‎

Lines changed: 25 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -27,11 +27,11 @@
2727
import java.nio.file.attribute.BasicFileAttributes;
2828
import java.nio.file.attribute.FileTime;
2929
import java.util.HashMap;
30-
import java.util.HashSet;
3130
import java.util.List;
3231
import java.util.Map;
3332
import java.util.Set;
3433
import java.util.concurrent.ExecutionException;
34+
import java.util.stream.Collectors;
3535
import java.util.stream.Stream;
3636

3737
/**
@@ -237,14 +237,16 @@ protected final void watcherThread() {
237237
key.reset();
238238

239239
if (key == settingsDirWatchKey) {
240-
// there may be multiple events for the same file - we only want to re-read once
241-
Set<Path> processedFiles = new HashSet<>();
242-
for (WatchEvent<?> e : events) {
243-
Path fullFile = settingsDir.resolve(e.context().toString());
244-
if (processedFiles.add(fullFile)) {
245-
if (fileChanged(fullFile)) {
246-
process(fullFile);
247-
}
240+
Set<Path> changedPaths = events.stream()
241+
.map(event -> settingsDir.resolve(event.context().toString()))
242+
.collect(Collectors.toSet());
243+
for (var changedPath : changedPaths) {
244+
// If a symlinked dir changed in the settings dir, it could be linked to other symlinks, so reprocess all files
245+
if (filesIsDirectory(changedPath) && filesIsSymbolicLink(changedPath)) {
246+
reprocessAllChangedFilesInSettingsDir();
247+
break;
248+
} else if (fileChanged(changedPath)) {
249+
process(changedPath);
248250
}
249251
}
250252
} else if (key == configDirWatchKey) {
@@ -257,14 +259,7 @@ protected final void watcherThread() {
257259
settingsDirWatchKey = enableDirectoryWatcher(settingsDirWatchKey, settingsDir);
258260

259261
// re-read the settings directory, and ping for any changes
260-
try (Stream<Path> files = filesList(settingsDir)) {
261-
for (var f = files.iterator(); f.hasNext();) {
262-
Path file = f.next();
263-
if (fileChanged(file)) {
264-
process(file);
265-
}
266-
}
267-
}
262+
reprocessAllChangedFilesInSettingsDir();
268263
} else if (settingsDirWatchKey != null) {
269264
settingsDirWatchKey.cancel();
270265
}
@@ -279,6 +274,17 @@ protected final void watcherThread() {
279274
}
280275
}
281276

277+
private void reprocessAllChangedFilesInSettingsDir() throws IOException, InterruptedException {
278+
try (Stream<Path> files = filesList(settingsDir)) {
279+
for (var f = files.iterator(); f.hasNext();) {
280+
Path file = f.next();
281+
if (fileChanged(file)) {
282+
process(file);
283+
}
284+
}
285+
}
286+
}
287+
282288
protected final synchronized void stopWatcher() {
283289
if (watching()) {
284290
logger.debug("stopping watcher ...");
@@ -378,6 +384,8 @@ private record FileUpdateState(long timestamp, String path, Object fileKey) {}
378384

379385
protected abstract boolean filesIsDirectory(Path path);
380386

387+
protected abstract boolean filesIsSymbolicLink(Path path);
388+
381389
protected abstract <A extends BasicFileAttributes> A filesReadAttributes(Path path, Class<A> clazz) throws IOException;
382390

383391
protected abstract Stream<Path> filesList(Path dir) throws IOException;

‎server/src/main/java/org/elasticsearch/lucene/grouping/GroupingDocValuesSelector.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -100,7 +100,7 @@ public void setNextReader(LeafReaderContext readerContext) throws IOException {
100100
public boolean advanceExact(int target) throws IOException {
101101
if (sorted.advanceExact(target)) {
102102
if (sorted.docValueCount() > 1) {
103-
throw new IllegalStateException(
103+
throw new IllegalArgumentException(
104104
"failed to extract doc:" + target + ", the grouping field must be single valued"
105105
);
106106
}
@@ -124,7 +124,7 @@ public long longValue() throws IOException {
124124
};
125125
}
126126
}
127-
default -> throw new IllegalStateException("unexpected doc values type " + type + "` for field `" + field + "`");
127+
default -> throw new IllegalArgumentException("unexpected doc values type " + type + "` for field `" + field + "`");
128128
}
129129
}
130130

@@ -200,7 +200,7 @@ public void setNextReader(LeafReaderContext readerContext) throws IOException {
200200
public boolean advanceExact(int target) throws IOException {
201201
if (sorted.advanceExact(target)) {
202202
if (sorted.docValueCount() > 1) {
203-
throw new IllegalStateException(
203+
throw new IllegalArgumentException(
204204
"failed to extract doc:" + target + ", the grouping field must be single valued"
205205
);
206206
}
@@ -233,7 +233,7 @@ public int getValueCount() {
233233
};
234234
}
235235
}
236-
default -> throw new IllegalStateException("unexpected doc values type " + type + "` for field `" + field + "`");
236+
default -> throw new IllegalArgumentException("unexpected doc values type " + type + "` for field `" + field + "`");
237237
}
238238
}
239239

‎server/src/main/java/org/elasticsearch/reservedstate/service/FileSettingsService.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -467,6 +467,11 @@ protected boolean filesIsDirectory(Path path) {
467467
return Files.isDirectory(path);
468468
}
469469

470+
@Override
471+
protected boolean filesIsSymbolicLink(Path path) {
472+
return Files.isSymbolicLink(path);
473+
}
474+
470475
@Override
471476
protected <A extends BasicFileAttributes> A filesReadAttributes(Path path, Class<A> clazz) throws IOException {
472477
return Files.readAttributes(path, clazz);

‎server/src/test/java/org/elasticsearch/common/file/AbstractFileWatchingServiceTests.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,11 @@ protected boolean filesIsDirectory(Path path) {
9898
return Files.isDirectory(path);
9999
}
100100

101+
@Override
102+
protected boolean filesIsSymbolicLink(Path path) {
103+
return Files.isSymbolicLink(path);
104+
}
105+
101106
@Override
102107
protected <A extends BasicFileAttributes> A filesReadAttributes(Path path, Class<A> clazz) throws IOException {
103108
return Files.readAttributes(path, clazz);

‎server/src/test/java/org/elasticsearch/common/file/MasterNodeFileWatchingServiceTests.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,11 @@ protected boolean filesIsDirectory(Path path) {
8080
return Files.isDirectory(path);
8181
}
8282

83+
@Override
84+
protected boolean filesIsSymbolicLink(Path path) {
85+
return Files.isSymbolicLink(path);
86+
}
87+
8388
@Override
8489
protected <A extends BasicFileAttributes> A filesReadAttributes(Path path, Class<A> clazz) throws IOException {
8590
return Files.readAttributes(path, clazz);

‎x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/ForkIT.java‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
import org.elasticsearch.action.index.IndexRequest;
1111
import org.elasticsearch.action.support.WriteRequest;
1212
import org.elasticsearch.common.settings.Settings;
13+
import org.elasticsearch.compute.operator.DriverProfile;
1314
import org.elasticsearch.xpack.esql.VerificationException;
1415
import org.elasticsearch.xpack.esql.parser.ParsingException;
1516
import org.junit.Before;
@@ -730,6 +731,35 @@ public void testOneSubQuery() {
730731
assertTrue(e.getMessage().contains("Fork requires at least two branches"));
731732
}
732733

734+
public void testProfile() {
735+
var query = """
736+
FROM test
737+
| FORK
738+
( WHERE content:"fox" | SORT id )
739+
( WHERE content:"dog" | SORT id )
740+
| SORT _fork, id
741+
| KEEP _fork, id, content
742+
""";
743+
744+
EsqlQueryRequest request = EsqlQueryRequest.syncEsqlQueryRequest();
745+
746+
request.pragmas(randomPragmas());
747+
request.query(query);
748+
request.profile(true);
749+
750+
try (var resp = run(request)) {
751+
EsqlQueryResponse.Profile profile = resp.profile();
752+
assertNotNull(profile);
753+
754+
List<String> descriptions = profile.drivers().stream().map(DriverProfile::description).sorted().toList();
755+
756+
assertEquals(
757+
List.of("data", "data", "main.final", "node_reduce", "node_reduce", "subplan-0.final", "subplan-1.final"),
758+
descriptions
759+
);
760+
}
761+
}
762+
733763
private void createAndPopulateIndex() {
734764
var indexName = "test";
735765
var client = client().admin().indices();

‎x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java‎

Lines changed: 30 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -195,7 +195,7 @@ public void execute(
195195

196196
// we have no sub plans, so we can just execute the given plan
197197
if (subplans == null || subplans.size() == 0) {
198-
executePlan(sessionId, rootTask, physicalPlan, configuration, foldContext, execInfo, listener, null);
198+
executePlan(sessionId, rootTask, physicalPlan, configuration, foldContext, execInfo, null, listener, null);
199199
return;
200200
}
201201

@@ -220,7 +220,7 @@ public void execute(
220220
var finalListener = ActionListener.runBefore(listener, () -> exchangeService.removeExchangeSourceHandler(sessionId));
221221
var computeContext = new ComputeContext(
222222
mainSessionId,
223-
"single",
223+
"main.final",
224224
LOCAL_CLUSTER,
225225
List.of(),
226226
configuration,
@@ -244,22 +244,33 @@ public void execute(
244244
) {
245245
runCompute(rootTask, computeContext, finalMainPlan, localListener.acquireCompute());
246246

247-
for (PhysicalPlan subplan : subplans) {
247+
for (int i = 0; i < subplans.size(); i++) {
248+
var subplan = subplans.get(i);
248249
var childSessionId = newChildSession(sessionId);
249250
ExchangeSinkHandler exchangeSink = exchangeService.createSinkHandler(childSessionId, queryPragmas.exchangeBufferSize());
250251
// funnel sub plan pages into the main plan exchange source
251252
mainExchangeSource.addRemoteSink(exchangeSink::fetchPageAsync, true, () -> {}, 1, ActionListener.noop());
252253
var subPlanListener = localListener.acquireCompute();
253254

254-
executePlan(childSessionId, rootTask, subplan, configuration, foldContext, execInfo, ActionListener.wrap(result -> {
255-
exchangeSink.addCompletionListener(
256-
ActionListener.running(() -> { exchangeService.finishSinkHandler(childSessionId, null); })
257-
);
258-
subPlanListener.onResponse(result.completionInfo());
259-
}, e -> {
260-
exchangeService.finishSinkHandler(childSessionId, e);
261-
subPlanListener.onFailure(e);
262-
}), () -> exchangeSink.createExchangeSink(() -> {}));
255+
executePlan(
256+
childSessionId,
257+
rootTask,
258+
subplan,
259+
configuration,
260+
foldContext,
261+
execInfo,
262+
"subplan-" + i,
263+
ActionListener.wrap(result -> {
264+
exchangeSink.addCompletionListener(
265+
ActionListener.running(() -> { exchangeService.finishSinkHandler(childSessionId, null); })
266+
);
267+
subPlanListener.onResponse(result.completionInfo());
268+
}, e -> {
269+
exchangeService.finishSinkHandler(childSessionId, e);
270+
subPlanListener.onFailure(e);
271+
}),
272+
() -> exchangeSink.createExchangeSink(() -> {})
273+
);
263274
}
264275
}
265276
}
@@ -272,6 +283,7 @@ public void executePlan(
272283
Configuration configuration,
273284
FoldContext foldContext,
274285
EsqlExecutionInfo execInfo,
286+
String profileQualifier,
275287
ActionListener<Result> listener,
276288
Supplier<ExchangeSink> exchangeSinkSupplier
277289
) {
@@ -309,7 +321,7 @@ public void executePlan(
309321
}
310322
var computeContext = new ComputeContext(
311323
newChildSession(sessionId),
312-
"single",
324+
profileDescription(profileQualifier, "single"),
313325
LOCAL_CLUSTER,
314326
List.of(),
315327
configuration,
@@ -395,7 +407,7 @@ public void executePlan(
395407
rootTask,
396408
new ComputeContext(
397409
sessionId,
398-
"final",
410+
profileDescription(profileQualifier, "final"),
399411
LOCAL_CLUSTER,
400412
List.of(),
401413
configuration,
@@ -611,6 +623,10 @@ String newChildSession(String session) {
611623
return session + "/" + childSessionIdGenerator.incrementAndGet();
612624
}
613625

626+
String profileDescription(String qualifier, String label) {
627+
return qualifier == null ? label : qualifier + "." + label;
628+
}
629+
614630
Runnable cancelQueryOnFailure(CancellableTask task) {
615631
return new RunOnce(() -> {
616632
LOGGER.debug("cancelling ESQL task {} on failure", task);

0 commit comments

Comments
 (0)