From acdcf9b6ae2ae46badcf04206acc6171b3f994af Mon Sep 17 00:00:00 2001 From: Bob Lantz Date: Tue, 27 Jan 2015 15:27:26 -0800 Subject: [PATCH] cluster: add batchStartup/Shutdown, cleanup --- bin/mn | 10 +++- examples/cluster.py | 97 +++++++++++++++++++++++++------------ mininet/clean.py | 114 ++++++++++++++++++++++++++------------------ mininet/node.py | 7 ++- 4 files changed, 145 insertions(+), 83 deletions(-) diff --git a/bin/mn b/bin/mn index bdb5ea3..d69b6b1 100755 --- a/bin/mn +++ b/bin/mn @@ -41,7 +41,8 @@ from functools import partial # Experimental! cluster edition prototype from mininet.examples.cluster import ( MininetCluster, RemoteHost, RemoteOVSSwitch, RemoteLink, - SwitchBinPlacer, RandomPlacer ) + SwitchBinPlacer, RandomPlacer, + ClusterCleanup ) from mininet.examples.clustercli import ClusterCLI PLACEMENT = { 'block': SwitchBinPlacer, 'random': RandomPlacer } @@ -281,6 +282,11 @@ class MininetRunner( object ): def begin( self ): "Create and run mininet." + if self.options.cluster: + servers = self.options.cluster.split( ',' ) + for server in servers: + ClusterCleanup.add( server ) + if self.options.clean: cleanup() exit() @@ -334,7 +340,7 @@ class MininetRunner( object ): warn( '*** WARNING: Experimental cluster mode!\n' '*** Using RemoteHost, RemoteOVSSwitch, RemoteLink\n' ) host, switch, link = RemoteHost, RemoteOVSSwitch, RemoteLink - Net = partial( MininetCluster, servers=cluster.split( ',' ), + Net = partial( MininetCluster, servers=servers, placement=PLACEMENT[ self.options.placement ] ) mn = Net( topo=topo, diff --git a/examples/cluster.py b/examples/cluster.py index 387c8b0..aa7ddfd 100755 --- a/examples/cluster.py +++ b/examples/cluster.py @@ -82,6 +82,7 @@ from mininet.topolib import TreeTopo from mininet.util import quietRun, errRun, retry from mininet.examples.clustercli import CLI from mininet.log import setLogLevel, debug, info, error +from mininet.clean import addCleanupCallback from signal import signal, SIGINT, SIG_IGN from subprocess import Popen, PIPE, STDOUT @@ -89,9 +90,51 @@ import os from random import randrange import sys import re - +from itertools import groupby +from operator import attrgetter from distutils.version import StrictVersion + +def findUser(): + "Try to return logged-in (usually non-root) user" + return ( + # If we're running sudo + os.environ.get( 'SUDO_USER', False ) or + # Logged-in user (if we have a tty) + ( quietRun( 'who am i' ).split() or [ False ] )[ 0 ] or + # Give up and return effective user + quietRun( 'whoami' ) ) + + +class ClusterCleanup( object ): + "Cleanup callback" + + inited = False + serveruser = {} + + @classmethod + def add( cls, server, user='' ): + "Add an entry to server: user dict" + if not cls.inited: + addCleanupCallback( cls.cleanup ) + if not user: + user = findUser() + cls.serveruser[ server ] = user + + @classmethod + def cleanup( cls ): + "Clean up" + info( '*** Cleaning up cluster\n' ) + for server, user in cls.serveruser.iteritems(): + if server == 'localhost': + # Handled by mininet.clean.cleanup() + continue + else: + cmd = [ 'su', user, '-c', + 'ssh %s@%s sudo mn -c' % ( user, server ) ] + info( cmd, '\n' ) + info( quietRun( cmd ) ) + # BL note: so little code is required for remote nodes, # we will probably just want to update the main Node() # class to enable it for remote access! However, there @@ -125,7 +168,8 @@ class RemoteMixin( object ): self.server = server if server else 'localhost' self.serverIP = ( serverIP if serverIP else self.findServerIP( self.server ) ) - self.user = user if user else self.findUser() + self.user = user if user else findUser() + ClusterCleanup.add( server=server, user=user ) if controlPath is True: # Set a default control path for shared SSH connections controlPath = '/tmp/mn-%r@%h:%p' @@ -148,17 +192,6 @@ class RemoteMixin( object ): self.shell, self.pid = None, None super( RemoteMixin, self ).__init__( name, **kwargs ) - @staticmethod - def findUser(): - "Try to return logged-in (usually non-root) user" - return ( - # If we're running sudo - os.environ.get( 'SUDO_USER', False ) or - # Logged-in user (if we have a tty) - ( quietRun( 'who am i' ).split() or [ False ] )[ 0 ] or - # Give up and return effective user - quietRun( 'whoami' ) ) - # Determine IP address of local host _ipMatchRegex = re.compile( r'\d+\.\d+\.\d+\.\d+' ) @@ -244,7 +277,7 @@ class RemoteMixin( object ): # Drop privileges cmd = [ 'sudo', '-E', '-u', self.user ] + cmd params.update( preexec_fn=self._ignoreSignal ) - debug( '_popen', ' '.join(cmd), params ) + debug( '_popen', cmd, '\n' ) popen = super( RemoteMixin, self )._popen( cmd, **params ) return popen @@ -257,13 +290,6 @@ class RemoteMixin( object ): kwargs.update( moveIntfFn=RemoteLink.moveIntf ) return super( RemoteMixin, self).addIntf( *args, **kwargs ) - def cleanup( self ): - "Help python collect its garbage." - # Intfs may end up in root NS - for intfName in self.intfNames(): - if self.name in intfName: - self.rcmd( 'ip link del ' + intfName ) - self.shell = None class RemoteNode( RemoteMixin, Node ): "A node on a remote server" @@ -282,7 +308,7 @@ class RemoteOVSSwitch( RemoteMixin, OVSSwitch ): def __init__( self, *args, **kwargs ): # No batch startup yet - kwargs.update( batch=False ) + kwargs.update( batch=True ) super( RemoteOVSSwitch, self ).__init__( *args, **kwargs ) def isOldOVS( self ): @@ -298,14 +324,24 @@ class RemoteOVSSwitch( RemoteMixin, OVSSwitch ): StrictVersion( '1.10' ) ) @classmethod - def batchStartup( cls, *_args, **_kwargs ): - "Not implemented yet" - return [] # no switches started - + def batchStartup( cls, switches, **_kwargs ): + "Start up switches in per-server batches" + for server, switchGroup in groupby( switches, attrgetter( 'server' ) ): + info( '(%s)' % server ) + group = tuple( switchGroup ) + switch = group[ 0 ] + OVSSwitch.batchStartup( group, run=switch.cmd ) + return switches + @classmethod - def batchShutdown( cls, *_args, **_kwargs ): - "Not implemented yet" - return [] # no switchest stopped + def batchShutdown( cls, switches, **_kwargs ): + "Stop switches in per-server batches" + for server, switchGroup in groupby( switches, attrgetter( 'server' ) ): + info( '(%s)' % server ) + group = tuple( switchGroup ) + switch = group[ 0 ] + OVSSwitch.batchShutdown( group, run=switch.rcmd ) + return switches class RemoteLink( Link ): @@ -325,6 +361,7 @@ class RemoteLink( Link ): def stop( self ): "Stop this link" + Link.stop( self ) if self.tunnel: self.tunnel.terminate() self.tunnel = None @@ -636,7 +673,7 @@ class MininetCluster( Mininet ): if not self.serverIP: self.serverIP = { server: RemoteMixin.findServerIP( server ) for server in self.servers } - self.user = params.pop( 'user', RemoteMixin.findUser() ) + self.user = params.pop( 'user', findUser() ) if params.pop( 'precheck' ): self.precheck() self.connections = {} diff --git a/mininet/clean.py b/mininet/clean.py index 4761721..707452e 100755 --- a/mininet/clean.py +++ b/mininet/clean.py @@ -38,60 +38,80 @@ def killprocs( pattern ): else: break -def cleanup(): - """Clean up junk which might be left over from old runs; - do fast stuff before slow dp and link removal!""" +class Cleanup( object ): + "Wrapper for cleanup()" - info("*** Removing excess controllers/ofprotocols/ofdatapaths/pings/noxes" - "\n") - zombies = 'controller ofprotocol ofdatapath ping nox_core lt-nox_core ' - zombies += 'ovs-openflowd ovs-controller udpbwtest mnexec ivs' - # Note: real zombie processes can't actually be killed, since they - # are already (un)dead. Then again, - # you can't connect to them either, so they're mostly harmless. - # Send SIGTERM first to give processes a chance to shutdown cleanly. - sh( 'killall ' + zombies + ' 2> /dev/null' ) - time.sleep( 1 ) - sh( 'killall -9 ' + zombies + ' 2> /dev/null' ) + callbacks = [] - # And kill off sudo mnexec - sh( 'pkill -9 -f "sudo mnexec"') + @classmethod + def cleanup( cls): + """Clean up junk which might be left over from old runs; + do fast stuff before slow dp and link removal!""" - info( "*** Removing junk from /tmp\n" ) - sh( 'rm -f /tmp/vconn* /tmp/vlogs* /tmp/*.out /tmp/*.log' ) + info("*** Removing excess controllers/ofprotocols/ofdatapaths/pings/noxes" + "\n") + zombies = 'controller ofprotocol ofdatapath ping nox_core lt-nox_core ' + zombies += 'ovs-openflowd ovs-controller udpbwtest mnexec ivs' + # Note: real zombie processes can't actually be killed, since they + # are already (un)dead. Then again, + # you can't connect to them either, so they're mostly harmless. + # Send SIGTERM first to give processes a chance to shutdown cleanly. + sh( 'killall ' + zombies + ' 2> /dev/null' ) + time.sleep( 1 ) + sh( 'killall -9 ' + zombies + ' 2> /dev/null' ) - info( "*** Removing old X11 tunnels\n" ) - cleanUpScreens() + # And kill off sudo mnexec + sh( 'pkill -9 -f "sudo mnexec"') - info( "*** Removing excess kernel datapaths\n" ) - dps = sh( "ps ax | egrep -o 'dp[0-9]+' | sed 's/dp/nl:/'" ).splitlines() - for dp in dps: - if dp: - sh( 'dpctl deldp ' + dp ) + info( "*** Removing junk from /tmp\n" ) + sh( 'rm -f /tmp/vconn* /tmp/vlogs* /tmp/*.out /tmp/*.log' ) - info( "*** Removing OVS datapaths" ) - dps = sh("ovs-vsctl --timeout=1 list-br").strip().splitlines() - if dps: - sh( "ovs-vsctl " + " -- ".join( "--if-exists del-br " + dp - for dp in dps if dp ) ) - # And in case the above didn't work... - dps = sh("ovs-vsctl --timeout=1 list-br").strip().splitlines() - for dp in dps: - sh( 'ovs-vsctl del-br ' + dp ) + info( "*** Removing old X11 tunnels\n" ) + cleanUpScreens() - info( "*** Removing all links of the pattern foo-ethX\n" ) - links = sh( "ip link show | " - "egrep -o '([-_.[:alnum:]]+-eth[[:digit:]]+)'" ).splitlines() - for link in links: - if link: - sh( "ip link del " + link ) + info( "*** Removing excess kernel datapaths\n" ) + dps = sh( "ps ax | egrep -o 'dp[0-9]+' | sed 's/dp/nl:/'" ).splitlines() + for dp in dps: + if dp: + sh( 'dpctl deldp ' + dp ) - info( "*** Killing stale mininet node processes\n" ) - killprocs( 'mininet:' ) + info( "*** Removing OVS datapaths" ) + dps = sh("ovs-vsctl --timeout=1 list-br").strip().splitlines() + if dps: + sh( "ovs-vsctl " + " -- ".join( "--if-exists del-br " + dp + for dp in dps if dp ) ) + # And in case the above didn't work... + dps = sh("ovs-vsctl --timeout=1 list-br").strip().splitlines() + for dp in dps: + sh( 'ovs-vsctl del-br ' + dp ) - info( "*** Shutting down stale tunnels\n" ) - killprocs( 'Tunnel=Ethernet' ) - killprocs( '.ssh/mn') - sh( 'rm -f ~/.ssh/mn/*' ) + info( "*** Removing all links of the pattern foo-ethX\n" ) + links = sh( "ip link show | " + "egrep -o '([-_.[:alnum:]]+-eth[[:digit:]]+)'" ).splitlines() + for link in links: + if link: + sh( "ip link del " + link ) - info( "*** Cleanup complete.\n" ) + info( "*** Killing stale mininet node processes\n" ) + killprocs( 'mininet:' ) + + info( "*** Shutting down stale tunnels\n" ) + killprocs( 'Tunnel=Ethernet' ) + killprocs( '.ssh/mn') + sh( 'rm -f ~/.ssh/mn/*' ) + + # Call any additional cleanup code if necessary + for callback in cls.callbacks: + callback() + + info( "*** Cleanup complete.\n" ) + + @classmethod + def addCleanupCallback( cls, callback ): + "Add cleanup callback" + if callback not in cls.callbacks: + cls.callbacks.append( callback ) + + +cleanup = Cleanup.cleanup +addCleanupCallback = Cleanup.addCleanupCallback diff --git a/mininet/node.py b/mininet/node.py index 0ccddff..3d459a3 100644 --- a/mininet/node.py +++ b/mininet/node.py @@ -157,8 +157,7 @@ class Node( object ): break self.pollOut.poll() self.waiting = False - self.cmd( 'stty -echo' ) - self.cmd( 'set +m' ) + self.cmd( 'stty -echo; set +m' ) def mountPrivateDirs( self ): "mount private directories" @@ -1270,7 +1269,7 @@ class OVSSwitch( Switch ): def batchShutdown( cls, switches, run=errRun ): "Shut down a list of OVS switches" delcmd = 'del-br %s' - if not cls.isOldOVS(): + if switches and not switches[ 0 ].isOldOVS(): delcmd = '--if-exists ' + delcmd # First, delete them all from ovsdb run( 'ovs-vsctl ' + @@ -1549,4 +1548,4 @@ def DefaultController( name, controllers=DefaultControllers, **kwargs ): controller = findController( controllers ) if not controller: raise Exception( 'Could not find a default OpenFlow controller' ) - return contr \ No newline at end of file + return controller( name, **kwargs )