src/cluster_connector.cpp - cassandra-cpp-driver - 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.
 */

 #include "cluster_connector.hpp"
 #include "dc_aware_policy.hpp"
 #include "protocol.hpp"
 #include "random.hpp"
 #include "round_robin_policy.hpp"

 using namespace datastax;
 using namespace datastax::internal;
 using namespace datastax::internal::core;

 namespace datastax { namespace internal { namespace core {

 /**
  * A task for running the connection process.
  */
 class RunResolveAndConnectCluster : public Task {
 public:
   RunResolveAndConnectCluster(const ClusterConnector::Ptr& connector)
       : connector_(connector) {}

   void run(EventLoop* event_loop) { connector_->internal_resolve_and_connect(); }

 private:
   ClusterConnector::Ptr connector_;
 };

 /**
  * A task for canceling the connection process.
  */
 class RunCancelCluster : public Task {
 public:
   RunCancelCluster(const ClusterConnector::Ptr& connector)
       : connector_(connector) {}

   void run(EventLoop* event_loop) { connector_->internal_cancel(); }

 private:
   ClusterConnector::Ptr connector_;
 };

 }}} // namespace datastax::internal::core

 ClusterConnector::ClusterConnector(const AddressVec& contact_points,
                                    ProtocolVersion protocol_version, const Callback& callback)
     : remaining_connector_count_(0)
     , contact_points_(contact_points)
     , protocol_version_(protocol_version)
     , listener_(NULL)
     , event_loop_(NULL)
     , random_(NULL)
     , metrics_(NULL)
     , callback_(callback)
     , error_code_(CLUSTER_OK)
     , ssl_error_code_(CASS_OK) {}

 ClusterConnector* ClusterConnector::with_listener(ClusterListener* listener) {
   listener_ = listener;
   return this;
 }

 ClusterConnector* ClusterConnector::with_random(Random* random) {
   random_ = random;
   return this;
 }

 ClusterConnector* ClusterConnector::with_metrics(Metrics* metrics) {
   metrics_ = metrics;
   return this;
 }

 void ClusterConnector::connect(EventLoop* event_loop) {
   event_loop_ = event_loop;
   event_loop_->add(new RunResolveAndConnectCluster(Ptr(this)));
 }

 void ClusterConnector::cancel() {
   if (event_loop_) {
     event_loop_->add(new RunCancelCluster(Ptr(this)));
   }
 }

 Cluster::Ptr ClusterConnector::release_cluster() {
   Cluster::Ptr temp(cluster_);
   cluster_.reset();
   return temp;
 }

 void ClusterConnector::internal_resolve_and_connect() {
   inc_ref();

   if (random_ && !contact_points_.empty()) {
     random_shuffle(contact_points_.begin(), contact_points_.end(), random_);
   }

   resolver_ = settings_.cluster_metadata_resolver_factory->new_instance(settings_);

   resolver_->resolve(event_loop_->loop(), contact_points_,
                      bind_callback(&ClusterConnector::on_resolve, this));
 }

 void ClusterConnector::internal_connect(const Address& address, ProtocolVersion version) {
   Host::Ptr host(new Host(address));
   ControlConnector::Ptr connector(
       new ControlConnector(host, version, bind_callback(&ClusterConnector::on_connect, this)));
   connectors_[address] = connector; // Keep track of the connectors so they can be canceled.
   connector->with_metrics(metrics_)
       ->with_settings(settings_.control_connection_settings)
       ->connect(event_loop_->loop());
 }

 void ClusterConnector::internal_cancel() {
   error_code_ = CLUSTER_CANCELED;
   if (resolver_) resolver_->cancel();
   for (ConnectorMap::iterator it = connectors_.begin(), end = connectors_.end(); it != end; ++it) {
     it->second->cancel();
   }
   if (cluster_) cluster_->close();
 }

 void ClusterConnector::finish() {
   callback_(this);
   if (cluster_) {
     // If the callback doesn't take possession of the cluster then we should
     // also clear the listener.
     cluster_->set_listener();
     cluster_->close();
   }
   // Explicitly release resources on the event loop thread.
   resolver_.reset();
   connectors_.clear();
   cluster_.reset();
   dec_ref();
 }

 void ClusterConnector::maybe_finish() {
   if (remaining_connector_count_ > 0 && --remaining_connector_count_ == 0) {
     finish();
   }
 }

 void ClusterConnector::on_error(ClusterConnector::ClusterError code, const String& message) {
   assert(code != CLUSTER_OK && "Notified error without an error");
   error_message_ = message;
   error_code_ = code;
   maybe_finish();
 }

 void ClusterConnector::on_resolve(ClusterMetadataResolver* resolver) {
   if (is_canceled()) {
     finish();
     return;
   }

   const AddressVec& resolved_contact_points(resolver->resolved_contact_points());

   if (resolved_contact_points.empty()) {
     error_code_ = CLUSTER_ERROR_NO_HOSTS_AVAILABLE;
     error_message_ = "Unable to connect to any contact points";
     finish();
     return;
   }

   local_dc_ = resolver->local_dc();
   remaining_connector_count_ = resolved_contact_points.size();
   for (AddressVec::const_iterator it = resolved_contact_points.begin(),
                                   end = resolved_contact_points.end();
        it != end; ++it) {
     internal_connect(*it, protocol_version_);
   }
 }

 void ClusterConnector::on_connect(ControlConnector* connector) {
   // Ignore logging protocol errors here because they're handled below
   if (!connector->is_ok() && !connector->is_canceled() && !connector->is_invalid_protocol()) {
     LOG_ERROR(
         "Unable to establish a control connection to host %s because of the following error: %s",
         connector->address().to_string().c_str(), connector->error_message().c_str());
   }

   // If the cluster object is successfully initialized or if the connector is
   // canceled then attempt to finish the connection process.
   if (cluster_ || is_canceled()) {
     maybe_finish();
     return;
   }

   // Otherwise, initialize the cluster and handle errors.
   if (connector->is_ok()) {
     const HostMap& hosts(connector->hosts());
     LoadBalancingPolicy::Vec policies;

     HostMap::const_iterator host_it = hosts.find(connector->address());
     if (host_it == hosts.end()) {
       // This error is unlikely to happen and it likely means that the local
       // metadata is corrupt or missing and the control connection process
       // would've probably failed before this happens.
       LOG_ERROR("Current control connection host %s not found in hosts metadata",
                 connector->address().to_string().c_str());
       on_error(CLUSTER_ERROR_NO_HOSTS_AVAILABLE,
                "Control connection host is not found in hosts metadata");
       return;
     }
     Host::Ptr connected_host(host_it->second);

     // Build policies list including the default policy
     LoadBalancingPolicy::Ptr default_policy(settings_.load_balancing_policy->new_instance());
     policies.push_back(default_policy);
     for (LoadBalancingPolicy::Vec::const_iterator it = settings_.load_balancing_policies.begin(),
                                                   end = settings_.load_balancing_policies.end();
          it != end; ++it) {
       policies.push_back(LoadBalancingPolicy::Ptr((*it)->new_instance()));
     }

     // Initialize all created policies
     for (LoadBalancingPolicy::Vec::const_iterator it = policies.begin(), end = policies.end();
          it != end; ++it) {
       LoadBalancingPolicy::Ptr policy(*it);
       policy->init(connected_host, hosts, random_, local_dc_);
       policy->register_handles(event_loop_->loop());
     }

     ScopedPtr<QueryPlan> query_plan(default_policy->new_query_plan("", NULL, NULL));
     if (!query_plan->compute_next()) { // No hosts in the query plan
       LOG_ERROR("Current control connection host %s has no hosts available in "
                 "it's query plan for the configured load balancing policy. If "
                 "using DC-aware check to see if the local datacenter is valid.",
                 connector->address().to_string().c_str());

       const char* message;
       if (dynamic_cast<DCAwarePolicy::DCAwareQueryPlan*>(query_plan.get()) !=
           NULL) { // Check if DC-aware
         message = "No hosts available for the control connection using the "
                   "DC-aware load balancing policy. "
                   "Check to see if the configured local datacenter is valid";
       } else {
         message = "No hosts available for the control connection using the "
                   "configured load balancing policy";
       }
       on_error(CLUSTER_ERROR_NO_HOSTS_AVAILABLE, message);
       return;
     }

     cluster_.reset(new Cluster(connector->release_connection(), listener_, event_loop_,
                                connected_host, hosts, connector->schema(), default_policy, policies,
                                local_dc_, connector->supported_options(), settings_));

     // Clear any connection errors and set the final negotiated protocol version.
     error_code_ = CLUSTER_OK;
     error_message_.clear();
     protocol_version_ = connector->protocol_version();

     // The cluster is initialized so the rest of the connectors can be canceled.
     for (ConnectorMap::iterator it = connectors_.begin(), end = connectors_.end(); it != end;
          ++it) {
       if (it->first != connector->address()) { // Not the current connector.
         it->second->cancel();
       }
     }

     maybe_finish();
   } else if (connector->is_invalid_protocol()) {
     ProtocolVersion lower_version(connector->protocol_version().previous());
     if (!lower_version.is_valid()) {
       LOG_ERROR(
           "Host %s does not support any valid protocol version (lowest supported version is %s)",
           connector->address().to_string().c_str(),
           ProtocolVersion::lowest_supported().to_string().c_str());
       on_error(CLUSTER_ERROR_INVALID_PROTOCOL, "Unable to find supported protocol version");
       return;
     }
     LOG_INFO("Host %s does not support protocol version %s. "
              "Trying protocol version %s...",
              connector->address().to_string().c_str(),
              connector->protocol_version().to_string().c_str(), lower_version.to_string().c_str());
     internal_connect(connector->address(), lower_version);
   } else if (connector->is_ssl_error()) {
     ssl_error_code_ = connector->ssl_error_code();
     on_error(CLUSTER_ERROR_SSL_ERROR, connector->error_message());
   } else if (connector->is_auth_error()) {
     on_error(CLUSTER_ERROR_AUTH_ERROR, connector->error_message());
   } else {
     assert(!connector->is_canceled() &&
            "The control connector should have an error and not be canceled");
     on_error(CLUSTER_ERROR_NO_HOSTS_AVAILABLE, connector->error_message());
   }
 }
	/*
	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.
	*/

	#include "cluster_connector.hpp"
	#include "dc_aware_policy.hpp"
	#include "protocol.hpp"
	#include "random.hpp"
	#include "round_robin_policy.hpp"

	using namespace datastax;
	using namespace datastax::internal;
	using namespace datastax::internal::core;

	namespace datastax { namespace internal { namespace core {

	/**
	* A task for running the connection process.
	*/
	class RunResolveAndConnectCluster : public Task {
	public:
	RunResolveAndConnectCluster(const ClusterConnector::Ptr& connector)
	: connector_(connector) {}

	void run(EventLoop* event_loop) { connector_->internal_resolve_and_connect(); }

	private:
	ClusterConnector::Ptr connector_;
	};

	/**
	* A task for canceling the connection process.
	*/
	class RunCancelCluster : public Task {
	public:
	RunCancelCluster(const ClusterConnector::Ptr& connector)
	: connector_(connector) {}

	void run(EventLoop* event_loop) { connector_->internal_cancel(); }

	private:
	ClusterConnector::Ptr connector_;
	};

	}}} // namespace datastax::internal::core

	ClusterConnector::ClusterConnector(const AddressVec& contact_points,
	ProtocolVersion protocol_version, const Callback& callback)
	: remaining_connector_count_(0)
	, contact_points_(contact_points)
	, protocol_version_(protocol_version)
	, listener_(NULL)
	, event_loop_(NULL)
	, random_(NULL)
	, metrics_(NULL)
	, callback_(callback)
	, error_code_(CLUSTER_OK)
	, ssl_error_code_(CASS_OK) {}

	ClusterConnector* ClusterConnector::with_listener(ClusterListener* listener) {
	listener_ = listener;
	return this;
	}

	ClusterConnector* ClusterConnector::with_random(Random* random) {
	random_ = random;
	return this;
	}

	ClusterConnector* ClusterConnector::with_metrics(Metrics* metrics) {
	metrics_ = metrics;
	return this;
	}

	void ClusterConnector::connect(EventLoop* event_loop) {
	event_loop_ = event_loop;
	event_loop_->add(new RunResolveAndConnectCluster(Ptr(this)));
	}

	void ClusterConnector::cancel() {
	if (event_loop_) {
	event_loop_->add(new RunCancelCluster(Ptr(this)));
	}
	}

	Cluster::Ptr ClusterConnector::release_cluster() {
	Cluster::Ptr temp(cluster_);
	cluster_.reset();
	return temp;
	}

	void ClusterConnector::internal_resolve_and_connect() {
	inc_ref();

	if (random_ && !contact_points_.empty()) {
	random_shuffle(contact_points_.begin(), contact_points_.end(), random_);
	}

	resolver_ = settings_.cluster_metadata_resolver_factory->new_instance(settings_);

	resolver_->resolve(event_loop_->loop(), contact_points_,
	bind_callback(&ClusterConnector::on_resolve, this));
	}

	void ClusterConnector::internal_connect(const Address& address, ProtocolVersion version) {
	Host::Ptr host(new Host(address));
	ControlConnector::Ptr connector(
	new ControlConnector(host, version, bind_callback(&ClusterConnector::on_connect, this)));
	connectors_[address] = connector; // Keep track of the connectors so they can be canceled.
	connector->with_metrics(metrics_)
	->with_settings(settings_.control_connection_settings)
	->connect(event_loop_->loop());
	}

	void ClusterConnector::internal_cancel() {
	error_code_ = CLUSTER_CANCELED;
	if (resolver_) resolver_->cancel();
	for (ConnectorMap::iterator it = connectors_.begin(), end = connectors_.end(); it != end; ++it) {
	it->second->cancel();
	}
	if (cluster_) cluster_->close();
	}

	void ClusterConnector::finish() {
	callback_(this);
	if (cluster_) {
	// If the callback doesn't take possession of the cluster then we should
	// also clear the listener.
	cluster_->set_listener();
	cluster_->close();
	}
	// Explicitly release resources on the event loop thread.
	resolver_.reset();
	connectors_.clear();
	cluster_.reset();
	dec_ref();
	}

	void ClusterConnector::maybe_finish() {
	if (remaining_connector_count_ > 0 && --remaining_connector_count_ == 0) {
	finish();
	}
	}

	void ClusterConnector::on_error(ClusterConnector::ClusterError code, const String& message) {
	assert(code != CLUSTER_OK && "Notified error without an error");
	error_message_ = message;
	error_code_ = code;
	maybe_finish();
	}

	void ClusterConnector::on_resolve(ClusterMetadataResolver* resolver) {
	if (is_canceled()) {
	finish();
	return;
	}

	const AddressVec& resolved_contact_points(resolver->resolved_contact_points());

	if (resolved_contact_points.empty()) {
	error_code_ = CLUSTER_ERROR_NO_HOSTS_AVAILABLE;
	error_message_ = "Unable to connect to any contact points";
	finish();
	return;
	}

	local_dc_ = resolver->local_dc();
	remaining_connector_count_ = resolved_contact_points.size();
	for (AddressVec::const_iterator it = resolved_contact_points.begin(),
	end = resolved_contact_points.end();
	it != end; ++it) {
	internal_connect(*it, protocol_version_);
	}
	}

	void ClusterConnector::on_connect(ControlConnector* connector) {
	// Ignore logging protocol errors here because they're handled below
	if (!connector->is_ok() && !connector->is_canceled() && !connector->is_invalid_protocol()) {
	LOG_ERROR(
	"Unable to establish a control connection to host %s because of the following error: %s",
	connector->address().to_string().c_str(), connector->error_message().c_str());
	}

	// If the cluster object is successfully initialized or if the connector is
	// canceled then attempt to finish the connection process.
	if (cluster_ \|\| is_canceled()) {
	maybe_finish();
	return;
	}

	// Otherwise, initialize the cluster and handle errors.
	if (connector->is_ok()) {
	const HostMap& hosts(connector->hosts());
	LoadBalancingPolicy::Vec policies;

	HostMap::const_iterator host_it = hosts.find(connector->address());
	if (host_it == hosts.end()) {
	// This error is unlikely to happen and it likely means that the local
	// metadata is corrupt or missing and the control connection process
	// would've probably failed before this happens.
	LOG_ERROR("Current control connection host %s not found in hosts metadata",
	connector->address().to_string().c_str());
	on_error(CLUSTER_ERROR_NO_HOSTS_AVAILABLE,
	"Control connection host is not found in hosts metadata");
	return;
	}
	Host::Ptr connected_host(host_it->second);

	// Build policies list including the default policy
	LoadBalancingPolicy::Ptr default_policy(settings_.load_balancing_policy->new_instance());
	policies.push_back(default_policy);
	for (LoadBalancingPolicy::Vec::const_iterator it = settings_.load_balancing_policies.begin(),
	end = settings_.load_balancing_policies.end();
	it != end; ++it) {
	policies.push_back(LoadBalancingPolicy::Ptr((*it)->new_instance()));
	}

	// Initialize all created policies
	for (LoadBalancingPolicy::Vec::const_iterator it = policies.begin(), end = policies.end();
	it != end; ++it) {
	LoadBalancingPolicy::Ptr policy(*it);
	policy->init(connected_host, hosts, random_, local_dc_);
	policy->register_handles(event_loop_->loop());
	}

	ScopedPtr<QueryPlan> query_plan(default_policy->new_query_plan("", NULL, NULL));
	if (!query_plan->compute_next()) { // No hosts in the query plan
	LOG_ERROR("Current control connection host %s has no hosts available in "
	"it's query plan for the configured load balancing policy. If "
	"using DC-aware check to see if the local datacenter is valid.",
	connector->address().to_string().c_str());

	const char* message;
	if (dynamic_cast<DCAwarePolicy::DCAwareQueryPlan*>(query_plan.get()) !=
	NULL) { // Check if DC-aware
	message = "No hosts available for the control connection using the "
	"DC-aware load balancing policy. "
	"Check to see if the configured local datacenter is valid";
	} else {
	message = "No hosts available for the control connection using the "
	"configured load balancing policy";
	}
	on_error(CLUSTER_ERROR_NO_HOSTS_AVAILABLE, message);
	return;
	}

	cluster_.reset(new Cluster(connector->release_connection(), listener_, event_loop_,
	connected_host, hosts, connector->schema(), default_policy, policies,
	local_dc_, connector->supported_options(), settings_));

	// Clear any connection errors and set the final negotiated protocol version.
	error_code_ = CLUSTER_OK;
	error_message_.clear();
	protocol_version_ = connector->protocol_version();

	// The cluster is initialized so the rest of the connectors can be canceled.
	for (ConnectorMap::iterator it = connectors_.begin(), end = connectors_.end(); it != end;
	++it) {
	if (it->first != connector->address()) { // Not the current connector.
	it->second->cancel();
	}
	}

	maybe_finish();
	} else if (connector->is_invalid_protocol()) {
	ProtocolVersion lower_version(connector->protocol_version().previous());
	if (!lower_version.is_valid()) {
	LOG_ERROR(
	"Host %s does not support any valid protocol version (lowest supported version is %s)",
	connector->address().to_string().c_str(),
	ProtocolVersion::lowest_supported().to_string().c_str());
	on_error(CLUSTER_ERROR_INVALID_PROTOCOL, "Unable to find supported protocol version");
	return;
	}
	LOG_INFO("Host %s does not support protocol version %s. "
	"Trying protocol version %s...",
	connector->address().to_string().c_str(),
	connector->protocol_version().to_string().c_str(), lower_version.to_string().c_str());
	internal_connect(connector->address(), lower_version);
	} else if (connector->is_ssl_error()) {
	ssl_error_code_ = connector->ssl_error_code();
	on_error(CLUSTER_ERROR_SSL_ERROR, connector->error_message());
	} else if (connector->is_auth_error()) {
	on_error(CLUSTER_ERROR_AUTH_ERROR, connector->error_message());
	} else {
	assert(!connector->is_canceled() &&
	"The control connector should have an error and not be canceled");
	on_error(CLUSTER_ERROR_NO_HOSTS_AVAILABLE, connector->error_message());
	}
	}