Register blobstore servers with ZooKeeper
@@ -34,7 +34,7 @@ from appscale.common.constants import LOG_FORMAT
from appscale.common.deployment_config import DeploymentConfig
from appscale.common.deployment_config import ConfigInaccessible
from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
-from kazoo.client import KazooClient
+from kazoo.client import KazooClient, KazooState, NodeExistsError
from StringIO import StringIO
sys.path.append(APPSCALE_PYTHON_APPSERVER)
@@ -73,6 +73,8 @@ GCS_UPLOAD_ID_HEADER = 'X-GUploader-UploadID'
# The chunk size to use for uploading files to GCS.
GCS_CHUNK_SIZE = 5 * 1024 * 1024 # 5MB
+BLOBSTORE_SERVERS_NODE = '/appscale/blobstore/servers'
+
# Global used for setting the datastore path when registering the DB
datastore_path = ""
@@ -423,6 +425,37 @@ class UploadHandler(tornado.web.RequestHandler):
return
+def register_location(zk_client, host, port):
+ """ Register service location with ZooKeeper. """
+ server_node = '{}/{}:{}'.format(BLOBSTORE_SERVERS_NODE, host, port)
+
+ def create_server_node():
+ """ Creates a server registration entry in ZooKeeper. """
+ try:
+ zk_client.retry(zk_client.create, server_node, ephemeral=True)
+ except NodeExistsError:
+ # If the server gets restarted, the old node may exist for a short time.
+ zk_client.retry(zk_client.delete, server_node)
+ zk_client.retry(zk_client.create, server_node, ephemeral=True)
+
+ logger.info('Blobstore server registered at {}'.format(server_node))
+
+ def zk_state_listener(state):
+ """ Handles changes to ZooKeeper connection state.
+
+ Args:
+ state: A string specifying the new ZooKeeper connection state.
+ """
+ if state == KazooState.CONNECTED:
+ tornado.ioloop.IOLoop.instance().add_callback(create_server_node)
+
+ zk_client.add_listener(zk_state_listener)
+ zk_client.ensure_path(BLOBSTORE_SERVERS_NODE)
+ # Since the client was started before adding the listener, make sure the
+ # server node gets created.
+ zk_state_listener(zk_client.state)
+
+
def main():
global datastore_path
global deployment_config
@@ -443,6 +476,8 @@ def main():
deployment_config = DeploymentConfig(zk_client)
setup_env()
+ register_location(zk_client, appscale_info.get_private_ip(), args.port)
+
http_server = tornado.httpserver.HTTPServer(
Application(), max_buffer_size=MAX_REQUEST_BUFF_SIZE, xheaders=True)
