SteNicholas commented on code in PR #3206:
URL: https://github.com/apache/celeborn/pull/3206#discussion_r2035172793
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/DeviceMonitor.scala:
##########
@@ -141,19 +141,34 @@ class LocalDeviceMonitor(
device.notifyObserversOnNonCriticalError(mountPoints,
DiskStatus.IO_HANG)
} else {
device.diskInfos.values().asScala.foreach { diskInfo =>
- if (checkDiskUsage && DeviceMonitor.highDiskUsage(conf,
diskInfo)) {
- logError(s"${diskInfo.mountPoint} high_disk_usage error,
notify observers")
-
device.notifyObserversOnHighDiskUsage(diskInfo.mountPoint)
- } else if (checkReadWrite &&
- DeviceMonitor.readWriteError(conf, diskInfo.dirs.head)) {
- logError(s"${diskInfo.mountPoint} read-write error,
notify observers")
- // We think that if one dir in device has read-write
problem, if possible all
- // dirs in this device have the problem
- device.notifyObserversOnNonCriticalError(
- List(diskInfo.mountPoint),
- DiskStatus.READ_OR_WRITE_FAILURE)
- } else if (nonCriticalErrorSum <=
device.notifyErrorThreshold * 0.5) {
- device.notifyObserversOnHealthy(diskInfo.mountPoint)
+ try {
+ if (checkDiskUsage && DeviceMonitor.highDiskUsage(conf,
diskInfo)) {
+ logError(s"${diskInfo.mountPoint} high_disk_usage
error, notify observers")
+
device.notifyObserversOnHighDiskUsage(diskInfo.mountPoint)
+ } else if (checkReadWrite &&
+ DeviceMonitor.readWriteError(conf,
diskInfo.dirs.head)) {
+ logError(s"${diskInfo.mountPoint} read-write error,
notify observers")
+ // We think that if one dir in device has read-write
problem, if possible all
+ // dirs in this device have the problem
+ device.notifyObserversOnNonCriticalError(
+ List(diskInfo.mountPoint),
+ DiskStatus.READ_OR_WRITE_FAILURE)
+ } else if (nonCriticalErrorSum <=
device.notifyErrorThreshold * 0.5) {
+ device.notifyObserversOnHealthy(diskInfo.mountPoint)
+ }
+ } catch {
+ case e: ExecutionException =>
+ e.getCause match {
+ case fse: FileSystemException =>
+ logError(
+ s"${diskInfo.mountPoint} critical error, notify
observers",
+ fse)
+ device.notifyObserversOnError(
+ List(diskInfo.mountPoint),
+ DiskStatus.CRITICAL_ERROR)
+ case throwable: Throwable =>
+ throw throwable
Review Comment:
@RexXiong, the exception would log in `Device check failed`.
```
try {
observedDevices.values().asScala.foreach(device => {
val mountPoints = device.diskInfos.keySet.asScala.toList
// tolerate time accuracy for better performance
val now = System.currentTimeMillis()
for (concurrentList <-
device.nonCriticalErrors.values().asScala) {
for (time <- concurrentList.asScala) {
if (now - time > device.notifyErrorExpireTimeout) {
concurrentList.remove(time)
}
}
}
val nonCriticalErrorSum =
device.nonCriticalErrors.values().asScala.map(_.size).sum
if (nonCriticalErrorSum > device.notifyErrorThreshold) {
logError(s"Device ${device.deviceInfo.name} has accumulated
$nonCriticalErrorSum non-critical " +
s"error within the past
${Utils.msDurationToString(device.notifyErrorExpireTimeout)} , its sum has " +
s"exceed the threshold (${device.notifyErrorThreshold}),
device monitor will notify error to " +
s"observed device.")
val mountPoints =
device.diskInfos.values().asScala.map(_.mountPoint).toList
device.notifyObserversOnError(mountPoints,
DiskStatus.CRITICAL_ERROR)
} else {
if (checkIoHang && device.ioHang()) {
logError(s"Encounter device io hang error!" +
s"${device.deviceInfo.name}, notify observers")
device.notifyObserversOnNonCriticalError(mountPoints,
DiskStatus.IO_HANG)
} else {
device.diskInfos.values().asScala.foreach { diskInfo =>
try {
if (checkDiskUsage &&
DeviceMonitor.highDiskUsage(conf, diskInfo)) {
logError(s"${diskInfo.mountPoint} high_disk_usage
error, notify observers")
device.notifyObserversOnHighDiskUsage(diskInfo.mountPoint)
} else if (checkReadWrite &&
DeviceMonitor.readWriteError(conf,
diskInfo.dirs.head)) {
logError(s"${diskInfo.mountPoint} read-write error,
notify observers")
// We think that if one dir in device has read-write
problem, if possible all
// dirs in this device have the problem
device.notifyObserversOnNonCriticalError(
List(diskInfo.mountPoint),
DiskStatus.READ_OR_WRITE_FAILURE)
} else if (nonCriticalErrorSum <=
device.notifyErrorThreshold * 0.5) {
device.notifyObserversOnHealthy(diskInfo.mountPoint)
}
} catch {
case e: ExecutionException =>
e.getCause match {
case fse: FileSystemException =>
logError(
s"${diskInfo.mountPoint} critical error,
notify observers",
fse)
device.notifyObserversOnError(
List(diskInfo.mountPoint),
DiskStatus.CRITICAL_ERROR)
case throwable: Throwable =>
throw throwable
}
}
}
}
}
})
} catch {
case t: Throwable =>
logError("Device check failed.", t)
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
