master

PhabricatorDatabaseRef.php
View Options

			<?php

			final class PhabricatorDatabaseRef
			extends Phobject {

			const STATUS_OKAY = 'okay';
			const STATUS_FAIL = 'fail';
			const STATUS_AUTH = 'auth';
			const STATUS_REPLICATION_CLIENT = 'replication-client';

			const REPLICATION_OKAY = 'okay';
			const REPLICATION_MASTER_REPLICA = 'master-replica';
			const REPLICATION_REPLICA_NONE = 'replica-none';
			const REPLICATION_SLOW = 'replica-slow';
			const REPLICATION_NOT_REPLICATING = 'not-replicating';

			const KEY_HEALTH = 'cluster.db.health';
			const KEY_REFS = 'cluster.db.refs';
			const KEY_INDIVIDUAL = 'cluster.db.individual';

			private $host;
			private $port;
			private $user;
			private $pass;
			private $disabled;
			private $isMaster;
			private $isIndividual;

			private $connectionLatency;
			private $connectionStatus;
			private $connectionMessage;
			private $connectionException;

			private $replicaStatus;
			private $replicaMessage;
			private $replicaDelay;

			private $healthRecord;
			private $didFailToConnect;

			private $isDefaultPartition;
			private $applicationMap = array();
			private $masterRef;
			private $replicaRefs = array();
			private $usePersistentConnections;

			public function setHost($host) {
			$this->host = $host;
			return $this;
			}

			public function getHost() {
			return $this->host;
			}

			public function setPort($port) {
			$this->port = $port;
			return $this;
			}

			public function getPort() {
			return $this->port;
			}

			public function setUser($user) {
			$this->user = $user;
			return $this;
			}

			public function getUser() {
			return $this->user;
			}

			public function setPass(PhutilOpaqueEnvelope $pass) {
			$this->pass = $pass;
			return $this;
			}

			public function getPass() {
			return $this->pass;
			}

			public function setIsMaster($is_master) {
			$this->isMaster = $is_master;
			return $this;
			}

			public function getIsMaster() {
			return $this->isMaster;
			}

			public function setDisabled($disabled) {
			$this->disabled = $disabled;
			return $this;
			}

			public function getDisabled() {
			return $this->disabled;
			}

			public function setConnectionLatency($connection_latency) {
			$this->connectionLatency = $connection_latency;
			return $this;
			}

			public function getConnectionLatency() {
			return $this->connectionLatency;
			}

			public function setConnectionStatus($connection_status) {
			$this->connectionStatus = $connection_status;
			return $this;
			}

			public function getConnectionStatus() {
			if ($this->connectionStatus === null) {
			throw new PhutilInvalidStateException('queryAll');
			}

			return $this->connectionStatus;
			}

			public function setConnectionMessage($connection_message) {
			$this->connectionMessage = $connection_message;
			return $this;
			}

			public function getConnectionMessage() {
			return $this->connectionMessage;
			}

			public function setReplicaStatus($replica_status) {
			$this->replicaStatus = $replica_status;
			return $this;
			}

			public function getReplicaStatus() {
			return $this->replicaStatus;
			}

			public function setReplicaMessage($replica_message) {
			$this->replicaMessage = $replica_message;
			return $this;
			}

			public function getReplicaMessage() {
			return $this->replicaMessage;
			}

			public function setReplicaDelay($replica_delay) {
			$this->replicaDelay = $replica_delay;
			return $this;
			}

			public function getReplicaDelay() {
			return $this->replicaDelay;
			}

			public function setIsIndividual($is_individual) {
			$this->isIndividual = $is_individual;
			return $this;
			}

			public function getIsIndividual() {
			return $this->isIndividual;
			}

			public function setIsDefaultPartition($is_default_partition) {
			$this->isDefaultPartition = $is_default_partition;
			return $this;
			}

			public function getIsDefaultPartition() {
			return $this->isDefaultPartition;
			}

			public function setUsePersistentConnections($use_persistent_connections) {
			$this->usePersistentConnections = $use_persistent_connections;
			return $this;
			}

			public function getUsePersistentConnections() {
			return $this->usePersistentConnections;
			}

			public function setApplicationMap(array $application_map) {
			$this->applicationMap = $application_map;
			return $this;
			}

			public function getApplicationMap() {
			return $this->applicationMap;
			}

			public function getPartitionStateForCommit() {
			$state = PhabricatorEnv::getEnvConfig('cluster.databases');
			foreach ($state as $key => $value) {
			// Don't store passwords, since we don't care if they differ and
			// users may find it surprising.
			unset($state[$key]['pass']);
			}

			return phutil_json_encode($state);
			}

			public function setMasterRef(PhabricatorDatabaseRef $master_ref) {
			$this->masterRef = $master_ref;
			return $this;
			}

			public function getMasterRef() {
			return $this->masterRef;
			}

			public function addReplicaRef(PhabricatorDatabaseRef $replica_ref) {
			$this->replicaRefs[] = $replica_ref;
			return $this;
			}

			public function getReplicaRefs() {
			return $this->replicaRefs;
			}

			public function getDisplayName() {
			return $this->getRefKey();
			}

			public function getRefKey() {
			$host = $this->getHost();

			$port = $this->getPort();
			if ($port) {
			return "{$host}:{$port}";
			}

			return $host;
			}

			public static function getConnectionStatusMap() {
			return array(
			self::STATUS_OKAY => array(
			'icon' => 'fa-exchange',
			'color' => 'green',
			'label' => pht('Okay'),
			),
			self::STATUS_FAIL => array(
			'icon' => 'fa-times',
			'color' => 'red',
			'label' => pht('Failed'),
			),
			self::STATUS_AUTH => array(
			'icon' => 'fa-key',
			'color' => 'red',
			'label' => pht('Invalid Credentials'),
			),
			self::STATUS_REPLICATION_CLIENT => array(
			'icon' => 'fa-eye-slash',
			'color' => 'yellow',
			'label' => pht('Missing Permission'),
			),
			);
			}

			public static function getReplicaStatusMap() {
			return array(
			self::REPLICATION_OKAY => array(
			'icon' => 'fa-download',
			'color' => 'green',
			'label' => pht('Okay'),
			),
			self::REPLICATION_MASTER_REPLICA => array(
			'icon' => 'fa-database',
			'color' => 'red',
			'label' => pht('Replicating Master'),
			),
			self::REPLICATION_REPLICA_NONE => array(
			'icon' => 'fa-download',
			'color' => 'red',
			'label' => pht('Not A Replica'),
			),
			self::REPLICATION_SLOW => array(
			'icon' => 'fa-hourglass',
			'color' => 'red',
			'label' => pht('Slow Replication'),
			),
			self::REPLICATION_NOT_REPLICATING => array(
			'icon' => 'fa-exclamation-triangle',
			'color' => 'red',
			'label' => pht('Not Replicating'),
			),
			);
			}

			public static function getClusterRefs() {
			$cache = PhabricatorCaches::getRequestCache();

			$refs = $cache->getKey(self::KEY_REFS);
			if (!$refs) {
			$refs = self::newRefs();
			$cache->setKey(self::KEY_REFS, $refs);
			}

			return $refs;
			}

			public static function getLiveIndividualRef() {
			$cache = PhabricatorCaches::getRequestCache();

			$ref = $cache->getKey(self::KEY_INDIVIDUAL);
			if (!$ref) {
			$ref = self::newIndividualRef();
			$cache->setKey(self::KEY_INDIVIDUAL, $ref);
			}

			return $ref;
			}

			public static function newRefs() {
			$default_port = PhabricatorEnv::getEnvConfig('mysql.port');
			$default_port = nonempty($default_port, 3306);

			$default_user = PhabricatorEnv::getEnvConfig('mysql.user');

			$default_pass = PhabricatorEnv::getEnvConfig('mysql.pass');
			$default_pass = phutil_string_cast($default_pass);
			$default_pass = new PhutilOpaqueEnvelope($default_pass);

			$config = PhabricatorEnv::getEnvConfig('cluster.databases');

			return id(new PhabricatorDatabaseRefParser())
			->setDefaultPort($default_port)
			->setDefaultUser($default_user)
			->setDefaultPass($default_pass)
			->newRefs($config);
			}

			public static function queryAll() {
			$refs = self::getActiveDatabaseRefs();
			return self::queryRefs($refs);
			}

			private static function queryRefs(array $refs) {
			foreach ($refs as $ref) {
			$conn = $ref->newManagementConnection();

			$t_start = microtime(true);
			$replica_status = false;
			try {
			$replica_status = queryfx_one($conn, 'SHOW SLAVE STATUS');
			$ref->setConnectionStatus(self::STATUS_OKAY);
			} catch (AphrontAccessDeniedQueryException $ex) {
			$ref->setConnectionStatus(self::STATUS_REPLICATION_CLIENT);
			$ref->setConnectionMessage(
			pht(
			'No permission to run "SHOW SLAVE STATUS". Grant this user '.
			'"REPLICATION CLIENT" permission to allow this server to '.
			'monitor replica health.'));
			} catch (AphrontInvalidCredentialsQueryException $ex) {
			$ref->setConnectionStatus(self::STATUS_AUTH);
			$ref->setConnectionMessage($ex->getMessage());
			} catch (AphrontQueryException $ex) {
			$ref->setConnectionStatus(self::STATUS_FAIL);

			$class = get_class($ex);
			$message = $ex->getMessage();
			$ref->setConnectionMessage(
			pht(
			'%s: %s',
			get_class($ex),
			$ex->getMessage()));
			}
			$t_end = microtime(true);
			$ref->setConnectionLatency($t_end - $t_start);

			if ($replica_status !== false) {
			$is_replica = (bool)$replica_status;
			if ($ref->getIsMaster() && $is_replica) {
			$ref->setReplicaStatus(self::REPLICATION_MASTER_REPLICA);
			$ref->setReplicaMessage(
			pht(
			'This host has a "master" role, but is replicating data from '.
			'another host ("%s")!',
			idx($replica_status, 'Master_Host')));
			} else if (!$ref->getIsMaster() && !$is_replica) {
			$ref->setReplicaStatus(self::REPLICATION_REPLICA_NONE);
			$ref->setReplicaMessage(
			pht(
			'This host has a "replica" role, but is not replicating data '.
			'from a master (no output from "SHOW SLAVE STATUS").'));
			} else {
			$ref->setReplicaStatus(self::REPLICATION_OKAY);
			}

			if ($is_replica) {
			$latency = idx($replica_status, 'Seconds_Behind_Master');
			if (!phutil_nonempty_string($latency)) {
			$ref->setReplicaStatus(self::REPLICATION_NOT_REPLICATING);
			} else {
			$latency = (int)$latency;
			$ref->setReplicaDelay($latency);
			if ($latency > 30) {
			$ref->setReplicaStatus(self::REPLICATION_SLOW);
			$ref->setReplicaMessage(
			pht(
			'This replica is lagging far behind the master. Data is at '.
			'risk!'));
			}
			}
			}
			}
			}

			return $refs;
			}

			public function newManagementConnection() {
			return $this->newConnection(
			array(
			'retries' => 0,
			'timeout' => 2,
			));
			}

			public function newApplicationConnection($database) {
			return $this->newConnection(
			array(
			'database' => $database,
			));
			}

			public function isSevered() {
			// If we only have an individual database, never sever our connection to
			// it, at least for now. It's possible that using the same severing rules
			// might eventually make sense to help alleviate load-related failures,
			// but we should wait for all the cluster stuff to stabilize first.
			if ($this->getIsIndividual()) {
			return false;
			}

			if ($this->didFailToConnect) {
			return true;
			}

			$record = $this->getHealthRecord();
			$is_healthy = $record->getIsHealthy();
			if (!$is_healthy) {
			return true;
			}

			return false;
			}

			public function isReachable(AphrontDatabaseConnection $connection) {
			$record = $this->getHealthRecord();
			$should_check = $record->getShouldCheck();

			if ($this->isSevered() && !$should_check) {
			return false;
			}

			$this->connectionException = null;
			try {
			$connection->openConnection();
			$reachable = true;
			} catch (AphrontSchemaQueryException $ex) {
			// We get one of these if the database we're trying to select does not
			// exist. In this case, just re-throw the exception. This is expected
			// during first-time setup, when databases like "config" will not exist
			// yet.
			throw $ex;
			} catch (Exception $ex) {
			$this->connectionException = $ex;
			$reachable = false;
			}

			if ($should_check) {
			$record->didHealthCheck($reachable);
			}

			if (!$reachable) {
			$this->didFailToConnect = true;
			}

			return $reachable;
			}

			public function checkHealth() {
			$health = $this->getHealthRecord();

			$should_check = $health->getShouldCheck();
			if ($should_check) {
			// This does an implicit health update.
			$connection = $this->newManagementConnection();
			$this->isReachable($connection);
			}

			return $this;
			}

			private function getHealthRecordCacheKey() {
			$host = $this->getHost();
			$port = $this->getPort();
			$key = self::KEY_HEALTH;

			return "{$key}({$host}, {$port})";
			}

			public function getHealthRecord() {
			if (!$this->healthRecord) {
			$this->healthRecord = new PhabricatorClusterServiceHealthRecord(
			$this->getHealthRecordCacheKey());
			}
			return $this->healthRecord;
			}

			public function getConnectionException() {
			return $this->connectionException;
			}

			public static function getActiveDatabaseRefs() {
			$refs = array();

			foreach (self::getMasterDatabaseRefs() as $ref) {
			$refs[] = $ref;
			}

			foreach (self::getReplicaDatabaseRefs() as $ref) {
			$refs[] = $ref;
			}

			return $refs;
			}

			public static function getAllMasterDatabaseRefs() {
			$refs = self::getClusterRefs();

			if (!$refs) {
			return array(self::getLiveIndividualRef());
			}

			$masters = array();
			foreach ($refs as $ref) {
			if ($ref->getIsMaster()) {
			$masters[] = $ref;
			}
			}

			return $masters;
			}

			public static function getMasterDatabaseRefs() {
			$refs = self::getAllMasterDatabaseRefs();
			return self::getEnabledRefs($refs);
			}

			public function isApplicationHost($database) {
			return isset($this->applicationMap[$database]);
			}

			public function loadRawMySQLConfigValue($key) {
			$conn = $this->newManagementConnection();

			try {
			$value = queryfx_one($conn, 'SELECT @@%C', $key);

			// NOTE: Although MySQL allows us to escape configuration values as if
			// they are column names, the escaping is included in the column name
			// of the return value: if we select "@@`x`", we get back a column named
			// "@@`x`", not "@@x" as we might expect.
			$value = head($value);

			} catch (AphrontQueryException $ex) {
			$value = null;
			}

			return $value;
			}

			public static function getMasterDatabaseRefForApplication($application) {
			$masters = self::getMasterDatabaseRefs();

			$application_master = null;
			$default_master = null;
			foreach ($masters as $master) {
			if ($master->isApplicationHost($application)) {
			$application_master = $master;
			break;
			}
			if ($master->getIsDefaultPartition()) {
			$default_master = $master;
			}
			}

			if ($application_master) {
			$masters = array($application_master);
			} else if ($default_master) {
			$masters = array($default_master);
			} else {
			$masters = array();
			}

			$masters = self::getEnabledRefs($masters);
			$master = head($masters);

			return $master;
			}

			public static function newIndividualRef() {
			$default_user = PhabricatorEnv::getEnvConfig('mysql.user');
			$default_pass = new PhutilOpaqueEnvelope(
			PhabricatorEnv::getEnvConfig('mysql.pass'));
			$default_host = PhabricatorEnv::getEnvConfig('mysql.host');
			$default_port = PhabricatorEnv::getEnvConfig('mysql.port');

			return id(new self())
			->setUser($default_user)
			->setPass($default_pass)
			->setHost($default_host)
			->setPort($default_port)
			->setIsIndividual(true)
			->setIsMaster(true)
			->setIsDefaultPartition(true)
			->setUsePersistentConnections(false);
			}

			public static function getAllReplicaDatabaseRefs() {
			$refs = self::getClusterRefs();

			if (!$refs) {
			return array();
			}

			$replicas = array();
			foreach ($refs as $ref) {
			if ($ref->getIsMaster()) {
			continue;
			}

			$replicas[] = $ref;
			}

			return $replicas;
			}

			public static function getReplicaDatabaseRefs() {
			$refs = self::getAllReplicaDatabaseRefs();
			return self::getEnabledRefs($refs);
			}

			private static function getEnabledRefs(array $refs) {
			foreach ($refs as $key => $ref) {
			if ($ref->getDisabled()) {
			unset($refs[$key]);
			}
			}
			return $refs;
			}

			public static function getReplicaDatabaseRefForApplication($application) {
			$replicas = self::getReplicaDatabaseRefs();

			$application_replicas = array();
			$default_replicas = array();
			foreach ($replicas as $replica) {
			$master = $replica->getMasterRef();

			if ($master->isApplicationHost($application)) {
			$application_replicas[] = $replica;
			}

			if ($master->getIsDefaultPartition()) {
			$default_replicas[] = $replica;
			}
			}

			if ($application_replicas) {
			$replicas = $application_replicas;
			} else {
			$replicas = $default_replicas;
			}

			$replicas = self::getEnabledRefs($replicas);

			// TODO: We may have multiple replicas to choose from, and could make
			// more of an effort to pick the "best" one here instead of always
			// picking the first one. Once we've picked one, we should try to use
			// the same replica for the rest of the request, though.

			return head($replicas);
			}

			private function newConnection(array $options) {
			// If we believe the database is unhealthy, don't spend as much time
			// trying to connect to it, since it's likely to continue to fail and
			// hammering it can only make the problem worse.
			$record = $this->getHealthRecord();
			if ($record->getIsHealthy()) {
			$default_retries = 3;
			$default_timeout = 10;
			} else {
			$default_retries = 0;
			$default_timeout = 2;
			}

			$spec = $options + array(
			'user' => $this->getUser(),
			'pass' => $this->getPass(),
			'host' => $this->getHost(),
			'port' => $this->getPort(),
			'database' => null,
			'retries' => $default_retries,
			'timeout' => $default_timeout,
			'persistent' => $this->getUsePersistentConnections(),
			);

			$is_cli = (php_sapi_name() == 'cli');

			$use_persistent = false;
			if (!empty($spec['persistent']) && !$is_cli) {
			$use_persistent = true;
			}
			unset($spec['persistent']);

			$connection = self::newRawConnection($spec);

			// If configured, use persistent connections. See T11672 for details.
			if ($use_persistent) {
			$connection->setPersistent($use_persistent);
			}

			// Unless this is a script running from the CLI, prevent any query from
			// running for more than 30 seconds. See T10849 for details.
			if (!$is_cli) {
			$connection->setQueryTimeout(30);
			}

			return $connection;
			}

			public static function newRawConnection(array $options) {
			if (extension_loaded('mysqli')) {
			return new AphrontMySQLiDatabaseConnection($options);
			} else {
			return new AphrontMySQLDatabaseConnection($options);
			}
			}

			}

Owner Packages

No Owners

phorge/src/infrastructure/cluster/PhabricatorDatabaseRef.phpe81eff68e581HEAD -> master

PhabricatorDatabaseRef.phpView Options

Owner Packages

phorge/src/infrastructure/cluster/PhabricatorDatabaseRef.php
e81eff68e581HEAD -> master

PhabricatorDatabaseRef.php
View Options