From: Sébastien Lesaint Date: Fri, 7 Jul 2017 14:59:16 +0000 (+0200) Subject: SONAR-9525 add IT on number of tasks processed in // X-Git-Tag: 6.6-RC1~881 X-Git-Url: https://source.dussan.org/?a=commitdiff_plain;h=e6ee887b01b83bb12aec85ab1dfdaee639b1d241;p=sonarqube.git SONAR-9525 add IT on number of tasks processed in // --- diff --git a/tests/plugins/fake-governance-plugin/src/main/java/FakeGovernancePlugin.java b/tests/plugins/fake-governance-plugin/src/main/java/FakeGovernancePlugin.java index 2d9c47cc854..2e236fded30 100644 --- a/tests/plugins/fake-governance-plugin/src/main/java/FakeGovernancePlugin.java +++ b/tests/plugins/fake-governance-plugin/src/main/java/FakeGovernancePlugin.java @@ -20,6 +20,8 @@ */ import org.sonar.api.Plugin; +import workerlatch.LatchControllerWorkerMeasureComputer; +import workerlatch.WorkerLatchMetrics; public class FakeGovernancePlugin implements Plugin { @@ -28,6 +30,8 @@ public class FakeGovernancePlugin implements Plugin { // Nothing should be loaded when the plugin is running within by the scanner if (isRunningInSQ()) { context.addExtension(FakeWorkerCountProviderImpl.class); + context.addExtension(WorkerLatchMetrics.class); + context.addExtension(LatchControllerWorkerMeasureComputer.class); } } diff --git a/tests/plugins/fake-governance-plugin/src/main/java/workerlatch/LatchControllerWorkerMeasureComputer.java b/tests/plugins/fake-governance-plugin/src/main/java/workerlatch/LatchControllerWorkerMeasureComputer.java new file mode 100644 index 00000000000..b7b04e57eed --- /dev/null +++ b/tests/plugins/fake-governance-plugin/src/main/java/workerlatch/LatchControllerWorkerMeasureComputer.java @@ -0,0 +1,120 @@ +/* + * SonarQube + * Copyright (C) 2009-2017 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 workerlatch; + +import java.io.File; +import java.io.IOException; +import java.io.RandomAccessFile; +import java.nio.MappedByteBuffer; +import java.nio.channels.FileChannel; +import org.sonar.api.ce.measure.Component; +import org.sonar.api.ce.measure.MeasureComputer; +import org.sonar.api.config.Configuration; +import org.sonar.api.utils.log.Logger; +import org.sonar.api.utils.log.Loggers; + +public class LatchControllerWorkerMeasureComputer implements MeasureComputer { + private static final Logger LOG = Loggers.get(LatchControllerWorkerMeasureComputer.class); + private static final String WORKER_LATCH_SHARED_MEMORY_FILE_PATH = "fakeGoverance.workerLatch.sharedMemoryFile"; + + private static final int WAIT = 200; // ms + private static final int MAX_WAIT_ROUND = 5 * 60 * 2; // 2 minutes + + private static final int MAX_SHARED_MEMORY = 1; + private static final byte UNLATCHED = (byte) 0x01; + + private final File sharedMemory; + + public LatchControllerWorkerMeasureComputer(Configuration configuration) { + this.sharedMemory = configuration.get(WORKER_LATCH_SHARED_MEMORY_FILE_PATH) + .map(path -> { + File file = new File(path); + if (file.exists() && file.isFile()) { + return file; + } + LOG.info("Latch sharedMemory file {} is not a file or does not exist", path); + return null; + }) + .orElse(null); + } + + private boolean isLatchEnabled() { + return sharedMemory != null; + } + + @Override + public MeasureComputerDefinition define(MeasureComputerDefinitionContext defContext) { + return defContext.newDefinitionBuilder() + .setOutputMetrics(WorkerLatchMetrics.METRIC_KEY) + .build(); + } + + @Override + public void compute(MeasureComputerContext context) { + Component component = context.getComponent(); + if (isLatchEnabled() && component.getType() == Component.Type.PROJECT) { + context.addMeasure(WorkerLatchMetrics.METRIC_KEY, waitForUnlatched(component.getKey())); + } else { + context.addMeasure(WorkerLatchMetrics.METRIC_KEY, false); + } + } + + private boolean waitForUnlatched(String key) { + RandomAccessFile randomAccessFile = null; + try { + randomAccessFile = new RandomAccessFile(sharedMemory, "rw"); + MappedByteBuffer mappedByteBuffer = randomAccessFile.getChannel().map(FileChannel.MapMode.READ_WRITE, 0, MAX_SHARED_MEMORY); + int i = 0; + boolean unlatched = isUnlatched(mappedByteBuffer); + while (!unlatched && i < MAX_WAIT_ROUND) { + waitInterruptedly(); + i++; + unlatched = isUnlatched(mappedByteBuffer); + } + LOG.info("Project {} unlatched={} i={}", key, unlatched, i); + + return true; + } catch (IOException e) { + LOG.error("Failed to read or write to shared memory", e); + return false; + } finally { + if (randomAccessFile != null) { + try { + randomAccessFile.close(); + } catch (IOException e) { + LOG.error("Failed to close randomAccessFile", e); + } + } + } + } + + private static void waitInterruptedly() { + try { + Thread.sleep(WAIT); + } catch (InterruptedException e) { + LOG.error("Wait was interrupted"); + } + } + + private boolean isUnlatched(MappedByteBuffer mappedByteBuffer) { + return mappedByteBuffer.get(0) == UNLATCHED; + } + +} diff --git a/tests/plugins/fake-governance-plugin/src/main/java/workerlatch/WorkerLatchMetrics.java b/tests/plugins/fake-governance-plugin/src/main/java/workerlatch/WorkerLatchMetrics.java new file mode 100644 index 00000000000..8689c9b1bba --- /dev/null +++ b/tests/plugins/fake-governance-plugin/src/main/java/workerlatch/WorkerLatchMetrics.java @@ -0,0 +1,37 @@ +/* + * SonarQube + * Copyright (C) 2009-2017 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 workerlatch; + +import java.util.Collections; +import java.util.List; +import org.sonar.api.measures.Metric; +import org.sonar.api.measures.Metrics; + +public class WorkerLatchMetrics implements Metrics { + static final String METRIC_KEY = "WORKER_LATCH"; + + @Override + public List getMetrics() { + return Collections.singletonList( + new Metric.Builder(METRIC_KEY, "Worker latch", Metric.ValueType.BOOL) + .setHidden(true) + .create()); + } +} diff --git a/tests/src/test/java/org/sonarqube/tests/Category5Suite.java b/tests/src/test/java/org/sonarqube/tests/Category5Suite.java index 905897e6c07..98587c47027 100644 --- a/tests/src/test/java/org/sonarqube/tests/Category5Suite.java +++ b/tests/src/test/java/org/sonarqube/tests/Category5Suite.java @@ -21,7 +21,7 @@ package org.sonarqube.tests; import org.junit.runner.RunWith; import org.junit.runners.Suite; -import org.sonarqube.tests.ce.WorkerCountTest; +import org.sonarqube.tests.ce.CeWorkersTest; import org.sonarqube.tests.qualityProfile.ActiveRuleEsResilienceTest; import org.sonarqube.tests.qualityProfile.BuiltInQualityProfilesNotificationTest; import org.sonarqube.tests.rule.RuleEsResilienceTest; @@ -58,7 +58,7 @@ import org.sonarqube.tests.user.UserEsResilienceTest; RuleEsResilienceTest.class, UserEsResilienceTest.class, // ce - WorkerCountTest.class + CeWorkersTest.class }) public class Category5Suite { diff --git a/tests/src/test/java/org/sonarqube/tests/ce/CeWorkersTest.java b/tests/src/test/java/org/sonarqube/tests/ce/CeWorkersTest.java new file mode 100644 index 00000000000..92a9482d852 --- /dev/null +++ b/tests/src/test/java/org/sonarqube/tests/ce/CeWorkersTest.java @@ -0,0 +1,196 @@ +/* + * SonarQube + * Copyright (C) 2009-2017 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.sonarqube.tests.ce; + +import com.google.common.collect.ImmutableList; +import com.sonar.orchestrator.Orchestrator; +import com.sonar.orchestrator.OrchestratorBuilder; +import com.sonar.orchestrator.build.SonarScanner; +import java.io.File; +import java.io.IOException; +import java.io.RandomAccessFile; +import java.nio.MappedByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.util.List; +import java.util.Random; +import java.util.Set; +import java.util.function.Function; +import java.util.stream.Collectors; +import java.util.stream.IntStream; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.sonarqube.ws.WsCe; +import org.sonarqube.ws.client.WsClient; +import org.sonarqube.ws.client.ce.ActivityWsRequest; +import util.ItUtils; + +import static com.google.common.collect.ImmutableSet.copyOf; +import static java.lang.String.valueOf; +import static java.util.stream.Collectors.toList; +import static java.util.stream.Collectors.toSet; +import static org.assertj.core.api.Assertions.assertThat; +import static util.ItUtils.newAdminWsClient; +import static util.ItUtils.pluginArtifact; +import static util.ItUtils.xooPlugin; + +public class CeWorkersTest { + /** 2 <= workerCount <= 5 */ + private static final int WORKER_COUNT = 2 + new Random().nextInt(4); + + private static final int WAIT = 200; // ms + private static final int MAX_WAIT_LOOP = 5 * 10; // 10s + + private static final byte BLOCKING = (byte) 0x00; + private static final byte UNLATCHED = (byte) 0x01; + + private static final String STATUS_PENDING = "PENDING"; + private static final String STATUS_IN_PROGRESS = "IN_PROGRESS"; + private static final String STATUS_SUCCESS = "SUCCESS"; + private static final String STATUS_FAILED = "FAILED"; + + private static final String STATUS_CANCELED = "CANCELED"; + + @ClassRule + public static TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private static File sharedMemory; + private static Orchestrator orchestrator; + + @BeforeClass + public static void setUp() throws Exception { + sharedMemory = temporaryFolder.newFile(); + + OrchestratorBuilder builder = Orchestrator.builderEnv() + .addPlugin(pluginArtifact("fake-governance-plugin")) + .setServerProperty("fakeGovernance.workerCount", valueOf(WORKER_COUNT)) + .setServerProperty("fakeGoverance.workerLatch.sharedMemoryFile", sharedMemory.getAbsolutePath()) + .addPlugin(xooPlugin()); + orchestrator = builder.build(); + orchestrator.start(); + } + + @AfterClass + public static void stop() { + if (orchestrator != null) { + orchestrator.stop(); + orchestrator = null; + } + } + + @Test + public void workerCount_is_modified_by_plugin() throws IOException { + Set line = Files.lines(orchestrator.getServer().getCeLogs().toPath()) + .filter(s -> s.contains("Compute Engine will use ")) + .collect(Collectors.toSet()); + assertThat(line) + .hasSize(1); + assertThat(line.iterator().next()).contains(valueOf(WORKER_COUNT)); + + assertThat(newAdminWsClient(orchestrator).ce().workerCount()) + .extracting(WsCe.WorkerCountResponse::getValue, WsCe.WorkerCountResponse::getCanSetWorkerCount) + .containsOnly(WORKER_COUNT, true); + } + + @Test + public void number_of_analysis_processed_in_parallel_is_equal_to_number_of_workers() throws IOException { + RandomAccessFile randomAccessFile = null; + try { + randomAccessFile = new RandomAccessFile(sharedMemory, "rw"); + MappedByteBuffer mappedByteBuffer = randomAccessFile.getChannel().map(FileChannel.MapMode.READ_WRITE, 0, 1); + // block any analysis which will run with the fake-governance-plugin + mappedByteBuffer.put(0, BLOCKING); + + // start analysis of WORKER_COUNT + 2 projects + List projectKeys = IntStream.range(0, WORKER_COUNT + 2).mapToObj(i -> "prj" + i).collect(toList()); + for (String projectKey : projectKeys) { + SonarScanner sonarRunner = SonarScanner.create(ItUtils.projectDir("shared/xoo-sample")) + .setProperties("sonar.projectKey", projectKey); + orchestrator.executeBuild(sonarRunner, false); + } + + List tasksList = waitForWsCallStatus(this::getTasksAllTasks, CeWorkersTest::threeTasksNotPending); + boolean expectedState = threeTasksNotPending(tasksList); + // let the blocked analyses finish running + mappedByteBuffer.put(0, UNLATCHED); + + assertThat(expectedState).as("Couldn't get to the expected CE queue state in time").isTrue(); + assertThat(tasksList.stream() + .filter(CeWorkersTest::pending) + .map(WsCe.Task::getComponentKey) + .collect(toSet())) + .isEqualTo(copyOf(projectKeys.subList(WORKER_COUNT, projectKeys.size()))); + assertThat(tasksList.stream() + .filter(CeWorkersTest::inProgress) + .map(WsCe.Task::getComponentKey) + .collect(toSet())) + .isEqualTo(copyOf(projectKeys.subList(0, WORKER_COUNT))); + + waitForWsCallStatus(this::getTasksAllTasks, tasks -> tasks.stream().noneMatch(CeWorkersTest::pending)); + } finally { + if (randomAccessFile != null) { + randomAccessFile.close(); + } + } + } + + private static boolean threeTasksNotPending(List tasksList) { + return tasksList.stream().filter(task -> !pending(task)).count() >= WORKER_COUNT; + } + + private static boolean pending(WsCe.Task task) { + return WsCe.TaskStatus.PENDING == task.getStatus(); + } + + private static boolean inProgress(WsCe.Task task) { + return WsCe.TaskStatus.IN_PROGRESS == task.getStatus(); + } + + private List getTasksAllTasks(WsClient wsClient) { + return wsClient.ce().activity(new ActivityWsRequest() + .setStatus(ImmutableList.of(STATUS_PENDING, STATUS_IN_PROGRESS, STATUS_SUCCESS, STATUS_FAILED, STATUS_CANCELED))) + .getTasksList(); + } + + private T waitForWsCallStatus(Function call, Function test) { + WsClient wsClient = ItUtils.newAdminWsClient(orchestrator); + int i = 0; + T returnValue = call.apply(wsClient); + boolean expectedState = test.apply(returnValue); + while (i < MAX_WAIT_LOOP && !expectedState) { + waitInterruptedly(); + i++; + returnValue = call.apply(wsClient); + expectedState = test.apply(returnValue); + } + return returnValue; + } + + private static void waitInterruptedly() { + try { + Thread.sleep(WAIT); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } +} diff --git a/tests/src/test/java/org/sonarqube/tests/ce/WorkerCountTest.java b/tests/src/test/java/org/sonarqube/tests/ce/WorkerCountTest.java deleted file mode 100644 index 3a8e766953d..00000000000 --- a/tests/src/test/java/org/sonarqube/tests/ce/WorkerCountTest.java +++ /dev/null @@ -1,69 +0,0 @@ -/* - * SonarQube - * Copyright (C) 2009-2017 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.sonarqube.tests.ce; - -import com.sonar.orchestrator.Orchestrator; -import com.sonar.orchestrator.OrchestratorBuilder; -import java.io.IOException; -import java.nio.file.Files; -import java.util.Set; -import java.util.stream.Collectors; -import org.junit.After; -import org.junit.Test; -import org.sonarqube.ws.WsCe.WorkerCountResponse; - -import static java.lang.Integer.parseInt; -import static org.assertj.core.api.Assertions.assertThat; -import static util.ItUtils.newAdminWsClient; -import static util.ItUtils.pluginArtifact; - -public class WorkerCountTest { - - private Orchestrator orchestrator; - - @After - public void stop() { - if (orchestrator != null) { - orchestrator.stop(); - orchestrator = null; - } - } - - @Test - public void workerCount_can_be_controlled_via_plugin() throws IOException { - String workerCount = "5"; - OrchestratorBuilder builder = Orchestrator.builderEnv() - .addPlugin(pluginArtifact("fake-governance-plugin")) - .setServerProperty("fakeGovernance.workerCount", workerCount); - orchestrator = builder.build(); - orchestrator.start(); - - Set line = Files.lines(orchestrator.getServer().getCeLogs().toPath()) - .filter(s -> s.contains("Compute Engine will use ")) - .collect(Collectors.toSet()); - assertThat(line) - .hasSize(1); - assertThat(line.iterator().next()).contains(workerCount); - - assertThat(newAdminWsClient(orchestrator).ce().workerCount()) - .extracting(WorkerCountResponse::getValue, WorkerCountResponse::getCanSetWorkerCount) - .containsOnly(parseInt(workerCount), true); - } -}