Dynamic Jenkins Job Scheduler

scheduler.groovy

// Schedule builds TeamCity style: Given a dependency relationship between builds,
// start all builds whose prerequisite builds have completed.

// This is intended to be run as a system groovy script in Jenkins.

// Unbound variables:
//   sleepSeconds - amount of time to sleep between polls. 5-15 seconds seem good.

// Every entry in the jobToRun map is expected to have this format:
//  JobId: [ JobName: <actual jenkins name of the job>,
//           WaitFor: [ <list of JobIds required to finish prior to launching this one> ],
//           Parameters: [ <key/value map of parameters> ]
//         ]
// Note that the same jenkins job can be invoked in many different ways using
// different parameters, hence the need for a JobId and specific tests that the parameters
// are what we want when looking for a build of a job in Jenkins.

// Normally, you would inject this or read this from a file.
def jobsToRun = [
  'dummy-job(1)': [ 'JobName': 'dummy-job',
                    'WaitFor': [],
                    'Parameters': [ 'Key': '1' ] ],
  'dummy-job(2)': [ 'JobName': 'dummy-job',
                    'WaitFor': [],
                    'Parameters': [ 'Key': '2' ] ],
  'dummy-job(3)': [ 'JobName': 'dummy-job',
                    'WaitFor': [],
                    'Parameters': [ 'Key': '3' ] ],
]

// These will fill up as we empty out jobsToRun...
def jobsQueued = [:]
def jobsFinished = [:]

// sleepSeconds are expected to be injected
def sleepPeriod = sleepSeconds.toInteger()*1000
def jlc = new jenkins.model.JenkinsLocationConfiguration()
def jenkinsUrl = jlc.url

while (jobsToRun || jobsQueued) {
  def refresh = false
  def queuable = false
  def jobsQueuable = jobsToRun.findAll { jobId, jobData ->
    jobData['WaitFor'].every { jobsFinished[it] }
  }

  jobsQueuable.each { jobId, jobData ->
    // Queue the job
    Thread.sleep(10)
    queuable = true
    if (!jobData.containsKey('JobProject')) {
      jobData['JobProject'] = hudson.model.Hudson.instance.getJob(jobData['JobName'])
    }
    def job = jobData['JobProject']
    if (job) {
      def params = jobData['Parameters'].collect { key, val ->
        new hudson.model.StringParameterValue(key, val)
      }
      def paramsAction = new hudson.model.ParametersAction(params)
      def cause = new hudson.model.Cause.UpstreamCause(build)
      def causeAction = new hudson.model.CauseAction(cause)

      jobData['Future'] = job.scheduleBuild2(0, causeAction, paramsAction)

      if (jobData['Future']) {
        // Shift job into Queued map
        println(" -> Queuing ${jobId}")
        jobsQueued[jobId] = jobData
        jobsToRun.remove(jobId)
        refresh = true
      } else {
        println(" -> Unable to schedule ${jobId} ${causeAction} ${paramsAction}")
      }
    } else {
      println(" -> Unable to retrieve job object for ${jobId}")
    }
  }

  if (jobsToRun && !jobsQueued && !queuable) {
    // Cyclic dependencies may cause this:
    println("No more jobs queued, but unable to queue new jobs")
    jobsToRun.each { jobId, jobData ->
      println("Job ${jobId} is waiting for:")
      jobData['WaitFor'].each { println("  ${it}") }
    }
    build.setResult(hudson.model.Result.FAILURE)
    jobsToRun = [:]
  }

  // sleeping here allows new jobs to be launched immediately after
  // finished jobs being detected
  try {
    Thread.sleep(sleepPeriod)
  } catch(e) {
    if (e in InterruptedException) {
      jobsToRun = [:]
      build.setResult(hudson.model.Result.ABORTED)
      jobsQueued.each { jobId, jobData ->
        jobData['Cancel'] = true
      }
      // Iterate a little bit faster once we detected a cancel
      sleepPeriod = 1000
    } else {
      throw(e)
    }
  }

  def jobsWithResult = jobsQueued.findAll { jobId, jobData ->
    // Check if done by finding a build which has a result, started after this job
    // and has all of our build parameters - doing this with a classic for loop
    // relying on the builds being returned in reverse chronological order to
    // fail fast once we hit a job launched prior to the scheduler job
    for (def candidate in jobData['JobProject'].builds) {
      if (candidate.startTimeInMillis < build.startTimeInMillis) return false  // bail on old builds
      Thread.sleep(10)
      if (jobData['Parameters'].every { key, val -> candidate.buildVariables[key] == val } ) {
        if (!jobData['Build']) {
          jobData['Build'] = candidate  // side effect, I know ....
          refresh = true
        }
        jobData['Result'] = candidate.result
        return candidate.result ? true : false  // bail when we found something
      }
    }
    return false  // nothing found
  }

  jobsWithResult.each { jobId, jobData ->
    println(" -> Finished ${jobId} (${jobData['Result']}) ${jenkinsUrl}${jobData['Build'].url}")
    // Shift job into Finished map
    jobsFinished[jobId] = jobData
    jobsQueued.remove(jobId)
    refresh = true
  }

  def jobsWithCancel = jobsQueued.findAll { jobId, jobData ->
    jobData['Cancel'] && !jobData['Result']
  }

  jobsWithCancel.each { jobId, jobData ->
    if (jobData['Build']) {
      println(" -> Stopping ${jobId}...")
      jobData['Build'].doStop()
      jobData['Cancel'] = false
    } else if (jobData['Unqueued'])  {
      // We performed the "jobData['Future'].cancel(true)" in the previous iteration
      // so that must be good.
      println(" -> Unqueue confirmed: ${jobId}")
      jobsFinished[jobId] = jobData
      jobsQueued.remove(jobId)
      refresh = true
    } else {
      // Attempt to unqueue, the next iteration will either see it as a running uncancelled
      // job, or as a job marked with 'Unqueued'
      jobData['Future'].cancel(true)
      println(" -> Unqueuing ${jobId}...")
      jobData['Unqueued'] = true
    }
  }

  if (refresh) {
    def summary = []
    if (jobsFinished) {
      summary << "completed: ${jobsFinished.size()}"
    }
    if (jobsQueued) {
      summary << "active: ${jobsQueued.size()}"
    }
    if (jobsToRun) {
      summary << "pending: ${jobsToRun.size()}"
    }
    build.description = 'Jobs '+summary.join(', ')
    if (jobsToRun) {
      println "==== Pending Jobs ===="
      jobsToRun.each { jobId, jobData ->
        println("${jobId}: Waits for:")
        jobData['WaitFor'].each { println("  ${it}") }
      }
    }
    if (jobsQueued) {
      println "==== Queued Jobs ===="
      jobsQueued.each { jobId, jobData ->
        if (jobData['Build']) {
          println("${jobId}: Running ${jenkinsUrl}${jobData['Build'].url}")
        } else {
          println("${jobId}: Queued")
        }
      }
      println "====================="
    } else {
      println "===== All Done ======"
    }
  }
}