Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 1 | import os |
| 2 | import base64 |
Tony Mack | 4fa85fb | 2013-09-25 14:39:57 -0400 | [diff] [blame] | 3 | from datetime import datetime |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 4 | from planetstack.config import Config |
Andy Bavier | e7abb62 | 2013-10-18 15:11:56 -0400 | [diff] [blame] | 5 | from util.logger import Logger, logging |
Sapan Bhatia | eba0843 | 2014-04-28 23:58:36 -0400 | [diff] [blame] | 6 | from observer.steps import * |
Sapan Bhatia | d9468eb | 2014-08-20 03:03:12 -0400 | [diff] [blame] | 7 | from django.db.models import F, Q |
Sapan Bhatia | 082ee36 | 2015-02-08 06:35:36 +0000 | [diff] [blame] | 8 | from core.models import * |
Sapan Bhatia | 8a51bf5 | 2015-01-29 20:55:40 +0000 | [diff] [blame] | 9 | import json |
| 10 | import time |
| 11 | import pdb |
Andy Bavier | e7abb62 | 2013-10-18 15:11:56 -0400 | [diff] [blame] | 12 | |
Andy Bavier | 04111b7 | 2013-10-22 16:47:10 -0400 | [diff] [blame] | 13 | logger = Logger(level=logging.INFO) |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 14 | |
Sapan Bhatia | b7b1b6e | 2015-02-08 06:36:32 +0000 | [diff] [blame] | 15 | def f7(seq): |
| 16 | seen = set() |
| 17 | seen_add = seen.add |
| 18 | return [ x for x in seq if not (x in seen or seen_add(x))] |
| 19 | |
| 20 | def elim_dups(backend_str): |
| 21 | strs = backend_str.split(' // ') |
| 22 | strs2 = f7(strs) |
| 23 | return ' // '.join(strs2) |
| 24 | |
Sapan Bhatia | 082ee36 | 2015-02-08 06:35:36 +0000 | [diff] [blame] | 25 | def deepgetattr(obj, attr): |
| 26 | return reduce(getattr, attr.split('.'), obj) |
| 27 | |
Sapan Bhatia | 13c7f11 | 2013-09-02 14:19:35 -0400 | [diff] [blame] | 28 | class FailedDependency(Exception): |
Tony Mack | ce79de0 | 2013-09-24 10:12:33 -0400 | [diff] [blame] | 29 | pass |
Sapan Bhatia | 13c7f11 | 2013-09-02 14:19:35 -0400 | [diff] [blame] | 30 | |
Tony Mack | e10fbe5 | 2015-01-03 19:37:39 -0500 | [diff] [blame] | 31 | class SyncStep(object): |
Tony Mack | ce79de0 | 2013-09-24 10:12:33 -0400 | [diff] [blame] | 32 | """ A PlanetStack Sync step. |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 33 | |
Tony Mack | ce79de0 | 2013-09-24 10:12:33 -0400 | [diff] [blame] | 34 | Attributes: |
| 35 | psmodel Model name the step synchronizes |
| 36 | dependencies list of names of models that must be synchronized first if the current model depends on them |
| 37 | """ |
| 38 | slow=False |
| 39 | def get_prop(prop): |
| 40 | try: |
| 41 | sync_config_dir = Config().sync_config_dir |
| 42 | except: |
| 43 | sync_config_dir = '/etc/planetstack/sync' |
| 44 | prop_config_path = '/'.join(sync_config_dir,self.name,prop) |
| 45 | return open(prop_config_path).read().rstrip() |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 46 | |
Tony Mack | ce79de0 | 2013-09-24 10:12:33 -0400 | [diff] [blame] | 47 | def __init__(self, **args): |
| 48 | """Initialize a sync step |
| 49 | Keyword arguments: |
| 50 | name -- Name of the step |
| 51 | provides -- PlanetStack models sync'd by this step |
| 52 | """ |
| 53 | dependencies = [] |
Tony Mack | 387a73f | 2013-09-18 07:59:14 -0400 | [diff] [blame] | 54 | self.driver = args.get('driver') |
Sapan Bhatia | eba0843 | 2014-04-28 23:58:36 -0400 | [diff] [blame] | 55 | self.error_map = args.get('error_map') |
| 56 | |
Tony Mack | ce79de0 | 2013-09-24 10:12:33 -0400 | [diff] [blame] | 57 | try: |
| 58 | self.soft_deadline = int(self.get_prop('soft_deadline_seconds')) |
| 59 | except: |
| 60 | self.soft_deadline = 5 # 5 seconds |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 61 | |
Tony Mack | ce79de0 | 2013-09-24 10:12:33 -0400 | [diff] [blame] | 62 | return |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 63 | |
Sapan Bhatia | e17bc5b | 2014-04-30 00:53:06 -0400 | [diff] [blame] | 64 | def fetch_pending(self, deletion=False): |
Sapan Bhatia | 2176566 | 2014-07-23 08:59:30 -0400 | [diff] [blame] | 65 | # This is the most common implementation of fetch_pending |
| 66 | # Steps should override it if they have their own logic |
| 67 | # for figuring out what objects are outstanding. |
Sapan Bhatia | 39a775f | 2015-01-29 20:58:25 +0000 | [diff] [blame] | 68 | main_obj = self.observes |
Sapan Bhatia | d9468eb | 2014-08-20 03:03:12 -0400 | [diff] [blame] | 69 | if (not deletion): |
Sapan Bhatia | 2176566 | 2014-07-23 08:59:30 -0400 | [diff] [blame] | 70 | objs = main_obj.objects.filter(Q(enacted__lt=F('updated')) | Q(enacted=None)) |
| 71 | else: |
| 72 | objs = main_obj.deleted_objects.all() |
| 73 | |
| 74 | return objs |
Sapan Bhatia | ca2e21f | 2013-10-02 01:10:02 -0400 | [diff] [blame] | 75 | #return Sliver.objects.filter(ip=None) |
Tony Mack | ce79de0 | 2013-09-24 10:12:33 -0400 | [diff] [blame] | 76 | |
Sapan Bhatia | ca2e21f | 2013-10-02 01:10:02 -0400 | [diff] [blame] | 77 | def check_dependencies(self, obj, failed): |
Tony Mack | ce79de0 | 2013-09-24 10:12:33 -0400 | [diff] [blame] | 78 | for dep in self.dependencies: |
Scott Baker | 105b6b7 | 2014-05-12 10:40:25 -0700 | [diff] [blame] | 79 | peer_name = dep[0].lower() + dep[1:] # django names are camelCased with the first letter lower |
Sapan Bhatia | 082ee36 | 2015-02-08 06:35:36 +0000 | [diff] [blame] | 80 | |
Sapan Bhatia | cfef6ef | 2014-08-20 03:04:03 -0400 | [diff] [blame] | 81 | try: |
Sapan Bhatia | 082ee36 | 2015-02-08 06:35:36 +0000 | [diff] [blame] | 82 | peer_object = deepgetattr(obj, peer_name) |
| 83 | try: |
| 84 | peer_objects = peer_object.all() |
| 85 | except AttributeError: |
| 86 | peer_objects = [peer_object] |
Sapan Bhatia | cfef6ef | 2014-08-20 03:04:03 -0400 | [diff] [blame] | 87 | except: |
Sapan Bhatia | 082ee36 | 2015-02-08 06:35:36 +0000 | [diff] [blame] | 88 | peer_objects = [] |
Sapan Bhatia | cfef6ef | 2014-08-20 03:04:03 -0400 | [diff] [blame] | 89 | |
Sapan Bhatia | 082ee36 | 2015-02-08 06:35:36 +0000 | [diff] [blame] | 90 | if (failed in peer_objects): |
| 91 | if (obj.backend_status!=failed.backend_status): |
| 92 | obj.backend_status = failed.backend_status |
Sapan Bhatia | 7e482de | 2014-08-22 03:05:13 -0400 | [diff] [blame] | 93 | obj.save(update_fields=['backend_status']) |
| 94 | raise FailedDependency("Failed dependency for %s:%s peer %s:%s failed %s:%s" % (obj.__class__.__name__, str(obj.pk), peer_object.__class__.__name__, str(peer_object.pk), failed.__class__.__name__, str(failed.pk))) |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 95 | |
Sapan Bhatia | 6082336 | 2014-04-30 00:52:32 -0400 | [diff] [blame] | 96 | def call(self, failed=[], deletion=False): |
| 97 | pending = self.fetch_pending(deletion) |
Tony Mack | ce79de0 | 2013-09-24 10:12:33 -0400 | [diff] [blame] | 98 | for o in pending: |
Sapan Bhatia | 8a51bf5 | 2015-01-29 20:55:40 +0000 | [diff] [blame] | 99 | sync_failed = False |
Tony Mack | 68e818d | 2013-09-25 13:34:17 -0400 | [diff] [blame] | 100 | try: |
Sapan Bhatia | de9f24c | 2015-02-10 17:21:33 -0500 | [diff] [blame] | 101 | backoff_disabled = Config().observer_backoff_disabled |
Sapan Bhatia | 2535359 | 2015-02-10 17:16:07 -0500 | [diff] [blame] | 102 | except: |
| 103 | backoff_disabled = 0 |
| 104 | |
| 105 | try: |
Sapan Bhatia | 8a51bf5 | 2015-01-29 20:55:40 +0000 | [diff] [blame] | 106 | scratchpad = json.loads(o.backend_register) |
| 107 | if (scratchpad): |
| 108 | next_run = scratchpad['next_run'] |
Sapan Bhatia | 2535359 | 2015-02-10 17:16:07 -0500 | [diff] [blame] | 109 | if (not backoff_disabled and next_run>time.time()): |
Sapan Bhatia | 8a51bf5 | 2015-01-29 20:55:40 +0000 | [diff] [blame] | 110 | sync_failed = True |
| 111 | print "BACKING OFF, exponent = %d"%scratchpad['exponent'] |
| 112 | except: |
| 113 | pass |
| 114 | |
| 115 | if (not sync_failed): |
Sapan Bhatia | eba0843 | 2014-04-28 23:58:36 -0400 | [diff] [blame] | 116 | try: |
Sapan Bhatia | 8a51bf5 | 2015-01-29 20:55:40 +0000 | [diff] [blame] | 117 | for f in failed: |
| 118 | self.check_dependencies(o,f) # Raises exception if failed |
| 119 | if (deletion): |
| 120 | self.delete_record(o) |
| 121 | o.delete(purge=True) |
| 122 | else: |
| 123 | self.sync_record(o) |
| 124 | o.enacted = datetime.now() # Is this the same timezone? XXX |
| 125 | scratchpad = {'next_run':0, 'exponent':0} |
| 126 | o.backend_register = json.dumps(scratchpad) |
| 127 | o.backend_status = "1 - OK" |
| 128 | o.save(update_fields=['enacted','backend_status','backend_register']) |
| 129 | except Exception,e: |
| 130 | logger.log_exc("sync step failed!") |
Sapan Bhatia | 6d42215 | 2015-02-08 06:31:42 +0000 | [diff] [blame] | 131 | try: |
| 132 | if (o.backend_status.startswith('2 - ')): |
| 133 | str_e = '%s // %r'%(o.backend_status[4:],e) |
Sapan Bhatia | b7b1b6e | 2015-02-08 06:36:32 +0000 | [diff] [blame] | 134 | str_e = elim_dups(str_e) |
Sapan Bhatia | 6d42215 | 2015-02-08 06:31:42 +0000 | [diff] [blame] | 135 | else: |
| 136 | str_e = '%r'%e |
| 137 | except: |
| 138 | str_e = '%r'%e |
| 139 | |
Sapan Bhatia | 9c308fc | 2014-08-22 03:07:59 -0400 | [diff] [blame] | 140 | try: |
Sapan Bhatia | 8a51bf5 | 2015-01-29 20:55:40 +0000 | [diff] [blame] | 141 | o.backend_status = '2 - %s'%self.error_map.map(str_e) |
Sapan Bhatia | 9c308fc | 2014-08-22 03:07:59 -0400 | [diff] [blame] | 142 | except: |
Sapan Bhatia | 8a51bf5 | 2015-01-29 20:55:40 +0000 | [diff] [blame] | 143 | o.backend_status = '2 - %s'%str_e |
Sapan Bhatia | eba0843 | 2014-04-28 23:58:36 -0400 | [diff] [blame] | 144 | |
Sapan Bhatia | 8a51bf5 | 2015-01-29 20:55:40 +0000 | [diff] [blame] | 145 | try: |
| 146 | scratchpad = json.loads(o.backend_register) |
| 147 | scratchpad['exponent'] |
| 148 | except: |
| 149 | scratchpad = {'next_run':0, 'exponent':0} |
| 150 | |
| 151 | # Second failure |
| 152 | if (scratchpad['exponent']): |
| 153 | delay = scratchpad['exponent'] * 600 # 10 minutes |
| 154 | if (delay<1440): |
| 155 | delay = 1440 |
| 156 | scratchpad['next_run'] = time.time() + delay |
| 157 | |
| 158 | scratchpad['exponent']+=1 |
| 159 | |
| 160 | o.backend_register = json.dumps(scratchpad) |
| 161 | |
| 162 | # TOFIX: |
| 163 | # DatabaseError: value too long for type character varying(140) |
| 164 | if (o.pk): |
| 165 | try: |
Sapan Bhatia | 6d42215 | 2015-02-08 06:31:42 +0000 | [diff] [blame] | 166 | o.backend_status = o.backend_status[:1024] |
Sapan Bhatia | 8a51bf5 | 2015-01-29 20:55:40 +0000 | [diff] [blame] | 167 | o.save(update_fields=['backend_status','backend_register']) |
| 168 | except: |
| 169 | print "Could not update backend status field!" |
| 170 | pass |
| 171 | sync_failed = True |
| 172 | |
| 173 | |
| 174 | if (sync_failed): |
Tony Mack | 68e818d | 2013-09-25 13:34:17 -0400 | [diff] [blame] | 175 | failed.append(o) |
Sapan Bhatia | ca2e21f | 2013-10-02 01:10:02 -0400 | [diff] [blame] | 176 | |
Tony Mack | ce79de0 | 2013-09-24 10:12:33 -0400 | [diff] [blame] | 177 | return failed |
Sapan Bhatia | 24836f1 | 2013-08-27 10:16:05 -0400 | [diff] [blame] | 178 | |
Tony Mack | 16f0474 | 2013-09-25 08:53:28 -0400 | [diff] [blame] | 179 | def __call__(self, **args): |
| 180 | return self.call(**args) |