backup/warcprox
1
0
Fork 0
mirror of https://github.com/internetarchive/warcprox.git synced 2025-01-18 13:22:09 +01:00
Code Issues Projects Releases Activity

add pid and first_heartbeat to service registry automatically

Browse Source
This commit is contained in:
Noah Levitt 2015-11-02 21:20:46 +00:00
parent 8a212327eb
commit a0f5c4870f
2 changed files with 16 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
rethinkstuff/__init__.py
View File

@ -5,6 +5,7 @@ import random
import time
import types
import socket
import os
class RethinkerWrapper(object):
logger = logging.getLogger('rethinkstuff.RethinkerWrapper')
@ -107,7 +108,9 @@ class ServiceRegistry(object):
'role': 'brozzler-worker',
'load': 0.5, # load score
'heartbeat_interval': 20.0,
'host': 'wbgrp-svc999.us.archive.org', # set in svcreg.heartbeat() as a fallback
'host': 'wbgrp-svc999.us.archive.org', # set in svcreg.heartbeat() as a fallback
'pid': 1234, # set in svcreg.heartbeat() as a fallback
'first_heartbeat': '2015-10-30T03:39:40.080814', # set in svcreg.heartbeat()
'last_heartbeat': '2015-10-30T05:54:35.422866', # set in svcreg.heartbeat()
... plus anything else you want...
}
@ -133,8 +136,13 @@ class ServiceRegistry(object):
def heartbeat(self, status_info):
'''Returns id of rethinkdb document (generated by rethinkdb on first insert)'''
status_info['last_heartbeat'] = r.now()
if not 'first_heartbeat' in status_info:
status_info['first_heartbeat'] = status_info['last_heartbeat']
status_info['last_heartbeat'] = r.now()
if not 'host' in status_info:
status_info['host'] = socket.gethostname()
if not 'pid' in status_info:
status_info['pid'] = os.getpid()
result = self.r.table('services').insert(status_info, conflict='replace', return_changes=True).run()
# XXX check
return result['changes'][0]['new_val']

8
tests/test_rethinker.py
View File

@ -7,6 +7,7 @@ import pytest
import rethinkdb
import time
import socket
import os
logging.basicConfig(stream=sys.stderr, level=logging.INFO,
format="%(asctime)s %(process)d %(levelname)s %(threadName)s %(name)s.%(funcName)s(%(filename)s:%(lineno)d) %(message)s")
@ -92,7 +93,6 @@ def test_slice(r, my_table):
assert n == 5
def test_service_registry(r):
# import pdb; pdb.set_trace()
svcreg = rethinkstuff.ServiceRegistry(r)
assert svcreg.available_service("yes-such-role") == None
svc0 = {
@ -114,6 +114,12 @@ def test_service_registry(r):
assert svc0["host"] == socket.gethostname()
assert svc1["host"] == socket.gethostname()
assert "pid" in svc0
assert "pid" in svc1
assert svc0["pid"] == os.getpid()
assert svc1["pid"] == os.getpid()
assert "first_heartbeat" in svc0
assert "first_heartbeat" in svc1
assert "last_heartbeat" in svc0
assert "last_heartbeat" in svc1