From 9e2c5f821824e7785187685743c4519843e6d73c Mon Sep 17 00:00:00 2001
From: Simon Mo <xmo@berkeley.edu>
Date: Mon, 28 Oct 2019 11:43:47 -0700
Subject: [PATCH] [Serve] Put global state in remote actor (#5937)

* Making progress

* Impl done, start debugging

* Tests all pass

* Add test, fix

* Update doc

* Fix type
---
 python/ray/experimental/serve/__init__.py     |  10 +-
 python/ray/experimental/serve/api.py          | 217 ++++++++++--------
 python/ray/experimental/serve/constants.py    |  15 ++
 python/ray/experimental/serve/context.py      |   1 +
 .../experimental/serve/examples/echo_full.py  |   6 +-
 python/ray/experimental/serve/global_state.py | 195 +++++++++-------
 python/ray/experimental/serve/handle.py       |   5 +-
 python/ray/experimental/serve/http_util.py    |   3 +-
 .../experimental/serve/kv_store_service.py    |  96 +++++++-
 python/ray/experimental/serve/metric.py       |   4 +-
 python/ray/experimental/serve/server.py       |  41 ++--
 python/ray/experimental/serve/task_runner.py  |   6 +-
 .../ray/experimental/serve/tests/conftest.py  |   7 +-
 .../ray/experimental/serve/tests/test_api.py  |   7 +-
 .../experimental/serve/tests/test_metric.py   |   1 -
 .../serve/tests/test_persistence.py           |  37 +++
 .../experimental/serve/tests/test_routing.py  |  31 ++-
 .../serve/tests/test_task_runner.py           |   9 +-
 python/ray/experimental/serve/utils.py        |  30 +++
 19 files changed, 480 insertions(+), 241 deletions(-)
 create mode 100644 python/ray/experimental/serve/tests/test_persistence.py

diff --git a/python/ray/experimental/serve/__init__.py b/python/ray/experimental/serve/__init__.py
index db1f5d9f1..7218feef1 100644
--- a/python/ray/experimental/serve/__init__.py
+++ b/python/ray/experimental/serve/__init__.py
@@ -2,11 +2,11 @@ import sys
 if sys.version_info < (3, 0):
     raise ImportError("serve is Python 3 only.")
 
-from ray.experimental.serve.api import (
-    init, create_backend, create_endpoint, link, split, rollback, get_handle,
-    global_state, stat, scale)  # noqa: E402
+from ray.experimental.serve.api import (init, create_backend, create_endpoint,
+                                        link, split, get_handle, stat,
+                                        scale)  # noqa: E402
 
 __all__ = [
-    "init", "create_backend", "create_endpoint", "link", "split", "rollback",
-    "get_handle", "global_state", "stat", "scale"
+    "init", "create_backend", "create_endpoint", "link", "split", "get_handle",
+    "stat", "scale"
 ]
diff --git a/python/ray/experimental/serve/api.py b/python/ray/experimental/serve/api.py
index 01bff7e31..b0d97fc16 100644
--- a/python/ray/experimental/serve/api.py
+++ b/python/ray/experimental/serve/api.py
@@ -1,68 +1,121 @@
 import inspect
+from functools import wraps
 
 import numpy as np
 
 import ray
+from ray.experimental.serve.constants import (
+    DEFAULT_HTTP_HOST, DEFAULT_HTTP_PORT, SERVE_NURSERY_NAME)
+from ray.experimental.serve.global_state import (GlobalState,
+                                                 start_initial_state)
+from ray.experimental.serve.kv_store_service import SQLiteKVStore
 from ray.experimental.serve.task_runner import RayServeMixin, TaskRunnerActor
-from ray.experimental.serve.utils import pformat_color_json, logger
-from ray.experimental.serve.global_state import GlobalState
-
-global_state = GlobalState()
+from ray.experimental.serve.utils import (block_until_http_ready,
+                                          get_random_letters)
+from ray.experimental.serve.exceptions import RayServeException
+global_state = None
 
 
-def init(blocking=False, object_store_memory=int(1e8), gc_window_seconds=3600):
+def _get_global_state():
+    """Used for internal purpose. Because just import serve.global_state
+    will always reference the original None object
+    """
+    return global_state
+
+
+def _ensure_connected(f):
+    @wraps(f)
+    def check(*args, **kwargs):
+        if _get_global_state() is None:
+            raise RayServeException("Please run serve.init to initialize or "
+                                    "connect to existing ray serve cluster.")
+        return f(*args, **kwargs)
+
+    return check
+
+
+def init(kv_store_connector=None,
+         kv_store_path="/tmp/ray_serve.db",
+         blocking=False,
+         http_host=DEFAULT_HTTP_HOST,
+         http_port=DEFAULT_HTTP_PORT,
+         ray_init_kwargs={"object_store_memory": int(1e8)},
+         gc_window_seconds=3600):
     """Initialize a serve cluster.
 
+    If serve cluster has already initialized, this function will just return.
+
     Calling `ray.init` before `serve.init` is optional. When there is not a ray
     cluster initialized, serve will call `ray.init` with `object_store_memory`
     requirement.
 
     Args:
+        kv_store_connector (callable): Function of (namespace) => TableObject.
+            We will use a SQLite connector that stores to /tmp by default.
+        kv_store_path (str, path): Path to the SQLite table.
         blocking (bool): If true, the function will wait for the HTTP server to
             be healthy, and other components to be ready before returns.
-        object_store_memory (int): Allocated shared memory size in bytes. The
-            default is 100MiB. The default is kept low for latency stability
-            reason.
+        http_host (str): Host for HTTP server. Default to "0.0.0.0".
+        http_port (int): Port for HTTP server. Default to 8000.
+        ray_init_kwargs (dict): Argument passed to ray.init, if there is no ray
+            connection. Default to {"object_store_memory": int(1e8)} for
+            performance stability reason
         gc_window_seconds(int): How long will we keep the metric data in
             memory. Data older than the gc_window will be deleted. The default
             is 3600 seconds, which is 1 hour.
     """
-    if not ray.is_initialized():
-        ray.init(object_store_memory=object_store_memory)
+    global global_state
 
-    # NOTE(simon): Currently the initialization order is fixed.
-    # HTTP server depends on the API server.
-    # Metric monitor depends on the router.
-    global_state.init_api_server()
-    global_state.init_router()
-    global_state.init_http_server()
-    global_state.init_metric_monitor()
+    # Noop if global_state is no longer None
+    if global_state is not None:
+        return
+
+    # Initialize ray if needed.
+    if not ray.is_initialized():
+        ray.init(**ray_init_kwargs)
+
+    # Try to get serve nursery if there exists
+    try:
+        ray.experimental.get_actor(SERVE_NURSERY_NAME)
+        global_state = GlobalState()
+        return
+    except ValueError:
+        pass
+
+    # Serve has not been initialized, perform init sequence
+    # Todo, move the db to session_dir
+    #    ray.worker._global_node.address_info["session_dir"]
+    def kv_store_connector(namespace):
+        return SQLiteKVStore(namespace, db_path=kv_store_path)
+
+    nursery = start_initial_state(kv_store_connector)
+
+    global_state = GlobalState(nursery)
+    global_state.init_or_get_http_server(host=http_host, port=http_port)
+    global_state.init_or_get_router()
+    global_state.init_or_get_metric_monitor(
+        gc_window_seconds=gc_window_seconds)
 
     if blocking:
-        global_state.wait_until_http_ready()
-        ray.get(global_state.router_actor_handle.is_ready.remote())
-        ray.get(global_state.kv_store_actor_handle.is_ready.remote())
-        ray.get(global_state.metric_monitor_handle.is_ready.remote())
+        block_until_http_ready("http://{}:{}".format(http_host, http_port))
 
 
-def create_endpoint(endpoint_name, route_expression, blocking=True):
+@_ensure_connected
+def create_endpoint(endpoint_name, route, blocking=True):
     """Create a service endpoint given route_expression.
 
     Args:
         endpoint_name (str): A name to associate to the endpoint. It will be
             used as key to set traffic policy.
-        route_expression (str): A string begin with "/". HTTP server will use
+        route (str): A string begin with "/". HTTP server will use
             the string to match the path.
         blocking (bool): If true, the function will wait for service to be
             registered before returning
     """
-    future = global_state.kv_store_actor_handle.register_service.remote(
-        route_expression, endpoint_name)
-    if blocking:
-        ray.get(future)
-    global_state.registered_endpoints.add(endpoint_name)
+    global_state.route_table.register_service(route, endpoint_name)
 
 
+@_ensure_connected
 def create_backend(func_or_class, backend_tag, *actor_init_args):
     """Create a backend using func_or_class and assign backend_tag.
 
@@ -91,45 +144,50 @@ def create_backend(func_or_class, backend_tag, *actor_init_args):
             "Backend must be a function or class, it is {}.".format(
                 type(func_or_class)))
 
-    global_state.backend_creators[backend_tag] = creator
-
-    global_state.registered_backends.add(backend_tag)
-
+    global_state.backend_table.register_backend(backend_tag, creator)
     scale(backend_tag, 1)
 
 
 def _start_replica(backend_tag):
-    assert backend_tag in global_state.registered_backends, (
+    assert backend_tag in global_state.backend_table.list_backends(), (
         "Backend {} is not registered.".format(backend_tag))
 
-    creator = global_state.backend_creators[backend_tag]
+    replica_tag = "{}#{}".format(backend_tag, get_random_letters(length=6))
+    creator = global_state.backend_table.get_backend_creator(backend_tag)
 
-    runner = creator()
-    setup_done = runner._ray_serve_setup.remote(
-        backend_tag, global_state.router_actor_handle, runner)
-    ray.get(setup_done)
-    runner._ray_serve_main_loop.remote()
+    # Create the runner in the nursery
+    [runner_handle] = ray.get(
+        global_state.actor_nursery_handle.start_actor_with_creator.remote(
+            creator, replica_tag))
 
-    global_state.backend_replicas[backend_tag].append(runner)
-    global_state.metric_monitor_handle.add_target.remote(runner)
+    # Setup the worker
+    ray.get(
+        runner_handle._ray_serve_setup.remote(
+            backend_tag, global_state.init_or_get_router(), runner_handle))
+    runner_handle._ray_serve_main_loop.remote()
+
+    # Register the worker in config tables as well as metric monitor
+    global_state.backend_table.add_replica(backend_tag, replica_tag)
+    global_state.init_or_get_metric_monitor().add_target.remote(runner_handle)
 
 
 def _remove_replica(backend_tag):
-    assert backend_tag in global_state.registered_backends, (
+    assert backend_tag in global_state.backend_table.list_backends(), (
         "Backend {} is not registered.".format(backend_tag))
-    assert len(global_state.backend_replicas[backend_tag]) > 0, (
+    assert len(global_state.backend_table.list_replicas(backend_tag)) > 0, (
         "Backend {} does not have enough replicas to be removed.".format(
             backend_tag))
 
-    replicas = global_state.backend_replicas[backend_tag]
-    oldest_replica_handle = replicas.popleft()
-
-    global_state.metric_monitor_handle.remove_target.remote(
-        oldest_replica_handle)
-    # explicitly terminate that actor
-    del oldest_replica_handle
+    replica_tag = global_state.backend_table.remove_replica(backend_tag)
+    [replica_handle] = ray.get(
+        global_state.actor_nursery_handle.get_handle.remote(replica_tag))
+    global_state.init_or_get_metric_monitor().remove_target.remote(
+        replica_handle)
+    ray.get(
+        global_state.actor_nursery_handle.remove_handle.remote(replica_tag))
 
 
+@_ensure_connected
 def scale(backend_tag, num_replicas):
     """Set the number of replicas for backend_tag.
 
@@ -137,11 +195,11 @@ def scale(backend_tag, num_replicas):
         backend_tag (str): A registered backend.
         num_replicas (int): Desired number of replicas
     """
-    assert backend_tag in global_state.registered_backends, (
+    assert backend_tag in global_state.backend_table.list_backends(), (
         "Backend {} is not registered.".format(backend_tag))
     assert num_replicas > 0, "Number of replicas must be greater than 1."
 
-    replicas = global_state.backend_replicas[backend_tag]
+    replicas = global_state.backend_table.list_replicas(backend_tag)
     current_num_replicas = len(replicas)
     delta_num_replicas = num_replicas - current_num_replicas
 
@@ -153,6 +211,7 @@ def scale(backend_tag, num_replicas):
             _remove_replica(backend_tag)
 
 
+@_ensure_connected
 def link(endpoint_name, backend_tag):
     """Associate a service endpoint with backend tag.
 
@@ -165,12 +224,10 @@ def link(endpoint_name, backend_tag):
 
     >>> serve.split("service-name", {"backend:v1": 1.0})
     """
-    assert endpoint_name in global_state.registered_endpoints
-
-    global_state.router_actor_handle.link.remote(endpoint_name, backend_tag)
-    global_state.policy_action_history[endpoint_name].append({backend_tag: 1})
+    split(endpoint_name, {backend_tag: 1.0})
 
 
+@_ensure_connected
 def split(endpoint_name, traffic_policy_dictionary):
     """Associate a service endpoint with traffic policy.
 
@@ -186,53 +243,27 @@ def split(endpoint_name, traffic_policy_dictionary):
         traffic_policy_dictionary (dict): a dictionary maps backend names
             to their traffic weights. The weights must sum to 1.
     """
-
-    # Perform dictionary checks
-    assert endpoint_name in global_state.registered_endpoints
+    assert endpoint_name in global_state.route_table.list_service().values()
 
     assert isinstance(traffic_policy_dictionary,
                       dict), "Traffic policy must be dictionary"
     prob = 0
     for backend, weight in traffic_policy_dictionary.items():
         prob += weight
-        assert (backend in global_state.registered_backends
+        assert (backend in global_state.backend_table.list_backends()
                 ), "backend {} is not registered".format(backend)
     assert np.isclose(
         prob, 1,
         atol=0.02), "weights must sum to 1, currently it sums to {}".format(
             prob)
 
-    global_state.router_actor_handle.set_traffic.remote(
+    global_state.policy_table.register_traffic_policy(
+        endpoint_name, traffic_policy_dictionary)
+    global_state.init_or_get_router().set_traffic.remote(
         endpoint_name, traffic_policy_dictionary)
-    global_state.policy_action_history[endpoint_name].append(
-        traffic_policy_dictionary)
-
-
-def rollback(endpoint_name):
-    """Rollback a traffic policy decision.
-
-    Args:
-        endpoint_name (str): A registered service endpoint.
-    """
-    assert endpoint_name in global_state.registered_endpoints
-    action_queues = global_state.policy_action_history[endpoint_name]
-    cur_policy, prev_policy = action_queues[-1], action_queues[-2]
-
-    logger.warning("""
-Current traffic policy is:
-{cur_policy}
-
-Will rollback to:
-{prev_policy}
-""".format(
-        cur_policy=pformat_color_json(cur_policy),
-        prev_policy=pformat_color_json(prev_policy)))
-
-    action_queues.pop()
-    global_state.router_actor_handle.set_traffic.remote(
-        endpoint_name, prev_policy)
 
 
+@_ensure_connected
 def get_handle(endpoint_name):
     """Retrieve RayServeHandle for service endpoint to invoke it from Python.
 
@@ -242,14 +273,15 @@ def get_handle(endpoint_name):
     Returns:
         RayServeHandle
     """
-    assert endpoint_name in global_state.registered_endpoints
+    assert endpoint_name in global_state.route_table.list_service().values()
 
     # Delay import due to it's dependency on global_state
     from ray.experimental.serve.handle import RayServeHandle
 
-    return RayServeHandle(global_state.router_actor_handle, endpoint_name)
+    return RayServeHandle(global_state.init_or_get_router(), endpoint_name)
 
 
+@_ensure_connected
 def stat(percentiles=[50, 90, 95],
          agg_windows_seconds=[10, 60, 300, 600, 3600]):
     """Retrieve metric statistics about ray serve system.
@@ -261,6 +293,5 @@ def stat(percentiles=[50, 90, 95],
             The longest aggregation window must be shorter or equal to the
             gc_window_seconds.
     """
-    return ray.get(
-        global_state.metric_monitor_handle.collect.remote(
-            percentiles, agg_windows_seconds))
+    return ray.get(global_state.init_or_get_metric_monitor().collect.remote(
+        percentiles, agg_windows_seconds))
diff --git a/python/ray/experimental/serve/constants.py b/python/ray/experimental/serve/constants.py
index e3aa5fc77..1b15c1e96 100644
--- a/python/ray/experimental/serve/constants.py
+++ b/python/ray/experimental/serve/constants.py
@@ -1,2 +1,17 @@
 #: The interval which http server refreshes its routing table
 HTTP_ROUTER_CHECKER_INTERVAL_S = 2
+
+#: Actor name used to register actor nursery
+SERVE_NURSERY_NAME = "SERVE_ACTOR_NURSERY"
+
+#: KVStore connector key in bootstrap config
+BOOTSTRAP_KV_STORE_CONN_KEY = "kv_store_connector"
+
+#: HTTP Address
+DEFAULT_HTTP_ADDRESS = "http://0.0.0.0:8000"
+
+#: HTTP Host
+DEFAULT_HTTP_HOST = "0.0.0.0"
+
+#: HTTP Port
+DEFAULT_HTTP_PORT = 8000
diff --git a/python/ray/experimental/serve/context.py b/python/ray/experimental/serve/context.py
index 43e1fd710..3221bfcc6 100644
--- a/python/ray/experimental/serve/context.py
+++ b/python/ray/experimental/serve/context.py
@@ -1,4 +1,5 @@
 from enum import IntEnum
+
 from ray.experimental.serve.exceptions import RayServeException
 
 
diff --git a/python/ray/experimental/serve/examples/echo_full.py b/python/ray/experimental/serve/examples/echo_full.py
index f8e939914..3d6e60276 100644
--- a/python/ray/experimental/serve/examples/echo_full.py
+++ b/python/ray/experimental/serve/examples/echo_full.py
@@ -2,11 +2,13 @@
 Full example of ray.serve module
 """
 
+import time
+
+import requests
+
 import ray
 import ray.experimental.serve as serve
 from ray.experimental.serve.utils import pformat_color_json
-import requests
-import time
 
 # initialize ray serve system.
 # blocking=True will wait for HTTP server to be ready to serve request.
diff --git a/python/ray/experimental/serve/global_state.py b/python/ray/experimental/serve/global_state.py
index 5fc733aff..be133e8f7 100644
--- a/python/ray/experimental/serve/global_state.py
+++ b/python/ray/experimental/serve/global_state.py
@@ -1,107 +1,134 @@
-import time
-from collections import defaultdict, deque
-
-import requests
-
 import ray
-from ray.experimental.serve.kv_store_service import KVStoreProxyActor
-from ray.experimental.serve.queues import CentralizedQueuesActor
-from ray.experimental.serve.utils import logger
-from ray.experimental.serve.server import HTTPActor
+from ray.experimental.serve.constants import (
+    BOOTSTRAP_KV_STORE_CONN_KEY, DEFAULT_HTTP_HOST, DEFAULT_HTTP_PORT,
+    SERVE_NURSERY_NAME)
+from ray.experimental.serve.kv_store_service import (
+    BackendTable, RoutingTable, TrafficPolicyTable)
 from ray.experimental.serve.metric import (MetricMonitor,
                                            start_metric_monitor_loop)
+from ray.experimental.serve.queues import CentralizedQueuesActor
+from ray.experimental.serve.server import HTTPActor
 
-# TODO(simon): Global state currently is designed to resides in the driver
-#     process. In the next iteration, we will move all mutable states into
-#     two actors: (1) namespaced key-value store backed by persistent store
-#     (2) actor supervisors holding all actor handles and is responsible
-#     for new actor instantiation and dead actor termination.
 
-LOG_PREFIX = "[Global State] "
+def start_initial_state(kv_store_connector):
+    nursery_handle = ActorNursery.remote()
+    ray.experimental.register_actor(SERVE_NURSERY_NAME, nursery_handle)
+
+    ray.get(
+        nursery_handle.store_bootstrap_state.remote(
+            BOOTSTRAP_KV_STORE_CONN_KEY, kv_store_connector))
+    return nursery_handle
+
+
+@ray.remote
+class ActorNursery:
+    """Initialize and store all actor handles.
+
+    Note:
+        This actor is necessary because ray will destory actors when the
+        original actor handle goes out of scope (when driver exit). Therefore
+        we need to initialize and store actor handles in a seperate actor.
+    """
+
+    def __init__(self):
+        # Dict: Actor handles -> tag
+        self.actor_handles = dict()
+
+        self.bootstrap_state = dict()
+
+    def start_actor(self, actor_cls, init_args, tag):
+        """Start an actor and add it to the nursery"""
+        handle = actor_cls.remote(*init_args)
+        self.actor_handles[handle] = tag
+        return [handle]
+
+    def start_actor_with_creator(self, creator, tag):
+        handle = creator()
+        self.actor_handles[handle] = tag
+        return [handle]
+
+    def get_all_handles(self):
+        return {tag: handle for handle, tag in self.actor_handles.items()}
+
+    def get_handle(self, actor_tag):
+        return [self.get_all_handles()[actor_tag]]
+
+    def remove_handle(self, actor_tag):
+        [handle] = self.get_handle(actor_tag)
+        self.actor_handles.pop(handle)
+        del handle
+
+    def store_bootstrap_state(self, key, value):
+        self.bootstrap_state[key] = value
+
+    def get_bootstrap_state_dict(self):
+        return self.bootstrap_state
 
 
 class GlobalState:
     """Encapsulate all global state in the serving system.
 
-    Warning:
-        Currently the state resides inside driver process. The state will be
-        moved into a key value stored service AND a supervisor service.
+    The information is fetch lazily from
+        1. A collection of namespaced key value stores
+        2. A actor supervisor service
     """
 
-    def __init__(self):
-        #: actor handle to KV store actor
-        self.kv_store_actor_handle = None
-        #: actor handle to HTTP server
-        self.http_actor_handle = None
-        #: actor handle the router actor
-        self.router_actor_handle = None
+    def __init__(self, actor_nursery_handle=None):
+        # Get actor nursery handle
+        if actor_nursery_handle is None:
+            actor_nursery_handle = ray.experimental.get_actor(
+                SERVE_NURSERY_NAME)
+        self.actor_nursery_handle = actor_nursery_handle
 
-        #: Set[str] list of backend names, used for deduplication
-        self.registered_backends = set()
-        #: Set[str] list of service endpoint names, used for deduplication
-        self.registered_endpoints = set()
+        # Connect to all the table
+        bootstrap_config = ray.get(
+            self.actor_nursery_handle.get_bootstrap_state_dict.remote())
+        kv_store_connector = bootstrap_config[BOOTSTRAP_KV_STORE_CONN_KEY]
+        self.route_table = RoutingTable(kv_store_connector)
+        self.backend_table = BackendTable(kv_store_connector)
+        self.policy_table = TrafficPolicyTable(kv_store_connector)
 
-        #: Mapping of endpoints -> a stack of traffic policy
-        self.policy_action_history = defaultdict(deque)
+        self.refresh_actor_handle_cache()
 
-        #: Backend creaters. Mapping backend_tag -> callable creator
-        self.backend_creators = dict()
-        #: Number of replicas per backend.
-        #  Mapping backend_tag -> deque(actor_handles)
-        self.backend_replicas = defaultdict(deque)
+    def refresh_actor_handle_cache(self):
+        self.actor_handle_cache = ray.get(
+            self.actor_nursery_handle.get_all_handles.remote())
 
-        #: HTTP address. Currently it's hard coded to localhost with port 8000
-        #  In future iteration, HTTP server will be started on every node and
-        #  use random/available port in a pre-defined port range. TODO(simon)
-        self.http_address = ""
+    def init_or_get_http_server(self,
+                                host=DEFAULT_HTTP_HOST,
+                                port=DEFAULT_HTTP_PORT):
+        if "http_server" not in self.actor_handle_cache:
+            [handle] = ray.get(
+                self.actor_nursery_handle.start_actor.remote(
+                    HTTPActor, init_args=(), tag="http_server"))
+            handle.run.remote(host=host, port=port)
+            self.refresh_actor_handle_cache()
+        return self.actor_handle_cache["http_server"]
 
-        #: Metric monitor handle
-        self.metric_monitor_handle = None
+    def init_or_get_router(self):
+        if "queue_actor" not in self.actor_handle_cache:
+            [handle] = ray.get(
+                self.actor_nursery_handle.start_actor.remote(
+                    CentralizedQueuesActor, init_args=(), tag="queue_actor"))
+            handle.register_self_handle.remote(handle)
+            self.refresh_actor_handle_cache()
 
-    def init_api_server(self):
-        logger.info(LOG_PREFIX + "Initalizing routing table")
-        self.kv_store_actor_handle = KVStoreProxyActor.remote()
-        logger.info((LOG_PREFIX + "Health checking routing table {}").format(
-            ray.get(self.kv_store_actor_handle.get_request_count.remote())), )
+        return self.actor_handle_cache["queue_actor"]
 
-    def init_http_server(self):
-        logger.info(LOG_PREFIX + "Initializing HTTP server")
-        self.http_actor_handle = HTTPActor.remote(self.kv_store_actor_handle,
-                                                  self.router_actor_handle)
-        self.http_actor_handle.run.remote(host="0.0.0.0", port=8000)
-        self.http_address = "http://localhost:8000"
+    def init_or_get_metric_monitor(self, gc_window_seconds=3600):
+        if "metric_monitor" not in self.actor_handle_cache:
+            [handle] = ray.get(
+                self.actor_nursery_handle.start_actor.remote(
+                    MetricMonitor,
+                    init_args=(gc_window_seconds, ),
+                    tag="metric_monitor"))
 
-    def init_router(self):
-        logger.info(LOG_PREFIX + "Initializing queuing system")
-        self.router_actor_handle = CentralizedQueuesActor.remote()
-        self.router_actor_handle.register_self_handle.remote(
-            self.router_actor_handle)
+            start_metric_monitor_loop.remote(handle)
 
-    def init_metric_monitor(self, gc_window_seconds=3600):
-        logger.info(LOG_PREFIX + "Initializing metric monitor")
-        self.metric_monitor_handle = MetricMonitor.remote(gc_window_seconds)
-        start_metric_monitor_loop.remote(self.metric_monitor_handle)
-        self.metric_monitor_handle.add_target.remote(self.router_actor_handle)
+            if "queue_actor" in self.actor_handle_cache:
+                handle.add_target.remote(
+                    self.actor_handle_cache["queue_actor"])
 
-    def wait_until_http_ready(self, num_retries=5, backoff_time_s=1):
-        http_is_ready = False
-        retries = num_retries
+            self.refresh_actor_handle_cache()
 
-        while not http_is_ready:
-            try:
-                resp = requests.get(self.http_address)
-                assert resp.status_code == 200
-                http_is_ready = True
-            except Exception:
-                pass
-
-            logger.debug((LOG_PREFIX + "Checking if HTTP server is ready."
-                          "{} retries left.").format(retries))
-            time.sleep(backoff_time_s)
-            # Exponential backoff
-            backoff_time_s *= 2
-            retries -= 1
-            if retries == 0:
-                raise Exception(
-                    "HTTP server not ready after {} retries.".format(
-                        num_retries))
+        return self.actor_handle_cache["metric_monitor"]
diff --git a/python/ray/experimental/serve/handle.py b/python/ray/experimental/serve/handle.py
index 4ce6a30e8..6102f8b0e 100644
--- a/python/ray/experimental/serve/handle.py
+++ b/python/ray/experimental/serve/handle.py
@@ -2,6 +2,7 @@ import ray
 from ray.experimental import serve
 from ray.experimental.serve.context import TaskContext
 from ray.experimental.serve.exceptions import RayServeException
+from ray.experimental.serve.constants import DEFAULT_HTTP_ADDRESS
 
 
 class RayServeHandle:
@@ -55,13 +56,13 @@ class RayServeHandle:
             return None
 
     def get_http_endpoint(self):
-        return serve.global_state.http_address
+        return DEFAULT_HTTP_ADDRESS
 
     def __repr__(self):
         return """
 RayServeHandle(
     Endpoint="{endpoint_name}",
-    URL="{http_endpoint}/{endpoint_name},
+    URL="{http_endpoint}/{endpoint_name}",
     Traffic={traffic_policy}
 )
 """.format(endpoint_name=self.endpoint_name,
diff --git a/python/ray/experimental/serve/http_util.py b/python/ray/experimental/serve/http_util.py
index 667aabd3c..16c877152 100644
--- a/python/ray/experimental/serve/http_util.py
+++ b/python/ray/experimental/serve/http_util.py
@@ -1,6 +1,7 @@
-import flask
 import io
 
+import flask
+
 
 def build_flask_request(asgi_scope_dict, request_body):
     """Build and return a flask request from ASGI payload
diff --git a/python/ray/experimental/serve/kv_store_service.py b/python/ray/experimental/serve/kv_store_service.py
index 2c7dffef4..423c572f7 100644
--- a/python/ray/experimental/serve/kv_store_service.py
+++ b/python/ray/experimental/serve/kv_store_service.py
@@ -1,7 +1,9 @@
 import json
+import sqlite3
 from abc import ABC
 
-import ray
+import cloudpickle as pickle
+
 import ray.experimental.internal_kv as ray_kv
 from ray.experimental.serve.utils import logger
 
@@ -128,9 +130,48 @@ class RayInternalKVStore(NamespacedKVStore):
         return data
 
 
-class KVStoreProxy:
-    def __init__(self, kv_class=InMemoryKVStore):
-        self.routing_table = kv_class(namespace="routes")
+class SQLiteKVStore(NamespacedKVStore):
+    def __init__(self, namespace, db_path):
+        self.namespace = namespace
+        self.conn = sqlite3.connect(db_path)
+
+        cursor = self.conn.cursor()
+        cursor.execute(
+            "CREATE TABLE IF NOT EXISTS {} (key TEXT UNIQUE, value TEXT)".
+            format(self.namespace))
+        self.conn.commit()
+
+    def put(self, key, value):
+        cursor = self.conn.cursor()
+        cursor.execute(
+            "INSERT OR REPLACE INTO {} (key, value) VALUES (?,?)".format(
+                self.namespace), (key, value))
+        self.conn.commit()
+
+    def get(self, key, default=None):
+        cursor = self.conn.cursor()
+        result = list(
+            cursor.execute(
+                "SELECT value FROM {} WHERE key = (?)".format(self.namespace),
+                (key, )))
+        if len(result) == 0:
+            return default
+        else:
+            # Due to UNIQUE constraint, there can only be one value.
+            value, *_ = result[0]
+            return value
+
+    def as_dict(self):
+        cursor = self.conn.cursor()
+        result = list(
+            cursor.execute("SELECT key, value FROM {}".format(self.namespace)))
+        return dict(result)
+
+
+# Tables
+class RoutingTable:
+    def __init__(self, kv_connector):
+        self.routing_table = kv_connector("routing_table")
         self.request_count = 0
 
     def register_service(self, route: str, service: str):
@@ -167,10 +208,45 @@ class KVStoreProxy:
         return self.request_count
 
 
-@ray.remote(num_cpus=0)
-class KVStoreProxyActor(KVStoreProxy):
-    def __init__(self, kv_class=RayInternalKVStore):
-        super().__init__(kv_class=kv_class)
+class BackendTable:
+    def __init__(self, kv_connector):
+        self.backend_table = kv_connector("backend_creator")
+        self.replica_table = kv_connector("replica_table")
 
-    def is_ready(self):
-        return True
+    def register_backend(self, backend_tag: str, backend_creator):
+        backend_creator_serialized = pickle.dumps(backend_creator)
+        self.backend_table.put(backend_tag, backend_creator_serialized)
+
+    def get_backend_creator(self, backend_tag):
+        return pickle.loads(self.backend_table.get(backend_tag))
+
+    def list_backends(self):
+        return list(self.backend_table.as_dict().keys())
+
+    def list_replicas(self, backend_tag: str):
+        return json.loads(self.replica_table.get(backend_tag, "[]"))
+
+    def add_replica(self, backend_tag: str, new_replica_tag: str):
+        replica_tags = self.list_replicas(backend_tag)
+        replica_tags.append(new_replica_tag)
+        self.replica_table.put(backend_tag, json.dumps(replica_tags))
+
+    def remove_replica(self, backend_tag):
+        replica_tags = self.list_replicas(backend_tag)
+        removed_replica = replica_tags.pop()
+        self.replica_table.put(backend_tag, json.dumps(replica_tags))
+        return removed_replica
+
+
+class TrafficPolicyTable:
+    def __init__(self, kv_connector):
+        self.traffic_policy_table = kv_connector("traffic_policy")
+
+    def register_traffic_policy(self, service_name, policy_dict):
+        self.traffic_policy_table.put(service_name, json.dumps(policy_dict))
+
+    def list_traffic_policy(self):
+        return {
+            service: json.loads(policy)
+            for service, policy in self.traffic_policy_table.as_dict()
+        }
diff --git a/python/ray/experimental/serve/metric.py b/python/ray/experimental/serve/metric.py
index b153261c2..901901489 100644
--- a/python/ray/experimental/serve/metric.py
+++ b/python/ray/experimental/serve/metric.py
@@ -1,9 +1,10 @@
 import time
 
-import ray
 import numpy as np
 import pandas as pd
 
+import ray
+
 
 @ray.remote(num_cpus=0)
 class MetricMonitor:
@@ -46,6 +47,7 @@ class MetricMonitor:
             handle._serve_metric.remote()
             for handle in self.actor_handles.values()
         ]
+        # TODO(simon): handle the possibility that an actor_handle is removed
         for handle_result in ray.get(result):
             for metric_name, metric_info in handle_result.items():
                 data_entry = {
diff --git a/python/ray/experimental/serve/server.py b/python/ray/experimental/serve/server.py
index f1ec15587..30f25b303 100644
--- a/python/ray/experimental/serve/server.py
+++ b/python/ray/experimental/serve/server.py
@@ -5,9 +5,9 @@ import uvicorn
 
 import ray
 from ray.experimental.async_api import _async_init, as_future
-from ray.experimental.serve.utils import BytesEncoder
 from ray.experimental.serve.constants import HTTP_ROUTER_CHECKER_INTERVAL_S
 from ray.experimental.serve.context import TaskContext
+from ray.experimental.serve.utils import BytesEncoder
 
 
 class JSONResponse:
@@ -55,21 +55,13 @@ class HTTPProxy:
     # blocks forever
     """
 
-    def __init__(self, kv_store_actor_handle, router_handle):
-        """
-        Args:
-            kv_store_actor_handle (ray.actor.ActorHandle): handle to routing
-               table actor. It will be used to populate routing table. It
-               should implement `handle.list_service()`
-            router_handle (ray.actor.ActorHandle): actor handle to push request
-               to. It should implement
-               `handle.enqueue_request.remote(endpoint, body)`
-        """
+    def __init__(self):
         assert ray.is_initialized()
 
-        self.admin_actor = kv_store_actor_handle
-        self.router = router_handle
-        self.route_table = dict()
+        # Delay import due to GlobalState depends on HTTP actor
+        from ray.experimental.serve.global_state import GlobalState
+        self.serve_global_state = GlobalState()
+        self.route_table_cache = dict()
 
         self.route_checker_should_shutdown = False
 
@@ -78,11 +70,8 @@ class HTTPProxy:
             if self.route_checker_should_shutdown:
                 return
 
-            try:
-                self.route_table = await as_future(
-                    self.admin_actor.list_service.remote())
-            except ray.exceptions.RayletError:  # Gracefully handle termination
-                return
+            self.route_table_cache = (
+                self.serve_global_state.route_table.list_service())
 
             await asyncio.sleep(interval)
 
@@ -122,10 +111,11 @@ class HTTPProxy:
         assert scope["type"] == "http"
         current_path = scope["path"]
         if current_path == "/":
-            await JSONResponse(self.route_table)(scope, receive, send)
+            await JSONResponse(self.route_table_cache)(scope, receive, send)
             return
 
-        if current_path not in self.route_table:
+        # TODO(simon): Use werkzeug route mapper to support variable path
+        if current_path not in self.route_table_cache:
             error_message = ("Path {} not found. "
                              "Please ping http://.../ for routing table"
                              ).format(current_path)
@@ -135,11 +125,12 @@ class HTTPProxy:
                 }, status_code=404)(scope, receive, send)
             return
 
-        endpoint_name = self.route_table[current_path]
+        endpoint_name = self.route_table_cache[current_path]
         http_body_bytes = await self.receive_http_body(scope, receive, send)
 
         result_object_id_bytes = await as_future(
-            self.router.enqueue_request.remote(
+            self.serve_global_state.init_or_get_router()
+            .enqueue_request.remote(
                 service=endpoint_name,
                 request_args=(scope, http_body_bytes),
                 request_kwargs=dict(),
@@ -157,8 +148,8 @@ class HTTPProxy:
 
 @ray.remote
 class HTTPActor:
-    def __init__(self, kv_store_actor_handle, router_handle):
-        self.app = HTTPProxy(kv_store_actor_handle, router_handle)
+    def __init__(self):
+        self.app = HTTPProxy()
 
     def run(self, host="0.0.0.0", port=8000):
         uvicorn.run(
diff --git a/python/ray/experimental/serve/task_runner.py b/python/ray/experimental/serve/task_runner.py
index eff3a9c5c..056442d88 100644
--- a/python/ray/experimental/serve/task_runner.py
+++ b/python/ray/experimental/serve/task_runner.py
@@ -1,9 +1,9 @@
-import traceback
 import time
+import traceback
 
 import ray
 from ray.experimental.serve import context as serve_context
-from ray.experimental.serve.context import TaskContext, FakeFlaskQuest
+from ray.experimental.serve.context import FakeFlaskQuest, TaskContext
 from ray.experimental.serve.http_util import build_flask_request
 
 
@@ -153,8 +153,6 @@ class TaskRunnerBackend(TaskRunner, RayServeMixin):
     for documentation purpose.
     """
 
-    pass
-
 
 @ray.remote
 class TaskRunnerActor(TaskRunnerBackend):
diff --git a/python/ray/experimental/serve/tests/conftest.py b/python/ray/experimental/serve/tests/conftest.py
index 688d7e976..69a757e46 100644
--- a/python/ray/experimental/serve/tests/conftest.py
+++ b/python/ray/experimental/serve/tests/conftest.py
@@ -1,3 +1,6 @@
+import os
+import tempfile
+
 import pytest
 
 import ray
@@ -6,8 +9,10 @@ from ray.experimental import serve
 
 @pytest.fixture(scope="session")
 def serve_instance():
-    serve.init(blocking=True)
+    _, new_db_path = tempfile.mkstemp(suffix=".test.db")
+    serve.init(kv_store_path=new_db_path, blocking=True)
     yield
+    os.remove(new_db_path)
 
 
 @pytest.fixture(scope="session")
diff --git a/python/ray/experimental/serve/tests/test_api.py b/python/ray/experimental/serve/tests/test_api.py
index 648aa6e6b..d4a974499 100644
--- a/python/ray/experimental/serve/tests/test_api.py
+++ b/python/ray/experimental/serve/tests/test_api.py
@@ -2,15 +2,14 @@ import time
 
 import requests
 
-import ray
 from ray.experimental import serve
 
 
 def test_e2e(serve_instance):
+    serve.init()  # so we have access to global state
     serve.create_endpoint("endpoint", "/api", blocking=True)
-    result = ray.get(
-        serve.global_state.kv_store_actor_handle.list_service.remote())
-    assert result == {"/api": "endpoint"}
+    result = serve.api._get_global_state().route_table.list_service()
+    assert result["/api"] == "endpoint"
 
     retry_count = 5
     timeout_sleep = 0.5
diff --git a/python/ray/experimental/serve/tests/test_metric.py b/python/ray/experimental/serve/tests/test_metric.py
index 427a3c4a8..fde360d00 100644
--- a/python/ray/experimental/serve/tests/test_metric.py
+++ b/python/ray/experimental/serve/tests/test_metric.py
@@ -2,7 +2,6 @@ import numpy as np
 import pytest
 
 import ray
-
 from ray.experimental.serve.metric import MetricMonitor
 
 
diff --git a/python/ray/experimental/serve/tests/test_persistence.py b/python/ray/experimental/serve/tests/test_persistence.py
new file mode 100644
index 000000000..037a79447
--- /dev/null
+++ b/python/ray/experimental/serve/tests/test_persistence.py
@@ -0,0 +1,37 @@
+import os
+import subprocess
+import tempfile
+
+import ray
+from ray.experimental import serve
+
+
+def test_new_driver(serve_instance):
+    script = """
+import ray
+ray.init(address="auto")
+
+from ray.experimental import serve
+serve.init()
+
+
+def function(flask_request):
+    return "OK!"
+
+serve.create_endpoint("driver", "/driver")
+serve.create_backend(function, "driver:v1")
+serve.link("driver", "driver:v1")
+"""
+
+    with tempfile.NamedTemporaryFile(mode="w", delete=False) as f:
+        path = f.name
+        f.write(script)
+
+    proc = subprocess.Popen(["python", path])
+    return_code = proc.wait(timeout=10)
+    assert return_code == 0
+
+    handle = serve.get_handle("driver")
+    assert ray.get(handle.remote()) == "OK!"
+
+    os.remove(path)
diff --git a/python/ray/experimental/serve/tests/test_routing.py b/python/ray/experimental/serve/tests/test_routing.py
index bc66cd136..7cbe237b2 100644
--- a/python/ray/experimental/serve/tests/test_routing.py
+++ b/python/ray/experimental/serve/tests/test_routing.py
@@ -1,5 +1,8 @@
-from ray.experimental.serve.kv_store_service import (InMemoryKVStore,
-                                                     RayInternalKVStore)
+import os
+import tempfile
+
+from ray.experimental.serve.kv_store_service import (
+    InMemoryKVStore, RayInternalKVStore, SQLiteKVStore)
 
 
 def test_default_in_memory_kv():
@@ -25,3 +28,27 @@ def test_ray_interal_kv(ray_instance):
     kv.put("1", 3)
     assert kv.get("1") == 3
     assert kv.as_dict() == {"1": 3}
+
+
+def test_sqlite_kv():
+    _, path = tempfile.mkstemp()
+
+    # Test get
+    kv = SQLiteKVStore("routing_table", db_path=path)
+    kv.put("/api", "api-endpoint")
+    assert kv.get("/api") == "api-endpoint"
+    assert kv.get("not-exist") is None
+
+    # Test namespace
+    kv2 = SQLiteKVStore("other_table", db_path=path)
+    kv2.put("/api", "api-endpoint-two")
+    assert kv2.get("/api") == "api-endpoint-two"
+
+    # Test as dict
+    assert kv.as_dict() == {"/api": "api-endpoint"}
+
+    # Test override
+    kv.put("/api", "api-new")
+    assert kv.get("/api") == "api-new"
+
+    os.remove(path)
diff --git a/python/ray/experimental/serve/tests/test_task_runner.py b/python/ray/experimental/serve/tests/test_task_runner.py
index 7746a93df..1d04f08d9 100644
--- a/python/ray/experimental/serve/tests/test_task_runner.py
+++ b/python/ray/experimental/serve/tests/test_task_runner.py
@@ -1,13 +1,10 @@
 import pytest
+
 import ray
+import ray.experimental.serve.context as context
 from ray.experimental.serve.queues import CentralizedQueuesActor
 from ray.experimental.serve.task_runner import (
-    RayServeMixin,
-    TaskRunner,
-    TaskRunnerActor,
-    wrap_to_ray_error,
-)
-import ray.experimental.serve.context as context
+    RayServeMixin, TaskRunner, TaskRunnerActor, wrap_to_ray_error)
 
 
 def test_runner_basic():
diff --git a/python/ray/experimental/serve/utils.py b/python/ray/experimental/serve/utils.py
index 60a0679fc..0e180952c 100644
--- a/python/ray/experimental/serve/utils.py
+++ b/python/ray/experimental/serve/utils.py
@@ -1,6 +1,10 @@
 import json
 import logging
+import random
+import string
+import time
 
+import requests
 from pygments import formatters, highlight, lexers
 
 
@@ -38,3 +42,29 @@ def pformat_color_json(d):
                               formatters.TerminalFormatter())
 
     return colorful_json
+
+
+def block_until_http_ready(http_endpoint, num_retries=5, backoff_time_s=1):
+    http_is_ready = False
+    retries = num_retries
+
+    while not http_is_ready:
+        try:
+            resp = requests.get(http_endpoint)
+            assert resp.status_code == 200
+            http_is_ready = True
+        except Exception:
+            pass
+
+        # Exponential backoff
+        time.sleep(backoff_time_s)
+        backoff_time_s *= 2
+
+        retries -= 1
+        if retries == 0:
+            raise Exception(
+                "HTTP server not ready after {} retries.".format(num_retries))
+
+
+def get_random_letters(length=6):
+    return "".join(random.choices(string.ascii_letters, k=length))