| from collections import OrderedDict |
| import json |
| import posixpath |
| import random |
| import threading |
| import traceback |
| import sys |
| |
| from mysos.common.cluster import get_cluster_path |
| from mysos.common.decorators import logged |
| |
| from .launcher import MySQLClusterLauncher |
| from .password import gen_password, PasswordBox |
| from .state import MySQLCluster, Scheduler, StateProvider |
| |
| import mesos.interface |
| import mesos.interface.mesos_pb2 as mesos_pb2 |
| from twitter.common import log |
| from twitter.common.collections.orderedset import OrderedSet |
| from twitter.common.metrics import AtomicGauge, MutatorGauge, Observable |
| from twitter.common.quantity import Amount, Data, Time |
| from twitter.common.quantity.parse_simple import InvalidData, parse_data |
| |
| |
| DEFAULT_TASK_CPUS = 1.0 |
| DEFAULT_TASK_DISK = Amount(2, Data.GB) |
| DEFAULT_TASK_MEM = Amount(512, Data.MB) |
| |
| # Refuse the offer for "eternity". |
| # NOTE: Using sys.maxint / 2 because sys.maxint causes rounding and precision loss when converted to |
| # double 'refuse_seconds' in the ProtoBuf and results in a negative duration on Mesos Master. |
| INCOMPATIBLE_ROLE_OFFER_REFUSE_DURATION = Amount(sys.maxint / 2, Time.NANOSECONDS) |
| |
| |
| class MysosScheduler(mesos.interface.Scheduler, Observable): |
| |
| class Error(Exception): pass |
| |
| class ClusterExists(Error): pass |
| class ClusterNotFound(Error): pass |
| class InvalidUser(Error): pass |
| class ServiceUnavailable(Error): pass |
| |
| def __init__( |
| self, |
| state, |
| state_provider, |
| framework_user, |
| executor_uri, |
| executor_cmd, |
| kazoo, |
| zk_url, |
| election_timeout, |
| admin_keypath, |
| scheduler_key, |
| installer_args=None, |
| backup_store_args=None, |
| executor_environ=None, |
| executor_source_prefix=None, |
| framework_role='*'): |
| """ |
| :param state: The Scheduler object. |
| :param state_provider: The StateProvider instance that the scheduler should use to |
| restore/persist states. |
| :param framework_user: See flags. |
| :param executor_uri: See flags. |
| :param executor_cmd: See flags. |
| :param framework_role: See flags. |
| :param election_timeout: See flags. |
| :param admin_keypath: See flags. |
| :param scheduler_key: Scheduler uses it to encrypt cluster passwords. |
| :param installer_args: See flags. |
| :param backup_store_args: See flags. |
| :param executor_environ: See flags. |
| :param executor_source_prefix: See flags. |
| :param kazoo: The Kazoo client for communicating MySQL cluster information between the |
| scheduler and the executors. |
| :param zk_url: ZooKeeper URL for used by the scheduler and the executors to access ZooKeeper. |
| """ |
| self._lock = threading.Lock() |
| |
| if not isinstance(state, Scheduler): |
| raise TypeError("'state' should be an instance of Scheduler") |
| self._state = state |
| |
| if not isinstance(state_provider, StateProvider): |
| raise TypeError("'state_provider' should be an instance of StateProvider") |
| self._state_provider = state_provider |
| |
| self._framework_user = framework_user |
| self._executor_uri = executor_uri |
| self._executor_cmd = executor_cmd |
| self._framework_role = framework_role |
| self._election_timeout = election_timeout |
| self._admin_keypath = admin_keypath |
| self._installer_args = installer_args |
| self._backup_store_args = backup_store_args |
| self._executor_environ = executor_environ |
| self._executor_source_prefix = executor_source_prefix |
| |
| self._driver = None # Will be set by registered(). |
| |
| # Use a subdir to avoid name collision with the state storage. |
| self._discover_zk_url = posixpath.join(zk_url, "discover") |
| self._kazoo = kazoo |
| |
| self._scheduler_key = scheduler_key |
| self._password_box = PasswordBox(scheduler_key) |
| |
| self._tasks = {} # {Task ID: cluster name} mappings. |
| self._launchers = OrderedDict() # Order-preserving {cluster name : MySQLClusterLauncher} |
| # mappings so cluster requests are fulfilled on a first come, |
| # first serve (FCFS) basis. |
| |
| self.stopped = threading.Event() # An event set when the scheduler is stopped. |
| self.connected = threading.Event() # An event set when the scheduler is first connected to |
| # Mesos. The scheduler tolerates later disconnections. |
| |
| self._cluster_count = self.metrics.register(AtomicGauge('cluster_count', 0)) |
| |
| # Total resources requested by the scheduler's clients. When a cluster is created its resources |
| # are added to the total; when it's deleted its resources are subtracted from the total. |
| # NOTE: These are 'requested' resources that are independent of resources offered by Mesos or |
| # allocated to or used by Mysos tasks running on Mesos cluster. |
| self._total_requested_cpus = self.metrics.register(MutatorGauge('total_requested_cpus', 0.)) |
| self._total_requested_mem_mb = self.metrics.register(MutatorGauge('total_requested_mem_mb', 0.)) |
| self._total_requested_disk_mb = self.metrics.register( |
| MutatorGauge('total_requested_disk_mb', 0.)) |
| |
| # --- Public interface. --- |
| def create_cluster(self, cluster_name, cluster_user, num_nodes, size=None, backup_id=None): |
| """ |
| :param cluster_name: Name of the cluster. |
| :param cluster_user: The user account on MySQL server. |
| :param num_nodes: Number of nodes in the cluster. |
| :param size: The size of instances in the cluster as a JSON dictionary of 'cpus', 'mem', |
| 'disk'. 'mem' and 'disk' are specified with data size units: kb, mb, gb, etc. If |
| given 'None' then app defaults are used. |
| :param backup_id: The 'backup_id' of the backup to restore from. If None then Mysos starts an |
| empty instance. |
| |
| :return: a tuple of the following: |
| - ZooKeeper URL for this Mysos cluster that can be used to resolve MySQL cluster info. |
| - The password for the specified user of the specified cluster. |
| |
| NOTE: |
| If the scheduler fails over after the cluster state is checkpointed, the new scheduler |
| instance will restore the state and continue to launch the cluster. TODO(jyx): Need to allow |
| users to retrieve the cluster info (e.g. the passwords) afterwards in case the user request |
| has failed but the cluster is successfully created eventually. |
| """ |
| with self._lock: |
| if not self._driver: |
| # 'self._driver' is set in (re)registered() after which the scheduler becomes "available". |
| # We need to get hold of the driver to recover the scheduler. |
| raise self.ServiceUnavailable("Service unavailable. Try again later") |
| |
| if cluster_name in self._state.clusters: |
| raise self.ClusterExists("Cluster '%s' already exists" % cluster_name) |
| |
| if not cluster_user: |
| raise self.InvalidUser('Invalid user name: %s' % cluster_user) |
| |
| num_nodes = int(num_nodes) |
| if num_nodes <= 0: |
| raise ValueError("Invalid number of cluster nodes: %s" % num_nodes) |
| |
| resources = parse_size(size) |
| log.info("Requested resources per instance for cluster %s: %s" % (resources, cluster_name)) |
| |
| self._total_requested_cpus.write( |
| self._total_requested_cpus.read() + resources['cpus'] * num_nodes) |
| self._total_requested_mem_mb.write( |
| self._total_requested_mem_mb.read() + resources['mem'].as_(Data.MB) * num_nodes) |
| self._total_requested_disk_mb.write( |
| self._total_requested_disk_mb.read() + resources['disk'].as_(Data.MB) * num_nodes) |
| |
| self._state.clusters.add(cluster_name) |
| self._state_provider.dump_scheduler_state(self._state) |
| |
| # Return the plaintext version to the client but store the encrypted version. |
| password = gen_password() |
| cluster = MySQLCluster( |
| cluster_name, |
| cluster_user, |
| self._password_box.encrypt(password), |
| num_nodes, |
| cpus=resources['cpus'], |
| mem=resources['mem'], |
| disk=resources['disk'], |
| backup_id=backup_id) |
| self._state_provider.dump_cluster_state(cluster) |
| |
| log.info("Creating launcher for cluster %s" % cluster_name) |
| self._launchers[cluster_name] = MySQLClusterLauncher( |
| self._driver, |
| cluster, |
| self._state_provider, |
| self._discover_zk_url, |
| self._kazoo, |
| self._framework_user, |
| self._executor_uri, |
| self._executor_cmd, |
| self._election_timeout, |
| self._admin_keypath, |
| self._scheduler_key, |
| installer_args=self._installer_args, |
| backup_store_args=self._backup_store_args, |
| executor_environ=self._executor_environ, |
| executor_source_prefix=self._executor_source_prefix, |
| framework_role=self._framework_role) |
| |
| self._cluster_count.increment() |
| |
| return get_cluster_path(self._discover_zk_url, cluster_name), password |
| |
| def delete_cluster(self, cluster_name, password): |
| """ |
| :return: ZooKeeper URL for this Mysos cluster that can be used to wait for the termination of |
| the cluster. |
| """ |
| with self._lock: |
| if not self._driver: |
| raise self.ServiceUnavailable("Service unavailable. Try again later") |
| |
| if cluster_name not in self._state.clusters: |
| raise self.ClusterNotFound("Cluster '%s' not found" % cluster_name) |
| |
| launcher = self._launchers[cluster_name] |
| launcher.kill(password) |
| log.info("Attempted to kill cluster %s" % cluster_name) |
| |
| self._cluster_count.decrement() |
| cluster_info = launcher.cluster_info |
| self._total_requested_cpus.write(self._total_requested_cpus.read() - cluster_info.total_cpus) |
| self._total_requested_mem_mb.write( |
| self._total_requested_mem_mb.read() - cluster_info.total_mem_mb) |
| self._total_requested_disk_mb.write( |
| self._total_requested_disk_mb.read() - cluster_info.total_disk_mb) |
| |
| if launcher.terminated: |
| log.info("Deleting the launcher for cluster %s directly because the cluster has already " |
| "terminated" % launcher.cluster_name) |
| self._delete_launcher(launcher) |
| |
| return get_cluster_path(self._discover_zk_url, cluster_name) |
| |
| @property |
| def clusters(self): |
| """ |
| A generator for information about clusters. |
| """ |
| with self._lock: |
| for launcher in self._launchers.values(): |
| yield launcher.cluster_info |
| |
| def _stop(self): |
| """Stop the scheduler.""" |
| |
| # 'self._driver' is set in registered() so it could be None if the scheduler is asked to stop |
| # before it is registered (e.g. an error is received). |
| if self._driver: |
| self._driver.stop(True) # Set failover to True. |
| self.stopped.set() |
| |
| # --- Mesos methods. --- |
| @logged |
| def registered(self, driver, frameworkId, masterInfo): |
| self._driver = driver |
| self._state.framework_info.id.value = frameworkId.value |
| self._state_provider.dump_scheduler_state(self._state) |
| |
| # Recover only after the scheduler is connected because it needs '_driver' to be assigned. This |
| # is blocking the scheduler driver thread but we do want further messages to be blocked until |
| # the scheduler state is fully recovered. |
| # TODO(jyx): If performance becomes an issue, we can also restore all the state data while the |
| # driver is connecting and proceed to recover all the internal state objects after the driver is |
| # connected. |
| try: |
| self._recover() |
| except Exception as e: |
| log.error("Stopping scheduler because: %s" % e) |
| log.error(traceback.format_exc()) |
| self._stop() |
| return |
| |
| self.connected.set() |
| |
| def _recover(self): |
| """ |
| Recover launchers for existing clusters. A newly created scheduler has no launcher to recover. |
| TODO(jyx): The recovery of clusters can potentially be parallelized. |
| """ |
| for cluster_name in OrderedSet(self._state.clusters): # Make a copy so we can remove dead |
| # entries while iterating the copy. |
| log.info("Recovering launcher for cluster %s" % cluster_name) |
| |
| cluster = self._state_provider.load_cluster_state(cluster_name) |
| if not cluster: |
| # The scheduler could have failed over before creating the launcher. The user request |
| # should have failed and there is no cluster state to restore. |
| log.info("Skipping cluster %s because its state cannot be found" % cluster_name) |
| self._state.clusters.remove(cluster_name) |
| self._state_provider.dump_scheduler_state(self._state) |
| continue |
| |
| for task_id in cluster.tasks: |
| self._tasks[task_id] = cluster.name # Reconstruct the 'tasks' map. |
| |
| # Order of launchers is preserved thanks to the OrderedSet. |
| # For recovered launchers we use the currently specified --framework_role and |
| # --executor_environ, etc., instead of saving it in cluster state so the change in flags can |
| # be picked up by existing clusters. |
| self._launchers[cluster.name] = MySQLClusterLauncher( |
| self._driver, |
| cluster, |
| self._state_provider, |
| self._discover_zk_url, |
| self._kazoo, |
| self._framework_user, |
| self._executor_uri, |
| self._executor_cmd, |
| self._election_timeout, |
| self._admin_keypath, |
| self._scheduler_key, |
| installer_args=self._installer_args, |
| backup_store_args=self._backup_store_args, |
| executor_environ=self._executor_environ, |
| executor_source_prefix=self._executor_source_prefix, |
| framework_role=self._framework_role) |
| |
| # Recover metrics from restored state. |
| self._cluster_count.increment() |
| |
| cluster_info = self._launchers[cluster.name].cluster_info |
| self._total_requested_cpus.write(self._total_requested_cpus.read() + cluster_info.total_cpus) |
| self._total_requested_mem_mb.write( |
| self._total_requested_mem_mb.read() + cluster_info.total_mem_mb) |
| self._total_requested_disk_mb.write( |
| self._total_requested_disk_mb.read() + cluster_info.total_disk_mb) |
| |
| log.info("Recovered %s clusters" % len(self._launchers)) |
| |
| @logged |
| def reregistered(self, driver, masterInfo): |
| self._driver = driver |
| self.connected.set() |
| # TODO(jyx): Reconcile tasks. |
| |
| @logged |
| def disconnected(self, driver): |
| pass |
| |
| @logged |
| def resourceOffers(self, driver, offers): |
| log.info('Got %d resource offers' % len(offers)) |
| |
| with self._lock: |
| # Current scheduling algorithm: randomly pick an offer and loop through the list of launchers |
| # until one decides to use this offer to launch a task. |
| # It's possible to launch multiple tasks on the same Mesos slave (in different batches of |
| # offers). |
| for offer in shuffled(offers): |
| task_id = None |
| # 'filters' is set when we need to create a special filter for incompatible roles. |
| filters = None |
| # For each offer, launchers are asked to launch a task in the order they are added (FCFS). |
| for name in self._launchers: |
| launcher = self._launchers[name] |
| try: |
| task_id, _ = launcher.launch(offer) |
| except MySQLClusterLauncher.IncompatibleRoleError as e: |
| # This "error" is not severe and we expect this to occur frequently only when Mysos |
| # first joins the cluster. For a running Mesos cluster this should be somewhat rare |
| # because we refuse the offer "forever". |
| filters = mesos_pb2.Filters() |
| filters.refuse_seconds = INCOMPATIBLE_ROLE_OFFER_REFUSE_DURATION.as_(Time.SECONDS) |
| break # No need to check with other launchers. |
| if task_id: |
| self._tasks[task_id] = launcher.cluster_name |
| # No need to check with other launchers. 'filters' remains unset. |
| break |
| if task_id: |
| break # Some launcher has used this offer. Move on to the next one. |
| |
| if filters: |
| log.info("Declining offer %s for %s because '%s'" % ( |
| offer.id.value, INCOMPATIBLE_ROLE_OFFER_REFUSE_DURATION, e)) |
| else: |
| log.info("Declining offer %s because no launcher can use this offer" % offer.id.value) |
| # Mesos scheduler Python binding doesn't deal with filters='None' properly. |
| # See https://issues.apache.org/jira/browse/MESOS-2567. |
| filters = mesos_pb2.Filters() |
| self._driver.declineOffer(offer.id, filters) |
| |
| @logged |
| def statusUpdate(self, driver, status): |
| with self._lock: |
| # Forward the status update to the corresponding launcher. |
| task_id = status.task_id.value |
| launcher = self._get_launcher_by_task_id(task_id) |
| if not launcher: |
| log.info("Cluster for task %s doesn't exist. It could have been removed" % task_id) |
| return |
| |
| try: |
| launcher.status_update(status) |
| except MySQLClusterLauncher.Error as e: |
| log.error("Status update failed due to launcher error: %s" % e.message) |
| self._stop() |
| |
| if launcher.terminated: |
| log.info("Deleting the launcher for cluster %s because the cluster has terminated" % |
| launcher.cluster_name) |
| self._delete_launcher(launcher) |
| |
| def _delete_launcher(self, launcher): |
| assert launcher.terminated |
| self._state.clusters.discard(launcher.cluster_name) |
| self._state_provider.dump_scheduler_state(self._state) |
| del self._launchers[launcher.cluster_name] |
| |
| @logged |
| def frameworkMessage(self, driver, executorId, slaveId, message): |
| log.info('Received framework message %s' % message) |
| task_id = executorId.value # task_id == executor_id in Mysos. |
| |
| launcher = self._get_launcher_by_task_id(task_id) |
| launcher.framework_message(task_id, slaveId.value, message) |
| |
| @logged |
| def slaveLost(self, driver, slaveId): |
| # We receive TASK_LOSTs when a slave is lost so we we don't need to handle it separately here. |
| pass |
| |
| @logged |
| def error(self, driver, message): |
| log.error('Received error from mesos: %s' % message) |
| self._stop() # SchedulerDriver aborts when an error message is received. |
| |
| def _get_launcher_by_task_id(self, task_id): |
| # TODO(jyx): Currently we don't delete entries from 'self._tasks' so a mapping can always |
| # be found but we should clean it up when tasks die. |
| assert task_id in self._tasks |
| cluster_name = self._tasks[task_id] |
| return self._launchers.get(cluster_name) # Cluster could have been removed. |
| |
| |
| def shuffled(li): |
| """Return a shuffled version of the list.""" |
| copy = li[:] |
| random.shuffle(copy) |
| return copy |
| |
| |
| def parse_size(size): |
| """Return the resources specified in 'size' as a dictionary.""" |
| if not size: |
| resources = dict(cpus=DEFAULT_TASK_CPUS, mem=DEFAULT_TASK_MEM, disk=DEFAULT_TASK_DISK) |
| else: |
| # TODO(jyx): Simplify this using T-shirt sizing |
| # (https://github.com/twitter/mysos/issues/14). |
| try: |
| resources_ = json.loads(size) |
| resources = dict( |
| cpus=float(resources_['cpus']), |
| mem=parse_data(resources_['mem']), |
| disk=parse_data(resources_['disk'])) |
| except (TypeError, KeyError, ValueError, InvalidData): |
| raise ValueError("'size' should be a JSON dictionary with keys 'cpus', 'mem' and 'disk'") |
| |
| return resources |
