 # 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.    

from datetime import datetime
from random import randint
from socket import gethostname
import logging
import threading
import time

from tashi.rpycservices.rpyctypes import Errors, InstanceState, HostState, TashiException
from tashi import boolean, convertExceptions, ConnectionManager, vmStates, timed, version, scrubString

class ClusterManagerService(object):
	"""RPC service for the ClusterManager"""
	
	def __init__(self, config, data, dfs):
		self.config = config
		self.data = data
		self.authAndEncrypt = boolean(config.get('Security', 'authAndEncrypt'))
		if self.authAndEncrypt:
			self.username = config.get('AccessNodeManager', 'username')
			self.password = config.get('AccessNodeManager', 'password')
		else:
			self.username = None
			self.password = None
		self.proxy = ConnectionManager(self.username, self.password, int(self.config.get('ClusterManager', 'nodeManagerPort')))
		self.dfs = dfs
		self.convertExceptions = boolean(config.get('ClusterManagerService', 'convertExceptions'))
		self.log = logging.getLogger(__name__)
		self.hostLastContactTime = {}
		#self.hostLastUpdateTime = {}
		self.instanceLastContactTime = {}
		self.expireHostTime = float(self.config.get('ClusterManagerService', 'expireHostTime'))
		self.allowDecayed = float(self.config.get('ClusterManagerService', 'allowDecayed'))
		self.allowMismatchedVersions = boolean(self.config.get('ClusterManagerService', 'allowMismatchedVersions'))
		self.maxMemory = int(self.config.get('ClusterManagerService', 'maxMemory'))
		self.maxCores = int(self.config.get('ClusterManagerService', 'maxCores'))
		self.allowDuplicateNames = boolean(self.config.get('ClusterManagerService', 'allowDuplicateNames'))
		now = self.__now()
		for instance in self.data.getInstances().itervalues():
			instanceId = instance.id
			instance = self.data.acquireInstance(instanceId)
			instance.decayed = False

			if instance.hostId is None:
				self.stateTransition(instance, None, InstanceState.Pending)
			else:
				self.stateTransition(instance, None, InstanceState.Orphaned)

			self.data.releaseInstance(instance)
		for host in self.data.getHosts().itervalues():
			hostId = host.id
			host = self.data.acquireHost(hostId)
			host.up = False
			host.decayed = False
			self.data.releaseHost(host)
		threading.Thread(target=self.monitorCluster).start()

	def stateTransition(self, instance, old, cur):
		if (old and instance.state != old):
			raise TashiException(d={'errno':Errors.IncorrectVmState,'msg':"VmState is not %s - it is %s" % (vmStates[old], vmStates[instance.state])})
		if (instance.state == cur):
			return

		instance.state = cur
		try:
			host = self.data.getHost(instance.hostId)
			vmId = instance.vmId
			self.proxy[host.name].vmStateChange(vmId, old, cur)
		except:
			#XXXstroucki append to a list?
			pass

	def __now(self):
		return time.time()

	def __downHost(self, host):
		self.log.warning('Host %s is down' % (host.name))
		host.up = False
		host.decayed = False

		self.__orphanInstances(host)

	def __upHost(self, host):
		self.log.warning('Host %s is up' % (host.name))
		host.up = True
		host.decayed = True

	def __orphanInstances(self, host):
		# expects lock to be held on host
		instances = [instance.id for instance in self.data.getInstances().itervalues() if instance.hostId == host.id]

		for instanceId in instances:
			instance = self.data.acquireInstance(instanceId)
			if instance.hostId == host.id:
				instance.decayed = True
				self.stateTransition(instance, None, InstanceState.Orphaned)

			self.data.releaseInstance(instance)

	def __checkHosts(self):
		# Check if hosts have been heard from recently
		# Otherwise, see if it is alive

		for hostId in self.hostLastContactTime.keys():
			if (self.hostLastContactTime[hostId] < (self.__now() - self.expireHostTime)):
				host = self.data.acquireHost(hostId)
				string = None
				try:
					string = self.proxy[host.name].liveCheck()
				except:
					pass

				if string != "alive":
					self.__downHost(host)
					del self.hostLastContactTime[hostId]
				else:
					self.__upHost(host)
					self.hostLastContactTime[hostId] = self.__now()

				self.data.releaseHost(host)

	def __checkInstances(self):
		# Reconcile instances with nodes

		# obtain a list of instances I know about
		myInstancesError = False
		try:
			myInstances = self.data.getInstances()
		except:
			myInstancesError = True
			self.log.warning('Failure communicating with my database')

		if myInstancesError == True:
			return

		# iterate through all hosts I believe are up
		for hostId in self.hostLastContactTime.keys():
			#self.log.warning("iterate %d" % hostId)
			host = self.data.acquireHost(hostId)
			if (self.hostLastContactTime[hostId] < (self.__now() - self.allowDecayed)):
				host.decayed = True

				self.log.info('Fetching state from host %s because it is decayed' % (host.name))
				
				myInstancesThisHost = [i for i in myInstances.values() if i.hostId == host.id]

				# get a list of VMs running on host
				try:
					hostProxy = self.proxy[host.name]
					remoteInstances = [hostProxy.getVmInfo(vmId) for vmId in hostProxy.listVms()]
				except:
					self.log.warning('Failure getting instances from host %s' % (host.name))
					self.data.releaseHost(host)
					continue

				# register instances I don't know about
				for instance in remoteInstances:
					if (instance.id not in myInstances):
						instance.hostId = host.id
						instance = self.data.registerInstance(instance)
						self.data.releaseInstance(instance)
				remoteInstanceIds = [i.id for i in remoteInstances]
				# remove instances that shouldn't be running
				for instance in myInstancesThisHost:
					if (instance.id not in remoteInstanceIds):
						# XXXstroucki before 20110902 excepted here with host lock
						try:
							instance = self.data.acquireInstance(instance.id)
						except:
							continue

						# XXXstroucki destroy?
						try:
							del self.instanceLastContactTime[instance.id]
						except:
							pass
						self.data.removeInstance(instance)

				self.hostLastContactTime[hostId] = self.__now()
				host.decayed = False

			self.data.releaseHost(host)
			#self.log.warning("iterate %d done" % hostId)
		
		# iterate through all VMs I believe are active
		for instanceId in self.instanceLastContactTime.keys():
			if (self.instanceLastContactTime[instanceId] < (self.__now() - self.allowDecayed)):
				try:
					instance = self.data.acquireInstance(instanceId)
				except:
					continue

				instance.decayed = True
				self.log.info('Fetching state on instance %s because it is decayed' % (instance.name))
				if instance.hostId is None: raise AssertionError

				# XXXstroucki check if host is down?
				host = self.data.getHost(instance.hostId)

				# get updated state on VM
				try:
					hostProxy = self.proxy[host.name]
					newInstance = hostProxy.getVmInfo(instance.vmId)
				except:
					self.log.warning('Failure getting data for instance %s from host %s' % (instance.name, host.name))
					self.data.releaseInstance(instance)
					continue

				# replace existing state with new state
				# XXXstroucki more?
				instance.state = newInstance.state
				self.instanceLastContactTime[instanceId] = self.__now()
				instance.decayed = False
				self.data.releaseInstance(instance)


	def monitorCluster(self):
		while True:
			sleepFor = min(self.expireHostTime, self.allowDecayed)

			try:
				self.__checkHosts()
				self.__checkInstances()
			except:
				self.log.exception('monitorCluster iteration failed')
			self.log.info("Sleeping for %d seconds" % sleepFor)
			time.sleep(sleepFor)


	def normalize(self, instance):
		instance.id = None
		instance.vmId = None
		instance.hostId = None
		instance.decayed = False
		instance.name = scrubString(instance.name, allowed="ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-.")
		instance.state = InstanceState.Pending
		# XXXstroucki At some point, check userId
		if (not self.allowDuplicateNames):
			for i in self.data.getInstances().itervalues():
				if (i.name == instance.name):
					raise TashiException(d={'errno':Errors.InvalidInstance,'msg':"The name %s is already in use" % (instance.name)})
		if (instance.cores < 1):
			raise TashiException(d={'errno':Errors.InvalidInstance,'msg':"Number of cores must be >= 1"})
		if (instance.cores > self.maxCores):
			raise TashiException(d={'errno':Errors.InvalidInstance,'msg':"Number of cores must be <= %d" % (self.maxCores)})
		if (instance.memory < 1):
			raise TashiException(d={'errno':Errors.InvalidInstance,'msg':"Amount of memory must be >= 1"})
		if (instance.memory > self.maxMemory):
			raise TashiException(d={'errno':Errors.InvalidInstance,'msg':"Amount of memory must be <= %d" % (self.maxMemory)})
		# Make sure disk spec is valid
		# Make sure network spec is valid
		# Ignore internal hints
		for hint in instance.hints:
			if (hint.startswith("__")):
				del instance.hints[hint]
		return instance
	
	def createVm(self, instance):
		"""Function to add a VM to the list of pending VMs"""
		# XXXstroucki: check for exception here
		instance = self.normalize(instance)
		instance = self.data.registerInstance(instance)
		self.data.releaseInstance(instance)
		return instance
	
	def shutdownVm(self, instanceId):
		instance = self.data.acquireInstance(instanceId)
		self.stateTransition(instance, InstanceState.Running, InstanceState.ShuttingDown)
		self.data.releaseInstance(instance)
		hostname = self.data.getHost(instance.hostId).name
		try:
			self.proxy[hostname].shutdownVm(instance.vmId)
		except Exception:
			self.log.exception('shutdownVm failed for host %s vmId %d' % (instance.name, instance.vmId))
			raise
		return
	
	def destroyVm(self, instanceId):
		instance = self.data.acquireInstance(instanceId)
		if (instance.state is InstanceState.Pending or instance.state is InstanceState.Held):
			self.data.removeInstance(instance)
		elif (instance.state is InstanceState.Activating):
			self.stateTransition(instance, InstanceState.Activating, InstanceState.Destroying)
			self.data.releaseInstance(instance)
		else:
			# XXXstroucki: This is a problem with keeping
			# clean state.
			self.stateTransition(instance, None, InstanceState.Destroying)
			if instance.hostId is None:
				self.data.removeInstance(instance)
			else:
				hostname = self.data.getHost(instance.hostId).name
				try:
					if hostname is not None:
						self.proxy[hostname].destroyVm(instance.vmId)
						self.data.releaseInstance(instance)
				except:
					self.log.exception('destroyVm failed on host %s vmId %s' % (hostname, str(instance.vmId)))
					self.data.removeInstance(instance)


		return
	
	def suspendVm(self, instanceId):
		instance = self.data.acquireInstance(instanceId)
		self.stateTransition(instance, InstanceState.Running, InstanceState.Suspending)
		self.data.releaseInstance(instance)
		hostname = self.data.getHost(instance.hostId).name
		destination = "suspend/%d_%s" % (instance.id, instance.name)
		try:
			self.proxy[hostname].suspendVm(instance.vmId, destination)
		except:
			self.log.exception('suspendVm failed for host %s vmId %d' % (hostname, instance.vmId))
			raise TashiException(d={'errno':Errors.UnableToSuspend, 'msg':'Failed to suspend %s' % (instance.name)})
		return
	
	def resumeVm(self, instanceId):
		instance = self.data.acquireInstance(instanceId)
		self.stateTransition(instance, InstanceState.Suspended, InstanceState.Pending)
		source = "suspend/%d_%s" % (instance.id, instance.name)
		instance.hints['__resume_source'] = source
		self.data.releaseInstance(instance)
		return instance
	
	def migrateVm(self, instanceId, targetHostId):
		instance = self.data.acquireInstance(instanceId)
		try:
			# FIXME: should these be acquire/release host?
			targetHost = self.data.getHost(targetHostId)
			sourceHost = self.data.getHost(instance.hostId)
			# FIXME: Are these the correct state transitions?
		except:
			self.data.releaseInstance(instance)
			raise
		from pprint import pformat
		self.log.info("A: %s" % (pformat(instance)))
		self.stateTransition(instance, InstanceState.Running, InstanceState.MigratePrep)
		self.data.releaseInstance(instance)
		try:
			# Prepare the target
			self.log.info("migrateVm: Calling prepSourceVm on source host %s" % sourceHost.name)
			self.proxy[sourceHost.name].prepSourceVm(instance.vmId)
			self.log.info("migrateVm: Calling prepReceiveVm on target host %s" % targetHost.name)
			cookie = self.proxy[targetHost.name].prepReceiveVm(instance, sourceHost)
			self.log.info("Debug: here")
		except Exception, e:
			self.log.exception('prepReceiveVm failed')
			raise
		instance = self.data.acquireInstance(instance.id)
		self.log.info("B: %s" % (pformat(instance)))
		self.stateTransition(instance, InstanceState.MigratePrep, InstanceState.MigrateTrans)
		self.log.info("C: %s" % (pformat(instance)))
		self.data.releaseInstance(instance)
		try:
			# Send the VM
			self.proxy[sourceHost.name].migrateVm(instance.vmId, targetHost, cookie)
		except Exception, e:
			self.log.exception('migrateVm failed')
			raise
		try:
			instance = self.data.acquireInstance(instance.id)
			instance.hostId = targetHost.id
		finally:
			self.data.releaseInstance(instance)

		try:
			# Notify the target
			vmId = self.proxy[targetHost.name].receiveVm(instance, cookie)
			self.log.info("D: %s notified" % targetHost.name)
		except Exception, e:
			self.log.exception('receiveVm failed')
			raise
		return
	
	def pauseVm(self, instanceId):
		instance = self.data.acquireInstance(instanceId)
		self.stateTransition(instance, InstanceState.Running, InstanceState.Pausing)
		self.data.releaseInstance(instance)
		hostname = self.data.getHost(instance.hostId).name
		try:
			self.proxy[hostname].pauseVm(instance.vmId)
		except Exception:
			self.log.exception('pauseVm failed on host %s with vmId %d' % (hostname, instance.vmId))
			raise
		instance = self.data.acquireInstance(instanceId)
		self.stateTransition(instance, InstanceState.Pausing, InstanceState.Paused)
		self.data.releaseInstance(instance)
		return

	def unpauseVm(self, instanceId):
		instance = self.data.acquireInstance(instanceId)
		self.stateTransition(instance, InstanceState.Paused, InstanceState.Unpausing)
		self.data.releaseInstance(instance)
		hostname = self.data.getHost(instance.hostId).name
		try:
			self.proxy[hostname].unpauseVm(instance.vmId)
		except Exception:
			self.log.exception('unpauseVm failed on host %s with vmId %d' % (hostname, instance.vmId))
			raise
		instance = self.data.acquireInstance(instanceId)
		self.stateTransition(instance, InstanceState.Unpausing, InstanceState.Running)
		self.data.releaseInstance(instance)
		return
	
	def getHosts(self):
		return self.data.getHosts().values()
	
	def getNetworks(self):
		return self.data.getNetworks().values()
	
	def getUsers(self):
		return self.data.getUsers().values()
	
	def getInstances(self):
		return self.data.getInstances().values()
	
	def vmmSpecificCall(self, instanceId, arg):
		instance = self.data.getInstance(instanceId)
		hostname = self.data.getHost(instance.hostId).name
		try:
			res = self.proxy[hostname].vmmSpecificCall(instance.vmId, arg)
		except Exception:
			self.log.exception('vmmSpecificCall failed on host %s with vmId %d' % (hostname, instance.vmId))
			raise
		return res
	
#	@timed
	def registerNodeManager(self, host, instances):
		"""Called by the NM every so often as a keep-alive/state polling -- state changes here are NOT AUTHORITATIVE"""

		# Handle a new registration
		if (host.id == None):
			hostList = [h for h in self.data.getHosts().itervalues() if h.name == host.name]
			if (len(hostList) != 1):
				raise TashiException(d={'errno':Errors.NoSuchHost, 'msg':'A host with name %s is not identifiable' % (host.name)})
			host.id = hostList[0].id

		# Check if remote host information matches mine
		oldHost = self.data.acquireHost(host.id)
		if (oldHost.name != host.name):
			self.data.releaseHost(oldHost)
			raise TashiException(d={'errno':Errors.NoSuchHostId, 'msg':'Host id and hostname mismatch'})

		if oldHost.up == False:
			self.__upHost(oldHost)
		self.hostLastContactTime[host.id] = time.time()
		#self.hostLastUpdateTime[host.id] = time.time()
		oldHost.version = host.version
		oldHost.memory = host.memory
		oldHost.cores = host.cores

		# compare whether CM / NM versions are compatible
		if (host.version != version and not self.allowMismatchedVersions):
			oldHost.state = HostState.VersionMismatch
		if (host.version == version and oldHost.state == HostState.VersionMismatch):
			oldHost.state = HostState.Normal

		# let the host communicate what it is running
		for instance in instances:
			self.log.info('Accounting: id %d host %d vmId %d user %d cores %d memory %d' % (instance.id, host.id, instance.vmId, instance.userId, instance.cores, instance.memory))
			self.instanceLastContactTime.setdefault(instance.id, 0)

		self.data.releaseHost(oldHost)
		return host.id
	
	def vmUpdate(self, instanceId, instance, oldState):
		try:
			oldInstance = self.data.acquireInstance(instanceId)
		except TashiException, e:
			# shouldn't have a lock to clean up after here
			if (e.errno == Errors.NoSuchInstanceId):
				self.log.warning('Got vmUpdate for unknown instanceId %d' % (instanceId))
				return
		except:
			self.log.exception("Could not acquire instance")
			raise

		self.instanceLastContactTime[instanceId] = time.time()
		oldInstance.decayed = False

		if (instance.state == InstanceState.Exited):
			# determine why a VM has exited
			hostname = self.data.getHost(oldInstance.hostId).name
			if (oldInstance.state not in [InstanceState.ShuttingDown, InstanceState.Destroying, InstanceState.Suspending]):
				self.log.warning('Unexpected exit on %s of instance %s (vmId %d)' % (hostname, oldInstance.name, oldInstance.vmId))
			if (oldInstance.state == InstanceState.Suspending):
				self.stateTransition(oldInstance, InstanceState.Suspending, InstanceState.Suspended)
				oldInstance.hostId = None
				oldInstance.vmId = None
				self.data.releaseInstance(oldInstance)
			else:
				del self.instanceLastContactTime[oldInstance.id]
				self.data.removeInstance(oldInstance)
		else:
			if (instance.state):
				# XXXstroucki does this matter?
				if (oldState and oldInstance.state != oldState):
					self.log.warning('Got vmUpdate of state from %s to %s, but the instance was previously %s' % (vmStates[oldState], vmStates[instance.state], vmStates[oldInstance.state]))
				oldInstance.state = instance.state
			if (instance.vmId):
				oldInstance.vmId = instance.vmId
			if (instance.hostId):
				oldInstance.hostId = instance.hostId
			if (instance.nics):
				for nic in instance.nics:
					if (nic.ip):
						for oldNic in oldInstance.nics:
							if (oldNic.mac == nic.mac):
								oldNic.ip = nic.ip

			self.data.releaseInstance(oldInstance)

		return "success"
	
	def activateVm(self, instanceId, host):
		dataHost = self.data.acquireHost(host.id)

		if (dataHost.name != host.name):
			self.data.releaseHost(dataHost)
			raise TashiException(d={'errno':Errors.HostNameMismatch,'msg':"Mismatched target host"})
		if (not dataHost.up):
			self.data.releaseHost(dataHost)
			raise TashiException(d={'errno':Errors.HostNotUp,'msg':"Target host is not up"})
		if (dataHost.state != HostState.Normal):
			self.data.releaseHost(dataHost)
			raise TashiException(d={'errno':Errors.HostStateError,'msg':"Target host state is not normal"})

		self.data.releaseHost(dataHost)
		instance = self.data.acquireInstance(instanceId)

		if ('__resume_source' in instance.hints):
			self.stateTransition(instance, InstanceState.Pending, InstanceState.Resuming)
		else:
			# XXXstroucki should held VMs be continually tried? Or be explicitly set back to pending?
			#self.stateTransition(instance, InstanceState.Pending, InstanceState.Activating)
			self.stateTransition(instance, None, InstanceState.Activating)

		instance.hostId = host.id
		self.data.releaseInstance(instance)

		try:
			if ('__resume_source' in instance.hints):
				vmId = self.proxy[host.name].resumeVm(instance, instance.hints['__resume_source'])
			else:
				vmId = self.proxy[host.name].instantiateVm(instance)
		except Exception, e:
			instance = self.data.acquireInstance(instanceId)
			if (instance.state is InstanceState.Destroying): # Special case for if destroyVm is called during initialization and initialization fails
				self.data.removeInstance(instance)
			else:
				# XXXstroucki what can we do about pending hosts in the scheduler?
				# put them at the end of the queue and keep trying?
				self.stateTransition(instance, None, InstanceState.Held)
				instance.hostId = None
				self.data.releaseInstance(instance)
			return "failure"

		instance = self.data.acquireInstance(instanceId)
		instance.vmId = vmId

		if (instance.state is InstanceState.Destroying): # Special case for if destroyVm is called during initialization
			try:
				self.proxy[host.name].destroyVm(vmId)
				self.data.removeInstance(instance)
			except Exception:
				self.log.exception('destroyVm failed for host %s vmId %d' % (host.name, instance.vmId))
				self.data.releaseInstance(instance)
				return "failure"
		else:
			if ('__resume_source' not in instance.hints):
				# XXXstroucki should we just wait for NM to update?
				#self.stateTransition(instance, InstanceState.Activating, InstanceState.Running)
				pass

		self.data.releaseInstance(instance)
		return "success"

        def registerHost(self, hostname, memory, cores, version):
                hostId, alreadyRegistered = self.data.registerHost(hostname, memory, cores, version)
                if alreadyRegistered:
                        self.log.info("Host %s is already registered, it was updated now" % hostname)
                else:
                        self.log.info("A host was registered - hostname: %s, version: %s, memory: %s, cores: %s" % (hostname, version, memory, cores))
                return hostId

        def unregisterHost(self, hostId):
                self.data.unregisterHost(hostId)
                self.log.info("Host %s was unregistered" % hostId)
                return