geode-assembly/src/acceptanceTest/java/org/apache/geode/metrics/function/executions/FunctionExecutionsTimerNoResultTest.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.util.stream.Collectors.toList;
 import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
 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.assertj.core.api.Assertions;
 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.RegionService;
 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.execute.Execution;
 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 for functions with hasResult=false
  */
 public class FunctionExecutionsTimerNoResultTest {

   private int locatorPort;
   private ClientCache clientCache;
   private Region<Object, Object> replicateRegion;
   private Region<Object, Object> partitionRegion;
   private FunctionToTimeWithoutResult functionWithNoResult;
   private Duration functionDuration;

   @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(3);

     locatorPort = availablePorts[0];
     int locatorJmxPort = availablePorts[1];
     int serverPort = availablePorts[2];

     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, FunctionToTimeWithoutResult.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 serverName = "server1";
     String startServerCommand =
         startServerCommand(serverName, serverPort, 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",
         "--name=" + partitionRegionName);

     gfshRule.execute(startLocatorCommand, startServerCommand, createReplicateRegionCommand,
         createPartitionRegionCommand);

     clientCache = new ClientCacheFactory()
         .addPoolLocator("localhost", locatorPort)
         .create();

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

     partitionRegion = clientCache
         .createClientRegionFactory(ClientRegionShortcut.PROXY)
         .create(partitionRegionName);

     functionWithNoResult = new FunctionToTimeWithoutResult();
     functionDuration = Duration.ofSeconds(1);
   }

   @After
   public void tearDown() {
     partitionRegion.close();
     replicateRegion.close();
     clientCache.close();

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

   @Test
   public void successTimerRecordsCountAndTotalTime_ifFunctionSucceeds_whenExecutedOnReplicateRegion() {
     executeFunctionThatSucceeds(onRegion(replicateRegion));

     await().untilAsserted(() -> {
       ExecutionsTimerValues value = successTimerValue();

       assertThat(value.count)
           .as("Number of successful executions")
           .isEqualTo(1);

       assertThat(value.totalTime)
           .as("Total time of successful executions")
           .isGreaterThan(functionDuration.toNanos());
     });
   }

   @Test
   public void failureTimerRecordsCountAndTotalTime_ifFunctionThrows_whenExecutedOnReplicateRegion() {
     executeFunctionThatThrows(onRegion(replicateRegion));

     await().untilAsserted(() -> {
       ExecutionsTimerValues value = failureTimerValue();

       assertThat(value.count)
           .as("Number of failed executions")
           .isEqualTo(1);

       assertThat(value.totalTime)
           .as("Total time of failed executions")
           .isGreaterThan(functionDuration.toNanos());
     });
   }

   @Test
   public void successTimerRecordsCountAndTotalTime_ifFunctionSucceeds_whenExecutedOnPartitionRegion() {
     executeFunctionThatSucceeds(onRegion(partitionRegion));

     await().untilAsserted(() -> {
       ExecutionsTimerValues value = successTimerValue();

       assertThat(value.count)
           .as("Number of successful executions")
           .isEqualTo(1);

       assertThat(value.totalTime)
           .as("Total time of successful executions")
           .isGreaterThan(functionDuration.toNanos());
     });
   }

   @Test
   public void failureTimerRecordsCountAndTotalTime_ifFunctionThrows_whenExecutedOnPartitionRegion() {
     executeFunctionThatThrows(onRegion(partitionRegion));

     await().untilAsserted(() -> {
       ExecutionsTimerValues value = failureTimerValue();

       Assertions.assertThat(value.count)
           .as("Number of failed executions")
           .isEqualTo(1);

       Assertions.assertThat(value.totalTime)
           .as("Total time of failed executions")
           .isGreaterThan(functionDuration.toNanos());
     });
   }

   @Test
   public void successTimerRecordsCountAndTotalTime_ifFunctionSucceeds_whenExecutedOnServer() {
     executeFunctionThatSucceeds(onServer(clientCache));

     await().untilAsserted(() -> {
       ExecutionsTimerValues value = successTimerValue();

       assertThat(value.count)
           .as("Number of successful executions")
           .isEqualTo(1);

       assertThat(value.totalTime)
           .as("Total time of successful executions")
           .isGreaterThan(functionDuration.toNanos());
     });
   }

   @Test
   public void failureTimerRecordsCountAndTotalTime_ifFunctionThrows_whenExecutedOnServer() {
     executeFunctionThatThrows(onServer(clientCache));

     await().untilAsserted(() -> {
       ExecutionsTimerValues value = failureTimerValue();

       assertThat(value.count)
           .as("Number of failed executions")
           .isEqualTo(1);

       assertThat(value.totalTime)
           .as("Total time of failed executions")
           .isGreaterThan(functionDuration.toNanos());
     });
   }

   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 executeFunctionThatSucceeds(Execution<? super String[], ?, ?> execution) {
     executeFunction(execution, true);
   }

   private void executeFunctionThatThrows(Execution<? super String[], ?, ?> execution) {
     executeFunction(execution, false);
   }

   private void executeFunction(Execution<? super String[], ?, ?> execution, boolean isSuccessful) {
     execution
         .setArguments(new String[] {
             String.valueOf(functionDuration.toMillis()),
             String.valueOf(isSuccessful)})
         .execute(functionWithNoResult);
   }

   private ExecutionsTimerValues successTimerValue() {
     return getExecutionsTimerValuesFromServer(true);
   }

   private ExecutionsTimerValues failureTimerValue() {
     return getExecutionsTimerValuesFromServer(false);
   }

   private ExecutionsTimerValues getExecutionsTimerValuesFromServer(boolean isSuccessful) {
     List<List<ExecutionsTimerValues>> timerValuesForEachServer =
         FunctionExecutionsTimerNoResultTest
             .<Void, List<ExecutionsTimerValues>>onServer(clientCache)
             .execute(new GetFunctionExecutionTimerValues())
             .getResult();

     List<ExecutionsTimerValues> values = timerValuesForEachServer.stream()
         .flatMap(List::stream)
         .filter(v -> v.functionId.equals(FunctionToTimeWithoutResult.ID))
         .filter(v -> v.succeeded == isSuccessful)
         .collect(toList());

     Assertions.assertThat(values)
         .hasSize(1);

     return values.get(0);
   }

   @SuppressWarnings("unchecked")
   private static <IN, OUT> Execution<IN, OUT, List<OUT>> onRegion(Region<?, ?> region) {
     return (Execution<IN, OUT, List<OUT>>) FunctionService.onRegion(region);
   }

   @SuppressWarnings("unchecked")
   private static <IN, OUT> Execution<IN, OUT, List<OUT>> onServer(RegionService regionService) {
     return (Execution<IN, OUT, List<OUT>>) FunctionService.onServer(regionService);
   }
 }
	/*
	* 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.util.stream.Collectors.toList;
	import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
	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.assertj.core.api.Assertions;
	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.RegionService;
	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.execute.Execution;
	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 for functions with hasResult=false
	*/
	public class FunctionExecutionsTimerNoResultTest {

	private int locatorPort;
	private ClientCache clientCache;
	private Region<Object, Object> replicateRegion;
	private Region<Object, Object> partitionRegion;
	private FunctionToTimeWithoutResult functionWithNoResult;
	private Duration functionDuration;

	@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(3);

	locatorPort = availablePorts[0];
	int locatorJmxPort = availablePorts[1];
	int serverPort = availablePorts[2];

	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, FunctionToTimeWithoutResult.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 serverName = "server1";
	String startServerCommand =
	startServerCommand(serverName, serverPort, 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",
	"--name=" + partitionRegionName);

	gfshRule.execute(startLocatorCommand, startServerCommand, createReplicateRegionCommand,
	createPartitionRegionCommand);

	clientCache = new ClientCacheFactory()
	.addPoolLocator("localhost", locatorPort)
	.create();

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

	partitionRegion = clientCache
	.createClientRegionFactory(ClientRegionShortcut.PROXY)
	.create(partitionRegionName);

	functionWithNoResult = new FunctionToTimeWithoutResult();
	functionDuration = Duration.ofSeconds(1);
	}

	@After
	public void tearDown() {
	partitionRegion.close();
	replicateRegion.close();
	clientCache.close();

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

	@Test
	public void successTimerRecordsCountAndTotalTime_ifFunctionSucceeds_whenExecutedOnReplicateRegion() {
	executeFunctionThatSucceeds(onRegion(replicateRegion));

	await().untilAsserted(() -> {
	ExecutionsTimerValues value = successTimerValue();

	assertThat(value.count)
	.as("Number of successful executions")
	.isEqualTo(1);

	assertThat(value.totalTime)
	.as("Total time of successful executions")
	.isGreaterThan(functionDuration.toNanos());
	});
	}

	@Test
	public void failureTimerRecordsCountAndTotalTime_ifFunctionThrows_whenExecutedOnReplicateRegion() {
	executeFunctionThatThrows(onRegion(replicateRegion));

	await().untilAsserted(() -> {
	ExecutionsTimerValues value = failureTimerValue();

	assertThat(value.count)
	.as("Number of failed executions")
	.isEqualTo(1);

	assertThat(value.totalTime)
	.as("Total time of failed executions")
	.isGreaterThan(functionDuration.toNanos());
	});
	}

	@Test
	public void successTimerRecordsCountAndTotalTime_ifFunctionSucceeds_whenExecutedOnPartitionRegion() {
	executeFunctionThatSucceeds(onRegion(partitionRegion));

	await().untilAsserted(() -> {
	ExecutionsTimerValues value = successTimerValue();

	assertThat(value.count)
	.as("Number of successful executions")
	.isEqualTo(1);

	assertThat(value.totalTime)
	.as("Total time of successful executions")
	.isGreaterThan(functionDuration.toNanos());
	});
	}

	@Test
	public void failureTimerRecordsCountAndTotalTime_ifFunctionThrows_whenExecutedOnPartitionRegion() {
	executeFunctionThatThrows(onRegion(partitionRegion));

	await().untilAsserted(() -> {
	ExecutionsTimerValues value = failureTimerValue();

	Assertions.assertThat(value.count)
	.as("Number of failed executions")
	.isEqualTo(1);

	Assertions.assertThat(value.totalTime)
	.as("Total time of failed executions")
	.isGreaterThan(functionDuration.toNanos());
	});
	}

	@Test
	public void successTimerRecordsCountAndTotalTime_ifFunctionSucceeds_whenExecutedOnServer() {
	executeFunctionThatSucceeds(onServer(clientCache));

	await().untilAsserted(() -> {
	ExecutionsTimerValues value = successTimerValue();

	assertThat(value.count)
	.as("Number of successful executions")
	.isEqualTo(1);

	assertThat(value.totalTime)
	.as("Total time of successful executions")
	.isGreaterThan(functionDuration.toNanos());
	});
	}

	@Test
	public void failureTimerRecordsCountAndTotalTime_ifFunctionThrows_whenExecutedOnServer() {
	executeFunctionThatThrows(onServer(clientCache));

	await().untilAsserted(() -> {
	ExecutionsTimerValues value = failureTimerValue();

	assertThat(value.count)
	.as("Number of failed executions")
	.isEqualTo(1);

	assertThat(value.totalTime)
	.as("Total time of failed executions")
	.isGreaterThan(functionDuration.toNanos());
	});
	}

	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 executeFunctionThatSucceeds(Execution<? super String[], ?, ?> execution) {
	executeFunction(execution, true);
	}

	private void executeFunctionThatThrows(Execution<? super String[], ?, ?> execution) {
	executeFunction(execution, false);
	}

	private void executeFunction(Execution<? super String[], ?, ?> execution, boolean isSuccessful) {
	execution
	.setArguments(new String[] {
	String.valueOf(functionDuration.toMillis()),
	String.valueOf(isSuccessful)})
	.execute(functionWithNoResult);
	}

	private ExecutionsTimerValues successTimerValue() {
	return getExecutionsTimerValuesFromServer(true);
	}

	private ExecutionsTimerValues failureTimerValue() {
	return getExecutionsTimerValuesFromServer(false);
	}

	private ExecutionsTimerValues getExecutionsTimerValuesFromServer(boolean isSuccessful) {
	List<List<ExecutionsTimerValues>> timerValuesForEachServer =
	FunctionExecutionsTimerNoResultTest
	.<Void, List<ExecutionsTimerValues>>onServer(clientCache)
	.execute(new GetFunctionExecutionTimerValues())
	.getResult();

	List<ExecutionsTimerValues> values = timerValuesForEachServer.stream()
	.flatMap(List::stream)
	.filter(v -> v.functionId.equals(FunctionToTimeWithoutResult.ID))
	.filter(v -> v.succeeded == isSuccessful)
	.collect(toList());

	Assertions.assertThat(values)
	.hasSize(1);

	return values.get(0);
	}

	@SuppressWarnings("unchecked")
	private static <IN, OUT> Execution<IN, OUT, List<OUT>> onRegion(Region<?, ?> region) {
	return (Execution<IN, OUT, List<OUT>>) FunctionService.onRegion(region);
	}

	@SuppressWarnings("unchecked")
	private static <IN, OUT> Execution<IN, OUT, List<OUT>> onServer(RegionService regionService) {
	return (Execution<IN, OUT, List<OUT>>) FunctionService.onServer(regionService);
	}
	}