From 87b9e4db5577cf149588d2b747603c9499562a66 Mon Sep 17 00:00:00 2001 From: Pierre Date: Tue, 9 Jun 2020 16:11:05 +0200 Subject: [PATCH] SONAR-13444 purge CE indexation tasks when triggering a disaster recovery --- .../java/org/sonar/db/ce/CeActivityDao.java | 4 ++ .../org/sonar/db/ce/CeActivityMapper.java | 4 ++ .../org/sonar/db/ce/CeActivityMapper.xml | 16 +++++- .../org/sonar/db/ce/CeActivityDaoTest.java | 56 +++++++++++++++++++ .../issue/index/AsyncIssueIndexingImpl.java | 29 +++++++++- .../index/AsyncIssueIndexingImplTest.java | 41 ++++++++++++++ 6 files changed, 145 insertions(+), 5 deletions(-) diff --git a/server/sonar-db-dao/src/main/java/org/sonar/db/ce/CeActivityDao.java b/server/sonar-db-dao/src/main/java/org/sonar/db/ce/CeActivityDao.java index 5b368d61fda..af9b1e4667f 100644 --- a/server/sonar-db-dao/src/main/java/org/sonar/db/ce/CeActivityDao.java +++ b/server/sonar-db-dao/src/main/java/org/sonar/db/ce/CeActivityDao.java @@ -61,6 +61,10 @@ public class CeActivityDao implements Dao { return mapper(dbSession).selectOlderThan(beforeDate); } + public List selectByTaskType(DbSession dbSession, String taskType) { + return mapper(dbSession).selectByTaskType(taskType); + } + public void deleteByUuids(DbSession dbSession, Set uuids) { executeLargeUpdates(uuids, mapper(dbSession)::deleteByUuids); } diff --git a/server/sonar-db-dao/src/main/java/org/sonar/db/ce/CeActivityMapper.java b/server/sonar-db-dao/src/main/java/org/sonar/db/ce/CeActivityMapper.java index c93ac7ab220..2b65ede52c3 100644 --- a/server/sonar-db-dao/src/main/java/org/sonar/db/ce/CeActivityMapper.java +++ b/server/sonar-db-dao/src/main/java/org/sonar/db/ce/CeActivityMapper.java @@ -46,4 +46,8 @@ public interface CeActivityMapper { @CheckForNull CeActivityDto selectLastByComponentUuidAndTaskType(@Param("componentUuid") String componentUuid, @Param("taskType") String taskType); + + short hasAnyFailedIssueSyncTask(); + + List selectByTaskType(@Param("taskType") String taskType); } diff --git a/server/sonar-db-dao/src/main/resources/org/sonar/db/ce/CeActivityMapper.xml b/server/sonar-db-dao/src/main/resources/org/sonar/db/ce/CeActivityMapper.xml index 3448087ba25..1beca1ae98d 100644 --- a/server/sonar-db-dao/src/main/resources/org/sonar/db/ce/CeActivityMapper.xml +++ b/server/sonar-db-dao/src/main/resources/org/sonar/db/ce/CeActivityMapper.xml @@ -19,7 +19,7 @@ (select count(1) from ce_task_message ctm where ctm.task_uuid = ca.uuid) as warningCount - + ca.uuid, ca.task_type as taskType, ca.component_uuid as componentUuid, @@ -39,7 +39,11 @@ ca.main_is_last_key as mainIsLastKey, ca.execution_time_ms as executionTimeMs, ca.error_message as errorMessage, - ca.error_type as errorType, + ca.error_type as errorType + + + + , @@ -244,4 +248,12 @@ and ca.is_last = ${_true} + + diff --git a/server/sonar-db-dao/src/test/java/org/sonar/db/ce/CeActivityDaoTest.java b/server/sonar-db-dao/src/test/java/org/sonar/db/ce/CeActivityDaoTest.java index 64d40a3d861..0e10b7b101b 100644 --- a/server/sonar-db-dao/src/test/java/org/sonar/db/ce/CeActivityDaoTest.java +++ b/server/sonar-db-dao/src/test/java/org/sonar/db/ce/CeActivityDaoTest.java @@ -742,6 +742,62 @@ public class CeActivityDaoTest { assertThat(result).isEmpty(); } + @Test + public void selectByTaskType() { + insert("TASK_1", CeTaskTypes.REPORT, MAINCOMPONENT_1, SUCCESS); + insert("TASK_2", CeTaskTypes.BRANCH_ISSUE_SYNC, MAINCOMPONENT_1, SUCCESS); + db.commit(); + + assertThat(underTest.selectByTaskType(db.getSession(), CeTaskTypes.REPORT)) + .extracting("uuid") + .containsExactly("TASK_1"); + assertThat(underTest.selectByTaskType(db.getSession(), CeTaskTypes.BRANCH_ISSUE_SYNC)) + .extracting("uuid") + .containsExactly("TASK_2"); + + assertThat(underTest.selectByTaskType(db.getSession(), "unknown-type")).isEmpty(); + } + + @Test + public void hasAnyFailedIssueSyncTask() { + assertThat(underTest.hasAnyFailedIssueSyncTask(db.getSession())).isFalse(); + + insert("TASK_1", REPORT, MAINCOMPONENT_1, SUCCESS); + insert("TASK_2", REPORT, MAINCOMPONENT_1, FAILED); + + ProjectDto projectDto1 = db.components() + .insertPrivateProjectDto(db.getDefaultOrganization(), branchDto -> branchDto.setNeedIssueSync(false)); + insert("TASK_3", CeTaskTypes.BRANCH_ISSUE_SYNC, projectDto1.getUuid(), projectDto1.getUuid(), SUCCESS); + + ProjectDto projectDto2 = db.components() + .insertPrivateProjectDto(db.getDefaultOrganization(), branchDto -> branchDto.setNeedIssueSync(false)); + insert("TASK_4", CeTaskTypes.BRANCH_ISSUE_SYNC, projectDto2.getUuid(), projectDto2.getUuid(), SUCCESS); + + assertThat(underTest.hasAnyFailedIssueSyncTask(db.getSession())).isFalse(); + + ProjectDto projectDto3 = db.components().insertPrivateProjectDto(db.getDefaultOrganization(), branchDto -> branchDto.setNeedIssueSync(false)); + insert("TASK_5", CeTaskTypes.BRANCH_ISSUE_SYNC, projectDto3.getUuid(), projectDto3.getUuid(), SUCCESS); + + BranchDto projectBranch = db.components() + .insertProjectBranch(projectDto3, branchDto -> branchDto.setNeedIssueSync(true)); + + insert("TASK_6", CeTaskTypes.BRANCH_ISSUE_SYNC, projectBranch.getUuid(), projectDto3.getUuid(), FAILED); + + //failed task and project branch still exists and need sync + assertThat(underTest.hasAnyFailedIssueSyncTask(db.getSession())).isTrue(); + + //assume branch has been re-analysed + db.getDbClient().branchDao().updateNeedIssueSync(db.getSession(), projectBranch.getUuid(), false); + + assertThat(underTest.hasAnyFailedIssueSyncTask(db.getSession())).isFalse(); + + //assume branch has been deleted + db.getDbClient().purgeDao().deleteBranch(db.getSession(), projectBranch.getUuid()); + + //associated branch does not exist, so there is no failures - either it has been deleted or purged or reanalysed + assertThat(underTest.hasAnyFailedIssueSyncTask(db.getSession())).isFalse(); + } + private CeActivityDto insert(String uuid, String type, @Nullable String mainComponentUuid, CeActivityDto.Status status) { return insert(uuid, type, mainComponentUuid, mainComponentUuid, status); } diff --git a/server/sonar-webserver-core/src/main/java/org/sonar/server/issue/index/AsyncIssueIndexingImpl.java b/server/sonar-webserver-core/src/main/java/org/sonar/server/issue/index/AsyncIssueIndexingImpl.java index ea1415756be..7dc3d84359e 100644 --- a/server/sonar-webserver-core/src/main/java/org/sonar/server/issue/index/AsyncIssueIndexingImpl.java +++ b/server/sonar-webserver-core/src/main/java/org/sonar/server/issue/index/AsyncIssueIndexingImpl.java @@ -20,6 +20,7 @@ package org.sonar.server.issue.index; import java.util.List; +import java.util.Set; import java.util.stream.Collectors; import org.sonar.api.server.ServerSide; import org.sonar.api.utils.log.Logger; @@ -28,6 +29,8 @@ import org.sonar.ce.queue.CeQueue; import org.sonar.ce.queue.CeTaskSubmit; import org.sonar.db.DbClient; import org.sonar.db.DbSession; +import org.sonar.db.ce.CeActivityDto; +import org.sonar.db.ce.CeQueueDto; import org.sonar.db.component.BranchDto; import static java.util.Collections.emptyMap; @@ -51,10 +54,10 @@ public class AsyncIssueIndexingImpl implements AsyncIssueIndexing { try (DbSession dbSession = dbClient.openSession(false)) { - dbClient.branchDao().updateAllNeedIssueSync(dbSession); - - // TODO check the queue for any BRANCH_ISSUE_SYNC existing task pending + // remove already existing indexation task, if any + removeExistingIndexationTasks(dbSession); + dbClient.branchDao().updateAllNeedIssueSync(dbSession); List branchInNeedOfIssueSync = dbClient.branchDao().selectBranchNeedingIssueSync(dbSession); if (branchInNeedOfIssueSync.isEmpty()) { @@ -75,6 +78,26 @@ public class AsyncIssueIndexingImpl implements AsyncIssueIndexing { } } + private void removeExistingIndexationTasks(DbSession dbSession) { + List uuids = dbClient.ceQueueDao().selectAllInAscOrder(dbSession).stream() + .filter(p -> p.getTaskType().equals(BRANCH_ISSUE_SYNC)) + .map(CeQueueDto::getUuid) + .collect(Collectors.toList()); + LOG.info(String.format("%s pending indexation task found to be deleted...", uuids.size())); + for (String uuid : uuids) { + dbClient.ceQueueDao().deleteByUuid(dbSession, uuid); + } + dbSession.commit(); + + Set ceUuids = dbClient.ceActivityDao().selectByTaskType(dbSession, BRANCH_ISSUE_SYNC).stream() + .map(CeActivityDto::getUuid) + .collect(Collectors.toSet()); + LOG.info(String.format("%s completed indexation task found to be deleted...", uuids.size())); + dbClient.ceActivityDao().deleteByUuids(dbSession, ceUuids); + dbSession.commit(); + LOG.info("Indexation task deletion complete."); + } + private CeTaskSubmit buildTaskSubmit(BranchDto branch) { return ceQueue.prepareSubmit() .setType(BRANCH_ISSUE_SYNC) diff --git a/server/sonar-webserver-core/src/test/java/org/sonar/server/issue/index/AsyncIssueIndexingImplTest.java b/server/sonar-webserver-core/src/test/java/org/sonar/server/issue/index/AsyncIssueIndexingImplTest.java index 996c544e77e..852e62e3694 100644 --- a/server/sonar-webserver-core/src/test/java/org/sonar/server/issue/index/AsyncIssueIndexingImplTest.java +++ b/server/sonar-webserver-core/src/test/java/org/sonar/server/issue/index/AsyncIssueIndexingImplTest.java @@ -32,6 +32,9 @@ import org.sonar.core.util.SequenceUuidFactory; import org.sonar.core.util.UuidFactory; import org.sonar.db.DbClient; import org.sonar.db.DbTester; +import org.sonar.db.ce.CeActivityDto; +import org.sonar.db.ce.CeActivityDto.Status; +import org.sonar.db.ce.CeQueueDto; import org.sonar.db.component.BranchDto; import static org.assertj.core.api.Assertions.assertThat; @@ -40,6 +43,8 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import static org.sonar.db.ce.CeTaskTypes.BRANCH_ISSUE_SYNC; +import static org.sonar.db.ce.CeTaskTypes.REPORT; import static org.sonar.db.component.BranchType.BRANCH; public class AsyncIssueIndexingImplTest { @@ -89,4 +94,40 @@ public class AsyncIssueIndexingImplTest { assertThat(logTester.logs(LoggerLevel.INFO)).contains("No branch found in need of issue sync"); } + @Test + public void remove_existing_indexation_task() { + CeQueueDto reportTask = new CeQueueDto(); + reportTask.setUuid("uuid_1"); + reportTask.setTaskType(REPORT); + dbClient.ceQueueDao().insert(dbTester.getSession(), reportTask); + + CeActivityDto reportActivity = new CeActivityDto(reportTask); + reportActivity.setStatus(Status.SUCCESS); + dbClient.ceActivityDao().insert(dbTester.getSession(), reportActivity); + CeQueueDto task = new CeQueueDto(); + task.setUuid("uuid_2"); + task.setTaskType(BRANCH_ISSUE_SYNC); + dbClient.ceQueueDao().insert(dbTester.getSession(), task); + + CeActivityDto activityDto = new CeActivityDto(task); + activityDto.setStatus(Status.SUCCESS); + dbClient.ceActivityDao().insert(dbTester.getSession(), activityDto); + + dbTester.commit(); + + underTest.triggerOnIndexCreation(); + + assertThat(dbClient.ceQueueDao().selectAllInAscOrder(dbTester.getSession())).extracting("uuid") + .containsExactly(reportTask.getUuid()); + assertThat(dbClient.ceActivityDao().selectByTaskType(dbTester.getSession(), BRANCH_ISSUE_SYNC)).isEmpty(); + + assertThat(dbClient.ceActivityDao().selectByTaskType(dbTester.getSession(), REPORT)).hasSize(1); + + assertThat(logTester.logs(LoggerLevel.INFO)) + .contains( + "1 pending indexation task found to be deleted...", + "1 completed indexation task found to be deleted...", + "Indexation task deletion complete."); + } + } -- 2.39.5