Added leader election and bus event to make sure only one pod is scheduling tasks

gradle/libs.versions.toml

@@ -43,6 +43,7 @@ spring-cloud-starter = { module = "org.springframework.cloud:spring-cloud-starte
 spring-cloud-starter-bus-kafka = { module = "org.springframework.cloud:spring-cloud-starter-bus-kafka" }
 spring-cloud-starter-config = { module = "org.springframework.cloud:spring-cloud-starter-config" }
 spring-cloud-config-server = { module = "org.springframework.cloud:spring-cloud-config-server" }
+spring-cloud-leader-election = { module = "org.springframework.cloud:spring-cloud-kubernetes-fabric8-leader" }
 spring-boot-starter = { module = "org.springframework.boot:spring-boot-starter" }
 spring-grpc = { module = "net.devh:grpc-spring-boot-starter", version.ref = "spring-grpc" }
 spring-ui = { module = "org.springdoc:springdoc-openapi-starter-webmvc-ui", version.ref = "spring-doc" }

module-server/server.gradle.kts

@@ -13,5 +13,6 @@ dependencies {
     annotationProcessor(libs.mapstruct.processor)
     kapt(libs.mapstruct.processor)
     implementation(libs.spring.cloud.starter.config)
-//    implementation(libs.spring.cloud.starter.bus.kafka)
+    implementation(libs.spring.cloud.leader.election)
+    implementation(libs.spring.cloud.starter.bus.kafka)
 }

module-server/src/main/kotlin/de/twomartens/timetable/MainApplication.kt

@@ -1,27 +1,40 @@
 package de.twomartens.timetable
 
-import de.twomartens.timetable.bahnApi.service.ScheduledTaskService
-import mu.KotlinLogging
+import de.twomartens.timetable.support.model.LeadershipStatus
 import org.springframework.boot.autoconfigure.SpringBootApplication
 import org.springframework.boot.context.event.ApplicationReadyEvent
 import org.springframework.boot.runApplication
+import org.springframework.cloud.kubernetes.commons.leader.LeaderProperties
 import org.springframework.context.event.EventListener
 import org.springframework.data.mongodb.config.EnableMongoAuditing
+import org.springframework.integration.leader.event.OnGrantedEvent
+import org.springframework.integration.leader.event.OnRevokedEvent
 import org.springframework.scheduling.annotation.EnableScheduling
 
 @EnableMongoAuditing
 @EnableScheduling
 @SpringBootApplication
 open class MainApplication(
-        private val scheduledTaskService: ScheduledTaskService
+        private val leadershipStatus: LeadershipStatus,
+        private val leaderProperties: LeaderProperties
 ) {
     @EventListener(ApplicationReadyEvent::class)
-    fun ready() {
-        scheduledTaskService.initializeScheduledTasks()
+    fun ready(event: ApplicationReadyEvent) {
+        setInitialLeadershipStatus()
     }
 
-    companion object {
-        private val log = KotlinLogging.logger {}
+    @EventListener(OnGrantedEvent::class)
+    fun onLeadershipGranted(event: OnGrantedEvent) {
+        leadershipStatus.isLeader = true
+    }
+
+    @EventListener(OnRevokedEvent::class)
+    fun onLeadershipRevoked(event: OnRevokedEvent) {
+        leadershipStatus.isLeader = false
+    }
+
+    private fun setInitialLeadershipStatus() {
+        leadershipStatus.isLeader = !leaderProperties.isEnabled
     }
 }
 

module-server/src/main/kotlin/de/twomartens/timetable/bahnApi/events/ScheduledTasksCreatedEvent.kt

@@ -0,0 +1,33 @@
+package de.twomartens.timetable.bahnApi.events
+
+import org.springframework.cloud.bus.event.Destination
+import org.springframework.cloud.bus.event.PathDestinationFactory
+import org.springframework.cloud.bus.event.RemoteApplicationEvent
+import java.time.Instant
+
+private const val TIMETABLE_SERVICE_NAME = "timetable"
+
+class ScheduledTasksCreatedEvent private constructor(source: Instant, originService: String,
+                                                     destination: Destination)
+    : RemoteApplicationEvent(source, originService, destination) {
+
+    private val creationTime: Instant
+
+    init {
+        creationTime = source
+    }
+
+    override fun getSource(): Instant {
+        return creationTime
+    }
+
+    companion object {
+        private val destinationFactory = PathDestinationFactory()
+
+        fun of(source: Instant): ScheduledTasksCreatedEvent {
+            return ScheduledTasksCreatedEvent(source,
+                    TIMETABLE_SERVICE_NAME,
+                    destinationFactory.getDestination(TIMETABLE_SERVICE_NAME))
+        }
+    }
+}

module-server/src/main/kotlin/de/twomartens/timetable/bahnApi/model/TaskFactory.kt

@@ -14,8 +14,7 @@ class TaskFactory {
 
     private val _tasksPerHour: MutableMap<HourAtDay, Int> = mutableMapOf()
 
-    fun initializeWithExistingTasks(existingTasks: List<ScheduledFetchTask>) {
-        resetCounter()
+    fun updateCounterWith(existingTasks: List<ScheduledFetchTask>) {
         countTasks(existingTasks)
     }
 

module-server/src/main/kotlin/de/twomartens/timetable/bahnApi/repository/ScheduledFetchTaskRepository.kt

@@ -5,7 +5,10 @@ import de.twomartens.timetable.bahnApi.model.db.ScheduledFetchTask
 import de.twomartens.timetable.types.HourAtDay
 import org.bson.types.ObjectId
 import org.springframework.data.mongodb.repository.MongoRepository
+import java.time.Instant
 
 interface ScheduledFetchTaskRepository : MongoRepository<ScheduledFetchTask, ObjectId> {
     fun findByEvaAndFetchedDateTime(eva: Eva, fetchedDateTime: HourAtDay): ScheduledFetchTask?
+
+    fun findAllByCreatedAfter(createdAt: Instant): List<ScheduledFetchTask>
 }

module-server/src/main/kotlin/de/twomartens/timetable/bahnApi/service/ScheduledTaskService.kt

@@ -1,39 +1,73 @@
 package de.twomartens.timetable.bahnApi.service
 
+import de.twomartens.timetable.bahnApi.events.ScheduledTasksCreatedEvent
 import de.twomartens.timetable.bahnApi.model.Eva
 import de.twomartens.timetable.bahnApi.model.FetchDates
 import de.twomartens.timetable.bahnApi.model.TaskFactory
 import de.twomartens.timetable.bahnApi.model.db.ScheduledFetchTask
 import de.twomartens.timetable.bahnApi.repository.ScheduledFetchTaskRepository
 import de.twomartens.timetable.model.db.TswRoute
+import de.twomartens.timetable.support.model.LeadershipStatus
+import de.twomartens.timetable.support.service.BusService
 import de.twomartens.timetable.types.Hour
 import de.twomartens.timetable.types.HourAtDay
 import mu.KotlinLogging
+import org.springframework.boot.context.event.ApplicationReadyEvent
+import org.springframework.cloud.kubernetes.commons.leader.LeaderProperties
+import org.springframework.context.event.EventListener
+import org.springframework.integration.leader.event.OnGrantedEvent
 import org.springframework.stereotype.Service
+import java.time.Instant
 import java.time.LocalDate
 
 @Service
 class ScheduledTaskService(
+        private val busService: BusService,
+        private val leadershipStatus: LeadershipStatus,
+        private val leaderProperties: LeaderProperties,
         private val scheduledFetchTaskRepository: ScheduledFetchTaskRepository,
         private val taskFactory: TaskFactory,
         private val taskScheduler: TaskScheduler
 ) {
-    fun initializeScheduledTasks() {
-        log.info { "Initialize persisted scheduled fetch tasks" }
-        val scheduledTasks = scheduledFetchTaskRepository.findAll()
-        taskFactory.initializeWithExistingTasks(scheduledTasks)
-        taskScheduler.scheduleFetchTasks(scheduledTasks)
+    private var createdTime: Instant = Instant.EPOCH
+    private var lastUpdate: Instant = Instant.EPOCH
+
+    @EventListener(ApplicationReadyEvent::class)
+    fun onApplicationReady(event: ApplicationReadyEvent) {
+        if (!leaderProperties.isEnabled) {
+            val updateTime = Instant.ofEpochMilli(event.timestamp)
+            updateTaskCounterAndScheduleTasksIfLeader(updateTime)
+        }
     }
 
-    fun scheduleTimetableFetch(tswRoute: TswRoute, fetchedDate: LocalDate) {
+    @EventListener(OnGrantedEvent::class)
+    fun onLeadershipGranted(event: OnGrantedEvent) {
+        val updateTime = Instant.ofEpochMilli(event.timestamp)
+        updateTaskCounterAndScheduleTasksIfLeader(updateTime)
+    }
+
+    @EventListener(ScheduledTasksCreatedEvent::class)
+    fun onScheduledTasksCreated(event: ScheduledTasksCreatedEvent) {
+        updateTaskCounterAndScheduleTasksIfLeader(event.source)
+    }
+
+    private fun updateTaskCounterAndScheduleTasksIfLeader(updateTime: Instant) {
+        log.info { "Update tasks from database and schedule if leader" }
+        val createdTasks = findTasksCreatedSince(lastUpdate)
+        updateCounterIfNotUpToDate(updateTime, createdTasks)
+        scheduleTasksIfLeader(createdTasks)
+        lastUpdate = updateTime
+    }
+
+    fun triggerTimetableFetch(tswRoute: TswRoute, fetchedDate: LocalDate) {
         log.info {
-            "Schedule timetable fetch: [route ${tswRoute.name}]"
+            "Trigger timetable fetch: [route ${tswRoute.name}]"
         }
         val fetchDates = calculateDatesToFetch(fetchedDate)
         val newTasks = buildScheduledTasks(tswRoute, fetchDates)
 
-        storeTasksInDatabase(newTasks)
-        taskScheduler.scheduleFetchTasks(newTasks)
+        storeTasksInDatabaseAndStoreCreationTime(newTasks)
+        publishTasksCreatedEvent()
     }
 
     private fun calculateDatesToFetch(fetchedDate: LocalDate): FetchDates {
@@ -67,8 +101,39 @@ class ScheduledTaskService(
         return newTasks
     }
 
-    private fun storeTasksInDatabase(newTasks: List<ScheduledFetchTask>) {
-        scheduledFetchTaskRepository.saveAll(newTasks)
+    private fun scheduleTasksIfLeader(tasksToSchedule: List<ScheduledFetchTask>) {
+        if (leadershipStatus.isLeader) {
+            taskScheduler.scheduleFetchTasks(tasksToSchedule)
+        }
+    }
+
+    private fun updateCounterIfNotUpToDate(updateTime: Instant,
+                                           createdTasks: List<ScheduledFetchTask>) {
+        if (notUpToDate(updateTime)) {
+            taskFactory.updateCounterWith(createdTasks)
+        }
+    }
+
+    private fun notUpToDate(updateTime: Instant) = createdTime != updateTime
+
+    private fun findTasksCreatedSince(lastScheduledTime: Instant): List<ScheduledFetchTask> {
+        return scheduledFetchTaskRepository.findAllByCreatedAfter(lastScheduledTime)
+    }
+
+    private fun storeTasksInDatabaseAndStoreCreationTime(newTasks: List<ScheduledFetchTask>) {
+        val savedEntities = scheduledFetchTaskRepository.saveAll(newTasks)
+        storeCreationTime(savedEntities)
+    }
+
+    private fun storeCreationTime(savedEntities: MutableList<ScheduledFetchTask>) {
+        if (savedEntities.isNotEmpty()) {
+            createdTime = savedEntities[savedEntities.lastIndex].created
+        }
+    }
+
+    private fun publishTasksCreatedEvent() {
+        val event = ScheduledTasksCreatedEvent.of(createdTime)
+        busService.publishEvent(event)
     }
 
     companion object {

module-server/src/main/kotlin/de/twomartens/timetable/configuration/PropertyConfiguration.kt

@@ -6,9 +6,11 @@ import de.twomartens.timetable.property.ServiceProperties
 import de.twomartens.timetable.property.StatusProbeProperties
 import de.twomartens.timetable.property.TimeProperties
 import org.springframework.boot.context.properties.EnableConfigurationProperties
+import org.springframework.cloud.kubernetes.commons.leader.LeaderProperties
 import org.springframework.context.annotation.Configuration
 
 @Configuration
 @EnableConfigurationProperties(RestTemplateTimeoutProperties::class, ServiceProperties::class,
-        StatusProbeProperties::class, TimeProperties::class, BahnApiProperties::class)
-open class PropertyConfiguration
+        StatusProbeProperties::class, TimeProperties::class, BahnApiProperties::class,
+        LeaderProperties::class)
+open class PropertyConfiguration

module-server/src/main/kotlin/de/twomartens/timetable/controller/TimetableController.kt

@@ -54,6 +54,6 @@ class TimetableController(
         val route = routeRepository.findByUserIdAndName(userId, routeName)
                 ?: throw ResponseStatusException(HttpStatus.BAD_REQUEST,
                         "Route name must belong to existing route")
-        scheduledTaskService.scheduleTimetableFetch(route, date)
+        scheduledTaskService.triggerTimetableFetch(route, date)
     }
 }

module-server/src/main/kotlin/de/twomartens/timetable/support/configuration/BusConfiguration.kt

@@ -0,0 +1,8 @@
+package de.twomartens.timetable.support.configuration
+
+import org.springframework.cloud.bus.jackson.RemoteApplicationEventScan
+import org.springframework.context.annotation.Configuration
+
+@Configuration
+@RemoteApplicationEventScan(basePackages = ["de.twomartens.timetable"])
+open class BusConfiguration

module-server/src/main/kotlin/de/twomartens/timetable/support/model/LeadershipStatus.kt

@@ -0,0 +1,8 @@
+package de.twomartens.timetable.support.model
+
+import org.springframework.stereotype.Component
+
+@Component
+class LeadershipStatus {
+    var isLeader = false
+}

module-server/src/main/kotlin/de/twomartens/timetable/support/service/BusService.kt

@@ -0,0 +1,17 @@
+package de.twomartens.timetable.support.service
+
+import org.springframework.beans.factory.ObjectProvider
+import org.springframework.cloud.bus.BusBridge
+import org.springframework.cloud.bus.event.RemoteApplicationEvent
+import org.springframework.stereotype.Service
+
+@Service
+class BusService(
+        private val busBridge: ObjectProvider<BusBridge>
+) {
+    fun publishEvent(event: RemoteApplicationEvent) {
+        busBridge.ifAvailable {
+            it.send(event)
+        }
+    }
+}

module-server/src/main/resources/application-dev.yaml

@@ -3,6 +3,10 @@ management.server.port: 12101
 
 
 spring:
+  cloud:
+    kubernetes:
+      leader:
+        enabled: false
   config:
     import:
       - "classpath:config/timetable.yaml"

module-server/src/main/resources/application.yaml

@@ -27,6 +27,9 @@ spring:
   cloud:
     config:
       name: timetable
+    kubernetes:
+      leader:
+        enabled: true
   main:
     banner-mode: off
   profiles: