pagespeed/system/redis_cache_cluster_setup.cc - incubator-pagespeed-mod - Git at Google

 /*
  * Copyright 2016 Google Inc.
  *
  * Licensed 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.
  */

 // Author: yeputons@google.com (Egor Suvorov)

 // Unit-test the redis interface in conjunction with Redis Cluster

 #include "pagespeed/system/redis_cache_cluster_setup.h"

 #include <cstddef>
 #include <cstdlib>
 #include <algorithm>

 #include "base/logging.h"
 #include "strings/stringpiece_utils.h"
 #include "pagespeed/kernel/base/basictypes.h"
 #include "pagespeed/kernel/base/posix_timer.h"
 #include "pagespeed/kernel/base/string.h"
 #include "pagespeed/kernel/base/string_util.h"
 #include "pagespeed/system/tcp_connection_for_testing.h"

 namespace net_instaweb {

 namespace RedisCluster {

 namespace {

 static const int kReconfigurationPropagationTimeoutMs = 10000;

 GoogleString ReadBulkString(TcpConnectionForTesting* conn) {
   GoogleString length_str_storage = conn->ReadLineCrLf();
   StringPiece length_str = length_str_storage;
   // Check that Redis answered with Bulk String
   CHECK(strings::StartsWith(length_str, "$"));
   length_str.remove_prefix(1);
   CHECK(strings::EndsWith(length_str, "\r\n"));
   length_str.remove_suffix(2);
   int length;
   CHECK(StringToInt(length_str, &length));
   GoogleString result =  conn->ReadBytes(length);
   CHECK_EQ("\r\n", conn->ReadLineCrLf());
   return result;
 }

 }  // namespace

 // TODO(cheesy): Instead of CLUSTER NODES, CLUSTER SLOTS provides the same
 // information in a machine readable format that is already parsed in
 // RedisCache::FetchClusterSlotMapping. The CLUSTER NODES part of this code
 // could be replaced with a call to the innards of FetchClusterSlotMapping.
 // redisReaderCreate and redisReaderGetReply would likely be needed to turn the
 // ReadBulkString result into a redisReply. See:
 // https://github.com/redis/hiredis/issues/59
 StringVector GetNodeConfig(TcpConnectionForTesting* conn) {
   conn->Send("CLUSTER INFO\r\n");
   GoogleString cluster_info = ReadBulkString(conn);
   if (cluster_info.find("cluster_state:ok\r\n") == GoogleString::npos) {
     return {};
   }

   conn->Send("CLUSTER NODES\r\n");
   GoogleString config_csv = ReadBulkString(conn);
   StringPieceVector lines;
   SplitStringPieceToVector(config_csv, "\r\n", &lines,
                             true /* omit_empty_strings */);
   StringVector current_config;
   for (StringPiece line : lines) {
     StringPieceVector fields;
     SplitStringPieceToVector(line, " ", &fields,
                               true /* omit_empty_strings */);
     CHECK_GE(fields.size(), 8);
     GoogleString node_descr;
     // See http://redis.io/commands/cluster-nodes. We take three fields
     // from node description (node id, ip:port, master/slave) plus
     // information about slots served.
     StrAppend(&node_descr, fields[0], " ", fields[1], " ", fields[3]);
     for (auto it = fields.begin() + 8; it != fields.end(); ++it) {
       StrAppend(&node_descr, " ", *it);
     }
     current_config.push_back(node_descr);
   }
   std::sort(current_config.begin(), current_config.end());
   return current_config;
 }

 void ResetConfiguration(StringVector* node_ids,
                         std::vector<int>* ports,
                         ConnectionList* connections) {
   // TODO(cheesy): These should be collapsed onto a single vector of
   // struct { conn, port, node_id }.
   CHECK_EQ(6, connections->size());
   CHECK_EQ(connections->size(), ports->size());
   CHECK_EQ(connections->size(), node_ids->size());

   LOG(INFO) << "Resetting Redis Cluster configuration back to default";

   // Flush the nodes which is required to re-configure the cluster.
   FlushAll(connections);

   // Reset all nodes.
   for (auto& conn : *connections) {
     conn->Send("CLUSTER RESET SOFT\r\n");
   }
   for (auto& conn : *connections) {
     CHECK_EQ("+OK\r\n", conn->ReadLineCrLf());
   }

   // Now make nodes know about each other.
   for (auto& conn : *connections) {
     for (int port : *ports) {
       conn->Send(
           StrCat("CLUSTER MEET 127.0.0.1 ", IntegerToString(port), "\r\n"));
     }
     for (int i = 0, n = ports->size(); i < n; ++i) {
       CHECK_EQ("+OK\r\n", conn->ReadLineCrLf());
     }
   }

   // And finally load slots configuration.
   // Some of these boundaries are explicitly probed in the SlotBoundaries
   // test. If you change the cluster layout, you must also change that test.
   static const int slot_ranges[] = { 0, 5500, 11000, 16384 };
   for (int i = 0; i < 3; i++) {
     GoogleString command = "CLUSTER ADDSLOTS";
     for (int slot = slot_ranges[i]; slot < slot_ranges[i + 1]; slot++) {
       StrAppend(&command, " ", IntegerToString(slot));
     }
     StrAppend(&command, "\r\n");

     auto& conn = (*connections)[i];
     conn->Send(command);
     CHECK_EQ("+OK\r\n", conn->ReadLineCrLf());
   }

   // Nodes learn about each other asynchronously in response to CLUSTER MEET
   // above, but if the system hasn't yet converged, REPLICATE will fail. We
   // poll the cluster config with GetNodeClusterConfig() until every node
   // knows about every other node.

   LOG(INFO) << "Reset Redis Cluster configuration back to default, "
                 "waiting for node propagation...";

   PosixTimer timer;
   int64 timeout_at_ms = timer.NowMs() + kReconfigurationPropagationTimeoutMs;

   bool propagated = false;
   while (!propagated && timer.NowMs() < timeout_at_ms) {
     size_t num_complete = 0;
     // For every connection, pull the node config and verify that it sees
     // the right number of nodes.
     for (auto& conn : *connections) {
       StringVector config = GetNodeConfig(conn.get());
       if (config.size() == connections->size()) {
         ++num_complete;
       } else {
         break;
       }
     }
     if (num_complete == connections->size()) {
       propagated = true;
     } else {
       timer.SleepMs(50);
     }
   }

   CHECK(propagated) << "All nodes did not report in after CLUSTER MEET";

   for (int i = 3; i < 6; i++) {
     auto& conn = (*connections)[i];
     conn->Send(StrCat("CLUSTER REPLICATE ", (*node_ids)[i - 3], "\r\n"));
     CHECK_EQ("+OK\r\n", conn->ReadLineCrLf());
   }

   // Now wait until all nodes report cluster as healthy and report same
   // cluster configuration.
   LOG(INFO) << "Reset Redis Cluster configuration back to default, "
                 "waiting for slot propagation...";
   timeout_at_ms = timer.NowMs() + kReconfigurationPropagationTimeoutMs;
   bool cluster_is_up = false;
   while (!cluster_is_up) {
     CHECK_LE(timer.NowMs(), timeout_at_ms)
         << "Redis Cluster configuration did not propagate in time";

     StringVector first_node_config;
     for (auto& conn : *connections) {
       StringVector current_config = GetNodeConfig(conn.get());
       if (current_config.empty() ||
           current_config.size() != connections->size()) {
         break;
       }
       // Check configs are the same on all nodes.
       cluster_is_up = true;
       if (first_node_config.empty()) {
         first_node_config = current_config;
       } else if (first_node_config != current_config) {
         cluster_is_up = false;
         break;
       }
     }
     if (!cluster_is_up) {
       timer.SleepMs(50);
     }
   }
   LOG(INFO) << "Redis Cluster is reset";
 }

 bool LoadConfiguration(StringVector* node_ids,
                        std::vector<int>* ports,
                         ConnectionList* connections) {
   // Parsing environment variables.
   // TODO(cheesy): We should discover the cluster IDs by querying the ports,
   // and not rely on the shell to set REDIS_CLUSTER_IDS for us.
   const char* ports_env = getenv("REDIS_CLUSTER_PORTS");
   const char* ids_env = getenv("REDIS_CLUSTER_IDS");
   if (!ports_env && !ids_env) {
     LOG(ERROR) << "Env variables REDIS_CLUSTER_* are not set. Use "
                 << "install/run_program_with_redis_cluster.sh for running "
                 << "these tests. Do not use real cluster; ALL DATA WILL "
                 << "BE ERASED DURING TESTS!";
     return false;
   }
   CHECK(ports_env) << "Env variable REDIS_CLUSTER_PORTS is unspecified";
   CHECK(ids_env) << "Env variable REDIS_CLUSTER_IDS is unspecified";

   StringPieceVector port_strs;
   StringPieceVector id_strs;
   SplitStringPieceToVector(ports_env, " ", &port_strs,
                             /* omit_empty_strings */ true);
   SplitStringPieceToVector(ids_env, " ", &id_strs,
                             /* omit_empty_strings */ true);
   CHECK_EQ(port_strs.size(), id_strs.size()) << "REDIS_CLUSTER_PORTS and "
                                                 "REDIS_CLUSTER_IDS have "
                                                 "different amount of items";
   CHECK_EQ(port_strs.size(), 6) << "Six Redis Cluster nodes are expected";

   for (auto port_str : port_strs) {
     int port;
     CHECK(StringToInt(port_str, &port)) << "Invalid port: " << port_str;
     ports->push_back(port);
   }
   for (StringPiece id : id_strs) {
     node_ids->push_back(id.as_string());
   }

   for (int port : *ports) {
     connections->emplace_back(new TcpConnectionForTesting());
     CHECK(connections->back()->Connect("localhost", port))
         << "Cannot connect to Redis Cluster node";
   }
   return true;
 }

 void FlushAll(TcpConnectionForTesting* conn) {
   conn->Send("FLUSHALL\r\n");
   GoogleString flushall_reply = conn->ReadLineCrLf();
   // We'll get READONLY from slave nodes, which isn't a problem.
   CHECK(flushall_reply == "+OK\r\n" ||
         strings::StartsWith(flushall_reply, "-READONLY"));
 }

 void FlushAll(ConnectionList* connections) {
   for (auto& conn : *connections) {
     FlushAll(conn.get());
   }
 }

 }  // namespace RedisCluster

 }  // namespace net_instaweb
	/*
	* Copyright 2016 Google Inc.
	*
	* Licensed 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.
	*/

	// Author: yeputons@google.com (Egor Suvorov)

	// Unit-test the redis interface in conjunction with Redis Cluster

	#include "pagespeed/system/redis_cache_cluster_setup.h"

	#include <cstddef>
	#include <cstdlib>
	#include <algorithm>

	#include "base/logging.h"
	#include "strings/stringpiece_utils.h"
	#include "pagespeed/kernel/base/basictypes.h"
	#include "pagespeed/kernel/base/posix_timer.h"
	#include "pagespeed/kernel/base/string.h"
	#include "pagespeed/kernel/base/string_util.h"
	#include "pagespeed/system/tcp_connection_for_testing.h"

	namespace net_instaweb {

	namespace RedisCluster {

	namespace {

	static const int kReconfigurationPropagationTimeoutMs = 10000;

	GoogleString ReadBulkString(TcpConnectionForTesting* conn) {
	GoogleString length_str_storage = conn->ReadLineCrLf();
	StringPiece length_str = length_str_storage;
	// Check that Redis answered with Bulk String
	CHECK(strings::StartsWith(length_str, "$"));
	length_str.remove_prefix(1);
	CHECK(strings::EndsWith(length_str, "\r\n"));
	length_str.remove_suffix(2);
	int length;
	CHECK(StringToInt(length_str, &length));
	GoogleString result = conn->ReadBytes(length);
	CHECK_EQ("\r\n", conn->ReadLineCrLf());
	return result;
	}

	} // namespace

	// TODO(cheesy): Instead of CLUSTER NODES, CLUSTER SLOTS provides the same
	// information in a machine readable format that is already parsed in
	// RedisCache::FetchClusterSlotMapping. The CLUSTER NODES part of this code
	// could be replaced with a call to the innards of FetchClusterSlotMapping.
	// redisReaderCreate and redisReaderGetReply would likely be needed to turn the
	// ReadBulkString result into a redisReply. See:
	// https://github.com/redis/hiredis/issues/59
	StringVector GetNodeConfig(TcpConnectionForTesting* conn) {
	conn->Send("CLUSTER INFO\r\n");
	GoogleString cluster_info = ReadBulkString(conn);
	if (cluster_info.find("cluster_state:ok\r\n") == GoogleString::npos) {
	return {};
	}

	conn->Send("CLUSTER NODES\r\n");
	GoogleString config_csv = ReadBulkString(conn);
	StringPieceVector lines;
	SplitStringPieceToVector(config_csv, "\r\n", &lines,
	true /* omit_empty_strings */);
	StringVector current_config;
	for (StringPiece line : lines) {
	StringPieceVector fields;
	SplitStringPieceToVector(line, " ", &fields,
	true /* omit_empty_strings */);
	CHECK_GE(fields.size(), 8);
	GoogleString node_descr;
	// See http://redis.io/commands/cluster-nodes. We take three fields
	// from node description (node id, ip:port, master/slave) plus
	// information about slots served.
	StrAppend(&node_descr, fields[0], " ", fields[1], " ", fields[3]);
	for (auto it = fields.begin() + 8; it != fields.end(); ++it) {
	StrAppend(&node_descr, " ", *it);
	}
	current_config.push_back(node_descr);
	}
	std::sort(current_config.begin(), current_config.end());
	return current_config;
	}

	void ResetConfiguration(StringVector* node_ids,
	std::vector<int>* ports,
	ConnectionList* connections) {
	// TODO(cheesy): These should be collapsed onto a single vector of
	// struct { conn, port, node_id }.
	CHECK_EQ(6, connections->size());
	CHECK_EQ(connections->size(), ports->size());
	CHECK_EQ(connections->size(), node_ids->size());

	LOG(INFO) << "Resetting Redis Cluster configuration back to default";

	// Flush the nodes which is required to re-configure the cluster.
	FlushAll(connections);

	// Reset all nodes.
	for (auto& conn : *connections) {
	conn->Send("CLUSTER RESET SOFT\r\n");
	}
	for (auto& conn : *connections) {
	CHECK_EQ("+OK\r\n", conn->ReadLineCrLf());
	}

	// Now make nodes know about each other.
	for (auto& conn : *connections) {
	for (int port : *ports) {
	conn->Send(
	StrCat("CLUSTER MEET 127.0.0.1 ", IntegerToString(port), "\r\n"));
	}
	for (int i = 0, n = ports->size(); i < n; ++i) {
	CHECK_EQ("+OK\r\n", conn->ReadLineCrLf());
	}
	}

	// And finally load slots configuration.
	// Some of these boundaries are explicitly probed in the SlotBoundaries
	// test. If you change the cluster layout, you must also change that test.
	static const int slot_ranges[] = { 0, 5500, 11000, 16384 };
	for (int i = 0; i < 3; i++) {
	GoogleString command = "CLUSTER ADDSLOTS";
	for (int slot = slot_ranges[i]; slot < slot_ranges[i + 1]; slot++) {
	StrAppend(&command, " ", IntegerToString(slot));
	}
	StrAppend(&command, "\r\n");

	auto& conn = (*connections)[i];
	conn->Send(command);
	CHECK_EQ("+OK\r\n", conn->ReadLineCrLf());
	}

	// Nodes learn about each other asynchronously in response to CLUSTER MEET
	// above, but if the system hasn't yet converged, REPLICATE will fail. We
	// poll the cluster config with GetNodeClusterConfig() until every node
	// knows about every other node.

	LOG(INFO) << "Reset Redis Cluster configuration back to default, "
	"waiting for node propagation...";

	PosixTimer timer;
	int64 timeout_at_ms = timer.NowMs() + kReconfigurationPropagationTimeoutMs;

	bool propagated = false;
	while (!propagated && timer.NowMs() < timeout_at_ms) {
	size_t num_complete = 0;
	// For every connection, pull the node config and verify that it sees
	// the right number of nodes.
	for (auto& conn : *connections) {
	StringVector config = GetNodeConfig(conn.get());
	if (config.size() == connections->size()) {
	++num_complete;
	} else {
	break;
	}
	}
	if (num_complete == connections->size()) {
	propagated = true;
	} else {
	timer.SleepMs(50);
	}
	}

	CHECK(propagated) << "All nodes did not report in after CLUSTER MEET";

	for (int i = 3; i < 6; i++) {
	auto& conn = (*connections)[i];
	conn->Send(StrCat("CLUSTER REPLICATE ", (*node_ids)[i - 3], "\r\n"));
	CHECK_EQ("+OK\r\n", conn->ReadLineCrLf());
	}

	// Now wait until all nodes report cluster as healthy and report same
	// cluster configuration.
	LOG(INFO) << "Reset Redis Cluster configuration back to default, "
	"waiting for slot propagation...";
	timeout_at_ms = timer.NowMs() + kReconfigurationPropagationTimeoutMs;
	bool cluster_is_up = false;
	while (!cluster_is_up) {
	CHECK_LE(timer.NowMs(), timeout_at_ms)
	<< "Redis Cluster configuration did not propagate in time";

	StringVector first_node_config;
	for (auto& conn : *connections) {
	StringVector current_config = GetNodeConfig(conn.get());
	if (current_config.empty() \|\|
	current_config.size() != connections->size()) {
	break;
	}
	// Check configs are the same on all nodes.
	cluster_is_up = true;
	if (first_node_config.empty()) {
	first_node_config = current_config;
	} else if (first_node_config != current_config) {
	cluster_is_up = false;
	break;
	}
	}
	if (!cluster_is_up) {
	timer.SleepMs(50);
	}
	}
	LOG(INFO) << "Redis Cluster is reset";
	}

	bool LoadConfiguration(StringVector* node_ids,
	std::vector<int>* ports,
	ConnectionList* connections) {
	// Parsing environment variables.
	// TODO(cheesy): We should discover the cluster IDs by querying the ports,
	// and not rely on the shell to set REDIS_CLUSTER_IDS for us.
	const char* ports_env = getenv("REDIS_CLUSTER_PORTS");
	const char* ids_env = getenv("REDIS_CLUSTER_IDS");
	if (!ports_env && !ids_env) {
	LOG(ERROR) << "Env variables REDIS_CLUSTER_* are not set. Use "
	<< "install/run_program_with_redis_cluster.sh for running "
	<< "these tests. Do not use real cluster; ALL DATA WILL "
	<< "BE ERASED DURING TESTS!";
	return false;
	}
	CHECK(ports_env) << "Env variable REDIS_CLUSTER_PORTS is unspecified";
	CHECK(ids_env) << "Env variable REDIS_CLUSTER_IDS is unspecified";

	StringPieceVector port_strs;
	StringPieceVector id_strs;
	SplitStringPieceToVector(ports_env, " ", &port_strs,
	/* omit_empty_strings */ true);
	SplitStringPieceToVector(ids_env, " ", &id_strs,
	/* omit_empty_strings */ true);
	CHECK_EQ(port_strs.size(), id_strs.size()) << "REDIS_CLUSTER_PORTS and "
	"REDIS_CLUSTER_IDS have "
	"different amount of items";
	CHECK_EQ(port_strs.size(), 6) << "Six Redis Cluster nodes are expected";

	for (auto port_str : port_strs) {
	int port;
	CHECK(StringToInt(port_str, &port)) << "Invalid port: " << port_str;
	ports->push_back(port);
	}
	for (StringPiece id : id_strs) {
	node_ids->push_back(id.as_string());
	}

	for (int port : *ports) {
	connections->emplace_back(new TcpConnectionForTesting());
	CHECK(connections->back()->Connect("localhost", port))
	<< "Cannot connect to Redis Cluster node";
	}
	return true;
	}

	void FlushAll(TcpConnectionForTesting* conn) {
	conn->Send("FLUSHALL\r\n");
	GoogleString flushall_reply = conn->ReadLineCrLf();
	// We'll get READONLY from slave nodes, which isn't a problem.
	CHECK(flushall_reply == "+OK\r\n" \|\|
	strings::StartsWith(flushall_reply, "-READONLY"));
	}

	void FlushAll(ConnectionList* connections) {
	for (auto& conn : *connections) {
	FlushAll(conn.get());
	}
	}

	} // namespace RedisCluster

	} // namespace net_instaweb