<commit_before>package com.ottogroup.bi.soda

import java.net.URLClassLoader
import java.nio.file.Paths
import java.util.Calendar
import java.util.concurrent.TimeUnit
import scala.Array.canBuildFrom
import scala.collection.mutable.HashMap
import scala.concurrent.duration.Duration
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.hadoop.security.UserGroupInformation
import org.apache.hadoop.yarn.conf.YarnConfiguration
import com.ottogroup.bi.soda.bottler.driver.FileSystemDriver.fileSystem
import com.ottogroup.bi.soda.dsl.Parameter.p
import com.typesafe.config.Config
import akka.actor.ActorSystem
import akka.actor.ExtendedActorSystem
import akka.actor.Extension
import akka.actor.ExtensionId
import akka.actor.ExtensionIdProvider
import com.ottogroup.bi.soda.dsl.views.DateParameterizationUtils
import com.ottogroup.bi.soda.bottler.driver.Driver
import com.ottogroup.bi.soda.dsl.Transformation

class SettingsImpl(val config: Config) extends Extension {

  println(config)
    
  val system = Settings.actorSystem

  val env = config.getString("soda.app.environment")

  val earliestDay = {
    val conf = config.getString("soda.app.earliestDay")
    val Array(year, month, day) = conf.split("-")
    DateParameterizationUtils.parametersToDay(p(year), p(month), p(day))
  }

  val latestDay = {
    val conf = config.getString("soda.app.latestDay")
    if (conf == "now") {
      val now = Calendar.getInstance()
      now.set(Calendar.HOUR_OF_DAY, 0)
      now.set(Calendar.MINUTE, 0)
      now.set(Calendar.SECOND, 0)
      now.set(Calendar.MILLISECOND, 0)
      now
    } else {
      val Array(year, month, day) = conf.split("-")
      DateParameterizationUtils.parametersToDay(p(year), p(month), p(day))
    }
  }

  val webserviceTimeOut: Duration =
  Duration(config.getDuration("soda.webservice.timeout", TimeUnit.MILLISECONDS),
      TimeUnit.MILLISECONDS)

  val port = config.getInt("soda.webservice.port")

  val jdbcUrl = config.getString("soda.metastore.jdbcUrl")

  val kerberosPrincipal = config.getString("soda.kerberos.principal")

  val metastoreUri = config.getString("soda.metastore.metastoreUri")

  val parsedViewAugmentorClass = config.getString("soda.app.parsedViewAugmentorClass")

  val availableTransformations = config.getObject("soda.transformations")

  val hadoopConf = new Configuration(true)

  val transformationVersioning = config.getBoolean("soda.versioning.transformations")

  val jobTrackerOrResourceManager = {
    val yarnConf = new YarnConfiguration(hadoopConf)
    if (yarnConf.get("yarn.resourcemanager.address") == null)
      config.getString("soda.hadoop.resourceManager")
    else
      yarnConf.get("yarn.resourcemanager.address")
  }

  val nameNode = if (hadoopConf.get("fs.defaultFS") == null)
    config.getString("soda.hadoop.nameNode")
  else
    hadoopConf.get("fs.defaultFS")

  val hiveActionTimeout = Duration.create(config.getDuration("soda.timeouts.hive", TimeUnit.SECONDS), TimeUnit.SECONDS)
  val oozieActionTimeout = Duration.create(config.getDuration("soda.timeouts.oozie", TimeUnit.SECONDS), TimeUnit.SECONDS)
  val fileActionTimeout = Duration.create(config.getDuration("soda.timeouts.file", TimeUnit.SECONDS), TimeUnit.SECONDS)
  val schemaActionTimeout = Duration.create(config.getDuration("soda.timeouts.schema", TimeUnit.SECONDS), TimeUnit.SECONDS)
  val dependencyTimout = Duration.create(config.getDuration("soda.timeouts.dependency", TimeUnit.SECONDS), TimeUnit.SECONDS)
  val materializeAllTimeout = Duration.create(config.getDuration("soda.timeouts.all", TimeUnit.SECONDS), TimeUnit.SECONDS)
  val retries = config.getInt("soda.action.retry")

  val userGroupInformation = {
    UserGroupInformation.setConfiguration(hadoopConf)
    val ugi = UserGroupInformation.getCurrentUser()
    ugi.setAuthenticationMethod(UserGroupInformation.AuthenticationMethod.KERBEROS)
    ugi.reloginFromKeytab();
    ugi
  }

  private val driverSettings: HashMap[String, DriverSettings] = HashMap[String, DriverSettings]()

  def getDriverSettings(d: Any with Driver[_]): DriverSettings = {
    getDriverSettings(d.transformationName)
  }

  def getDriverSettings[T <: Transformation](t: T): DriverSettings = {
    val name = t.getClass.getSimpleName.toLowerCase.replaceAll("transformation", "").replaceAll("\\$", "")
    getDriverSettings(name)
  }
  
  def getDriverSettings(n : String) : DriverSettings = {
    if (!driverSettings.contains(n)) {     
      val confName = "soda.transformations." + n
      driverSettings.put(n, new DriverSettings(config.getConfig(confName), n))
    }

    driverSettings(n)
  }
}

object Settings extends ExtensionId[SettingsImpl] with ExtensionIdProvider {
  val actorSystem = ActorSystem("soda")

  override def lookup = Settings

  override def createExtension(system: ExtendedActorSystem) =
    new SettingsImpl(system.settings.config)

  override def get(system: ActorSystem): SettingsImpl = super.get(system)
  
  def apply() = {
    super.apply(actorSystem)
  }
}

class DriverSettings(val config: Config, val name: String) {
  val location = Settings().nameNode + config.getString("location")<commit_msg>added per-transformation settings<commit_after>package com.ottogroup.bi.soda

import java.net.URLClassLoader
import java.nio.file.Paths
import java.util.Calendar
import java.util.concurrent.TimeUnit
import scala.Array.canBuildFrom
import scala.collection.mutable.HashMap
import scala.concurrent.duration.Duration
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.hadoop.security.UserGroupInformation
import org.apache.hadoop.yarn.conf.YarnConfiguration
import com.ottogroup.bi.soda.bottler.driver.FileSystemDriver.fileSystem
import com.ottogroup.bi.soda.dsl.Parameter.p
import com.typesafe.config.Config
import akka.actor.ActorSystem
import akka.actor.ExtendedActorSystem
import akka.actor.Extension
import akka.actor.ExtensionId
import akka.actor.ExtensionIdProvider
import com.ottogroup.bi.soda.dsl.views.DateParameterizationUtils
import com.ottogroup.bi.soda.bottler.driver.Driver
import com.ottogroup.bi.soda.dsl.Transformation

class SettingsImpl(val config: Config) extends Extension {

  println(config)
    
  val system = Settings.actorSystem

  val env = config.getString("soda.app.environment")

  val earliestDay = {
    val conf = config.getString("soda.app.earliestDay")
    val Array(year, month, day) = conf.split("-")
    DateParameterizationUtils.parametersToDay(p(year), p(month), p(day))
  }

  val latestDay = {
    val conf = config.getString("soda.app.latestDay")
    if (conf == "now") {
      val now = Calendar.getInstance()
      now.set(Calendar.HOUR_OF_DAY, 0)
      now.set(Calendar.MINUTE, 0)
      now.set(Calendar.SECOND, 0)
      now.set(Calendar.MILLISECOND, 0)
      now
    } else {
      val Array(year, month, day) = conf.split("-")
      DateParameterizationUtils.parametersToDay(p(year), p(month), p(day))
    }
  }

  val webserviceTimeOut: Duration =
  Duration(config.getDuration("soda.webservice.timeout", TimeUnit.MILLISECONDS),
      TimeUnit.MILLISECONDS)

  val port = config.getInt("soda.webservice.port")

  val jdbcUrl = config.getString("soda.metastore.jdbcUrl")

  val kerberosPrincipal = config.getString("soda.kerberos.principal")

  val metastoreUri = config.getString("soda.metastore.metastoreUri")

  val parsedViewAugmentorClass = config.getString("soda.app.parsedViewAugmentorClass")

  val availableTransformations = config.getObject("soda.transformations")

  val hadoopConf = new Configuration(true)

  val transformationVersioning = config.getBoolean("soda.versioning.transformations")

  val jobTrackerOrResourceManager = {
    val yarnConf = new YarnConfiguration(hadoopConf)
    if (yarnConf.get("yarn.resourcemanager.address") == null)
      config.getString("soda.hadoop.resourceManager")
    else
      yarnConf.get("yarn.resourcemanager.address")
  }

  val nameNode = if (hadoopConf.get("fs.defaultFS") == null)
    config.getString("soda.hadoop.nameNode")
  else
    hadoopConf.get("fs.defaultFS")

  val hiveActionTimeout = Duration.create(config.getDuration("soda.timeouts.hive", TimeUnit.SECONDS), TimeUnit.SECONDS)
  val oozieActionTimeout = Duration.create(config.getDuration("soda.timeouts.oozie", TimeUnit.SECONDS), TimeUnit.SECONDS)
  val fileActionTimeout = Duration.create(config.getDuration("soda.timeouts.file", TimeUnit.SECONDS), TimeUnit.SECONDS)
  val schemaActionTimeout = Duration.create(config.getDuration("soda.timeouts.schema", TimeUnit.SECONDS), TimeUnit.SECONDS)
  val dependencyTimout = Duration.create(config.getDuration("soda.timeouts.dependency", TimeUnit.SECONDS), TimeUnit.SECONDS)
  val materializeAllTimeout = Duration.create(config.getDuration("soda.timeouts.all", TimeUnit.SECONDS), TimeUnit.SECONDS)
  val retries = config.getInt("soda.action.retry")

  val userGroupInformation = {
    UserGroupInformation.setConfiguration(hadoopConf)
    val ugi = UserGroupInformation.getCurrentUser()
    ugi.setAuthenticationMethod(UserGroupInformation.AuthenticationMethod.KERBEROS)
    ugi.reloginFromKeytab();
    ugi
  }

  private val driverSettings: HashMap[String, DriverSettings] = HashMap[String, DriverSettings]()

  def getDriverSettings(d: Any with Driver[_]): DriverSettings = {
    getDriverSettings(d.transformationName)
  }

  def getDriverSettings[T <: Transformation](t: T): DriverSettings = {
    val name = t.getClass.getSimpleName.toLowerCase.replaceAll("transformation", "").replaceAll("\\$", "")
    getDriverSettings(name)
  }
  
  def getDriverSettings(n : String) : DriverSettings = {
    if (!driverSettings.contains(n)) {     
      val confName = "soda.transformations." + n
      driverSettings.put(n, new DriverSettings(config.getConfig(confName), n))
    }

    driverSettings(n)
  }
  
  def getTransformationSetting(typ: String, setting: String) = {
    val confName = s"soda.transformations.${typ}.transformation.${setting}"
    config.getString(confName)
  }
  
}

object Settings extends ExtensionId[SettingsImpl] with ExtensionIdProvider {
  val actorSystem = ActorSystem("soda")

  override def lookup = Settings

  override def createExtension(system: ExtendedActorSystem) =
    new SettingsImpl(system.settings.config)

  override def get(system: ActorSystem): SettingsImpl = super.get(system)
  
  def apply() = {
    super.apply(actorSystem)
  }
}

class DriverSettings(val config: Config, val name: String) {
  val location = Settings().nameNode + config.getString("location")
{'fraction_non_alphanumeric': 0.11397338403041825, 'fraction_numerical': 0.0007604562737642585, 'mean_word_length': 4.9106741573033705, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2975200', 'n_tokens_mistral': 3301, 'n_tokens_neox': 3230, 'n_words': 554}