From 2e9f806278251ffae25b0d63fef32021930e9e89 Mon Sep 17 00:00:00 2001 From: Paul Mck Date: Thu, 27 May 2021 18:15:28 +0300 Subject: [PATCH 1/3] fix: use explicit ModelVersion encoding, because of ADT --- .../api/http/controller/events/SSEController.scala | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/src/main/scala/io/hydrosphere/serving/manager/api/http/controller/events/SSEController.scala b/src/main/scala/io/hydrosphere/serving/manager/api/http/controller/events/SSEController.scala index acda5368..754ded9a 100644 --- a/src/main/scala/io/hydrosphere/serving/manager/api/http/controller/events/SSEController.scala +++ b/src/main/scala/io/hydrosphere/serving/manager/api/http/controller/events/SSEController.scala @@ -1,7 +1,6 @@ package io.hydrosphere.serving.manager.api.http.controller.events import java.util.UUID - import akka.actor.ActorSystem import akka.http.scaladsl.marshalling.sse.EventStreamMarshalling._ import akka.http.scaladsl.model.sse.ServerSentEvent @@ -10,7 +9,6 @@ import akka.stream.scaladsl.Source import cats.effect.{ConcurrentEffect, ContextShift} import streamz.converter._ import io.circe.syntax._ - import io.hydrosphere.serving.manager.api.http.controller.AkkaHttpControllerDsl import io.hydrosphere.serving.manager.api.http.controller.application.ApplicationView import io.hydrosphere.serving.manager.api.http.controller.servable.ServableView @@ -18,11 +16,10 @@ import io.hydrosphere.serving.manager.discovery._ import io.hydrosphere.serving.manager.domain.application.ApplicationEvents import io.hydrosphere.serving.manager.domain.deploy_config import io.hydrosphere.serving.manager.domain.deploy_config.DeploymentConfigurationEvents -import io.hydrosphere.serving.manager.domain.model_version.ModelVersionEvents +import io.hydrosphere.serving.manager.domain.model_version.{ModelVersion, ModelVersionEvents} import io.hydrosphere.serving.manager.domain.monitoring.MetricSpecEvents import io.hydrosphere.serving.manager.domain.servable.ServableEvents - import scala.concurrent.ExecutionContext import scala.concurrent.duration._ @@ -31,7 +28,7 @@ class SSEController[F[_]]( modelSubscriber: ModelVersionEvents.Subscriber[F], servableSubscriber: ServableEvents.Subscriber[F], metricSpecSubscriber: MetricSpecEvents.Subscriber[F], - depSubscriber: DeploymentConfigurationEvents.Subscriber[F], + depSubscriber: DeploymentConfigurationEvents.Subscriber[F] )(implicit F: ConcurrentEffect[F], cs: ContextShift[F], @@ -120,7 +117,10 @@ object SSEController { case DiscoveryEvent.ItemUpdate(items) => items.map { mv => ServerSentEvent( - data = mv.asJson.spaces2, + data = mv match { + case m: ModelVersion.Internal => m.asJson.spaces2 + case m: ModelVersion.External => m.asJson.spaces2 + }, `type` = "ModelUpdate" ) } @@ -146,7 +146,7 @@ object SSEController { case DiscoveryEvent.ItemRemove(items) => items.map { i => ServerSentEvent( - data = i.toString, + data = i, `type` = "ApplicationRemove" ) } From 5877b2376a0f08fc4471ea80a4a4e9825d22fed3 Mon Sep 17 00:00:00 2001 From: Paul Mck Date: Mon, 7 Jun 2021 10:30:29 +0300 Subject: [PATCH 2/3] chore: add new threads --- .../io/hydrosphere/serving/manager/App.scala | 1 + .../io/hydrosphere/serving/manager/Core.scala | 6 ++-- .../serving/manager/api/http/HttpServer.scala | 31 ++++++++++--------- .../controller/events/SSEController.scala | 12 +++---- .../manager/domain/model/ModelService.scala | 12 +++++-- .../model_build/ModelVersionBuilder.scala | 8 ++--- 6 files changed, 38 insertions(+), 32 deletions(-) diff --git a/src/main/scala/io/hydrosphere/serving/manager/App.scala b/src/main/scala/io/hydrosphere/serving/manager/App.scala index 8ba0c9d2..90a57527 100644 --- a/src/main/scala/io/hydrosphere/serving/manager/App.scala +++ b/src/main/scala/io/hydrosphere/serving/manager/App.scala @@ -78,6 +78,7 @@ object App { implicit val storageOps = StorageOps.default[F] implicit val uuidGen = UUIDGenerator.default[F]() implicit val dc = dockerClient + for { rngF <- Resource.liftF(RNG.default[F]) cloudDriver = diff --git a/src/main/scala/io/hydrosphere/serving/manager/Core.scala b/src/main/scala/io/hydrosphere/serving/manager/Core.scala index 1369e13e..af725998 100644 --- a/src/main/scala/io/hydrosphere/serving/manager/Core.scala +++ b/src/main/scala/io/hydrosphere/serving/manager/Core.scala @@ -79,8 +79,9 @@ object Core { servableRepo: ServableRepository[F], appRepo: ApplicationRepository[F], buildLogsRepo: BuildLogRepository[F], - monitoringRepo: MonitoringRepository[F] - ): F[Core[F]] = { + monitoringRepo: MonitoringRepository[F], + cs: ContextShift[F] + ): F[Core[F]] = for { buildLoggingService <- BuildLoggingService.make[F]() core <- { @@ -127,5 +128,4 @@ object Core { } } } yield core - } } diff --git a/src/main/scala/io/hydrosphere/serving/manager/api/http/HttpServer.scala b/src/main/scala/io/hydrosphere/serving/manager/api/http/HttpServer.scala index 8f545b11..2344a6eb 100644 --- a/src/main/scala/io/hydrosphere/serving/manager/api/http/HttpServer.scala +++ b/src/main/scala/io/hydrosphere/serving/manager/api/http/HttpServer.scala @@ -14,6 +14,7 @@ import io.hydrosphere.serving.manager.api.http.controller.AkkaHttpControllerDsl import io.hydrosphere.serving.manager.config.ApplicationConfig import io.hydrosphere.serving.manager.util.AsyncUtil +import java.util.concurrent.Executors import scala.collection.immutable.Seq import scala.concurrent.ExecutionContext @@ -23,24 +24,26 @@ trait HttpServer[F[_]] { object HttpServer extends AkkaHttpControllerDsl { - def akkaBased[F[_] : Async]( - config: ApplicationConfig, - modelRoutes: Route, - applicationRoutes: Route, - hostSelectorRoutes: Route, - servableRoutes: Route, - sseRoutes: Route, - monitoringRoutes: Route, - externalModelRoutes: Route, - deploymentConfRoutes: Route, + def akkaBased[F[_]: Async]( + config: ApplicationConfig, + modelRoutes: Route, + applicationRoutes: Route, + hostSelectorRoutes: Route, + servableRoutes: Route, + sseRoutes: Route, + monitoringRoutes: Route, + externalModelRoutes: Route, + deploymentConfRoutes: Route )(implicit - as: ActorSystem, - am: ActorMaterializer, - ec: ExecutionContext + as: ActorSystem, + am: ActorMaterializer ): HttpServer[F] = { + val es = Executors.newCachedThreadPool() + implicit val ec: ExecutionContext = ExecutionContext.fromExecutor(es) + val controllerRoutes: Route = pathPrefix("v2") { handleExceptions(commonExceptionHandler) { - modelRoutes ~ + modelRoutes ~ externalModelRoutes ~ applicationRoutes ~ hostSelectorRoutes ~ diff --git a/src/main/scala/io/hydrosphere/serving/manager/api/http/controller/events/SSEController.scala b/src/main/scala/io/hydrosphere/serving/manager/api/http/controller/events/SSEController.scala index 754ded9a..05b451e4 100644 --- a/src/main/scala/io/hydrosphere/serving/manager/api/http/controller/events/SSEController.scala +++ b/src/main/scala/io/hydrosphere/serving/manager/api/http/controller/events/SSEController.scala @@ -1,10 +1,8 @@ package io.hydrosphere.serving.manager.api.http.controller.events import java.util.UUID -import akka.actor.ActorSystem import akka.http.scaladsl.marshalling.sse.EventStreamMarshalling._ import akka.http.scaladsl.model.sse.ServerSentEvent -import akka.stream.ActorMaterializer import akka.stream.scaladsl.Source import cats.effect.{ConcurrentEffect, ContextShift} import streamz.converter._ @@ -14,12 +12,12 @@ import io.hydrosphere.serving.manager.api.http.controller.application.Applicatio import io.hydrosphere.serving.manager.api.http.controller.servable.ServableView import io.hydrosphere.serving.manager.discovery._ import io.hydrosphere.serving.manager.domain.application.ApplicationEvents -import io.hydrosphere.serving.manager.domain.deploy_config import io.hydrosphere.serving.manager.domain.deploy_config.DeploymentConfigurationEvents import io.hydrosphere.serving.manager.domain.model_version.{ModelVersion, ModelVersionEvents} import io.hydrosphere.serving.manager.domain.monitoring.MetricSpecEvents import io.hydrosphere.serving.manager.domain.servable.ServableEvents +import java.util.concurrent.Executors import scala.concurrent.ExecutionContext import scala.concurrent.duration._ @@ -31,12 +29,10 @@ class SSEController[F[_]]( depSubscriber: DeploymentConfigurationEvents.Subscriber[F] )(implicit F: ConcurrentEffect[F], - cs: ContextShift[F], - ec: ExecutionContext, - actorSystem: ActorSystem + cs: ContextShift[F] ) extends AkkaHttpControllerDsl { - - implicit val am = ActorMaterializer.create(actorSystem) + val executor = Executors.newCachedThreadPool() + implicit val ec: ExecutionContext = ExecutionContext.fromExecutor(executor) def subscribe = pathPrefix("events") { diff --git a/src/main/scala/io/hydrosphere/serving/manager/domain/model/ModelService.scala b/src/main/scala/io/hydrosphere/serving/manager/domain/model/ModelService.scala index 298314b5..0e23b0c7 100644 --- a/src/main/scala/io/hydrosphere/serving/manager/domain/model/ModelService.scala +++ b/src/main/scala/io/hydrosphere/serving/manager/domain/model/ModelService.scala @@ -2,7 +2,7 @@ package io.hydrosphere.serving.manager.domain.model import java.nio.file.Path import cats.data.OptionT -import cats.effect.Clock +import cats.effect.{Clock, ContextShift} import cats.implicits._ import cats.{Monad, MonadError} import io.hydrosphere.serving.manager.api.http.controller.model._ @@ -20,6 +20,9 @@ import io.hydrosphere.serving.manager.util.DeferredResult import io.hydrosphere.serving.manager.util.InstantClockSyntax._ import org.apache.logging.log4j.scala.Logging +import java.util.concurrent.Executors +import scala.concurrent.ExecutionContext + trait ModelService[F[_]] { def get(modelId: Long): F[Model] @@ -48,9 +51,12 @@ object ModelService { appRepo: ApplicationRepository[F], servableRepo: ServableRepository[F], fetcher: ModelFetcher[F], - modelVersionBuilder: ModelVersionBuilder[F] + modelVersionBuilder: ModelVersionBuilder[F], + cs: ContextShift[F] ): ModelService[F] = new ModelService[F] with Logging { + val ex = Executors.newCachedThreadPool() + val ec = ExecutionContext.fromExecutor(ex) def deleteModel(modelId: Long): F[Model] = for { @@ -84,7 +90,7 @@ object ModelService { .leftMap(x => InvalidRequest(x.toList.mkString)) ) parentModel <- createIfNecessary(versionMetadata.modelName) - b <- modelVersionBuilder.build(parentModel, versionMetadata, modelPath) + b <- cs.evalOn(ec)(modelVersionBuilder.build(parentModel, versionMetadata, modelPath)) } yield b def createIfNecessary(modelName: String): F[Model] = diff --git a/src/main/scala/io/hydrosphere/serving/manager/domain/model_build/ModelVersionBuilder.scala b/src/main/scala/io/hydrosphere/serving/manager/domain/model_build/ModelVersionBuilder.scala index 07ae20df..8450d8da 100644 --- a/src/main/scala/io/hydrosphere/serving/manager/domain/model_build/ModelVersionBuilder.scala +++ b/src/main/scala/io/hydrosphere/serving/manager/domain/model_build/ModelVersionBuilder.scala @@ -49,7 +49,7 @@ object ModelVersionBuilder { _ <- handleBuild(init, modelFileStructure, handler).flatMap(deferred.complete).start } yield DeferredResult(init, deferred) - def initialVersion(model: Model, metadata: ModelVersionMetadata) = + private def initialVersion(model: Model, metadata: ModelVersionMetadata) = for { version <- modelVersionService.getNextModelVersion(model.id) image = imageRepository.getImage(metadata.modelName, version.toString) @@ -70,7 +70,7 @@ object ModelVersionBuilder { modelVersion <- modelVersionRepository.create(mv) } yield mv.copy(id = modelVersion.id) - def buildImage(buildPath: Path, image: DockerImage, handler: ProgressHandler) = + private def buildImage(buildPath: Path, image: DockerImage, handler: ProgressHandler) = for { imageId <- dockerClient.build( buildPath, @@ -82,7 +82,7 @@ object ModelVersionBuilder { res <- dockerClient.inspectImage(imageId) } yield res.id().stripPrefix("sha256:") - def handleBuild( + private def handleBuild( mv: ModelVersion.Internal, modelFileStructure: ModelFileStructure, handler: ProgressHandler @@ -111,7 +111,7 @@ object ModelVersionBuilder { } } - def prepare( + private def prepare( modelVersion: ModelVersion.Internal, modelFileStructure: ModelFileStructure ): F[ModelFileStructure] = From c2cb1612912f5f27783498c6f3bfdf45c9240bf7 Mon Sep 17 00:00:00 2001 From: Paul Mck Date: Mon, 7 Jun 2021 17:02:35 +0300 Subject: [PATCH 3/3] chore: naming --- .../serving/manager/api/http/HttpServer.scala | 27 +++++++++++-------- .../controller/events/SSEController.scala | 4 +-- 2 files changed, 18 insertions(+), 13 deletions(-) diff --git a/src/main/scala/io/hydrosphere/serving/manager/api/http/HttpServer.scala b/src/main/scala/io/hydrosphere/serving/manager/api/http/HttpServer.scala index 2344a6eb..3d8b3298 100644 --- a/src/main/scala/io/hydrosphere/serving/manager/api/http/HttpServer.scala +++ b/src/main/scala/io/hydrosphere/serving/manager/api/http/HttpServer.scala @@ -38,8 +38,8 @@ object HttpServer extends AkkaHttpControllerDsl { as: ActorSystem, am: ActorMaterializer ): HttpServer[F] = { - val es = Executors.newCachedThreadPool() - implicit val ec: ExecutionContext = ExecutionContext.fromExecutor(es) + val ex = Executors.newCachedThreadPool() + implicit val ec: ExecutionContext = ExecutionContext.fromExecutor(ex) val controllerRoutes: Route = pathPrefix("v2") { handleExceptions(commonExceptionHandler) { @@ -55,13 +55,17 @@ object HttpServer extends AkkaHttpControllerDsl { } val buildInfoRoute = pathPrefix("buildinfo") { - complete(HttpResponse( - status = StatusCodes.OK, - entity = HttpEntity(ContentTypes.`application/json`, BuildInfo.toJson) - )) + complete( + HttpResponse( + status = StatusCodes.OK, + entity = HttpEntity(ContentTypes.`application/json`, BuildInfo.toJson) + ) + ) } - val routes: Route = CorsDirectives.cors(CorsSettings.defaultSettings.withAllowedMethods(Seq(GET, POST, HEAD, OPTIONS, PUT, DELETE))) { + val routes: Route = CorsDirectives.cors( + CorsSettings.defaultSettings.withAllowedMethods(Seq(GET, POST, HEAD, OPTIONS, PUT, DELETE)) + ) { pathPrefix("health") { complete("OK") } ~ @@ -70,9 +74,10 @@ object HttpServer extends AkkaHttpControllerDsl { } } new HttpServer[F] { - override def start(): F[Http.ServerBinding] = AsyncUtil.futureAsync { - Http().bindAndHandle(routes, "0.0.0.0", config.port) - } + override def start(): F[Http.ServerBinding] = + AsyncUtil.futureAsync { + Http().bindAndHandle(routes, "0.0.0.0", config.port) + } } } -} \ No newline at end of file +} diff --git a/src/main/scala/io/hydrosphere/serving/manager/api/http/controller/events/SSEController.scala b/src/main/scala/io/hydrosphere/serving/manager/api/http/controller/events/SSEController.scala index 05b451e4..68462ff5 100644 --- a/src/main/scala/io/hydrosphere/serving/manager/api/http/controller/events/SSEController.scala +++ b/src/main/scala/io/hydrosphere/serving/manager/api/http/controller/events/SSEController.scala @@ -31,8 +31,8 @@ class SSEController[F[_]]( F: ConcurrentEffect[F], cs: ContextShift[F] ) extends AkkaHttpControllerDsl { - val executor = Executors.newCachedThreadPool() - implicit val ec: ExecutionContext = ExecutionContext.fromExecutor(executor) + val ex = Executors.newCachedThreadPool() + implicit val ec: ExecutionContext = ExecutionContext.fromExecutor(ex) def subscribe = pathPrefix("events") {