From e0e793ae1e70ea27dae9d909ced0e980d649844d Mon Sep 17 00:00:00 2001 From: Jim Martens Date: Sat, 28 Oct 2023 13:43:50 +0200 Subject: [PATCH] Added leader election and bus event to make sure only one pod is scheduling tasks --- gradle/libs.versions.toml | 1 + module-server/server.gradle.kts | 3 +- .../twomartens/timetable/MainApplication.kt | 27 ++++-- .../events/ScheduledTasksCreatedEvent.kt | 33 +++++++ .../timetable/bahnApi/model/TaskFactory.kt | 3 +- .../ScheduledFetchTaskRepository.kt | 3 + .../bahnApi/service/ScheduledTaskService.kt | 87 ++++++++++++++++--- .../configuration/PropertyConfiguration.kt | 6 +- .../controller/TimetableController.kt | 2 +- .../support/configuration/BusConfiguration.kt | 8 ++ .../support/model/LeadershipStatus.kt | 8 ++ .../timetable/support/service/BusService.kt | 17 ++++ .../src/main/resources/application-dev.yaml | 4 + .../src/main/resources/application.yaml | 3 + 14 files changed, 181 insertions(+), 24 deletions(-) create mode 100644 module-server/src/main/kotlin/de/twomartens/timetable/bahnApi/events/ScheduledTasksCreatedEvent.kt create mode 100644 module-server/src/main/kotlin/de/twomartens/timetable/support/configuration/BusConfiguration.kt create mode 100644 module-server/src/main/kotlin/de/twomartens/timetable/support/model/LeadershipStatus.kt create mode 100644 module-server/src/main/kotlin/de/twomartens/timetable/support/service/BusService.kt diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 650fe55..712e209 100644 --- a/gradle/libs.versions.toml +++ b/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" } diff --git a/module-server/server.gradle.kts b/module-server/server.gradle.kts index 89c558c..8a9fdae 100644 --- a/module-server/server.gradle.kts +++ b/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) } diff --git a/module-server/src/main/kotlin/de/twomartens/timetable/MainApplication.kt b/module-server/src/main/kotlin/de/twomartens/timetable/MainApplication.kt index cba4a89..d2469bd 100644 --- a/module-server/src/main/kotlin/de/twomartens/timetable/MainApplication.kt +++ b/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 } } diff --git a/module-server/src/main/kotlin/de/twomartens/timetable/bahnApi/events/ScheduledTasksCreatedEvent.kt b/module-server/src/main/kotlin/de/twomartens/timetable/bahnApi/events/ScheduledTasksCreatedEvent.kt new file mode 100644 index 0000000..e83d8b8 --- /dev/null +++ b/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)) + } + } +} \ No newline at end of file diff --git a/module-server/src/main/kotlin/de/twomartens/timetable/bahnApi/model/TaskFactory.kt b/module-server/src/main/kotlin/de/twomartens/timetable/bahnApi/model/TaskFactory.kt index 36a6408..9fbb416 100644 --- a/module-server/src/main/kotlin/de/twomartens/timetable/bahnApi/model/TaskFactory.kt +++ b/module-server/src/main/kotlin/de/twomartens/timetable/bahnApi/model/TaskFactory.kt @@ -14,8 +14,7 @@ class TaskFactory { private val _tasksPerHour: MutableMap = mutableMapOf() - fun initializeWithExistingTasks(existingTasks: List) { - resetCounter() + fun updateCounterWith(existingTasks: List) { countTasks(existingTasks) } diff --git a/module-server/src/main/kotlin/de/twomartens/timetable/bahnApi/repository/ScheduledFetchTaskRepository.kt b/module-server/src/main/kotlin/de/twomartens/timetable/bahnApi/repository/ScheduledFetchTaskRepository.kt index 138c34e..bf5e111 100644 --- a/module-server/src/main/kotlin/de/twomartens/timetable/bahnApi/repository/ScheduledFetchTaskRepository.kt +++ b/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 { fun findByEvaAndFetchedDateTime(eva: Eva, fetchedDateTime: HourAtDay): ScheduledFetchTask? + + fun findAllByCreatedAfter(createdAt: Instant): List } diff --git a/module-server/src/main/kotlin/de/twomartens/timetable/bahnApi/service/ScheduledTaskService.kt b/module-server/src/main/kotlin/de/twomartens/timetable/bahnApi/service/ScheduledTaskService.kt index 0c0e8d4..f0ed954 100644 --- a/module-server/src/main/kotlin/de/twomartens/timetable/bahnApi/service/ScheduledTaskService.kt +++ b/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) { - scheduledFetchTaskRepository.saveAll(newTasks) + private fun scheduleTasksIfLeader(tasksToSchedule: List) { + if (leadershipStatus.isLeader) { + taskScheduler.scheduleFetchTasks(tasksToSchedule) + } + } + + private fun updateCounterIfNotUpToDate(updateTime: Instant, + createdTasks: List) { + if (notUpToDate(updateTime)) { + taskFactory.updateCounterWith(createdTasks) + } + } + + private fun notUpToDate(updateTime: Instant) = createdTime != updateTime + + private fun findTasksCreatedSince(lastScheduledTime: Instant): List { + return scheduledFetchTaskRepository.findAllByCreatedAfter(lastScheduledTime) + } + + private fun storeTasksInDatabaseAndStoreCreationTime(newTasks: List) { + val savedEntities = scheduledFetchTaskRepository.saveAll(newTasks) + storeCreationTime(savedEntities) + } + + private fun storeCreationTime(savedEntities: MutableList) { + if (savedEntities.isNotEmpty()) { + createdTime = savedEntities[savedEntities.lastIndex].created + } + } + + private fun publishTasksCreatedEvent() { + val event = ScheduledTasksCreatedEvent.of(createdTime) + busService.publishEvent(event) } companion object { diff --git a/module-server/src/main/kotlin/de/twomartens/timetable/configuration/PropertyConfiguration.kt b/module-server/src/main/kotlin/de/twomartens/timetable/configuration/PropertyConfiguration.kt index 98da0dd..aee3b4a 100644 --- a/module-server/src/main/kotlin/de/twomartens/timetable/configuration/PropertyConfiguration.kt +++ b/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 \ No newline at end of file + StatusProbeProperties::class, TimeProperties::class, BahnApiProperties::class, + LeaderProperties::class) +open class PropertyConfiguration diff --git a/module-server/src/main/kotlin/de/twomartens/timetable/controller/TimetableController.kt b/module-server/src/main/kotlin/de/twomartens/timetable/controller/TimetableController.kt index 2848223..d7db7e8 100644 --- a/module-server/src/main/kotlin/de/twomartens/timetable/controller/TimetableController.kt +++ b/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) } } \ No newline at end of file diff --git a/module-server/src/main/kotlin/de/twomartens/timetable/support/configuration/BusConfiguration.kt b/module-server/src/main/kotlin/de/twomartens/timetable/support/configuration/BusConfiguration.kt new file mode 100644 index 0000000..eed7088 --- /dev/null +++ b/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 diff --git a/module-server/src/main/kotlin/de/twomartens/timetable/support/model/LeadershipStatus.kt b/module-server/src/main/kotlin/de/twomartens/timetable/support/model/LeadershipStatus.kt new file mode 100644 index 0000000..c85fe3a --- /dev/null +++ b/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 +} diff --git a/module-server/src/main/kotlin/de/twomartens/timetable/support/service/BusService.kt b/module-server/src/main/kotlin/de/twomartens/timetable/support/service/BusService.kt new file mode 100644 index 0000000..781a640 --- /dev/null +++ b/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 +) { + fun publishEvent(event: RemoteApplicationEvent) { + busBridge.ifAvailable { + it.send(event) + } + } +} \ No newline at end of file diff --git a/module-server/src/main/resources/application-dev.yaml b/module-server/src/main/resources/application-dev.yaml index 6e2e3c0..5a9da5b 100644 --- a/module-server/src/main/resources/application-dev.yaml +++ b/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" diff --git a/module-server/src/main/resources/application.yaml b/module-server/src/main/resources/application.yaml index a57518c..3287a6b 100644 --- a/module-server/src/main/resources/application.yaml +++ b/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: