Skip to content

Commit 85dd4e6

Browse files
committed
feat(core): add LINSTOR controller and satellite helpers (#12)
Signed-off-by: Ronan Abhamon <ronan.abhamon@vates.tech>
1 parent 3fa2ea1 commit 85dd4e6

7 files changed

Lines changed: 383 additions & 1 deletion

File tree

‎src/xcp_storage/backends/linstor/__init__.py‎

Whitespace-only changes.
Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,71 @@
1+
# Copyright (C) 2026 Vates SAS
2+
#
3+
# This program is free software: you can redistribute it and/or modify
4+
# it under the terms of the GNU General Public License as published by
5+
# the Free Software Foundation, either version 3 of the License, or
6+
# (at your option) any later version.
7+
# This program is distributed in the hope that it will be useful,
8+
# but WITHOUT ANY WARRANTY; without even the implied warranty of
9+
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
10+
# GNU General Public License for more details.
11+
#
12+
# You should have received a copy of the GNU General Public License
13+
# along with this program. If not, see <https://www.gnu.org/licenses/>.
14+
15+
from xcp_storage.backends.linstor.satellite import LINSTOR_SATELLITE_PORT_PLAIN, LINSTOR_SATELLITE_PORT_SSL
16+
from xcp_storage.utils.process import run_command
17+
from xcp_storage.utils.service import (
18+
is_service_active,
19+
restart_service,
20+
start_service,
21+
stop_service,
22+
)
23+
24+
from xcp_storage.typing import Final, List
25+
26+
# ==============================================================================
27+
28+
LINSTOR_CONTROLLER_PORT_PLAIN: Final = 3370
29+
LINSTOR_CONTROLLER_PORT_SSL: Final = 3371
30+
31+
# ------------------------------------------------------------------------------
32+
33+
_EXEC_PATH_SS: Final = "/usr/sbin/ss"
34+
35+
_SERVICE_LINSTOR_CONTROLLER: Final = "linstor-controller"
36+
37+
# ------------------------------------------------------------------------------
38+
39+
class LinstorController:
40+
@staticmethod
41+
def get_addresses() -> List[str]:
42+
stdout = run_command([
43+
_EXEC_PATH_SS, "-tnpH", "state", "established",
44+
f"( sport = :{LINSTOR_SATELLITE_PORT_PLAIN} or sport = :{LINSTOR_SATELLITE_PORT_SSL} )"
45+
], expected_ret_code=0)
46+
return [
47+
line.split()[3].rsplit(":", 1)[0]
48+
for line in stdout.splitlines()
49+
]
50+
51+
@classmethod
52+
def get_uri(cls) -> str:
53+
# TODO(XCPNG-3033): On caller side, check that an IP address from the current pool is returned.
54+
addresses = cls.get_addresses()
55+
return "linstor://" + addresses[0] if addresses else ""
56+
57+
@staticmethod
58+
def is_running() -> bool:
59+
return is_service_active(_SERVICE_LINSTOR_CONTROLLER)
60+
61+
@staticmethod
62+
def start() -> None:
63+
start_service(_SERVICE_LINSTOR_CONTROLLER)
64+
65+
@staticmethod
66+
def stop() -> None:
67+
stop_service(_SERVICE_LINSTOR_CONTROLLER)
68+
69+
@staticmethod
70+
def restart() -> None:
71+
restart_service(_SERVICE_LINSTOR_CONTROLLER)
Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
# Copyright (C) 2026 Vates SAS
2+
#
3+
# This program is free software: you can redistribute it and/or modify
4+
# it under the terms of the GNU General Public License as published by
5+
# the Free Software Foundation, either version 3 of the License, or
6+
# (at your option) any later version.
7+
# This program is distributed in the hope that it will be useful,
8+
# but WITHOUT ANY WARRANTY; without even the implied warranty of
9+
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
10+
# GNU General Public License for more details.
11+
#
12+
# You should have received a copy of the GNU General Public License
13+
# along with this program. If not, see <https://www.gnu.org/licenses/>.
14+
15+
from xcp_storage.utils.service import (
16+
disable_and_stop_service,
17+
enable_and_start_service,
18+
is_service_active,
19+
)
20+
21+
from xcp_storage.typing import Final
22+
23+
# ==============================================================================
24+
25+
LINSTOR_SATELLITE_PORT_PLAIN: Final = 3366
26+
LINSTOR_SATELLITE_PORT_SSL: Final = 3367
27+
28+
# ------------------------------------------------------------------------------
29+
30+
_SERVICE_LINSTOR_SATELLITE: Final = "linstor-satellite"
31+
32+
# ------------------------------------------------------------------------------
33+
34+
class LinstorSatellite:
35+
@staticmethod
36+
def is_running() -> bool:
37+
return is_service_active(_SERVICE_LINSTOR_SATELLITE)
38+
39+
@staticmethod
40+
def enable_and_start() -> None:
41+
enable_and_start_service(_SERVICE_LINSTOR_SATELLITE)
42+
43+
@staticmethod
44+
def disable_and_stop() -> None:
45+
disable_and_stop_service(_SERVICE_LINSTOR_SATELLITE)

‎src/xcp_storage/network/socket.py‎

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -198,13 +198,18 @@ def create_client_sock(
198198
reuse_address: bool = True,
199199
keep_alive: bool = True,
200200
timeout: Optional[float] = None,
201-
ssl_context: Optional[ssl.SSLContext] = None
201+
ssl_context: Optional[ssl.SSLContext] = None,
202+
source_address: Optional[str] = None,
203+
source_port: int = 0
202204
) -> socket.socket:
203205
family, connect = format_address(address, port)
204206
sock = _create_stream_sock(address, family, bind=False, reuse_address=reuse_address, ssl_context=ssl_context)
205207
_normalize_and_set_sock_timeout(sock, timeout)
206208

207209
try:
210+
if source_address:
211+
_, bind = format_address(source_address, source_port)
212+
sock.bind(bind)
208213
sock.connect(connect)
209214
except OSError as e:
210215
with contextlib.suppress(Exception):
@@ -318,6 +323,12 @@ def socket_wait_readable(sock: socket.socket, *, timeout: Optional[float] = None
318323
def get_socket_family_str(sock: socket.socket) -> str:
319324
return _FAMILY_TO_STR.get(sock.family, "Unknown")
320325

326+
def get_socket_address(sock: socket.socket) -> Optional[str]:
327+
try:
328+
return sock.getsockname()[0]
329+
except OSError:
330+
return None
331+
321332
def get_socket_port(sock: socket.socket) -> Optional[int]:
322333
try:
323334
return sock.getsockname()[1]
@@ -371,6 +382,10 @@ def close(self) -> None:
371382
def family_str(self) -> str:
372383
return get_socket_family_str(self.sock)
373384

385+
@property
386+
def address(self) -> Optional[str]:
387+
return get_socket_address(self.sock)
388+
374389
@property
375390
def port(self) -> Optional[int]:
376391
return get_socket_port(self.sock)
Lines changed: 189 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,189 @@
1+
# Copyright (C) 2026 Vates SAS
2+
#
3+
# This program is free software: you can redistribute it and/or modify
4+
# it under the terms of the GNU General Public License as published by
5+
# the Free Software Foundation, either version 3 of the License, or
6+
# (at your option) any later version.
7+
# This program is distributed in the hope that it will be useful,
8+
# but WITHOUT ANY WARRANTY; without even the implied warranty of
9+
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
10+
# GNU General Public License for more details.
11+
#
12+
# You should have received a copy of the GNU General Public License
13+
# along with this program. If not, see <https://www.gnu.org/licenses/>.
14+
15+
import contextlib
16+
from unittest.mock import (
17+
MagicMock,
18+
patch,
19+
)
20+
21+
import pytest
22+
23+
from xcp_storage.backends.linstor.controller import _SERVICE_LINSTOR_CONTROLLER, LinstorController
24+
from xcp_storage.network.socket import (
25+
create_client_sock,
26+
create_server_sock,
27+
Socket,
28+
SocketError,
29+
)
30+
31+
from xcp_storage.typing import (
32+
Callable,
33+
cast,
34+
Final,
35+
Iterator,
36+
NamedTuple,
37+
Optional,
38+
Tuple,
39+
)
40+
41+
# ==============================================================================
42+
43+
MODULE_NAME: Final = "xcp_storage.backends.linstor.controller"
44+
45+
# ------------------------------------------------------------------------------
46+
47+
class TestLinstorControllerGetAddresses:
48+
class LoopbackSpec(NamedTuple):
49+
listener_address: str
50+
client_address: str
51+
uri_client_address: str
52+
53+
LOOPBACK_ADDRESSES: Final = ["127.0.0.2", "127.0.0.3", "127.0.0.4"]
54+
55+
IPV4: Final = LoopbackSpec("127.0.0.1", LOOPBACK_ADDRESSES[0], LOOPBACK_ADDRESSES[0])
56+
IPV6: Final = LoopbackSpec("::1", "::1", "[::1]")
57+
58+
ListenFunc = Callable[..., Socket]
59+
ConnectFunc = Callable[..., Tuple[Socket, Socket]]
60+
PatchListenerPortFunc = Callable[[Socket], "contextlib.AbstractContextManager[None]"]
61+
62+
@staticmethod
63+
@contextlib.contextmanager
64+
def patch_satellite_ports(plain_port: Optional[int], ssl_port: Optional[int]) -> Iterator[None]:
65+
with patch.multiple(MODULE_NAME, LINSTOR_SATELLITE_PORT_PLAIN=plain_port, LINSTOR_SATELLITE_PORT_SSL=ssl_port):
66+
yield
67+
68+
@pytest.fixture(params=[IPV4, IPV6], ids=["ipv4", "ipv6"])
69+
def loopback_spec(self, request: pytest.FixtureRequest) -> LoopbackSpec:
70+
spec = cast(TestLinstorControllerGetAddresses.LoopbackSpec, request.param)
71+
if spec is self.IPV6:
72+
try:
73+
create_server_sock(spec.listener_address, 0).close()
74+
except SocketError:
75+
pytest.skip("IPv6 is not available.")
76+
return spec
77+
78+
@pytest.fixture
79+
def sockets(self) -> Iterator[contextlib.ExitStack]:
80+
with contextlib.ExitStack() as stack:
81+
yield stack
82+
83+
@pytest.fixture
84+
def listen(self, sockets: contextlib.ExitStack) -> ListenFunc:
85+
def impl(address: str = "127.0.0.1") -> Socket:
86+
return sockets.enter_context(Socket(create_server_sock(address, 0)))
87+
return impl
88+
89+
@pytest.fixture
90+
def connect(self, sockets: contextlib.ExitStack) -> ConnectFunc:
91+
def impl(listener: Socket, client_address: str) -> Tuple[Socket, Socket]:
92+
client = sockets.enter_context(Socket(create_client_sock(
93+
cast(str, listener.address),
94+
cast(int, listener.port),
95+
source_address=client_address,
96+
keep_alive=False
97+
)))
98+
accepted, _ = listener.sock.accept()
99+
return client, sockets.enter_context(Socket(accepted))
100+
return impl
101+
102+
@pytest.fixture(params=["plain", "ssl"])
103+
def patch_listener_port(self, request: pytest.FixtureRequest, listen: ListenFunc) -> PatchListenerPortFunc:
104+
unused_port = listen().port
105+
106+
def impl(listener: Socket) -> "contextlib.AbstractContextManager[None]":
107+
ports = (listener.port, unused_port)
108+
return self.patch_satellite_ports(*(ports if request.param == "plain" else reversed(ports)))
109+
return impl
110+
111+
def test_no_connection(self, listen: ListenFunc, loopback_spec: LoopbackSpec) -> None:
112+
listener = listen(loopback_spec.listener_address)
113+
114+
with self.patch_satellite_ports(listener.port, listener.port):
115+
assert LinstorController.get_addresses() == []
116+
assert LinstorController.get_uri() == ""
117+
118+
def test_closed_connection(
119+
self,
120+
listen: ListenFunc,
121+
connect: ConnectFunc,
122+
patch_listener_port: PatchListenerPortFunc,
123+
loopback_spec: LoopbackSpec
124+
) -> None:
125+
listener = listen(loopback_spec.listener_address)
126+
client, accepted = connect(listener, loopback_spec.client_address)
127+
128+
accepted.close()
129+
client.close()
130+
131+
with patch_listener_port(listener):
132+
assert LinstorController.get_addresses() == []
133+
134+
def test_established_connection(
135+
self,
136+
listen: ListenFunc,
137+
connect: ConnectFunc,
138+
patch_listener_port: PatchListenerPortFunc,
139+
loopback_spec: LoopbackSpec
140+
) -> None:
141+
listener = listen(loopback_spec.listener_address)
142+
connect(listener, loopback_spec.client_address)
143+
144+
with patch_listener_port(listener):
145+
assert LinstorController.get_addresses() == [loopback_spec.uri_client_address]
146+
assert LinstorController.get_uri() == f"linstor://{loopback_spec.uri_client_address}"
147+
148+
# ------------------------------------------------------------------------------
149+
150+
@patch(f"{MODULE_NAME}.LinstorController.get_addresses")
151+
class TestLinstorControllerGetUri:
152+
def test_no_address(self, mock_get_addresses: MagicMock) -> None:
153+
mock_get_addresses.return_value = []
154+
assert LinstorController.get_uri() == ""
155+
156+
def test_one_address(self, mock_get_addresses: MagicMock) -> None:
157+
mock_get_addresses.return_value = ["10.10.0.13"]
158+
assert LinstorController.get_uri() == "linstor://10.10.0.13"
159+
160+
def test_two_addresses(self, mock_get_addresses: MagicMock) -> None:
161+
mock_get_addresses.return_value = ["10.10.0.13", "10.10.0.14"]
162+
assert LinstorController.get_uri() == "linstor://10.10.0.13"
163+
164+
def test_ipv6_address(self, mock_get_addresses: MagicMock) -> None:
165+
mock_get_addresses.return_value = ["[fe80::2]"]
166+
assert LinstorController.get_uri() == "linstor://[fe80::2]"
167+
168+
# ------------------------------------------------------------------------------
169+
170+
class TestLinstorControllerService:
171+
@patch(f"{MODULE_NAME}.is_service_active", return_value=True)
172+
def test_running(self, mock_is_service_active: MagicMock) -> None:
173+
assert LinstorController.is_running()
174+
mock_is_service_active.assert_called_once_with(_SERVICE_LINSTOR_CONTROLLER)
175+
176+
@patch(f"{MODULE_NAME}.is_service_active", return_value=False)
177+
def test_not_running(self, mock_is_service_active: MagicMock) -> None:
178+
assert not LinstorController.is_running()
179+
mock_is_service_active.assert_called_once_with(_SERVICE_LINSTOR_CONTROLLER)
180+
181+
@pytest.mark.parametrize("service_name, function_name", [
182+
("start", "start_service"),
183+
("stop", "stop_service"),
184+
("restart", "restart_service")
185+
])
186+
def test_service_command(self, service_name: str, function_name: str) -> None:
187+
with patch(f"{MODULE_NAME}.{function_name}") as mock_function_name:
188+
getattr(LinstorController, service_name)()
189+
mock_function_name.assert_called_once_with(_SERVICE_LINSTOR_CONTROLLER)

0 commit comments

Comments
 (0)