WX-1566 Fix Morgan's call cache file hash CPU thrash Cromwell crash #7419
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| package cromwell.engine.io.nio | ||
|
|
||
| import cats.effect.IO | ||
| import cloud.nio.spi.{FileHash, HashType} | ||
| import common.util.StringUtil.EnhancedString | ||
| import cromwell.core.path.Path | ||
| import cromwell.filesystems.blob.BlobPath | ||
| import cromwell.filesystems.drs.DrsPath | ||
| import cromwell.filesystems.gcs.GcsPath | ||
| import cromwell.filesystems.http.HttpPath | ||
| import cromwell.filesystems.s3.S3Path | ||
| import cromwell.util.TryWithResource.tryWithResource | ||
|
|
||
| import scala.util.Try | ||
|
|
||
| object NioHashing { | ||
|
|
||
| def hash(file: Path): IO[String] = | ||
| // If there is no hash accessible from the file storage system, | ||
| // we'll read the file and generate the hash ourselves if we can. | ||
| getStoredHash(file) | ||
| .flatMap { | ||
| case Some(storedHash) => IO.pure(storedHash) | ||
| case None => | ||
| if (canHashLocally(file)) | ||
| generateMd5FileHashForPath(file) | ||
| else | ||
| IO.raiseError( | ||
| new Exception( | ||
| s"File of type ${file.getClass.getSimpleName} requires hash in object metadata, not present for ${file.pathAsString.maskSensitiveUri}" | ||
| ) | ||
| ) | ||
| } | ||
| .map(_.hash) | ||
|
|
||
| def getStoredHash(file: Path): IO[Option[FileHash]] = | ||
| file match { | ||
| case gcsPath: GcsPath => getFileHashForGcsPath(gcsPath).map(Option(_)) | ||
| case blobPath: BlobPath => getFileHashForBlobPath(blobPath) | ||
| case drsPath: DrsPath => | ||
| IO { | ||
| // We assume all DRS files have a stored hash; this will throw | ||
| // if the file does not. | ||
| drsPath.getFileHash | ||
| }.map(Option(_)) | ||
| case s3Path: S3Path => | ||
| IO { | ||
| Option(FileHash(HashType.S3Etag, s3Path.eTag)) | ||
| } | ||
| case _ => IO.pure(None) | ||
| } | ||
|
|
||
| /** | ||
| * In some scenarios like SFS it is appropriate for Cromwell to hash files using its own CPU power. | ||
| * | ||
| * In cloud scenarios, we don't want this because the files are huge, downloading them is slow & expensive, | ||
| * and the extreme CPU usage destabilizes the instance. [WX-1566] | ||
| * | ||
| * Cromwell is fundamentally supposed to be a job scheduler, and heavy computation should take place elsewhere. | ||
| * | ||
| * @param file The path to consider for local hashing | ||
| */ | ||
| private def canHashLocally(file: Path) = | ||
| file match { | ||
| case _: HttpPath => false | ||
| case _: BlobPath => false | ||
| case _ => true | ||
| } | ||
|
|
||
| private def generateMd5FileHashForPath(path: Path): IO[FileHash] = delayedIoFromTry { | ||
| tryWithResource(() => path.newInputStream) { inputStream => | ||
| FileHash(HashType.Md5, org.apache.commons.codec.digest.DigestUtils.md5Hex(inputStream)) | ||
| } | ||
| } | ||
|
|
||
| private def getFileHashForGcsPath(gcsPath: GcsPath): IO[FileHash] = delayedIoFromTry { | ||
| gcsPath.objectBlobId.map(id => FileHash(HashType.GcsCrc32c, gcsPath.cloudStorage.get(id).getCrc32c)) | ||
| } | ||
|
|
||
| private def getFileHashForBlobPath(blobPath: BlobPath): IO[Option[FileHash]] = delayedIoFromTry { | ||
| blobPath.md5HexString.map(md5 => md5.map(FileHash(HashType.Md5, _))) | ||
| } | ||
|
|
||
| /** | ||
| * Lazy evaluation of a Try in a delayed IO. This avoids accidentally eagerly evaluating the Try. | ||
| * | ||
| * IMPORTANT: Use this instead of IO.fromTry to make sure the Try will be reevaluated if the | ||
| * IoCommand is retried. | ||
| */ | ||
| private def delayedIoFromTry[A](t: => Try[A]): IO[A] = IO[A](t.get) | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The
NioFlowclass is supposed to handle all 10 file methods but it was getting to be mostly hashing, hence the split-out.It also wasn't really clear what hashing methods you were actually supposed to call, and which ones existed solely for use by other hashing-related methods. Now the interface is clear, you can either
hashorgetStoredHash.