This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 33a10541b3 [python] Log FileIO cache and Jindo stream health at a
bounded rate (#10580)
33a10541b3 is described below
commit 33a10541b38a6d88321630bf6cb809b03d1fdeda
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Sun Oct 18 16:50:10 2026 +0800
[python] Log FileIO cache and Jindo stream health at a bounded rate (#10580)
---
.../pypaimon/catalog/rest/rest_token_file_io.py | 65 +++++++++++++
.../filesystem/jindo_file_system_handler.py | 40 ++++++++
.../pypaimon/tests/jindo_file_system_test.py | 64 +++++++++++++
.../pypaimon/tests/rest/rest_token_file_io_test.py | 105 +++++++++++++++++++++
4 files changed, 274 insertions(+)
diff --git a/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py
b/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py
index fbb39480ed..94e044652e 100644
--- a/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py
+++ b/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py
@@ -59,6 +59,12 @@ class RESTTokenFileIO(FileIO):
# Objects whose lock a forked child resets.
_INSTANCES = weakref.WeakSet()
+ # Cache misses log a summary at most this often; idle processes log
nothing.
+ _STATS_LOG_INTERVAL = 600
+ _FILE_IO_STATS = {"created": 0, "evicted": 0, "warned": False, "next_log":
0.0,
+ "fds": None, "next_scan": 0.0}
+ _SHORT_TOKEN_STATS = {"count": 0, "next_log": 0.0}
+
def __init__(self, identifier: Identifier, path: str,
catalog_options: Optional[Union[dict, Options]] = None):
self.identifier = identifier
@@ -146,10 +152,44 @@ class RESTTokenFileIO(FileIO):
file_io = cache.get(cache_key)
if file_io is None:
file_io = FileIO.get(self.path,
self._merged_options(token))
+ self._log_file_io_created(cache)
# Re-inserting on every access expires idle entries only, like
expireAfterAccess.
cache[cache_key] = file_io
return file_io, token
+ def _log_file_io_created(self, cache):
+ # Runs under the cache lock right before the new FileIO is inserted.
+ stats = RESTTokenFileIO._FILE_IO_STATS
+ stats["created"] += 1
+ cache.expire()
+ if len(cache) >= cache.maxsize:
+ stats["evicted"] += 1
+ now = time.monotonic()
+ quiet = now < stats["next_log"]
+ if quiet and stats["warned"]:
+ return
+ fds = self._sample_open_files(stats, now)
+ crowded = stats["evicted"] > 0 or (fds is not None and fds[0] * 5 >
fds[1] * 4)
+ # The first sign of trouble is logged at once; anything else waits for
the interval.
+ if quiet and not crowded:
+ return
+ self.log.log(
+ logging.WARNING if crowded else logging.INFO,
+ "RESTTokenFileIO cache holds %d/%d FileIOs and %d tokens; %d
FileIOs created and "
+ "%d evicted for size since the last summary; open files %s",
+ min(len(cache) + 1, cache.maxsize), cache.maxsize,
len(self._TOKEN_CACHE),
+ stats["created"], stats["evicted"], "unknown" if fds is None else
"%d/%d" % fds)
+ stats.update(created=0, evicted=0, warned=crowded,
+ next_log=now + self._STATS_LOG_INTERVAL)
+
+ @staticmethod
+ def _sample_open_files(stats, now):
+ # Listing /proc/self/fd takes about 0.5 ms per 1,000 descriptors, so
scan at most 1% of the time.
+ if now >= stats["next_scan"]:
+ stats["fds"] = _open_files()
+ stats["next_scan"] = now + 100 * (time.monotonic() - now)
+ return stats["fds"]
+
def _get_file_io_scope(self) -> str:
# FileIO.get binds its result to the path (e.g. the OSS bucket) and
the catalog options.
if self._file_io_scope is None:
@@ -385,6 +425,23 @@ class RESTTokenFileIO(FileIO):
merged_token_dict =
self._merge_token_with_catalog_options(response.token)
new_token = RESTToken(merged_token_dict, response.expires_at_millis)
self.token = new_token
+ if self._is_token_expired(new_token):
+ self._log_short_token(new_token)
+
+ def _log_short_token(self, token: RESTToken):
+ with self._TOKEN_LOCKS_LOCK:
+ stats = RESTTokenFileIO._SHORT_TOKEN_STATS
+ stats["count"] += 1
+ now = time.monotonic()
+ if now < stats["next_log"]:
+ return
+ count = stats["count"]
+ stats.update(count=0, next_log=now + self._STATS_LOG_INTERVAL)
+ self.log.warning(
+ "Data token for [%s] expires in %d s, within the %d s refresh
margin, so every "
+ "access refreshes it; %d such tokens in the last %d s",
+ self.identifier, (token.expire_at_millis - int(time.time() *
1000)) // 1000,
+ RESTApi.TOKEN_EXPIRATION_SAFE_TIME_MILLIS // 1000, count,
self._STATS_LOG_INTERVAL)
def valid_token(self):
self.try_to_refresh_token()
@@ -399,6 +456,14 @@ class RESTTokenFileIO(FileIO):
pass
+def _open_files():
+ try:
+ import resource
+ return len(os.listdir("/proc/self/fd")),
resource.getrlimit(resource.RLIMIT_NOFILE)[0]
+ except (ImportError, OSError):
+ return None
+
+
if hasattr(os, "register_at_fork"):
# Like logging: a fork waits for the cache lock, and the child starts with
free locks.
os.register_at_fork(before=RESTTokenFileIO._before_fork,
diff --git a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py
b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py
index 2219497a80..93d255f4a0 100644
--- a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py
+++ b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py
@@ -57,6 +57,42 @@ _CASE_SENSITIVE_JINDO_CONFIG_KEYS = {
}
_KNOWN_FILE_SIZE_CACHE_MAX_ENTRIES = 4096
+_LOG = logging.getLogger(__name__)
+
+
+class _StreamStats:
+ """Counts open streams of JindoFileSystemHandler, not those PVFS opens
through pyjindo.ossfs."""
+
+ # Reentrant: garbage collection may finalize a stream while this thread
holds it.
+ _lock = threading.RLock()
+ _pid = os.getpid()
+ open = 0
+ milestone = 256
+
+ @classmethod
+ def _get_lock(cls):
+ # A parent thread may hold the lock across a fork, so a child takes a
new one.
+ if cls._pid != os.getpid():
+ cls._lock, cls._pid = threading.RLock(), os.getpid()
+ return cls._lock
+
+ @classmethod
+ def opened(cls):
+ with cls._get_lock():
+ cls.open += 1
+ count = cls.open
+ report = count >= cls.milestone
+ if report:
+ cls.milestone *= 2
+ if report:
+ _LOG.info("%d streams of JindoFileSystemHandler are open; each
holds native buffers until closed",
+ count)
+
+ @classmethod
+ def closed(cls):
+ with cls._get_lock():
+ cls.open -= 1
+
def _jindo_config_value(value) -> str:
if isinstance(value, bool):
@@ -138,6 +174,7 @@ class JindoInputFile:
# handler) so evicting the handler's FileIO from a cache cannot free
it.
self._owner = owner
self._closed = False
+ _StreamStats.opened()
def __del__(self):
# CPython releases instance attributes in dict layout order, which is
@@ -191,6 +228,7 @@ class JindoInputFile:
# The native stream is closed: a file object that stays
# referenced no longer needs to pin the native store.
self._owner = None
+ _StreamStats.closed()
def __enter__(self):
return self
@@ -207,6 +245,7 @@ class JindoOutputFile:
# stream is open.
self._owner = owner
self._closed = False
+ _StreamStats.opened()
def __del__(self):
# See JindoInputFile.__del__: close while ``_owner`` still pins the
@@ -244,6 +283,7 @@ class JindoOutputFile:
# The native stream is closed: a file object that stays
# referenced no longer needs to pin the native store.
self._owner = None
+ _StreamStats.closed()
def __enter__(self):
return self
diff --git a/paimon-python/pypaimon/tests/jindo_file_system_test.py
b/paimon-python/pypaimon/tests/jindo_file_system_test.py
index 3b872003a9..96743573d0 100644
--- a/paimon-python/pypaimon/tests/jindo_file_system_test.py
+++ b/paimon-python/pypaimon/tests/jindo_file_system_test.py
@@ -19,11 +19,13 @@ import gc
import io
import os
import pickle
+import signal
import threading
import time
import types
import unittest
import uuid
+import warnings
import weakref
from unittest import mock
@@ -95,6 +97,68 @@ class JindoInputFileTest(unittest.TestCase):
self.assertEqual([60], stream.reads)
+class JindoStreamStatsTest(unittest.TestCase):
+
+ def setUp(self):
+ stats = jindo_module._StreamStats
+ for name in ("open", "milestone"):
+ self.addCleanup(setattr, stats, name, getattr(stats, name))
+ stats.open, stats.milestone = 0, 2
+
+ def test_logs_only_when_open_streams_pass_a_power_of_two(self):
+ with self.assertLogs(jindo_module.__name__, level="INFO") as logs:
+ files = [JindoInputFile(_RecordingInputStream(), owner=object())
for _ in range(3)]
+ # The output stream is the one that passes 4.
+ files.append(JindoOutputFile(_RecordingInputStream(),
owner=object()))
+ self.assertEqual(["2 streams of JindoFileSystemHandler are open",
+ "4 streams of JindoFileSystemHandler are open"],
+ [r.getMessage().split(";")[0] for r in logs.records])
+
+ files.pop().close()
+ self.assertEqual(3, jindo_module._StreamStats.open)
+ files.pop(0).close()
+ # Dropping an unclosed stream closes it, so it no longer counts as
open.
+ files.pop(0)
+ gc.collect()
+ self.assertEqual(1, jindo_module._StreamStats.open)
+ files[0].close()
+ files[0].close()
+ self.assertEqual(0, jindo_module._StreamStats.open)
+
+ @unittest.skipUnless(hasattr(os, "fork"), "needs os.fork")
+ def
test_forked_child_counts_streams_while_a_parent_thread_holds_the_lock(self):
+ held, release = threading.Event(), threading.Event()
+
+ def hold():
+ with jindo_module._StreamStats._lock:
+ held.set()
+ release.wait()
+
+ thread = threading.Thread(target=hold)
+ thread.start()
+ held.wait()
+ try:
+ with warnings.catch_warnings():
+ # Python 3.12+ warns about forking with other threads alive,
which this test does on purpose.
+ warnings.simplefilter("ignore", DeprecationWarning)
+ pid = os.fork()
+ if pid == 0:
+ code = 1
+ try:
+ signal.alarm(10)
+ JindoInputFile(_RecordingInputStream(),
owner=object()).close()
+ code = 0 if jindo_module._StreamStats.open == 0 else 2
+ finally:
+ os._exit(code)
+ status = os.waitpid(pid, 0)[1]
+ finally:
+ release.set()
+ thread.join()
+ self.assertFalse(os.WIFSIGNALED(status) and os.WTERMSIG(status) ==
signal.SIGALRM,
+ "the forked child blocked on a lock held by a parent
thread")
+ self.assertTrue(os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0,
status)
+
+
class JindoConfigTest(unittest.TestCase):
def test_input_file_seek_returns_position(self):
diff --git a/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py
b/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py
index a7b88220d0..ad76d12d03 100644
--- a/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py
+++ b/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py
@@ -22,6 +22,7 @@ import signal
import tempfile
import threading
import time
+import types
import unittest
import warnings
import weakref
@@ -244,6 +245,110 @@ class RESTTokenFileIOTest(unittest.TestCase):
"the forked child blocked on a lock held by a parent
thread")
self.assertTrue(os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0,
status)
+ def _touch_with_summary_stats(self, cache, stats, clock, open_files,
steps):
+ # Each step sets the clock and open files, then reads through a FileIO
with that token name.
+ module = 'pypaimon.catalog.rest.rest_token_file_io'
+ with patch.object(RESTTokenFileIO, '_FILE_IO_CACHE', cache), \
+ patch.object(RESTTokenFileIO, '_FILE_IO_STATS', stats), \
+ patch.object(RESTTokenFileIO, 'try_to_refresh_token'), \
+ patch(module + '.time',
types.SimpleNamespace(monotonic=lambda: clock[0], time=time.time)), \
+ patch(_FILE_IO_GET, side_effect=_FakeFileIO), \
+ self.assertLogs(module, level='INFO') as logs:
+ for now, files, name in steps:
+ clock[0] = now
+ if files is not None:
+ open_files[0] = files
+ if name is None:
+ cache.clear()
+ continue
+ self._file_io_with_token(
+ self.warehouse_path, self.catalog_options,
self._valid_token(name)).exists("x")
+ return logs.records
+
+ def test_cache_misses_log_one_summary_per_interval(self):
+ cache = TTLCache(maxsize=3, ttl=1e9)
+ stats = dict(RESTTokenFileIO._FILE_IO_STATS, created=0, evicted=0,
warned=False, next_log=0.0,
+ fds=None, next_scan=0.0)
+ open_files = [(10, 1000)]
+ scan = MagicMock(side_effect=lambda: open_files[0])
+ with patch('pypaimon.catalog.rest.rest_token_file_io._open_files',
scan):
+ records = self._touch_with_summary_stats(cache, stats, [0.0],
open_files, [
+ (1000, None, "a"),
+ (1100, None, "b"),
+ # Crowded descriptors warn at once, even within the interval
of an INFO summary.
+ (1200, (900, 1000), "c"),
+ # After a warning, misses neither log nor list descriptors
until the interval ends.
+ (1300, None, "d"),
+ (1801, (10, 1000), "e"),
+ (2402, None, None),
+ (2402, None, "f"),
+ ])
+
+ self.assertEqual(['INFO', 'WARNING', 'WARNING', 'INFO'], [r.levelname
for r in records])
+ messages = [r.getMessage() for r in records]
+ self.assertIn("holds 1/3 FileIOs", messages[0])
+ self.assertIn("1 FileIOs created and 0 evicted", messages[0])
+ self.assertIn("open files 10/1000", messages[0])
+ self.assertIn("holds 3/3 FileIOs", messages[1])
+ self.assertIn("2 FileIOs created and 0 evicted", messages[1])
+ self.assertIn("open files 900/1000", messages[1])
+ self.assertIn("2 FileIOs created and 2 evicted", messages[2])
+ self.assertIn("open files 10/1000", messages[2])
+ self.assertIn("1 FileIOs created and 0 evicted", messages[3])
+ self.assertEqual(5, scan.call_count)
+
+ def test_open_files_are_listed_at_most_one_percent_of_the_time(self):
+ clock = [100.0]
+ stats = {"fds": None, "next_scan": 0.0}
+
+ def scan():
+ clock[0] += 0.002
+ return 10, 1000
+
+ module = 'pypaimon.catalog.rest.rest_token_file_io'
+ with patch(module + '._open_files', side_effect=scan) as open_files, \
+ patch(module + '.time',
types.SimpleNamespace(monotonic=lambda: clock[0], time=time.time)):
+ for now in (100.0, 100.1, 100.19, 100.21):
+ clock[0] = now
+ self.assertEqual((10, 1000),
RESTTokenFileIO._sample_open_files(stats, now))
+ # A 2 ms scan at 100.0 defers the next one to 100.2.
+ self.assertEqual(2, open_files.call_count)
+
+ def
test_cache_miss_reports_unknown_open_files_when_they_cannot_be_listed(self):
+ real_listdir = os.listdir
+
+ def listdir(path="."):
+ if path == "/proc/self/fd":
+ raise OSError("not listable")
+ return real_listdir(path)
+
+ stats = dict(RESTTokenFileIO._FILE_IO_STATS, created=0, evicted=0,
warned=False, next_log=0.0,
+ fds=None, next_scan=0.0)
+ with patch('os.listdir', side_effect=listdir):
+ records = self._touch_with_summary_stats(
+ TTLCache(maxsize=3, ttl=1e9), stats, [0.0], [None], [(1000,
None, "a")])
+ self.assertEqual(1, len(records))
+ self.assertEqual('INFO', records[0].levelname)
+ self.assertIn("open files unknown", records[0].getMessage())
+
+ def test_short_lived_token_warns_once_per_interval(self):
+ file_io = RESTTokenFileIO(self.identifier, self.warehouse_path,
self.catalog_options)
+ file_io.api_instance = MagicMock()
+ now = int(time.time() * 1000)
+ lifetimes = [7_200_000, 1_800_000, 1_800_000, 1_800_000]
+ responses = [MagicMock(token={'ak': 'test-ak'}, expires_at_millis=now
+ lifetime)
+ for lifetime in lifetimes]
+ file_io.api_instance.load_table_token.side_effect = responses
+
+ with patch.object(RESTTokenFileIO, '_SHORT_TOKEN_STATS', {"count": 0,
"next_log": 0.0}), \
+ self.assertLogs('pypaimon.catalog.rest.rest_token_file_io',
level='WARNING') as logs:
+ for _ in lifetimes:
+ file_io.refresh_token()
+
+ self.assertEqual(1, len(logs.records))
+ self.assertIn("within the 3600 s refresh margin",
logs.records[0].getMessage())
+ self.assertIn("1 such tokens", logs.records[0].getMessage())
+
@unittest.skipUnless(hasattr(os, "register_at_fork"), "needs
os.register_at_fork")
def test_fork_while_another_thread_creates_file_io(self):
token = self._valid_token("shared")
