Feature/368 scalable scheduler (#370)

* Add ComputeInstance repository

* Rename ComputeInstance -> SchedulerInstance

* Rename ping -> heartbeat

* Add scheduler instance id to workflow

* Implement workflow balancing

* Fix suddenly failing test

* Add tests

* Add integration test

* Add logging

* Only process sensors of assigned workflows

* Fix Thread sleep in the wrong block

* Only select dags from assigned workflows

* Fix test

* Add more logging

* Fixes

* Only consider active instances for rebalancing

* Fix instancesSteady check

* Delete deactivated instances

* Tidy up PR

* Rename drop -> release

* Add FK to workflow table, don't change scheduler instance id on updating workflow

* Guard assignWorkflows against concurrent execution

* Merge branch 'develop' into feature/368-scalable-scheduler

* Use liquibase

* Modify changeset version 0.3.3 to 0.3.4

Author: kevinwallimann

README.md

@@ -81,11 +81,13 @@ appUniqueId=
 # Core properties.
 # How many threads to use for each part of the "scheduler".
 # Heart beat interval in milliseconds.
+# Lag threshold, before instance is deactivated by another instance.
 scheduler.thread.pool.size=10
 scheduler.sensors.thread.pool.size=20
 scheduler.executors.thread.pool.size=30
 scheduler.jobs.parallel.number=100
 scheduler.heart.beat=5000
+scheduler.lag.threshold=20000
 ```
 ```
 #Kafka sensor properties. Not all are required. Adjust according to your use case.

src/main/resources/application.properties

@@ -48,6 +48,7 @@ scheduler.sensors.thread.pool.size=20
 scheduler.executors.thread.pool.size=30
 scheduler.jobs.parallel.number=100
 scheduler.heart.beat=5000
+scheduler.lag.threshold=20000
 
 #Kafka sensor properties.
 kafkaSource.group.id=hyper_drive_${appUniqueId}

src/main/resources/db_scripts/db_script_latest.sql

@@ -19,6 +19,7 @@ create table "workflow" (
   "project" VARCHAR NOT NULL,
   "created" TIMESTAMP NOT NULL,
   "updated" TIMESTAMP,
+  "scheduler_instance_id" BIGINT,
   "id" BIGSERIAL NOT NULL PRIMARY KEY
 );
 
@@ -98,6 +99,12 @@ create table "job_template" (
   "form_config" VARCHAR NOT NULL DEFAULT 'unknown'
 );
 
+create table "scheduler_instance" (
+  "id" BIGSERIAL NOT NULL PRIMARY KEY,
+  "status" VARCHAR NOT NULL,
+  "last_heartbeat" TIMESTAMP NOT NULL
+);
+
 alter table "job_instance"
   add constraint "job_instance_dag_instance_fk"
   foreign key("dag_instance_id")
@@ -146,6 +153,12 @@ alter table "dag_instance"
   references "workflow"("id")
   on update NO ACTION on delete NO ACTION;
 
+alter table "workflow"
+  add constraint "workflow_scheduler_instance_fk"
+  foreign key("scheduler_instance_id")
+  references "scheduler_instance"("id")
+  on update NO ACTION on delete NO ACTION;
+
 create view "dag_run_view" AS
 select
     dag_instance.id as "id",
@@ -174,3 +187,5 @@ values ('Generic Shell Job', 'Shell', 'Shell', '{}', '{}', '{}');
 
 CREATE INDEX job_instance_dag_instance_idx ON job_instance (dag_instance_id);
 CREATE INDEX dag_instance_workflow_id_idx ON dag_instance (workflow_id);
+CREATE INDEX workflow_scheduler_inst_id_idx ON workflow (scheduler_instance_id);
+

src/main/resources/db_scripts/liquibase/db.changelog.yml

@@ -44,3 +44,7 @@ databaseChangeLog:
   - include:
       relativeToChangelogFile: true
       file: v0.3.3.add-indexes.yml
+  - include:
+      relativeToChangelogFile: true
+      file: v0.3.4.add-scheduler-instances.yml
+

src/main/resources/db_scripts/liquibase/v0.3.4.add-scheduler-instances.sql

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2018 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+create table "scheduler_instance" (
+  "id" BIGSERIAL NOT NULL PRIMARY KEY,
+  "status" VARCHAR NOT NULL,
+  "last_heartbeat" TIMESTAMP NOT NULL
+);
+
+alter table "workflow" add column "scheduler_instance_id" BIGINT;
+
+alter table "workflow"
+  add constraint "workflow_scheduler_instance_fk"
+  foreign key("scheduler_instance_id")
+  references "scheduler_instance"("id")
+  on update NO ACTION on delete NO ACTION;
+
+CREATE INDEX workflow_scheduler_inst_id_idx ON workflow (scheduler_instance_id);

src/main/resources/db_scripts/liquibase/v0.3.4.add-scheduler-instances.yml

@@ -0,0 +1,25 @@
+#
+# Copyright 2018 ABSA Group Limited
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+databaseChangeLog:
+  - changeSet:
+      id: v0.3.4.add-scheduler-instances
+      logicalFilePath: v0.3.4.add-scheduler-instances
+      author: [email protected]
+      context: default
+      changes:
+        - sqlFile:
+            relativeToChangelogFile: true
+            path: v0.3.4.add-scheduler-instances.sql

src/main/scala/za/co/absa/hyperdrive/trigger/models/SchedulerInstance.scala

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2018 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.hyperdrive.trigger.models
+
+import java.time.LocalDateTime
+
+import za.co.absa.hyperdrive.trigger.models.enums.SchedulerInstanceStatuses.SchedulerInstanceStatus
+
+case class SchedulerInstance(
+  id: Long = 0,
+  status: SchedulerInstanceStatus,
+  lastHeartbeat: LocalDateTime
+)

src/main/scala/za/co/absa/hyperdrive/trigger/models/Workflow.scala

@@ -23,6 +23,7 @@ case class Workflow(
   project: String,
   created: LocalDateTime = LocalDateTime.now(),
   updated: Option[LocalDateTime],
+  schedulerInstanceId: Option[Long] = None,
   id: Long = 0
 )
 
@@ -36,6 +37,7 @@ case class WorkflowJoined(
   project: String,
   created: LocalDateTime = LocalDateTime.now(),
   updated: Option[LocalDateTime],
+  schedulerInstanceId: Option[Long] = None,
   sensor: Sensor,
   dagDefinitionJoined: DagDefinitionJoined,
   id: Long = 0
@@ -47,6 +49,7 @@ case class WorkflowJoined(
       project = this.project,
       created = this.created,
       updated = this.updated,
+      schedulerInstanceId = this.schedulerInstanceId,
       id = this.id
     )
   }

src/main/scala/za/co/absa/hyperdrive/trigger/models/enums/SchedulerInstanceStatuses.scala

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2018 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.hyperdrive.trigger.models.enums
+
+object SchedulerInstanceStatuses {
+
+  sealed abstract class SchedulerInstanceStatus(val name: String) {
+    override def toString: String = name
+  }
+
+  case object Active extends SchedulerInstanceStatus("Active")
+  case object Deactivated extends SchedulerInstanceStatus("Deactivated")
+
+  val statuses: Set[SchedulerInstanceStatus] = Set(Active, Deactivated)
+
+}

src/main/scala/za/co/absa/hyperdrive/trigger/models/tables/JdbcTypeMapper.scala

@@ -17,7 +17,7 @@ package za.co.absa.hyperdrive.trigger.models.tables
 
 import java.io.StringWriter
 
-import za.co.absa.hyperdrive.trigger.models.enums.{DBOperation, DagInstanceStatuses, JobStatuses, JobTypes, SensorTypes}
+import za.co.absa.hyperdrive.trigger.models.enums.{DBOperation, DagInstanceStatuses, SchedulerInstanceStatuses, JobStatuses, JobTypes, SensorTypes}
 import za.co.absa.hyperdrive.trigger.models.enums.SensorTypes.SensorType
 import za.co.absa.hyperdrive.trigger.models.enums.JobStatuses.JobStatus
 import za.co.absa.hyperdrive.trigger.models.enums.JobTypes.JobType
@@ -27,6 +27,7 @@ import za.co.absa.hyperdrive.trigger.ObjectMapperSingleton
 import za.co.absa.hyperdrive.trigger.models.WorkflowJoined
 import za.co.absa.hyperdrive.trigger.models.enums.DBOperation.DBOperation
 import za.co.absa.hyperdrive.trigger.models.enums.DagInstanceStatuses.DagInstanceStatus
+import za.co.absa.hyperdrive.trigger.models.enums.SchedulerInstanceStatuses.SchedulerInstanceStatus
 
 import scala.collection.immutable.SortedMap
 import scala.util.Try
@@ -84,6 +85,14 @@ trait JdbcTypeMapper {
       )
     )
 
+  implicit lazy val instanceStatusMapper: JdbcType[SchedulerInstanceStatus] =
+    MappedColumnType.base[SchedulerInstanceStatus, String](
+      status => status.name,
+      statusName => SchedulerInstanceStatuses.statuses.find(_.name == statusName).getOrElse(
+        throw new Exception(s"Couldn't find SchedulerInstanceStatus: $statusName")
+      )
+    )
+
   implicit lazy val payloadMapper: JdbcType[JsValue] =
     MappedColumnType.base[JsValue, String](
       payload => payload.toString(),