cluster: add batchStartup/Shutdown, cleanup

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,

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 ):
+    def batchStartup( cls, switches, **_kwargs ):
-        "Not implemented yet"
+        "Start up switches in per-server batches"
-        return []  # no switches started
+        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 ):
+    def batchShutdown( cls, switches, **_kwargs ):
-        "Not implemented yet"
+        "Stop switches in per-server batches"
-        return []  # no switchest stopped
+        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 = {}

mininet/clean.py

@@ -38,60 +38,80 @@ def killprocs( pattern ):
         else:
             break
 
-def cleanup():
+class Cleanup( object ):
-    """Clean up junk which might be left over from old runs;
+    "Wrapper for cleanup()"
-       do fast stuff before slow dp and link removal!"""
 
-    info("*** Removing excess controllers/ofprotocols/ofdatapaths/pings/noxes"
+    callbacks = []
-         "\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' )
 
-    # And kill off sudo mnexec
+    @classmethod
-    sh( 'pkill -9 -f "sudo mnexec"')
+    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" )
+        info("*** Removing excess controllers/ofprotocols/ofdatapaths/pings/noxes"
-    sh( 'rm -f /tmp/vconn* /tmp/vlogs* /tmp/*.out /tmp/*.log' )
+             "\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" )
+        # And kill off sudo mnexec
-    cleanUpScreens()
+        sh( 'pkill -9 -f "sudo mnexec"')
 
-    info( "*** Removing excess kernel datapaths\n" )
+        info( "*** Removing junk from /tmp\n" )
-    dps = sh( "ps ax | egrep -o 'dp[0-9]+' | sed 's/dp/nl:/'" ).splitlines()
+        sh( 'rm -f /tmp/vconn* /tmp/vlogs* /tmp/*.out /tmp/*.log' )
-    for dp in dps:
-        if dp:
-            sh( 'dpctl deldp ' + dp )
 
-    info( "***  Removing OVS datapaths" )
+        info( "*** Removing old X11 tunnels\n" )
-    dps = sh("ovs-vsctl --timeout=1 list-br").strip().splitlines()
+        cleanUpScreens()
-    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 all links of the pattern foo-ethX\n" )
+        info( "*** Removing excess kernel datapaths\n" )
-    links = sh( "ip link show | "
+        dps = sh( "ps ax | egrep -o 'dp[0-9]+' | sed 's/dp/nl:/'" ).splitlines()
-                "egrep -o '([-_.[:alnum:]]+-eth[[:digit:]]+)'" ).splitlines()
+        for dp in dps:
-    for link in links:
+            if dp:
-        if link:
+                sh( 'dpctl deldp ' + dp )
-            sh( "ip link del " + link )
 
-    info( "*** Killing stale mininet node processes\n" )
+        info( "***  Removing OVS datapaths" )
-    killprocs( 'mininet:' )
+        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" )
+        info( "*** Removing all links of the pattern foo-ethX\n" )
-    killprocs( 'Tunnel=Ethernet' )
+        links = sh( "ip link show | "
-    killprocs( '.ssh/mn')
+                    "egrep -o '([-_.[:alnum:]]+-eth[[:digit:]]+)'" ).splitlines()
-    sh( 'rm -f ~/.ssh/mn/*' )
+        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

mininet/node.py

@@ -157,8 +157,7 @@ class Node( object ):
                 break
             self.pollOut.poll()
         self.waiting = False
-        self.cmd( 'stty -echo' )
+        self.cmd( 'stty -echo; set +m' )
-        self.cmd( '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
+    return controller( name, **kwargs )