Config parsing improvements (close #1252)

modules/common-transformer-stream/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/transformer/stream/common/CliConfig.scala

@@ -15,17 +15,16 @@ package com.snowplowanalytics.snowplow.rdbloader.transformer.stream.common
 
 import cats.data.EitherT
 import cats.effect.Sync
-
-import com.snowplowanalytics.snowplow.rdbloader.common.config.TransformerCliConfig
+import cats.implicits._
+import com.snowplowanalytics.snowplow.rdbloader.common.config.{ConfigUtils, TransformerCliConfig}
 
 object CliConfig {
 
-  implicit def configParsable[F[_]: Sync]: TransformerCliConfig.Parsable[F, Config] =
-    new TransformerCliConfig.Parsable[F, Config] {
-      def fromString(conf: String): EitherT[F, String, Config] =
-        Config.fromString(conf)
-    }
-
   def loadConfigFrom[F[_]: Sync](name: String, description: String)(args: Seq[String]): EitherT[F, String, CliConfig] =
-    TransformerCliConfig.loadConfigFrom[F, Config](name, description, args)
+    for {
+      raw <- EitherT.fromEither[F](TransformerCliConfig.command(name, description).parse(args).leftMap(_.show))
+      appConfig <- Config.parse(raw.config)
+      resolverConfig <- ConfigUtils.parseJsonF[F](raw.igluConfig)
+      duplicatesStorageConfig <- raw.duplicateStorageConfig.traverse(d => ConfigUtils.parseJsonF(d))
+    } yield TransformerCliConfig(resolverConfig, duplicatesStorageConfig, appConfig)
 }

modules/common-transformer-stream/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/transformer/stream/common/Config.scala

@@ -13,16 +13,14 @@
 package com.snowplowanalytics.snowplow.rdbloader.transformer.stream.common
 
 import java.net.URI
-
 import cats.implicits._
 import cats.data.EitherT
 import cats.effect.Sync
-
+import com.snowplowanalytics.snowplow.rdbloader.common.config.args.HoconOrPath
 import io.circe._
 import io.circe.generic.semiauto._
 
 import scala.concurrent.duration.{Duration, FiniteDuration}
-
 import com.snowplowanalytics.snowplow.rdbloader.common.telemetry.Telemetry
 import com.snowplowanalytics.snowplow.rdbloader.common.config.{ConfigUtils, TransformerConfig}
 import com.snowplowanalytics.snowplow.rdbloader.common.config.implicits._
@@ -45,13 +43,13 @@ final case class Config(
 
 object Config {
 
-  def fromString[F[_]: Sync](conf: String): EitherT[F, String, Config] =
-    fromString(conf, impureDecoders)
+  def parse[F[_]: Sync](config: HoconOrPath): EitherT[F, String, Config] =
+    parse(config, impureDecoders)
 
-  def fromString[F[_]: Sync](conf: String, decoders: Decoders): EitherT[F, String, Config] = {
+  def parse[F[_]: Sync](config: HoconOrPath, decoders: Decoders): EitherT[F, String, Config] = {
     import decoders._
     for {
-      config <- ConfigUtils.fromStringF[F, Config](conf)
+      config <- ConfigUtils.parseAppConfigF[F, Config](config)
       _ <- EitherT.fromEither[F](TransformerConfig.formatsCheck(config.formats))
     } yield config
   }

modules/common-transformer-stream/src/test/scala/com/snowplowanalytics/snowplow/rdbloader/transformer/stream/common/ConfigUtils.scala

@@ -12,29 +12,28 @@
  */
 package com.snowplowanalytics.snowplow.rdbloader.transformer.stream.common
 
-import java.nio.file.{Files, Paths}
 import cats.effect.IO
-import io.circe.Decoder
-import com.snowplowanalytics.snowplow.rdbloader.common.config.Region
+import cats.effect.unsafe.implicits.global
 import com.snowplowanalytics.snowplow.rdbloader.common.RegionSpec
+import com.snowplowanalytics.snowplow.rdbloader.common.config.args.HoconOrPath
+import com.snowplowanalytics.snowplow.rdbloader.common.config.Region
+import io.circe.Decoder
 
-import cats.effect.unsafe.implicits.global
+import java.nio.file.{Path, Paths}
 
 object ConfigUtils {
-  def getConfig[A](confPath: String, parse: String => Either[String, A]): Either[String, A] =
-    parse(readResource(confPath))
+  def getConfigFromResource[A](resourcePath: String, parse: HoconOrPath => Either[String, A]): Either[String, A] =
+    parse(Right(pathOf(resourcePath)))
 
-  def readResource(resourcePath: String): String = {
-    val configExamplePath = Paths.get(getClass.getResource(resourcePath).toURI)
-    Files.readString(configExamplePath)
-  }
+  def pathOf(resource: String): Path =
+    Paths.get(getClass.getResource(resource).toURI)
 
   def testDecoders: Config.Decoders = new Config.Decoders {
     implicit def regionDecoder: Decoder[Region] =
       RegionSpec.testRegionConfigDecoder
   }
 
-  def testParseStreamConfig(conf: String): Either[String, Config] =
-    Config.fromString[IO](conf, testDecoders).value.unsafeRunSync()
+  def testParseStreamConfig(config: HoconOrPath): Either[String, Config] =
+    Config.parse[IO](config, testDecoders).value.unsafeRunSync()
 
 }

modules/common/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/common/config/ConfigUtils.scala

@@ -14,84 +14,72 @@
  */
 package com.snowplowanalytics.snowplow.rdbloader.common.config
 
-import java.nio.charset.StandardCharsets.UTF_8
-import java.util.Base64
-
-import cats.Id
+import cats.data.EitherT
 import cats.effect.Sync
-import cats.data.{EitherT, ValidatedNel}
 import cats.syntax.either._
 import cats.syntax.show._
-
-import io.circe.parser.parse
+import com.snowplowanalytics.snowplow.rdbloader.common.config.args._
+import com.typesafe.config.{Config => TypesafeConfig, ConfigFactory}
 import io.circe._
+import io.circe.config.syntax.CirceConfigOps
 
-import pureconfig.module.circe._
-import pureconfig._
-import pureconfig.error._
-
-import com.snowplowanalytics.iglu.client.Resolver
+import java.nio.file.{Files, Path}
+import scala.jdk.CollectionConverters._
 
 object ConfigUtils {
-  private val Base64Decoder = Base64.getDecoder
 
-  def fromStringF[F[_]: Sync, A](conf: String)(implicit d: Decoder[A]): EitherT[F, String, A] =
-    EitherT(Sync[F].delay(fromString(conf)))
+  def parseAppConfigF[F[_]: Sync, A: Decoder](in: HoconOrPath): EitherT[F, String, A] =
+    EitherT(Sync[F].delay(parseAppConfig(in)))
 
-  def fromString[A](conf: String)(implicit d: Decoder[A]): Either[String, A] =
-    Either
-      .catchNonFatal {
-        val source = ConfigSource.string(conf)
-        namespaced(
-          ConfigSource.default(namespaced(source.withFallback(namespaced(ConfigSource.default))))
-        )
-      }
-      .leftMap(error => ConfigReaderFailures(CannotParse(s"Not valid HOCON. ${error.getMessage}", None)))
-      .flatMap { config =>
-        config
-          .load[Json]
-          .flatMap { json =>
-            json.as[A].leftMap(failure => ConfigReaderFailures(CannotParse(failure.show, None)))
-          }
-      }
-      .leftMap(_.prettyPrint())
+  def parseJsonF[F[_]: Sync](in: HoconOrPath): EitherT[F, String, Json] =
+    EitherT(Sync[F].delay(parseJson(in)))
+
+  def parseAppConfig[A: Decoder](in: HoconOrPath): Either[String, A] =
+    parseHoconOrPath(in, appConfigFallbacks)
+
+  def parseJson(in: HoconOrPath): Either[String, Json] =
+    parseHoconOrPath[Json](in, identity)
 
-  def base64decode(str: String): Either[String, String] =
+  def hoconFromString(str: String): Either[String, DecodedHocon] =
     Either
-      .catchOnly[IllegalArgumentException](Base64Decoder.decode(str))
-      .map(arr => new String(arr, UTF_8))
+      .catchNonFatal(DecodedHocon(ConfigFactory.parseString(str)))
       .leftMap(_.getMessage)
 
-  object Base64Json {
-    def decode(str: String): ValidatedNel[String, Json] =
-      base64decode(str)
-        .flatMap(str => parse(str).leftMap(_.show))
-        .toValidatedNel
+  private def parseHoconOrPath[A: Decoder](
+    config: HoconOrPath,
+    fallbacks: TypesafeConfig => TypesafeConfig
+  ): Either[String, A] =
+    config match {
+      case Left(hocon) =>
+        resolve[A](hocon, fallbacks)
+      case Right(path) =>
+        readTextFrom(path)
+          .flatMap(hoconFromString)
+          .flatMap(hocon => resolve[A](hocon, fallbacks))
+    }
+
+  private def resolve[A: Decoder](hocon: DecodedHocon, fallbacks: TypesafeConfig => TypesafeConfig): Either[String, A] = {
+    val either = for {
+      resolved <- Either.catchNonFatal(hocon.value.resolve()).leftMap(_.getMessage)
+      merged <- Either.catchNonFatal(fallbacks(resolved)).leftMap(_.getMessage)
+      parsed <- merged.as[A].leftMap(_.show)
+    } yield parsed
+    either.leftMap(e => s"Cannot resolve config: $e")
   }
 
-  /**
-   * Optionally give precedence to configs wrapped in a "snowplow" block. To help avoid polluting
-   * config namespace
-   */
-  private def namespaced(configObjSource: ConfigObjectSource): ConfigObjectSource =
-    ConfigObjectSource {
-      for {
-        configObj <- configObjSource.value()
-        conf = configObj.toConfig
-      } yield
-        if (conf.hasPath(Namespace))
-          conf.getConfig(Namespace).withFallback(conf.withoutPath(Namespace))
-        else
-          conf
-    }
+  private def readTextFrom(path: Path): Either[String, String] =
+    Either
+      .catchNonFatal(Files.readAllLines(path).asScala.mkString("\n"))
+      .leftMap(e => s"Error reading ${path.toAbsolutePath} file from filesystem: ${e.getMessage}")
 
-  def validateResolverJson(resolverJson: Json): ValidatedNel[String, Json] =
-    Resolver
-      .parse[Id](resolverJson)
-      .leftMap(_.show)
-      .map(_ => resolverJson)
-      .toValidatedNel
+  private def appConfigFallbacks(config: TypesafeConfig): TypesafeConfig =
+    namespaced(ConfigFactory.load(namespaced(config.withFallback(namespaced(ConfigFactory.load())))))
 
-  // Used as an option prefix when reading system properties.
-  val Namespace = "snowplow"
+  private def namespaced(config: TypesafeConfig): TypesafeConfig = {
+    val namespace = "snowplow"
+    if (config.hasPath(namespace))
+      config.getConfig(namespace).withFallback(config.withoutPath(namespace))
+    else
+      config
+  }
 }

modules/common/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/common/config/TransformerCliConfig.scala

@@ -13,14 +13,10 @@
  */
 package com.snowplowanalytics.snowplow.rdbloader.common.config
 
-import cats.Monad
 import cats.implicits._
-import cats.data.EitherT
-
 import io.circe.Json
-
 import com.monovore.decline.{Command, Opts}
-
+import com.snowplowanalytics.snowplow.rdbloader.common.config.args.HoconOrPath
 import com.snowplowanalytics.snowplow.rdbloader.generated.BuildInfo
 
 case class TransformerCliConfig[C](
@@ -30,29 +26,21 @@ case class TransformerCliConfig[C](
 )
 object TransformerCliConfig {
 
-  trait Parsable[F[_], C] {
-    def fromString(conf: String): EitherT[F, String, C]
-  }
-
   case class RawConfig(
-    igluConfig: Json,
-    duplicateStorageConfig: Option[Json],
-    config: String
+    igluConfig: HoconOrPath,
+    duplicateStorageConfig: Option[HoconOrPath],
+    config: HoconOrPath
   )
 
   val igluConfigOpt = Opts
-    .option[String]("iglu-config", "Base64-encoded Iglu Client JSON config", metavar = "<base64>")
-    .mapValidated(ConfigUtils.Base64Json.decode)
-    .mapValidated(ConfigUtils.validateResolverJson)
+    .option[HoconOrPath]("iglu-config", "Base64-encoded Iglu Client HOCON config", metavar = "resolver.hocon")
 
   val duplicatesOpt = Opts
-    .option[String]("duplicate-storage-config", "Base64-encoded Events Manifest JSON config", metavar = "<base64>")
-    .mapValidated(ConfigUtils.Base64Json.decode)
+    .option[HoconOrPath]("duplicate-storage-config", "Base64-encoded Events Manifest JSON config", metavar = "<base64>")
     .orNone
 
-  def configOpt = Opts
-    .option[String]("config", "base64-encoded config HOCON", "c", "config.hocon")
-    .mapValidated(x => ConfigUtils.base64decode(x).toValidatedNel)
+  val configOpt = Opts
+    .option[HoconOrPath]("config", "base64-encoded config HOCON", "c", "config.hocon")
 
   def rawConfigOpt: Opts[RawConfig] =
     (igluConfigOpt, duplicatesOpt, configOpt).mapN { (iglu, dupeStorage, target) =>
@@ -61,17 +49,4 @@ object TransformerCliConfig {
 
   def command(name: String, description: String): Command[RawConfig] =
     Command(s"$name-${BuildInfo.version}", description)(rawConfigOpt)
-
-  def loadConfigFrom[F[_], C](
-    name: String,
-    description: String,
-    args: Seq[String]
-  )(implicit M: Monad[F],
-    P: Parsable[F, C]
-  ): EitherT[F, String, TransformerCliConfig[C]] =
-    for {
-      raw <- EitherT.fromEither[F](command(name, description).parse(args).leftMap(_.show))
-      conf <- P.fromString(raw.config)
-    } yield TransformerCliConfig(raw.igluConfig, raw.duplicateStorageConfig, conf)
-
 }

modules/common/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/common/config/args.scala

@@ -0,0 +1,59 @@
+/*
+ * Copyright (c) 2012-2023 Snowplow Analytics Ltd. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at
+ * http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and
+ * limitations there under.
+ */
+package com.snowplowanalytics.snowplow.rdbloader.common.config
+import cats.data.{NonEmptyList, ValidatedNel}
+import cats.implicits._
+import com.monovore.decline.Argument
+import com.typesafe.config.{Config => TypesafeConfig}
+
+import java.nio.charset.StandardCharsets
+import java.nio.file.Path
+import java.util.Base64
+
+object args {
+
+  final case class DecodedHocon(value: TypesafeConfig) extends AnyVal
+
+  type HoconOrPath = Either[DecodedHocon, Path]
+
+  implicit val hoconOrPathArg: Argument[HoconOrPath] =
+    new Argument[HoconOrPath] {
+      def read(string: String): ValidatedNel[String, HoconOrPath] = {
+        val hocon = Argument[DecodedHocon].read(string).map(_.asLeft)
+        val path = Argument[Path].read(string).map(_.asRight)
+        val error = show"Value $string cannot be parsed as Base64 hocon neither as FS path"
+        hocon.orElse(path).leftMap(_ => NonEmptyList.one(error))
+      }
+
+      def defaultMetavar: String = "input"
+    }
+
+  implicit val decodedHoconArg: Argument[DecodedHocon] =
+    new Argument[DecodedHocon] {
+      def read(string: String): ValidatedNel[String, DecodedHocon] =
+        tryToDecodeString(string)
+          .leftMap(_.getMessage)
+          .flatMap(ConfigUtils.hoconFromString)
+          .toValidatedNel
+
+      def defaultMetavar: String = "base64"
+    }
+
+  private def tryToDecodeString(string: String): Either[IllegalArgumentException, String] =
+    Either
+      .catchOnly[IllegalArgumentException](Base64.getDecoder.decode(string))
+      .map(bytes => new String(bytes, StandardCharsets.UTF_8))
+
+}