+
+ def do_wait_nodes(self):
+ for guid, node in self._elements.iteritems():
+ if isinstance(node, self._node.Node):
+ # Just inject configuration stuff
+ node.home_path = "nepi-node-%s" % (guid,)
+ node.ident_path = self.sliceSSHKey
+ node.slicename = self.slicename
+
+ # Show the magic
+ self._logger.info("PlanetLab Node %s configured at %s", guid, node.hostname)
+
+ try:
+ for guid, node in self._elements.iteritems():
+ if isinstance(node, self._node.Node):
+ self._logger.info("Waiting for Node %s configured at %s", guid, node.hostname)
+
+ node.wait_provisioning(
+ (20*60 if node._node_id in self._just_provisioned else 60)
+ )
+
+ self._logger.info("READY Node %s at %s", guid, node.hostname)
+
+ # Prepare dependency installer now
+ node.prepare_dependencies()
+ except self._node.UnresponsiveNodeError:
+ # Uh...
+ self._logger.warn("UNRESPONSIVE Node %s", node.hostname)
+
+ # Mark all dead nodes (which are unresponsive) on the blacklist
+ # and re-raise
+ for guid, node in self._elements.iteritems():
+ if isinstance(node, self._node.Node):
+ if not node.is_alive():
+ self._logger.warn("Blacklisting %s for unresponsiveness", node.hostname)
+ self._blacklist.add(node._node_id)
+ node.unassign_node()
+
+ try:
+ self._save_blacklist()
+ except:
+ # not important...
+ import traceback
+ traceback.print_exc()
+
+ raise
+
+ def do_spanning_deployment_plan(self):
+ # Create application groups by collecting all applications
+ # based on their hash - the hash should contain everything that
+ # defines them and the platform they're built
+
+ def dephash(app):
+ return (
+ frozenset((app.depends or "").split(' ')),
+ frozenset((app.sources or "").split(' ')),
+ app.build,
+ app.install,
+ app.node.architecture,
+ app.node.operatingSystem,
+ app.node.pl_distro,
+ )
+
+ depgroups = collections.defaultdict(list)
+
+ for element in self._elements.itervalues():
+ if isinstance(element, self._app.Dependency):
+ depgroups[dephash(element)].append(element)
+ elif isinstance(element, self._node.Node):
+ deps = element._yum_dependencies
+ if deps:
+ depgroups[dephash(deps)].append(deps)
+
+ # Set up spanning deployment for those applications that
+ # have been deployed in several nodes.
+ for dh, group in depgroups.iteritems():
+ if len(group) > 1:
+ # Pick root (deterministically)
+ root = min(group, key=lambda app:app.node.hostname)
+
+ # Obtain all IPs in numeric format
+ # (which means faster distance computations)
+ for dep in group:
+ dep._ip = socket.gethostbyname(dep.node.hostname)
+ dep._ip_n = struct.unpack('!L', socket.inet_aton(dep._ip))[0]
+
+ # Compute plan
+ # NOTE: the plan is an iterator
+ plan = mst.mst(
+ group,
+ lambda a,b : ipaddr2.ipdistn(a._ip_n, b._ip_n),
+ root = root,
+ maxbranching = 2)
+
+ # Re-sign private key
+ try:
+ tempprk, temppuk, tmppass = self._make_temp_private_key()
+ except TempKeyError:
+ continue
+
+ # Set up slaves
+ plan = list(plan)
+ for slave, master in plan:
+ slave.set_master(master)
+ slave.install_keys(tempprk, temppuk, tmppass)
+
+ # We don't need the user's passphrase anymore
+ self.sliceSSHKeyPass = None
+
+ def _make_temp_private_key(self):
+ # Get the user's key's passphrase
+ if not self.sliceSSHKeyPass:
+ if 'SSH_ASKPASS' in os.environ:
+ proc = subprocess.Popen(
+ [ os.environ['SSH_ASKPASS'],
+ "Please type the passphrase for the %s SSH identity file. "
+ "The passphrase will be used to re-cipher the identity file with "
+ "a random 256-bit key for automated chain deployment on the "
+ "%s PlanetLab slice" % (
+ os.path.basename(self.sliceSSHKey),
+ self.slicename
+ ) ],
+ stdin = open("/dev/null"),
+ stdout = subprocess.PIPE,
+ stderr = subprocess.PIPE)
+ out,err = proc.communicate()
+ self.sliceSSHKeyPass = out.strip()
+
+ if not self.sliceSSHKeyPass:
+ raise TempKeyError
+
+ # Create temporary key files
+ prk = tempfile.NamedTemporaryFile(
+ dir = self.root_directory,
+ prefix = "pl_deploy_tmpk_",
+ suffix = "")
+
+ puk = tempfile.NamedTemporaryFile(
+ dir = self.root_directory,
+ prefix = "pl_deploy_tmpk_",
+ suffix = ".pub")
+
+ # Create secure 256-bits temporary passphrase
+ passphrase = ''.join(map(chr,[rng.randint(0,255)
+ for rng in (random.SystemRandom(),)
+ for i in xrange(32)] )).encode("hex")
+
+ # Copy keys
+ oprk = open(self.sliceSSHKey, "rb")
+ opuk = open(self.sliceSSHKey+".pub", "rb")
+ shutil.copymode(oprk.name, prk.name)
+ shutil.copymode(opuk.name, puk.name)
+ shutil.copyfileobj(oprk, prk)
+ shutil.copyfileobj(opuk, puk)
+ prk.flush()
+ puk.flush()
+ oprk.close()
+ opuk.close()
+
+ # A descriptive comment
+ comment = "%s#NEPI_INTERNAL@%s" % (self.authUser, self.slicename)
+
+ # Recipher keys
+ proc = subprocess.Popen(
+ ["ssh-keygen", "-p",
+ "-f", prk.name,
+ "-P", self.sliceSSHKeyPass,
+ "-N", passphrase,
+ "-C", comment ],
+ stdout = subprocess.PIPE,
+ stderr = subprocess.PIPE,
+ stdin = subprocess.PIPE
+ )
+ out, err = proc.communicate()
+
+ if err:
+ raise RuntimeError, "Problem generating keys: \n%s\n%r" % (
+ out, err)
+
+ prk.seek(0)
+ puk.seek(0)
+
+ # Change comment on public key
+ puklines = puk.readlines()
+ puklines[0] = puklines[0].split(' ')
+ puklines[0][-1] = comment+'\n'
+ puklines[0] = ' '.join(puklines[0])
+ puk.seek(0)
+ puk.truncate()
+ puk.writelines(puklines)
+ del puklines
+ puk.flush()
+
+ return prk, puk, passphrase
+