geode-assembly/src/acceptanceTest/java/org/apache/geode/metrics/CacheGetsTimerTest.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;

 import static java.io.File.pathSeparatorChar;
 import static java.util.concurrent.TimeUnit.NANOSECONDS;
 import static java.util.stream.Collectors.toList;
 import static org.apache.geode.cache.execute.FunctionService.onServer;
 import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_MANAGER;
 import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPorts;
 import static org.apache.geode.test.compiler.ClassBuilder.writeJarFromClasses;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.catchThrowable;

 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.Serializable;
 import java.nio.file.Path;
 import java.util.Collection;
 import java.util.List;
 import java.util.Properties;

 import io.micrometer.core.instrument.Timer;
 import org.junit.After;
 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.execute.Function;
 import org.apache.geode.cache.execute.FunctionContext;
 import org.apache.geode.distributed.DistributedMember;
 import org.apache.geode.rules.ServiceJarRule;
 import org.apache.geode.security.AuthInitialize;
 import org.apache.geode.security.AuthenticationFailedException;
 import org.apache.geode.security.NotAuthorizedException;
 import org.apache.geode.security.ResourcePermission;
 import org.apache.geode.test.junit.rules.gfsh.GfshRule;

 public class CacheGetsTimerTest {
   private int locatorPort;
   private ClientCache clientCache;
   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();

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

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

   @Test
   public void hitTimerRecordsCountAndTotalTime_ifGetsPerformedOnReplicateRegionWithExistingKey()
       throws IOException {
     givenClusterWithTimeStatisticsEnabled();

     for (int i = 0; i < 5; i++) {
       replicateRegion.put(i, i);
       replicateRegion.get(i);
     }

     TimerValue hitTimerValue = hitTimerValueForRegion(replicateRegion);

     assertThat(hitTimerValue.region)
         .as("Cache gets hit timer region tag value")
         .isEqualTo(replicateRegion.getName());

     assertThat(hitTimerValue.result)
         .as("Cache gets hit timer result tag value")
         .isEqualTo("hit");

     assertThat(hitTimerValue.count)
         .as("Cache gets hit timer count for region " + replicateRegion.getName())
         .isEqualTo(5);

     assertThat(hitTimerValue.totalTime)
         .as("Cache gets hit timer total time for region " + replicateRegion.getName())
         .isGreaterThan(0);
   }

   @Test
   public void missTimerRecordsCountAndTotalTime_ifGetsPerformedOnReplicateRegionWithMissingKey()
       throws IOException {
     givenClusterWithTimeStatisticsEnabled();

     for (int i = 0; i < 5; i++) {
       replicateRegion.get(i);
     }

     TimerValue missTimerValue = missTimerValueForRegion(replicateRegion);

     assertThat(missTimerValue.region)
         .as("Cache gets miss timer region tag value")
         .isEqualTo(replicateRegion.getName());

     assertThat(missTimerValue.result)
         .as("Cache gets miss timer result tag value")
         .isEqualTo("miss");

     assertThat(missTimerValue.count)
         .as("Cache gets miss timer count for region " + replicateRegion.getName())
         .isEqualTo(5);

     assertThat(missTimerValue.totalTime)
         .as("Cache gets miss timer total time for region " + replicateRegion.getName())
         .isGreaterThan(0);
   }

   @Test
   public void hitTimerRecordsCountAndTotalTime_ifGetsPerformedOnPartitionRegionWithExistingKey()
       throws IOException {
     givenClusterWithTimeStatisticsEnabled();

     for (int i = 0; i < 5; i++) {
       partitionRegion.put(i, i);
       partitionRegion.get(i);
     }

     TimerValue hitTimerValue = hitTimerValueForRegion(partitionRegion);

     assertThat(hitTimerValue.region)
         .as("Cache gets hit timer region tag value")
         .isEqualTo(partitionRegion.getName());

     assertThat(hitTimerValue.result)
         .as("Cache gets hit timer result tag value")
         .isEqualTo("hit");

     assertThat(hitTimerValue.count)
         .as("Cache gets hit timer count for region " + partitionRegion.getName())
         .isEqualTo(5);

     assertThat(hitTimerValue.totalTime)
         .as("Cache gets hit timer total time for region " + partitionRegion.getName())
         .isGreaterThan(0);
   }

   @Test
   public void missTimerRecordsCountAndTotalTime_ifGetsPerformedOnPartitionRegionWithMissingKey()
       throws IOException {
     givenClusterWithTimeStatisticsEnabled();

     for (int i = 0; i < 5; i++) {
       partitionRegion.get(i);
     }

     TimerValue missTimerValue = missTimerValueForRegion(partitionRegion);

     assertThat(missTimerValue.region)
         .as("Cache gets miss timer region tag value")
         .isEqualTo(partitionRegion.getName());

     assertThat(missTimerValue.result)
         .as("Cache gets miss timer result tag value")
         .isEqualTo("miss");

     assertThat(missTimerValue.count)
         .as("Cache gets miss timer count for region " + partitionRegion.getName())
         .isEqualTo(5);

     assertThat(missTimerValue.totalTime)
         .as("Cache gets miss timer total time for region " + partitionRegion.getName())
         .isGreaterThan(0);
   }

   @Test
   public void timersExistWithInitialValues_ifNoGetsPerformedOnReplicateRegion() throws IOException {
     givenClusterWithTimeStatisticsEnabled();

     assertThat(allTimerValuesForRegion(replicateRegion))
         .as("All timer values for region " + replicateRegion.getName())
         .hasSize(2)
         .anyMatch(tv -> tv.result.equals("hit"))
         .anyMatch(tv -> tv.result.equals("miss"))
         .allMatch(tv -> tv.count == 0, "All timers have count of zero")
         .allMatch(tv -> tv.totalTime == 0, "All timers have total time of zero");
   }

   @Test
   public void allTimersRemoved_ifReplicateRegionDestroyed() throws IOException {
     givenClusterWithTimeStatisticsEnabled();

     assertThat(allTimerValuesForRegion(replicateRegion))
         .as("All timer values before destroying region " + replicateRegion.getName())
         .hasSize(2);

     replicateRegion.destroyRegion();

     assertThat(allTimerValuesForRegion(replicateRegion))
         .as("All timer values after destroying region " + replicateRegion.getName())
         .isEmpty();
   }

   @Test
   public void timersExistWithInitialValues_ifNoGetsPerformedOnPartitionRegion() throws IOException {
     givenClusterWithTimeStatisticsEnabled();

     assertThat(allTimerValuesForRegion(partitionRegion))
         .as("All timer values for region " + partitionRegion.getName())
         .hasSize(2)
         .anyMatch(tv -> tv.result.equals("hit"))
         .anyMatch(tv -> tv.result.equals("miss"))
         .allMatch(tv -> tv.count == 0, "All timers have count of zero")
         .allMatch(tv -> tv.totalTime == 0, "All timers have total time of zero");
   }

   @Test
   public void allTimersRemoved_ifPartitionRegionDestroyed() throws IOException {
     givenClusterWithTimeStatisticsEnabled();

     assertThat(allTimerValuesForRegion(partitionRegion))
         .as("All timer values before destroying region " + partitionRegion.getName())
         .hasSize(2);

     partitionRegion.destroyRegion();

     assertThat(allTimerValuesForRegion(partitionRegion))
         .as("All timer values after destroying region " + partitionRegion.getName())
         .isEmpty();
   }

   @Test
   public void timersRecordCountForReplicateRegion_ifTimeStatisticsDisabled() throws IOException {
     givenClusterWithTimeStatisticsDisabled();

     replicateRegion.put("existing-key", "existing-value");
     replicateRegion.get("existing-key");
     replicateRegion.get("missing-key");

     assertThat(allTimerValuesForRegion(replicateRegion))
         .as("All timer values for region " + replicateRegion.getName())
         .hasSize(2)
         .anyMatch(tv -> tv.result.equals("hit"))
         .anyMatch(tv -> tv.result.equals("miss"))
         .allMatch(tv -> tv.count == 1, "All timers have count of one")
         .allMatch(tv -> tv.totalTime == 0, "All timers have total time of zero");
   }

   @Test
   public void timersRecordCountForPartitionRegion_ifTimeStatisticsDisabled() throws IOException {
     givenClusterWithTimeStatisticsDisabled();

     partitionRegion.put("existing-key", "existing-value");
     partitionRegion.get("existing-key");
     partitionRegion.get("missing-key");

     assertThat(allTimerValuesForRegion(partitionRegion))
         .as("All timer values for region " + partitionRegion.getName())
         .hasSize(2)
         .anyMatch(tv -> tv.result.equals("hit"))
         .anyMatch(tv -> tv.result.equals("miss"))
         .allMatch(tv -> tv.count == 1, "All timers have count of one")
         .allMatch(tv -> tv.totalTime == 0, "All timers have total time of zero");
   }

   @Test
   public void timersDoNotRecord_ifGetThrows() throws IOException {
     givenClusterWithSecurityThatDeniesAllGetOperations();

     Throwable thrown = catchThrowable(() -> replicateRegion.get("key"));

     assertThat(thrown)
         .as("Exception from get operation")
         .hasCauseInstanceOf(NotAuthorizedException.class);

     assertThat(allTimerValuesForRegion(replicateRegion))
         .as("All timer values for region " + replicateRegion.getName())
         .hasSize(2)
         .allMatch(tv -> tv.count == 0, "All timers have count of zero");
   }

   private void givenClusterWithTimeStatisticsEnabled() throws IOException {
     startCluster(true, false);
   }

   private void givenClusterWithTimeStatisticsDisabled() throws IOException {
     startCluster(false, false);
   }

   private void givenClusterWithSecurityThatDeniesAllGetOperations() throws IOException {
     startCluster(true, true);
   }

   private void startCluster(boolean enableTimeStatistics, boolean enableSecurity)
       throws IOException {
     int[] availablePorts = 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 helpersJarPath = temporaryFolder.getRoot().toPath()
         .resolve("helpers.jar").toAbsolutePath();
     writeJarFromClasses(helpersJarPath.toFile(), TimerValue.class,
         FetchCacheGetsTimerValues.class, DenyAllDataRead.class, ClientSecurityConfig.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,
         "--classpath=" + helpersJarPath);

     String serverName = "server";
     String startServerCommand = String.join(" ",
         "start server",
         "--name=" + serverName,
         "--dir=" + temporaryFolder.newFolder(serverName).getAbsolutePath(),
         "--server-port=" + serverPort,
         "--locators=localhost[" + locatorPort + "]",
         "--classpath=" + serviceJarPath + pathSeparatorChar + helpersJarPath);

     if (enableTimeStatistics) {
       startServerCommand += " --enable-time-statistics";
     }

     if (enableSecurity) {
       File securityPropertiesFile = createSecurityPropertiesFile();
       String securityPropertiesFileOption =
           " --security-properties-file=" + securityPropertiesFile.getAbsolutePath();
       startLocatorCommand += securityPropertiesFileOption;
       startServerCommand += securityPropertiesFileOption;
     }

     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);

     ClientCacheFactory clientCacheFactory = new ClientCacheFactory();
     if (enableSecurity) {
       clientCacheFactory.set("security-client-auth-init", ClientSecurityConfig.class.getName());
     }

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

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

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

   private File createSecurityPropertiesFile() throws IOException {
     Properties securityProperties = ClientSecurityConfig.securityProperties();
     File securityPropertiesFile = gfshRule.getTemporaryFolder().newFile("security.properties");
     securityProperties.store(new FileOutputStream(securityPropertiesFile), null);
     return securityPropertiesFile;
   }

   private TimerValue hitTimerValueForRegion(Region<?, ?> region) {
     return timerValueForRegionAndResult(region, "hit");
   }

   private TimerValue missTimerValueForRegion(Region<?, ?> region) {
     return timerValueForRegionAndResult(region, "miss");
   }

   private TimerValue timerValueForRegionAndResult(Region<?, ?> region, String resultTagValue) {
     List<TimerValue> cacheGetsTimerValues = allTimerValuesForRegion(region).stream()
         .filter(tv -> tv.result.equals(resultTagValue))
         .collect(toList());

     assertThat(cacheGetsTimerValues)
         .as("Timers for region " + region.getName() + " with result " + resultTagValue)
         .hasSize(1);

     return cacheGetsTimerValues.get(0);
   }

   private List<TimerValue> allTimerValuesForRegion(Region<?, ?> region) {
     @SuppressWarnings("unchecked")
     List<List<TimerValue>> timerValuesFromAllServers =
         (List<List<TimerValue>>) onServer(clientCache)
             .execute(new FetchCacheGetsTimerValues())
             .getResult();

     assertThat(timerValuesFromAllServers)
         .hasSize(1);

     return timerValuesFromAllServers.get(0).stream()
         .filter(tv -> tv.region.equals(region.getName()))
         .collect(toList());
   }

   static class TimerValue implements Serializable {
     final long count;
     final double totalTime;
     final String region;
     final String result;

     TimerValue(long count, double totalTime, String region, String result) {
       this.count = count;
       this.totalTime = totalTime;
       this.region = region;
       this.result = result;
     }

     @Override
     public String toString() {
       return "TimerValue{" +
           "count=" + count +
           ", totalTime=" + totalTime +
           ", region='" + region + '\'' +
           ", result='" + result + '\'' +
           '}';
     }
   }

   static class FetchCacheGetsTimerValues implements Function<Void> {
     private static final String ID = "FetchCacheGetsTimerValues";

     @Override
     public void execute(FunctionContext<Void> context) {
       Collection<Timer> timers = SimpleMetricsPublishingService.getRegistry()
           .find("geode.cache.gets")
           .timers();

       List<TimerValue> result = timers.stream()
           .map(FetchCacheGetsTimerValues::toTimerValues)
           .collect(toList());

       context.getResultSender().lastResult(result);
     }

     @Override
     public String getId() {
       return ID;
     }

     private static TimerValue toTimerValues(Timer t) {
       String region = t.getId().getTag("region");
       String result = t.getId().getTag("result");

       return new TimerValue(
           t.count(),
           t.totalTime(NANOSECONDS),
           region,
           result);
     }
   }

   public static class DenyAllDataRead implements org.apache.geode.security.SecurityManager {
     @Override
     public Object authenticate(Properties credentials) throws AuthenticationFailedException {
       String userName = credentials.getProperty(USER_NAME);
       if (userName == null) {
         throw new AuthenticationFailedException("No user name provided");
       }
       return userName;
     }

     @Override
     public boolean authorize(Object principal, ResourcePermission permission) {
       return !isDataRead(permission);
     }

     private static boolean isDataRead(ResourcePermission permission) {
       return ResourcePermission.Resource.DATA.equals(permission.getResource())
           && ResourcePermission.Operation.READ.equals(permission.getOperation());
     }
   }

   public static class ClientSecurityConfig implements AuthInitialize {
     @Override
     public Properties getCredentials(Properties securityProps, DistributedMember server,
         boolean isPeer) throws AuthenticationFailedException {
       return securityProperties();
     }

     static Properties securityProperties() {
       Properties securityProperties = new Properties();
       securityProperties.setProperty(SECURITY_MANAGER, DenyAllDataRead.class.getName());
       securityProperties.setProperty("security-username", "user");
       securityProperties.setProperty("security-password", "pass");
       return securityProperties;
     }
   }
 }
	/*
	* 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;

	import static java.io.File.pathSeparatorChar;
	import static java.util.concurrent.TimeUnit.NANOSECONDS;
	import static java.util.stream.Collectors.toList;
	import static org.apache.geode.cache.execute.FunctionService.onServer;
	import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_MANAGER;
	import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPorts;
	import static org.apache.geode.test.compiler.ClassBuilder.writeJarFromClasses;
	import static org.assertj.core.api.Assertions.assertThat;
	import static org.assertj.core.api.Assertions.catchThrowable;

	import java.io.File;
	import java.io.FileOutputStream;
	import java.io.IOException;
	import java.io.Serializable;
	import java.nio.file.Path;
	import java.util.Collection;
	import java.util.List;
	import java.util.Properties;

	import io.micrometer.core.instrument.Timer;
	import org.junit.After;
	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.execute.Function;
	import org.apache.geode.cache.execute.FunctionContext;
	import org.apache.geode.distributed.DistributedMember;
	import org.apache.geode.rules.ServiceJarRule;
	import org.apache.geode.security.AuthInitialize;
	import org.apache.geode.security.AuthenticationFailedException;
	import org.apache.geode.security.NotAuthorizedException;
	import org.apache.geode.security.ResourcePermission;
	import org.apache.geode.test.junit.rules.gfsh.GfshRule;

	public class CacheGetsTimerTest {
	private int locatorPort;
	private ClientCache clientCache;
	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();

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

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

	@Test
	public void hitTimerRecordsCountAndTotalTime_ifGetsPerformedOnReplicateRegionWithExistingKey()
	throws IOException {
	givenClusterWithTimeStatisticsEnabled();

	for (int i = 0; i < 5; i++) {
	replicateRegion.put(i, i);
	replicateRegion.get(i);
	}

	TimerValue hitTimerValue = hitTimerValueForRegion(replicateRegion);

	assertThat(hitTimerValue.region)
	.as("Cache gets hit timer region tag value")
	.isEqualTo(replicateRegion.getName());

	assertThat(hitTimerValue.result)
	.as("Cache gets hit timer result tag value")
	.isEqualTo("hit");

	assertThat(hitTimerValue.count)
	.as("Cache gets hit timer count for region " + replicateRegion.getName())
	.isEqualTo(5);

	assertThat(hitTimerValue.totalTime)
	.as("Cache gets hit timer total time for region " + replicateRegion.getName())
	.isGreaterThan(0);
	}

	@Test
	public void missTimerRecordsCountAndTotalTime_ifGetsPerformedOnReplicateRegionWithMissingKey()
	throws IOException {
	givenClusterWithTimeStatisticsEnabled();

	for (int i = 0; i < 5; i++) {
	replicateRegion.get(i);
	}

	TimerValue missTimerValue = missTimerValueForRegion(replicateRegion);

	assertThat(missTimerValue.region)
	.as("Cache gets miss timer region tag value")
	.isEqualTo(replicateRegion.getName());

	assertThat(missTimerValue.result)
	.as("Cache gets miss timer result tag value")
	.isEqualTo("miss");

	assertThat(missTimerValue.count)
	.as("Cache gets miss timer count for region " + replicateRegion.getName())
	.isEqualTo(5);

	assertThat(missTimerValue.totalTime)
	.as("Cache gets miss timer total time for region " + replicateRegion.getName())
	.isGreaterThan(0);
	}

	@Test
	public void hitTimerRecordsCountAndTotalTime_ifGetsPerformedOnPartitionRegionWithExistingKey()
	throws IOException {
	givenClusterWithTimeStatisticsEnabled();

	for (int i = 0; i < 5; i++) {
	partitionRegion.put(i, i);
	partitionRegion.get(i);
	}

	TimerValue hitTimerValue = hitTimerValueForRegion(partitionRegion);

	assertThat(hitTimerValue.region)
	.as("Cache gets hit timer region tag value")
	.isEqualTo(partitionRegion.getName());

	assertThat(hitTimerValue.result)
	.as("Cache gets hit timer result tag value")
	.isEqualTo("hit");

	assertThat(hitTimerValue.count)
	.as("Cache gets hit timer count for region " + partitionRegion.getName())
	.isEqualTo(5);

	assertThat(hitTimerValue.totalTime)
	.as("Cache gets hit timer total time for region " + partitionRegion.getName())
	.isGreaterThan(0);
	}

	@Test
	public void missTimerRecordsCountAndTotalTime_ifGetsPerformedOnPartitionRegionWithMissingKey()
	throws IOException {
	givenClusterWithTimeStatisticsEnabled();

	for (int i = 0; i < 5; i++) {
	partitionRegion.get(i);
	}

	TimerValue missTimerValue = missTimerValueForRegion(partitionRegion);

	assertThat(missTimerValue.region)
	.as("Cache gets miss timer region tag value")
	.isEqualTo(partitionRegion.getName());

	assertThat(missTimerValue.result)
	.as("Cache gets miss timer result tag value")
	.isEqualTo("miss");

	assertThat(missTimerValue.count)
	.as("Cache gets miss timer count for region " + partitionRegion.getName())
	.isEqualTo(5);

	assertThat(missTimerValue.totalTime)
	.as("Cache gets miss timer total time for region " + partitionRegion.getName())
	.isGreaterThan(0);
	}

	@Test
	public void timersExistWithInitialValues_ifNoGetsPerformedOnReplicateRegion() throws IOException {
	givenClusterWithTimeStatisticsEnabled();

	assertThat(allTimerValuesForRegion(replicateRegion))
	.as("All timer values for region " + replicateRegion.getName())
	.hasSize(2)
	.anyMatch(tv -> tv.result.equals("hit"))
	.anyMatch(tv -> tv.result.equals("miss"))
	.allMatch(tv -> tv.count == 0, "All timers have count of zero")
	.allMatch(tv -> tv.totalTime == 0, "All timers have total time of zero");
	}

	@Test
	public void allTimersRemoved_ifReplicateRegionDestroyed() throws IOException {
	givenClusterWithTimeStatisticsEnabled();

	assertThat(allTimerValuesForRegion(replicateRegion))
	.as("All timer values before destroying region " + replicateRegion.getName())
	.hasSize(2);

	replicateRegion.destroyRegion();

	assertThat(allTimerValuesForRegion(replicateRegion))
	.as("All timer values after destroying region " + replicateRegion.getName())
	.isEmpty();
	}

	@Test
	public void timersExistWithInitialValues_ifNoGetsPerformedOnPartitionRegion() throws IOException {
	givenClusterWithTimeStatisticsEnabled();

	assertThat(allTimerValuesForRegion(partitionRegion))
	.as("All timer values for region " + partitionRegion.getName())
	.hasSize(2)
	.anyMatch(tv -> tv.result.equals("hit"))
	.anyMatch(tv -> tv.result.equals("miss"))
	.allMatch(tv -> tv.count == 0, "All timers have count of zero")
	.allMatch(tv -> tv.totalTime == 0, "All timers have total time of zero");
	}

	@Test
	public void allTimersRemoved_ifPartitionRegionDestroyed() throws IOException {
	givenClusterWithTimeStatisticsEnabled();

	assertThat(allTimerValuesForRegion(partitionRegion))
	.as("All timer values before destroying region " + partitionRegion.getName())
	.hasSize(2);

	partitionRegion.destroyRegion();

	assertThat(allTimerValuesForRegion(partitionRegion))
	.as("All timer values after destroying region " + partitionRegion.getName())
	.isEmpty();
	}

	@Test
	public void timersRecordCountForReplicateRegion_ifTimeStatisticsDisabled() throws IOException {
	givenClusterWithTimeStatisticsDisabled();

	replicateRegion.put("existing-key", "existing-value");
	replicateRegion.get("existing-key");
	replicateRegion.get("missing-key");

	assertThat(allTimerValuesForRegion(replicateRegion))
	.as("All timer values for region " + replicateRegion.getName())
	.hasSize(2)
	.anyMatch(tv -> tv.result.equals("hit"))
	.anyMatch(tv -> tv.result.equals("miss"))
	.allMatch(tv -> tv.count == 1, "All timers have count of one")
	.allMatch(tv -> tv.totalTime == 0, "All timers have total time of zero");
	}

	@Test
	public void timersRecordCountForPartitionRegion_ifTimeStatisticsDisabled() throws IOException {
	givenClusterWithTimeStatisticsDisabled();

	partitionRegion.put("existing-key", "existing-value");
	partitionRegion.get("existing-key");
	partitionRegion.get("missing-key");

	assertThat(allTimerValuesForRegion(partitionRegion))
	.as("All timer values for region " + partitionRegion.getName())
	.hasSize(2)
	.anyMatch(tv -> tv.result.equals("hit"))
	.anyMatch(tv -> tv.result.equals("miss"))
	.allMatch(tv -> tv.count == 1, "All timers have count of one")
	.allMatch(tv -> tv.totalTime == 0, "All timers have total time of zero");
	}

	@Test
	public void timersDoNotRecord_ifGetThrows() throws IOException {
	givenClusterWithSecurityThatDeniesAllGetOperations();

	Throwable thrown = catchThrowable(() -> replicateRegion.get("key"));

	assertThat(thrown)
	.as("Exception from get operation")
	.hasCauseInstanceOf(NotAuthorizedException.class);

	assertThat(allTimerValuesForRegion(replicateRegion))
	.as("All timer values for region " + replicateRegion.getName())
	.hasSize(2)
	.allMatch(tv -> tv.count == 0, "All timers have count of zero");
	}

	private void givenClusterWithTimeStatisticsEnabled() throws IOException {
	startCluster(true, false);
	}

	private void givenClusterWithTimeStatisticsDisabled() throws IOException {
	startCluster(false, false);
	}

	private void givenClusterWithSecurityThatDeniesAllGetOperations() throws IOException {
	startCluster(true, true);
	}

	private void startCluster(boolean enableTimeStatistics, boolean enableSecurity)
	throws IOException {
	int[] availablePorts = 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 helpersJarPath = temporaryFolder.getRoot().toPath()
	.resolve("helpers.jar").toAbsolutePath();
	writeJarFromClasses(helpersJarPath.toFile(), TimerValue.class,
	FetchCacheGetsTimerValues.class, DenyAllDataRead.class, ClientSecurityConfig.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,
	"--classpath=" + helpersJarPath);

	String serverName = "server";
	String startServerCommand = String.join(" ",
	"start server",
	"--name=" + serverName,
	"--dir=" + temporaryFolder.newFolder(serverName).getAbsolutePath(),
	"--server-port=" + serverPort,
	"--locators=localhost[" + locatorPort + "]",
	"--classpath=" + serviceJarPath + pathSeparatorChar + helpersJarPath);

	if (enableTimeStatistics) {
	startServerCommand += " --enable-time-statistics";
	}

	if (enableSecurity) {
	File securityPropertiesFile = createSecurityPropertiesFile();
	String securityPropertiesFileOption =
	" --security-properties-file=" + securityPropertiesFile.getAbsolutePath();
	startLocatorCommand += securityPropertiesFileOption;
	startServerCommand += securityPropertiesFileOption;
	}

	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);

	ClientCacheFactory clientCacheFactory = new ClientCacheFactory();
	if (enableSecurity) {
	clientCacheFactory.set("security-client-auth-init", ClientSecurityConfig.class.getName());
	}

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

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

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

	private File createSecurityPropertiesFile() throws IOException {
	Properties securityProperties = ClientSecurityConfig.securityProperties();
	File securityPropertiesFile = gfshRule.getTemporaryFolder().newFile("security.properties");
	securityProperties.store(new FileOutputStream(securityPropertiesFile), null);
	return securityPropertiesFile;
	}

	private TimerValue hitTimerValueForRegion(Region<?, ?> region) {
	return timerValueForRegionAndResult(region, "hit");
	}

	private TimerValue missTimerValueForRegion(Region<?, ?> region) {
	return timerValueForRegionAndResult(region, "miss");
	}

	private TimerValue timerValueForRegionAndResult(Region<?, ?> region, String resultTagValue) {
	List<TimerValue> cacheGetsTimerValues = allTimerValuesForRegion(region).stream()
	.filter(tv -> tv.result.equals(resultTagValue))
	.collect(toList());

	assertThat(cacheGetsTimerValues)
	.as("Timers for region " + region.getName() + " with result " + resultTagValue)
	.hasSize(1);

	return cacheGetsTimerValues.get(0);
	}

	private List<TimerValue> allTimerValuesForRegion(Region<?, ?> region) {
	@SuppressWarnings("unchecked")
	List<List<TimerValue>> timerValuesFromAllServers =
	(List<List<TimerValue>>) onServer(clientCache)
	.execute(new FetchCacheGetsTimerValues())
	.getResult();

	assertThat(timerValuesFromAllServers)
	.hasSize(1);

	return timerValuesFromAllServers.get(0).stream()
	.filter(tv -> tv.region.equals(region.getName()))
	.collect(toList());
	}

	static class TimerValue implements Serializable {
	final long count;
	final double totalTime;
	final String region;
	final String result;

	TimerValue(long count, double totalTime, String region, String result) {
	this.count = count;
	this.totalTime = totalTime;
	this.region = region;
	this.result = result;
	}

	@Override
	public String toString() {
	return "TimerValue{" +
	"count=" + count +
	", totalTime=" + totalTime +
	", region='" + region + '\'' +
	", result='" + result + '\'' +
	'}';
	}
	}

	static class FetchCacheGetsTimerValues implements Function<Void> {
	private static final String ID = "FetchCacheGetsTimerValues";

	@Override
	public void execute(FunctionContext<Void> context) {
	Collection<Timer> timers = SimpleMetricsPublishingService.getRegistry()
	.find("geode.cache.gets")
	.timers();

	List<TimerValue> result = timers.stream()
	.map(FetchCacheGetsTimerValues::toTimerValues)
	.collect(toList());

	context.getResultSender().lastResult(result);
	}

	@Override
	public String getId() {
	return ID;
	}

	private static TimerValue toTimerValues(Timer t) {
	String region = t.getId().getTag("region");
	String result = t.getId().getTag("result");

	return new TimerValue(
	t.count(),
	t.totalTime(NANOSECONDS),
	region,
	result);
	}
	}

	public static class DenyAllDataRead implements org.apache.geode.security.SecurityManager {
	@Override
	public Object authenticate(Properties credentials) throws AuthenticationFailedException {
	String userName = credentials.getProperty(USER_NAME);
	if (userName == null) {
	throw new AuthenticationFailedException("No user name provided");
	}
	return userName;
	}

	@Override
	public boolean authorize(Object principal, ResourcePermission permission) {
	return !isDataRead(permission);
	}

	private static boolean isDataRead(ResourcePermission permission) {
	return ResourcePermission.Resource.DATA.equals(permission.getResource())
	&& ResourcePermission.Operation.READ.equals(permission.getOperation());
	}
	}

	public static class ClientSecurityConfig implements AuthInitialize {
	@Override
	public Properties getCredentials(Properties securityProps, DistributedMember server,
	boolean isPeer) throws AuthenticationFailedException {
	return securityProperties();
	}

	static Properties securityProperties() {
	Properties securityProperties = new Properties();
	securityProperties.setProperty(SECURITY_MANAGER, DenyAllDataRead.class.getName());
	securityProperties.setProperty("security-username", "user");
	securityProperties.setProperty("security-password", "pass");
	return securityProperties;
	}
	}
	}