blob: 638c1eb6f808e122745789b0053ecc1422f296c3 [file] [log] [blame]
Sapan Bhatia24836f12013-08-27 10:16:05 -04001import os
2import base64
Tony Mack4fa85fb2013-09-25 14:39:57 -04003from datetime import datetime
Sapan Bhatia24836f12013-08-27 10:16:05 -04004from planetstack.config import Config
Andy Baviere7abb622013-10-18 15:11:56 -04005from util.logger import Logger, logging
Sapan Bhatiaeba08432014-04-28 23:58:36 -04006from observer.steps import *
Sapan Bhatiad9468eb2014-08-20 03:03:12 -04007from django.db.models import F, Q
Sapan Bhatia082ee362015-02-08 06:35:36 +00008from core.models import *
Sapan Bhatia8a51bf52015-01-29 20:55:40 +00009import json
10import time
11import pdb
Andy Baviere7abb622013-10-18 15:11:56 -040012
Andy Bavier04111b72013-10-22 16:47:10 -040013logger = Logger(level=logging.INFO)
Sapan Bhatia24836f12013-08-27 10:16:05 -040014
Sapan Bhatiab7b1b6e2015-02-08 06:36:32 +000015def 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
20def elim_dups(backend_str):
21 strs = backend_str.split(' // ')
22 strs2 = f7(strs)
23 return ' // '.join(strs2)
24
Sapan Bhatia082ee362015-02-08 06:35:36 +000025def deepgetattr(obj, attr):
26 return reduce(getattr, attr.split('.'), obj)
27
Sapan Bhatia13c7f112013-09-02 14:19:35 -040028class FailedDependency(Exception):
Tony Mackce79de02013-09-24 10:12:33 -040029 pass
Sapan Bhatia13c7f112013-09-02 14:19:35 -040030
Tony Macke10fbe52015-01-03 19:37:39 -050031class SyncStep(object):
Tony Mackce79de02013-09-24 10:12:33 -040032 """ A PlanetStack Sync step.
Sapan Bhatia24836f12013-08-27 10:16:05 -040033
Tony Mackce79de02013-09-24 10:12:33 -040034 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 Bhatia24836f12013-08-27 10:16:05 -040046
Tony Mackce79de02013-09-24 10:12:33 -040047 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 Mack387a73f2013-09-18 07:59:14 -040054 self.driver = args.get('driver')
Sapan Bhatiaeba08432014-04-28 23:58:36 -040055 self.error_map = args.get('error_map')
56
Tony Mackce79de02013-09-24 10:12:33 -040057 try:
58 self.soft_deadline = int(self.get_prop('soft_deadline_seconds'))
59 except:
60 self.soft_deadline = 5 # 5 seconds
Sapan Bhatia24836f12013-08-27 10:16:05 -040061
Tony Mackce79de02013-09-24 10:12:33 -040062 return
Sapan Bhatia24836f12013-08-27 10:16:05 -040063
Sapan Bhatiae17bc5b2014-04-30 00:53:06 -040064 def fetch_pending(self, deletion=False):
Sapan Bhatia21765662014-07-23 08:59:30 -040065 # 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 Bhatia39a775f2015-01-29 20:58:25 +000068 main_obj = self.observes
Sapan Bhatiad9468eb2014-08-20 03:03:12 -040069 if (not deletion):
Sapan Bhatia21765662014-07-23 08:59:30 -040070 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 Bhatiaca2e21f2013-10-02 01:10:02 -040075 #return Sliver.objects.filter(ip=None)
Tony Mackce79de02013-09-24 10:12:33 -040076
Sapan Bhatiaca2e21f2013-10-02 01:10:02 -040077 def check_dependencies(self, obj, failed):
Tony Mackce79de02013-09-24 10:12:33 -040078 for dep in self.dependencies:
Scott Baker105b6b72014-05-12 10:40:25 -070079 peer_name = dep[0].lower() + dep[1:] # django names are camelCased with the first letter lower
Sapan Bhatia082ee362015-02-08 06:35:36 +000080
Sapan Bhatiacfef6ef2014-08-20 03:04:03 -040081 try:
Sapan Bhatia082ee362015-02-08 06:35:36 +000082 peer_object = deepgetattr(obj, peer_name)
83 try:
84 peer_objects = peer_object.all()
85 except AttributeError:
86 peer_objects = [peer_object]
Sapan Bhatiacfef6ef2014-08-20 03:04:03 -040087 except:
Sapan Bhatia082ee362015-02-08 06:35:36 +000088 peer_objects = []
Sapan Bhatiacfef6ef2014-08-20 03:04:03 -040089
Sapan Bhatia082ee362015-02-08 06:35:36 +000090 if (failed in peer_objects):
91 if (obj.backend_status!=failed.backend_status):
92 obj.backend_status = failed.backend_status
Sapan Bhatia7e482de2014-08-22 03:05:13 -040093 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 Bhatia24836f12013-08-27 10:16:05 -040095
Sapan Bhatia60823362014-04-30 00:52:32 -040096 def call(self, failed=[], deletion=False):
97 pending = self.fetch_pending(deletion)
Tony Mackce79de02013-09-24 10:12:33 -040098 for o in pending:
Sapan Bhatia8a51bf52015-01-29 20:55:40 +000099 sync_failed = False
Tony Mack68e818d2013-09-25 13:34:17 -0400100 try:
Sapan Bhatia8a51bf52015-01-29 20:55:40 +0000101 scratchpad = json.loads(o.backend_register)
102 if (scratchpad):
103 next_run = scratchpad['next_run']
104 if (next_run>time.time()):
105 sync_failed = True
106 print "BACKING OFF, exponent = %d"%scratchpad['exponent']
107 except:
108 pass
109
110 if (not sync_failed):
Sapan Bhatiaeba08432014-04-28 23:58:36 -0400111 try:
Sapan Bhatia8a51bf52015-01-29 20:55:40 +0000112 for f in failed:
113 self.check_dependencies(o,f) # Raises exception if failed
114 if (deletion):
115 self.delete_record(o)
116 o.delete(purge=True)
117 else:
118 self.sync_record(o)
119 o.enacted = datetime.now() # Is this the same timezone? XXX
120 scratchpad = {'next_run':0, 'exponent':0}
121 o.backend_register = json.dumps(scratchpad)
122 o.backend_status = "1 - OK"
123 o.save(update_fields=['enacted','backend_status','backend_register'])
124 except Exception,e:
125 logger.log_exc("sync step failed!")
Sapan Bhatia6d422152015-02-08 06:31:42 +0000126 try:
127 if (o.backend_status.startswith('2 - ')):
128 str_e = '%s // %r'%(o.backend_status[4:],e)
Sapan Bhatiab7b1b6e2015-02-08 06:36:32 +0000129 str_e = elim_dups(str_e)
Sapan Bhatia6d422152015-02-08 06:31:42 +0000130 else:
131 str_e = '%r'%e
132 except:
133 str_e = '%r'%e
134
Sapan Bhatia9c308fc2014-08-22 03:07:59 -0400135 try:
Sapan Bhatia8a51bf52015-01-29 20:55:40 +0000136 o.backend_status = '2 - %s'%self.error_map.map(str_e)
Sapan Bhatia9c308fc2014-08-22 03:07:59 -0400137 except:
Sapan Bhatia8a51bf52015-01-29 20:55:40 +0000138 o.backend_status = '2 - %s'%str_e
Sapan Bhatiaeba08432014-04-28 23:58:36 -0400139
Sapan Bhatia8a51bf52015-01-29 20:55:40 +0000140 try:
141 scratchpad = json.loads(o.backend_register)
142 scratchpad['exponent']
143 except:
144 scratchpad = {'next_run':0, 'exponent':0}
145
146 # Second failure
147 if (scratchpad['exponent']):
148 delay = scratchpad['exponent'] * 600 # 10 minutes
149 if (delay<1440):
150 delay = 1440
151 scratchpad['next_run'] = time.time() + delay
152
153 scratchpad['exponent']+=1
154
155 o.backend_register = json.dumps(scratchpad)
156
157 # TOFIX:
158 # DatabaseError: value too long for type character varying(140)
159 if (o.pk):
160 try:
Sapan Bhatia6d422152015-02-08 06:31:42 +0000161 o.backend_status = o.backend_status[:1024]
Sapan Bhatia8a51bf52015-01-29 20:55:40 +0000162 o.save(update_fields=['backend_status','backend_register'])
163 except:
164 print "Could not update backend status field!"
165 pass
166 sync_failed = True
167
168
169 if (sync_failed):
Tony Mack68e818d2013-09-25 13:34:17 -0400170 failed.append(o)
Sapan Bhatiaca2e21f2013-10-02 01:10:02 -0400171
Tony Mackce79de02013-09-24 10:12:33 -0400172 return failed
Sapan Bhatia24836f12013-08-27 10:16:05 -0400173
Tony Mack16f04742013-09-25 08:53:28 -0400174 def __call__(self, **args):
175 return self.call(**args)