CORE-308 Use okhttp rite

src/main/scala/org/broadinstitute/dsde/firecloud/Boot.scala

@@ -128,7 +128,8 @@ object Boot extends IOApp with LazyLogging {
             userServiceConstructor,
             shareLogServiceConstructor,
             managedGroupServiceConstructor,
-            oauth2Config
+            oauth2Config,
+            app.samDAO
           )
         }
       } yield service

src/main/scala/org/broadinstitute/dsde/firecloud/FireCloudApiService.scala

@@ -14,6 +14,7 @@ import akka.http.scaladsl.server.directives.{DebuggingDirectives, LogEntry, Logg
 import akka.http.scaladsl.server.{Directive, Directive0, ExceptionHandler, RouteResult}
 import akka.stream.Materializer
 import com.typesafe.scalalogging.LazyLogging
+import org.broadinstitute.dsde.firecloud.dataaccess.SamDAO
 import org.broadinstitute.dsde.firecloud.model.{ModelSchema, UserInfo, WithAccessToken}
 import org.broadinstitute.dsde.firecloud.service._
 import org.broadinstitute.dsde.firecloud.utils.StandardUserInfoDirectives
@@ -234,7 +235,8 @@ class FireCloudApiServiceImpl(
   val userServiceConstructor: (UserInfo) => UserService,
   val shareLogServiceConstructor: () => ShareLogService,
   val managedGroupServiceConstructor: (WithAccessToken) => ManagedGroupService,
-  val oidcConfig: OpenIDConnectConfiguration
+  val oidcConfig: OpenIDConnectConfiguration,
+  val samDao: SamDAO
 )(implicit
   val actorRefFactory: ActorRefFactory,
   val executionContext: ExecutionContext,

src/main/scala/org/broadinstitute/dsde/firecloud/dataaccess/HttpSamDAO.scala

@@ -7,6 +7,7 @@ import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
 import akka.http.scaladsl.model.{HttpRequest, StatusCode, StatusCodes}
 import akka.http.scaladsl.unmarshalling.Unmarshal
 import akka.stream.Materializer
+import okhttp3.Dispatcher
 import org.broadinstitute.dsde.firecloud.{FireCloudConfig, FireCloudExceptionWithErrorReport}
 import org.broadinstitute.dsde.firecloud.model.ErrorReportExtensions.FCErrorReport
 import org.broadinstitute.dsde.firecloud.model.ManagedGroupRoles.ManagedGroupRole
@@ -27,11 +28,11 @@ import org.broadinstitute.dsde.firecloud.model.{
   WorkbenchUserInfo
 }
 import org.broadinstitute.dsde.firecloud.utils.RestJsonClient
-import org.broadinstitute.dsde.rawls.{RawlsException, RawlsExceptionWithErrorReport}
+import org.broadinstitute.dsde.rawls.RawlsException
 import org.broadinstitute.dsde.rawls.model.{ErrorReport, RawlsUserEmail, WorkspaceJsonSupport}
 import org.broadinstitute.dsde.workbench.client.sam.{ApiCallback, ApiClient, ApiException}
 import org.broadinstitute.dsde.workbench.client.sam.api.{ResourcesApi, UsersApi}
-import org.broadinstitute.dsde.workbench.client.sam.model.BulkMembershipUpdateRequestV2
+import org.broadinstitute.dsde.workbench.client.sam.model.{BulkMembershipUpdateRequestV2, UserStatusInfo}
 import org.broadinstitute.dsde.workbench.model.WorkbenchIdentityJsonSupport._
 import org.broadinstitute.dsde.workbench.model.google.GoogleProject
 import org.broadinstitute.dsde.workbench.model.{WorkbenchEmail, WorkbenchGroupName, WorkbenchUserId}
@@ -57,6 +58,10 @@ class HttpSamDAO(implicit
     with SprayJsonSupport {
 
   val timeout: FiniteDuration = 1.minute
+  private val dispatcher = new Dispatcher()
+  dispatcher.setMaxRequests(1000)
+  dispatcher.setMaxRequestsPerHost(100)
+  private val httpClient = new ApiClient().getHttpClient.newBuilder().dispatcher(dispatcher).build()
 
   override def listWorkspaceResources(implicit userInfo: WithAccessToken): Future[Seq[UserPolicy]] =
     authedRequestToObject[Seq[UserPolicy]](Get(samListResources("workspace")),
@@ -214,8 +219,16 @@ class HttpSamDAO(implicit
     callback.future.map(_ => ())
   }
 
+  override def getUserStatus(user: WithAccessToken): Future[UserStatusInfo] = {
+    val apiClient = newApiClient(user)
+    val sam = new UsersApi(apiClient)
+    val callback = new SamApiCallback[UserStatusInfo]("getUserStatusInfo")
+    sam.getUserStatusInfoAsync(callback)
+    callback.future
+  }
+
   private def newApiClient(user: WithAccessToken) = {
-    val apiClient = new ApiClient()
+    val apiClient = new ApiClient(httpClient)
     apiClient.setAccessToken(user.accessToken.token)
     apiClient.setBasePath(FireCloudConfig.Sam.baseUrl)
     apiClient

src/main/scala/org/broadinstitute/dsde/firecloud/dataaccess/SamDAO.scala

@@ -18,7 +18,7 @@ import org.broadinstitute.dsde.firecloud.model.{
   WorkbenchUserInfo
 }
 import org.broadinstitute.dsde.rawls.model.{ErrorReportSource, RawlsUserEmail}
-import org.broadinstitute.dsde.workbench.client.sam.model.BulkMembershipUpdateRequestV2
+import org.broadinstitute.dsde.workbench.client.sam.model.{BulkMembershipUpdateRequestV2, UserStatusInfo}
 import org.broadinstitute.dsde.workbench.model.google.GoogleProject
 import org.broadinstitute.dsde.workbench.model.{WorkbenchEmail, WorkbenchGroupName, WorkbenchUserId}
 import org.broadinstitute.dsde.workbench.util.health.Subsystems
@@ -146,4 +146,6 @@ trait SamDAO extends LazyLogging with ReportsSubsystemStatus {
   val serviceName = SamDAO.serviceName
 
   def bulkUpdateGroups(request: List[BulkMembershipUpdateRequestV2], user: WithAccessToken): Future[Unit]
+
+  def getUserStatus(user: WithAccessToken): Future[UserStatusInfo]
 }

src/main/scala/org/broadinstitute/dsde/firecloud/utils/EnabledUserDirectives.scala

@@ -7,7 +7,8 @@ import akka.http.scaladsl.server.{RequestContext, Route, RouteResult}
 import akka.http.scaladsl.unmarshalling.Unmarshal
 import akka.stream.Materializer
 import com.typesafe.scalalogging.LazyLogging
-import org.broadinstitute.dsde.firecloud.model.UserInfo
+import org.broadinstitute.dsde.firecloud.dataaccess.SamDAO
+import org.broadinstitute.dsde.firecloud.model.{UserInfo, WithAccessToken}
 import org.broadinstitute.dsde.firecloud.{FireCloudConfig, FireCloudExceptionWithErrorReport}
 import org.broadinstitute.dsde.rawls.model.{ErrorReport, ErrorReportSource}
 import org.broadinstitute.dsde.workbench.client.sam.api.UsersApi
@@ -25,6 +26,8 @@ trait EnabledUserDirectives extends LazyLogging with SprayJsonSupport with Statu
   implicit val executionContext: ExecutionContext
   implicit val materializer: Materializer
 
+  val samDao: SamDAO
+
   /**
     * Queries Sam to see if the current user is enabled. If the user is disabled,
     * responds with Forbidden and prevents the rest of the route from executing.
@@ -33,18 +36,21 @@ trait EnabledUserDirectives extends LazyLogging with SprayJsonSupport with Statu
     * @param samBaseUrl where to find Sam - used for unit testing
     * @return n/a
     */
-  def requireEnabledUser(userInfo: UserInfo, samBaseUrl: String = FireCloudConfig.Sam.baseUrl)(
+  def requireEnabledUser(userInfo: UserInfo)(
     innerRoute: RequestContext => Future[RouteResult]
   ): Route =
     extractUri { uri =>
-      onComplete(getUserEnabled(userInfo.accessToken.token, samBaseUrl)) {
+      onComplete(getUserEnabled(userInfo)) {
         case Success(true) =>
           logger.debug(s"User ${userInfo.userEmail} is enabled: $uri")
           innerRoute
         case Success(false) =>
           logger.warn(s"User ${userInfo.userEmail} is disabled: $uri")
           // the 401/"User is disabled." response mirrors what Sam returns in this case.
           throwErrorReport(StatusCodes.Unauthorized, "User is disabled.")
+        case Failure(fcerr: FireCloudExceptionWithErrorReport)
+            if fcerr.errorReport.statusCode.contains(StatusCodes.NotFound) =>
+          throwErrorReport(StatusCodes.Unauthorized, "User is not registered.")
         case Failure(fcerr: FireCloudExceptionWithErrorReport) =>
           logger.error(
             s"FireCloudExceptionWithErrorReport exception checking enabled status for user ${userInfo.userEmail}: (${fcerr.getMessage}) while calling $uri",
@@ -59,14 +65,10 @@ trait EnabledUserDirectives extends LazyLogging with SprayJsonSupport with Statu
             apiex
           )
           val code = statusCodeFrom(apiex.getCode, Option(StatusCodes.InternalServerError))
-          if (code == StatusCodes.NotFound) {
-            throwErrorReport(StatusCodes.Unauthorized, "User is not registered.")
-          } else {
-            val message =
-              if (Option(apiex.getMessage).isEmpty || apiex.getMessage.isEmpty) code.defaultMessage()
-              else apiex.getMessage
-            throwErrorReport(code, message)
-          }
+          val message =
+            if (Option(apiex.getMessage).isEmpty || apiex.getMessage.isEmpty) code.defaultMessage()
+            else apiex.getMessage
+          throwErrorReport(code, message)
         case Failure(ex) =>
           logger.error(
             s"Unexpected exception checking enabled status for user ${userInfo.userEmail}: (${ex.getMessage}) while calling $uri",
@@ -76,64 +78,8 @@ trait EnabledUserDirectives extends LazyLogging with SprayJsonSupport with Statu
       }
     }
 
-  private class SamApiCallback[T](functionName: String = "userStatusInfo") extends ApiCallback[T] {
-    private val promise = Promise[T]()
-
-    override def onFailure(e: ApiException,
-                           statusCode: Int,
-                           responseHeaders: java.util.Map[String, java.util.List[String]]
-    ): Unit = {
-      val response = e.getResponseBody
-      // attempt to propagate an ErrorReport from Sam. If we can't understand Sam's response as an ErrorReport,
-      // create our own error message.
-      import org.broadinstitute.dsde.rawls.model.WorkspaceJsonSupport.ErrorReportFormat
-      toFutureTry(Unmarshal(response).to[ErrorReport]) flatMap {
-        case Success(err) =>
-          logger.error(s"Sam call to $functionName failed with error $err")
-          throw new FireCloudExceptionWithErrorReport(err)
-        case Failure(_) =>
-          // attempt to extract something useful from the response entity, even though it's not an ErrorReport
-          toFutureTry(Unmarshal(response).to[String]) map { maybeString =>
-            val stringErrMsg = maybeString match {
-              case Success(stringErr) => stringErr
-              case Failure(_)         => response
-            }
-            throw new FireCloudExceptionWithErrorReport(
-              ErrorReport(statusCodeFrom(statusCode, Option(StatusCodes.InternalServerError)),
-                          s"Sam call to $functionName failed with error '$stringErrMsg'"
-              )
-            )
-          }
-      }
-      promise.failure(e)
-    }
-    override def onSuccess(result: T,
-                           statusCode: Int,
-                           responseHeaders: java.util.Map[String, java.util.List[String]]
-    ): Unit = promise.success(result)
-
-    override def onUploadProgress(bytesWritten: Long, contentLength: Long, done: Boolean): Unit = ()
-
-    override def onDownloadProgress(bytesRead: Long, contentLength: Long, done: Boolean): Unit = ()
-
-    def future: Future[T] = promise.future
-  }
-
-  private def getUserEnabled(token: String, samBaseUrl: String): Future[Boolean] = {
-    val apiClient = new ApiClient()
-    apiClient.setAccessToken(token)
-    apiClient.setBasePath(samBaseUrl)
-    val sam = new UsersApi(apiClient)
-    val callback = new SamApiCallback[UserStatusInfo]
-    // TODO: should we use retries here?
-    /* TODO: instead of directly calling Sam's userStatusInfo API, should we call a "real" api and, if that api returns
-        401, echo its response? That would allow Orch to always give the same error response that Sam gives, but
-        it is likely to be more expensive in the common case of the user being enabled.
-     */
-    sam.getUserStatusInfoAsync(callback)
-    // TODO: is it enough to check this one `getEnabled` boolean?
-    callback.future map (userStatus => Boolean.unbox(userStatus.getEnabled))
-  }
+  private def getUserEnabled(user: WithAccessToken): Future[Boolean] =
+    samDao.getUserStatus(user) map (userStatus => Boolean.unbox(userStatus.getEnabled))
 
   /**
     * Constructs and throws a FireCloudExceptionWithErrorReport in response to the

src/test/scala/org/broadinstitute/dsde/firecloud/dataaccess/MockSamDAO.scala

@@ -19,7 +19,7 @@ import org.broadinstitute.dsde.firecloud.model.{
 }
 import org.broadinstitute.dsde.workbench.util.health.SubsystemStatus
 import org.broadinstitute.dsde.rawls.model.{ErrorReport, RawlsUserEmail}
-import org.broadinstitute.dsde.workbench.client.sam.model.BulkMembershipUpdateRequestV2
+import org.broadinstitute.dsde.workbench.client.sam.model.{BulkMembershipUpdateRequestV2, UserStatusInfo}
 import org.broadinstitute.dsde.workbench.model.google.GoogleProject
 import org.broadinstitute.dsde.workbench.model.{
   AzureB2CId,
@@ -206,4 +206,20 @@ class MockSamDAO extends SamDAO {
   ): Future[Seq[WorkbenchUserInfo]] = Future.successful(Seq())
 
   override def bulkUpdateGroups(request: List[BulkMembershipUpdateRequestV2], user: WithAccessToken): Future[Unit] = ???
+
+  override def getUserStatus(user: WithAccessToken): Future[UserStatusInfo] =
+    user.accessToken.token match {
+      case MockSamDAO.disabledUserToken =>
+        Future.successful(new UserStatusInfo().enabled(false))
+      case MockSamDAO.unregisteredUserToken =>
+        Future.failed(new FireCloudExceptionWithErrorReport(ErrorReport(StatusCodes.NotFound, "User not found")))
+      case _ =>
+        Future.successful(new UserStatusInfo().enabled(true))
+    }
+
+}
+
+object MockSamDAO {
+  val disabledUserToken = "disabled"
+  val unregisteredUserToken = "unregistered"
 }