Orchestrator rolling updates with job definiton (#564)

Co-authored-by: Tomas Valenta <[email protected]>

Author: jakubno

packages/nomad/main.tf

@@ -317,8 +317,9 @@ data "external" "orchestrator_checksum" {
   }
 }
 
-resource "nomad_job" "orchestrator" {
-  jobspec = templatefile("${path.module}/orchestrator.hcl", {
+
+locals {
+  orchestrator_envs = {
     gcp_zone         = var.gcp_zone
     port             = var.orchestrator_port
     proxy_port       = var.orchestrator_proxy_port
@@ -336,7 +337,45 @@ resource "nomad_job" "orchestrator" {
     clickhouse_username          = var.clickhouse_username
     clickhouse_password          = var.clickhouse_password
     clickhouse_database          = var.clickhouse_database
-  })
+  }
+
+  orchestrator_job_check = templatefile("${path.module}/orchestrator.hcl", merge(
+    local.orchestrator_envs,
+    {
+      latest_orchestrator_job_id = "placeholder",
+    }
+  ))
+}
+
+
+
+resource "random_id" "orchestrator_job" {
+  keepers = {
+    # Use both the orchestrator job (including vars) definition and the latest orchestrator checksum to detect changes
+    orchestrator_job = sha256("${local.orchestrator_job_check}-${data.external.orchestrator_checksum.result.hex}")
+  }
+
+  byte_length = 8
+}
+
+resource "nomad_variable" "orchestrator_hash" {
+  path = "nomad/jobs"
+  items = {
+    latest_orchestrator_job_id = random_id.orchestrator_job.hex
+  }
+}
+
+resource "nomad_job" "orchestrator" {
+  deregister_on_id_change = false
+
+  jobspec = templatefile("${path.module}/orchestrator.hcl", merge(
+    local.orchestrator_envs,
+    {
+      latest_orchestrator_job_id = random_id.orchestrator_job.hex
+    }
+  ))
+
+  depends_on = [nomad_variable.orchestrator_hash]
 }
 
 data "google_storage_bucket_object" "template_manager" {

packages/nomad/orchestrator.hcl

@@ -1,16 +1,10 @@
-job "orchestrator" {
+job "orchestrator-${latest_orchestrator_job_id}" {
   type = "system"
-  datacenters = ["${gcp_zone}"]
+  node_pool = "default"
 
   priority = 90
 
   group "client-orchestrator" {
-    network {
-      port "orchestrator" {
-        static = "${port}"
-      }
-    }
-
     service {
       name = "orchestrator"
       port = "${port}"
@@ -30,9 +24,42 @@ job "orchestrator" {
       port = "${proxy_port}"
     }
 
+    task "check-placement" {
+      driver = "raw_exec"
+
+      lifecycle {
+        hook = "prestart"
+        sidecar = false
+      }
+
+      restart {
+        attempts = 0
+      }
+
+      template {
+        destination = "local/check-placement.sh"
+        data = <<EOT
+#!/bin/bash
+
+if [ "{{with nomadVar "nomad/jobs" }}{{ .latest_orchestrator_job_id }}{{ end }}" != "${latest_orchestrator_job_id}" ]; then
+  echo "This orchestrator is not the latest version, exiting"
+  exit 1
+fi
+EOT
+      }
+      
+      config {
+        command = "local/check-placement.sh"
+      }
+    }
+
     task "start" {
       driver = "raw_exec"
 
+      restart {
+        attempts = 0
+      }
+
       env {
         NODE_ID                      = "$${node.unique.name}"
         CONSUL_TOKEN                 = "${consul_acl_token}"
@@ -50,7 +77,7 @@ job "orchestrator" {
 
       config {
         command = "/bin/bash"
-        args    = ["-c", " chmod +x local/orchestrator && local/orchestrator --port ${port} --proxy-port ${proxy_port}"]
+        args    = ["-c", " chmod +x local/orchestrator && local/orchestrator --port ${port} --proxy-port ${proxy_port} --wait 0"]
       }
 
       artifact {

packages/orchestrator/Makefile

@@ -46,7 +46,7 @@ run-debug:
 	ORCHESTRATOR_SERVICES=$(ORCHESTRATOR_SERVICES) \
 	GCP_DOCKER_REPOSITORY_NAME=$(GCP_DOCKER_REPOSITORY_NAME) \
 	GOOGLE_SERVICE_ACCOUNT_BASE64=$(GOOGLE_SERVICE_ACCOUNT_BASE64) \
-	./bin/orchestrator
+	./bin/orchestrator --wait 0
 
 .PHONY: upload/orchestrator
 upload/orchestrator:

packages/orchestrator/main.go

@@ -12,6 +12,7 @@ import (
 	"os/signal"
 	"slices"
 	"syscall"
+	"time"
 
 	"go.opentelemetry.io/otel"
 	"go.uber.org/zap"
@@ -41,6 +42,7 @@ type Closeable interface {
 const (
 	defaultPort      = 5008
 	defaultProxyPort = 5007
+	defaultWait      = 30
 
 	version = "0.1.0"
 
@@ -53,6 +55,7 @@ var commitSHA string
 func main() {
 	port := flag.Uint("port", defaultPort, "orchestrator server port")
 	proxyPort := flag.Uint("proxy-port", defaultProxyPort, "orchestrator proxy port")
+	wait := flag.Uint("wait", defaultWait, "orchestrator proxy port")
 	flag.Parse()
 
 	if *port > math.MaxUint16 {
@@ -63,6 +66,12 @@ func main() {
 		log.Fatalf("%d is larger than maximum possible proxy port %d", proxyPort, math.MaxInt16)
 	}
 
+	// TODO: Remove after the orchestrator is fully migrated to the new job definition
+	if *wait > 0 {
+		log.Printf("waiting %d seconds before starting orchestrator", *wait)
+		time.Sleep(time.Duration(*wait) * time.Second)
+	}
+
 	success := run(*port, *proxyPort)
 
 	log.Println("Stopping orchestrator, success:", success)