planetstack/ec2_observer/syncstep.py

   1 import os
   2 import base64
   3 from datetime import datetime
   4 from planetstack.config import Config
   5 from util.logger import Logger, logging
   6 from observer.steps import *
   7
   8 logger = Logger(level=logging.INFO)
   9
  10 class FailedDependency(Exception):
  11     pass
  12
  13 class SyncStep:
  14     """ A PlanetStack Sync step.
  15
  16     Attributes:
  17         psmodel        Model name the step synchronizes
  18         dependencies    list of names of models that must be synchronized first if the current model depends on them
  19     """
  20     slow=False
  21     def get_prop(prop):
  22         try:
  23             sync_config_dir = Config().sync_config_dir
  24         except:
  25             sync_config_dir = '/etc/planetstack/sync'
  26         prop_config_path = '/'.join(sync_config_dir,self.name,prop)
  27         return open(prop_config_path).read().rstrip()
  28
  29     def __init__(self, **args):
  30         """Initialize a sync step
  31            Keyword arguments:
  32                    name -- Name of the step
  33                 provides -- PlanetStack models sync'd by this step
  34         """
  35         dependencies = []
  36         self.driver = args.get('driver')
  37         self.error_map = args.get('error_map')
  38
  39         try:
  40             self.soft_deadline = int(self.get_prop('soft_deadline_seconds'))
  41         except:
  42             self.soft_deadline = 5 # 5 seconds
  43
  44         return
  45
  46     def fetch_pending(self, deletion=False):
  47         # This is the most common implementation of fetch_pending
  48         # Steps should override it if they have their own logic
  49         # for figuring out what objects are outstanding.
  50         main_obj = self.provides[0]
  51         if (not deleted):
  52             objs = main_obj.objects.filter(Q(enacted__lt=F('updated')) | Q(enacted=None))
  53         else:
  54             objs = main_obj.deleted_objects.all()
  55
  56         return objs
  57         #return Sliver.objects.filter(ip=None)
  58
  59     def check_dependencies(self, obj, failed):
  60         for dep in self.dependencies:
  61             peer_name = dep[0].lower() + dep[1:]    # django names are camelCased with the first letter lower
  62             peer_object = getattr(obj, peer_name)
  63
  64             # peer_object can be None, and if so there
  65             # is no object-level dependency
  66             if (peer_object and peer_object.pk==failed.pk):
  67                 raise FailedDependency
  68
  69     def call(self, failed=[], deletion=False):
  70         pending = self.fetch_pending(deletion)
  71         for o in pending:
  72             try:
  73                 for f in failed:
  74                     self.check_dependencies(o,f) # Raises exception if failed
  75                 if (deletion):
  76                     self.delete_record(o)
  77                     o.delete(purge=True)
  78                 else:
  79                     self.sync_record(o)
  80                     o.enacted = datetime.now() # Is this the same timezone? XXX
  81                     o.backend_status = "OK"
  82                     o.save(update_fields=['enacted'])
  83             except Exception,e:
  84                 try:
  85                     o.backend_status = self.error_map.map(str(e))
  86                 except:
  87                     o.backend_status = str(e)
  88
  89                 if (o.pk):
  90                     o.save(update_fields=['backend_status'])
  91
  92                 logger.log_exc("sync step failed!")
  93                 failed.append(o)
  94
  95         return failed
  96
  97     def __call__(self, **args):
  98         return self.call(**args)