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.

Sign up for GitHub

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

[Backport] Introduce FS Health HEALTHY threshold to fail stuck node #1269

Merged
merged 7 commits into from
Feb 7, 2022
Merged

[Backport] Introduce FS Health HEALTHY threshold to fail stuck node #1269

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
8bd1907
Introduce FS Health HEALTHY threshold to fail stuck node (#1167)
Bukhtawar Sep 17, 2021
3cc45b7
Merge remote-tracking branch 'upstream/1.x' into 1.x-fs
Bukhtawar Sep 29, 2021
84f4f95
Merge branch '1.x' into 1.x-fs
Bukhtawar Feb 7, 2022
a957b31
Fix up settings
Bukhtawar Feb 7, 2022
3c93bbb
Fix up tests
Bukhtawar Feb 7, 2022
1702995
Fix up tests
Bukhtawar Feb 7, 2022
c087a3c
Fix up tests
Bukhtawar Feb 7, 2022
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: 2 additions & 1 deletion server/src/main/java/org/opensearch/common/settings/ClusterSettings.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -601,7 +601,8 @@ public void apply(Settings value, Settings current, Settings previous) {
ShardIndexingPressureMemoryManager.MAX_OUTSTANDING_REQUESTS,
NodeLoadAwareAllocationDecider.CLUSTER_ROUTING_ALLOCATION_LOAD_AWARENESS_PROVISIONED_CAPACITY_SETTING,
NodeLoadAwareAllocationDecider.CLUSTER_ROUTING_ALLOCATION_LOAD_AWARENESS_SKEW_FACTOR_SETTING,
NodeLoadAwareAllocationDecider.CLUSTER_ROUTING_ALLOCATION_LOAD_AWARENESS_ALLOW_UNASSIGNED_PRIMARIES_SETTING
NodeLoadAwareAllocationDecider.CLUSTER_ROUTING_ALLOCATION_LOAD_AWARENESS_ALLOW_UNASSIGNED_PRIMARIES_SETTING,
FsHealthService.HEALTHY_TIMEOUT_SETTING
)
)
);
Expand Down
66 changes: 56 additions & 10 deletions server/src/main/java/org/opensearch/monitor/fs/FsHealthService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,8 @@
import java.nio.file.StandardOpenOption;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.LongSupplier;
import java.util.stream.Collectors;

Expand All @@ -78,6 +80,9 @@ public class FsHealthService extends AbstractLifecycleComponent implements NodeH
private final NodeEnvironment nodeEnv;
private final LongSupplier currentTimeMillisSupplier;
private volatile Scheduler.Cancellable scheduledFuture;
private volatile TimeValue healthyTimeoutThreshold;
private final AtomicLong lastRunStartTimeMillis = new AtomicLong(Long.MIN_VALUE);
private final AtomicBoolean checkInProgress = new AtomicBoolean();

@Nullable
private volatile Set<Path> unhealthyPaths;
Expand All @@ -90,7 +95,7 @@ public class FsHealthService extends AbstractLifecycleComponent implements NodeH
);
public static final Setting<TimeValue> REFRESH_INTERVAL_SETTING = Setting.timeSetting(
"monitor.fs.health.refresh_interval",
TimeValue.timeValueSeconds(120),
TimeValue.timeValueSeconds(60),
TimeValue.timeValueMillis(1),
Setting.Property.NodeScope
);
Expand All @@ -101,15 +106,24 @@ public class FsHealthService extends AbstractLifecycleComponent implements NodeH
Setting.Property.NodeScope,
Setting.Property.Dynamic
);
public static final Setting<TimeValue> HEALTHY_TIMEOUT_SETTING = Setting.timeSetting(
"monitor.fs.health.healthy_timeout_threshold",
TimeValue.timeValueSeconds(60),
TimeValue.timeValueMillis(1),
Setting.Property.NodeScope,
Setting.Property.Dynamic
);

public FsHealthService(Settings settings, ClusterSettings clusterSettings, ThreadPool threadPool, NodeEnvironment nodeEnv) {
this.threadPool = threadPool;
this.enabled = ENABLED_SETTING.get(settings);
this.refreshInterval = REFRESH_INTERVAL_SETTING.get(settings);
this.slowPathLoggingThreshold = SLOW_PATH_LOGGING_THRESHOLD_SETTING.get(settings);
this.currentTimeMillisSupplier = threadPool::relativeTimeInMillis;
this.healthyTimeoutThreshold = HEALTHY_TIMEOUT_SETTING.get(settings);
this.nodeEnv = nodeEnv;
clusterSettings.addSettingsUpdateConsumer(SLOW_PATH_LOGGING_THRESHOLD_SETTING, this::setSlowPathLoggingThreshold);
clusterSettings.addSettingsUpdateConsumer(HEALTHY_TIMEOUT_SETTING, this::setHealthyTimeoutThreshold);
clusterSettings.addSettingsUpdateConsumer(ENABLED_SETTING, this::setEnabled);
}

Expand All @@ -134,6 +148,10 @@ public void setSlowPathLoggingThreshold(TimeValue slowPathLoggingThreshold) {
this.slowPathLoggingThreshold = slowPathLoggingThreshold;
}

public void setHealthyTimeoutThreshold(TimeValue healthyTimeoutThreshold) {
this.healthyTimeoutThreshold = healthyTimeoutThreshold;
}

@Override
public StatusInfo getHealth() {
StatusInfo statusInfo;
Expand All @@ -142,14 +160,17 @@ public StatusInfo getHealth() {
statusInfo = new StatusInfo(HEALTHY, "health check disabled");
} else if (brokenLock) {
statusInfo = new StatusInfo(UNHEALTHY, "health check failed due to broken node lock");
} else if (unhealthyPaths == null) {
statusInfo = new StatusInfo(HEALTHY, "health check passed");
} else {
String info = "health check failed on ["
+ unhealthyPaths.stream().map(k -> k.toString()).collect(Collectors.joining(","))
+ "]";
statusInfo = new StatusInfo(UNHEALTHY, info);
}
} else if (checkInProgress.get()
&& currentTimeMillisSupplier.getAsLong() - lastRunStartTimeMillis.get() > healthyTimeoutThreshold.millis()) {
statusInfo = new StatusInfo(UNHEALTHY, "healthy threshold breached");
} else if (unhealthyPaths == null) {
statusInfo = new StatusInfo(HEALTHY, "health check passed");
} else {
String info = "health check failed on ["
+ unhealthyPaths.stream().map(k -> k.toString()).collect(Collectors.joining(","))
+ "]";
statusInfo = new StatusInfo(UNHEALTHY, info);
}
return statusInfo;
}

Expand All @@ -164,13 +185,22 @@ class FsHealthMonitor implements Runnable {

@Override
public void run() {
boolean checkEnabled = enabled;
try {
if (enabled) {
if (checkEnabled) {
setLastRunStartTimeMillis();
boolean started = checkInProgress.compareAndSet(false, true);
assert started;
monitorFSHealth();
logger.debug("health check succeeded");
}
} catch (Exception e) {
logger.error("health check failed", e);
} finally {
if (checkEnabled) {
boolean completed = checkInProgress.compareAndSet(true, false);
assert completed;
}
}
}

Expand Down Expand Up @@ -205,6 +235,18 @@ private void monitorFSHealth() {
slowPathLoggingThreshold
);
}
if (elapsedTime > healthyTimeoutThreshold.millis()) {
logger.error(
"health check of [{}] failed, took [{}ms] which is above the healthy threshold of [{}]",
path,
elapsedTime,
healthyTimeoutThreshold
);
if (currentUnhealthyPaths == null) {
currentUnhealthyPaths = new HashSet<>(1);
}
currentUnhealthyPaths.add(path);
}
}
} catch (Exception ex) {
logger.error(new ParameterizedMessage("health check of [{}] failed", path), ex);
Expand All @@ -218,4 +260,8 @@ private void monitorFSHealth() {
brokenLock = false;
}
}

private void setLastRunStartTimeMillis() {
lastRunStartTimeMillis.getAndUpdate(l -> Math.max(l, currentTimeMillisSupplier.getAsLong()));
}
}
83 changes: 78 additions & 5 deletions server/src/test/java/org/opensearch/monitor/fs/FsHealthServiceTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
import org.opensearch.common.settings.ClusterSettings;
import org.opensearch.common.settings.Settings;
import org.opensearch.env.NodeEnvironment;
import org.opensearch.monitor.StatusInfo;
import org.opensearch.test.OpenSearchTestCase;
import org.opensearch.test.MockLogAppender;
import org.opensearch.test.junit.annotations.TestLogging;
Expand All @@ -61,6 +62,7 @@
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import static org.hamcrest.Matchers.equalTo;
import static org.opensearch.monitor.StatusInfo.Status.HEALTHY;
import static org.opensearch.monitor.StatusInfo.Status.UNHEALTHY;
import static org.opensearch.node.Node.NODE_NAME_SETTING;
Expand Down Expand Up @@ -172,7 +174,7 @@ public void testLoggingOnHungIO() throws Exception {
}

// disrupt file system
disruptFileSystemProvider.injectIOException.set(true);
disruptFileSystemProvider.injectIODelay.set(true);
fsHealthService.new FsHealthMonitor().run();
assertEquals(env.nodeDataPaths().length, disruptFileSystemProvider.getInjectedPathCount());
assertBusy(mockAppender::assertAllExpectationsMatched);
Expand All @@ -182,6 +184,66 @@ public void testLoggingOnHungIO() throws Exception {
}
}

public void testFailsHealthOnHungIOBeyondHealthyTimeout() throws Exception {
long healthyTimeoutThreshold = randomLongBetween(500, 1000);
long refreshInterval = randomLongBetween(500, 1000);
long slowLogThreshold = randomLongBetween(100, 200);
long delayBetweenChecks = 100;
final Settings settings = Settings.builder()
.put(FsHealthService.HEALTHY_TIMEOUT_SETTING.getKey(), healthyTimeoutThreshold + "ms")
.put(FsHealthService.REFRESH_INTERVAL_SETTING.getKey(), refreshInterval + "ms")
.put(FsHealthService.SLOW_PATH_LOGGING_THRESHOLD_SETTING.getKey(), slowLogThreshold + "ms")
.put(ThreadPool.ESTIMATED_TIME_INTERVAL_SETTING.getKey(), 0)// we need to verify exact time
.build();
FileSystem fileSystem = PathUtils.getDefaultFileSystem();
TestThreadPool testThreadPool = new TestThreadPool(getClass().getName(), settings);
FileSystemFsyncHungProvider disruptFileSystemProvider = new FileSystemFsyncHungProvider(fileSystem, testThreadPool);
fileSystem = disruptFileSystemProvider.getFileSystem(null);
PathUtilsForTesting.installMock(fileSystem);
final ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
try (NodeEnvironment env = newNodeEnvironment()) {
FsHealthService fsHealthService = new FsHealthService(settings, clusterSettings, testThreadPool, env);
logger.info("--> Initial health status prior to the first monitor run");
StatusInfo fsHealth = fsHealthService.getHealth();
assertEquals(HEALTHY, fsHealth.getStatus());
assertEquals("health check passed", fsHealth.getInfo());
logger.info("--> First monitor run");
fsHealthService.new FsHealthMonitor().run();
fsHealth = fsHealthService.getHealth();
assertEquals(HEALTHY, fsHealth.getStatus());
assertEquals("health check passed", fsHealth.getInfo());
logger.info("--> Disrupt file system");
disruptFileSystemProvider.injectIODelay.set(true);
final FsHealthService fsHealthSrvc = new FsHealthService(settings, clusterSettings, testThreadPool, env);
fsHealthSrvc.doStart();
waitUntil(
() -> fsHealthSrvc.getHealth().getStatus() == UNHEALTHY,
healthyTimeoutThreshold + (2 * refreshInterval),
TimeUnit.MILLISECONDS
);
fsHealth = fsHealthSrvc.getHealth();
assertEquals(UNHEALTHY, fsHealth.getStatus());
assertEquals("healthy threshold breached", fsHealth.getInfo());
int disruptedPathCount = disruptFileSystemProvider.getInjectedPathCount();
assertThat(disruptedPathCount, equalTo(1));
logger.info("--> Fix file system disruption");
disruptFileSystemProvider.injectIODelay.set(false);
waitUntil(
() -> fsHealthSrvc.getHealth().getStatus() == HEALTHY,
delayBetweenChecks + (4 * refreshInterval),
TimeUnit.MILLISECONDS
);
fsHealth = fsHealthSrvc.getHealth();
assertEquals(HEALTHY, fsHealth.getStatus());
assertEquals("health check passed", fsHealth.getInfo());
assertEquals(disruptedPathCount, disruptFileSystemProvider.getInjectedPathCount());
fsHealthSrvc.doStop();
} finally {
PathUtilsForTesting.teardown();
ThreadPool.terminate(testThreadPool, 500, TimeUnit.MILLISECONDS);
}
}

public void testFailsHealthOnSinglePathFsyncFailure() throws IOException {
FileSystem fileSystem = PathUtils.getDefaultFileSystem();
FileSystemFsyncIOExceptionProvider disruptFsyncFileSystemProvider = new FileSystemFsyncIOExceptionProvider(fileSystem);
Expand Down Expand Up @@ -349,18 +411,25 @@ public void force(boolean metaData) throws IOException {

private static class FileSystemFsyncHungProvider extends FilterFileSystemProvider {

AtomicBoolean injectIOException = new AtomicBoolean();
AtomicBoolean injectIODelay = new AtomicBoolean();
AtomicInteger injectedPaths = new AtomicInteger();

private final long delay;
private final ThreadPool threadPool;
private static final long AWAIT_BUSY_THRESHOLD = 100L;

FileSystemFsyncHungProvider(FileSystem inner, long delay, ThreadPool threadPool) {
super("disrupt_fs_health://", inner);
this.delay = delay;
this.threadPool = threadPool;
}

FileSystemFsyncHungProvider(FileSystem inner, ThreadPool threadPool) {
super("disrupt_fs_health://", inner);
this.threadPool = threadPool;
this.delay = Long.MAX_VALUE;
}

public int getInjectedPathCount() {
return injectedPaths.get();
}
Expand All @@ -370,17 +439,21 @@ public FileChannel newFileChannel(Path path, Set<? extends OpenOption> options,
return new FilterFileChannel(super.newFileChannel(path, options, attrs)) {
@Override
public void force(boolean metaData) throws IOException {
if (injectIOException.get()) {
if (injectIODelay.get()) {
if (path.getFileName().toString().equals(FsHealthService.FsHealthMonitor.TEMP_FILE_NAME)) {
injectedPaths.incrementAndGet();
final long startTimeMillis = threadPool.relativeTimeInMillis();
long timeInMillis = 1;
long maxWaitTimeMillis = startTimeMillis + delay >= 0 ? startTimeMillis + delay : Long.MAX_VALUE;// long
// overflow
do {
try {
Thread.sleep(delay);
Thread.sleep(timeInMillis);
} catch (InterruptedException e) {
throw new AssertionError(e);
}
} while (threadPool.relativeTimeInMillis() <= startTimeMillis + delay);
timeInMillis = Math.min(AWAIT_BUSY_THRESHOLD, timeInMillis * 2);
} while (threadPool.relativeTimeInMillis() <= maxWaitTimeMillis && injectIODelay.get());
}
}
super.force(metaData);
Expand Down