geode-redis/src/integrationTest/java/org/apache/geode/redis/internal/executor/server/AbstractRedisInfoStatsIntegrationTest.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.redis.internal.executor.server;

 import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
 import static org.assertj.core.api.Assertions.assertThat;

 import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;

 import com.google.common.util.concurrent.AtomicDouble;
 import org.assertj.core.data.Offset;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import redis.clients.jedis.Jedis;

 import org.apache.geode.internal.statistics.EnabledStatisticsClock;
 import org.apache.geode.internal.statistics.StatisticsClock;
 import org.apache.geode.test.awaitility.GeodeAwaitility;
 import org.apache.geode.test.dunit.rules.RedisPortSupplier;

 public abstract class AbstractRedisInfoStatsIntegrationTest implements RedisPortSupplier {

   private static final int TIMEOUT = (int) GeodeAwaitility.getTimeout().toMillis();
   private static final String EXISTING_HASH_KEY = "Existing_Hash";
   private static final String EXISTING_STRING_KEY = "Existing_String";
   private static final String EXISTING_SET_KEY_1 = "Existing_Set_1";
   private static final String EXISTING_SET_KEY_2 = "Existing_Set_2";

   private Jedis jedis;
   private static long START_TIME;
   private static StatisticsClock statisticsClock;

   private long preTestConnectionsReceived = 0;
   private long preTestConnectedClients = 0;

   private static final String COMMANDS_PROCESSED = "total_commands_processed";
   private static final String TOTAL_CONNECTIONS_RECEIVED = "total_connections_received";
   private static final String CONNECTED_CLIENTS = "connected_clients";
   private static final String OPS_PERFORMED_OVER_LAST_SECOND = "instantaneous_ops_per_sec";
   private static final String TOTAL_NETWORK_BYTES_READ = "total_net_input_bytes";
   private static final String NETWORK_KB_READ_OVER_LAST_SECOND = "instantaneous_input_kbps";
   private static final String UPTIME_IN_DAYS = "uptime_in_days";
   private static final String UPTIME_IN_SECONDS = "uptime_in_seconds";

   private static final AtomicInteger numInfoCalled = new AtomicInteger(0);


   // ------------------- Setup -------------------------- //
   @BeforeClass
   public static void beforeClass() {
     statisticsClock = new EnabledStatisticsClock();
     START_TIME = statisticsClock.getTime();
   }

   @Before
   public void before() {
     jedis = new Jedis("localhost", getPort(), TIMEOUT);
     numInfoCalled.set(0);

     long preSetupCommandsProcessed = Long.valueOf(getInfo(jedis).get(COMMANDS_PROCESSED));

     jedis.set(EXISTING_STRING_KEY, "A_Value");
     jedis.hset(EXISTING_HASH_KEY, "Field1", "Value1");
     jedis.sadd(EXISTING_SET_KEY_1, "m1", "m2", "m3");
     jedis.sadd(EXISTING_SET_KEY_2, "m4", "m5", "m6");

     // the info command increments command processed so we need to account for that.
     // the +1 is needed because info returns the number of commands processed before that call to
     // info
     await().atMost(Duration.ofSeconds(2))
         .untilAsserted(() -> assertThat(
             Long.valueOf(getInfo(jedis).get(COMMANDS_PROCESSED)) - numInfoCalled.get() + 1)
                 .isEqualTo(preSetupCommandsProcessed + 4));

     preTestConnectionsReceived = Long.valueOf(getInfo(jedis).get(TOTAL_CONNECTIONS_RECEIVED));
     preTestConnectedClients = Long.valueOf(getInfo(jedis).get(CONNECTED_CLIENTS));
     numInfoCalled.set(0);
   }

   @After
   public void after() {
     jedis.flushAll();
     await().atMost(Duration.ofSeconds(5))
         .untilAsserted(
             () -> assertThat(Long.valueOf(getInfo(jedis).get(CONNECTED_CLIENTS))).isEqualTo(1));
     jedis.close();
   }

   // ------------------- Stats Section -------------------------- //

   // note: see AbstractHitsMissesIntegrationTest for testing of hits/misses

   @Test
   public void commandsProcessed_shouldIncrement_givenSuccessfulCommand() {
     long initialCommandsProcessed = Long.valueOf(getInfo(jedis).get(COMMANDS_PROCESSED));
     jedis.ttl("key");

     validateCommandsProcessed(jedis, initialCommandsProcessed, 1);
   }

   @Test
   public void opsPerformedOverLastSecond_ShouldUpdate_givenOperationsOccurring() {
     int NUMBER_SECONDS_TO_RUN = 3;

     AtomicInteger numberOfCommandsExecuted = new AtomicInteger();
     AtomicDouble actualCommandsProcessedOverLastSecond = new AtomicDouble();

     await().during(Duration.ofSeconds(NUMBER_SECONDS_TO_RUN)).until(() -> {
       jedis.set("key", "value");
       numberOfCommandsExecuted.getAndIncrement();
       actualCommandsProcessedOverLastSecond.set(
           Double.valueOf(getInfo(jedis).get(OPS_PERFORMED_OVER_LAST_SECOND)));

       return true;
     });

     long expected = (numberOfCommandsExecuted.get() + numInfoCalled.get()) / NUMBER_SECONDS_TO_RUN;

     assertThat(actualCommandsProcessedOverLastSecond.get())
         .isCloseTo(expected,
             Offset.offset(getTenPercentOf(actualCommandsProcessedOverLastSecond.get())));

     // if time passes w/o operations
     await().during(NUMBER_SECONDS_TO_RUN, TimeUnit.SECONDS).until(() -> true);

     assertThat(Double.valueOf(getInfo(jedis).get(OPS_PERFORMED_OVER_LAST_SECOND))).isEqualTo(0D);
   }

   @Test
   public void networkBytesRead_shouldIncrementBySizeOfCommandSent() {
     long initialNetworkBytesRead = Long.valueOf(getInfo(jedis).get(TOTAL_NETWORK_BYTES_READ));
     String infoCommandString = "*3\r\n$3\r\ninfo\r\n";
     String respCommandString = "*3\r\n$3\r\nset\r\n$3\r\nkey\r\n$5\r\nvalue\r\n";

     jedis.set("key", "value");

     validateNetworkBytesRead(jedis, initialNetworkBytesRead,
         respCommandString.length() + infoCommandString.length());
   }

   @Test
   public void networkKiloBytesReadOverLastSecond_shouldBeCloseToBytesReadOverLastSecond() {

     double REASONABLE_SOUNDING_OFFSET = .8;
     int NUMBER_SECONDS_TO_RUN = 2;
     String RESP_COMMAND_STRING = "*3\r\n$3\r\nset\r\n$3\r\nkey\r\n$5\r\nvalue\r\n";
     int BYTES_SENT_PER_COMMAND = RESP_COMMAND_STRING.length();
     AtomicInteger totalBytesSent = new AtomicInteger();
     AtomicReference<Double> actual_kbs = new AtomicReference<>((double) 0);

     await().during(Duration.ofSeconds(NUMBER_SECONDS_TO_RUN)).until(() -> {
       jedis.set("key", "value");
       totalBytesSent.addAndGet(BYTES_SENT_PER_COMMAND);
       actual_kbs.set(Double.valueOf(getInfo(jedis).get(NETWORK_KB_READ_OVER_LAST_SECOND)));
       return true;
     });

     double expectedBytesReceived = totalBytesSent.get() / NUMBER_SECONDS_TO_RUN;
     double expected_kbs = expectedBytesReceived / 1000;

     assertThat(actual_kbs.get()).isCloseTo(expected_kbs, Offset.offset(REASONABLE_SOUNDING_OFFSET));

     // if time passes w/o operations
     await()
         .during(NUMBER_SECONDS_TO_RUN, TimeUnit.SECONDS)
         .until(() -> true);

     assertThat(Double.valueOf(getInfo(jedis).get(NETWORK_KB_READ_OVER_LAST_SECOND))).isEqualTo(0);

   }

   // todo test rejected connections
   @Test
   public void should_UpdateRejectedConnections() {
     Jedis jedis2 = new Jedis("localhost", getPort(), TIMEOUT);
     Jedis jedis3 = new Jedis("localhost", getPort(), TIMEOUT);

     jedis2.ping();
     jedis3.ping();

     validateConnectedClients(jedis, preTestConnectedClients, 2);

     jedis2.close();
     jedis3.close();

     validateConnectedClients(jedis, preTestConnectedClients, 0);
   }

   // ------------------- Clients Section -------------------------- //

   @Test
   public void connectedClients_incrAndDecrWhenClientConnectsAndDisconnects() {
     Jedis jedis2 = new Jedis("localhost", getPort(), TIMEOUT);
     jedis2.ping();

     validateConnectedClients(jedis, preTestConnectedClients, 1);

     jedis2.close();

     validateConnectedClients(jedis, preTestConnectedClients, 0);
   }

   @Test
   public void totalConnectionsReceivedStat_shouldIncrement_whenNewConnectionOccurs() {
     Jedis jedis2 = new Jedis("localhost", getPort(), TIMEOUT);
     jedis2.ping();

     validateConnectionsReceived(jedis, preTestConnectionsReceived, 1);

     jedis2.close();

     validateConnectedClients(jedis, preTestConnectedClients, 0);
   }

   // ------------------- Server Section -------------------------- //

   @Test
   public void upTimeInDays_shouldBeEqualToTimeSinceStartInDays() {
     long startTimeInNanos = getStartTime();
     long currentTimeInNanos = getCurrentTime();

     long expectedNanos = currentTimeInNanos - startTimeInNanos;
     long expectedDays = TimeUnit.NANOSECONDS.toDays(expectedNanos);

     assertThat(Long.valueOf(getInfo(jedis).get(UPTIME_IN_DAYS))).isEqualTo(expectedDays);
   }

   @Test
   public void uptimeInSeconds_shouldReturnTimeSinceStartInSeconds() {
     long serverUptimeAtStartOfTestInNanos = getCurrentTime();
     long statsUpTimeAtStartOfTest = Long.valueOf(getInfo(jedis).get(UPTIME_IN_SECONDS));

     await().during(Duration.ofSeconds(3)).until(() -> true);

     long expectedNanos = getCurrentTime() - serverUptimeAtStartOfTestInNanos;
     long expectedSeconds = TimeUnit.NANOSECONDS.toSeconds(expectedNanos);

     assertThat(Long.valueOf(getInfo(jedis).get(UPTIME_IN_SECONDS)) - statsUpTimeAtStartOfTest)
         .isCloseTo(expectedSeconds, Offset.offset(1l));
   }

   // ------------------- Helper Methods ----------------------------- //
   public long getStartTime() {
     return START_TIME;
   }

   public long getCurrentTime() {
     return this.statisticsClock.getTime();
   }

   private double getTenPercentOf(Double value) {
     return Math.ceil(value * .1);
   }

   /**
    * Convert the values returned by the INFO command into a basic param:value map.
    */
   static synchronized Map<String, String> getInfo(Jedis jedis) {
     Map<String, String> results = new HashMap<>();
     String rawInfo = jedis.info();
     numInfoCalled.incrementAndGet();

     for (String line : rawInfo.split("\r\n")) {
       int colonIndex = line.indexOf(":");
       if (colonIndex > 0) {
         String key = line.substring(0, colonIndex);
         String value = line.substring(colonIndex + 1);
         results.put(key, value);
       }
     }

     return results;
   }

   private void validateNetworkBytesRead(Jedis jedis, long initialNetworkBytesRead,
       int responseLength) {
     await().atMost(Duration.ofSeconds(2)).untilAsserted(
         () -> assertThat(Long.valueOf(getInfo(jedis).get(TOTAL_NETWORK_BYTES_READ)))
             .isEqualTo(initialNetworkBytesRead + responseLength));
   }

   private void validateCommandsProcessed(Jedis jedis, long initialCommandsProcessed, int diff) {
     await().atMost(Duration.ofSeconds(2)).untilAsserted(
         () -> assertThat(
             Long.valueOf(getInfo(jedis).get(COMMANDS_PROCESSED)) - numInfoCalled.get() + 1)
                 .isEqualTo(initialCommandsProcessed + diff));
   }

   private void validateConnectedClients(Jedis jedis, long initialConnectedClients, int diff) {
     await().atMost(Duration.ofSeconds(2)).untilAsserted(
         () -> assertThat(Long.valueOf(getInfo(jedis).get(CONNECTED_CLIENTS)))
             .isEqualTo(initialConnectedClients + diff));
   }

   private void validateConnectionsReceived(Jedis jedis, long initialConnectionsReceived, int diff) {
     assertThat(Long.valueOf(getInfo(jedis).get(TOTAL_CONNECTIONS_RECEIVED)))
         .isEqualTo(initialConnectionsReceived + diff);
   }
 }
	/*
	* 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.redis.internal.executor.server;

	import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
	import static org.assertj.core.api.Assertions.assertThat;

	import java.time.Duration;
	import java.util.HashMap;
	import java.util.Map;
	import java.util.concurrent.TimeUnit;
	import java.util.concurrent.atomic.AtomicInteger;
	import java.util.concurrent.atomic.AtomicReference;

	import com.google.common.util.concurrent.AtomicDouble;
	import org.assertj.core.data.Offset;
	import org.junit.After;
	import org.junit.Before;
	import org.junit.BeforeClass;
	import org.junit.Test;
	import redis.clients.jedis.Jedis;

	import org.apache.geode.internal.statistics.EnabledStatisticsClock;
	import org.apache.geode.internal.statistics.StatisticsClock;
	import org.apache.geode.test.awaitility.GeodeAwaitility;
	import org.apache.geode.test.dunit.rules.RedisPortSupplier;

	public abstract class AbstractRedisInfoStatsIntegrationTest implements RedisPortSupplier {

	private static final int TIMEOUT = (int) GeodeAwaitility.getTimeout().toMillis();
	private static final String EXISTING_HASH_KEY = "Existing_Hash";
	private static final String EXISTING_STRING_KEY = "Existing_String";
	private static final String EXISTING_SET_KEY_1 = "Existing_Set_1";
	private static final String EXISTING_SET_KEY_2 = "Existing_Set_2";

	private Jedis jedis;
	private static long START_TIME;
	private static StatisticsClock statisticsClock;

	private long preTestConnectionsReceived = 0;
	private long preTestConnectedClients = 0;

	private static final String COMMANDS_PROCESSED = "total_commands_processed";
	private static final String TOTAL_CONNECTIONS_RECEIVED = "total_connections_received";
	private static final String CONNECTED_CLIENTS = "connected_clients";
	private static final String OPS_PERFORMED_OVER_LAST_SECOND = "instantaneous_ops_per_sec";
	private static final String TOTAL_NETWORK_BYTES_READ = "total_net_input_bytes";
	private static final String NETWORK_KB_READ_OVER_LAST_SECOND = "instantaneous_input_kbps";
	private static final String UPTIME_IN_DAYS = "uptime_in_days";
	private static final String UPTIME_IN_SECONDS = "uptime_in_seconds";

	private static final AtomicInteger numInfoCalled = new AtomicInteger(0);


	// ------------------- Setup -------------------------- //
	@BeforeClass
	public static void beforeClass() {
	statisticsClock = new EnabledStatisticsClock();
	START_TIME = statisticsClock.getTime();
	}

	@Before
	public void before() {
	jedis = new Jedis("localhost", getPort(), TIMEOUT);
	numInfoCalled.set(0);

	long preSetupCommandsProcessed = Long.valueOf(getInfo(jedis).get(COMMANDS_PROCESSED));

	jedis.set(EXISTING_STRING_KEY, "A_Value");
	jedis.hset(EXISTING_HASH_KEY, "Field1", "Value1");
	jedis.sadd(EXISTING_SET_KEY_1, "m1", "m2", "m3");
	jedis.sadd(EXISTING_SET_KEY_2, "m4", "m5", "m6");

	// the info command increments command processed so we need to account for that.
	// the +1 is needed because info returns the number of commands processed before that call to
	// info
	await().atMost(Duration.ofSeconds(2))
	.untilAsserted(() -> assertThat(
	Long.valueOf(getInfo(jedis).get(COMMANDS_PROCESSED)) - numInfoCalled.get() + 1)
	.isEqualTo(preSetupCommandsProcessed + 4));

	preTestConnectionsReceived = Long.valueOf(getInfo(jedis).get(TOTAL_CONNECTIONS_RECEIVED));
	preTestConnectedClients = Long.valueOf(getInfo(jedis).get(CONNECTED_CLIENTS));
	numInfoCalled.set(0);
	}

	@After
	public void after() {
	jedis.flushAll();
	await().atMost(Duration.ofSeconds(5))
	.untilAsserted(
	() -> assertThat(Long.valueOf(getInfo(jedis).get(CONNECTED_CLIENTS))).isEqualTo(1));
	jedis.close();
	}

	// ------------------- Stats Section -------------------------- //

	// note: see AbstractHitsMissesIntegrationTest for testing of hits/misses

	@Test
	public void commandsProcessed_shouldIncrement_givenSuccessfulCommand() {
	long initialCommandsProcessed = Long.valueOf(getInfo(jedis).get(COMMANDS_PROCESSED));
	jedis.ttl("key");

	validateCommandsProcessed(jedis, initialCommandsProcessed, 1);
	}

	@Test
	public void opsPerformedOverLastSecond_ShouldUpdate_givenOperationsOccurring() {
	int NUMBER_SECONDS_TO_RUN = 3;

	AtomicInteger numberOfCommandsExecuted = new AtomicInteger();
	AtomicDouble actualCommandsProcessedOverLastSecond = new AtomicDouble();

	await().during(Duration.ofSeconds(NUMBER_SECONDS_TO_RUN)).until(() -> {
	jedis.set("key", "value");
	numberOfCommandsExecuted.getAndIncrement();
	actualCommandsProcessedOverLastSecond.set(
	Double.valueOf(getInfo(jedis).get(OPS_PERFORMED_OVER_LAST_SECOND)));

	return true;
	});

	long expected = (numberOfCommandsExecuted.get() + numInfoCalled.get()) / NUMBER_SECONDS_TO_RUN;

	assertThat(actualCommandsProcessedOverLastSecond.get())
	.isCloseTo(expected,
	Offset.offset(getTenPercentOf(actualCommandsProcessedOverLastSecond.get())));

	// if time passes w/o operations
	await().during(NUMBER_SECONDS_TO_RUN, TimeUnit.SECONDS).until(() -> true);

	assertThat(Double.valueOf(getInfo(jedis).get(OPS_PERFORMED_OVER_LAST_SECOND))).isEqualTo(0D);
	}

	@Test
	public void networkBytesRead_shouldIncrementBySizeOfCommandSent() {
	long initialNetworkBytesRead = Long.valueOf(getInfo(jedis).get(TOTAL_NETWORK_BYTES_READ));
	String infoCommandString = "*3\r\n$3\r\ninfo\r\n";
	String respCommandString = "*3\r\n$3\r\nset\r\n$3\r\nkey\r\n$5\r\nvalue\r\n";

	jedis.set("key", "value");

	validateNetworkBytesRead(jedis, initialNetworkBytesRead,
	respCommandString.length() + infoCommandString.length());
	}

	@Test
	public void networkKiloBytesReadOverLastSecond_shouldBeCloseToBytesReadOverLastSecond() {

	double REASONABLE_SOUNDING_OFFSET = .8;
	int NUMBER_SECONDS_TO_RUN = 2;
	String RESP_COMMAND_STRING = "*3\r\n$3\r\nset\r\n$3\r\nkey\r\n$5\r\nvalue\r\n";
	int BYTES_SENT_PER_COMMAND = RESP_COMMAND_STRING.length();
	AtomicInteger totalBytesSent = new AtomicInteger();
	AtomicReference<Double> actual_kbs = new AtomicReference<>((double) 0);

	await().during(Duration.ofSeconds(NUMBER_SECONDS_TO_RUN)).until(() -> {
	jedis.set("key", "value");
	totalBytesSent.addAndGet(BYTES_SENT_PER_COMMAND);
	actual_kbs.set(Double.valueOf(getInfo(jedis).get(NETWORK_KB_READ_OVER_LAST_SECOND)));
	return true;
	});

	double expectedBytesReceived = totalBytesSent.get() / NUMBER_SECONDS_TO_RUN;
	double expected_kbs = expectedBytesReceived / 1000;

	assertThat(actual_kbs.get()).isCloseTo(expected_kbs, Offset.offset(REASONABLE_SOUNDING_OFFSET));

	// if time passes w/o operations
	await()
	.during(NUMBER_SECONDS_TO_RUN, TimeUnit.SECONDS)
	.until(() -> true);

	assertThat(Double.valueOf(getInfo(jedis).get(NETWORK_KB_READ_OVER_LAST_SECOND))).isEqualTo(0);

	}

	// todo test rejected connections
	@Test
	public void should_UpdateRejectedConnections() {
	Jedis jedis2 = new Jedis("localhost", getPort(), TIMEOUT);
	Jedis jedis3 = new Jedis("localhost", getPort(), TIMEOUT);

	jedis2.ping();
	jedis3.ping();

	validateConnectedClients(jedis, preTestConnectedClients, 2);

	jedis2.close();
	jedis3.close();

	validateConnectedClients(jedis, preTestConnectedClients, 0);
	}

	// ------------------- Clients Section -------------------------- //

	@Test
	public void connectedClients_incrAndDecrWhenClientConnectsAndDisconnects() {
	Jedis jedis2 = new Jedis("localhost", getPort(), TIMEOUT);
	jedis2.ping();

	validateConnectedClients(jedis, preTestConnectedClients, 1);

	jedis2.close();

	validateConnectedClients(jedis, preTestConnectedClients, 0);
	}

	@Test
	public void totalConnectionsReceivedStat_shouldIncrement_whenNewConnectionOccurs() {
	Jedis jedis2 = new Jedis("localhost", getPort(), TIMEOUT);
	jedis2.ping();

	validateConnectionsReceived(jedis, preTestConnectionsReceived, 1);

	jedis2.close();

	validateConnectedClients(jedis, preTestConnectedClients, 0);
	}

	// ------------------- Server Section -------------------------- //

	@Test
	public void upTimeInDays_shouldBeEqualToTimeSinceStartInDays() {
	long startTimeInNanos = getStartTime();
	long currentTimeInNanos = getCurrentTime();

	long expectedNanos = currentTimeInNanos - startTimeInNanos;
	long expectedDays = TimeUnit.NANOSECONDS.toDays(expectedNanos);

	assertThat(Long.valueOf(getInfo(jedis).get(UPTIME_IN_DAYS))).isEqualTo(expectedDays);
	}

	@Test
	public void uptimeInSeconds_shouldReturnTimeSinceStartInSeconds() {
	long serverUptimeAtStartOfTestInNanos = getCurrentTime();
	long statsUpTimeAtStartOfTest = Long.valueOf(getInfo(jedis).get(UPTIME_IN_SECONDS));

	await().during(Duration.ofSeconds(3)).until(() -> true);

	long expectedNanos = getCurrentTime() - serverUptimeAtStartOfTestInNanos;
	long expectedSeconds = TimeUnit.NANOSECONDS.toSeconds(expectedNanos);

	assertThat(Long.valueOf(getInfo(jedis).get(UPTIME_IN_SECONDS)) - statsUpTimeAtStartOfTest)
	.isCloseTo(expectedSeconds, Offset.offset(1l));
	}

	// ------------------- Helper Methods ----------------------------- //
	public long getStartTime() {
	return START_TIME;
	}

	public long getCurrentTime() {
	return this.statisticsClock.getTime();
	}

	private double getTenPercentOf(Double value) {
	return Math.ceil(value * .1);
	}

	/**
	* Convert the values returned by the INFO command into a basic param:value map.
	*/
	static synchronized Map<String, String> getInfo(Jedis jedis) {
	Map<String, String> results = new HashMap<>();
	String rawInfo = jedis.info();
	numInfoCalled.incrementAndGet();

	for (String line : rawInfo.split("\r\n")) {
	int colonIndex = line.indexOf(":");
	if (colonIndex > 0) {
	String key = line.substring(0, colonIndex);
	String value = line.substring(colonIndex + 1);
	results.put(key, value);
	}
	}

	return results;
	}

	private void validateNetworkBytesRead(Jedis jedis, long initialNetworkBytesRead,
	int responseLength) {
	await().atMost(Duration.ofSeconds(2)).untilAsserted(
	() -> assertThat(Long.valueOf(getInfo(jedis).get(TOTAL_NETWORK_BYTES_READ)))
	.isEqualTo(initialNetworkBytesRead + responseLength));
	}

	private void validateCommandsProcessed(Jedis jedis, long initialCommandsProcessed, int diff) {
	await().atMost(Duration.ofSeconds(2)).untilAsserted(
	() -> assertThat(
	Long.valueOf(getInfo(jedis).get(COMMANDS_PROCESSED)) - numInfoCalled.get() + 1)
	.isEqualTo(initialCommandsProcessed + diff));
	}

	private void validateConnectedClients(Jedis jedis, long initialConnectedClients, int diff) {
	await().atMost(Duration.ofSeconds(2)).untilAsserted(
	() -> assertThat(Long.valueOf(getInfo(jedis).get(CONNECTED_CLIENTS)))
	.isEqualTo(initialConnectedClients + diff));
	}

	private void validateConnectionsReceived(Jedis jedis, long initialConnectionsReceived, int diff) {
	assertThat(Long.valueOf(getInfo(jedis).get(TOTAL_CONNECTIONS_RECEIVED)))
	.isEqualTo(initialConnectionsReceived + diff);
	}
	}