SONAR-12007 & SONAR-12008 tasks can be executed by any node of the cluster

SONAR-12007 & SONAR-12008 tasks can be executed by any node of the cluster

server/sonar-server/src/main/java/org/sonar/server/platform/platformlevel/PlatformLevel1.java

@@ -61,6 +61,7 @@ import org.sonar.server.rule.index.RuleIndex;
 import org.sonar.server.setting.ThreadLocalSettings;
 import org.sonar.server.user.SystemPasscodeImpl;
 import org.sonar.server.user.ThreadLocalUserSession;
+import org.sonar.server.util.GlobalLockManager;
 import org.sonar.server.util.OkHttpClientProvider;
 
 import static org.sonar.core.extension.CoreExtensionsInstaller.noAdditionalSideFilter;
@@ -128,6 +129,8 @@ public class PlatformLevel1 extends PlatformLevel {
       // issues
       IssueIndex.class,
 
+      GlobalLockManager.class,
+
       new OkHttpClientProvider(),
 
       CoreExtensionRepositoryImpl.class,

server/sonar-server/src/main/java/org/sonar/server/platform/platformlevel/PlatformLevel4.java

@@ -519,11 +519,14 @@ public class PlatformLevel4 extends PlatformLevel {
       HttpRequestIdModule.class,
 
       RecoveryIndexer.class,
-      ProjectIndexersImpl.class);
+      ProjectIndexersImpl.class,
 
-    // telemetry
-    add(TelemetryDataLoader.class);
-    addIfStartupLeader(TelemetryDaemon.class, TelemetryClient.class);
+      // telemetry
+      TelemetryDataLoader.class,
+      TelemetryDaemon.class,
+      TelemetryClient.class
+
+    );
 
     // system info
     add(WebSystemInfoModule.class);

server/sonar-server/src/main/java/org/sonar/server/telemetry/TelemetryDaemon.java

@@ -36,6 +36,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.GlobalLockManager;
 
 import static org.sonar.api.utils.DateUtils.formatDate;
 import static org.sonar.api.utils.DateUtils.parseDate;
@@ -50,21 +51,26 @@ public class TelemetryDaemon implements Startable {
   private static final int SEVEN_DAYS = 7 * 24 * 60 * 60 * 1_000;
   private static final String I_PROP_LAST_PING = "telemetry.lastPing";
   private static final String I_PROP_OPT_OUT = "telemetry.optOut";
+  private static final String LOCK_NAME = "TelemetryStat";
   private static final Logger LOG = Loggers.get(TelemetryDaemon.class);
+  private static final String LOCK_DELAY_SEC = "sonar.telemetry.lock.delay";
 
   private final TelemetryDataLoader dataLoader;
   private final TelemetryClient telemetryClient;
+  private final GlobalLockManager lockManager;
   private final Configuration config;
   private final InternalProperties internalProperties;
   private final System2 system2;
 
   private ScheduledExecutorService executorService;
 
-  public TelemetryDaemon(TelemetryDataLoader dataLoader, TelemetryClient telemetryClient, Configuration config, InternalProperties internalProperties, System2 system2) {
+  public TelemetryDaemon(TelemetryDataLoader dataLoader, TelemetryClient telemetryClient, Configuration config,
+    InternalProperties internalProperties, GlobalLockManager lockManager, System2 system2) {
     this.dataLoader = dataLoader;
     this.telemetryClient = telemetryClient;
     this.config = config;
     this.internalProperties = internalProperties;
+    this.lockManager = lockManager;
     this.system2 = system2;
   }
 
@@ -113,6 +119,11 @@ public class TelemetryDaemon implements Startable {
   private Runnable telemetryCommand() {
     return () -> {
       try {
+
+        if (!lockManager.tryLock(LOCK_NAME, lockDuration())) {
+          return;
+        }
+
         long now = system2.now();
         if (shouldUploadStatistics(now)) {
           uploadStatistics();
@@ -157,4 +168,8 @@ public class TelemetryDaemon implements Startable {
     return config.getInt(SONAR_TELEMETRY_FREQUENCY_IN_SECONDS.getKey())
       .orElseThrow(() -> new IllegalStateException(String.format("Setting '%s' must be provided.", SONAR_TELEMETRY_FREQUENCY_IN_SECONDS)));
   }
+
+  private int lockDuration(){
+    return config.getInt(LOCK_DELAY_SEC).orElse(60);
+  }
 }

server/sonar-server/src/main/java/org/sonar/server/util/GlobalLockManager.java

@@ -0,0 +1,59 @@
+/*
+ * SonarQube
+ * Copyright (C) 2009-2019 SonarSource SA
+ * mailto:info AT sonarsource DOT com
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ */
+package org.sonar.server.util;
+
+import org.sonar.api.ce.ComputeEngineSide;
+import org.sonar.api.server.ServerSide;
+import org.sonar.db.DbClient;
+import org.sonar.db.DbSession;
+
+/**
+ * Provide a simple mechanism to manage global locks across multiple nodes running in a cluster.
+ * In the target use case multiple nodes try to execute something at around the same time,
+ * and only the first should succeed, and the rest do nothing.
+ */
+@ComputeEngineSide
+@ServerSide
+public class GlobalLockManager {
+
+  static final int DEFAULT_LOCK_DURATION_SECONDS = 180;
+
+  private final DbClient dbClient;
+
+  public GlobalLockManager(DbClient dbClient) {
+    this.dbClient = dbClient;
+  }
+
+  /**
+   * Try to acquire a lock on the given name in the default namespace,
+   * using the generic locking mechanism of {@see org.sonar.db.property.InternalPropertiesDao}.
+   */
+  public boolean tryLock(String name) {
+    return tryLock(name, DEFAULT_LOCK_DURATION_SECONDS);
+  }
+
+  public boolean tryLock(String name, int durationSecond) {
+    try (DbSession dbSession = dbClient.openSession(false)) {
+      boolean success = dbClient.internalPropertiesDao().tryLock(dbSession, name, durationSecond);
+      dbSession.commit();
+      return success;
+    }
+  }
+}

server/sonar-server/src/test/java/org/sonar/server/telemetry/TelemetryDaemonTest.java

@@ -29,6 +29,7 @@ import org.junit.After;
 import org.junit.Rule;
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
+import org.mockito.internal.matchers.Any;
 import org.sonar.api.config.internal.MapSettings;
 import org.sonar.api.utils.internal.TestSystem2;
 import org.sonar.api.utils.log.LogTester;
@@ -50,6 +51,7 @@ import org.sonar.server.property.MapInternalProperties;
 import org.sonar.server.tester.UserSessionRule;
 import org.sonar.server.user.index.UserIndex;
 import org.sonar.server.user.index.UserIndexer;
+import org.sonar.server.util.GlobalLockManager;
 import org.sonar.updatecenter.common.Version;
 
 import static java.util.Arrays.asList;
@@ -57,6 +59,8 @@ import static java.util.Collections.emptySet;
 import static org.apache.commons.lang.RandomStringUtils.randomAlphabetic;
 import static org.apache.commons.lang.RandomStringUtils.randomAlphanumeric;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.after;
 import static org.mockito.Mockito.mock;
@@ -93,6 +97,7 @@ public class TelemetryDaemonTest {
 
   private TelemetryClient client = mock(TelemetryClient.class);
   private InternalProperties internalProperties = spy(new MapInternalProperties());
+  private final GlobalLockManager lockManager = mock(GlobalLockManager.class);
   private FakeServer server = new FakeServer();
   private PluginRepository pluginRepository = mock(PluginRepository.class);
   private TestSystem2 system2 = new TestSystem2().setNow(System.currentTimeMillis());
@@ -103,12 +108,12 @@ public class TelemetryDaemonTest {
 
   private final TelemetryDataLoader communityDataLoader = new TelemetryDataLoader(server, db.getDbClient(), pluginRepository, new UserIndex(es.client(), system2),
     new ProjectMeasuresIndex(es.client(), null, system2), editionProvider, new DefaultOrganizationProviderImpl(db.getDbClient()), internalProperties, null);
-  private TelemetryDaemon communityUnderTest = new TelemetryDaemon(communityDataLoader, client, settings.asConfig(), internalProperties, system2);
+  private TelemetryDaemon communityUnderTest = new TelemetryDaemon(communityDataLoader, client, settings.asConfig(), internalProperties, lockManager, system2);
 
   private final LicenseReader licenseReader = mock(LicenseReader.class);
   private final TelemetryDataLoader commercialDataLoader = new TelemetryDataLoader(server, db.getDbClient(), pluginRepository, new UserIndex(es.client(), system2),
     new ProjectMeasuresIndex(es.client(), null, system2), editionProvider, new DefaultOrganizationProviderImpl(db.getDbClient()), internalProperties, licenseReader);
-  private TelemetryDaemon commercialUnderTest = new TelemetryDaemon(commercialDataLoader, client, settings.asConfig(), internalProperties, system2);
+  private TelemetryDaemon commercialUnderTest = new TelemetryDaemon(commercialDataLoader, client, settings.asConfig(), internalProperties, lockManager, system2);
 
   @After
   public void tearDown() {
@@ -124,6 +129,7 @@ public class TelemetryDaemonTest {
     List<PluginInfo> plugins = asList(newPlugin("java", "4.12.0.11033"), newPlugin("scmgit", "1.2"), new PluginInfo("other"));
     when(pluginRepository.getPluginInfos()).thenReturn(plugins);
     when(editionProvider.get()).thenReturn(Optional.of(EditionProvider.Edition.DEVELOPER));
+    when(lockManager.tryLock(any(), anyInt())).thenReturn(true);
 
     IntStream.range(0, 3).forEach(i -> db.users().insertUser());
     db.users().insertUser(u -> u.setActive(false));
@@ -175,6 +181,7 @@ public class TelemetryDaemonTest {
   @Test
   public void take_biggest_long_living_branches() throws IOException {
     initTelemetrySettingsToDefaultValues();
+    when(lockManager.tryLock(any(), anyInt())).thenReturn(true);
     settings.setProperty("sonar.telemetry.frequencyInSeconds", "1");
     server.setId("AU-TpxcB-iU5OvuD2FL7").setVersion("7.5.4");
     MetricDto ncloc = db.measures().insertMetric(m -> m.setKey(NCLOC_KEY));
@@ -197,6 +204,7 @@ public class TelemetryDaemonTest {
   @Test
   public void send_data_via_client_at_startup_after_initial_delay() throws IOException {
     initTelemetrySettingsToDefaultValues();
+    when(lockManager.tryLock(any(), anyInt())).thenReturn(true);
     settings.setProperty("sonar.telemetry.frequencyInSeconds", "1");
     communityUnderTest.start();
 
@@ -206,6 +214,7 @@ public class TelemetryDaemonTest {
   @Test
   public void data_contains_no_license_type_on_community_edition() throws IOException {
     initTelemetrySettingsToDefaultValues();
+    when(lockManager.tryLock(any(), anyInt())).thenReturn(true);
     settings.setProperty("sonar.telemetry.frequencyInSeconds", "1");
 
     communityUnderTest.start();
@@ -219,6 +228,7 @@ public class TelemetryDaemonTest {
     initTelemetrySettingsToDefaultValues();
     settings.setProperty("sonar.telemetry.frequencyInSeconds", "1");
     when(licenseReader.read()).thenReturn(Optional.empty());
+    when(lockManager.tryLock(any(), anyInt())).thenReturn(true);
 
     commercialUnderTest.start();
 
@@ -234,6 +244,7 @@ public class TelemetryDaemonTest {
     LicenseReader.License license = mock(LicenseReader.License.class);
     when(license.getType()).thenReturn(licenseType);
     when(licenseReader.read()).thenReturn(Optional.of(license));
+    when(lockManager.tryLock(any(), anyInt())).thenReturn(true);
 
     commercialUnderTest.start();
 
@@ -246,6 +257,7 @@ public class TelemetryDaemonTest {
   @Test
   public void check_if_should_send_data_periodically() throws IOException {
     initTelemetrySettingsToDefaultValues();
+    when(lockManager.tryLock(any(), anyInt())).thenReturn(true);
     long now = system2.now();
     long sixDaysAgo = now - (ONE_DAY * 6L);
     long sevenDaysAgo = now - (ONE_DAY * 7L);
@@ -261,6 +273,7 @@ public class TelemetryDaemonTest {
   @Test
   public void send_server_id_and_version() throws IOException {
     initTelemetrySettingsToDefaultValues();
+    when(lockManager.tryLock(any(), anyInt())).thenReturn(true);
     settings.setProperty("sonar.telemetry.frequencyInSeconds", "1");
     String id = randomAlphanumeric(40);
     String version = randomAlphanumeric(10);
@@ -275,6 +288,7 @@ public class TelemetryDaemonTest {
   @Test
   public void send_server_installation_date_and_installation_version() throws IOException {
     initTelemetrySettingsToDefaultValues();
+    when(lockManager.tryLock(any(), anyInt())).thenReturn(true);
     settings.setProperty("sonar.telemetry.frequencyInSeconds", "1");
     String installationVersion = "7.9.BEST.LTS.EVER";
     Long installationDate = 1546300800000L; // 2019/01/01
@@ -290,6 +304,7 @@ public class TelemetryDaemonTest {
   @Test
   public void do_not_send_server_installation_details_if_missing_property() throws IOException {
     initTelemetrySettingsToDefaultValues();
+    when(lockManager.tryLock(any(), anyInt())).thenReturn(true);
     settings.setProperty("sonar.telemetry.frequencyInSeconds", "1");
 
     communityUnderTest.start();
@@ -301,6 +316,7 @@ public class TelemetryDaemonTest {
   @Test
   public void do_not_send_data_if_last_ping_earlier_than_one_week_ago() throws IOException {
     initTelemetrySettingsToDefaultValues();
+    when(lockManager.tryLock(any(), anyInt())).thenReturn(true);
     settings.setProperty("sonar.telemetry.frequencyInSeconds", "1");
     long now = system2.now();
     long sixDaysAgo = now - (ONE_DAY * 6L);
@@ -314,6 +330,7 @@ public class TelemetryDaemonTest {
   @Test
   public void send_data_if_last_ping_is_one_week_ago() throws IOException {
     initTelemetrySettingsToDefaultValues();
+    when(lockManager.tryLock(any(), anyInt())).thenReturn(true);
     settings.setProperty("sonar.telemetry.frequencyInSeconds", "1");
     long today = parseDate("2017-08-01").getTime();
     system2.setNow(today + 15 * ONE_HOUR);
@@ -330,6 +347,7 @@ public class TelemetryDaemonTest {
   @Test
   public void opt_out_sent_once() throws IOException {
     initTelemetrySettingsToDefaultValues();
+    when(lockManager.tryLock(any(), anyInt())).thenReturn(true);
     settings.setProperty("sonar.telemetry.frequencyInSeconds", "1");
     settings.setProperty("sonar.telemetry.enable", "false");
     communityUnderTest.start();

server/sonar-server/src/test/java/org/sonar/server/util/GlobalLockManagerTest.java

@@ -0,0 +1,76 @@
+/*
+ * SonarQube
+ * Copyright (C) 2009-2019 SonarSource SA
+ * mailto:info AT sonarsource DOT com
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ */
+package org.sonar.server.util;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.sonar.api.utils.System2;
+import org.sonar.db.DbTester;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.sonar.server.util.GlobalLockManager.DEFAULT_LOCK_DURATION_SECONDS;
+
+public class GlobalLockManagerTest {
+
+  private final System2 system2 = mock(System2.class);
+
+  @Rule
+  public final DbTester dbTester = DbTester.create(system2);
+
+  private final GlobalLockManager underTest = new GlobalLockManager(dbTester.getDbClient());
+
+  @Test
+  public void tryLock_succeeds_when_created_for_the_first_time() {
+    assertThat(underTest.tryLock("newName")).isTrue();
+  }
+
+  @Test
+  public void tryLock_fails_when_previous_lock_is_too_recent() {
+    String name = "newName";
+    assertThat(underTest.tryLock(name)).isTrue();
+    assertThat(underTest.tryLock(name)).isFalse();
+  }
+
+  @Test
+  public void tryLock_succeeds_when_previous_lock_is_old_enough() {
+    String name = "newName";
+    long firstLock = 0;
+    long longEnoughAfterFirstLock = firstLock + DEFAULT_LOCK_DURATION_SECONDS * 1000;
+    long notLongEnoughAfterFirstLock = longEnoughAfterFirstLock - 1;
+
+    when(system2.now()).thenReturn(firstLock);
+    assertThat(underTest.tryLock(name)).isTrue();
+
+    when(system2.now()).thenReturn(notLongEnoughAfterFirstLock);
+    assertThat(underTest.tryLock(name)).isFalse();
+
+    when(system2.now()).thenReturn(longEnoughAfterFirstLock);
+    assertThat(underTest.tryLock(name)).isTrue();
+  }
+
+  @Test
+  public void locks_with_different_name_are_independent() {
+    assertThat(underTest.tryLock("newName1")).isTrue();
+    assertThat(underTest.tryLock("newName2")).isTrue();
+  }
+
+}