Add pagination to history and admin logs

src/execution/logging.py

@@ -117,6 +117,7 @@ def __init__(self, output_folder, log_name_creator, authorizer):
 
         self._visited_files = set()
         self._ids_to_file_map = {}
+        self._entries_cache = {}
         self._output_loggers = {}
 
         file_utils.prepare_folder(output_folder)
@@ -183,33 +184,34 @@ def write_post_execution_info(self, execution_id, exit_code):
 
         log_file_path = os.path.join(self._output_folder, filename)
 
-        logger.set_close_callback(lambda: self._write_post_execution_info(log_file_path, exit_code))
+        def close_cb():
+            self._write_post_execution_info(log_file_path, exit_code)
+            cached_entry = self._entries_cache.get(execution_id)
+            if cached_entry:
+                cached_entry.exit_code = int(exit_code)
+
+        logger.set_close_callback(close_cb)
 
     def get_history_entries(self, user_id, *, system_call=False):
         self._renew_files_cache()
 
         result = []
 
-        for file in self._ids_to_file_map.values():
-            history_entry = self._extract_history_entry(file)
-            if history_entry is not None and self._can_access_entry(history_entry, user_id, system_call):
-                result.append(history_entry)
+        for entry in self._entries_cache.values():
+            if entry is not None and self._can_access_entry(entry, user_id, system_call):
+                result.append(entry)
 
         return result
 
     def find_history_entry(self, execution_id, user_id):
         self._renew_files_cache()
 
-        file = self._ids_to_file_map.get(execution_id)
-        if file is None:
+        entry = self._entries_cache.get(execution_id)
+        if entry is None:
             LOGGER.warning('find_history_entry: file for %s id not found', execution_id)
             return None
 
-        entry = self._extract_history_entry(file)
-        if entry is None:
-            LOGGER.warning('find_history_entry: cannot parse file for %s', execution_id)
-
-        elif not self._can_access_entry(entry, user_id):
+        if not self._can_access_entry(entry, user_id):
             message = 'User ' + user_id + ' has no access to execution #' + str(execution_id)
             LOGGER.warning('%s. Original user: %s', message, entry.user_id)
             raise AccessProhibitedException(message)
@@ -261,6 +263,7 @@ def _renew_files_cache(self):
         for obsolete_id in obsolete_ids:
             LOGGER.info('Logs for execution #' + obsolete_id + ' were deleted')
             del cache[obsolete_id]
+            self._entries_cache.pop(obsolete_id, None)
 
         for file in os.listdir(self._output_folder):
             if not file.lower().endswith('.log'):
@@ -276,6 +279,7 @@ def _renew_files_cache(self):
                 continue
 
             cache[entry.id] = file
+            self._entries_cache[entry.id] = entry
 
     @staticmethod
     def _create_log_identifier(audit_name, script_name, start_time):

src/tests/web/server_test.py

@@ -313,6 +313,174 @@ def start_server(self, port, address, *, xsrf_protection=XSRF_PROTECTION_TOKEN):
         self._admin_session.cookies['username'] = create_signed_value(cookie_secret, 'username', 'admin_user') \
             .decode('utf8')
 
+    def _create_mock_history_entry(self, entry_id, user_name='normal_user', script_name='script1', start_time_ms=None, exit_code=0):
+        from execution.logging import HistoryEntry
+        from datetime import datetime, timezone
+        entry = HistoryEntry()
+        entry.id = str(entry_id)
+        entry.user_name = user_name
+        entry.user_id = user_name
+        entry.script_name = script_name
+        entry.command = 'python script.py'
+        entry.output_format = 'terminal'
+        entry.exit_code = exit_code
+        if start_time_ms is not None:
+            entry.start_time = datetime.fromtimestamp(start_time_ms / 1000.0, tz=timezone.utc)
+        return entry
+
+    def test_history_short_log_pagination_empty(self):
+        self.start_server(12345, '127.0.0.1')
+        server._tornado_app.execution_logging_service.get_history_entries.return_value = []
+
+        response = self.request('GET', 'http://127.0.0.1:12345/history/execution_log/short?page=1&size=25')
+        self.assertEqual(response, {
+            'records': [],
+            'total': 0,
+            'page': 1,
+            'pageSize': 25,
+            'totalPages': 1
+        })
+
+    def test_history_short_log_pagination_slicing_and_sorting(self):
+        self.start_server(12345, '127.0.0.1')
+        entries = [
+            self._create_mock_history_entry('e1', start_time_ms=1000),
+            self._create_mock_history_entry('e2', start_time_ms=3000),
+            self._create_mock_history_entry('e3', start_time_ms=2000),
+        ]
+        server._tornado_app.execution_logging_service.get_history_entries.return_value = entries
+
+        page1 = self.request('GET', 'http://127.0.0.1:12345/history/execution_log/short?page=1&size=10')
+        self.assertEqual(page1['total'], 3)
+        self.assertEqual(page1['page'], 1)
+        self.assertEqual(page1['pageSize'], 10)
+        self.assertEqual(page1['totalPages'], 1)
+        self.assertEqual([r['id'] for r in page1['records']], ['e2', 'e3', 'e1'])
+
+    def test_history_short_log_pagination_multiple_pages(self):
+        self.start_server(12345, '127.0.0.1')
+        entries = [self._create_mock_history_entry(f'e{i}', start_time_ms=i * 1000) for i in range(1, 35)]
+        server._tornado_app.execution_logging_service.get_history_entries.return_value = entries
+
+        res_p1 = self.request('GET', 'http://127.0.0.1:12345/history/execution_log/short?page=1&size=10')
+        self.assertEqual(res_p1['total'], 34)
+        self.assertEqual(res_p1['page'], 1)
+        self.assertEqual(res_p1['pageSize'], 10)
+        self.assertEqual(res_p1['totalPages'], 4)
+        self.assertEqual(len(res_p1['records']), 10)
+        self.assertEqual(res_p1['records'][0]['id'], 'e34')
+
+        res_p4 = self.request('GET', 'http://127.0.0.1:12345/history/execution_log/short?page=4&size=10')
+        self.assertEqual(res_p4['page'], 4)
+        self.assertEqual(len(res_p4['records']), 4)
+        self.assertEqual(res_p4['records'][-1]['id'], 'e1')
+
+    def test_history_short_log_pagination_invalid_size_fallback(self):
+        self.start_server(12345, '127.0.0.1')
+        server._tornado_app.execution_logging_service.get_history_entries.return_value = []
+
+        response = self.request('GET', 'http://127.0.0.1:12345/history/execution_log/short?page=1&size=999')
+        self.assertEqual(response['pageSize'], 25)
+
+        response_abc = self.request('GET', 'http://127.0.0.1:12345/history/execution_log/short?page=1&size=abc')
+        self.assertEqual(response_abc['pageSize'], 25)
+
+    def test_history_short_log_pagination_invalid_page_fallback(self):
+        self.start_server(12345, '127.0.0.1')
+        server._tornado_app.execution_logging_service.get_history_entries.return_value = []
+
+        response = self.request('GET', 'http://127.0.0.1:12345/history/execution_log/short?page=-5&size=25')
+        self.assertEqual(response['page'], 1)
+
+        response_invalid = self.request('GET', 'http://127.0.0.1:12345/history/execution_log/short?page=xyz&size=25')
+        self.assertEqual(response_invalid['page'], 1)
+
+    def test_history_short_log_legacy_unpaginated(self):
+        self.start_server(12345, '127.0.0.1')
+        entries = [self._create_mock_history_entry('e1', start_time_ms=1000)]
+        server._tornado_app.execution_logging_service.get_history_entries.return_value = entries
+
+        response = self.request('GET', 'http://127.0.0.1:12345/history/execution_log/short')
+        self.assertIsInstance(response, list)
+        self.assertEqual(len(response), 1)
+        self.assertEqual(response[0]['id'], 'e1')
+
+    def test_history_short_log_pagination_page_out_of_bounds(self):
+        self.start_server(12345, '127.0.0.1')
+        entries = [self._create_mock_history_entry(f'e{i}', start_time_ms=i * 1000) for i in range(1, 31)]
+        server._tornado_app.execution_logging_service.get_history_entries.return_value = entries
+
+        response = self.request('GET', 'http://127.0.0.1:12345/history/execution_log/short?page=99&size=10')
+        self.assertEqual(response['page'], 3)
+        self.assertEqual(response['totalPages'], 3)
+        self.assertEqual(len(response['records']), 10)
+
+    def test_history_short_log_pagination_with_none_start_time(self):
+        self.start_server(12345, '127.0.0.1')
+        entries = [
+            self._create_mock_history_entry('e_none', start_time_ms=None),
+            self._create_mock_history_entry('e_valid', start_time_ms=5000)
+        ]
+        server._tornado_app.execution_logging_service.get_history_entries.return_value = entries
+
+        response = self.request('GET', 'http://127.0.0.1:12345/history/execution_log/short?page=1&size=10')
+        self.assertEqual([r['id'] for r in response['records']], ['e_valid', 'e_none'])
+
+    def test_history_short_log_pagination_all_supported_sizes(self):
+        self.start_server(12345, '127.0.0.1')
+        server._tornado_app.execution_logging_service.get_history_entries.return_value = []
+
+        for size in [10, 25, 50, 100, 250, 500]:
+            response = self.request('GET', f'http://127.0.0.1:12345/history/execution_log/short?page=1&size={size}')
+            self.assertEqual(response['pageSize'], size)
+
+    def test_history_short_log_pagination_running_script_status(self):
+        self.start_server(12345, '127.0.0.1')
+        entries = [
+            self._create_mock_history_entry('running_1', start_time_ms=2000, exit_code=None),
+            self._create_mock_history_entry('finished_1', start_time_ms=1000, exit_code=0)
+        ]
+        server._tornado_app.execution_logging_service.get_history_entries.return_value = entries
+        server._tornado_app.execution_service.is_running.side_effect = lambda entry_id, user: entry_id == 'running_1'
+
+        response = self.request('GET', 'http://127.0.0.1:12345/history/execution_log/short?page=1&size=10')
+        records = response['records']
+        self.assertEqual(records[0]['id'], 'running_1')
+        self.assertEqual(records[0]['status'], 'running')
+        self.assertEqual(records[1]['id'], 'finished_1')
+        self.assertEqual(records[1]['status'], 'finished')
+
+    def test_isolated_paginate_empty(self):
+        result = server.paginate_history_entries([], '1', '25')
+        self.assertEqual(result, {
+            'records': [],
+            'total': 0,
+            'page': 1,
+            'pageSize': 25,
+            'totalPages': 1
+        })
+
+    def test_isolated_paginate_legacy_unpaginated_returns_list(self):
+        entries = [self._create_mock_history_entry('e1', start_time_ms=1000)]
+        result = server.paginate_history_entries(entries, None, None)
+        self.assertIsInstance(result, list)
+        self.assertEqual(len(result), 1)
+
+    def test_isolated_paginate_running_checker_only_called_for_sliced_page(self):
+        entries = [self._create_mock_history_entry(f'e{i}', start_time_ms=i * 1000) for i in range(1, 100)]
+        checked_ids = []
+
+        def mock_checker(entry_id):
+            checked_ids.append(entry_id)
+            return entry_id == 'e99'
+
+        result = server.paginate_history_entries(entries, '1', '10', is_running_checker=mock_checker)
+        self.assertEqual(result['total'], 99)
+        self.assertEqual(result['page'], 1)
+        self.assertEqual(len(result['records']), 10)
+        self.assertEqual(len(checked_ids), 10)
+        self.assertEqual(checked_ids[0], 'e99')
+
     def start_loop(self):
         io_loop = IOLoop.current()
         self.ioloop_thread = threading.Thread(target=io_loop.start)

src/web/server.py

@@ -2,11 +2,13 @@
 import asyncio
 import json
 import logging.config
+import math
 import os
 import signal
 import ssl
 import time
 import urllib
+from datetime import datetime, timezone
 from urllib.parse import urlencode
 
 import tornado.concurrent
@@ -679,18 +681,74 @@ def post(self):
             file_utils.write_file(file_path, value)
 
 
+ALLOWED_PAGE_SIZES = {10, 25, 50, 100, 250, 500}
+DEFAULT_PAGE_SIZE = 25
+
+
+def paginate_history_entries(entries, page_arg, size_arg, is_running_checker=None):
+    def _get_sort_key(entry):
+        if entry.start_time is None:
+            return datetime.min.replace(tzinfo=timezone.utc)
+        return entry.start_time
+
+    entries.sort(key=_get_sort_key, reverse=True)
+
+    if page_arg is None and size_arg is None:
+        running_ids = [e.id for e in entries if is_running_checker and is_running_checker(e.id)]
+        return to_short_execution_log(entries, running_ids)
+
+    try:
+        size = int(size_arg) if size_arg is not None else DEFAULT_PAGE_SIZE
+        if size not in ALLOWED_PAGE_SIZES:
+            size = DEFAULT_PAGE_SIZE
+    except (ValueError, TypeError):
+        size = DEFAULT_PAGE_SIZE
+
+    try:
+        page = int(page_arg) if page_arg is not None else 1
+        if page < 1:
+            page = 1
+    except (ValueError, TypeError):
+        page = 1
+
+    total_count = len(entries)
+    total_pages = math.ceil(total_count / size) if total_count > 0 else 1
+
+    if page > total_pages and total_pages > 0:
+        page = total_pages
+
+    start_idx = (page - 1) * size
+    end_idx = start_idx + size
+    page_entries = entries[start_idx:end_idx]
+
+    running_ids = [e.id for e in page_entries if is_running_checker and is_running_checker(e.id)]
+
+    return {
+        'records': to_short_execution_log(page_entries, running_ids),
+        'total': total_count,
+        'page': page,
+        'pageSize': size,
+        'totalPages': total_pages
+    }
+
+
 class GetShortHistoryEntriesHandler(BaseRequestHandler):
     @check_authorization
     @inject_user
     def get(self, user):
+        page_arg = self.get_argument('page', None)
+        size_arg = self.get_argument('size', None)
+
         history_entries = self.application.execution_logging_service.get_history_entries(user.user_id)
-        running_script_ids = []
-        for entry in history_entries:
-            if self.application.execution_service.is_running(entry.id, user):
-                running_script_ids.append(entry.id)
 
-        short_logs = to_short_execution_log(history_entries, running_script_ids)
-        self.write(json.dumps(short_logs))
+        result = paginate_history_entries(
+            history_entries,
+            page_arg,
+            size_arg,
+            is_running_checker=lambda entry_id: self.application.execution_service.is_running(entry_id, user)
+        )
+
+        self.write(json.dumps(result))
 
 
 class GetLongHistoryEntryHandler(BaseRequestHandler):
@@ -895,7 +953,8 @@ def init(server_config: ServerConfig,
         asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
     io_loop = tornado.ioloop.IOLoop.current()
 
-    global _http_server
+    global _http_server, _tornado_app
+    _tornado_app = application
     _http_server = httpserver.HTTPServer(
         application,
         ssl_options=ssl_context,