dataflow_job.tf

locals {
  foo_directory = "../../../../services/foo/target"
  jobName        = "company-${var.environment}-datalayer-foo-app-v1"
  region         = "europe-west1"
  //parameters for the java jar 
  dataflow_parameters = { 
    runner                = "DataflowRunner"
    jobName               = local.jobName
    update                = data.external.running_job.result.name == local.jobName ? true : false
    project               = var.project
    kafkaInputEnabled     = "true"
    serviceAccount        = module.dataflow_job.service_account.email
    region                = local.region
    zone                  = "${local.region}-b"
    workerMachineType     = "n1-standard-1"
    diskSizeGb            = "30"
    maxWorkers            = "1" 
    streaming             = "true"
    gcpTempLocation       = "gs://${google_storage_bucket.default.name}/${google_storage_bucket_object.temp_subfolder.name}"
    stagingLocation       = "gs://${google_storage_bucket.default.name}/${google_storage_bucket_object.staging_subfolder.name}"
    pubSubFailureTopic    = google_pubsub_topic.failure_topic.name
    deadLetterDatasetName = google_bigquery_dataset.dead_letter_dataset.id
    env                   = var.environment
    jobOwner              = "DeLorean"
  }

  //  turns the map above into a list of parameters to the java process
  parameters_string = join(" ", formatlist("--%s=%s", keys(local.dataflow_parameters), values(local.dataflow_parameters)))
}

data "external" "running_job" {
  working_dir = local.foo_directory
  //check for existing job 
  program     = ["bash", "-c", "gcloud dataflow jobs list --status active --filter='name=${local.jobName}' --format=json | jq '.[0]'"]
}


resource "null_resource" "deploy_job" {
  triggers = { 
    jar_hash = filesha256("${local.foo_directory}/foo-bundled-0.0.2.jar")
  }

  provisioner "local-exec" {
    command     = "java -cp foo-bundled-0.0.2.jar com.company.datalayer.pipelines.MetlGrid ${local.parameters_string}"
    working_dir = local.foo_directory
  }
}