Skip to content

Commit cc5aa91

Browse files
jfredenjfreden
authored
Ensure config reload on ..data symlink switch for CSI driver support (elastic#127628)
* Reprocess all changed files in settings dir if symlink dir is updated
1 parent 6bf5316 commit cc5aa91

File tree

6 files changed

+81
-20
lines changed

6 files changed

+81
-20
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/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);

0 commit comments

Comments
 (0)