make user's logger used everywhere

core/src/it/scala/com/criteo/cuttle/DatabaseITest.scala

@@ -3,12 +3,14 @@ package com.criteo.cuttle
 import java.time.{Instant, LocalDateTime, ZoneOffset}
 import java.time.temporal.ChronoUnit
 
+import scala.concurrent.Future
+
 import cats.effect.IO
-import com.criteo.cuttle.Auth.User
 import doobie.implicits._
 import doobie.scalatest.IOChecker
 
-import scala.concurrent.Future
+import com.criteo.cuttle.Auth.User
+
 
 class DatabaseITest extends DatabaseSuite with IOChecker with TestScheduling {
   val dbConfig = DatabaseConfig(

core/src/it/scala/com/criteo/cuttle/DatabaseSuite.scala

@@ -8,7 +8,15 @@ import org.scalatest.{BeforeAndAfter, FunSuite}
 class DatabaseSuite extends FunSuite with BeforeAndAfter {
   val dbName = "cuttle_it_test"
 
-  val queries: Queries = new Queries {}
+  implicit val logger: Logger = new Logger {
+    override def debug(message: => String): Unit = ()
+    override def info(message: => String): Unit = ()
+    override def warn(message: => String): Unit = ()
+    override def error(message: => String): Unit = ()
+    override def trace(message: => String): Unit = ()
+  }
+
+  val queries: Queries = Queries(logger)
 
   private val dbConfig = DatabaseConfig(
     Seq(DBLocation("localhost", 3388)),

core/src/main/scala/com/criteo/cuttle/Database.scala

@@ -14,6 +14,7 @@ import io.circe.parser._
 import cats.data.NonEmptyList
 import cats.implicits._
 import cats.effect.{IO, Resource}
+import doobie.util.log
 
 import com.criteo.cuttle.ExecutionStatus._
 import com.criteo.cuttle.events.{Event, JobSuccessForced}
@@ -65,18 +66,6 @@ object DatabaseConfig {
 }
 
 private[cuttle] object Database {
-
-  implicit val ExecutionStatusMeta: Meta[ExecutionStatus] =
-    Meta[Boolean].imap(x => if (x) ExecutionSuccessful else ExecutionFailed: ExecutionStatus) {
-      case ExecutionSuccessful => true
-      case ExecutionFailed     => false
-      case x                   => sys.error(s"Unexpected ExecutionLog status to write in database: $x")
-    }
-
-  implicit val JsonMeta: Meta[Json] = Meta[String].imap(x => parse(x).fold(e => throw e, identity))(
-    x => x.noSpaces
-  )
-
   val schemaEvolutions = List(
     sql"""
       CREATE TABLE executions (
@@ -213,7 +202,7 @@ private[cuttle] object Database {
   private[cuttle] def reset(): Unit =
     connections.clear()
 
-  def connect(dbConfig: DatabaseConfig): XA = {
+  def connect(dbConfig: DatabaseConfig)(implicit logger: Logger): XA = {
     // FIXME we shouldn't use allocated as it's unsafe instead we have to flatMap on the Resource[HikariTransactor]
     val (transactor, releaseIO) = newHikariTransactor(dbConfig).allocated.unsafeRunSync
     logger.debug("Allocated new Hikari transactor")
@@ -229,8 +218,12 @@ private[cuttle] object Database {
   }
 }
 
-private[cuttle] trait Queries {
-  import Database._
+private[cuttle] case class Queries(logger: Logger) {
+  implicit val logHandler: log.LogHandler = DoobieLogsHandler(logger).handler
+
+  implicit val JsonMeta: Meta[Json] = Meta[String].imap(x => parse(x).fold(e => throw e, identity))(
+    x => x.noSpaces
+  )
 
   def logExecution(e: ExecutionLog, logContext: ConnectionIO[String]): ConnectionIO[Int] =
     for {

core/src/main/scala/com/criteo/cuttle/DoobieLogsHandler.scala

@@ -6,7 +6,7 @@ case class DoobieLogsHandler(private val logger: Logger) {
   implicit val handler: LogHandler = LogHandler {
 
     case Success(s, a, e1, e2) =>
-      logger.debug(s"""
+      logger.trace(s"""
                      | Successful Statement Execution:
                      |
                      | ${s.lines.dropWhile(_.trim.isEmpty).mkString("\n  ")}

core/src/main/scala/com/criteo/cuttle/ExecutionStreams.scala

@@ -45,8 +45,6 @@ private[cuttle] object ExecutionStreams {
   private val maxExecutionLogSizeProp = "com.criteo.cuttle.maxExecutionLogSize"
   private val maxExecutionLogSize = sys.props.get(maxExecutionLogSizeProp).map(_.toInt).getOrElse(524288)
 
-  logger.info(s"Transient execution streams go to $transientStorage")
-
   private def logFile(id: ExecutionId): File = new File(transientStorage, id)
 
   private def getWriter(id: ExecutionId): PrintWriter = {

core/src/main/scala/com/criteo/cuttle/Executor.scala

@@ -9,9 +9,8 @@ import java.util.{Timer, TimerTask}
 import scala.concurrent.duration._
 import scala.concurrent.stm._
 import scala.concurrent.{Future, Promise}
-import scala.reflect.{classTag, ClassTag}
+import scala.reflect.{ClassTag, classTag}
 import scala.util._
-
 import cats.Eq
 import cats.effect.IO
 import cats.implicits._
@@ -21,12 +20,12 @@ import io.circe._
 import io.circe.java8.time._
 import io.circe.syntax._
 import lol.http.PartialService
-
 import com.criteo.cuttle.Auth._
 import com.criteo.cuttle.ExecutionStatus._
 import com.criteo.cuttle.ThreadPools.{SideEffectThreadPool, _}
 import com.criteo.cuttle.Metrics._
 import com.criteo.cuttle.platforms.ExecutionPool
+import doobie.util.Meta
 
 /** The strategy to use to retry stuck executions.
   *
@@ -77,6 +76,13 @@ object RetryStrategy {
 
 private[cuttle] sealed trait ExecutionStatus
 private[cuttle] object ExecutionStatus {
+  implicit val ExecutionStatusMeta: Meta[ExecutionStatus] =
+    Meta[Boolean].imap(x => if (x) ExecutionSuccessful else ExecutionFailed: ExecutionStatus) {
+      case ExecutionSuccessful => true
+      case ExecutionFailed     => false
+      case x                   => sys.error(s"Unexpected ExecutionLog status to write in database: $x")
+    }
+
   case object ExecutionSuccessful extends ExecutionStatus
   case object ExecutionFailed extends ExecutionStatus
   case object ExecutionRunning extends ExecutionStatus
@@ -413,9 +419,7 @@ class Executor[S <: Scheduling] private[cuttle] (val platforms: Seq[ExecutionPla
 
   private implicit val contextOrdering: Ordering[S#Context] = Ordering.by(c => c: SchedulingContext)
 
-  private val queries = new Queries {
-    val appLogger: Logger = logger
-  }
+  private val queries = Queries(logger)
 
   // TODO: move to the scheduler
   private[cuttle] val runningState = TMap.empty[Execution[S], Future[Completed]]

core/src/main/scala/com/criteo/cuttle/Logger.scala

@@ -4,6 +4,7 @@ package com.criteo.cuttle
 trait Logger {
   def debug(message: => String): Unit
   def info(message: => String): Unit
-  def warning(message: => String): Unit
+  def warn(message: => String): Unit
   def error(message: => String): Unit
+  def trace(message: => String): Unit
 }

core/src/main/scala/com/criteo/cuttle/package.scala

@@ -4,7 +4,9 @@ import scala.concurrent._
 
 import cats.effect.IO
 import cats.free._
-import doobie.imports._
+import io.circe.Json
+import io.circe.parser._
+import doobie._
 
 package cuttle {
 
@@ -61,12 +63,7 @@ package object cuttle {
     */
   implicit def scopedExecutionContext(implicit execution: Execution[_]): ExecutionContext = execution.executionContext
 
-  /** Default implicit logger that output everything to __stdout__ */
-  implicit val logger = new Logger {
-    def logMe(message: => String, level: String) = println(s"${java.time.Instant.now}\t${level}\t${message}")
-    override def info(message: => String): Unit = logMe(message, "INFO")
-    override def debug(message: => String): Unit = logMe(message, "DEBUG")
-    override def warning(message: => String): Unit = logMe(message, "WARNING")
-    override def error(message: => String): Unit = logMe(message, "ERROR")
-  }
+  implicit val JsonMeta: Meta[Json] = Meta[String].imap(x => parse(x).fold(e => throw e, identity _))(
+    x => x.noSpaces
+  )
 }

core/src/test/scala/com/criteo/cuttle/ExecutorSpec.scala

@@ -12,6 +12,7 @@ import org.scalatest.FunSuite
 import com.criteo.cuttle.ThreadPools.Implicits.sideEffectThreadPool
 import com.criteo.cuttle.ThreadPools.Implicits.sideEffectContextShift
 import com.criteo.cuttle.ThreadPools._
+import com.criteo.cuttle.Utils.logger
 
 import com.criteo.cuttle.Metrics.Prometheus
 

core/src/test/scala/com/criteo/cuttle/Utils.scala

@@ -0,0 +1,11 @@
+package com.criteo.cuttle
+
+object Utils {
+  implicit val logger: Logger = new Logger {
+    override def debug(message: => String): Unit = ()
+    override def info(message: => String): Unit = ()
+    override def warn(message: => String): Unit = ()
+    override def error(message: => String): Unit = ()
+    override def trace(message: => String): Unit = ()
+  }
+}

cron/src/it/scala/com/criteo/cuttle/cron/UIITest.scala

@@ -10,6 +10,7 @@ import org.scalatest.{FlatSpec, Matchers}
 import com.criteo.cuttle.Auth.User
 import com.criteo.cuttle.{Job, _}
 import com.criteo.cuttle.ThreadPools.Implicits.sideEffectThreadPool
+import com.criteo.cuttle.cron.Utils.logger
 
 /**
   * These tests don't verify anything but the rendering for "/" and "/executions" pages.

cron/src/it/scala/com/criteo/cuttle/cron/Utils.scala

@@ -0,0 +1,12 @@
+package com.criteo.cuttle.cron
+import com.criteo.cuttle.Logger
+
+object Utils {
+  implicit val logger: Logger = new Logger {
+    override def debug(message: => String): Unit = ()
+    override def info(message: => String): Unit = ()
+    override def warn(message: => String): Unit = ()
+    override def error(message: => String): Unit = ()
+    override def trace(message: => String): Unit = ()
+  }
+}

cron/src/main/scala/com/criteo/cuttle/cron/CronProject.scala

@@ -40,7 +40,7 @@ class CronProject private[cuttle] (val name: String,
     retryStrategy: RetryStrategy = CronProject.retryStrategy
   ): Unit = {
     logger.info("Connecting to database")
-    implicit val transactor = com.criteo.cuttle.Database.connect(databaseConfig)
+    implicit val transactor = com.criteo.cuttle.Database.connect(databaseConfig)(logger)
 
     logger.info("Creating Executor")
     val executor = new Executor[CronScheduling](platforms, transactor, logger, name, version)(retryStrategy)

cron/src/main/scala/com/criteo/cuttle/cron/CronScheduler.scala

@@ -27,7 +27,8 @@ case class CronScheduler(logger: Logger) extends Scheduler[CronScheduling] {
 
   override val name = "cron"
 
-  private val queries = new Queries {}
+  private val queries = Queries(logger)
+
   private val state = CronState(logger)
 
   private def logState = IO(logger.debug(state.toString()))

examples/src/main/scala/com/criteo/cuttle/examples/package.scala

@@ -0,0 +1,14 @@
+package com.criteo.cuttle
+
+package object examples {
+
+  /** Default implicit logger that output everything to __stdout__ */
+  implicit val logger = new Logger {
+    def logMe(message: => String, level: String) = println(s"${java.time.Instant.now}\t${level}\t${message}")
+    override def info(message: => String): Unit = logMe(message, "INFO")
+    override def debug(message: => String): Unit = logMe(message, "DEBUG")
+    override def warn(message: => String): Unit = logMe(message, "WARNING")
+    override def error(message: => String): Unit = logMe(message, "ERROR")
+    override def trace(message: => String): Unit = logMe(message, "ERROR")
+  }
+}

timeseries/src/main/scala/com/criteo/cuttle/timeseries/CuttleProject.scala

@@ -35,7 +35,7 @@ class CuttleProject private[cuttle] (val name: String,
     retryStrategy: RetryStrategy = RetryStrategy.ExponentialBackoffRetryStrategy,
     paused: Boolean = false
   ): Unit = {
-    val xa = CuttleDatabase.connect(databaseConfig)
+    val xa = CuttleDatabase.connect(databaseConfig)(logger)
     val executor = new Executor[TimeSeries](platforms, xa, logger, name, version)(retryStrategy)
 
     if (paused) {
@@ -69,7 +69,7 @@ class CuttleProject private[cuttle] (val name: String,
     databaseConfig: DatabaseConfig = DatabaseConfig.fromEnv,
     retryStrategy: RetryStrategy = RetryStrategy.ExponentialBackoffRetryStrategy
   ): (Service, () => Unit) = {
-    val xa = CuttleDatabase.connect(databaseConfig)
+    val xa = CuttleDatabase.connect(databaseConfig)(logger)
     val executor = new Executor[TimeSeries](platforms, xa, logger, name, version)(retryStrategy)
 
     val startScheduler = () => {

timeseries/src/main/scala/com/criteo/cuttle/timeseries/Database.scala

@@ -17,7 +17,6 @@ import doobie.implicits._
 
 private[timeseries] object Database {
   import TimeSeriesUtils._
-  import com.criteo.cuttle.Database._
 
   import intervals.{Interval, IntervalMap}
 

timeseries/src/main/scala/com/criteo/cuttle/timeseries/TimeSeriesApp.scala

@@ -165,7 +165,6 @@ private[timeseries] case class TimeSeriesApp(project: CuttleProject, executor: E
       Ok(Prometheus.serialize(metrics))
 
     case GET at url"/api/executions/status/$kind?limit=$l&offset=$o&events=$events&sort=$sort&order=$order&jobs=$jobs" =>
-      logger.debug(s"Retreiving $kind executions with sse mode $events")
       val jobIds = parseJobIds(jobs)
       val limit = Try(l.toInt).toOption.getOrElse(25)
       val offset = Try(o.toInt).toOption.getOrElse(0)
@@ -343,7 +342,7 @@ private[timeseries] case class TimeSeriesApp(project: CuttleProject, executor: E
         "end" -> interval.hi.asJson
       )
   }
-  private val queries = new Queries {}
+  private val queries = Queries(project.logger)
 
   private trait ExecutionPeriod {
     val period: Interval[Instant]

timeseries/src/main/scala/com/criteo/cuttle/timeseries/TimeSeriesScheduler.scala

@@ -346,9 +346,7 @@ case class TimeSeriesScheduler(logger: Logger) extends Scheduler[TimeSeries] {
     _pausedJobs()
   }
 
-  private val queries = new Queries {
-    val appLogger: Logger = logger
-  }
+  private val queries = Queries(logger)
 
   private[timeseries] def state: (State, Set[Backfill]) = atomic { implicit txn =>
     (_state(), _backfills())

timeseries/src/test/scala/com/criteo/cuttle/timeseries/TimeSeriesSchedulerIntegrationTests.scala

@@ -1,7 +1,6 @@
 package com.criteo.cuttle.timeseries
 
 import java.time.ZoneOffset.UTC
-import java.time.temporal.ChronoUnit.HOURS
 import java.time.{Duration, Instant, LocalDate}
 import java.util.concurrent.TimeUnit
 
@@ -17,6 +16,7 @@ import com.wix.mysql.distribution.Version._
 import com.criteo.cuttle.platforms.local._
 import com.criteo.cuttle.timeseries.TimeSeriesUtils.{Run, TimeSeriesJob}
 import com.criteo.cuttle.{Auth, Database => CuttleDatabase, _}
+import Utils.logger
 
 object TimeSeriesSchedulerIntegrationTests {
   // TODO: turn this into a unit test. This is not done for now as the thread pool responsible for checking the lock on

timeseries/src/test/scala/com/criteo/cuttle/timeseries/TimeSeriesSchedulerSpec.scala

@@ -6,10 +6,11 @@ import scala.concurrent.Future
 
 import org.scalatest.FunSuite
 
-import com.criteo.cuttle.{logger, Completed, Job, TestScheduling}
+import com.criteo.cuttle.{Completed, Job, TestScheduling}
 import com.criteo.cuttle.timeseries.JobState.{Done, Todo}
 import com.criteo.cuttle.timeseries.TimeSeriesUtils.State
 import com.criteo.cuttle.timeseries.intervals.{Interval, IntervalMap}
+import com.criteo.cuttle.Utils.logger
 
 class TimeSeriesSchedulerSpec extends FunSuite with TestScheduling {
   private val scheduling: TimeSeries = hourly(date"2017-03-25T02:00:00Z")
@@ -38,7 +39,7 @@ class TimeSeriesSchedulerSpec extends FunSuite with TestScheduling {
         Interval(date"2017-03-25T02:00:00Z", date"2017-03-25T05:00:00Z") -> Done("")
       )
     )
-    val (stateSnapshot, newBackfills, completedBackfills) =
+    val (_, newBackfills, completedBackfills) =
       scheduler.collectCompletedJobs(state, Set(backfill), completed = Set.empty)
     assert(newBackfills.equals(Set(backfill)))
     assert(completedBackfills.isEmpty)
@@ -51,7 +52,7 @@ class TimeSeriesSchedulerSpec extends FunSuite with TestScheduling {
         Interval(date"2017-03-25T02:00:00Z", date"2017-03-25T05:00:00Z") -> Done("")
       )
     )
-    val (stateSnapshot, newBackfills, completedBackfills) = scheduler.collectCompletedJobs(
+    val (_, newBackfills, completedBackfills) = scheduler.collectCompletedJobs(
       state,
       Set(backfill),
       completed = Set(

timeseries/src/test/scala/com/criteo/cuttle/timeseries/TimeSeriesSpec.scala

@@ -8,6 +8,7 @@ import org.scalatest.FunSuite
 import com.criteo.cuttle._
 import com.criteo.cuttle.timeseries.intervals.Bound.{Finite, Top}
 import com.criteo.cuttle.timeseries.intervals._
+import com.criteo.cuttle.Utils.logger
 
 class TimeSeriesSpec extends FunSuite with TestScheduling {
   val scheduling: TimeSeries = hourly(date"2017-03-25T02:00:00Z")