CoursierJob.scala

import tv.teads.lib.conf._
import tv.teads.lib.job.{Job, JobValidation}


trait CoursierJob[T <: JobParameters] extends Job[T] with JobValidation {
  /**
   * Example cli:
   *   coursier launch `JOB` -- `SPARK_OPTS` -- `JOB_ARGS`
   *
   * where
   *   `JOB` could be "tv.teads.finance:lib-finance-jobs:0.1.0"
   *   `SPARK_OTPS` are spark options like `--class` or `--conf`, `--master`
   *   `JOB_ARGS` are args for the actual job logic, such as `--job.dryrun=true`
   */
  override def main(args: Array[String]) = {
    val idx = args.indexOf("--")
    assert(idx >= 0)

    val sparkOpts = args.take(idx)
    val jobArgs = args.drop(idx + 1)

    val jobParams = buildJobParameters(jobArgs)

    def actualPrevalidate(): Option[Seq[String]] = {
      val conf = HierarchicalConfiguration(
        List(CommandLineConfiguration(args), TypesafeConfiguration()))

      val result = prevalidateJob(conf)
      if (result.canRun) {
        Some(result.additionalJobArgs)
      } else {
        None
      }
    }

    def prevalidate(): Option[Seq[String]] =
      if (jobParams.isDryRun) {
        Some(Seq())
      } else {
        actualPrevalidate()
      }

    def classPath(): Seq[String] = {
      val b = new collection.mutable.ArrayBuffer[String]
      def helper(cl: ClassLoader): Unit =
        if (cl != null) {
          cl match {
            case u: java.net.URLClassLoader =>
              b ++= u
                .getURLs
                .filter(_.getProtocol == "file")
                .map(_.getPath)
          }
          helper(cl.getParent)
        }
      helper(Thread.currentThread().getContextClassLoader)
      b.toVector
    }

    prevalidate() match {
      case None =>
        Console.err.println(s"Nothing to process for ${this.name}")
        sys.exit(0)

      case Some(extraArgs) =>
        val mainJar: String = getClass.getProtectionDomain.getCodeSource.getLocation.getPath
        val extraJars: Seq[String] = classPath()

        org.apache.spark.deploy.SparkSubmit.main(
          sparkOpts ++
            Seq(mainJar) ++
            jobArgs ++
            extraArgs
        )
    }
  }
}