SONAR-19084 periodically schedule the GitHub sync task.

6 files changed, 66 insertions, 39 deletions

diff --git a/server/sonar-ce-common/src/it/java/org/sonar/ce/queue/CeQueueImplIT.java b/server/sonar-ce-common/src/it/java/org/sonar/ce/queue/CeQueueImplIT.java
index 0823582b04d..d89b20a568b 100644
--- a/server/sonar-ce-common/src/it/java/org/sonar/ce/queue/CeQueueImplIT.java
+++ b/server/sonar-ce-common/src/it/java/org/sonar/ce/queue/CeQueueImplIT.java
@@ -56,6 +56,7 @@ import static org.assertj.core.api.Assertions.tuple;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 import static org.sonar.ce.queue.CeQueue.SubmitOption.UNIQUE_QUEUE_PER_MAIN_COMPONENT;
+import static org.sonar.ce.queue.CeQueue.SubmitOption.UNIQUE_QUEUE_PER_TASK_TYPE;
 
 public class CeQueueImplIT {
 
@@ -209,6 +210,22 @@ public class CeQueueImplIT {
   }
 
   @Test
+  public void submit_with_UNIQUE_QUEUE_PER_TASK_TYPE_does_not_create_task_when_there_is_a_task_with_the_same_type() {
+    String mainComponentUuid = randomAlphabetic(5);
+    CeTaskSubmit taskSubmit = createTaskSubmit("some type", newComponent(mainComponentUuid), null);
+    String[] uuids = IntStream.range(0, 2 + new Random().nextInt(5))
+      .mapToObj(i -> insertPendingInQueue(newComponent(mainComponentUuid)))
+      .map(CeQueueDto::getUuid)
+      .toArray(String[]::new);
+    Optional<CeTask> task = underTest.submit(taskSubmit, UNIQUE_QUEUE_PER_TASK_TYPE);
+
+    assertThat(task).isEmpty();
+    assertThat(db.getDbClient().ceQueueDao().selectAllInAscOrder(db.getSession()))
+      .extracting(CeQueueDto::getUuid)
+      .containsOnly(uuids);
+  }
+
+  @Test
   public void massSubmit_returns_tasks_for_each_CeTaskSubmit_populated_from_CeTaskSubmit_and_creates_CeQueue_row_for_each() {
     String mainComponentUuid = randomAlphabetic(10);
     CeTaskSubmit taskSubmit1 = createTaskSubmit(CeTaskTypes.REPORT, newComponent(mainComponentUuid), "submitter uuid");
diff --git a/server/sonar-ce-common/src/main/java/org/sonar/ce/queue/CeQueue.java b/server/sonar-ce-common/src/main/java/org/sonar/ce/queue/CeQueue.java
index 10b649a1298..adee3c00e93 100644
--- a/server/sonar-ce-common/src/main/java/org/sonar/ce/queue/CeQueue.java
+++ b/server/sonar-ce-common/src/main/java/org/sonar/ce/queue/CeQueue.java
@@ -124,7 +124,8 @@ public interface CeQueue {
   int clear();
 
   enum SubmitOption {
-    UNIQUE_QUEUE_PER_MAIN_COMPONENT
+    UNIQUE_QUEUE_PER_MAIN_COMPONENT,
+    UNIQUE_QUEUE_PER_TASK_TYPE
   }
 
   enum WorkersPauseStatus {
diff --git a/server/sonar-ce-common/src/main/java/org/sonar/ce/queue/CeQueueImpl.java b/server/sonar-ce-common/src/main/java/org/sonar/ce/queue/CeQueueImpl.java
index b124b509c02..e0af41a2682 100644
--- a/server/sonar-ce-common/src/main/java/org/sonar/ce/queue/CeQueueImpl.java
+++ b/server/sonar-ce-common/src/main/java/org/sonar/ce/queue/CeQueueImpl.java
@@ -46,6 +46,7 @@ import org.sonar.db.DbSession;
 import org.sonar.db.ce.CeActivityDto;
 import org.sonar.db.ce.CeQueueDto;
 import org.sonar.db.ce.CeTaskCharacteristicDto;
+import org.sonar.db.ce.CeTaskQuery;
 import org.sonar.db.ce.DeleteIf;
 import org.sonar.db.component.ComponentDto;
 import org.sonar.db.user.UserDto;
@@ -55,7 +56,6 @@ import org.sonar.server.property.InternalProperties;
 import static com.google.common.base.Preconditions.checkState;
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.singleton;
-import static java.util.Optional.of;
 import static java.util.Optional.ofNullable;
 import static org.sonar.ce.queue.CeQueue.SubmitOption.UNIQUE_QUEUE_PER_MAIN_COMPONENT;
 import static org.sonar.core.util.stream.MoreCollectors.toEnumSet;
@@ -95,24 +95,39 @@ public class CeQueueImpl implements CeQueue {
 
   private Optional<CeTask> submit(CeTaskSubmit submission, EnumSet<SubmitOption> submitOptions) {
     try (DbSession dbSession = dbClient.openSession(false)) {
-      if (submitOptions.contains(UNIQUE_QUEUE_PER_MAIN_COMPONENT)
-        && submission.getComponent()
-          .map(component -> dbClient.ceQueueDao().countByStatusAndMainComponentUuid(dbSession, PENDING, component.getMainComponentUuid()) > 0)
-          .orElse(false)) {
-        return Optional.empty();
-      }
-      CeQueueDto taskDto = addToQueueInDb(dbSession, submission);
+      Optional<CeTask> ceTask = submit(dbSession, submission, submitOptions);
       dbSession.commit();
+      return ceTask;
+    }
+  }
 
-      Map<String, ComponentDto> componentsByUuid = loadComponentDtos(dbSession, taskDto);
-      if (componentsByUuid.isEmpty()) {
-        return of(convertToTask(dbSession, taskDto, submission.getCharacteristics(), null, null));
+  private Optional<CeTask> submit(DbSession dbSession, CeTaskSubmit submission, EnumSet<SubmitOption> submitOptions) {
+    CeTaskQuery query = new CeTaskQuery();
+    for (SubmitOption option : submitOptions) {
+      switch (option) {
+        case UNIQUE_QUEUE_PER_MAIN_COMPONENT -> submission.getComponent()
+          .flatMap(component -> Optional.ofNullable(component.getMainComponentUuid()))
+          .ifPresent(mainComponentUuid -> query.setMainComponentUuid(mainComponentUuid).setStatuses(List.of(PENDING.name())));
+        case UNIQUE_QUEUE_PER_TASK_TYPE -> query.setType(submission.getType());
       }
+    }
+
+    boolean queryNonEmpty = query.getMainComponentUuids() != null || query.getStatuses() != null || query.getType() != null;
+    if (queryNonEmpty && dbClient.ceQueueDao().countByQuery(dbSession, query) > 0) {
+      return Optional.empty();
+    }
+    CeQueueDto inserted = addToQueueInDb(dbSession, submission);
+    return Optional.of(convertQueueDtoToTask(dbSession, inserted, submission));
+  }
 
-      return of(convertToTask(dbSession, taskDto, submission.getCharacteristics(),
-        ofNullable(taskDto.getComponentUuid()).map(componentsByUuid::get).orElse(null),
-        ofNullable(taskDto.getMainComponentUuid()).map(componentsByUuid::get).orElse(null)));
+  private CeTask convertQueueDtoToTask(DbSession dbSession, CeQueueDto queueDto, CeTaskSubmit submission) {
+    Map<String, ComponentDto> componentsByUuid = loadComponentDtos(dbSession, queueDto);
+    if (componentsByUuid.isEmpty()) {
+      return convertToTask(dbSession, queueDto, submission.getCharacteristics(), null, null);
     }
+    return convertToTask(dbSession, queueDto, submission.getCharacteristics(),
+      ofNullable(queueDto.getComponentUuid()).map(componentsByUuid::get).orElse(null),
+      ofNullable(queueDto.getMainComponentUuid()).map(componentsByUuid::get).orElse(null));
   }
 
   Map<String, ComponentDto> loadComponentDtos(DbSession dbSession, CeQueueDto taskDto) {
@@ -183,7 +198,7 @@ public class CeQueueImpl implements CeQueue {
     return Arrays.stream(options).collect(toEnumSet(SubmitOption.class));
   }
 
-  private CeQueueDto addToQueueInDb(DbSession dbSession, CeTaskSubmit submission) {
+  private void insertCharacteristics(DbSession dbSession, CeTaskSubmit submission) {
     for (Map.Entry<String, String> characteristic : submission.getCharacteristics().entrySet()) {
       CeTaskCharacteristicDto characteristicDto = new CeTaskCharacteristicDto();
       characteristicDto.setUuid(uuidFactory.create());
@@ -192,6 +207,10 @@ public class CeQueueImpl implements CeQueue {
       characteristicDto.setValue(characteristic.getValue());
       dbClient.ceTaskCharacteristicsDao().insert(dbSession, characteristicDto);
     }
+  }
+
+  private CeQueueDto addToQueueInDb(DbSession dbSession, CeTaskSubmit submission) {
+    insertCharacteristics(dbSession, submission);
 
     CeQueueDto dto = new CeQueueDto();
     dto.setUuid(submission.getUuid());
@@ -262,7 +281,7 @@ public class CeQueueImpl implements CeQueue {
     updateExecutionFields(activityDto);
     remove(dbSession, task, activityDto);
   }
-  
+
   protected long updateExecutionFields(CeActivityDto activityDto) {
     Long startedAt = activityDto.getStartedAt();
     if (startedAt == null) {
@@ -354,7 +373,6 @@ public class CeQueueImpl implements CeQueue {
       .setCharacteristics(characteristics)
       .setSubmitter(resolveSubmitter(dbSession, taskDto.getSubmitterUuid()));
 
-
     String componentUuid = taskDto.getComponentUuid();
     if (component != null) {
       builder.setComponent(new CeTask.Component(component.uuid(), component.getKey(), component.name()));
diff --git a/server/sonar-db-dao/src/main/java/org/sonar/db/Pagination.java b/server/sonar-db-dao/src/main/java/org/sonar/db/Pagination.java
index d4a68eaddaf..bf1f260add7 100644
--- a/server/sonar-db-dao/src/main/java/org/sonar/db/Pagination.java
+++ b/server/sonar-db-dao/src/main/java/org/sonar/db/Pagination.java
@@ -27,6 +27,8 @@ import static com.google.common.base.Preconditions.checkArgument;
 public final class Pagination {
   private static final Pagination ALL = new Builder(1).andSize(Integer.MAX_VALUE);
 
+  private static final Pagination FIRST = new Builder(1).andSize(1);
+
   private final int page;
   private final int pageSize;
 
@@ -39,6 +41,10 @@ public final class Pagination {
     return ALL;
   }
 
+  public static Pagination first() {
+    return FIRST;
+  }
+
   public static Builder forPage(int page) {
     return new Builder(page);
   }
diff --git a/server/sonar-server-common/src/main/java/org/sonar/server/util/AbstractStoppableScheduledExecutorServiceImpl.java b/server/sonar-server-common/src/main/java/org/sonar/server/util/AbstractStoppableScheduledExecutorServiceImpl.java
index 7699f9842d6..f4ab38e716a 100644
--- a/server/sonar-server-common/src/main/java/org/sonar/server/util/AbstractStoppableScheduledExecutorServiceImpl.java
+++ b/server/sonar-server-common/src/main/java/org/sonar/server/util/AbstractStoppableScheduledExecutorServiceImpl.java
@@ -24,9 +24,9 @@ import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 
-public class AbstractStoppableScheduledExecutorServiceImpl<T extends ScheduledExecutorService> extends AbstractStoppableExecutorService<T>
+public abstract class AbstractStoppableScheduledExecutorServiceImpl<T extends ScheduledExecutorService> extends AbstractStoppableExecutorService<T>
   implements StoppableScheduledExecutorService {
-  public AbstractStoppableScheduledExecutorServiceImpl(T delegate) {
+  protected AbstractStoppableScheduledExecutorServiceImpl(T delegate) {
     super(delegate);
   }
 
diff --git a/server/sonar-webserver-core/src/main/java/org/sonar/server/telemetry/TelemetryDaemon.java b/server/sonar-webserver-core/src/main/java/org/sonar/server/telemetry/TelemetryDaemon.java
index 2c423c1d14a..974219952ba 100644
--- a/server/sonar-webserver-core/src/main/java/org/sonar/server/telemetry/TelemetryDaemon.java
+++ b/server/sonar-webserver-core/src/main/java/org/sonar/server/telemetry/TelemetryDaemon.java
@@ -27,7 +27,6 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
-import org.sonar.api.Startable;
 import org.sonar.api.config.Configuration;
 import org.sonar.api.server.ServerSide;
 import org.sonar.api.utils.System2;
@@ -35,6 +34,7 @@ import org.sonar.api.utils.log.Logger;
 import org.sonar.api.utils.log.Loggers;
 import org.sonar.api.utils.text.JsonWriter;
 import org.sonar.server.property.InternalProperties;
+import org.sonar.server.util.AbstractStoppableScheduledExecutorServiceImpl;
 import org.sonar.server.util.GlobalLockManager;
 
 import static org.sonar.process.ProcessProperties.Property.SONAR_TELEMETRY_ENABLE;
@@ -42,7 +42,7 @@ import static org.sonar.process.ProcessProperties.Property.SONAR_TELEMETRY_FREQU
 import static org.sonar.process.ProcessProperties.Property.SONAR_TELEMETRY_URL;
 
 @ServerSide
-public class TelemetryDaemon implements Startable {
+public class TelemetryDaemon extends AbstractStoppableScheduledExecutorServiceImpl<ScheduledExecutorService> {
   private static final String THREAD_NAME_PREFIX = "sq-telemetry-service-";
   private static final int ONE_DAY = 24 * 60 * 60 * 1_000;
   private static final String I_PROP_LAST_PING = "telemetry.lastPing";
@@ -60,10 +60,9 @@ public class TelemetryDaemon implements Startable {
   private final InternalProperties internalProperties;
   private final System2 system2;
 
-  private ScheduledExecutorService executorService;
-
   public TelemetryDaemon(TelemetryDataLoader dataLoader, TelemetryDataJsonWriter dataJsonWriter, TelemetryClient telemetryClient, Configuration config,
     InternalProperties internalProperties, GlobalLockManager lockManager, System2 system2) {
+    super(Executors.newSingleThreadScheduledExecutor(newThreadFactory()));
     this.dataLoader = dataLoader;
     this.dataJsonWriter = dataJsonWriter;
     this.telemetryClient = telemetryClient;
@@ -90,22 +89,8 @@ public class TelemetryDaemon implements Startable {
       return;
     }
     LOG.info("Sharing of SonarQube statistics is enabled.");
-    executorService = Executors.newSingleThreadScheduledExecutor(newThreadFactory());
     int frequencyInSeconds = frequency();
-    executorService.scheduleWithFixedDelay(telemetryCommand(), frequencyInSeconds, frequencyInSeconds, TimeUnit.SECONDS);
-  }
-
-  @Override
-  public void stop() {
-    try {
-      if (executorService == null) {
-        return;
-      }
-      executorService.shutdown();
-      executorService.awaitTermination(5, TimeUnit.SECONDS);
-    } catch (InterruptedException e) {
-      Thread.currentThread().interrupt();
-    }
+    scheduleWithFixedDelay(telemetryCommand(), frequencyInSeconds, frequencyInSeconds, TimeUnit.SECONDS);
   }
 
   private static ThreadFactory newThreadFactory() {