Xrayによるリクエストとレスポンスの実行時間の可視化

dev-requirements.txt

@@ -14,3 +14,6 @@ pytest==2.8.2
 pytest-cov==2.2.0
 python-coveralls==2.9.1
 pyzmq==14.4.1
+
+# X-Ray SDK
+aws-xray-sdk==2.12.0

mfr/extensions/unoconv/export.py

@@ -9,29 +9,33 @@
                                              UNOCONV_BIN,
                                              UNOCONV_TIMEOUT)
 
+from aws_xray_sdk.core import xray_recorder
 
 class UnoconvExporter(BaseExporter):
 
     def export(self):
-        try:
-            run([
-                UNOCONV_BIN,
-                '-n',
-                '-c', 'socket,host={},port={};urp;StarOffice.ComponentContext'.format(ADDRESS, PORT),
-                '-f', self.format,
-                '-o', self.output_file_path,
-                '-vvv',
-                self.source_file_path
-            ], check=True, timeout=UNOCONV_TIMEOUT)
-        except CalledProcessError as err:
-            name, extension = splitext(basename(self.source_file_path))
-            raise SubprocessError(
-                'Unable to export the file in the requested format, please try again later.',
-                process='unoconv',
-                cmd=str(err.cmd),
-                returncode=err.returncode,
-                path=str(self.source_file_path),
-                code=HTTPStatus.BAD_REQUEST,
-                extension=extension or '',
-                exporter_class='unoconv',
-            )
+        xray_recorder.begin_segment('UNOCONV')
+        with xray_recorder.in_subsegment(f'{ADDRESS}:{PORT}'):
+            try:
+                run([
+                    UNOCONV_BIN,
+                    '-n',
+                    '-c', 'socket,host={},port={};urp;StarOffice.ComponentContext'.format(ADDRESS, PORT),
+                    '-f', self.format,
+                    '-o', self.output_file_path,
+                    '-vvv',
+                    self.source_file_path
+                ], check=True, timeout=UNOCONV_TIMEOUT)
+            except CalledProcessError as err:
+                name, extension = splitext(basename(self.source_file_path))
+                raise SubprocessError(
+                    'Unable to export the file in the requested format, please try again later.',
+                    process='unoconv',
+                    cmd=str(err.cmd),
+                    returncode=err.returncode,
+                    path=str(self.source_file_path),
+                    code=HTTPStatus.BAD_REQUEST,
+                    extension=extension or '',
+                    exporter_class='unoconv',
+                )
+        xray_recorder.end_segment()

mfr/server/app.py

@@ -1,14 +1,3 @@
-import time
-import signal
-import asyncio
-import logging
-from functools import partial
-
-import tornado.web
-import tornado.httpserver
-import tornado.platform.asyncio
-from raven.contrib.tornado import AsyncSentryClient
-
 from mfr import settings
 from mfr.server import settings as server_settings
 from mfr.server.handlers.export import ExportHandler
@@ -17,8 +6,12 @@
 from mfr.server.handlers.exporters import ExportersHandler
 from mfr.server.handlers.renderers import RenderersHandler
 from mfr.server.handlers.core import ExtensionsStaticFileHandler
+from mfr.server.handlers.core import XrayStaticFileHandler
 from mfr.version import __version__
 
+from aws_xray_sdk.core import xray_recorder
+from aws_xray_sdk.core import patch_all
+
 logger = logging.getLogger(__name__)
 access_logger = logging.getLogger('tornado.access')
 
@@ -46,9 +39,18 @@ def almost_apache_style_log(handler):
 
 
 def make_app(debug):
+    xray_recorder.configure(
+        service='mfr',
+        daemon_address='192.168.168.167:2000',
+        sampling=False,
+        context_missing='LOG_ERROR',
+        plugins=('EC2Plugin',),
+        dynamic_naming='*.perfin.rdm.nii.ac.jp',
+    )
+    patch_all()
     app = tornado.web.Application(
         [
-            (r'/static/(.*)', tornado.web.StaticFileHandler, {'path': server_settings.STATIC_PATH}),
+            (r'/static/(.*)', XrayStaticFileHandler, {'path': server_settings.STATIC_PATH}),
             (r'/assets/(.*?)/(.*\..*)', ExtensionsStaticFileHandler),
             (r'/export', ExportHandler),
             (r'/exporters', ExportersHandler),
@@ -87,4 +89,4 @@ def serve():
 
     signal.signal(signal.SIGTERM, partial(sig_handler))
     asyncio.get_event_loop().set_debug(server_settings.DEBUG)
-    asyncio.get_event_loop().run_forever()
+    asyncio.get_event_loop().run_forever()

mfr/server/handlers/core.py

@@ -16,6 +16,8 @@
 from mfr.server import settings
 from mfr.core.metrics import MetricsRecord
 from mfr.core import utils, exceptions, remote_logging
+from aws_xray_sdk.core import xray_recorder
+from aws_xray_sdk.core.models import http
 
 CORS_ACCEPT_HEADERS = [
     'Range',
@@ -75,6 +77,31 @@ def options(self):
         if self.request.headers.get('Origin'):
             self.set_header('Access-Control-Allow-Methods', 'GET, PUT, POST, DELETE'),
 
+class XrayStaticFileHandler(tornado.web.StaticFileHandler):
+    def prepare(self):
+        self.segment = xray_recorder.begin_segment('mfr')
+        self.segment.put_http_meta(http.URL, self.request.full_url())
+        self.segment.put_http_meta(http.METHOD, self.request.method)
+        super().prepare()
+
+    def on_finish(self):
+        if hasattr(self, 'segment'):
+            self.segment.put_http_meta(http.STATUS, self.get_status())
+            xray_recorder.end_segment()
+        super().on_finish()
+
+class XrayHandler(tornado.web.RequestHandler):
+    def prepare(self):
+        self.segment = xray_recorder.begin_segment('mfr')
+        self.segment.put_http_meta(http.URL, self.request.full_url())
+        self.segment.put_http_meta(http.METHOD, self.request.method)
+        super().prepare()
+
+    def on_finish(self):
+        if hasattr(self, 'segment'):
+            self.segment.put_http_meta(http.STATUS, self.get_status())
+            xray_recorder.end_segment()
+        super().on_finish()
 
 class BaseHandler(CorsMixin, tornado.web.RequestHandler, SentryMixin):
     """Base class for the Render and Export handlers.  Fetches the file metadata for the file
@@ -102,6 +129,9 @@ async def prepare(self):
         """Builds an MFR provider instance, to which it passes the the ``url`` query parameter.
         From that, the file metadata is extracted.  Also builds cached waterbutler providers.
         """
+        self.segment = xray_recorder.begin_segment('mfr')
+        self.segment.put_http_meta(http.URL, self.request.full_url())
+        self.segment.put_http_meta(http.METHOD, self.request.method)
         if self.request.method == 'OPTIONS':
             return
 
@@ -215,6 +245,10 @@ def log_exception(self, typ, value, tb):
                                       exc_info=(typ, value, tb))
 
     def on_finish(self):
+        if hasattr(self, 'segment'):
+            self.segment.put_http_meta(http.STATUS, self.get_status())
+            xray_recorder.end_segment()
+
         if self.request.method not in self.ALLOWED_METHODS:
             return
 
@@ -277,7 +311,7 @@ def _all_metrics(self):
         return metrics
 
 
-class ExtensionsStaticFileHandler(tornado.web.StaticFileHandler, CorsMixin):
+class ExtensionsStaticFileHandler(XrayStaticFileHandler, CorsMixin):
     """Extensions static path definitions
     """
 
@@ -290,14 +324,16 @@ def initialize(self):
         }
 
     async def get(self, module_name, path):
-        try:
-            super().initialize(self.modules[module_name])
-            return await super().get(path)
-        except Exception:
-            self.set_status(404)
+        with xray_recorder.in_segment('get_module'):
+            try:
+                super().initialize(self.modules[module_name])
+                return await super().get(path)
+            except Exception:
+                self.set_status(404)
 
-        try:
-            super().initialize(settings.STATIC_PATH)
-            return await super().get(path)
-        except Exception:
-            self.set_status(404)
+        with xray_recorder.in_segment('get_static'):
+            try:
+                super().initialize(settings.STATIC_PATH)
+                return await super().get(path)
+            except Exception:
+                self.set_status(404)

mfr/server/handlers/export.py

@@ -8,6 +8,7 @@
 from mfr.core import utils
 from mfr.server import settings
 from mfr.server.handlers import core
+from aws_xray_sdk.core import xray_recorder
 
 logger = logging.getLogger(__name__)
 
@@ -18,6 +19,7 @@ class ExportHandler(core.BaseHandler):
     ALLOWED_METHODS = ['GET']
 
     async def prepare(self):
+        xray_recorder.begin_segment('mfr')
         if self.request.method not in self.ALLOWED_METHODS:
             return
 
@@ -57,49 +59,53 @@ async def prepare(self):
 
     async def get(self):
         """Export a file to the format specified via the associated extension library"""
-
-        # File is already in the requested format
-        if self.metadata.ext.lower() == ".{}".format(self.format.lower()):
-            await self.write_stream(await self.provider.download())
-            logger.info('Exported {} with no conversion.'.format(self.format))
-            self.metrics.add('export.conversion', 'noop')
-            return
-
-        if settings.CACHE_ENABLED:
-            try:
-                cached_stream = await self.cache_provider.download(self.cache_file_path)
-            except DownloadError as e:
-                assert e.code == 404, 'Non-404 DownloadError {!r}'.format(e)
-                logger.info('No cached file found; Starting export [{}]'.format(self.cache_file_path))
-                self.metrics.add('cache_file.result', 'miss')
-            else:
-                logger.info('Cached file found; Sending downstream [{}]'.format(self.cache_file_path))
-                self.metrics.add('cache_file.result', 'hit')
+        with xray_recorder.in_subsegment('Export File'):
+            # File is already in the requested format
+            if self.metadata.ext.lower() == ".{}".format(self.format.lower()):
+                await self.write_stream(await self.provider.download())
+                logger.info('Exported {} with no conversion.'.format(self.format))
+                self.metrics.add('export.conversion', 'noop')
+                return
+
+        with xray_recorder.in_subsegment('Cache Check'):
+            if settings.CACHE_ENABLED:
+                try:
+                    cached_stream = await self.cache_provider.download(self.cache_file_path)
+                except DownloadError as e:
+                    assert e.code == 404, 'Non-404 DownloadError {!r}'.format(e)
+                    logger.info('No cached file found; Starting export [{}]'.format(self.cache_file_path))
+                    self.metrics.add('cache_file.result', 'miss')
+                else:
+                    logger.info('Cached file found; Sending downstream [{}]'.format(self.cache_file_path))
+                    self.metrics.add('cache_file.result', 'hit')
+                    self._set_headers()
+                    return await self.write_stream(cached_stream)
+
+        with xray_recorder.in_subsegment('Upload File'):
+            await self.local_cache_provider.upload(
+                await self.provider.download(),
+                self.source_file_path
+            )
+
+        with xray_recorder.in_subsegment('Exporter Function'):
+            exporter = utils.make_exporter(
+                self.metadata.ext,
+                self.source_file_path.full_path,
+                self.output_file_path.full_path,
+                self.format,
+                self.metadata,
+            )
+
+            self.extension_metrics.add('class', exporter._get_module_name())
+
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(None, exporter.export)
+            self.exporter_metrics = exporter.exporter_metrics
+
+        with xray_recorder.in_subsegment('Write Stream'):
+            with open(self.output_file_path.full_path, 'rb') as fp:
                 self._set_headers()
-                return await self.write_stream(cached_stream)
-
-        await self.local_cache_provider.upload(
-            await self.provider.download(),
-            self.source_file_path
-        )
-
-        exporter = utils.make_exporter(
-            self.metadata.ext,
-            self.source_file_path.full_path,
-            self.output_file_path.full_path,
-            self.format,
-            self.metadata,
-        )
-
-        self.extension_metrics.add('class', exporter._get_module_name())
-
-        loop = asyncio.get_event_loop()
-        await loop.run_in_executor(None, exporter.export)
-        self.exporter_metrics = exporter.exporter_metrics
-
-        with open(self.output_file_path.full_path, 'rb') as fp:
-            self._set_headers()
-            await self.write_stream(waterbutler.core.streams.FileStreamReader(fp))
+                await self.write_stream(waterbutler.core.streams.FileStreamReader(fp))
 
     async def _cache_and_clean(self):
         if settings.CACHE_ENABLED and os.path.exists(self.output_file_path.full_path):
@@ -121,3 +127,6 @@ def _set_headers(self):
         self.set_header('Content-Disposition', 'attachment;filename="{}"'.format('{}.{}'.format(self.metadata.name.replace('"', '\\"'), self.format)))
         if self.metadata.content_type:
             self.set_header('Content-Type', self.metadata.content_type)
+
+    def on_finish(self):
+        xray_recorder.end_segment()

mfr/server/handlers/exporters.py

@@ -1,8 +1,9 @@
 import pkg_resources
 import tornado.web
+from mfr.server.handlers import core
 
 
-class ExportersHandler(tornado.web.RequestHandler):
+class ExportersHandler(core.XrayHandler):
 
     def get(self):
         """List available exporters"""