change: Remove leader election and Kafka

This commit is contained in:
2025-03-20 17:57:29 +01:00
parent 10597b0b53
commit fa1f4b47b9
18 changed files with 17 additions and 175 deletions
-1
View File
@@ -40,7 +40,6 @@ spring-boot-oauth2-resource-server = { module = "org.springframework.boot:spring
spring-boot-config = { module = "org.springframework.boot:spring-boot-configuration-processor", version.ref = "spring-boot" } spring-boot-config = { module = "org.springframework.boot:spring-boot-configuration-processor", version.ref = "spring-boot" }
spring-cloud = { module = "org.springframework.cloud:spring-cloud-dependencies", version.ref = "spring-cloud" } spring-cloud = { module = "org.springframework.cloud:spring-cloud-dependencies", version.ref = "spring-cloud" }
spring-cloud-starter = { module = "org.springframework.cloud:spring-cloud-starter" } spring-cloud-starter = { module = "org.springframework.cloud:spring-cloud-starter" }
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-starter-config = { module = "org.springframework.cloud:spring-cloud-starter-config" }
spring-cloud-config-server = { module = "org.springframework.cloud:spring-cloud-config-server" } 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-cloud-leader-election = { module = "org.springframework.cloud:spring-cloud-kubernetes-fabric8-leader" }
-2
View File
@@ -18,6 +18,4 @@ dependencies {
implementation(libs.spring.openapi) implementation(libs.spring.openapi)
implementation(libs.spring.boot.mongo) implementation(libs.spring.boot.mongo)
implementation(libs.spring.cloud.leader.election)
implementation(libs.spring.cloud.starter.bus.kafka)
} }
@@ -1,29 +0,0 @@
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 = 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))
}
}
}
@@ -3,9 +3,7 @@ package de.twomartens.timetable.bahnApi.property
import io.swagger.v3.oas.annotations.media.Schema import io.swagger.v3.oas.annotations.media.Schema
import org.springframework.boot.context.properties.ConfigurationProperties import org.springframework.boot.context.properties.ConfigurationProperties
import org.springframework.boot.context.properties.bind.ConstructorBinding import org.springframework.boot.context.properties.bind.ConstructorBinding
import org.springframework.cloud.context.config.annotation.RefreshScope
@RefreshScope
@ConfigurationProperties(prefix = "de.twomartens.timetable.bahn-api") @ConfigurationProperties(prefix = "de.twomartens.timetable.bahn-api")
@Schema(description = "Properties to configure the BahnAPI") @Schema(description = "Properties to configure the BahnAPI")
data class BahnApiProperties @ConstructorBinding constructor( data class BahnApiProperties @ConstructorBinding constructor(
@@ -1,8 +1,5 @@
package de.twomartens.timetable.bahnApi.service package de.twomartens.timetable.bahnApi.service
import de.twomartens.support.model.LeadershipStatus
import de.twomartens.support.service.BusService
import de.twomartens.timetable.bahnApi.events.ScheduledTasksCreatedEvent
import de.twomartens.timetable.bahnApi.model.Eva import de.twomartens.timetable.bahnApi.model.Eva
import de.twomartens.timetable.bahnApi.model.FetchDates import de.twomartens.timetable.bahnApi.model.FetchDates
import de.twomartens.timetable.bahnApi.model.TaskFactory import de.twomartens.timetable.bahnApi.model.TaskFactory
@@ -13,20 +10,15 @@ import de.twomartens.timetable.types.Hour
import de.twomartens.timetable.types.HourAtDay import de.twomartens.timetable.types.HourAtDay
import mu.KotlinLogging import mu.KotlinLogging
import org.springframework.boot.context.event.ApplicationReadyEvent import org.springframework.boot.context.event.ApplicationReadyEvent
import org.springframework.cloud.kubernetes.commons.leader.LeaderProperties
import org.springframework.context.event.EventListener import org.springframework.context.event.EventListener
import org.springframework.data.mongodb.core.BulkOperations import org.springframework.data.mongodb.core.BulkOperations
import org.springframework.data.mongodb.core.MongoTemplate import org.springframework.data.mongodb.core.MongoTemplate
import org.springframework.integration.leader.event.OnGrantedEvent
import org.springframework.stereotype.Service import org.springframework.stereotype.Service
import java.time.Instant import java.time.Instant
import java.time.LocalDate import java.time.LocalDate
@Service @Service
class ScheduledTaskService( class ScheduledTaskService(
private val busService: BusService,
private val leadershipStatus: LeadershipStatus,
private val leaderProperties: LeaderProperties,
private val scheduledFetchTaskRepository: ScheduledFetchTaskRepository, private val scheduledFetchTaskRepository: ScheduledFetchTaskRepository,
private val taskFactory: TaskFactory, private val taskFactory: TaskFactory,
private val fetchTaskScheduler: FetchTaskScheduler, private val fetchTaskScheduler: FetchTaskScheduler,
@@ -37,30 +29,16 @@ class ScheduledTaskService(
@EventListener(ApplicationReadyEvent::class) @EventListener(ApplicationReadyEvent::class)
fun onApplicationReady(event: ApplicationReadyEvent) { fun onApplicationReady(event: ApplicationReadyEvent) {
if (!leaderProperties.isEnabled) { log.info { "Application ready" }
log.info { "Leader election disabled and application ready" }
val updateTime = Instant.ofEpochMilli(event.timestamp) val updateTime = Instant.ofEpochMilli(event.timestamp)
updateTaskCounterAndScheduleTasksIfLeader(updateTime) updateTaskCounterAndScheduleTasks(updateTime)
}
} }
@EventListener(OnGrantedEvent::class) private fun updateTaskCounterAndScheduleTasks(updateTime: Instant) {
fun onLeadershipGranted(event: OnGrantedEvent) { log.info { "Update tasks from database and schedule" }
log.info { "Granted leadership" }
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) val createdTasks = findTasksCreatedSince(lastUpdate)
updateCounterIfNotUpToDate(updateTime, createdTasks) updateCounterIfNotUpToDate(updateTime, createdTasks)
scheduleTasksIfLeader(createdTasks) scheduleTasks(createdTasks)
lastUpdate = updateTime lastUpdate = updateTime
} }
@@ -72,7 +50,6 @@ class ScheduledTaskService(
val newTasks = buildScheduledTasks(tswRoute, fetchDates) val newTasks = buildScheduledTasks(tswRoute, fetchDates)
storeTasksInDatabaseAndStoreCreationTime(newTasks) storeTasksInDatabaseAndStoreCreationTime(newTasks)
publishTasksCreatedEvent()
} }
private fun calculateDatesToFetch(fetchedDate: LocalDate): FetchDates { private fun calculateDatesToFetch(fetchedDate: LocalDate): FetchDates {
@@ -109,11 +86,9 @@ class ScheduledTaskService(
return newTasks return newTasks
} }
private fun scheduleTasksIfLeader(tasksToSchedule: List<ScheduledFetchTask>) { private fun scheduleTasks(tasksToSchedule: List<ScheduledFetchTask>) {
if (leadershipStatus.isLeader) {
fetchTaskScheduler.scheduleFetchTasks(tasksToSchedule) fetchTaskScheduler.scheduleFetchTasks(tasksToSchedule)
} }
}
private fun updateCounterIfNotUpToDate(updateTime: Instant, private fun updateCounterIfNotUpToDate(updateTime: Instant,
createdTasks: List<ScheduledFetchTask>) { createdTasks: List<ScheduledFetchTask>) {
@@ -139,11 +114,6 @@ class ScheduledTaskService(
.execute() .execute()
} }
private fun publishTasksCreatedEvent() {
val event = ScheduledTasksCreatedEvent.of(createdTime)
busService.publishEvent(event)
}
companion object { companion object {
private val log = KotlinLogging.logger {} private val log = KotlinLogging.logger {}
} }
-3
View File
@@ -18,7 +18,4 @@ dependencies {
kapt(libs.mapstruct.processor) kapt(libs.mapstruct.processor)
implementation(libs.bundles.spring.boot.security) implementation(libs.bundles.spring.boot.security)
implementation(libs.spring.cloud.starter.config)
implementation(libs.spring.cloud.leader.election)
implementation(libs.spring.cloud.starter.bus.kafka)
} }
@@ -1,44 +1,15 @@
package de.twomartens.timetable package de.twomartens.timetable
import de.twomartens.support.model.LeadershipStatus
import org.springframework.boot.autoconfigure.SpringBootApplication import org.springframework.boot.autoconfigure.SpringBootApplication
import org.springframework.boot.context.event.ApplicationReadyEvent
import org.springframework.boot.runApplication 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.data.mongodb.config.EnableMongoAuditing
import org.springframework.integration.leader.event.OnGrantedEvent
import org.springframework.integration.leader.event.OnRevokedEvent
import org.springframework.scheduling.annotation.EnableScheduling import org.springframework.scheduling.annotation.EnableScheduling
@EnableMongoAuditing @EnableMongoAuditing
@EnableScheduling @EnableScheduling
@SpringBootApplication(scanBasePackages = ["de.twomartens.support", "de.twomartens.timetable"]) @SpringBootApplication(scanBasePackages = ["de.twomartens.support", "de.twomartens.timetable"])
open class MainApplication( open class MainApplication() {}
private val leadershipStatus: LeadershipStatus,
private val leaderProperties: LeaderProperties
) {
@EventListener(ApplicationReadyEvent::class)
fun ready(event: ApplicationReadyEvent) {
setInitialLeadershipStatus()
}
@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
}
}
fun main(args: Array<String>) { fun main(args: Array<String>) {
runApplication<MainApplication>(*args) runApplication<MainApplication>(*args)
} }
@@ -1,8 +0,0 @@
package de.twomartens.timetable.configuration
import org.springframework.cloud.bus.jackson.RemoteApplicationEventScan
import org.springframework.context.annotation.Configuration
@Configuration
@RemoteApplicationEventScan(basePackages = ["de.twomartens.timetable", "de.twomartens.support"])
open class BusConfiguration
@@ -7,12 +7,10 @@ import de.twomartens.support.property.TimeProperties
import de.twomartens.timetable.bahnApi.property.BahnApiProperties import de.twomartens.timetable.bahnApi.property.BahnApiProperties
import de.twomartens.timetable.property.ServiceProperties import de.twomartens.timetable.property.ServiceProperties
import org.springframework.boot.context.properties.EnableConfigurationProperties import org.springframework.boot.context.properties.EnableConfigurationProperties
import org.springframework.cloud.kubernetes.commons.leader.LeaderProperties
import org.springframework.context.annotation.Configuration import org.springframework.context.annotation.Configuration
@Configuration @Configuration
@EnableConfigurationProperties(RestTemplateTimeoutProperties::class, ServiceProperties::class, @EnableConfigurationProperties(RestTemplateTimeoutProperties::class, ServiceProperties::class,
StatusProbeProperties::class, TimeProperties::class, BahnApiProperties::class, StatusProbeProperties::class, TimeProperties::class, BahnApiProperties::class,
HealthCheckProperties::class, HealthCheckProperties::class)
LeaderProperties::class)
open class PropertyConfiguration open class PropertyConfiguration
@@ -5,11 +5,9 @@ import de.twomartens.timetable.model.common.LanguageCode
import de.twomartens.timetable.types.NonEmptyString import de.twomartens.timetable.types.NonEmptyString
import io.swagger.v3.oas.annotations.media.Schema import io.swagger.v3.oas.annotations.media.Schema
import org.springframework.boot.context.properties.ConfigurationProperties import org.springframework.boot.context.properties.ConfigurationProperties
import org.springframework.cloud.context.config.annotation.RefreshScope
@RefreshScope
@ConfigurationProperties(prefix = "de.twomartens.timetable") @ConfigurationProperties(prefix = "de.twomartens.timetable")
@Schema(description = "Properties, to configure this Application") @Schema(description = "Properties, to configure this Application")
class ServiceProperties { open class ServiceProperties {
lateinit var countryNames: Map<CountryCode, Map<LanguageCode, NonEmptyString>> lateinit var countryNames: Map<CountryCode, Map<LanguageCode, NonEmptyString>>
} }
@@ -1,13 +1,2 @@
server.port: 12100 server.port: 12100
management.server.port: 12101 management.server.port: 12101
spring:
cloud:
kubernetes:
leader:
enabled: false
config:
import:
- "classpath:config/timetable.yaml"
- "optional:configserver:${CONFIGSERVER_SCHEME:http}://${CONFIGSERVER_HOST:localhost}:${CONFIGSERVER_PORT:8888}${CONFIGSERVER_PREFIX:/config}"
@@ -1,7 +0,0 @@
spring:
config:
import:
- "configserver:${CONFIGSERVER_SCHEME:http}://${CONFIGSERVER_HOST:configserver}:${CONFIGSERVER_PORT:8888}${CONFIGSERVER_PREFIX:/config}"
kafka:
bootstrap-servers:
- ${KAFKA_HOST:kafka}:${KAFKA_PORT:9092}
@@ -26,12 +26,6 @@ management:
http.server.requests: true http.server.requests: true
spring: spring:
cloud:
config:
name: timetable
kubernetes:
leader:
enabled: true
main: main:
banner-mode: off banner-mode: off
profiles: profiles:
@@ -84,6 +78,12 @@ de.twomartens.timetable:
bahn-api: bahn-api:
client-id: ${DB_CLIENT_ID} client-id: ${DB_CLIENT_ID}
client-secret: ${DB_CLIENT_SECRET} client-secret: ${DB_CLIENT_SECRET}
greeting: "Hello %s, on millenium"
statusprobe:
scheduleDuration: 60
maxKafkaFailureDuration: 5
maxBlobFailureCount: 5
maxFailurePercent: 50
de.twomartens.support.health.greeting: "Good morning" de.twomartens.support.health.greeting: "Good morning"
time: time:
@@ -1,8 +0,0 @@
package de.twomartens.support.model
import org.springframework.stereotype.Component
@Component
class LeadershipStatus {
var isLeader = false
}
@@ -4,16 +4,13 @@ import io.swagger.v3.oas.annotations.media.Schema
import org.springframework.boot.context.properties.ConfigurationProperties import org.springframework.boot.context.properties.ConfigurationProperties
import org.springframework.boot.context.properties.bind.ConstructorBinding import org.springframework.boot.context.properties.bind.ConstructorBinding
import org.springframework.boot.convert.DurationUnit import org.springframework.boot.convert.DurationUnit
import org.springframework.cloud.context.config.annotation.RefreshScope
import java.time.Duration import java.time.Duration
import java.time.temporal.ChronoUnit import java.time.temporal.ChronoUnit
@RefreshScope
@ConfigurationProperties(prefix = "de.twomartens.timetable.statusprobe") @ConfigurationProperties(prefix = "de.twomartens.timetable.statusprobe")
@Schema(description = "Properties, to configure this Application") @Schema(description = "Properties, to configure this Application")
data class StatusProbeProperties @ConstructorBinding constructor( data class StatusProbeProperties @ConstructorBinding constructor(
@DurationUnit(ChronoUnit.SECONDS) val scheduleDuration: Duration, @DurationUnit(ChronoUnit.SECONDS) val scheduleDuration: Duration,
@DurationUnit(ChronoUnit.MINUTES) val maxKafkaFailureDuration: Duration,
val maxBlobFailureCount: Int, val maxBlobFailureCount: Int,
val maxFailurePercent: Int val maxFailurePercent: Int
) )
@@ -3,10 +3,8 @@ package de.twomartens.support.property
import io.swagger.v3.oas.annotations.media.Schema import io.swagger.v3.oas.annotations.media.Schema
import org.springframework.boot.context.properties.ConfigurationProperties import org.springframework.boot.context.properties.ConfigurationProperties
import org.springframework.boot.context.properties.bind.ConstructorBinding import org.springframework.boot.context.properties.bind.ConstructorBinding
import org.springframework.cloud.context.config.annotation.RefreshScope
import java.time.ZoneId import java.time.ZoneId
@RefreshScope
@ConfigurationProperties(prefix = "time") @ConfigurationProperties(prefix = "time")
@Schema(description = "Properties to configure time") @Schema(description = "Properties to configure time")
data class TimeProperties @ConstructorBinding constructor( data class TimeProperties @ConstructorBinding constructor(
@@ -1,17 +0,0 @@
package de.twomartens.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)
}
}
}
-2
View File
@@ -10,6 +10,4 @@ dependencies {
implementation(libs.spring.openapi) implementation(libs.spring.openapi)
implementation(libs.bundles.spring.boot.security) implementation(libs.bundles.spring.boot.security)
implementation(libs.spring.cloud.leader.election)
implementation(libs.spring.cloud.starter.bus.kafka)
} }