Merge branch 'release/2.3.1'

.travis.yml

@@ -4,7 +4,7 @@ jdk:
   - oraclejdk8
 
 scala:
-  - 2.11.2
+  - 2.11.4
 
 addons:
   postgresql: "9.3"
@@ -14,4 +14,6 @@ before_script:
   - psql -c "CREATE USER octoparts_app WITH PASSWORD '';" -U postgres
   - psql -c "GRANT ALL PRIVILEGES ON DATABASE octoparts_test to octoparts_app;" -U postgres
 
-script: "sbt coveralls test"
+script: "sbt clean coverage test"
+
+after_success: "sbt coverageAggregate coveralls"

app/com/m3/octoparts/aggregator/handler/Handler.scala

@@ -1,5 +1,6 @@
 package com.m3.octoparts.aggregator.handler
 
+import com.m3.octoparts.aggregator.PartRequestInfo
 import com.m3.octoparts.model.PartResponse
 import com.m3.octoparts.model.config.ShortPartParam
 
@@ -21,5 +22,6 @@ trait Handler {
   // Used primarily for creating a PartResponse, but also for logging purposes
   def partId: String
 
-  def process(arguments: HandlerArguments): Future[PartResponse]
+  def process(partRequestInfo: PartRequestInfo, arguments: HandlerArguments): Future[PartResponse]
+
 }

app/com/m3/octoparts/aggregator/handler/HttpPartRequestHandler.scala

@@ -2,6 +2,7 @@ package com.m3.octoparts.aggregator.handler
 
 import java.net.{ URI, URLEncoder }
 
+import com.m3.octoparts.aggregator.PartRequestInfo
 import com.m3.octoparts.http._
 import com.m3.octoparts.hystrix._
 import com.m3.octoparts.model.{ HttpMethod, PartResponse }
@@ -47,12 +48,13 @@ trait HttpPartRequestHandler extends Handler {
    * contain a Failure instead of Success. Make sure to transform with
    * .recover
    *
+   * @param partRequestInfo info about the request, used for generating HTTP headers for request tracing
    * @param hArgs Preparsed HystrixArguments
    * @return Future[PartResponse]
    */
-  def process(hArgs: HandlerArguments): Future[PartResponse] = {
+  def process(partRequestInfo: PartRequestInfo, hArgs: HandlerArguments): Future[PartResponse] = {
     hystrixExecutor.future {
-      createBlockingHttpRetrieve(hArgs).retrieve()
+      createBlockingHttpRetrieve(partRequestInfo, hArgs).retrieve()
     }.map {
       createPartResponse
     }
@@ -64,18 +66,27 @@ trait HttpPartRequestHandler extends Handler {
    * @param hArgs Handler arguments
    * @return a command object that will perform an HTTP request on demand
    */
-  def createBlockingHttpRetrieve(hArgs: HandlerArguments): BlockingHttpRetrieve = {
+  def createBlockingHttpRetrieve(partRequestInfo: PartRequestInfo, hArgs: HandlerArguments): BlockingHttpRetrieve = {
     new BlockingHttpRetrieve {
       val httpClient = handler.httpClient
       def method = httpMethod
       val uri = new URI(buildUri(hArgs))
       val maybeBody = hArgs.collectFirst {
         case (p, values) if p.paramType == ParamType.Body && values.nonEmpty => values.head
       }
-      val headers = collectHeaders(hArgs)
+      val headers = collectHeaders(hArgs) ++ buildTracingHeaders(partRequestInfo)
     }
   }
 
+  private def buildTracingHeaders(partRequestInfo: PartRequestInfo): Seq[(String, String)] = {
+    import HttpPartRequestHandler._
+    Seq(
+      AggregateRequestIdHeader -> partRequestInfo.requestMeta.id,
+      PartRequestIdHeader -> partRequestInfo.partRequestId,
+      PartIdHeader -> partRequestInfo.partRequest.partId
+    )
+  }
+
   /**
    * Transforms a HttpResponse case class into a PartResponse
    * @param httpResp HttpResponse
@@ -141,7 +152,8 @@ trait HttpPartRequestHandler extends Handler {
       maybeParamsVal.getOrElse("")
     }
     val kvs = for {
-      (p, values) <- hArgs if p.paramType == ParamType.Query
+      // toSeq because we don't want the result to be a map (with unique keys)
+      (p, values) <- hArgs.toSeq if p.paramType == ParamType.Query
       v <- values
     } yield p.outputName -> v
     baseUri.addParams(kvs.toSeq)
@@ -157,4 +169,10 @@ trait HttpPartRequestHandler extends Handler {
   private def interpolate(stringToInterpolate: String)(replacer: String => String) =
     PlaceholderReplacer.replaceAllIn(stringToInterpolate, { m => replacer(m.group(1)) })
 
+}
+
+object HttpPartRequestHandler {
+  val AggregateRequestIdHeader = "X-OCTOPARTS-PARENT-REQUEST-ID"
+  val PartRequestIdHeader = "X-OCTOPARTS-REQUEST-ID"
+  val PartIdHeader = "X-OCTOPARTS-PART-ID"
 }

app/com/m3/octoparts/aggregator/service/PartRequestServiceBase.scala

@@ -1,12 +1,11 @@
 package com.m3.octoparts.aggregator.service
 
+import com.beachape.logging.LTSVLogger
 import com.m3.octoparts.aggregator.PartRequestInfo
 import com.m3.octoparts.aggregator.handler.HttpHandlerFactory
 import com.m3.octoparts.model.PartResponse
 import com.m3.octoparts.model.config.{ HttpPartConfig, ShortPartParam }
 import com.m3.octoparts.repository.ConfigsRepository
-import skinny.logging.Logging
-import skinny.util.LTSV
 
 import scala.concurrent.{ ExecutionContext, Future }
 
@@ -16,7 +15,7 @@ import scala.concurrent.{ ExecutionContext, Future }
  * Implements the basic shared methods but leaves some methods to be implemented in
  * children classes / decorators
  */
-trait PartRequestServiceBase extends RequestParamSupport with Logging {
+trait PartRequestServiceBase extends RequestParamSupport {
   implicit def executionContext: ExecutionContext
 
   def repository: ConfigsRepository
@@ -50,7 +49,7 @@ trait PartRequestServiceBase extends RequestParamSupport with Logging {
    */
   private def unsupported(pReq: PartRequestInfo): Future[PartResponse] = {
     val partId = pReq.partRequest.partId
-    warn(LTSV.dump("Request Id" -> pReq.requestMeta.id, "Requested PartId" -> partId, "Error" -> "not found"))
+    LTSVLogger.warn("Request Id" -> pReq.requestMeta.id, "Requested PartId" -> partId, "Error" -> "not found")
     Future.successful(PartResponse(partId, pReq.partRequestId, errors = Seq(unsupportedMsg(partId))))
   }
 
@@ -82,7 +81,7 @@ trait PartRequestServiceBase extends RequestParamSupport with Logging {
    */
   protected def processWithConfig(ci: HttpPartConfig, partRequestInfo: PartRequestInfo, params: Map[ShortPartParam, Seq[String]]): Future[PartResponse] = {
     val handler = handlerFactory.makeHandler(ci)
-    val fResp = handler.process(params)
+    val fResp = handler.process(partRequestInfo, params)
     fResp.map {
       resp =>
         val respWithId = resp.copy(id = partRequestInfo.partRequestId)

app/com/m3/octoparts/cache/PartResponseCachingSupport.scala

@@ -1,13 +1,12 @@
 package com.m3.octoparts.cache
 
+import com.beachape.logging.LTSVLogger
 import com.m3.octoparts.aggregator.PartRequestInfo
 import com.m3.octoparts.aggregator.service.PartRequestServiceBase
 import com.m3.octoparts.cache.directive.{ CacheDirective, CacheDirectiveGenerator }
 import com.m3.octoparts.model.PartResponse
 import com.m3.octoparts.model.config._
 import org.apache.http.HttpStatus
-import skinny.logging.Logging
-import skinny.util.LTSV
 import com.m3.octoparts.cache.RichCacheControl._
 
 import scala.concurrent.Future
@@ -30,7 +29,7 @@ private[cache] object PartResponseCachingSupport {
   }
 }
 
-trait PartResponseCachingSupport extends PartRequestServiceBase with Logging {
+trait PartResponseCachingSupport extends PartRequestServiceBase {
   import PartResponseCachingSupport._
 
   def cacheOps: CacheOps
@@ -67,14 +66,14 @@ trait PartResponseCachingSupport extends PartRequestServiceBase with Logging {
     case ce: CacheException => {
       ce.getCause match {
         case te: shade.TimeoutException =>
-          warn(LTSV.dump("Memcached error" -> "timed out", "cache key" -> ce.key.toString))
+          LTSVLogger.warn("Memcached error" -> "timed out", "cache key" -> ce.key.toString)
         case other =>
-          error(LTSV.dump("Memcached error" -> other.getClass.getSimpleName, "cache key" -> ce.key.toString), other)
+          LTSVLogger.error(other, "Memcached error" -> other.getClass.getSimpleName, "cache key" -> ce.key.toString)
       }
       super.processWithConfig(ci, partRequestInfo, params)
     }
     case NonFatal(e) => {
-      error(LTSV.dump("Memcached error" -> e.getClass.getSimpleName), e)
+      LTSVLogger.error(e, "Memcached error" -> e.getClass.getSimpleName)
       super.processWithConfig(ci, partRequestInfo, params)
     }
   }

app/com/m3/octoparts/cache/memcached/MemcachedCacheOps.scala

@@ -6,18 +6,15 @@ import com.m3.octoparts.cache.key.{ PartCacheKey, VersionCacheKey }
 import com.m3.octoparts.cache.versioning._
 import com.m3.octoparts.model.PartResponse
 import shade.memcached.Codec
-import skinny.logging.Logging
 
 import scala.concurrent.duration.Duration
 import scala.concurrent.{ ExecutionContext, Future }
-import scala.util.control.NonFatal
 
 class MemcachedCacheOps(
   cache: Cache,
   latestVersionCache: LatestVersionCache)(implicit executionContext: ExecutionContext)
     extends CacheOps
-    with TtlCalculator
-    with Logging {
+    with TtlCalculator {
 
   import com.m3.octoparts.cache.versioning.LatestVersionCache.Version
 
@@ -48,7 +45,7 @@ class MemcachedCacheOps(
     }
 
     def insertPartResponse(cacheKey: PartCacheKey, partResponse: PartResponse, ttl: Option[Duration]): Future[Unit] =
-      cache.put[String](cacheKey, Json.toJson(partResponse).toString, calculateTtl(partResponse.cacheControl, ttl))
+      cache.put[String](cacheKey, Json.toJson(partResponse).toString(), calculateTtl(partResponse.cacheControl, ttl))
   }
 
   object CombinedVersionLookup {

app/com/m3/octoparts/http/InstrumentedHttpClient.scala

@@ -6,7 +6,6 @@ import java.nio.charset.{ Charset, StandardCharsets }
 import com.beachape.logging.LTSVLogger
 import com.codahale.metrics.{ Gauge, MetricRegistry }
 import com.m3.octoparts.OctopartsMetricsRegistry
-import com.m3.octoparts.util.TimingSupport
 import org.apache.http.{ HttpClientConnection, HttpRequest }
 import org.apache.http.client.HttpClient
 import org.apache.http.client.config.{ CookieSpecs, RequestConfig }
@@ -16,8 +15,6 @@ import org.apache.http.impl.client.HttpClientBuilder
 import org.apache.http.impl.conn.PoolingHttpClientConnectionManager
 import org.apache.http.pool.PoolStats
 import org.apache.http.protocol.{ HttpContext, HttpRequestExecutor }
-import skinny.logging.Logging
-import skinny.util.LTSV
 
 import scala.concurrent.duration._
 
@@ -37,8 +34,6 @@ class InstrumentedHttpClient(
   socketTimeout: Duration = 10.seconds,
   defaultEncoding: Charset = StandardCharsets.UTF_8)
     extends HttpClientLike
-    with TimingSupport
-    with Logging
     with Closeable {
   import InstrumentedHttpClient._
 
@@ -71,18 +66,7 @@ class InstrumentedHttpClient(
    * @param request HttpUriRequest
    * @return HttpResponse
    */
-  def retrieve(request: HttpUriRequest): HttpResponse = {
-    debug(LTSV.dump("HTTP request" -> request.toString))
-    time {
-      httpClient.execute(request, responseHandler)
-    } {
-      (httpResponse, duration) =>
-        debug(LTSV.dump(
-          "HTTP status" -> httpResponse.status.toString,
-          "HTTP response time" -> duration.toString
-        ))
-    }
-  }
+  def retrieve(request: HttpUriRequest): HttpResponse = httpClient.execute(request, responseHandler)
 
   /**
    * A [[PoolingHttpClientConnectionManager]] which monitors the number of open connections.

app/com/m3/octoparts/repository/ConfigImporter.scala

@@ -1,10 +1,10 @@
 package com.m3.octoparts.repository
 
+import com.beachape.logging.LTSVLogger
 import com.m3.octoparts.model.config.json
 import com.m3.octoparts.model.config._
 import com.m3.octoparts.repository.config._
 import scalikejdbc._
-import skinny.util.LTSV
 
 import scala.concurrent.Future
 
@@ -38,7 +38,7 @@ trait ConfigImporter {
             saveWithSession(ThreadPoolConfigRepository, ThreadPoolConfig.fromJsonModel(threadPoolConfig)).map(threadPoolConfig.threadPoolKey -> _)
           } {
             tpcWasThere =>
-              debug(LTSV.dump("Thread pool" -> tpcWasThere.threadPoolKey, "Action" -> "Skipping import"))
+              LTSVLogger.debug("Thread pool" -> tpcWasThere.threadPoolKey, "Action" -> "Skipping import")
               Future.successful(threadPoolConfig.threadPoolKey -> tpcWasThere.id.get)
           }
         }
@@ -63,7 +63,7 @@ trait ConfigImporter {
               getWithSession(CacheGroupRepository, sqls.eq(CacheGroupRepository.defaultAlias.id, cacheGroupId)).map(jCacheGroup.name -> _.get)
             }
           } { cgWasThere =>
-            debug(LTSV.dump("Cache group" -> cgWasThere.name, "Action" -> "Skipping import"))
+            LTSVLogger.debug("Cache group" -> cgWasThere.name, "Action" -> "Skipping import")
             Future.successful(cgWasThere.name -> cgWasThere)
           }
         }
@@ -116,7 +116,7 @@ trait ConfigImporter {
      * @return the [[HttpPartConfig.partId]] if inserted, else [[None]]
      */
     private[repository] def insertConfigIfMissing(jpart: json.HttpPartConfig)(implicit session: DBSession): Future[Option[String]] = {
-      info(LTSV.dump("part" -> jpart.toString, "action" -> "insert if missing"))
+      LTSVLogger.info("part" -> jpart.toString, "action" -> "insert if missing")
 
       val oldConfig = getWithSession(HttpPartConfigRepository, sqls.eq(HttpPartConfigRepository.defaultAlias.partId, jpart.partId))
       oldConfig.flatMap { mbConfigWasThere =>

app/com/m3/octoparts/repository/DBConfigsRepository.scala

@@ -1,14 +1,13 @@
 package com.m3.octoparts.repository
 
+import com.beachape.logging.LTSVLogger
 import com.m3.octoparts.model.config._
 import com.m3.octoparts.repository.config._
 import play.api.Play
 import play.api.libs.concurrent.Akka
 import scalikejdbc._
-import skinny.logging.Logging
 import skinny.orm.SkinnyCRUDMapper
 import skinny.orm.feature.associations.Association
-import skinny.util.LTSV
 import com.m3.octoparts.future.RichFutureWithTiming._
 
 import scala.concurrent.{ Future, blocking }
@@ -45,7 +44,7 @@ object DBContext {
 
 }
 
-trait ImmutableDBRepository extends ConfigsRepository with Logging {
+trait ImmutableDBRepository extends ConfigsRepository {
   import DBContext._
 
   // Configs
@@ -81,7 +80,7 @@ trait ImmutableDBRepository extends ConfigsRepository with Logging {
     blocking {
       val ret = mapper.joins(joins: _*).includes(includes: _*).findBy(where)
       ret.foreach {
-        ci => debug(LTSV.dump("Table" -> mapper.tableName, "where" -> where.toString(), "Data" -> ret.toString))
+        ci => LTSVLogger.debug("Table" -> mapper.tableName, "where" -> where.toString(), "Data" -> ret.toString)
       }
       ret
     }
@@ -95,7 +94,7 @@ trait ImmutableDBRepository extends ConfigsRepository with Logging {
                                                includes: Seq[Association[A]] = Nil)(implicit session: DBSession = ReadOnlyAutoSession): Future[Seq[A]] = Future {
     blocking {
       val ret = mapper.joins(joins: _*).includes(includes: _*).findAll()
-      debug(LTSV.dump("Table" -> mapper.tableName, "Retrieved records" -> ret.length.toString))
+      LTSVLogger.debug("Table" -> mapper.tableName, "Retrieved records" -> ret.length.toString)
       ret
     }
   }.measure("DB_GET")
@@ -109,13 +108,13 @@ trait ImmutableDBRepository extends ConfigsRepository with Logging {
                                                  includes: Seq[Association[A]] = Nil)(implicit session: DBSession = ReadOnlyAutoSession): Future[Seq[A]] = Future {
     blocking {
       val ret = mapper.joins(joins: _*).includes(includes: _*).findAllBy(where)
-      debug(LTSV.dump("Table" -> mapper.tableName, "Retrieved records" -> ret.length.toString))
+      LTSVLogger.debug("Table" -> mapper.tableName, "Retrieved records" -> ret.length.toString)
       ret
     }
   }.measure("DB_GET")
 }
 
-trait MutableDBRepository extends MutableConfigsRepository with Logging {
+trait MutableDBRepository extends MutableConfigsRepository {
   import DBContext._
 
   def save[A <: ConfigModel[A]: ConfigMapper](obj: A): Future[Long] = DB.futureLocalTx { implicit session => saveWithSession(implicitly[ConfigMapper[A]], obj) }
@@ -136,7 +135,7 @@ trait MutableDBRepository extends MutableConfigsRepository with Logging {
   private[repository] def saveWithSession[A <: ConfigModel[A]](mapper: ConfigMapper[A], model: A)(implicit session: DBSession = AutoSession): Future[Long] = Future {
     blocking {
       val id = mapper.save(model)
-      info(LTSV.dump("Table" -> mapper.tableName, "Data" -> model.toString, "Action" -> "Saved"))
+      LTSVLogger.info("Table" -> mapper.tableName, "Data" -> model.toString, "Action" -> "Saved")
       id
     }
   }.measure("DB_UPDATE")
@@ -148,7 +147,7 @@ trait MutableDBRepository extends MutableConfigsRepository with Logging {
   private[repository] def deleteWithSession(mapper: SkinnyCRUDMapper[_], where: SQLSyntax)(implicit session: DBSession = AutoSession): Future[Int] = Future {
     blocking {
       val count = mapper.deleteBy(where)
-      info(LTSV.dump("Table" -> mapper.tableName, "where" -> where.toString(), "count" -> count.toString, "Action" -> "Deleted"))
+      LTSVLogger.info("Table" -> mapper.tableName, "where" -> where.toString(), "count" -> count.toString, "Action" -> "Deleted")
       count
     }
   }.measure("DB_UPDATE")
@@ -159,7 +158,7 @@ trait MutableDBRepository extends MutableConfigsRepository with Logging {
   private[repository] def deleteAllWithSession(mapper: SkinnyCRUDMapper[_])(implicit session: DBSession = AutoSession): Future[Int] = Future {
     blocking {
       val count = mapper.deleteAll
-      warn(LTSV.dump("Table" -> mapper.tableName, "count" -> count.toString, "Action" -> "Truncated"))
+      LTSVLogger.warn("Table" -> mapper.tableName, "count" -> count.toString, "Action" -> "Truncated")
       count
     }
   }.measure("DB_UPDATE")