From 03c72b40761ebf7078e17a45e10d2accb8cc23a3 Mon Sep 17 00:00:00 2001 From: Ivan Nikolaev-Axenov Date: Mon, 1 Jun 2026 12:08:50 +0300 Subject: [PATCH] job cleaner cron added --- .../config/IndicSearchConfig.java | 26 +++++ .../config/settings/ApplicationSettings.java | 13 ++- .../settings/cron/JobCleanerSettings.java | 22 +++++ .../cron/CronChangeListener.java | 19 ++++ .../nbch/indicsearchservice/cron/CronJob.java | 5 + .../cron/CronJobManager.java | 78 +++++++++++++++ .../indicsearchservice/cron/CronJobName.java | 5 + .../indicsearchservice/cron/JobCleaner.java | 97 +++++++++++++++++++ .../event/CronChangedEvent.java | 9 ++ .../manager/IndicSearchJobManager.java | 15 +-- .../processor/IndicSearchProcessor.java | 9 -- .../service/SettingsService.java | 43 +++++--- 12 files changed, 309 insertions(+), 32 deletions(-) create mode 100644 src/main/java/ru/nbch/indicsearchservice/config/IndicSearchConfig.java create mode 100644 src/main/java/ru/nbch/indicsearchservice/config/settings/cron/JobCleanerSettings.java create mode 100644 src/main/java/ru/nbch/indicsearchservice/cron/CronChangeListener.java create mode 100644 src/main/java/ru/nbch/indicsearchservice/cron/CronJob.java create mode 100644 src/main/java/ru/nbch/indicsearchservice/cron/CronJobManager.java create mode 100644 src/main/java/ru/nbch/indicsearchservice/cron/CronJobName.java create mode 100644 src/main/java/ru/nbch/indicsearchservice/cron/JobCleaner.java create mode 100644 src/main/java/ru/nbch/indicsearchservice/event/CronChangedEvent.java diff --git a/src/main/java/ru/nbch/indicsearchservice/config/IndicSearchConfig.java b/src/main/java/ru/nbch/indicsearchservice/config/IndicSearchConfig.java new file mode 100644 index 0000000..a5895cc --- /dev/null +++ b/src/main/java/ru/nbch/indicsearchservice/config/IndicSearchConfig.java @@ -0,0 +1,26 @@ +package ru.nbch.indicsearchservice.config; + +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import ru.nbch.indicsearchservice.model.SearchJob; + +@Configuration +public class IndicSearchConfig { + private final ExecutorService searchJobExecutor = Executors.newFixedThreadPool(5); + private final Map searchJobs = new ConcurrentHashMap<>(); + + @Bean("searchJobExecutor") + public ExecutorService searchJobExecutor() { + return searchJobExecutor; + } + + @Bean("searchJobsMap") + public Map searchJobsMap() { + return searchJobs; + } +} diff --git a/src/main/java/ru/nbch/indicsearchservice/config/settings/ApplicationSettings.java b/src/main/java/ru/nbch/indicsearchservice/config/settings/ApplicationSettings.java index 6b365b2..7359712 100644 --- a/src/main/java/ru/nbch/indicsearchservice/config/settings/ApplicationSettings.java +++ b/src/main/java/ru/nbch/indicsearchservice/config/settings/ApplicationSettings.java @@ -5,6 +5,7 @@ import java.nio.file.Path; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.NestedConfigurationProperty; import org.springframework.context.annotation.Configuration; +import ru.nbch.indicsearchservice.config.settings.cron.JobCleanerSettings; @Configuration @ConfigurationProperties(prefix = "app") @@ -13,9 +14,11 @@ public class ApplicationSettings { private DataSourceSettings dataSource; private DataSourceSettings indicDataSource = new DataSourceSettings(); - private File storageDirectory = Path.of("/opt/app/indic-search-service-storage").toAbsolutePath().toFile(); + @NestedConfigurationProperty + private JobCleanerSettings jobCleaner = new JobCleanerSettings(); + public DataSourceSettings getDataSource() { return dataSource; } @@ -39,4 +42,12 @@ public class ApplicationSettings { public void setStorageDirectory(File storageDirectory) { this.storageDirectory = storageDirectory; } + + public JobCleanerSettings getJobCleaner() { + return jobCleaner; + } + + public void setJobCleaner(JobCleanerSettings jobCleaner) { + this.jobCleaner = jobCleaner; + } } diff --git a/src/main/java/ru/nbch/indicsearchservice/config/settings/cron/JobCleanerSettings.java b/src/main/java/ru/nbch/indicsearchservice/config/settings/cron/JobCleanerSettings.java new file mode 100644 index 0000000..ee67ad6 --- /dev/null +++ b/src/main/java/ru/nbch/indicsearchservice/config/settings/cron/JobCleanerSettings.java @@ -0,0 +1,22 @@ +package ru.nbch.indicsearchservice.config.settings.cron; + +public class JobCleanerSettings { + private boolean enabled = true; + private String cron; + + public boolean isEnabled() { + return enabled; + } + + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } + + public String getCron() { + return cron; + } + + public void setCron(String cron) { + this.cron = cron; + } +} diff --git a/src/main/java/ru/nbch/indicsearchservice/cron/CronChangeListener.java b/src/main/java/ru/nbch/indicsearchservice/cron/CronChangeListener.java new file mode 100644 index 0000000..26df3df --- /dev/null +++ b/src/main/java/ru/nbch/indicsearchservice/cron/CronChangeListener.java @@ -0,0 +1,19 @@ +package ru.nbch.indicsearchservice.cron; + +import org.springframework.context.event.EventListener; +import org.springframework.stereotype.Component; +import ru.nbch.indicsearchservice.event.CronChangedEvent; + +@Component +public class CronChangeListener { + private final CronJobManager cronJobManager; + + public CronChangeListener(CronJobManager cronJobManager) { + this.cronJobManager = cronJobManager; + } + + @EventListener + public void onCronChanged(CronChangedEvent event) { + cronJobManager.rescheduleNow(event.jobName(), event.newCron()); + } +} diff --git a/src/main/java/ru/nbch/indicsearchservice/cron/CronJob.java b/src/main/java/ru/nbch/indicsearchservice/cron/CronJob.java new file mode 100644 index 0000000..319e9a4 --- /dev/null +++ b/src/main/java/ru/nbch/indicsearchservice/cron/CronJob.java @@ -0,0 +1,5 @@ +package ru.nbch.indicsearchservice.cron; + +public interface CronJob { + CronJobName getCronJobName(); +} diff --git a/src/main/java/ru/nbch/indicsearchservice/cron/CronJobManager.java b/src/main/java/ru/nbch/indicsearchservice/cron/CronJobManager.java new file mode 100644 index 0000000..52841cd --- /dev/null +++ b/src/main/java/ru/nbch/indicsearchservice/cron/CronJobManager.java @@ -0,0 +1,78 @@ +package ru.nbch.indicsearchservice.cron; + +import jakarta.annotation.PostConstruct; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledFuture; +import java.util.function.Supplier; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.annotation.EnableScheduling; +import org.springframework.scheduling.support.CronTrigger; +import org.springframework.stereotype.Component; +import ru.nbch.indicsearchservice.config.settings.ApplicationSettings; +import ru.nbch.indicsearchservice.service.SettingsService; + +@Component +@EnableScheduling +public class CronJobManager { + private final TaskScheduler taskScheduler; + private final ApplicationSettings settings; + private final List cronJobs; + private final SettingsService settingsService; + + private final Map tasks = new ConcurrentHashMap<>(); + private final Map> futures = new ConcurrentHashMap<>(); + + public CronJobManager(TaskScheduler taskScheduler, + ApplicationSettings settings, + List cronJobs, + SettingsService settingsService) { + this.taskScheduler = taskScheduler; + this.settings = settings; + this.cronJobs = cronJobs; + this.settingsService = settingsService; + + for (CronJob cronJob : cronJobs) { + if (cronJob instanceof Runnable runnableCronJob) { + tasks.put(cronJob.getCronJobName(), runnableCronJob); + } + } + } + + @PostConstruct + public void scheduleAll() { + for (CronJob cronJob : cronJobs) { + if (cronJob instanceof Runnable) { + if (cronJob instanceof JobCleaner) { + Optional cron = settingsService.getOptionalSettingsValueByName("jobCleaner.cron"); + if (cron.isPresent()) { + settings.getJobCleaner().setCron(cron.get()); + schedule(cronJob.getCronJobName(), cron::get); + } + } else { + return; + } + } + } + } + + private void schedule(CronJobName jobName, Supplier cronSupplier) { + Runnable task = tasks.get(jobName); + CronTrigger trigger = new CronTrigger(cronSupplier.get()); + ScheduledFuture f = taskScheduler.schedule(task, trigger); + futures.put(jobName, f); + } + + public void rescheduleNow(CronJobName jobName, String newCron) { + ScheduledFuture f = futures.get(jobName); + if (f != null) { + f.cancel(false); + } + CronTrigger trigger = CronTrigger.resumeLenientExecution(newCron, Instant.now()); + ScheduledFuture nf = taskScheduler.schedule(tasks.get(jobName), trigger); + futures.put(jobName, nf); + } +} diff --git a/src/main/java/ru/nbch/indicsearchservice/cron/CronJobName.java b/src/main/java/ru/nbch/indicsearchservice/cron/CronJobName.java new file mode 100644 index 0000000..5187f07 --- /dev/null +++ b/src/main/java/ru/nbch/indicsearchservice/cron/CronJobName.java @@ -0,0 +1,5 @@ +package ru.nbch.indicsearchservice.cron; + +public enum CronJobName { + JOB_CLEANER +} diff --git a/src/main/java/ru/nbch/indicsearchservice/cron/JobCleaner.java b/src/main/java/ru/nbch/indicsearchservice/cron/JobCleaner.java new file mode 100644 index 0000000..c1c628a --- /dev/null +++ b/src/main/java/ru/nbch/indicsearchservice/cron/JobCleaner.java @@ -0,0 +1,97 @@ +package ru.nbch.indicsearchservice.cron; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.time.LocalDateTime; +import java.util.Iterator; +import java.util.Map; +import java.util.Optional; +import java.util.UUID; +import org.apache.commons.io.FilenameUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.stereotype.Component; +import ru.nbch.indicsearchservice.config.settings.ApplicationSettings; +import ru.nbch.indicsearchservice.exception.InternalException; +import ru.nbch.indicsearchservice.model.SearchJob; +import ru.nbch.indicsearchservice.service.SettingsService; + +@Component +public class JobCleaner implements Runnable, CronJob { + private static final Logger log = LoggerFactory.getLogger(JobCleaner.class); + + private final ApplicationSettings settings; + private final Map searchJobs; + + public JobCleaner(ApplicationSettings settings, + SettingsService settingsService, + @Qualifier("searchJobsMap") Map searchJobs) { + this.settings = settings; + this.searchJobs = searchJobs; + + Optional enabled = settingsService.getOptionalSettingsValueByName("jobCleaner.enabled"); + if (enabled.isPresent()) { + settings.getJobCleaner().setEnabled(Boolean.parseBoolean(enabled.get())); + } else { + log.warn("There is no jobCleaner.enabled in SETTINGS table!"); + } + } + + @Override + public void run() { + LocalDateTime hourBefore = LocalDateTime.now().minusMinutes(1); + + // Очищаем Map + Iterator> iterator = searchJobs.entrySet().iterator(); + while (iterator.hasNext()) { + Map.Entry entry = iterator.next(); + SearchJob searchJob = entry.getValue(); + + if (searchJob.getFinishedAt().isBefore(hourBefore)) { + iterator.remove(); + } + } + + // Очищаем папку + for (File file : Optional.ofNullable(settings.getStorageDirectory().listFiles()).orElse(new File[0])) { + String fileNameWithoutExtension = FilenameUtils.getBaseName(file.getName()); + String fileExtension = FilenameUtils.getExtension(file.getName()); + + UUID parsedUuid = null; + if (fileNameWithoutExtension.startsWith("AGGREGATED-")) { + try { + parsedUuid = UUID.fromString(fileNameWithoutExtension.replace("AGGREGATED-", "")); + } catch (Exception e) { + log.warn("Could not parse UUID from file name: {}, removing", fileNameWithoutExtension, e); + } + } else if (fileExtension.startsWith("json") || fileExtension.startsWith("xlsx")) { + try { + parsedUuid = UUID.fromString(fileNameWithoutExtension); + } catch (Exception e) { + log.warn("Could not parse UUID from file name: {}, removing", fileNameWithoutExtension, e); + } + } else { + try { + parsedUuid = UUID.fromString(fileNameWithoutExtension.substring(fileNameWithoutExtension.lastIndexOf('_') + 1)); + } catch (Exception e) { + log.warn("Could not parse UUID from file name: {}, removing", fileNameWithoutExtension, e); + } + } + + if (parsedUuid == null || !searchJobs.containsKey(parsedUuid)) { + try { + Files.deleteIfExists(file.toPath()); + } catch (IOException e) { + throw new InternalException("Failed to delete file " + file + " " + e); + } + } + } + } + + @Override + public CronJobName getCronJobName() { + return CronJobName.JOB_CLEANER; + } +} diff --git a/src/main/java/ru/nbch/indicsearchservice/event/CronChangedEvent.java b/src/main/java/ru/nbch/indicsearchservice/event/CronChangedEvent.java new file mode 100644 index 0000000..a8bdd20 --- /dev/null +++ b/src/main/java/ru/nbch/indicsearchservice/event/CronChangedEvent.java @@ -0,0 +1,9 @@ +package ru.nbch.indicsearchservice.event; + +import ru.nbch.indicsearchservice.cron.CronJobName; + +public record CronChangedEvent( + CronJobName jobName, + String newCron +) { +} diff --git a/src/main/java/ru/nbch/indicsearchservice/manager/IndicSearchJobManager.java b/src/main/java/ru/nbch/indicsearchservice/manager/IndicSearchJobManager.java index 70b304f..fbcf995 100644 --- a/src/main/java/ru/nbch/indicsearchservice/manager/IndicSearchJobManager.java +++ b/src/main/java/ru/nbch/indicsearchservice/manager/IndicSearchJobManager.java @@ -2,9 +2,8 @@ package ru.nbch.indicsearchservice.manager; import java.util.Map; import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.http.ResponseEntity; import org.springframework.stereotype.Component; import ru.nbch.indicsearchservice.dto.request.AddSearchJobRequestDto; @@ -15,11 +14,15 @@ import ru.nbch.indicsearchservice.processor.IndicSearchProcessor; @Component public class IndicSearchJobManager { - private final ExecutorService executor = Executors.newFixedThreadPool(5); - private final Map searchJobs = new ConcurrentHashMap<>(); + private final ExecutorService searchJobExecutor; + private final Map searchJobs; private final IndicSearchProcessor indicSearchProcessor; - public IndicSearchJobManager(IndicSearchProcessor indicSearchProcessor) { + public IndicSearchJobManager(@Qualifier("searchJobExecutor") ExecutorService searchJobExecutor, + @Qualifier("searchJobsMap") Map searchJobs, + IndicSearchProcessor indicSearchProcessor) { + this.searchJobExecutor = searchJobExecutor; + this.searchJobs = searchJobs; this.indicSearchProcessor = indicSearchProcessor; } @@ -34,7 +37,7 @@ public class IndicSearchJobManager { request.getMemberCode() ); searchJobs.put(uuid, newSearchJob); - executor.submit(() -> indicSearchProcessor.aggregateTables(newSearchJob)); + searchJobExecutor.submit(() -> indicSearchProcessor.aggregateTables(newSearchJob)); return uuid; } diff --git a/src/main/java/ru/nbch/indicsearchservice/processor/IndicSearchProcessor.java b/src/main/java/ru/nbch/indicsearchservice/processor/IndicSearchProcessor.java index 9d0025f..e68dfe4 100644 --- a/src/main/java/ru/nbch/indicsearchservice/processor/IndicSearchProcessor.java +++ b/src/main/java/ru/nbch/indicsearchservice/processor/IndicSearchProcessor.java @@ -170,9 +170,6 @@ public class IndicSearchProcessor { } catch (IOException e) { log.error("Failed to get temp excel file size!", e); try { - if (aggregatedFile != null) { - Files.deleteIfExists(aggregatedFile); - } Files.deleteIfExists(exportedFile); } catch (IOException ignored) { } @@ -185,18 +182,12 @@ public class IndicSearchProcessor { @Override public void close() throws IOException { super.close(); - if (aggregatedFile != null) { - Files.deleteIfExists(aggregatedFile); - } Files.deleteIfExists(exportedFile); } }); } catch (IOException e) { log.error("Failed to create exported file resource!", e); try { - if (aggregatedFile != null) { - Files.deleteIfExists(aggregatedFile); - } Files.deleteIfExists(exportedFile); } catch (IOException ignored) { } diff --git a/src/main/java/ru/nbch/indicsearchservice/service/SettingsService.java b/src/main/java/ru/nbch/indicsearchservice/service/SettingsService.java index 69db2e6..562024c 100644 --- a/src/main/java/ru/nbch/indicsearchservice/service/SettingsService.java +++ b/src/main/java/ru/nbch/indicsearchservice/service/SettingsService.java @@ -11,6 +11,8 @@ import java.util.Optional; import org.springframework.context.ApplicationEventPublisher; import org.springframework.stereotype.Service; import ru.nbch.indicsearchservice.config.settings.ApplicationSettings; +import ru.nbch.indicsearchservice.cron.CronJobName; +import ru.nbch.indicsearchservice.event.CronChangedEvent; import ru.nbch.indicsearchservice.event.DataSourceChangedEvent; import ru.nbch.indicsearchservice.exception.InternalException; import ru.nbch.indicsearchservice.model.Settings; @@ -19,6 +21,9 @@ import ru.nbch.indicsearchservice.util.DataSourceUpdater; @Service public class SettingsService { + private static final String STORAGE_SETTINGS = "indic-search-service-storage"; // todo: change key + private static final String JOB_CLEANER_SETTINGS = "jobCleaner"; + private final SettingsMapper settingsMapper; private final ApplicationSettings localSettings; private final ApplicationEventPublisher publisher; @@ -31,12 +36,6 @@ public class SettingsService { this.publisher = publisher; } - public String getSettingsValueByName(String name) { - return settingsMapper.findByName(name) - .orElseThrow(() -> new InternalException("Setting with key: " + name + " was not found in database!")) - .getValue(); - } - public Map> getSettingsValuesByNames(List names) { if (names == null || names.isEmpty()) { return Map.of(); @@ -68,17 +67,29 @@ public class SettingsService { } public void updateLocalSettings() { - Optional newStorageDirectory = getOptionalSettingsValueByName("indic-search-service-storage"); // todo: change key + Map> newSettings = getSettingsValuesByNames(List.of( + STORAGE_SETTINGS, + JOB_CLEANER_SETTINGS + ".enabled", + JOB_CLEANER_SETTINGS + ".cron" + )); + + if (newSettings.get(STORAGE_SETTINGS).isPresent()) { + String newStorageDirectory = newSettings.get(STORAGE_SETTINGS).get(); + try { + Files.createDirectories(Paths.get(newStorageDirectory)); + } catch (IOException e) { + throw new InternalException("Failed to create directory " + newStorageDirectory + " " + e); + } + localSettings.setStorageDirectory(new File(newStorageDirectory)); + } + + if (newSettings.get(JOB_CLEANER_SETTINGS + ".enabled").isPresent()) { + localSettings.getJobCleaner().setEnabled(Boolean.parseBoolean(newSettings.get(JOB_CLEANER_SETTINGS + ".enabled").get())); + } + if (newSettings.get(JOB_CLEANER_SETTINGS + ".cron").isPresent()) { + publisher.publishEvent(new CronChangedEvent(CronJobName.JOB_CLEANER, newSettings.get(JOB_CLEANER_SETTINGS + ".cron").get())); + } publisher.publishEvent(new DataSourceChangedEvent(DataSourceUpdater.DataSourceType.INDIC)); - - if (newStorageDirectory.isPresent()) { - try { - Files.createDirectories(Paths.get(newStorageDirectory.get())); - } catch (IOException e) { - throw new InternalException("Failed to create directory " + newStorageDirectory.get() + " " + e); - } - localSettings.setStorageDirectory(new File(newStorageDirectory.get())); - } } }