Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
11 changes: 4 additions & 7 deletions hail/hail/package.mill
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ object `package` extends HailScalaModule:
build.hail.utils,
build.hail.memory(if buildMode == CI then "debug" else "release"),
)

override def resources: T[Seq[PathRef]] =
super.resources() ++ Seq(
BuildCtx.withFilesystemCheckerDisabled {
Expand All @@ -43,26 +43,23 @@ object `package` extends HailScalaModule:
|package object hail {
| // FIXME: This should be a compiler switch/flag
| val EmitTracebackInAssertions = ${buildMode != Release}
|
|
| val PipVersion = "${Settings.hailMajorMinorVersion}.${Settings.hailPatchVersion}"
| val Revision = "${VcsVersion.vcsState().currentRevision}"
| val SparkVersion = "${Env.sparkVersion()}"
|
|
| // FIXME: probably should use tags or something to choose English name
| val PrettyVersion = PipVersion + "-" + Revision.substring(0, 12)
|}""".stripMargin,
createFolders = true,
)

Seq(
PathRef(Task.dest),
build.hail.`ir-gen`.generate(),
)
}

override def unmanagedClasspath: T[Seq[PathRef]] =
Seq(build.hail.shadeazure.assembly())

// omit unmanagedClasspath from the jar
override def jar: T[PathRef] =
Task {
Expand Down
42 changes: 0 additions & 42 deletions hail/hail/shadeazure/package.mill

This file was deleted.

97 changes: 0 additions & 97 deletions hail/hail/src/is/hail/services/oauth2.scala
Original file line number Diff line number Diff line change
@@ -1,20 +1,9 @@
package is.hail.services

import is.hail.services.oauth2.AzureCloudCredentials.AzureTokenRefreshMinutes
import is.hail.services.oauth2.AzureCloudCredentials.EnvVars.AzureApplicationCredentials
import is.hail.shadedazure.com.azure.core.credential.{
AccessToken, TokenCredential, TokenRequestContext,
}
import is.hail.shadedazure.com.azure.identity.{
ClientSecretCredentialBuilder, DefaultAzureCredentialBuilder,
}
import is.hail.utils.{jsonToBytes, using}

import scala.jdk.CollectionConverters._

import java.io.{ByteArrayInputStream, Serializable}
import java.nio.file.{Files, Path}
import java.time.OffsetDateTime

import com.google.auth.oauth2.GoogleCredentials
import org.json4s.{DefaultFormats, Formats, JValue}
Expand Down Expand Up @@ -43,15 +32,13 @@ object oauth2 {
jvalue <- JsonMethods.parseOpt(identity)
} yield (jvalue \ "idp").extract[String] match {
case "Google" => GoogleCloudCredentials.fromJson(jvalue \ "credentials")
case "Microsoft" => AzureCloudCredentials.fromJson(jvalue \ "credentials")
case other => throw new IllegalArgumentException(s"Unknown identity provider: '$other'")
}

def CloudCredentials(keyPath: Option[Path], env: Map[String, String] = sys.env)
: CloudCredentials =
env.get("HAIL_CLOUD") match {
case None | Some("gcp") => GoogleCloudCredentials(keyPath)
case Some("azure") => AzureCloudCredentials(keyPath, env)
case Some(cloud) => throw new IllegalArgumentException(s"Unknown cloud: '$cloud'")
}

Expand Down Expand Up @@ -83,88 +70,4 @@ object oauth2 {
}
}
}

sealed trait AzureCloudCredentials extends CloudCredentials {

def value: TokenCredential
def scopes: Array[String]

@transient private[this] var token: AccessToken = _

override def accessToken: String = {
refreshIfRequired()
token.getToken
}

override def scoped(scopes: Array[String]): AzureCloudCredentials

private[this] def refreshIfRequired(): Unit =
if (!isExpired) token.getToken
else synchronized {
if (isExpired) {
token = value.getTokenSync(new TokenRequestContext().setScopes(scopes.toSeq.asJava))
}

token.getToken: Unit
}

private[this] def isExpired: Boolean =
token == null || OffsetDateTime.now.plusMinutes(AzureTokenRefreshMinutes).isAfter(
token.getExpiresAt
)
}

object AzureCloudCredentials {
object EnvVars {
val AzureApplicationCredentials = "AZURE_APPLICATION_CREDENTIALS"
}

val DefaultOAuth2Scopes: Array[String] =
Array(".default")

private[AzureCloudCredentials] val AzureTokenRefreshMinutes = 5L

def fromJson(jv: JValue, scopes: Array[String] = DefaultOAuth2Scopes): AzureCloudCredentials =
AzureClientSecretCredentials(
clientId = (jv \ "appId").extract[String],
tenantId = (jv \ "tenant").extract[String],
secret = (jv \ "password").extract[String],
scopes = scopes,
)

def apply(keyPath: Option[Path], env: Map[String, String] = sys.env): AzureCloudCredentials =
keyPath.orElse(env.get(AzureApplicationCredentials).map(Path.of(_))) match {
case Some(path) =>
using(Files.newInputStream(path)) { in =>
fromJson(JsonMethods.parse(in), DefaultOAuth2Scopes)
}
case None =>
AzureDefaultCredentials(DefaultOAuth2Scopes)
}
}

private case class AzureDefaultCredentials(scopes: Array[String]) extends AzureCloudCredentials {
@transient override lazy val value: TokenCredential =
new DefaultAzureCredentialBuilder().build()

override def scoped(scopes: Array[String]): AzureDefaultCredentials =
copy(scopes)
}

private case class AzureClientSecretCredentials(
clientId: String,
tenantId: String,
secret: String,
scopes: Array[String],
) extends AzureCloudCredentials {
@transient override lazy val value: TokenCredential =
new ClientSecretCredentialBuilder()
.clientId(clientId)
.clientSecret(secret)
.tenantId(tenantId)
.build()

override def scoped(scopes: Array[String]): AzureClientSecretCredentials =
copy(scopes = scopes)
}
}
32 changes: 2 additions & 30 deletions hail/hail/src/is/hail/services/package.scala
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
package is.hail

import is.hail.services.requests.ClientResponseException
import is.hail.shadedazure.com.azure.storage.common.implementation.Constants
import is.hail.utils._

import scala.annotation.tailrec
Expand Down Expand Up @@ -47,15 +46,8 @@ package object services extends Logging {
}

@tailrec
def isLimitedRetriesError(_e: Throwable): Boolean = {
// An exception is a "retry once error" if a rare, known bug in a dependency or in a cloud
// provider can manifest as this exception *and* that manifestation is indistinguishable from a
// true error.
val e = is.hail.shadedazure.reactor.core.Exceptions.unwrap(_e)
def isLimitedRetriesError(e: Throwable): Boolean = {
e match {
case e: RuntimeException
if e.getMessage != null && e.getMessage == Constants.STREAM_CLOSED =>
true
case e: SocketException
if e.getMessage != null && (e.getMessage.contains(
"Connection reset"
Expand Down Expand Up @@ -87,14 +79,7 @@ package object services extends Logging {
}

@tailrec
def isTransientError(_e: Throwable): Boolean = {
// ReactiveException is package private inside reactore.core.Exception so we cannot access
// it directly for an isInstance check. AFAICT, this is the only way to check if we received
// a ReactiveException.
//
// If the argument is a ReactiveException, it returns its cause. If the argument is not a
// ReactiveException it returns the exception unmodified.
val e = is.hail.shadedazure.reactor.core.Exceptions.unwrap(_e)
def isTransientError(e: Throwable): Boolean = {
e match {
case _: java.nio.channels.ClosedByInterruptException =>
true
Expand Down Expand Up @@ -153,19 +138,6 @@ package object services extends Logging {
case e: EOFException
if e.getMessage != null && e.getMessage.contains("SSL peer shut down incorrectly") =>
true
case e: IllegalStateException
if e.getMessage.contains("Timeout on blocking read") ||
e.getMessage.contains("Faulted stream due to underlying sink write failure") =>
/* Caused by: java.lang.IllegalStateException: Timeout on blocking read for 30000000000
* NANOSECONDS */
/* reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:123) */
// reactor.core.publisher.Mono.block(Mono.java:1727)
/* com.azure.storage.common.implementation.StorageImplUtils.blockWithOptionalTimeout(StorageImplUtils.java:130) */
/* com.azure.storage.blob.specialized.BlobClientBase.downloadStreamWithResponse(BlobClientBase.java:731) */
// is.hail.io.fs.AzureStorageFS$$anon$1.fill(AzureStorageFS.scala:152)
// is.hail.io.fs.FSSeekableInputStream.read(FS.scala:141)
// ...
true
case e: java.net.SocketTimeoutException
if e.getMessage != null && e.getMessage.contains("connect timed out") =>
true
Expand Down
1 change: 0 additions & 1 deletion hail/scripts/test-gcp.sh
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,6 @@ spark-submit \
./hail-test.jar \
execute \
--select-package=is.hail \
--exclude-package=is.hail.shadedazure \
--exclude-package=is.hail.io.fs \
--exclude-package=is.hail.services \
--fail-if-no-tests \
Expand Down
Loading