geode-assembly/src/acceptanceTest/java/org/apache/geode/metrics/function/executions/FunctionExecutionsTimerClusterTest.java - geode - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
  * agreements. See the NOTICE file distributed with this work for additional information regarding
  * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance with the License. You may obtain a
  * copy of the License at
  *
  * http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software distributed under the License
  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  * or implied. See the License for the specific language governing permissions and limitations under
  * the License.
  */
 package org.apache.geode.metrics.function.executions;


 import static java.io.File.pathSeparatorChar;
 import static java.lang.Boolean.TRUE;
 import static java.util.stream.Collectors.toList;
 import static org.apache.geode.test.compiler.ClassBuilder.writeJarFromClasses;
 import static org.assertj.core.api.Assertions.assertThat;

 import java.io.IOException;
 import java.nio.file.Path;
 import java.time.Duration;
 import java.util.List;

 import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;

 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.client.ClientCache;
 import org.apache.geode.cache.client.ClientCacheFactory;
 import org.apache.geode.cache.client.ClientRegionShortcut;
 import org.apache.geode.cache.client.Pool;
 import org.apache.geode.cache.client.PoolManager;
 import org.apache.geode.cache.execute.Execution;
 import org.apache.geode.cache.execute.Function;
 import org.apache.geode.cache.execute.FunctionService;
 import org.apache.geode.internal.AvailablePortHelper;
 import org.apache.geode.metrics.MetricsPublishingService;
 import org.apache.geode.metrics.SimpleMetricsPublishingService;
 import org.apache.geode.rules.ServiceJarRule;
 import org.apache.geode.test.junit.rules.gfsh.GfshRule;

 /**
  * Acceptance tests for function executions timer on cluster with two servers and one locator
  */
 public class FunctionExecutionsTimerClusterTest {

   private int locatorPort;
   private ClientCache clientCache;
   private Pool server1Pool;
   private Pool server2Pool;
   private Pool multiServerPool;
   private Region<Object, Object> replicateRegion;
   private Region<Object, Object> partitionRegion;

   @Rule
   public GfshRule gfshRule = new GfshRule();

   @Rule
   public TemporaryFolder temporaryFolder = new TemporaryFolder();

   @Rule
   public ServiceJarRule serviceJarRule = new ServiceJarRule();

   @Before
   public void setUp() throws IOException {
     int[] availablePorts = AvailablePortHelper.getRandomAvailableTCPPorts(4);

     locatorPort = availablePorts[0];
     int locatorJmxPort = availablePorts[1];
     int server1Port = availablePorts[2];
     int server2Port = availablePorts[3];

     Path serviceJarPath = serviceJarRule.createJarFor("metrics-publishing-service.jar",
         MetricsPublishingService.class, SimpleMetricsPublishingService.class);

     Path functionsJarPath = temporaryFolder.getRoot().toPath()
         .resolve("functions.jar").toAbsolutePath();
     writeJarFromClasses(functionsJarPath.toFile(), GetFunctionExecutionTimerValues.class,
         FunctionToTimeWithResult.class, ExecutionsTimerValues.class, ThreadSleep.class);

     String startLocatorCommand = String.join(" ",
         "start locator",
         "--name=" + "locator",
         "--dir=" + temporaryFolder.newFolder("locator").getAbsolutePath(),
         "--port=" + locatorPort,
         "--http-service-port=0",
         "--J=-Dgemfire.jmx-manager-port=" + locatorJmxPort);

     String server1Name = "server1";
     String server2Name = "server2";
     String startServer1Command =
         startServerCommand(server1Name, server1Port, serviceJarPath, functionsJarPath);
     String startServer2Command =
         startServerCommand(server2Name, server2Port, serviceJarPath, functionsJarPath);

     String replicateRegionName = "ReplicateRegion";
     String createReplicateRegionCommand = String.join(" ",
         "create region",
         "--type=REPLICATE",
         "--name=" + replicateRegionName);

     String partitionRegionName = "PartitionRegion";
     String createPartitionRegionCommand = String.join(" ",
         "create region",
         "--type=PARTITION",
         "--groups=" + server1Name,
         "--name=" + partitionRegionName);

     gfshRule.execute(startLocatorCommand, startServer1Command, startServer2Command,
         createReplicateRegionCommand, createPartitionRegionCommand);

     clientCache = new ClientCacheFactory().create();

     server1Pool = PoolManager.createFactory()
         .addServer("localhost", server1Port)
         .create("server1Pool");

     server2Pool = PoolManager.createFactory()
         .addServer("localhost", server2Port)
         .create("server2Pool");

     multiServerPool = PoolManager.createFactory()
         .addServer("localhost", server1Port)
         .addServer("localhost", server2Port)
         .create("multiServerPool");

     replicateRegion = clientCache
         .createClientRegionFactory(ClientRegionShortcut.PROXY)
         .setPoolName(multiServerPool.getName())
         .create(replicateRegionName);

     partitionRegion = clientCache
         .createClientRegionFactory(ClientRegionShortcut.PROXY)
         .setPoolName(server1Pool.getName())
         .create(partitionRegionName);
   }

   @After
   public void tearDown() {
     if (clientCache != null) {
       clientCache.close();
     }
     if (multiServerPool != null) {
       multiServerPool.destroy();
     }
     if (server1Pool != null) {
       server1Pool.destroy();
     }
     if (server2Pool != null) {
       server2Pool.destroy();
     }

     String connectToLocatorCommand = "connect --locator=localhost[" + locatorPort + "]";
     String shutdownCommand = "shutdown --include-locators=true";
     gfshRule.execute(connectToLocatorCommand, shutdownCommand);
   }

   @Test
   public void timersRecordCountAndTotalTime_ifFunctionExecutedOnReplicateRegion() {
     FunctionToTimeWithResult function = new FunctionToTimeWithResult();
     Duration functionDuration = Duration.ofSeconds(1);
     executeFunctionOnReplicateRegion(function, functionDuration);

     List<ExecutionsTimerValues> values = getAllExecutionsTimerValues(function.getId());

     assertThat(getAggregateCount(values))
         .as("Number of function executions across all servers")
         .isEqualTo(1);

     assertThat(getAggregateTotalTime(values))
         .as("Total time of function executions across all servers")
         .isBetween((double) functionDuration.toNanos(), ((double) functionDuration.toNanos()) * 2);
   }

   @Test
   public void timersRecordCountAndTotalTime_ifFunctionExecutedOnReplicateRegionMultipleTimes() {
     FunctionToTimeWithResult function = new FunctionToTimeWithResult();
     Duration functionDuration = Duration.ofSeconds(1);
     int numberOfExecutions = 10;

     for (int i = 0; i < numberOfExecutions; i++) {
       executeFunctionOnReplicateRegion(function, functionDuration);
     }

     List<ExecutionsTimerValues> values = getAllExecutionsTimerValues(function.getId());

     double expectedMinimumTotalTime = ((double) functionDuration.toNanos()) * numberOfExecutions;
     double expectedMaximumTotalTime = expectedMinimumTotalTime * 2;

     assertThat(getAggregateCount(values))
         .as("Number of function executions across all servers")
         .isEqualTo(numberOfExecutions);

     assertThat(getAggregateTotalTime(values))
         .as("Total time of function executions across all servers")
         .isBetween(expectedMinimumTotalTime, expectedMaximumTotalTime);
   }

   @Test
   public void timersRecordCountAndTotalTime_ifFunctionExecutedOnPartitionRegionMultipleTimes() {
     FunctionToTimeWithResult function = new FunctionToTimeWithResult();
     Duration functionDuration = Duration.ofSeconds(1);
     int numberOfExecutions = 10;

     for (int i = 0; i < numberOfExecutions; i++) {
       executeFunctionOnPartitionRegion(function, functionDuration);
     }

     List<ExecutionsTimerValues> server1Values = getServer1ExecutionsTimerValues(function.getId());
     List<ExecutionsTimerValues> server2Values = getServer2ExecutionsTimerValues(function.getId());

     double expectedMinimumTotalTime = ((double) functionDuration.toNanos()) * numberOfExecutions;
     double expectedMaximumTotalTime = expectedMinimumTotalTime * 2;

     assertThat(getAggregateCount(server1Values))
         .as("Number of function executions on server 1")
         .isEqualTo(numberOfExecutions);

     assertThat(getAggregateTotalTime(server1Values))
         .as("Total time of function executions on server 1")
         .isBetween(expectedMinimumTotalTime, expectedMaximumTotalTime);

     assertThat(getAggregateCount(server2Values))
         .as("Number of function executions on server 2")
         .isEqualTo(0);

     assertThat(getAggregateTotalTime(server2Values))
         .as("Total time of function executions on server 2")
         .isEqualTo(0);
   }

   private String startServerCommand(String serverName, int serverPort, Path serviceJarPath,
       Path functionsJarPath)
       throws IOException {
     return String.join(" ",
         "start server",
         "--name=" + serverName,
         "--groups=" + serverName,
         "--dir=" + temporaryFolder.newFolder(serverName).getAbsolutePath(),
         "--server-port=" + serverPort,
         "--locators=localhost[" + locatorPort + "]",
         "--classpath=" + serviceJarPath + pathSeparatorChar + functionsJarPath);
   }

   private void executeFunctionOnReplicateRegion(Function<? super String[]> function,
       Duration duration) {
     executeFunctionOnRegion(function, duration, replicateRegion);
   }

   private void executeFunctionOnPartitionRegion(Function<? super String[]> function,
       Duration duration) {
     executeFunctionOnRegion(function, duration, partitionRegion);
   }

   private void executeFunctionOnRegion(Function<? super String[]> function, Duration duration,
       Region<?, ?> region) {
     @SuppressWarnings("unchecked")
     Execution<String[], Object, List<Object>> execution =
         (Execution<String[], Object, List<Object>>) FunctionService.onRegion(region);

     execution
         .setArguments(new String[] {String.valueOf(duration.toMillis()), TRUE.toString()})
         .execute(function)
         .getResult();
   }

   private List<ExecutionsTimerValues> getServer1ExecutionsTimerValues(String functionId) {
     return getExecutionsTimerValuesFromPool(functionId, server1Pool);
   }

   private List<ExecutionsTimerValues> getServer2ExecutionsTimerValues(String functionId) {
     return getExecutionsTimerValuesFromPool(functionId, server2Pool);
   }

   private List<ExecutionsTimerValues> getAllExecutionsTimerValues(String functionId) {
     return getExecutionsTimerValuesFromPool(functionId, multiServerPool);
   }

   private List<ExecutionsTimerValues> getExecutionsTimerValuesFromPool(String functionId,
       Pool pool) {
     @SuppressWarnings("unchecked")
     Execution<Void, List<ExecutionsTimerValues>, List<List<ExecutionsTimerValues>>> functionExecution =
         (Execution<Void, List<ExecutionsTimerValues>, List<List<ExecutionsTimerValues>>>) FunctionService
             .onServers(pool);

     List<List<ExecutionsTimerValues>> timerValuesForEachServer = functionExecution
         .execute(new GetFunctionExecutionTimerValues())
         .getResult();

     return timerValuesForEachServer.stream()
         .flatMap(List::stream)
         .filter(v -> v.functionId.equals(functionId))
         .collect(toList());
   }

   private static Long getAggregateCount(List<ExecutionsTimerValues> values) {
     return values.stream().map(x -> x.count).reduce(0L, Long::sum);
   }

   private static Double getAggregateTotalTime(List<ExecutionsTimerValues> values) {
     return values.stream().map(x -> x.totalTime).reduce(0.0, Double::sum);
   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more contributor license
	* agreements. See the NOTICE file distributed with this work for additional information regarding
	* copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance with the License. You may obtain a
	* copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software distributed under the License
	* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
	* or implied. See the License for the specific language governing permissions and limitations under
	* the License.
	*/
	package org.apache.geode.metrics.function.executions;


	import static java.io.File.pathSeparatorChar;
	import static java.lang.Boolean.TRUE;
	import static java.util.stream.Collectors.toList;
	import static org.apache.geode.test.compiler.ClassBuilder.writeJarFromClasses;
	import static org.assertj.core.api.Assertions.assertThat;

	import java.io.IOException;
	import java.nio.file.Path;
	import java.time.Duration;
	import java.util.List;

	import org.junit.After;
	import org.junit.Before;
	import org.junit.Rule;
	import org.junit.Test;
	import org.junit.rules.TemporaryFolder;

	import org.apache.geode.cache.Region;
	import org.apache.geode.cache.client.ClientCache;
	import org.apache.geode.cache.client.ClientCacheFactory;
	import org.apache.geode.cache.client.ClientRegionShortcut;
	import org.apache.geode.cache.client.Pool;
	import org.apache.geode.cache.client.PoolManager;
	import org.apache.geode.cache.execute.Execution;
	import org.apache.geode.cache.execute.Function;
	import org.apache.geode.cache.execute.FunctionService;
	import org.apache.geode.internal.AvailablePortHelper;
	import org.apache.geode.metrics.MetricsPublishingService;
	import org.apache.geode.metrics.SimpleMetricsPublishingService;
	import org.apache.geode.rules.ServiceJarRule;
	import org.apache.geode.test.junit.rules.gfsh.GfshRule;

	/**
	* Acceptance tests for function executions timer on cluster with two servers and one locator
	*/
	public class FunctionExecutionsTimerClusterTest {

	private int locatorPort;
	private ClientCache clientCache;
	private Pool server1Pool;
	private Pool server2Pool;
	private Pool multiServerPool;
	private Region<Object, Object> replicateRegion;
	private Region<Object, Object> partitionRegion;

	@Rule
	public GfshRule gfshRule = new GfshRule();

	@Rule
	public TemporaryFolder temporaryFolder = new TemporaryFolder();

	@Rule
	public ServiceJarRule serviceJarRule = new ServiceJarRule();

	@Before
	public void setUp() throws IOException {
	int[] availablePorts = AvailablePortHelper.getRandomAvailableTCPPorts(4);

	locatorPort = availablePorts[0];
	int locatorJmxPort = availablePorts[1];
	int server1Port = availablePorts[2];
	int server2Port = availablePorts[3];

	Path serviceJarPath = serviceJarRule.createJarFor("metrics-publishing-service.jar",
	MetricsPublishingService.class, SimpleMetricsPublishingService.class);

	Path functionsJarPath = temporaryFolder.getRoot().toPath()
	.resolve("functions.jar").toAbsolutePath();
	writeJarFromClasses(functionsJarPath.toFile(), GetFunctionExecutionTimerValues.class,
	FunctionToTimeWithResult.class, ExecutionsTimerValues.class, ThreadSleep.class);

	String startLocatorCommand = String.join(" ",
	"start locator",
	"--name=" + "locator",
	"--dir=" + temporaryFolder.newFolder("locator").getAbsolutePath(),
	"--port=" + locatorPort,
	"--http-service-port=0",
	"--J=-Dgemfire.jmx-manager-port=" + locatorJmxPort);

	String server1Name = "server1";
	String server2Name = "server2";
	String startServer1Command =
	startServerCommand(server1Name, server1Port, serviceJarPath, functionsJarPath);
	String startServer2Command =
	startServerCommand(server2Name, server2Port, serviceJarPath, functionsJarPath);

	String replicateRegionName = "ReplicateRegion";
	String createReplicateRegionCommand = String.join(" ",
	"create region",
	"--type=REPLICATE",
	"--name=" + replicateRegionName);

	String partitionRegionName = "PartitionRegion";
	String createPartitionRegionCommand = String.join(" ",
	"create region",
	"--type=PARTITION",
	"--groups=" + server1Name,
	"--name=" + partitionRegionName);

	gfshRule.execute(startLocatorCommand, startServer1Command, startServer2Command,
	createReplicateRegionCommand, createPartitionRegionCommand);

	clientCache = new ClientCacheFactory().create();

	server1Pool = PoolManager.createFactory()
	.addServer("localhost", server1Port)
	.create("server1Pool");

	server2Pool = PoolManager.createFactory()
	.addServer("localhost", server2Port)
	.create("server2Pool");

	multiServerPool = PoolManager.createFactory()
	.addServer("localhost", server1Port)
	.addServer("localhost", server2Port)
	.create("multiServerPool");

	replicateRegion = clientCache
	.createClientRegionFactory(ClientRegionShortcut.PROXY)
	.setPoolName(multiServerPool.getName())
	.create(replicateRegionName);

	partitionRegion = clientCache
	.createClientRegionFactory(ClientRegionShortcut.PROXY)
	.setPoolName(server1Pool.getName())
	.create(partitionRegionName);
	}

	@After
	public void tearDown() {
	if (clientCache != null) {
	clientCache.close();
	}
	if (multiServerPool != null) {
	multiServerPool.destroy();
	}
	if (server1Pool != null) {
	server1Pool.destroy();
	}
	if (server2Pool != null) {
	server2Pool.destroy();
	}

	String connectToLocatorCommand = "connect --locator=localhost[" + locatorPort + "]";
	String shutdownCommand = "shutdown --include-locators=true";
	gfshRule.execute(connectToLocatorCommand, shutdownCommand);
	}

	@Test
	public void timersRecordCountAndTotalTime_ifFunctionExecutedOnReplicateRegion() {
	FunctionToTimeWithResult function = new FunctionToTimeWithResult();
	Duration functionDuration = Duration.ofSeconds(1);
	executeFunctionOnReplicateRegion(function, functionDuration);

	List<ExecutionsTimerValues> values = getAllExecutionsTimerValues(function.getId());

	assertThat(getAggregateCount(values))
	.as("Number of function executions across all servers")
	.isEqualTo(1);

	assertThat(getAggregateTotalTime(values))
	.as("Total time of function executions across all servers")
	.isBetween((double) functionDuration.toNanos(), ((double) functionDuration.toNanos()) * 2);
	}

	@Test
	public void timersRecordCountAndTotalTime_ifFunctionExecutedOnReplicateRegionMultipleTimes() {
	FunctionToTimeWithResult function = new FunctionToTimeWithResult();
	Duration functionDuration = Duration.ofSeconds(1);
	int numberOfExecutions = 10;

	for (int i = 0; i < numberOfExecutions; i++) {
	executeFunctionOnReplicateRegion(function, functionDuration);
	}

	List<ExecutionsTimerValues> values = getAllExecutionsTimerValues(function.getId());

	double expectedMinimumTotalTime = ((double) functionDuration.toNanos()) * numberOfExecutions;
	double expectedMaximumTotalTime = expectedMinimumTotalTime * 2;

	assertThat(getAggregateCount(values))
	.as("Number of function executions across all servers")
	.isEqualTo(numberOfExecutions);

	assertThat(getAggregateTotalTime(values))
	.as("Total time of function executions across all servers")
	.isBetween(expectedMinimumTotalTime, expectedMaximumTotalTime);
	}

	@Test
	public void timersRecordCountAndTotalTime_ifFunctionExecutedOnPartitionRegionMultipleTimes() {
	FunctionToTimeWithResult function = new FunctionToTimeWithResult();
	Duration functionDuration = Duration.ofSeconds(1);
	int numberOfExecutions = 10;

	for (int i = 0; i < numberOfExecutions; i++) {
	executeFunctionOnPartitionRegion(function, functionDuration);
	}

	List<ExecutionsTimerValues> server1Values = getServer1ExecutionsTimerValues(function.getId());
	List<ExecutionsTimerValues> server2Values = getServer2ExecutionsTimerValues(function.getId());

	double expectedMinimumTotalTime = ((double) functionDuration.toNanos()) * numberOfExecutions;
	double expectedMaximumTotalTime = expectedMinimumTotalTime * 2;

	assertThat(getAggregateCount(server1Values))
	.as("Number of function executions on server 1")
	.isEqualTo(numberOfExecutions);

	assertThat(getAggregateTotalTime(server1Values))
	.as("Total time of function executions on server 1")
	.isBetween(expectedMinimumTotalTime, expectedMaximumTotalTime);

	assertThat(getAggregateCount(server2Values))
	.as("Number of function executions on server 2")
	.isEqualTo(0);

	assertThat(getAggregateTotalTime(server2Values))
	.as("Total time of function executions on server 2")
	.isEqualTo(0);
	}

	private String startServerCommand(String serverName, int serverPort, Path serviceJarPath,
	Path functionsJarPath)
	throws IOException {
	return String.join(" ",
	"start server",
	"--name=" + serverName,
	"--groups=" + serverName,
	"--dir=" + temporaryFolder.newFolder(serverName).getAbsolutePath(),
	"--server-port=" + serverPort,
	"--locators=localhost[" + locatorPort + "]",
	"--classpath=" + serviceJarPath + pathSeparatorChar + functionsJarPath);
	}

	private void executeFunctionOnReplicateRegion(Function<? super String[]> function,
	Duration duration) {
	executeFunctionOnRegion(function, duration, replicateRegion);
	}

	private void executeFunctionOnPartitionRegion(Function<? super String[]> function,
	Duration duration) {
	executeFunctionOnRegion(function, duration, partitionRegion);
	}

	private void executeFunctionOnRegion(Function<? super String[]> function, Duration duration,
	Region<?, ?> region) {
	@SuppressWarnings("unchecked")
	Execution<String[], Object, List<Object>> execution =
	(Execution<String[], Object, List<Object>>) FunctionService.onRegion(region);

	execution
	.setArguments(new String[] {String.valueOf(duration.toMillis()), TRUE.toString()})
	.execute(function)
	.getResult();
	}

	private List<ExecutionsTimerValues> getServer1ExecutionsTimerValues(String functionId) {
	return getExecutionsTimerValuesFromPool(functionId, server1Pool);
	}

	private List<ExecutionsTimerValues> getServer2ExecutionsTimerValues(String functionId) {
	return getExecutionsTimerValuesFromPool(functionId, server2Pool);
	}

	private List<ExecutionsTimerValues> getAllExecutionsTimerValues(String functionId) {
	return getExecutionsTimerValuesFromPool(functionId, multiServerPool);
	}

	private List<ExecutionsTimerValues> getExecutionsTimerValuesFromPool(String functionId,
	Pool pool) {
	@SuppressWarnings("unchecked")
	Execution<Void, List<ExecutionsTimerValues>, List<List<ExecutionsTimerValues>>> functionExecution =
	(Execution<Void, List<ExecutionsTimerValues>, List<List<ExecutionsTimerValues>>>) FunctionService
	.onServers(pool);

	List<List<ExecutionsTimerValues>> timerValuesForEachServer = functionExecution
	.execute(new GetFunctionExecutionTimerValues())
	.getResult();

	return timerValuesForEachServer.stream()
	.flatMap(List::stream)
	.filter(v -> v.functionId.equals(functionId))
	.collect(toList());
	}

	private static Long getAggregateCount(List<ExecutionsTimerValues> values) {
	return values.stream().map(x -> x.count).reduce(0L, Long::sum);
	}

	private static Double getAggregateTotalTime(List<ExecutionsTimerValues> values) {
	return values.stream().map(x -> x.totalTime).reduce(0.0, Double::sum);
	}
	}