Skip to content

Commit f5af4d2

Browse files
gaudybGaudy BlancoCopilot
authored
Fix cannot release un-acquired lock in Blob logger (#2457)
* fix issue-2170 * semversioner change * test changes * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * fix syntax mismatch --------- Co-authored-by: Gaudy Blanco <gaudy-microsoft@MacBook-Pro-m4-Gaudy-For-Work.local> Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
1 parent 27a3675 commit f5af4d2

4 files changed

Lines changed: 189 additions & 20 deletions

File tree

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
{
2+
"type": "patch",
3+
"description": "Fix 2170 - cannot release un-acquired lock in Blob logger"
4+
}

packages/graphrag/graphrag/logger/blob_workflow_logger.py

Lines changed: 20 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@
66
import json
77
import logging
88
from datetime import datetime, timezone
9-
from pathlib import Path
109
from typing import Any
1110

1211
from azure.identity import DefaultAzureCredential
@@ -41,6 +40,7 @@ def __init__(
4140

4241
self._connection_string = connection_string
4342
self.account_url = account_url
43+
self._base_dir = base_dir
4444

4545
if self._connection_string:
4646
self._blob_service_client = BlobServiceClient.from_connection_string(
@@ -56,18 +56,8 @@ def __init__(
5656
credential=DefaultAzureCredential(),
5757
)
5858

59-
if blob_name == "":
60-
blob_name = f"report/{datetime.now(tz=timezone.utc).strftime('%Y-%m-%d-%H:%M:%S:%f')}.logs.json"
61-
62-
self._blob_name = str(Path(base_dir or "") / blob_name)
6359
self._container_name = container_name
64-
self._blob_client = self._blob_service_client.get_blob_client(
65-
self._container_name, self._blob_name
66-
)
67-
if not self._blob_client.exists():
68-
self._blob_client.create_append_blob()
69-
70-
self._num_blocks = 0 # refresh block counter
60+
self._rotate_blob(blob_name or None)
7161

7262
def emit(self, record) -> None:
7363
"""Emit a log record to blob storage."""
@@ -98,17 +88,27 @@ def _get_log_type(self, level: int) -> str:
9888
return "warning"
9989
return "log"
10090

91+
def _rotate_blob(self, blob_name: str | None = None):
92+
"""Create a new blob file when the current one reaches max block count."""
93+
if not blob_name:
94+
blob_name = f"report/{datetime.now(tz=timezone.utc).strftime('%Y-%m-%d-%H:%M:%S:%f')}.logs.json"
95+
blob_path = blob_name.lstrip("/")
96+
if self._base_dir:
97+
blob_path = f"{self._base_dir.strip('/')}/{blob_path}"
98+
self._blob_name = blob_path
99+
self._blob_client = self._blob_service_client.get_blob_client(
100+
self._container_name, self._blob_name
101+
)
102+
if not self._blob_client.exists():
103+
self._blob_client.create_append_blob()
104+
self._num_blocks = 0
105+
101106
def _write_log(self, log: dict[str, Any]):
102107
"""Write log data to blob storage."""
103108
# create a new file when block count hits close 25k
104-
if (
105-
self._num_blocks >= self._max_block_count
106-
): # Check if block count exceeds 25k
107-
self.__init__(
108-
self._connection_string,
109-
self._container_name,
110-
account_url=self.account_url,
111-
)
109+
110+
if self._num_blocks >= self._max_block_count:
111+
self._rotate_blob()
112112

113113
blob_client = self._blob_service_client.get_blob_client(
114114
self._container_name, self._blob_name

tests/unit/logger/__init__.py

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
# Copyright (c) 2024 Microsoft Corporation.
2+
# Licensed under the MIT License
Lines changed: 163 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,163 @@
1+
# Copyright (c) 2024 Microsoft Corporation.
2+
# Licensed under the MIT License
3+
4+
"""Unit tests for the BlobWorkflowLogger."""
5+
6+
import json
7+
from unittest.mock import MagicMock, patch
8+
9+
import pytest
10+
from graphrag.logger.blob_workflow_logger import BlobWorkflowLogger
11+
12+
13+
@pytest.fixture
14+
def mock_blob_service():
15+
"""Create a mock BlobServiceClient and related objects."""
16+
with patch(
17+
"graphrag.logger.blob_workflow_logger.BlobServiceClient"
18+
) as mock_bsc_cls:
19+
mock_blob_client = MagicMock()
20+
mock_blob_client.exists.return_value = False
21+
22+
mock_bsc = MagicMock()
23+
mock_bsc.get_blob_client.return_value = mock_blob_client
24+
25+
mock_bsc_cls.return_value = mock_bsc
26+
27+
yield mock_bsc, mock_blob_client
28+
29+
30+
@pytest.fixture
31+
def mock_credential():
32+
"""Mock DefaultAzureCredential."""
33+
with patch(
34+
"graphrag.logger.blob_workflow_logger.DefaultAzureCredential"
35+
) as mock_cred:
36+
yield mock_cred
37+
38+
39+
def _make_logger() -> BlobWorkflowLogger:
40+
return BlobWorkflowLogger(
41+
connection_string=None,
42+
container_name="test-container",
43+
blob_name="test.logs.json",
44+
base_dir="logs",
45+
account_url="https://test.blob.core.windows.net",
46+
)
47+
48+
49+
def test_init_requires_container_name(mock_credential):
50+
"""Test that a missing container name raises a ValueError."""
51+
with pytest.raises(ValueError, match="No container name provided"):
52+
BlobWorkflowLogger(
53+
connection_string=None,
54+
container_name=None,
55+
account_url="https://test.blob.core.windows.net",
56+
)
57+
58+
59+
def test_init_requires_connection_or_account_url():
60+
"""Test that missing both connection string and account url raises."""
61+
with pytest.raises(ValueError, match="No storage account blob url provided"):
62+
BlobWorkflowLogger(
63+
connection_string=None,
64+
container_name="test-container",
65+
)
66+
67+
68+
def test_init_creates_append_blob(mock_blob_service, mock_credential):
69+
"""Test that a new append blob is created on initialization."""
70+
_mock_bsc, mock_blob_client = mock_blob_service
71+
72+
logger = _make_logger()
73+
74+
mock_blob_client.create_append_blob.assert_called_once()
75+
assert logger._num_blocks == 0 # noqa: SLF001
76+
assert logger._base_dir == "logs" # noqa: SLF001
77+
78+
79+
def test_init_uses_provided_blob_name(mock_blob_service, mock_credential):
80+
"""Test that the provided blob name is honored and nested under base_dir."""
81+
logger = _make_logger()
82+
83+
assert logger._blob_name == "logs/test.logs.json" # noqa: SLF001
84+
85+
86+
def test_rotate_blob_does_not_reinitialize_handler(mock_blob_service, mock_credential):
87+
"""Test that blob rotation does not call __init__ on the handler.
88+
89+
This verifies the fix for issue #2170 where calling self.__init__()
90+
during rotation caused 'cannot release un-acquired lock' errors.
91+
"""
92+
mock_bsc, _mock_blob_client = mock_blob_service
93+
94+
logger = _make_logger()
95+
96+
# Simulate reaching max block count
97+
logger._num_blocks = logger._max_block_count # noqa: SLF001
98+
99+
# Store reference to the original lock to verify it's not replaced
100+
original_lock = logger.lock
101+
102+
# Write a log entry, which should trigger rotation
103+
logger._write_log({"type": "log", "data": "test message"}) # noqa: SLF001
104+
105+
# Verify the lock was NOT replaced (i.e., __init__ was not called)
106+
assert logger.lock is original_lock
107+
108+
# Verify block counter was reset (rotation happened)
109+
# After rotation (reset to 0) + 1 write = 1
110+
assert logger._num_blocks == 1 # noqa: SLF001
111+
112+
# Verify a new blob client was created during rotation
113+
assert mock_bsc.get_blob_client.call_count > 1
114+
115+
116+
def test_rotate_blob_creates_new_blob_name(mock_blob_service, mock_credential):
117+
"""Test that rotation generates a new blob name."""
118+
_mock_bsc, _mock_blob_client = mock_blob_service
119+
120+
logger = _make_logger()
121+
122+
original_blob_name = logger._blob_name # noqa: SLF001
123+
124+
# Trigger rotation
125+
logger._rotate_blob() # noqa: SLF001
126+
127+
# New blob name should be different (auto-generated with timestamp)
128+
assert logger._blob_name != original_blob_name # noqa: SLF001
129+
assert logger._num_blocks == 0 # noqa: SLF001
130+
131+
132+
def test_write_log_appends_block(mock_blob_service, mock_credential):
133+
"""Test that writing a log appends an encoded block and increments count."""
134+
_mock_bsc, mock_blob_client = mock_blob_service
135+
136+
logger = _make_logger()
137+
138+
logger._write_log({"type": "log", "data": "hello"}) # noqa: SLF001
139+
140+
mock_blob_client.append_block.assert_called_once()
141+
(payload,), _kwargs = mock_blob_client.append_block.call_args
142+
decoded = json.loads(payload.decode("utf-8"))
143+
assert decoded == {"type": "log", "data": "hello"}
144+
assert logger._num_blocks == 1 # noqa: SLF001
145+
146+
147+
@pytest.mark.parametrize(
148+
("level", "expected"),
149+
[
150+
(logging_level, expected)
151+
for logging_level, expected in [
152+
(40, "error"), # logging.ERROR
153+
(30, "warning"), # logging.WARNING
154+
(20, "log"), # logging.INFO
155+
(10, "log"), # logging.DEBUG
156+
]
157+
],
158+
)
159+
def test_get_log_type(mock_blob_service, mock_credential, level, expected):
160+
"""Test that log levels map to the correct log type string."""
161+
logger = _make_logger()
162+
163+
assert logger._get_log_type(level) == expected # noqa: SLF001

0 commit comments

Comments
 (0)