Changeset 27

Timestamp:

25/11/07 20:50:23 (4 years ago)

Author:

daedalus

Message:

* Added 'authoritarian' mode, which allows you permit/deny each and every command

that will be issued to a remote device. Designed for testing new change scripts.

* Added a 'command_timeout' parameter to Command Provisioners, so you can override

the default command timeout of 300 seconds by setting an attribute in the config
file

* Added proper handling of the 'onfail: continue' idea. You can now specify that

if a change fails, and it is a pre-requisite for some other change, processing
will continue despite the failure.

* Added the onfail: retry feature. If a change fails, you can specify that it should

be retried. You can also set a max number of retries. The default is 3.

Location:

trunk

Files:

• change.py (modified) (8 diffs)
• confloader.py (modified) (15 diffs)
• debug.py (modified) (1 diff)
• etc/netapp-provision-demo.xml (modified) (3 diffs)
• modipy.py (modified) (2 diffs)
• options.py (modified) (3 diffs)
• provisioner.py (modified) (19 diffs)
• test_ssh.py (modified) (1 diff)
• util.py (modified) (1 diff)

trunk/change.py

@@ -61 +61 @@
 import debug
 
-log = logging.getLogger('configulator')
+log = logging.getLogger('modipy')
 
 from zope.interface import Interface, implements
@@ -74 +74 @@
     'backout_ok': 4,
     'backout_failed': 5,
+    'retry': 6,
     }
 
@@ -86 +87 @@
     """
 
-    def __init__(self, name, devices=[], serial_mode=True, backout_all=False, continue_on_fail=False, **kwargs):
+    def __init__(self, name, devices=[], serial_mode=True, backout_all=False, on_fail_continue=False, **kwargs):
         """
         @param devices: a list of devices that this change should be applied to.
@@ -95 +96 @@
         L{Change} to a device fails, the L{Change} will be backed out for all other
         devices that have so far succeeded. This makes a L{Change} atomic.
-        @param continue_on_fail: Boolean. If set to True, if application of a
+        @param on_fail_continue: Boolean. If set to True, if application of a
         L{Change} to a device fails, the change will be backed out from that device,
         but the system will continue attempting to apply the change to the rest
@@ -175 +176 @@
     implements( IChange, )
 
-    def __init__(self, name, devices=[], serial_mode=True, backout_all=False, continue_on_fail=False,
+    def __init__(self, name, devices=[], serial_mode=True, backout_all=False,
+                 on_fail_continue=False,
+                 on_fail_retry=False,
+                 max_retries=3,
                  pre_impl=None,
                  impl=None,
@@ -191 +195 @@
         self.serial_mode = serial_mode
         self.backout_all = backout_all
-        self.continue_on_fail = continue_on_fail
-
+        self.on_fail_continue = on_fail_continue
+        self.on_fail_retry = on_fail_retry
+        self.max_retries = max_retries
+        self.retries = 0
         self.pre_impl = pre_impl
         self.impl = impl
@@ -206 +212 @@
         pass
 
-    def __str__(self):
+    def __repr__(self):
         return '<%s: %s>' % ( self.__class__, self.name )
 
@@ -427 +433 @@
 
         namespace = self.namespace
+
+    def can_retry(self):
+        """
+        Check to see if this change is allowed to retry after failing.
+        """
+        if self.on_fail_retry:
+            log.debug("Onfail-retry is enabled")
+            if self.retries < self.max_retries:
+                log.debug("Retries: %d is less than retry max: %s. Will retry change.", self.retries, self.max_retries)
+                self.retries += 1
+                return True
+            else:
+                log.debug("Too many retries for change. Will not retry again.")
+            pass
+        return False
         
 if __name__ == '__main__':

trunk/confloader.py

@@ -22 +22 @@
 import debug
 
-log = logging.getLogger('configulator')
+log = logging.getLogger('modipy')
 
 xinclude_re = re.compile(r'.*<xi:include href=[\'\"](?P<uri>.*)[\'\"].*')
@@ -78 +78 @@
 
     def update(self, dict):
-        log.debug("Updating a namespace!")
+        for key in dict:
+            self.namespace[key] = dict[key]
 
     def __iter__(self):
@@ -102 +103 @@
                 return self.namespace.next()
             except AttributeError:
-                log.debug("i have no parent!")
                 raise StopIteration
             
@@ -114 +114 @@
         return items
 
-
 class ConfigLoader:
     """
@@ -121 +120 @@
     """
 
-    def __init__(self, configfile=None, devices=[]):
+    def __init__(self, options=None, devices=[]):
 
         self.doc = None
+
+        self.options = options
 
         self.provisioners = {}
@@ -138 +139 @@
         log.debug("created ConfigLoader")
 
-        if configfile:
-            self.parse(configfile)
+        if options.configfile:
+            self.parse(options.configfile)
             pass
 
@@ -196 +197 @@
 ##             self.dependency_tree = element
             
-    def parse(self, configfile=None):
+    def parse(self, configfile):
         """
         Parse my configuration file.
         """
-        if configfile is None:
-            configfile = self.configfile
-            pass
-        
         self.tree = etree.parse(configfile)
 
@@ -291 +288 @@
         prov_klass = node.attrib['type']
         prov_name = node.attrib['name']
+
+        # Copy the attribs into a dictionary for use as kwargs
+        kwargs = {}
+        for key in node.attrib:
+            if key not in ['type', 'name']:
+                kwargs[key] = node.attrib[key]
+
         log.debug("Attempting to create a '%s' provisioner", prov_klass)
         prov_module = __import__('provisioner')
         klass = getattr( prov_module, prov_klass )
-        provisioner = klass()
+        try:
+            provisioner = klass(prov_name, authoritarian=self.options.authoritarian, **kwargs)
+        except TypeError, e:
+            log.error("Incorrect parameter supplied for provisioner of type '%s'", prov_klass)
+            raise e
 
         log.debug("created provisioner '%s': %s", prov_name, provisioner)
@@ -334 +342 @@
         change_klass = node.attrib['type']
         change_name = node.attrib['name']
+
         log.debug("Attempting to create a '%s' change", change_klass)
 
@@ -341 +350 @@
             log.info("change module not yet imported. Importing...")
             change_module = __import__('change')
-            
-        klass = getattr( change_module, change_klass )
-
+            pass
+
+        klass = getattr(change_module, change_klass)
         change = klass(change_name)
 
@@ -451 +460 @@
                 log.error("Iterator '%s' is not defined", itername)
                 raise
-        
+
+        # Add an optional onfail mode
+        log.debug("Checking for 'onfail' attribute...")
+        try:
+            onfail = node.attrib['onfail']
+            log.debug("onfail node found.")
+            if onfail == 'continue':
+                change.on_fail_continue = True
+                log.debug("Processing will continue if this change fails")
+
+            elif onfail == 'retry':
+                change.on_fail_retry = True
+                log.debug("This change will retry on failure")
+                try:
+                    max_retries = int(node.attrib['max_retries'])
+                    change.max_retries = max_retries
+                except KeyError:
+                    pass
+                log.debug("  This change will retry at most %d times", change.max_retries)
+        except KeyError:
+            pass
+
         # If the change doesn't have a provisioner, use the first one
         # in our config by default.
@@ -497 +527 @@
         log.debug("pending changes: %s", self.pending_changes)
         for change in self.pending_changes:
+            log.debug("testing change %s", change)
             if len(change.pre_requisites) == 0:
                 changelist.append(change)
-                log.debug("change '%s' has no pre-reqs. Adding.", change)
+                log.debug("change '%s' has no pre-reqs. Adding to execution queue.", change)
                 continue
+
             else:
                 all_prereqs = True
-                log.debug("checking change pre-requsisites: %s", change.pre_requisites)
                 for prereq in change.pre_requisites:
                     log.debug("change '%s' has a pre-req of '%s'", change, prereq)
                     if prereq not in self.change_success:
-                        log.debug("pre-req '%s' not complete yet. skipping", prereq)
+
+                        # If a prereq has failed, doesn't need to retry, and is
+                        # marked as 'onfail:continue', then we treat it as if
+                        # this prereq has been met.
+                        if prereq.state not in [ CHANGE_STATE['pending'],
+                                                 CHANGE_STATE['retry'],
+                                                 ] and prereq.on_fail_continue:
+                            log.debug("prereq failed, but is marked onfail:continue.")
+                            continue
+
+                        log.debug("pre-req '%s' not complete yet.", prereq)
                         all_prereqs = False
                         break
                     pass
 
-                log.debug("All pre-reqs for '%s' have completed. Adding.", change)
                 if all_prereqs:
+                    log.debug("All pre-reqs for '%s' have completed. Adding to execution queue.", change)
                     changelist.append(change)
                 else:
@@ -547 +588 @@
             self.backout_failure.append(change)
             self.pending_changes.remove(change)
-            
+
+        elif change.state == CHANGE_STATE['retry']:
+            # Change will be retried
+            pass
+        
         else:
             log.error("Unknown/unhandled change state '%s'", change.state)
@@ -647 +692 @@
             d = change.provisioner.perform_change(None, change, self.cfgldr.global_namespace)
             d.addCallback(self.change_complete, change)
-            d.addErrback(self.change_complete, change)
+            d.addErrback(self.change_failure, change)
             dlist.append(d)
             pass
@@ -662 +707 @@
         self.cfgldr.change_complete(change)
 
-    def change_failure(self, failure):
+    def change_failure(self, failure, change):
         log.error("Major change failure!")
         tlog.err(failure)
+        self.change_complete(failure, change)
 
     def print_stats(self, ignored):

trunk/debug.py

@@ -32 +32 @@
     handler = logging.handlers.RotatingFileHandler(filename=filename, maxBytes=10e6, backupCount=10)
     handler.setFormatter(formatter)
-    log = logging.getLogger('configulator')
+    log = logging.getLogger('modipy')
     log.addHandler(handler)
     

trunk/etc/netapp-provision-demo.xml

@@ -15 +15 @@
 <provisioner
   name='netapp_provisioner'
-  type='MultiConnectingProvisioner'>
+  type='MultiConnectingProvisioner'
+  command_timeout='5'>
 
   <command>ssh -i /home/daedalus/.ssh/configulator -o BatchMode=yes -o ServerAliveInterval=0 root@%(device.ipaddress)s "%(command.send)s"</command>
@@ -60 +61 @@
 </iterator>
 
-<change name="create_root_volume_primary" type="CommandChange">
+<change name="create_root_volume_primary" type="CommandChange" onfail='continue'>
 
   <target>%(primary_filer)s</target>
@@ -99 +100 @@
 </change>
 
-<change name="create_project_volume_primary" type="CommandChange" iterator="project_volumes">
+<change name="create_project_volume_primary" type="CommandChange" iterator="project_volumes"
+  onfail='retry' max_retries='2'>
 
   <prereq>create_root_volume_primary</prereq>

trunk/modipy.py

@@ -11 +11 @@
 
 import logging
-log = logging.getLogger('configulator')
+log = logging.getLogger('modipy')
 
 optparser = ChangeOptions()
@@ -17 +17 @@
 
 try:
-    cfgldr = ConfigLoader(optparser.options.configfile, optparser.args)
+    cfgldr = ConfigLoader(optparser.options, optparser.args)
 except Exception, e:
     log.error("Cannot load configuration: %s", e)

trunk/options.py

@@ -18 +18 @@
 from twisted.python import log as tlog
 
-log = logging.getLogger('configulator')
+log = logging.getLogger('modipy')
 
 class BaseOptions(optparse.OptionParser):
@@ -32 +32 @@
         help_license = 'Display the license agreement and exit.'
         help_debug = "Set the output debug level to: debug, info, warn, error, or critical."
-        help_logfile = "Log to specified file instead of default logfile"
-        help_no_logfile = "Disable logging to logfile"
+#        help_logfile = "Log to specified file instead of default logfile"
+#        help_no_logfile = "Disable logging to logfile"
 
+        
         self.add_option('', '--license',       dest='license', action='store_true', help=help_license)    
         self.add_option('', '--debug',         dest='debug', type='choice', choices=('debug', 'info', 'warn', 'error', 'critical'), metavar='LEVEL', default='info', help=help_debug)
-        self.add_option('', '--logfile',       dest='logfile', type='string', help=help_logfile)
-        self.add_option('', '--no-logfile',       dest='no-logfile', action='store_true', default=False, help=help_no_logfile)
-        
+#        self.add_option('', '--logfile',       dest='logfile', type='string', help=help_logfile)
+#        self.add_option('', '--no-logfile',       dest='no-logfile', action='store_true', default=False, help=help_no_logfile)
+
         self.addOptions()
 
@@ -75 +76 @@
 
     def addOptions(self):
+        help_authoritarian = "Authoritarian mode. Confirm every change command."
         help_configfile = "The configuration file to load"
-        #help_loadonly = "Load the configuration file and exit. Used to test parsing."
+        help_loadonly = "Load the configuration file and exit. Used to test parsing."
         
+
+        self.add_option('-a', '--authoritarian',  dest='authoritarian', action='store_true', default=False, help=help_authoritarian)        
         self.add_option('-c', '--configfile', dest='configfile', type='string', help=help_configfile)
-        #self.add_option('', '--loadonly', dest='loadonly', action='store_true', default=False, help=help_configfile)
+        self.add_option('', '--loadonly', dest='loadonly', action='store_true', default=False, help=help_loadonly)
 
     def check_values(self, options, args):

trunk/provisioner.py

@@ -17 +17 @@
 from twisted.internet import protocol
 from twisted.python import log as tlog
-from twisted.internet.error import ProcessDone, ProcessTerminated
+from twisted.internet.error import ProcessDone, ProcessTerminated, AlreadyCancelled, AlreadyCalled
 
 from change import ChangeFailed, ChangeConditionFailure, CHANGE_STATE
@@ -26 +26 @@
 
 import debug
-log = logging.getLogger('configulator')
+log = logging.getLogger('modipy')
 
 from twisted.internet.base import DelayedCall
@@ -63 +63 @@
     implements( IProvisioner, )
 
-    def __init__(self, name='', namespace={}):
+    def __init__(self, name='', namespace={}, authoritarian=False, **kwargs):
         self.name = name
         self.connectedDevice = None
         self.namespace = namespace
+        self.authoritarian = authoritarian
+
+    def parse_config_node(self, node):
+        pass
 
     def perform_change(self, ignored, change, namespace={}):
@@ -109 +113 @@
 
                 log.debug("applying change with namespace: %s", namespace)
-                
-##                 for key in self.namespace:
-##                     if namespace.has_key(key):
-##                         log.warning("Conflicting namespace keys: %s from %s conflicts with %s from global namespace" % (key, self, key) )
-##                         pass
-##                     namespace[key] = self.namespace[key]
-
-##                 # Then the change namespace overrides
-##                 for key in change.namespace:
-##                     if namespace.has_key(key):
-##                         log.warning("Conflicting namespace keys: %s from %s conflicts with %s from global namespace" % (key, change, key) )
-##                         pass
-##                     namespace[key] = change.namespace[key]
-                    
-##                 for key in device.namespace:
-##                     if namespace.has_key[key]:
-##                         log.warning("Conflicting namespace keys: %s from %s conflicts with %s from global namespace" % (key, device, key) )
-##                         pass
-##                     namespace[key] = device.namespace[key]
-                        
-                d.addCallback( self.apply_change, device, change, namespace)
+                d.addCallback(self.apply_change, device, change, namespace)
 
                 d.addCallback(self.change_complete_success, change, namespace)
@@ -175 +159 @@
         d = self.backout_change(None, device, change, namespace)
 
-        if change.continue_on_fail:
+        if change.on_fail_continue:
             log.info("Attempting to continue, despite failure...")
             change.set_state('partial_failure')
@@ -191 +175 @@
     def change_failure(self, failure, change, namespace):
         log.debug("Change failure for %s: %s", change.name, failure.value)
-        if change.state == CHANGE_STATE['pending']:
-            change.state = CHANGE_STATE['total_failure']
+        if change.state in [ CHANGE_STATE['pending'], CHANGE_STATE['retry'] ]:
+            # If the change is marked as 'on_fail: retry', let it retry
+            if change.can_retry():
+                change.state = CHANGE_STATE['retry']
+            else:
+                change.state = CHANGE_STATE['total_failure']
         else:
             change.state = CHANGE_STATE['backout_failed']
@@ -199 +187 @@
 
     def change_complete_success(self, result, change, namespace):
-        if change.state == CHANGE_STATE['pending']:
+        if change.state in [ CHANGE_STATE['pending'], CHANGE_STATE['retry'] ]:
             change.state = CHANGE_STATE['success']
             log.info("Change '%s' was a success!", change.name)
@@ -258 +246 @@
     command_re = re.compile(r'^(?P<start>.*)"(?P<quoted>.*)"(?P<end>.*)$')
 
+    def parse_config_node(self, node):
+        Provisioner.parse_config_node(self, node)
+        pass
+
+    def __init__(self, name='', namespace={}, authoritarian=False, command_timeout=300):
+        Provisioner.__init__(self, name, namespace, authoritarian)
+        self.command_timeout = int(command_timeout)
+
     def split_command(self, cmdstring):
         """
@@ -298 +294 @@
     device via STDIN, and receives the output from STDOUT and STDERR.
     """
+    def parse_config_node(self, node):
+        CommandProvisioner.parse_config_node(self, node)
 
     def connect(self, device):
@@ -327 +325 @@
         log.error("Connection to child timed out!")
         self.ep.transport.loseConnection()
-        self.childConnectingDefer.errback( ValueError("Connection timed out") )
+        self.childConnectingDefer.errback( Exception("Connection timed out") )
 
     def childConnectSuccess(self):
@@ -454 +452 @@
         # check the expr to see if we should process it
         if expr is None:
-            log.debug("No expression to wait for. Running command '%s' immediately." % cmdstring)
             if cmdstring is not None:
-                self.transport.writeToChild(0, '%s\n' % cmdstring)
-                self.data_wait_timeout = reactor.callLater(10, self.data_wait_too_long)
+                log.debug("No expression to wait for. Running command '%s' immediately." % cmdstring)
+                self.issue_command(cmdstring)
+            else:
+                log.debug("cmdstring is 'None', so nothing to do")
 
         else:
@@ -477 +476 @@
                 
                 if cmdstring is not None:
-                    log.debug("running command: %s" % cmdstring)
-                    self.transport.writeToChild(0, '%s\n' % cmdstring)
-                    self.data_wait_timeout = reactor.callLater(10, self.data_wait_too_long)
+                    self.issue_command(cmdstring)
                     pass
                 else:
@@ -494 +491 @@
             pass
 
+    def issue_command(self, cmdstring):
+        """
+        Write the command to the remote device.
+        """
+        # If we're in authoritarian mode, wait for confirmation
+        # that we should execute the command.
+        if self.parent.authoritarian:
+            log.debug("Authoritarian mode. Waiting for ok to proceed...")
+            isok = raw_input("Issue command (y/n)[n]?> ")
+            if isok.startswith('y'):
+                log.debug("Ok! Let's continue!")
+            else:
+                log.info(" Bailing out at your command.")
+                self.waitingForCommand.errback("User requested manual bailout.")
+                return
+            pass
+        log.debug("running command: %s" % cmdstring)                    
+        self.transport.writeToChild(0, '%s\n' % cmdstring)
+        self.data_wait_timeout = reactor.callLater(10, self.data_wait_too_long)
 
     def data_wait_too_long(self):
@@ -516 +532 @@
         Deals with additional configuration that I expect to be provided with.
         """
+        ConnectingProvisioner.parse_config_node(self, node)
         log.debug("Parsing config node: %s", node)
         # parse command definition
@@ -572 +589 @@
                 cmd = self.split_command(command)
                 log.debug("command is: %s", cmd)
-                d.addCallback(self.spawn_command, cmd)
+                d.addCallback(self.spawn_command, cmd, self.command_timeout)
                 d.addCallbacks(self.command_completed, self.command_failed)
                 pass
@@ -586 +603 @@
         will fire based on the results.
         """
+        log.debug("spawning command...")
+        self.waitingForCommand = defer.Deferred()
+        # If we're in authoritarian mode, wait for confirmation
+        # that we should execute the command.
+        log.debug("Checking for authoritarian mode...")
+        if self.authoritarian:
+            log.debug("Authoritarian mode. Waiting for ok to proceed...")
+            isok = raw_input("Issue command: %s\n(y/n)[n]?> " % ' '.join(cmd) )
+            if isok.startswith('y'):
+                log.debug("Ok! Let's continue!")
+            else:
+                log.info("Bailing out at your command.")
+                self.exitcode = -1
+                self.cmdoutput = 'User requested manual bailout'
+                self.waitingForCommand.errback( Exception("User requested manual bailout.") )
+                return self.waitingForCommand
+            pass
+
         log.debug("running command remotely: %s" % cmd)
-        self.waitingForCommand = defer.Deferred()
         self.ep = SingleCommandProtocol(self, timeout)
         self.p = reactor.spawnProcess(self.ep, cmd[0], cmd)
+        log.debug("Spawned process: %s", self.p)
+
         return self.waitingForCommand
 
@@ -602 +638 @@
         self.cmdoutput += result[1]
         log.debug("current results: exit '%d', output: %s", self.exitcode, self.cmdoutput)
-
+        log.error("Process: %s", self.p)
+        
     def command_failed(self, failure):
         errorstr = "%s: %s" % (self.exitcode, self.cmdoutput)
         log.error("Command failed: %s", errorstr)
-        self.all_commands_defer.errback( Exception(errorstr) )
+        log.error("Process: %s", self.p)
+        self.exitcode = result[0]
+        self.cmdoutput += result[1]
+        #self.all_commands_defer.errback( Exception(errorstr) )
 
     def all_commands_done(self, result):
@@ -650 +690 @@
 
     def processEnded(self, status_object):
-        #log.debug("process ended: %s" % status_object)
+        log.debug("process ended: %s" % status_object)
         try:
             if isinstance(status_object.value, ProcessDone):
@@ -659 +699 @@
             
             elif isinstance(status_object.value, ProcessTerminated):
-                log.debug("process did not exit ok: %s: %s" % (status_object.value.exitCode, self.databuf))
+                log.debug("process did not exit ok: %s: [%s: %s] %s" % (status_object.value.exitCode, status_object.value.signal, status_object.value.status, self.databuf))
                 self.exitCode = status_object.value.exitCode
                 self.timeout.cancel()
                 self.parent.waitingForCommand.callback( (self.exitCode, self.databuf) )
+                pass
                 
         except Exception, e:
-            log.error("erk: not good")
-            traceback.print_exc(e)
-            self.timeout.cancel()
-            self.parent.waitingForCommand.errback( e )
+            log.error("Exception raised during end of process processing. That's quite bad.")
+            raise
+##             traceback.print_exc(e)
+##             self.timeout.cancel()
+##             self.parent.waitingForCommand.errback( e )
 
     def timedOut(self):

trunk/test_ssh.py

@@ -11 +11 @@
 
 import logging
-log = logging.getLogger('configulator')
+log = logging.getLogger('modipy')
 log.setLevel(logging.DEBUG)
 

trunk/util.py

@@ -3 +3 @@
 #
 import logging
-log = logging.getLogger('configulator')
+log = logging.getLogger('modipy')
 
 def substituteVariables(str, namespace={}):