Merge branch 'release/2.5'

.travis.yml

@@ -0,0 +1,32 @@
+language: scala
+
+jdk:
+  - oraclejdk8
+
+scala:
+  - 2.11.6
+env:
+  - SBT_OPTS="-XX:+UseConcMarkSweepGC -XX:+CMSClassUnloadingEnabled -XX:PermSize=256m -XX:MaxPermSize=512m -Xms128m -Xmx512m"
+  - JAVA_OPTS="-XX:+UseConcMarkSweepGC -XX:+CMSClassUnloadingEnabled -XX:PermSize=256m -XX:MaxPermSize=512m -Xms1024m -Xmx1024m"
+
+addons:
+  postgresql: "9.3"
+
+before_script:
+  - psql -c "CREATE DATABASE octoparts_test WITH ENCODING 'UTF8';" -U postgres
+  - 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 clean compile
+  - sbt coverage test
+  - find $HOME/.sbt -name "*.lock" -type f -delete && find $HOME/.ivy2/cache -name "*[\[\]\(\)]*.properties" -type f -delete
+
+after_success: sbt coverageAggregate coveralls
+
+sudo: false
+
+cache:
+  directories:
+  - $HOME/.sbt
+  - $HOME/.ivy2

LICENSE.txt

@@ -0,0 +1,13 @@
+Copyright 2014 M3, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+either express or implied. See the License for the specific language
+governing permissions and limitations under the License.

README.md

@@ -1,3 +1,5 @@
-# Octoparts
+# Octoparts [![Build Status](https://travis-ci.org/m3dev/octoparts.svg?branch=develop)](https://travis-ci.org/m3dev/octoparts) [![Coverage Status](https://coveralls.io/repos/m3dev/octoparts/badge.svg?branch=develop)](https://coveralls.io/r/m3dev/octoparts?branch=develop)
 
 [See documentation](http://m3dev.github.io/octoparts/)
+
+Also see these [Lightning talk slides](https://docs.google.com/presentation/d/1dgbLSaEyWydGX6SaPtGeKXX-6p-0OuUvnWFpiOqgoQo/edit?usp=sharing) from ScalaMatsuri 2014 for a quick explanation of what Octoparts is all about.

app/com/m3/octoparts/Global.scala

@@ -2,27 +2,26 @@ package com.m3.octoparts
 
 import java.io.File
 import java.util.concurrent.TimeUnit
-
-import _root_.controllers.ControllersModule
+import com.beachape.zipkin.ZipkinHeaderFilter
 import com.kenshoo.play.metrics.MetricsFilter
-import com.m3.octoparts.cache.CacheModule
-import com.m3.octoparts.http.HttpModule
-import com.m3.octoparts.hystrix.{ HystrixMetricsLogger, HystrixModule }
-import com.m3.octoparts.logging.PartRequestLogger
-import com.m3.octoparts.repository.RepositoriesModule
+import com.m3.octoparts.hystrix.{ KeyAndBuilderValuesHystrixPropertiesStrategy, HystrixMetricsLogger }
+import com.beachape.logging.LTSVLogger
+import com.m3.octoparts.wiring.Amalgamated
+import com.netflix.hystrix.strategy.HystrixPlugins
+import com.softwaremill.macwire.MacwireMacros._
+import com.twitter.zipkin.gen.Span
 import com.typesafe.config.ConfigFactory
 import com.wordnik.swagger.config.{ ConfigFactory => SwaggerConfigFactory }
 import com.wordnik.swagger.model.ApiInfo
+import org.apache.commons.lang3.StringUtils
 import play.api._
 import play.api.libs.concurrent.Akka
 import play.api.mvc._
-import scaldi.Module
-import scaldi.play.ScaldiSupport
 
-import scala.collection.concurrent.TrieMap
 import scala.concurrent.duration._
+import scala.util.control.NonFatal
 
-object Global extends WithFilters(MetricsFilter) with ScaldiSupport {
+object Global extends WithFilters(ZipkinHeaderFilter(ZipkinServiceHolder.ZipkinService), MetricsFilter) {
 
   val info = ApiInfo(
     title = "Octoparts",
@@ -34,33 +33,14 @@ object Global extends WithFilters(MetricsFilter) with ScaldiSupport {
 
   SwaggerConfigFactory.config.setApiInfo(info)
 
-  def applicationModule =
-    aggregator.module ::
-      new RepositoriesModule ::
-      new CacheModule ::
-      new HystrixModule ::
-      new HttpModule ::
-      new ControllersModule ::
-      new Module {
-        // Random stuff that doesn't belong in other modules
-        bind[PartRequestLogger] to PartRequestLogger
-      }
+  lazy val Components = new Amalgamated(Play.current)
+  lazy val InjectionModule = wiredInModule(Components)
 
-  /**
-   * For each entry, V.getClass == K
-   */
-  private val controllerCache = TrieMap[Class[_], Any]()
-
-  /**
-   * Caches controller instantiation which was shown to be expensive because of ScalDI.
-   */
-  override def getControllerInstance[A](clazz: Class[A]): A = {
-    controllerCache.getOrElseUpdate(clazz, super.getControllerInstance(clazz)).asInstanceOf[A]
-  }
-
-  override def onStop(app: Application) = {
-    controllerCache.clear()
-    super.onStop(app)
+  override def getControllerInstance[A](controllerClass: Class[A]): A = {
+    InjectionModule.lookup(controllerClass) match {
+      case head :: Nil => head
+      case _ => super.getControllerInstance(controllerClass)
+    }
   }
 
   // Load environment-specific application.${env}.conf, merged with the generic application.conf
@@ -73,27 +53,71 @@ object Global extends WithFilters(MetricsFilter) with ScaldiSupport {
         case (_, env) => env
       }
     }
-    Logger.debug(s"Play environment = $playEnv (mode = $mode, application.env = ${config.getString("application.env")}). Loading extra config from application.$playEnv.conf, if it exists.")
+    LTSVLogger.debug("Play environment" -> playEnv, "mode" -> mode, "application.env" -> config.getString("application.env"), "message" -> "Loading extra config...")
     val modeSpecificConfig = config ++ Configuration(ConfigFactory.load(s"application.$playEnv.conf"))
     super.onLoadConfig(modeSpecificConfig, path, classloader, mode)
   }
 
   override def onStart(app: Application) = {
+    // Need to do this as early as possible, before Hystrix gets instantiated
+    setHystrixPropertiesStrategy()
+
     super.onStart(app)
 
     startPeriodicTasks(app)
+    checkForDodgyPartIds()
   }
 
   /**
    * Register any tasks that should be run on the global Akka scheduler.
    * These tasks will automatically stop running when the app shuts down.
    */
-  def startPeriodicTasks(implicit app: Application): Unit = {
+  private def startPeriodicTasks(implicit app: Application): Unit = {
     import play.api.libs.concurrent.Execution.Implicits.defaultContext
 
     val hystrixLoggingInterval = app.configuration.underlying.getDuration("hystrix.logging.intervalMs", TimeUnit.MILLISECONDS).toInt.millis
     Akka.system.scheduler.schedule(hystrixLoggingInterval, hystrixLoggingInterval) {
       HystrixMetricsLogger.logHystrixMetrics()
     }
   }
+
+  /**
+   * Check if there are any registered parts with leading/trailing spaces in their partIds.
+   * Output warning logs if we find any, as they can be a nightmare to debug and are best avoided.
+   */
+  private def checkForDodgyPartIds(): Unit = {
+    import play.api.libs.concurrent.Execution.Implicits.defaultContext
+    implicit val emptySpan = new Span() // empty span -> doesn't trace
+    val configsRepo = Components.configsRepository
+    for {
+      configs <- configsRepo.findAllConfigs()
+      config <- configs
+    } {
+      val trimmed = StringUtils.strip(config.partId)
+      if (trimmed != config.partId) {
+        LTSVLogger.warn("message" -> "This partId is suspicious - it has leading/trailing spaces", "partId" -> s"'${config.partId}'")
+      }
+    }
+  }
+
+  /**
+   * Tries to set the Hystrix properties strategy to [[KeyAndBuilderValuesHystrixPropertiesStrategy]]
+   *
+   * Resist the temptation to do a HystrixPlugins.getInstance().getPropertiesStrategy first to do
+   * checking, as that actually also sets the strategy if it isn't already set.
+   */
+  def setHystrixPropertiesStrategy(): Unit = {
+    // If it's defined, we don't need to set anything
+    if (sys.props.get("hystrix.plugin.HystrixPropertiesStrategy.implementation").isEmpty) {
+      LTSVLogger.info("-Dhystrix.plugin.HystrixPropertiesStrategy.implementation is not set. Defaulting to" -> "com.m3.octoparts.hystrix.KeyAndBuilderValuesHystrixPropertiesStrategy")
+      try {
+        HystrixPlugins.getInstance().registerPropertiesStrategy(new KeyAndBuilderValuesHystrixPropertiesStrategy)
+      } catch {
+        case NonFatal(e) =>
+          val currentStrategy = HystrixPlugins.getInstance().getPropertiesStrategy.getClass
+          LTSVLogger.info(e, "Current Hystrix Properties Strategy:" -> currentStrategy)
+      }
+    }
+  }
+
 }

app/com/m3/octoparts/OctopartsMetricsRegistry.scala

@@ -0,0 +1,11 @@
+package com.m3.octoparts
+
+import com.codahale.metrics.SharedMetricRegistries
+import play.api.Play
+
+object OctopartsMetricsRegistry {
+  /**
+   * Same as [[com.kenshoo.play.metrics.MetricsRegistry.defaultRegistry]] when a Play app is running; uses the default name ("default") otherwise
+   */
+  val default = SharedMetricRegistries.getOrCreate(Play.maybeApplication.flatMap(_.configuration.getString("metrics.name")).getOrElse("default"))
+}

app/com/m3/octoparts/ZipkinServiceHolder.scala

@@ -0,0 +1,60 @@
+package com.m3.octoparts
+
+import java.net.InetAddress
+
+import com.beachape.zipkin.services.{ BraveZipkinService, NoopZipkinService, ZipkinServiceLike }
+import com.github.kristofa.brave.zipkin.ZipkinSpanCollector
+import play.api.{ Logger, Play }
+
+import scala.util.{ Failure, Success, Try }
+
+object ZipkinServiceHolder {
+
+  private implicit val app = play.api.Play.current
+  private implicit val ex = play.api.libs.concurrent.Execution.Implicits.defaultContext
+
+  val ZipkinService: ZipkinServiceLike = if (Play.isTest) {
+    NoopZipkinService
+  } else {
+    val maybeService = for {
+      zipkinHost <- Play.configuration.getString("zipkin.host")
+      zipkinPort <- Play.configuration.getInt("zipkin.port")
+      zipkinRate <- Play.configuration.getDouble("zipkin.sampleRate")
+      env <- Play.configuration.getString("application.env")
+    } yield {
+      Try {
+        val zipkinSpanCollector = new ZipkinSpanCollector(zipkinHost, zipkinPort)
+        sys.addShutdownHook(zipkinSpanCollector.close())
+        val currentHostName = InetAddress.getLocalHost.getHostName
+        val currentRunningPort = Play.configuration.getInt("http.port").getOrElse(9000)
+        new BraveZipkinService(
+          hostIp = currentHostName,
+          hostPort = currentRunningPort,
+          serviceName = s"Octoparts - $env",
+          collector = zipkinSpanCollector,
+          clientTraceFilters = Seq(_.isSetParent_id),
+          serverTraceFilters = Seq(
+            { s =>
+              val name = s.getName
+              !(name.startsWith("OPTION") || name.startsWith("GET - /assets"))
+            },
+            { s => s.isSetParent_id || (zipkinRate > scala.util.Random.nextDouble()) }
+          )
+        )
+      }
+
+    }
+    maybeService match {
+      case Some(Success(zipkinService)) => zipkinService
+      case Some(Failure(e)) => {
+        Logger.error("Could not create the Zipkin service", e)
+        NoopZipkinService
+      }
+      case None => {
+        Logger.warn("Zipkin configs are missing in the current environment, falling back to NoopZipkinService")
+        NoopZipkinService
+      }
+    }
+  }
+
+}

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

@@ -1,7 +1,9 @@
 package com.m3.octoparts.aggregator.handler
 
+import com.m3.octoparts.aggregator.PartRequestInfo
 import com.m3.octoparts.model.PartResponse
-import com.m3.octoparts.model.config._
+import com.m3.octoparts.model.config.ShortPartParam
+import com.twitter.zipkin.gen.Span
 
 import scala.concurrent.Future
 
@@ -16,10 +18,11 @@ import scala.concurrent.Future
  */
 trait Handler {
 
-  type HandlerArguments = Map[ShortPartParam, String]
+  type HandlerArguments = Map[ShortPartParam, Seq[String]]
 
   // 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)(implicit parentSpan: Span): Future[PartResponse]
+
 }

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

@@ -1,9 +1,15 @@
 package com.m3.octoparts.aggregator.handler
 
+import com.beachape.zipkin.services.ZipkinServiceLike
 import com.m3.octoparts.model.config.HttpPartConfig
 
 trait HttpHandlerFactory {
 
+  /**
+   * For tracing Http request times
+   */
+  implicit def zipkinService: ZipkinServiceLike
+
   /**
    * @param config a HttpCommandConfig entry
    * @return a handler ready to be used