Merge pull request #933 from mozilla-services/refactor/632

refactor: begin tearing apart AutopushSettings

autopush/base.py

@@ -1,21 +1,41 @@
 import sys
 import uuid
+from typing import TYPE_CHECKING
 
 import cyclone.web
 from twisted.logger import Logger
 from twisted.python import failure
 
+if TYPE_CHECKING:  # pragma: nocover
+    from autopush.db import DatabaseManager  # noqa
+    from autopush.metrics import IMetrics  # noqa
+    from autopush.settings import AutopushSettings  # noqa
+
 
 class BaseHandler(cyclone.web.RequestHandler):
     """Base cyclone RequestHandler for autopush"""
 
     log = Logger()
 
-    def initialize(self, ap_settings):
-        """Setup basic attributes from AutopushSettings"""
-        self.ap_settings = ap_settings
+    def initialize(self):
+        """Initialize info from the client"""
         self._client_info = self._init_info()
 
+    @property
+    def ap_settings(self):
+        # type: () -> AutopushSettings
+        return self.application.ap_settings
+
+    @property
+    def db(self):
+        # type: () -> DatabaseManager
+        return self.application.db
+
+    @property
+    def metrics(self):
+        # type: () -> IMetrics
+        return self.db.metrics
+
     def _init_info(self):
         return dict(
             ami_id=self.ap_settings.ami_id,

autopush/db.py

@@ -37,6 +37,11 @@
 import uuid
 from functools import wraps
 
+from attr import (
+    attrs,
+    attrib,
+    Factory
+)
 from boto.exception import JSONResponseError, BotoServerError
 from boto.dynamodb2.exceptions import (
     ConditionalCheckFailedException,
@@ -51,14 +56,19 @@
     Any,
     Callable,
     Dict,
+    Generator,
     Iterable,
     List,
     Optional,
     Set,
     TypeVar,
     Tuple,
 )
+from twisted.internet.defer import Deferred  # noqa
+from twisted.internet.defer import inlineCallbacks, returnValue
+from twisted.internet.threads import deferToThread
 
+import autopush.metrics
 from autopush.exceptions import AutopushException
 from autopush.metrics import IMetrics  # noqa
 from autopush.types import ItemLike  # noqa
@@ -853,3 +863,136 @@ def clear_node(self, item):
             return True
         except ConditionalCheckFailedException:
             return False
+
+
+@attrs
+class DatabaseManager(object):
+    """Provides database access"""
+
+    storage = attrib()  # type: Storage
+    router = attrib()   # type: Router
+
+    metrics = attrib()  # type: IMetrics
+
+    message_tables = attrib(default=Factory(dict))  # type: Dict[str, Message]
+    current_msg_month = attrib(default=None)        # type: Optional[str]
+    current_month = attrib(default=None)            # type: Optional[int]
+
+    _message_prefix = attrib(default="message")  # type: str
+
+    @classmethod
+    def from_settings(cls, settings):
+        router_table = get_router_table(
+            settings.router_tablename,
+            settings.router_read_throughput,
+            settings.router_write_throughput
+        )
+        storage_table = get_storage_table(
+            settings.storage_tablename,
+            settings.storage_read_throughput,
+            settings.storage_write_throughput
+        )
+        get_rotating_message_table(
+            settings.message_tablename,
+            message_read_throughput=settings.message_read_throughput,
+            message_write_throughput=settings.message_write_throughput
+        )
+        metrics = autopush.metrics.from_settings(settings)
+        return cls(
+            storage=Storage(storage_table, metrics),
+            router=Router(router_table, metrics),
+            message_prefix=settings.message_tablename,
+            metrics=metrics
+        )
+
+    def setup(self, preflight_uaid):
+        # type: (str) -> None
+        """Setup metrics, message tables and perform preflight_check"""
+        self.metrics.start()
+
+        # Used to determine whether a connection is out of date with current
+        # db objects. There are three noteworty cases:
+        # 1 "Last Month" the table requires a rollover.
+        # 2 "This Month" the most common case.
+        # 3 "Next Month" where the system will soon be rolling over, but with
+        #   timing, some nodes may roll over sooner. Ensuring the next month's
+        #   table is present before the switchover is the main reason for this,
+        #   just in case some nodes do switch sooner.
+        self.create_initial_message_tables()
+
+        preflight_check(self.storage, self.router, preflight_uaid)
+
+    @property
+    def message(self):
+        # type: () -> Message
+        """Property that access the current message table"""
+        return self.message_tables[self.current_msg_month]
+
+    @message.setter
+    def message(self, value):
+        # type: (Message) -> None
+        """Setter to set the current message table"""
+        self.message_tables[self.current_msg_month] = value
+
+    def _tomorrow(self):
+        # type: () -> datetime.date
+        return datetime.date.today() + datetime.timedelta(days=1)
+
+    def create_initial_message_tables(self):
+        """Initializes a dict of the initial rotating messages tables.
+
+        An entry for last months table, an entry for this months table,
+        an entry for tomorrow, if tomorrow is a new month.
+
+        """
+        today = datetime.date.today()
+        last_month = get_rotating_message_table(self._message_prefix, -1)
+        this_month = get_rotating_message_table(self._message_prefix)
+        self.current_month = today.month
+        self.current_msg_month = this_month.table_name
+        self.message_tables = {
+            last_month.table_name: Message(last_month, self.metrics),
+            this_month.table_name: Message(this_month, self.metrics)
+        }
+        if self._tomorrow().month != today.month:
+            next_month = get_rotating_message_table(self._message_prefix,
+                                                    delta=1)
+            self.message_tables[next_month.table_name] = Message(
+                next_month, self.metrics)
+
+    @inlineCallbacks
+    def update_rotating_tables(self):
+        # type: () -> Generator
+        """This method is intended to be tasked to run periodically off the
+        twisted event hub to rotate tables.
+
+        When today is a new month from yesterday, then we swap out all the
+        table objects on the settings object.
+
+        """
+        today = datetime.date.today()
+        tomorrow = self._tomorrow()
+        if ((tomorrow.month != today.month) and
+                sorted(self.message_tables.keys())[-1] != tomorrow.month):
+            next_month = yield deferToThread(
+                get_rotating_message_table,
+                self._message_prefix, 0, tomorrow
+            )
+            self.message_tables[next_month.table_name] = Message(
+                next_month, self.metrics)
+
+        if today.month == self.current_month:
+            # No change in month, we're fine.
+            returnValue(False)
+
+        # Get tables for the new month, and verify they exist before we try to
+        # switch over
+        message_table = yield deferToThread(get_rotating_message_table,
+                                            self._message_prefix)
+
+        # Both tables found, safe to switch-over
+        self.current_month = today.month
+        self.current_msg_month = message_table.table_name
+        self.message_tables[self.current_msg_month] = Message(
+            message_table, self.metrics)
+        returnValue(True)

autopush/diagnostic_cli.py

@@ -6,6 +6,7 @@
 import configargparse
 from twisted.logger import Logger
 
+from autopush.db import DatabaseManager
 from autopush.main import AutopushMultiService
 from autopush.main_argparse import add_shared_args
 from autopush.settings import AutopushSettings
@@ -19,12 +20,15 @@ class EndpointDiagnosticCLI(object):
 
     def __init__(self, sysargs, use_files=True):
         args = self._load_args(sysargs, use_files)
-        self._settings = AutopushSettings(
+        self._settings = settings = AutopushSettings(
             crypto_key=args.crypto_key,
             router_tablename=args.router_tablename,
             storage_tablename=args.storage_tablename,
             message_tablename=args.message_tablename,
+            statsd_host=None,
         )
+        self.db = DatabaseManager.from_settings(settings)
+        self.db.setup(settings.preflight_uaid)
         self._endpoint = args.endpoint
         self._pp = pprint.PrettyPrinter(indent=4)
 
@@ -56,20 +60,21 @@ def run(self):
         api_ver, token = md.get("api_ver", "v1"), md["token"]
 
         parsed = self._settings.parse_endpoint(
+            self.db.metrics,
             token=token,
             version=api_ver,
         )
         uaid, chid = parsed["uaid"], parsed["chid"]
 
         print("UAID: {}\nCHID: {}\n".format(uaid, chid))
 
-        rec = self._settings.router.get_uaid(uaid)
+        rec = self.db.router.get_uaid(uaid)
         print("Router record:")
         self._pp.pprint(rec._data)
         print("\n")
 
         mess_table = rec["current_month"]
-        chans = self._settings.message_tables[mess_table].all_channels(uaid)
+        chans = self.db.message_tables[mess_table].all_channels(uaid)
         print("Channels in message table:")
         self._pp.pprint(chans)
 

autopush/http.py

@@ -2,6 +2,7 @@
 from typing import (  # noqa
     Any,
     Callable,
+    Dict,
     Optional,
     Sequence,
     Tuple,
@@ -11,6 +12,9 @@
 import cyclone.web
 
 from autopush.base import BaseHandler
+from autopush.db import DatabaseManager
+from autopush.router import routers_from_settings
+from autopush.router.interface import IRouter  # noqa
 from autopush.settings import AutopushSettings  # noqa
 from autopush.ssl import AutopushSSLContextFactory
 from autopush.web.health import (
@@ -53,45 +57,47 @@ class BaseHTTPFactory(cyclone.web.Application):
     )
 
     def __init__(self,
-                 ap_settings,
-                 handlers=None,
-                 log_function=skip_request_logging,
+                 ap_settings,    # type: AutopushSettings
+                 db,             # type: DatabaseManager
+                 routers,        # type: Dict[str, IRouter]
+                 handlers=None,  # type: APHandlers
+                 log_function=skip_request_logging,  # type: CycloneLogger
                  **kwargs):
-        # type: (AutopushSettings, APHandlers, CycloneLogger, **Any) -> None
+        # type: (...) -> None
         self.ap_settings = ap_settings
+        self.db = db
+        self.routers = routers
         self.noisy = ap_settings.debug
 
         cyclone.web.Application.__init__(
             self,
+            handlers=self.ap_handlers if handlers is None else handlers,
             default_host=self._hostname,
             debug=ap_settings.debug,
             log_function=log_function,
             **kwargs
         )
-        self.add_ap_handlers(
-            self.ap_handlers if handlers is None else handlers)
-
-    def add_ap_handlers(self, handlers):
-        # type: (APHandlers) -> None
-        """Add BaseHandlers w/ their appropriate handler kwargs"""
-        h_kwargs = dict(ap_settings=self.ap_settings)
-        self.add_handlers(
-            ".*$",
-            [(pattern, handler, h_kwargs) for pattern, handler in handlers]
-        )
 
     def add_health_handlers(self):
         """Add the health check HTTP handlers"""
-        self.add_ap_handlers(self.health_ap_handlers)
+        self.add_handlers(".*$", self.health_ap_handlers)
 
     @property
     def _hostname(self):
         return self.ap_settings.hostname
 
     @classmethod
-    def for_handler(cls, handler_cls, *args, **kwargs):
-        # type: (Type[BaseHandler], *Any, **Any) -> BaseHTTPFactory
-        """Create a cyclone app around a specific handler_cls.
+    def for_handler(cls,
+                    handler_cls,    # Type[BaseHTTPFactory]
+                    ap_settings,    # type: AutopushSettings
+                    db=None,        # type: Optional[DatabaseManager]
+                    routers=None,   # type: Optional[Dict[str, IRouter]]
+                    **kwargs):
+        # type: (...) -> BaseHTTPFactory
+        """Create a cyclone app around a specific handler_cls for tests.
+
+        Creates an uninitialized (no setup() called) DatabaseManager
+        from settings if one isn't specified.
 
         handler_cls must be included in ap_handlers or a ValueError is
         thrown.
@@ -101,7 +107,17 @@ def for_handler(cls, handler_cls, *args, **kwargs):
             raise ValueError("handler_cls incompatibile with handlers kwarg")
         for pattern, handler in cls.ap_handlers + cls.health_ap_handlers:
             if handler is handler_cls:
-                return cls(handlers=[(pattern, handler)], *args, **kwargs)
+                if db is None:
+                    db = DatabaseManager.from_settings(ap_settings)
+                if routers is None:
+                    routers = routers_from_settings(ap_settings, db)
+                return cls(
+                    ap_settings,
+                    db=db,
+                    routers=routers,
+                    handlers=[(pattern, handler)],
+                    **kwargs
+                )
         raise ValueError("{!r} not in ap_handlers".format(
             handler_cls))  # pragma: nocover