blob: 247d4c99aad542fb35105b17c9adb8e1630b6f2f [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 Bhatia8a51bf52015-01-29 20:55:40 +00008import json
9import time
10import pdb
Andy Baviere7abb622013-10-18 15:11:56 -040011
Andy Bavier04111b72013-10-22 16:47:10 -040012logger = Logger(level=logging.INFO)
Sapan Bhatia24836f12013-08-27 10:16:05 -040013
Sapan Bhatia13c7f112013-09-02 14:19:35 -040014class FailedDependency(Exception):
Tony Mackce79de02013-09-24 10:12:33 -040015 pass
Sapan Bhatia13c7f112013-09-02 14:19:35 -040016
Tony Macke10fbe52015-01-03 19:37:39 -050017class SyncStep(object):
Tony Mackce79de02013-09-24 10:12:33 -040018 """ A PlanetStack Sync step.
Sapan Bhatia24836f12013-08-27 10:16:05 -040019
Tony Mackce79de02013-09-24 10:12:33 -040020 Attributes:
21 psmodel Model name the step synchronizes
22 dependencies list of names of models that must be synchronized first if the current model depends on them
23 """
24 slow=False
25 def get_prop(prop):
26 try:
27 sync_config_dir = Config().sync_config_dir
28 except:
29 sync_config_dir = '/etc/planetstack/sync'
30 prop_config_path = '/'.join(sync_config_dir,self.name,prop)
31 return open(prop_config_path).read().rstrip()
Sapan Bhatia24836f12013-08-27 10:16:05 -040032
Tony Mackce79de02013-09-24 10:12:33 -040033 def __init__(self, **args):
34 """Initialize a sync step
35 Keyword arguments:
36 name -- Name of the step
37 provides -- PlanetStack models sync'd by this step
38 """
39 dependencies = []
Tony Mack387a73f2013-09-18 07:59:14 -040040 self.driver = args.get('driver')
Sapan Bhatiaeba08432014-04-28 23:58:36 -040041 self.error_map = args.get('error_map')
42
Tony Mackce79de02013-09-24 10:12:33 -040043 try:
44 self.soft_deadline = int(self.get_prop('soft_deadline_seconds'))
45 except:
46 self.soft_deadline = 5 # 5 seconds
Sapan Bhatia24836f12013-08-27 10:16:05 -040047
Tony Mackce79de02013-09-24 10:12:33 -040048 return
Sapan Bhatia24836f12013-08-27 10:16:05 -040049
Sapan Bhatiae17bc5b2014-04-30 00:53:06 -040050 def fetch_pending(self, deletion=False):
Sapan Bhatia21765662014-07-23 08:59:30 -040051 # This is the most common implementation of fetch_pending
52 # Steps should override it if they have their own logic
53 # for figuring out what objects are outstanding.
Sapan Bhatia39a775f2015-01-29 20:58:25 +000054 main_obj = self.observes
Sapan Bhatiad9468eb2014-08-20 03:03:12 -040055 if (not deletion):
Sapan Bhatia21765662014-07-23 08:59:30 -040056 objs = main_obj.objects.filter(Q(enacted__lt=F('updated')) | Q(enacted=None))
57 else:
58 objs = main_obj.deleted_objects.all()
59
60 return objs
Sapan Bhatiaca2e21f2013-10-02 01:10:02 -040061 #return Sliver.objects.filter(ip=None)
Tony Mackce79de02013-09-24 10:12:33 -040062
Sapan Bhatiaca2e21f2013-10-02 01:10:02 -040063 def check_dependencies(self, obj, failed):
Tony Mackce79de02013-09-24 10:12:33 -040064 for dep in self.dependencies:
Scott Baker105b6b72014-05-12 10:40:25 -070065 peer_name = dep[0].lower() + dep[1:] # django names are camelCased with the first letter lower
Sapan Bhatiacfef6ef2014-08-20 03:04:03 -040066 try:
67 peer_object = getattr(obj, peer_name)
68 except:
69 peer_object = None
70
Scott Baker0652f4e2014-08-19 15:49:27 -070071 if (peer_object and peer_object.pk==failed.pk and type(peer_object)==type(failed)):
Sapan Bhatia7e482de2014-08-22 03:05:13 -040072 if (obj.backend_status!=peer_object.backend_status):
73 obj.backend_status = peer_object.backend_status
74 obj.save(update_fields=['backend_status'])
75 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 -040076
Sapan Bhatia60823362014-04-30 00:52:32 -040077 def call(self, failed=[], deletion=False):
78 pending = self.fetch_pending(deletion)
Tony Mackce79de02013-09-24 10:12:33 -040079 for o in pending:
Sapan Bhatia8a51bf52015-01-29 20:55:40 +000080 sync_failed = False
Tony Mack68e818d2013-09-25 13:34:17 -040081 try:
Sapan Bhatia8a51bf52015-01-29 20:55:40 +000082 scratchpad = json.loads(o.backend_register)
83 if (scratchpad):
84 next_run = scratchpad['next_run']
85 if (next_run>time.time()):
86 sync_failed = True
87 print "BACKING OFF, exponent = %d"%scratchpad['exponent']
88 except:
89 pass
90
91 if (not sync_failed):
Sapan Bhatiaeba08432014-04-28 23:58:36 -040092 try:
Sapan Bhatia8a51bf52015-01-29 20:55:40 +000093 for f in failed:
94 self.check_dependencies(o,f) # Raises exception if failed
95 if (deletion):
96 self.delete_record(o)
97 o.delete(purge=True)
98 else:
99 self.sync_record(o)
100 o.enacted = datetime.now() # Is this the same timezone? XXX
101 scratchpad = {'next_run':0, 'exponent':0}
102 o.backend_register = json.dumps(scratchpad)
103 o.backend_status = "1 - OK"
104 o.save(update_fields=['enacted','backend_status','backend_register'])
105 except Exception,e:
106 logger.log_exc("sync step failed!")
107 str_e = '%r'%e
Sapan Bhatia9c308fc2014-08-22 03:07:59 -0400108 try:
Sapan Bhatia8a51bf52015-01-29 20:55:40 +0000109 o.backend_status = '2 - %s'%self.error_map.map(str_e)
Sapan Bhatia9c308fc2014-08-22 03:07:59 -0400110 except:
Sapan Bhatia8a51bf52015-01-29 20:55:40 +0000111 o.backend_status = '2 - %s'%str_e
Sapan Bhatiaeba08432014-04-28 23:58:36 -0400112
Sapan Bhatia8a51bf52015-01-29 20:55:40 +0000113 try:
114 scratchpad = json.loads(o.backend_register)
115 scratchpad['exponent']
116 except:
117 scratchpad = {'next_run':0, 'exponent':0}
118
119 # Second failure
120 if (scratchpad['exponent']):
121 delay = scratchpad['exponent'] * 600 # 10 minutes
122 if (delay<1440):
123 delay = 1440
124 scratchpad['next_run'] = time.time() + delay
125
126 scratchpad['exponent']+=1
127
128 o.backend_register = json.dumps(scratchpad)
129
130 # TOFIX:
131 # DatabaseError: value too long for type character varying(140)
132 if (o.pk):
133 try:
134 o.backend_status = o.backend_status[:140]
135 o.save(update_fields=['backend_status','backend_register'])
136 except:
137 print "Could not update backend status field!"
138 pass
139 sync_failed = True
140
141
142 if (sync_failed):
Tony Mack68e818d2013-09-25 13:34:17 -0400143 failed.append(o)
Sapan Bhatiaca2e21f2013-10-02 01:10:02 -0400144
Tony Mackce79de02013-09-24 10:12:33 -0400145 return failed
Sapan Bhatia24836f12013-08-27 10:16:05 -0400146
Tony Mack16f04742013-09-25 08:53:28 -0400147 def __call__(self, **args):
148 return self.call(**args)