entropy.services.system.interfaces

65

66 - class FakeServiceInterface:

67 - def __init__(self, *args, **kwargs):

68 pass

69

70 - class BuiltInSystemManagerExecutorCommands:

71

72 - def __init__(self, SystemManagerExecutorInstance, *args, **kwargs):

73 self.SystemManagerExecutor = SystemManagerExecutorInstance 74 self.available_commands = { 75 'hello_world': { 76 'func': self.hello_world, 77 'args': 0, 78 } 79 }

80

81 - def hello_world(self):

82 rc = subprocess.call('echo hello world', shell = True) 83 return True, rc

84 85 86 queue_file = 'system_manager_queue' 87 pinboard_file = "system_manager_pinboard" 88 queue_ext_rc_dir = "system_manager_rc" 89 STDOUT_STORAGE_DIR = os.path.join(etpConst['dumpstoragedir'], 'system_manager_stdout')

90 - def __init__(self, EntropyInterface, do_ssl = False, stdout_logging = True, entropy_interface_kwargs = {}, **kwargs):

91 92 self.started = False 93 self.queue_loaded = False 94 from entropy.misc import TimeScheduled 95 self.TimeScheduled = TimeScheduled 96 97 import entropy.tools as entropyTools 98 import entropy.dump as dumpTools 99 import threading 100 self.entropyTools, self.dumpTools, self.threading = entropyTools, dumpTools, threading 101 from datetime import datetime 102 self.datetime = datetime 103 import copy 104 self.copy = copy 105 from entropy.services.system.commands import Base 106 self.setup_stdout_storage_dir() 107 108 if 'external_cmd_classes' not in kwargs: 109 kwargs['external_cmd_classes'] = [] 110 kwargs['external_cmd_classes'].insert(0, Base) 111 112 self.Entropy = EntropyInterface(**entropy_interface_kwargs) 113 self.Text = TextInterface() 114 self.SystemExecutor = TaskExecutor(self, self.Entropy) 115 116 self.ExecutorCommandClasses = [(self.BuiltInSystemManagerExecutorCommands, [], {},)] 117 self.ExecutorCommandInstances = [] 118 if 'external_executor_cmd_classes' in kwargs: 119 self.ExecutorCommandClasses += kwargs.pop('external_executor_cmd_classes') 120 self.handle_executor_command_classes_initialization() 121 122 self.QueueProcessor = None 123 self.QueueLock = self.threading.Lock() 124 self.PinboardLock = self.threading.Lock() 125 self.ForkLock = self.threading.Lock() 126 self.do_ssl = do_ssl 127 128 self.PinboardData = {} 129 self.load_pinboard() 130 131 self.done_queue_keys = ['processed', 'errored'] 132 self.removable_queue_keys = ['processed', 'errored', 'queue'] 133 self.processing_queue_keys = ['processing'] 134 self.dict_queue_keys = ['queue', 'processing', 'processed', 'errored'] 135 self.ManagerQueueStdInOut = {} 136 self.ManagerQueue = { 137 'queue': {}, 138 'queue_order': [], 139 'processing': {}, 140 'processing_order': [], 141 'processed': {}, 142 'processed_order': [], 143 'errored' : {}, 144 'errored_order': [], 145 'pause': True 146 } 147 self.load_queue() 148 self.queue_loaded = True 149 if self.ManagerQueue['processing'] or self.ManagerQueue['processing_order']: 150 self.ManagerQueue['processing'].clear() 151 del self.ManagerQueue['processing_order'][:] 152 self.store_queue() 153 154 SocketHost.__init__( 155 self, 156 self.FakeServiceInterface, 157 sock_output = self.Text, 158 ssl = do_ssl, 159 **kwargs 160 ) 161 self.stdout_logging = stdout_logging 162 # no way, we MUST fork requests, otherwise weird things will happen when more than 163 # one user is connected 164 # self.fork_requests = False 165 self.load_queue_processor() 166 # here we can put anything that must be loaded before the queue processor execution 167 self.play_queue() 168 self.started = True

169

170 - def __del__(self):

171 if hasattr(self, 'queue_loaded'): 172 if self.queue_loaded: 173 self.store_queue()

174

175 - def handle_executor_command_classes_initialization(self):

176 for myclass, args, kwargs in self.ExecutorCommandClasses: 177 myintf = myclass(self.SystemExecutor, *args,**kwargs) 178 if hasattr(myintf, 'available_commands'): 179 self.SystemExecutor.register(myintf.available_commands) 180 self.ExecutorCommandInstances.append(myintf) 181 else: 182 del myintf

183

184 - def setup_stdout_storage_dir(self):

185 if os.path.isfile(self.STDOUT_STORAGE_DIR) or os.path.islink(self.STDOUT_STORAGE_DIR): 186 os.remove(self.STDOUT_STORAGE_DIR) 187 if not os.path.isdir(self.STDOUT_STORAGE_DIR): 188 os.makedirs(self.STDOUT_STORAGE_DIR, 0o775) 189 if etpConst['entropygid'] != None: 190 const_setup_perms(self.STDOUT_STORAGE_DIR, etpConst['entropygid'])

191

192 - def load_pinboard(self):

193 obj = self.get_stored_pinboard() 194 if isinstance(obj, dict): 195 self.PinboardData = obj 196 return True 197 return False

198

199 - def get_stored_pinboard(self):

200 return self.dumpTools.loadobj(self.pinboard_file)

201

202 - def store_pinboard(self):

203 self.dumpTools.dumpobj(self.pinboard_file, self.PinboardData)

204

205 - def add_to_pinboard(self, note, extended_text):

206 with self.PinboardLock: 207 mydata = { 208 'note': note, 209 'extended_text': extended_text, 210 'ts': self.get_ts(), 211 'done': False, 212 } 213 pinboard_id = self.get_pinboard_id() 214 self.PinboardData[pinboard_id] = mydata 215 self.store_pinboard()

216

217 - def remove_from_pinboard(self, pinboard_id):

218 with self.PinboardLock: 219 if pinboard_id in self.PinboardData: 220 self.PinboardData.pop(pinboard_id) 221 self.store_pinboard() 222 return True 223 return False

224

225 - def set_pinboard_item_status(self, pinboard_id, status):

226 with self.PinboardLock: 227 if pinboard_id in self.PinboardData: 228 self.PinboardData[pinboard_id]['done'] = status 229 self.store_pinboard() 230 return True 231 return False

232

233 - def get_pinboard_id(self):

234 numbers = list(self.PinboardData.keys()) 235 if numbers: 236 number = max(numbers)+1 237 else: 238 number = 1 239 return number

240

241 - def get_pinboard_data(self):

242 with self.PinboardLock: 243 return self.PinboardData.copy()

244

245 - def load_queue_processor(self):

246 self.QueueProcessor = self.TimeScheduled(2, self.queue_processor) 247 self.QueueProcessor.start()

248

249 - def get_stored_queue(self):

250 return self.dumpTools.loadobj(self.queue_file)

251

252 - def load_queue(self):

253 obj = self.get_stored_queue() 254 if isinstance(obj, dict): 255 self.ManagerQueue = obj 256 return True 257 return False

258

259 - def store_queue(self):

260 self.dumpTools.dumpobj(self.queue_file, self.ManagerQueue)

261

262 - def load_queue_ext_rc(self, queue_id):

263 return self.dumpTools.loadobj(os.path.join(self.queue_ext_rc_dir, str(queue_id)))

264

265 - def store_queue_ext_rc(self, queue_id, rc):

266 return self.dumpTools.dumpobj(os.path.join(self.queue_ext_rc_dir, str(queue_id)), rc)

267

268 - def remove_queue_ext_rc(self, queue_id):

269 return self.dumpTools.removeobj(os.path.join(self.queue_ext_rc_dir, str(queue_id)))

270

271 - def get_ts(self):

272 return self.datetime.fromtimestamp(time.time())

273

274 - def swap_items_in_queue(self, queue_id1, queue_id2):

275 self.load_queue() 276 item1, key1 = self._get_item_by_queue_id(queue_id1) 277 item2, key2 = self._get_item_by_queue_id(queue_id2) 278 if key1 != key2: 279 return False 280 t_item = item1.copy() 281 item1.clear() 282 item1.update(item2) 283 item2.clear() 284 item2.update(t_item) 285 # fix the _order 286 queue_id1_idx = self.ManagerQueue[key1+"_order"].index(queue_id1) 287 queue_id2_idx = self.ManagerQueue[key2+"_order"].index(queue_id2) 288 self.ManagerQueue[key1+"_order"][queue_id1_idx] = queue_id2 289 self.ManagerQueue[key2+"_order"][queue_id2_idx] = queue_id1 290 self.store_queue() 291 return True

292 293

294 - def add_to_queue(self, command_name, command_text, user_id, group_id, function, args, kwargs, do_parallel, extended_result, interactive = False):

295 296 if function not in self.SystemExecutor.available_commands: 297 return -1 298 299 self.load_queue() 300 queue_id = self.generate_unique_queue_id() 301 if interactive: 302 self.ManagerQueueStdInOut[queue_id] = os.pipe() 303 myqueue_dict = { 304 'queue_id': queue_id, 305 'command_name': command_name, 306 'command_desc': self.valid_commands[command_name]['desc'], 307 'command_text': command_text, 308 'call': function, 309 'args': self.copy.deepcopy(args), 310 'kwargs': self.copy.deepcopy(kwargs), 311 'user_id': user_id, 312 'group_id': group_id, 313 'stdout': self.assign_unique_stdout_file(queue_id), 314 'queue_ts': "%s" % (self.get_ts(),), 315 'kill': False, 316 'processing_pid': None, 317 'do_parallel': do_parallel, 318 'interactive': False, 319 } 320 if extended_result: 321 myqueue_dict['extended_result'] = None 322 self.ManagerQueue['queue'][queue_id] = myqueue_dict 323 self.ManagerQueue['queue_order'].append(queue_id) 324 self.store_queue() 325 return queue_id

326

327 - def remove_from_queue(self, queue_ids):

328 self.load_queue() 329 removed = False 330 for key in self.ManagerQueue: 331 if key not in self.dict_queue_keys: 332 continue 333 for queue_id in queue_ids: 334 item = None 335 try: 336 item = self.ManagerQueue[key].pop(queue_id) 337 except KeyError: 338 continue 339 if item: 340 self.flush_item(item, queue_id) 341 if queue_id in self.ManagerQueue[key+"_order"]: 342 self.ManagerQueue[key+"_order"].remove(queue_id) 343 removed = True 344 self.remove_queue_ext_rc(queue_id) 345 if removed: self.store_queue() 346 return removed

347

348 - def kill_processing_queue_id(self, queue_id):

349 self.load_queue() 350 item, key = self._get_item_by_queue_id(queue_id) 351 if key in self.processing_queue_keys: 352 item['kill'] = True 353 self.store_queue()

354

355 - def pause_queue(self):

356 self.load_queue() 357 self.ManagerQueue['pause'] = True 358 self.store_queue()

359

360 - def play_queue(self):

361 self.load_queue() 362 self.ManagerQueue['pause'] = False 363 self.store_queue()

364

365 - def flush_item(self, item, queue_id):

366 if not isinstance(item, dict): 367 return False 368 if 'stdout' in item: 369 stdout = item['stdout'] 370 if (os.path.isfile(stdout) and os.access(stdout, os.W_OK)): 371 os.remove(stdout) 372 if 'interactive' in item: 373 if item['interactive'] and (queue_id in self.ManagerQueueStdInOut): 374 stdin, stdout = self.ManagerQueueStdInOut.pop(queue_id) 375 os.close(stdin) 376 os.close(stdout) 377 return True

378

379 - def assign_unique_stdout_file(self, queue_id):

380 stdout = os.path.join(self.STDOUT_STORAGE_DIR, "%d.%s" % (queue_id, "stdout",)) 381 if os.path.isfile(stdout): 382 os.remove(stdout) 383 count = 0 384 orig_stdout = stdout 385 while os.path.lexists(stdout): 386 count += 1 387 stdout = "%s.%d" % (orig_stdout, count,) 388 return stdout

389

390 - def generate_unique_queue_id(self):

391 current_ids = set() 392 for key in self.ManagerQueue: 393 if not key.endswith("_order"): 394 continue 395 current_ids |= set(self.ManagerQueue[key]) 396 while True: 397 try: 398 queue_id = abs(hash(os.urandom(1))) 399 except NotImplementedError: 400 random.seed() 401 queue_id = random.randint(1000000000000000000, 9999999999999999999) 402 if queue_id not in current_ids: 403 return queue_id

404

405 - def get_item_by_queue_id(self, queue_id, copy = False):

406 self.load_queue() 407 item, key = self._get_item_by_queue_id(queue_id) 408 if copy: item = self._queue_copy_obj(item) 409 return item, key

410

411 - def _get_item_by_queue_id(self, queue_id):

412 for key in self.dict_queue_keys: 413 item = self.ManagerQueue[key].get(queue_id) 414 if item != None: 415 return item, key 416 return None, None

417

418 - def _pop_item_from_queue(self):

419 try: 420 if self.ManagerQueue['queue_order']: 421 queue_id = self.ManagerQueue['queue_order'].pop(0) 422 return self.ManagerQueue['queue'].pop(queue_id), queue_id 423 except (IndexError, KeyError,): 424 self.entropyTools.print_traceback() 425 return None, None

426

427 - def _queue_copy_obj(self, obj):

428 if isinstance(obj, (dict, set, frozenset)): 429 return obj.copy() 430 elif isinstance(obj, (list, tuple)): 431 return obj[:] 432 return obj

433

434 - def queue_processor(self, fork_data = None):

435 436 try: 437 self._queue_processor(fork_data) 438 except: 439 if self.QueueLock.locked() and not fork_data: 440 self.QueueLock.release() 441 raise

442

443 - def _queue_processor(self, fork_data):

444 445 # queue processing is stopped until there's a process running 446 if self.ForkLock.locked(): return 447 448 with self.ForkLock: 449 with self.QueueLock: 450 451 if fork_data: 452 command_data, queue_id = self._queue_copy_obj(fork_data) 453 else: 454 self.load_queue() 455 if self.ManagerQueue['pause']: return 456 if not self.ManagerQueue['queue_order']: return 457 command_data, queue_id = self._pop_item_from_queue() 458 if not command_data: return 459 command_data = self._queue_copy_obj(command_data) 460 command_data['processing_ts'] = "%s" % (self.get_ts(),) 461 self.ManagerQueue['processing'][queue_id] = command_data 462 self.ManagerQueue['processing_order'].append(queue_id) 463 self.store_queue() 464 465 self.remove_queue_ext_rc(queue_id) 466 try: 467 if command_data.get('do_parallel') and not fork_data: 468 t = ParallelTask(self.queue_processor, fork_data = (command_data, queue_id,)) 469 t.start() 470 return 471 done, result = self.SystemExecutor.execute_task(command_data) 472 except Exception as e: 473 if self.QueueLock.locked(): self.QueueLock.release() 474 self.entropyTools.print_traceback() 475 done = False 476 result = (False, str(e),) 477 478 if 'extended_result' in command_data and done: 479 try: 480 command_data['result'], extended_result = self._queue_copy_obj(result) 481 self.store_queue_ext_rc(queue_id, extended_result) 482 except TypeError: 483 done = False 484 command_data['result'] = 'wrong tuple split from queue processor (1)' 485 self.store_queue_ext_rc(queue_id, None) 486 else: 487 command_data['result'] = self._queue_copy_obj(result) 488 489 with self.ForkLock: 490 with self.QueueLock: 491 492 self.load_queue() 493 494 if not done: 495 try: 496 self.ManagerQueue['processing'].pop(queue_id) 497 except KeyError: 498 pass 499 if queue_id in self.ManagerQueue['processing_order']: 500 self.ManagerQueue['processing_order'].remove(queue_id) 501 command_data['errored_ts'] = "%s" % (self.get_ts(),) 502 self.ManagerQueue['errored'][queue_id] = command_data 503 self.ManagerQueue['errored_order'].append(queue_id) 504 self.store_queue() 505 return 506 507 try: 508 done, cmd_result = result 509 except TypeError: 510 done = False 511 command_data['result'] = 'wrong tuple split from queue processor (2)' 512 513 if not done: 514 try: 515 self.ManagerQueue['processing'].pop(queue_id) 516 except KeyError: 517 pass 518 if queue_id in self.ManagerQueue['processing_order']: 519 self.ManagerQueue['processing_order'].remove(queue_id) 520 command_data['errored_ts'] = "%s" % (self.get_ts(),) 521 self.ManagerQueue['errored'][queue_id] = command_data 522 self.ManagerQueue['errored_order'].append(queue_id) 523 self.store_queue() 524 return 525 526 try: 527 self.ManagerQueue['processing'].pop(queue_id) 528 except KeyError: 529 pass 530 if queue_id in self.ManagerQueue['processing_order']: 531 self.ManagerQueue['processing_order'].remove(queue_id) 532 command_data['processed_ts'] = "%s" % (self.get_ts(),) 533 self.ManagerQueue['processed'][queue_id] = command_data 534 self.ManagerQueue['processed_order'].append(queue_id) 535 self.store_queue()

536 537

538 - def killall(self):

539 SocketHost.killall(self) 540 if self.QueueProcessor != None: 541 self.QueueProcessor.kill()

Source Code for Module entropy.services.system.interfaces