Merge pull request #41 from criteo/upgrade_deps

jedirandy · web-flow · commit 1f45d42d0e89 · 2018-01-09T10:59:47.000+01:00
Upgrade dependencies
diff --git a/build.sbt b/build.sbt
@@ -3,8 +3,8 @@ import scala.sys.process.Process
 lazy val commonSettings = Seq(
   organization := "com.criteo",
   version := "0.4.8",
-  scalaVersion := "2.12.2",
-  crossScalaVersions := Seq("2.11.8", "2.12.2"),
+  scalaVersion := "2.12.4",
+  crossScalaVersions := Seq("2.11.8", "2.12.4"),
   credentials += Credentials(
     "Sonatype Nexus Repository Manager",
     "oss.sonatype.org",
@@ -92,14 +92,16 @@ lazy val root = (project in file("."))
   .settings(
     name := "slab",
     libraryDependencies ++= Seq(
-      "org.json4s" %% "json4s-native" % "3.4.2",
-      "org.scalatest" %% "scalatest" % "3.0.1" % Test,
-      "org.mockito" % "mockito-core" % "2.7.0" % Test,
       "org.slf4j" % "slf4j-api" % "1.7.25",
-      "com.criteo.lolhttp" %% "lolhttp" % "0.6.1",
+      "org.json4s" %% "json4s-native" % "3.4.2",
+      "com.criteo.lolhttp" %% "lolhttp" % "0.9.1",
       "com.github.cb372" %% "scalacache-core" % "0.9.4",
       "com.github.cb372" %% "scalacache-caffeine" % "0.9.4",
-      "com.chuusai" %% "shapeless" % "2.3.2"
+      "com.chuusai" %% "shapeless" % "2.3.2",
+      "org.typelevel" %% "cats-core" % "1.0.1",
+      "org.typelevel" %% "cats-effect" % "0.7",
+      "org.scalatest" %% "scalatest" % "3.0.1" % Test,
+      "org.mockito" % "mockito-core" % "2.7.0" % Test
     )
   )
 
diff --git a/example/src/main/scala/com/criteo/slab/example/GraphiteLauncher.scala b/example/src/main/scala/com/criteo/slab/example/GraphiteLauncher.scala
@@ -19,7 +19,7 @@ object GraphiteLauncher {
       host <- sys.env.get("GRAPHITE_HOST")
       port <- sys.env.get("GRAPHITE_PORT").map(_.toInt)
       webHost <- sys.env.get("GRAPHITE_WEB_HOST")
-    } yield new GraphiteStore(host, port, webHost, Duration.ofSeconds(60), Some("slab.example"))
+    } yield new GraphiteStore(host, port, webHost, Duration.ofSeconds(60), Some("slab.example"), Some("slab.example.slo"))
     implicit val store = maybeStore match {
       case Some(s) =>
         logger.info("[Slab Example] using Graphite store")
diff --git a/example/src/main/scala/com/criteo/slab/example/Launcher.scala b/example/src/main/scala/com/criteo/slab/example/Launcher.scala
@@ -5,6 +5,7 @@ package com.criteo.slab.example
 
 import java.net.URLDecoder
 
+import cats.effect.IO
 import com.criteo.slab.app.StateService.NotFoundError
 import com.criteo.slab.app.WebServer
 import com.criteo.slab.lib.InMemoryStore
@@ -32,14 +33,16 @@ object Launcher {
       .withRoutes(stateService => {
         case GET at "/api/heartbeat" => Ok("ok")
         case GET at url"/api/boards/$board/status" =>
-          stateService
-            .current(URLDecoder.decode(board, "UTF-8")).map(view => Ok(view.status.name))
-            .recover {
-              case NotFoundError(message) => NotFound(message)
-              case e =>
-                logger.error(e.getMessage, e)
-                InternalServerError()
-            }
+          IO.fromFuture(IO(
+            stateService
+              .current(URLDecoder.decode(board, "UTF-8")).map(view => Ok(view.status.name))
+              .recover {
+                case NotFoundError(message) => NotFound(message)
+                case e =>
+                  logger.error(e.getMessage, e)
+                  InternalServerError()
+              }
+          ))
       })
       // Attach a board to the server
       .attach(board)
diff --git a/src/main/scala/com/criteo/slab/app/WebServer.scala b/src/main/scala/com/criteo/slab/app/WebServer.scala
@@ -4,6 +4,8 @@ import java.net.URLDecoder
 import java.text.DecimalFormat
 import java.time.{Duration, Instant}
 
+import cats.Eval
+import cats.effect.IO
 import com.criteo.slab.app.StateService.NotFoundError
 import com.criteo.slab.core.Executor.{FetchBoardHistory, FetchBoardHourlySlo, RunBoard}
 import com.criteo.slab.core._
@@ -16,7 +18,6 @@ import shapeless.HList
 import shapeless.poly.Case3
 
 import scala.concurrent.{ExecutionContext, Future}
-import scala.util.Try
 import scala.util.control.NonFatal
 
 /** Slab Web server
@@ -30,7 +31,7 @@ import scala.util.control.NonFatal
 case class WebServer(
                       val pollingInterval: Int = 60,
                       val statsDays: Int = 730,
-                      private val routeGenerator: StateService => PartialFunction[Request, Future[Response]] = _ => PartialFunction.empty,
+                      private val routeGenerator: StateService => PartialFunction[Request, IO[Response]] = _ => PartialFunction.empty,
                       private val executors: List[Executor[_]] = List.empty
                     )(implicit ec: ExecutionContext) {
   /**
@@ -71,7 +72,7 @@ case class WebServer(
     * @param generator A function that takes StateService and returns routes
     * @return Web server with the created routes
     */
-  def withRoutes(generator: StateService => PartialFunction[Request, Future[Response]]) = this.copy(routeGenerator = generator)
+  def withRoutes(generator: StateService => PartialFunction[Request, IO[Response]]) = this.copy(routeGenerator = generator)
 
   private val logger = LoggerFactory.getLogger(this.getClass)
 
@@ -89,68 +90,80 @@ case class WebServer(
 
   private lazy val boards = executors.map(_.board)
 
-  private val routes: PartialFunction[Request, Future[Response]] = {
+  private val routes: PartialFunction[Request, IO[Response]] = {
     // Configs of boards
     case GET at url"/api/boards" => {
       Ok(boards.map { board => BoardConfig(board.title, board.layout, board.links, board.slo) }.toJSON).map(jsonContentType)
     }
     // Current board view
     case GET at url"/api/boards/$board" => {
       val boardName = URLDecoder.decode(board, "UTF-8")
-      stateService
-        .current(boardName)
-        .map((_: ReadableView).toJSON)
-        .map(Ok(_))
-        .map(jsonContentType)
-        .recoverWith(errorHandler)
+      IO.fromFuture(IO(
+        stateService
+          .current(boardName)
+          .map((_: ReadableView).toJSON)
+          .map(Ok(_))
+          .map(jsonContentType)
+          .recover(errorHandler)
+      ))
     }
     // Snapshot of the given time point
     case GET at url"/api/boards/$board/snapshot/$timestamp" => {
       val boardName = URLDecoder.decode(board, "UTF-8")
-      executors.find(_.board.title == boardName).fold(Future.successful(NotFound(s"Board $boardName does not exist"))) { executor =>
-        Try(timestamp.toLong).map(Instant.ofEpochMilli).toOption.fold(
-          Future.successful(BadRequest("invalid timestamp"))
-        ) { dateTime =>
-          executor.apply(Some(Context(dateTime)))
-            .map((_: ReadableView).toJSON)
-            .map(Ok(_))
-            .map(jsonContentType)
+      executors.find(_.board.title == boardName).fold(IO(NotFound(s"Board $boardName does not exist"))) { executor =>
+        IO(Instant.ofEpochMilli(timestamp.toLong)).attempt.flatMap {
+          case Left(_) => IO(BadRequest("invalid timestamp"))
+          case Right(dateTime) =>
+            IO.fromFuture(IO(
+              executor.apply(Some(Context(dateTime)))
+                .map((_: ReadableView).toJSON)
+                .map(Ok(_))
+                .map(jsonContentType)
+                .recover(errorHandler)
+            ))
         }
-      }.recoverWith(errorHandler)
+      }
     }
     // History of last 24 hours
     case GET at url"/api/boards/$board/history?last" => {
       val boardName = URLDecoder.decode(board, "UTF-8")
-      stateService
-        .history(boardName)
-        .map(h => Ok(h.toJSON))
-        .map(jsonContentType)
-        .recoverWith(errorHandler)
+      IO.fromFuture(IO(
+        stateService
+          .history(boardName)
+          .map(h => Ok(h.toJSON))
+          .map(jsonContentType)
+          .recover(errorHandler)
+      ))
     }
     // History of the given range
     case GET at url"/api/boards/$board/history?from=$fromTS&until=$untilTS" => {
       val boardName = URLDecoder.decode(board, "UTF-8")
-      executors.find(_.board.title == boardName).fold(Future.successful(NotFound(s"Board $boardName does not exist"))) { executor =>
-        val range = for {
-          from <- Try(fromTS.toLong).map(Instant.ofEpochMilli).toOption
-          until <- Try(untilTS.toLong).map(Instant.ofEpochMilli).toOption
-        } yield (from, until)
-        range.fold(Future.successful(BadRequest("Invalid timestamp"))) { case (from, until) =>
-          executor.fetchHistory(from, until)
-            .map(_.toMap.mapValues(_.status.name).toJSON)
-            .map(Ok(_))
-            .map(jsonContentType)
+      executors.find(_.board.title == boardName).fold(IO(NotFound(s"Board $boardName does not exist"))) { executor =>
+        IO(
+          Instant.ofEpochMilli(fromTS.toLong) -> Instant.ofEpochMilli(untilTS.toLong)
+        ).attempt.flatMap {
+          case Left(_) => IO(BadRequest("Invalid timestamp"))
+          case Right((from, until)) =>
+            IO.fromFuture(IO(
+              executor.fetchHistory(from, until)
+                .map(_.toMap.mapValues(_.status.name).toJSON)
+                .map(Ok(_))
+                .map(jsonContentType)
+                .recover(errorHandler)
+            ))
         }
-      }.recoverWith(errorHandler)
+      }
     }
     // Stats of the board
     case GET at url"/api/boards/$board/stats" => {
       val boardName = URLDecoder.decode(board, "UTF-8")
-      stateService.stats(boardName)
-        .map(_.mapValues(decimalFormat.format)).map(_.toJSON)
-        .map(Ok(_))
-        .map(jsonContentType)
-        .recoverWith(errorHandler)
+      IO.fromFuture(IO(
+        stateService.stats(boardName)
+          .map(_.mapValues(decimalFormat.format)).map(_.toJSON)
+          .map(Ok(_))
+          .map(jsonContentType)
+          .recover(errorHandler)
+      ))
     }
     // Static resources
     case GET at url"/$file.$ext" => {
@@ -161,14 +174,14 @@ case class WebServer(
     }
   }
 
-  private def notFound: PartialFunction[Request, Future[Response]] = {
+  private def notFound: PartialFunction[Request, IO[Response]] = {
     case anyReq => {
       logger.info(s"${anyReq.method.toString} ${anyReq.url} not found")
       Response(404)
     }
   }
 
-  private def errorHandler: PartialFunction[Throwable, Future[Response]] = {
+  private def errorHandler: PartialFunction[Throwable, Response] = {
     case f: NotFoundError =>
       NotFound(f.message)
     case NonFatal(e) =>
@@ -178,7 +191,7 @@ case class WebServer(
 
   private def jsonContentType(res: Response) = res.addHeaders(HttpString("content-type") -> HttpString("application/json"))
 
-  private def routeLogger(router: Request => Future[Response]) = (request: Request) => {
+  private def routeLogger(router: Request => IO[Response]) = (request: Request) => {
     val start = Instant.now()
     router(request) map { res =>
       val duration = Duration.between(start, Instant.now)
diff --git a/src/main/scala/com/criteo/slab/utils/HttpUtils.scala b/src/main/scala/com/criteo/slab/utils/HttpUtils.scala
@@ -4,7 +4,8 @@ import java.net.{URL, URLEncoder}
 import java.time.Instant
 import java.util.concurrent.TimeUnit.SECONDS
 
-import lol.http.{Client, ContentDecoder, HttpString, Get, Response}
+import cats.effect.IO
+import lol.http.{Client, ContentDecoder, Get, HttpString, Response}
 import org.slf4j.LoggerFactory
 
 import scala.concurrent.duration.{Duration, FiniteDuration}
@@ -39,12 +40,12 @@ object HttpUtils {
     val fullURL = s"${url.getProtocol}://${url.getHost}:${port}$path"
     logger.info(s"Requesting $fullURL")
     val start = Instant.now
-    Client(url.getHost, port, url.getProtocol) runAndStop { client =>
+    Client(url.getHost, port, url.getProtocol).runAndStop { client =>
       client.run(request, timeout = timeout) { res =>
         logger.info(s"Response from $fullURL, status: ${res.status}, ${Instant.now.toEpochMilli - start.toEpochMilli}ms")
         handleResponse(res, fullURL)
-      } recoverWith handleError(fullURL)
-    }
+      }
+    }.unsafeToFuture().recoverWith(handleError(fullURL))
   }
 
   /** Make a HTTP Get client of which the connection is kept open
@@ -73,19 +74,21 @@ object HttpUtils {
 
   private def encodeURI(in: String) = URLEncoder.encode(in, "UTF-8").replace("+", "%20")
 
-  private def handleResponse[A: ContentDecoder](res: Response, url: String)(implicit ec: ExecutionContext): Future[A] = {
+  private def handleResponse[A: ContentDecoder](res: Response, url: String)(implicit ec: ExecutionContext): IO[A] = {
     if (res.status < 400)
       res.readAs[A]
     else
-      res
-        .readAs[String]
-        .recoverWith { case e =>
-          logger.error(e.getMessage, e)
-          Future.successful("Unable to get the message")
+      res.readAs[String].attempt
+        .map {
+          case Left(e) =>
+            logger.error(e.getMessage, e)
+            "Unable to get the message"
+          case Right(message) =>
+            message
         }
         .flatMap { message =>
           logger.info(s"Request to $url has failed, status: ${res.status}, message: $message")
-          Future.failed(FailedRequestException(res))
+          IO.raiseError(FailedRequestException(res))
         }
   }
 
@@ -103,7 +106,7 @@ object HttpUtils {
   case class FailedRequestException(response: Response) extends Exception
 
   /**
-    * It be used to send GET requests using the same client
+    * Send GET requests using the same HTTP client
     *
     * @param client         The client
     * @param defaultHeaders Default request headers
@@ -122,7 +125,7 @@ object HttpUtils {
       client.run(request, timeout = timeout) { res: Response =>
         logger.info(s"Response from $fullURL, status: ${res.status}, ${Instant.now.toEpochMilli - start.toEpochMilli}ms")
         handleResponse(res, fullURL)
-      } recoverWith handleError(fullURL)
+      }.unsafeToFuture().recoverWith(handleError(fullURL))
     }
   }
 
