Adding init and close methods in authenticator trait (#54)

Author: ashishagg

commons/src/main/scala/com/expedia/www/haystack/collector/commons/config/ConfigurationLoader.scala

@@ -20,55 +20,104 @@ package com.expedia.www.haystack.collector.commons.config
 import java.io.File
 import java.util.Properties
 
-import com.typesafe.config.{Config, ConfigFactory}
+import com.typesafe.config.{Config, ConfigFactory, ConfigRenderOptions, ConfigValueType}
 import org.apache.kafka.clients.producer.ProducerConfig
 import org.apache.kafka.clients.producer.ProducerConfig.{KEY_SERIALIZER_CLASS_CONFIG, VALUE_SERIALIZER_CLASS_CONFIG}
 import org.apache.kafka.common.serialization.ByteArraySerializer
+import org.slf4j.LoggerFactory
 
 import scala.collection.JavaConversions._
+import scala.collection.JavaConverters._
 
 object ConfigurationLoader {
 
-  private val ENV_NAME_PREFIX = "HAYSTACK_PROP_"
+  private val LOGGER = LoggerFactory.getLogger(ConfigurationLoader.getClass)
+
+  private[haystack] val ENV_NAME_PREFIX = "HAYSTACK_PROP_"
 
   /**
     * Load and return the configuration
     * if overrides_config_path env variable exists, then we load that config file and use base conf as fallback,
     * else we load the config from env variables(prefixed with haystack) and use base conf as fallback
     *
+    * @param resourceName name of the resource file to be loaded. Default value is `config/base.conf`
+    * @param envNamePrefix env variable prefix to override config values. Default is `HAYSTACK_PROP_`
+    *
+    * @return an instance of com.typesafe.Config
     */
-  lazy val loadAppConfig: Config = {
-    val baseConfig = ConfigFactory.load("config/base.conf")
-
-    sys.env.get("HAYSTACK_OVERRIDES_CONFIG_PATH") match {
-      case Some(path) => ConfigFactory.parseFile(new File(path)).withFallback(baseConfig)
-      case _ => loadFromEnvVars().withFallback(baseConfig)
+  def loadConfigFileWithEnvOverrides(resourceName : String = "config/base.conf",
+                                     envNamePrefix : String = ENV_NAME_PREFIX) : Config = {
+
+    require(resourceName != null && resourceName.length > 0 , "resourceName is required")
+    require(envNamePrefix != null && envNamePrefix.length > 0 , "envNamePrefix is required")
+
+    val baseConfig = ConfigFactory.load(resourceName)
+
+    val keysWithArrayValues = baseConfig.entrySet()
+      .asScala
+      .filter(_.getValue.valueType() == ConfigValueType.LIST)
+      .map(_.getKey)
+      .toSet
+
+    val config = sys.env.get("HAYSTACK_OVERRIDES_CONFIG_PATH") match {
+      case Some(overrideConfigPath) =>
+        val overrideConfig = ConfigFactory.parseFile(new File(overrideConfigPath))
+        ConfigFactory
+          .parseMap(parsePropertiesFromMap(sys.env, keysWithArrayValues, envNamePrefix).asJava)
+          .withFallback(overrideConfig)
+          .withFallback(baseConfig)
+          .resolve()
+      case _ => ConfigFactory
+        .parseMap(parsePropertiesFromMap(sys.env, keysWithArrayValues, envNamePrefix).asJava)
+        .withFallback(baseConfig)
+        .resolve()
     }
+
+    // In key-value pairs that contain 'password' in the key, replace the value with asterisks
+    LOGGER.info(config.root()
+      .render(ConfigRenderOptions.defaults().setOriginComments(false))
+      .replaceAll("(?i)(\\\".*password\\\"\\s*:\\s*)\\\".+\\\"", "$1********"))
+
+    config
   }
 
   /**
-    * @return new config object with haystack specific environment variables
+    *  @return new config object with haystack specific environment variables
     */
-  private def loadFromEnvVars(): Config = {
-    val envMap = sys.env.filter {
-      case (envName, _) => isHaystackEnvVar(envName)
+  private[haystack] def parsePropertiesFromMap(envVars: Map[String, String],
+                                               keysWithArrayValues: Set[String],
+                                               envNamePrefix: String): Map[String, Object] = {
+    envVars.filter {
+      case (envName, _) => envName.startsWith(envNamePrefix)
     } map {
-      case (envName, envValue) => (transformEnvVarName(envName), envValue)
+      case (envName, envValue) =>
+        val key = transformEnvVarName(envName, envNamePrefix)
+        if (keysWithArrayValues.contains(key)) (key, transformEnvVarArrayValue(envValue)) else (key, envValue)
     }
-
-    ConfigFactory.parseMap(envMap)
   }
 
-  private def isHaystackEnvVar(env: String): Boolean = env.startsWith(ENV_NAME_PREFIX)
-
   /**
     * converts the env variable to HOCON format
     * for e.g. env variable HAYSTACK_KAFKA_STREAMS_NUM_STREAM_THREADS gets converted to kafka.streams.num.stream.threads
     * @param env environment variable name
-    * @return
+    * @return variable name that complies with hocon key
     */
-  private def transformEnvVarName(env: String): String = {
-    env.replaceFirst(ENV_NAME_PREFIX, "").toLowerCase.replace("_", ".")
+  private def transformEnvVarName(env: String, envNamePrefix: String): String = {
+    env.replaceFirst(envNamePrefix, "").toLowerCase.replace("_", ".")
+  }
+
+  /**
+    * converts the env variable value to iterable object if it starts and ends with '[' and ']' respectively.
+    * @param env environment variable value
+    * @return string or iterable object
+    */
+  private def transformEnvVarArrayValue(env: String): java.util.List[String] = {
+    if (env.startsWith("[") && env.endsWith("]")) {
+      import scala.collection.JavaConverters._
+      env.substring(1, env.length - 1).split(',').filter(str => (str != null) && str.nonEmpty).toList.asJava
+    } else {
+      throw new RuntimeException("config key is of array type, so it should start and end with '[', ']' respectively")
+    }
   }
 
   def kafkaProducerConfig(config: Config): KafkaProduceConfiguration = {
@@ -97,6 +146,5 @@ object ConfigurationLoader {
   def extractorConfiguration(config: Config): ExtractorConfiguration = {
     val extractor = config.getConfig("extractor")
     ExtractorConfiguration(outputFormat = if (extractor.hasPath("output.format")) Format.withName(extractor.getString("output.format")) else Format.PROTO)
-
   }
 }

http/src/main/scala/com/expedia/www/haystack/http/span/collector/ProjectConfiguration.scala

@@ -26,18 +26,21 @@ import scala.reflect.ClassTag
 case class HttpConfiguration(host: String = "127.0.0.1", port: Int = 8080, authenticator: Authenticator = NoopAuthenticator)
 
 object ProjectConfiguration {
-  val config: Config = ConfigurationLoader.loadAppConfig
+  val config: Config = ConfigurationLoader.loadConfigFileWithEnvOverrides()
 
   def kafkaProducerConfig(): KafkaProduceConfiguration = ConfigurationLoader.kafkaProducerConfig(config)
   def extractorConfig(): ExtractorConfiguration = ConfigurationLoader.extractorConfiguration(config)
 
-  def httpConfig(): HttpConfiguration = {
+  lazy val httpConfig: HttpConfiguration = {
     val authenticator = if(config.hasPath("http.authenticator")) {
       toInstance[Authenticator](config.getString("http.authenticator"))
     } else {
       NoopAuthenticator
     }
 
+    // initialize the
+    authenticator.init(config)
+
     HttpConfiguration(config.getString("http.host"), config.getInt("http.port"), authenticator)
   }
 

http/src/main/scala/com/expedia/www/haystack/http/span/collector/WebServer.scala

@@ -35,6 +35,7 @@ import org.slf4j.LoggerFactory
 import scala.concurrent.duration._
 import scala.concurrent.{Await, Future}
 import scala.sys._
+import scala.util.Try
 
 object WebServer extends App with MetricsSupport {
   val LOGGER = LoggerFactory.getLogger(WebServer.getClass)
@@ -45,7 +46,7 @@ object WebServer extends App with MetricsSupport {
     ProjectConfiguration.extractorConfig(),
     LoggerFactory.getLogger(classOf[ProtoSpanExtractor]))
 
-  private val http = ProjectConfiguration.httpConfig()
+  private val http = ProjectConfiguration.httpConfig
 
   // setup actor system
   implicit val system = ActorSystem("span-collector", ProjectConfiguration.config)
@@ -123,7 +124,8 @@ object WebServer extends App with MetricsSupport {
   }
 
   def close(): Unit = {
-    kafkaSink.close()
+    Try(kafkaSink.close())
+    Try(http.authenticator.close())
     materializer.shutdown()
     system.terminate()
     jmxReporter.close()

http/src/main/scala/com/expedia/www/haystack/http/span/collector/WebServer.scala.orig

@@ -17,8 +17,13 @@
 
 package com.expedia.www.haystack.http.span.collector.authenticator
 
+import java.io.Closeable
+
 import akka.http.scaladsl.model.HttpRequest
+import com.typesafe.config.Config
 
-trait Authenticator {
+trait Authenticator extends Closeable {
   def apply(req: HttpRequest): Boolean
+
+  def init(config: Config): Unit
 }

http/src/main/scala/com/expedia/www/haystack/http/span/collector/authenticator/Authenticator.scala

@@ -18,7 +18,12 @@
 package com.expedia.www.haystack.http.span.collector.authenticator
 
 import akka.http.scaladsl.model.HttpRequest
+import com.typesafe.config.Config
 
 object NoopAuthenticator extends Authenticator {
   override def apply(req: HttpRequest): Boolean = true
+
+  override def init(config: Config): Unit = ()
+
+  override def close(): Unit = ()
 }

http/src/main/scala/com/expedia/www/haystack/http/span/collector/authenticator/NoopAuthenticator.scala

@@ -28,7 +28,7 @@ import scala.concurrent.duration._
 
 object ProjectConfiguration {
 
-  private val config = ConfigurationLoader.loadAppConfig
+  private val config = ConfigurationLoader.loadConfigFileWithEnvOverrides()
 
   def healthStatusFile(): Option[String] = if(config.hasPath("health.status.path")) Some(config.getString("health.status.path")) else None