Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Added code to remove exceptions every Saturday from history that is t… #1453

Merged
merged 2 commits into from
Mar 4, 2024
Merged

Added code to remove exceptions every Saturday from history that is t… #1453

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
046adb0
Added code to remove exceptions every Saturday from history that is t…
oblodgett Mar 1, 2024
5f6b03f
Update JobScheduler.java
oblodgett Mar 4, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
120 changes: 80 additions & 40 deletions src/main/java/org/alliancegenome/curation_api/jobs/JobScheduler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,12 @@

import java.io.File;
import java.time.ZonedDateTime;
import java.util.Collections;
import java.util.Comparator;

import org.alliancegenome.curation_api.dao.loads.BulkLoadDAO;
import org.alliancegenome.curation_api.dao.loads.BulkLoadFileDAO;
import org.alliancegenome.curation_api.dao.loads.BulkLoadFileExceptionDAO;
import org.alliancegenome.curation_api.dao.loads.BulkLoadGroupDAO;
import org.alliancegenome.curation_api.enums.JobStatus;
import org.alliancegenome.curation_api.jobs.events.PendingBulkLoadFileJobEvent;
Expand All @@ -14,9 +17,12 @@
import org.alliancegenome.curation_api.jobs.util.SlackNotifier;
import org.alliancegenome.curation_api.model.entities.bulkloads.BulkLoad;
import org.alliancegenome.curation_api.model.entities.bulkloads.BulkLoadFile;
import org.alliancegenome.curation_api.model.entities.bulkloads.BulkLoadFileException;
import org.alliancegenome.curation_api.model.entities.bulkloads.BulkLoadFileHistory;
import org.alliancegenome.curation_api.model.entities.bulkloads.BulkLoadGroup;
import org.alliancegenome.curation_api.model.entities.bulkloads.BulkScheduledLoad;
import org.alliancegenome.curation_api.response.SearchResponse;
import org.alliancegenome.curation_api.util.ProcessDisplayHelper;
import org.eclipse.microprofile.config.inject.ConfigProperty;

import com.cronutils.model.Cron;
Expand All @@ -37,48 +43,41 @@
@ApplicationScoped
public class JobScheduler {

//@Inject
//EventBus bus;

@Inject
Event<PendingBulkLoadJobEvent> pendingJobEvents;

@Inject
Event<StartedBulkLoadJobEvent> startedJobEvents;

@Inject
Event<StartedBulkLoadFileJobEvent> startedFileJobEvents;

@Inject
BulkLoadFileDAO bulkLoadFileDAO;
@Inject
BulkLoadGroupDAO groupDAO;
@Inject
BulkLoadDAO bulkLoadDAO;
@Inject
SlackNotifier slackNotifier;

@ConfigProperty(name = "bulk.data.loads.schedulingEnabled")
Boolean loadSchedulingEnabled;

@ConfigProperty(name = "reindex.schedulingEnabled", defaultValue = "false")
Boolean reindexSchedulingEnabled;
// @Inject
// EventBus bus;

@Inject Event<PendingBulkLoadJobEvent> pendingJobEvents;

@Inject Event<StartedBulkLoadJobEvent> startedJobEvents;

@Inject Event<StartedBulkLoadFileJobEvent> startedFileJobEvents;

@Inject BulkLoadFileDAO bulkLoadFileDAO;
@Inject BulkLoadGroupDAO groupDAO;
@Inject BulkLoadDAO bulkLoadDAO;
@Inject BulkLoadFileExceptionDAO bulkLoadFileExceptionDAO;
@Inject SlackNotifier slackNotifier;

@ConfigProperty(name = "bulk.data.loads.schedulingEnabled") Boolean loadSchedulingEnabled;

@ConfigProperty(name = "reindex.schedulingEnabled", defaultValue = "false") Boolean reindexSchedulingEnabled;

private ZonedDateTime lastCheck = null;

@PostConstruct
public void init() {
// Set any running jobs to failed as the server has restarted
//Log.info("Init: ");
// Log.info("Init: ");
SearchResponse<BulkLoadGroup> groups = groupDAO.findAll();
for (BulkLoadGroup g : groups.getResults()) {
if (g.getLoads().size() > 0) {
for (BulkLoad b : g.getLoads()) {
for (BulkLoadFile bf : b.getLoadFiles()) {
if (bf.getBulkloadStatus() == null || bf.getBulkloadStatus().isRunning() || bf.getBulkloadStatus().isStarted() || bf.getLocalFilePath() != null) {
if(bf.getLocalFilePath() != null) {
if (bf.getLocalFilePath() != null) {
File file = new File(bf.getLocalFilePath());
if(file.exists()) file.delete();
if (file.exists())
file.delete();
}
bf.setLocalFilePath(null);
bf.setErrorMessage("Failed due to server start up: Process never finished before the server restarted");
Expand Down Expand Up @@ -150,41 +149,82 @@ public void scheduleCronGroupJobs() {
}

public void pendingJobs(@Observes PendingBulkLoadJobEvent event) {
//Log.info("pendingJobs: " + event.getId());
// Log.info("pendingJobs: " + event.getId());
BulkLoad load = bulkLoadDAO.find(event.getId());
if(load != null) {
if (load != null) {
if (load.getBulkloadStatus().isPending()) {
load.setBulkloadStatus(load.getBulkloadStatus().getNextStatus());
bulkLoadDAO.merge(load);
//Log.info("Firing Start Event: " + load.getId());
// Log.info("Firing Start Event: " + load.getId());
startedJobEvents.fire(new StartedBulkLoadJobEvent(load.getId()));
}
}
}

public void pendingFileJobs(@Observes PendingBulkLoadFileJobEvent event) {
//Log.info("pendingFileJobs: " + event.getId());
// Log.info("pendingFileJobs: " + event.getId());
BulkLoadFile fileLoad = bulkLoadFileDAO.find(event.getId());
if(fileLoad != null) {
if (fileLoad != null) {
if (fileLoad.getBulkloadStatus().isPending()) {
fileLoad.setBulkloadStatus(fileLoad.getBulkloadStatus().getNextStatus());
bulkLoadFileDAO.merge(fileLoad);
//Log.info("Firing Start Event: " + fileLoad.getId());
// Log.info("Firing Start Event: " + fileLoad.getId());
startedFileJobEvents.fire(new StartedBulkLoadFileJobEvent(fileLoad.getId()));
}
}
}

@Scheduled(cron = "0 0 0 ? * SUN")

@Scheduled(cron = "0 0 0 ? * SAT")
public void cleanUpFileExceptions() {
SearchResponse<BulkLoadGroup> groups = groupDAO.findAll();

Comparator<BulkLoadFile> c = new Comparator<BulkLoadFile>() {
public int compare(BulkLoadFile blf1, BulkLoadFile blf2) {
if (blf1.getDateLastLoaded() == null && blf2.getDateLastLoaded() == null) { return 0; }
if (blf1.getDateLastLoaded() == null) { return 1; }
if (blf2.getDateLastLoaded() == null) { return -1; }
return blf2.getDateLastLoaded().compareTo(blf1.getDateLastLoaded());
}
};

int historyExceptionsToKeep = 3;

ProcessDisplayHelper ph = new ProcessDisplayHelper();
ph.startProcess("Bulk Load File History Exception Cleanup");

for (BulkLoadGroup g : groups.getResults()) {
Log.debug("Group Clean Up: " + g.getName());
if (g.getLoads().size() > 0) {
for (BulkLoad b : g.getLoads()) {
Log.debug("Load Clean Up: " + b.getName());
Collections.sort(b.getLoadFiles(), c);
if (b.getLoadFiles().size() > historyExceptionsToKeep) {
for (int i = historyExceptionsToKeep; i < b.getLoadFiles().size(); i++) {
Log.debug("File Clean Up: " + b.getLoadFiles().get(i).getDateLastLoaded());
for(BulkLoadFileHistory h: b.getLoadFiles().get(i).getHistory()) {
for(BulkLoadFileException e: h.getExceptions()) {
bulkLoadFileExceptionDAO.remove(e.getId());
ph.progressProcess();
}
}
}
}
}
}
}
ph.finishProcess();
}

@Scheduled(cron = "0 0 0 ? * SAT")
public void runMassIndexerEverything() {
// Not sure what is going to happen when this time's out but should run anyway
// Defaults taken from the API endpoint
// DAO used doesn't matter they all have this method

if(reindexSchedulingEnabled){
if (reindexSchedulingEnabled) {
Log.info("Scheduled mass reindexing initiated.");
bulkLoadDAO.reindexEverything(1000, 10000, 0, 4, 7200, 1);
}
else{
} else {
Log.info("Scheduled mass reindexing not initiated (reindex scheduling not enabled).");
}
}
Expand Down
Loading