Coverage for drivers/CephFSSR.py : 21%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#!/usr/bin/env python3
2#
3# Original work copyright (C) Citrix systems
4# Modified work copyright (C) Vates SAS and XCP-ng community
5#
6# This program is free software; you can redistribute it and/or modify
7# it under the terms of the GNU Lesser General Public License as published
8# by the Free Software Foundation; version 2.1 only.
9#
10# This program is distributed in the hope that it will be useful,
11# but WITHOUT ANY WARRANTY; without even the implied warranty of
12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13# GNU Lesser General Public License for more details.
14#
15# You should have received a copy of the GNU Lesser General Public License
16# along with this program; if not, write to the Free Software Foundation, Inc.,
17# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
18#
19# CEPHFSSR: Based on FileSR, mounts ceph fs share
21from sm_typing import override
23import errno
24import os
25import socket
26import syslog as _syslog
27import xmlrpc.client
28from syslog import syslog
30# careful with the import order here
31# FileSR has a circular dependency:
32# FileSR -> blktap2 -> lvutil -> EXTSR -> FileSR
33# importing in this order seems to avoid triggering the issue.
34import SR
35import SRCommand
36import FileSR
37# end of careful
38import VDI
39import cleanup
40import util
41import vhdutil
42import xs_errors
43from lock import Lock
45CAPABILITIES = ["SR_PROBE", "SR_UPDATE",
46 "VDI_CREATE", "VDI_DELETE", "VDI_ATTACH", "VDI_DETACH",
47 "VDI_UPDATE", "VDI_CLONE", "VDI_SNAPSHOT", "VDI_RESIZE", "VDI_MIRROR",
48 "VDI_GENERATE_CONFIG",
49 "VDI_RESET_ON_BOOT/2", "ATOMIC_PAUSE"]
51CONFIGURATION = [
52 ['server', 'Ceph server(s) (required, ex: "192.168.0.12" or "10.10.10.10,10.10.10.26")'],
53 ['serverpath', 'Ceph FS path (required, ex: "/")'],
54 ['serverport', 'ex: 6789'],
55 ['options', 'Ceph FS client name, and secretfile (required, ex: "name=admin,secretfile=/etc/ceph/admin.secret")']
56]
58DRIVER_INFO = {
59 'name': 'CephFS VHD',
60 'description': 'SR plugin which stores disks as VHD files on a CephFS storage',
61 'vendor': 'Vates SAS',
62 'copyright': '(C) 2020 Vates SAS',
63 'driver_version': '1.0',
64 'required_api_version': '1.0',
65 'capabilities': CAPABILITIES,
66 'configuration': CONFIGURATION
67}
69DRIVER_CONFIG = {"ATTACH_FROM_CONFIG_WITH_TAPDISK": True}
71# The mountpoint for the directory when performing an sr_probe. All probes
72# are guaranteed to be serialised by xapi, so this single mountpoint is fine.
73PROBE_MOUNTPOINT = os.path.join(SR.MOUNT_BASE, "probe")
76class CephFSException(Exception):
77 def __init__(self, errstr):
78 self.errstr = errstr
81# mountpoint = /var/run/sr-mount/CephFS/uuid
82# linkpath = mountpoint/uuid - path to SR directory on share
83# path = /var/run/sr-mount/uuid - symlink to SR directory on share
84class CephFSSR(FileSR.FileSR):
85 """Ceph file-based storage repository"""
87 DRIVER_TYPE = 'cephfs'
89 @override
90 @staticmethod
91 def handles(sr_type) -> bool:
92 # fudge, because the parent class (FileSR) checks for smb to alter its behavior
93 return sr_type == CephFSSR.DRIVER_TYPE or sr_type == 'smb'
95 @override
96 def load(self, sr_uuid) -> None:
97 if not self._is_ceph_available():
98 raise xs_errors.XenError(
99 'SRUnavailable',
100 opterr='ceph is not installed'
101 )
103 self.ops_exclusive = FileSR.OPS_EXCLUSIVE
104 self.lock = Lock(vhdutil.LOCK_TYPE_SR, self.uuid)
105 self.sr_vditype = SR.DEFAULT_TAP
106 self.driver_config = DRIVER_CONFIG
107 if 'server' not in self.dconf:
108 raise xs_errors.XenError('ConfigServerMissing')
109 self.remoteserver = self.dconf['server']
110 self.remotepath = self.dconf['serverpath']
111 # if serverport is not specified, use default 6789
112 if 'serverport' not in self.dconf:
113 self.remoteport = "6789"
114 else:
115 self.remoteport = self.dconf['serverport']
116 if self.sr_ref and self.session is not None:
117 self.sm_config = self.session.xenapi.SR.get_sm_config(self.sr_ref)
118 else:
119 self.sm_config = self.srcmd.params.get('sr_sm_config') or {}
120 self.mountpoint = os.path.join(SR.MOUNT_BASE, 'CephFS', sr_uuid)
121 self.linkpath = os.path.join(self.mountpoint, sr_uuid or "")
122 self.path = os.path.join(SR.MOUNT_BASE, sr_uuid)
123 self._check_o_direct()
125 def checkmount(self):
126 return util.ioretry(lambda: ((util.pathexists(self.mountpoint) and
127 util.ismount(self.mountpoint)) and
128 util.pathexists(self.path)))
130 def mount(self, mountpoint=None):
131 """Mount the remote ceph export at 'mountpoint'"""
132 if mountpoint is None:
133 mountpoint = self.mountpoint
134 elif not util.is_string(mountpoint) or mountpoint == "":
135 raise CephFSException("mountpoint not a string object")
137 try:
138 if not util.ioretry(lambda: util.isdir(mountpoint)):
139 util.ioretry(lambda: util.makedirs(mountpoint))
140 except util.CommandException as inst:
141 raise CephFSException("Failed to make directory: code is %d" % inst.code)
143 try:
144 options = []
145 if 'options' in self.dconf:
146 options.append(self.dconf['options'])
147 if options:
148 options = ['-o', ','.join(options)]
149 acc = []
150 for server in self.remoteserver.split(','):
151 try:
152 addr_info = socket.getaddrinfo(server, 0)[0]
153 except Exception:
154 continue
156 acc.append('[' + server + ']' if addr_info[0] == socket.AF_INET6 else server)
158 remoteserver = ','.join(acc)
159 command = ["mount", '-t', 'ceph', remoteserver + ":" + self.remoteport + ":" + self.remotepath, mountpoint] + options
160 util.ioretry(lambda: util.pread(command), errlist=[errno.EPIPE, errno.EIO], maxretry=2, nofail=True)
161 except util.CommandException as inst:
162 syslog(_syslog.LOG_ERR, 'CephFS mount failed ' + inst.__str__())
163 raise CephFSException("mount failed with return code %d" % inst.code)
165 # Sanity check to ensure that the user has at least RO access to the
166 # mounted share. Windows sharing and security settings can be tricky.
167 try:
168 util.listdir(mountpoint)
169 except util.CommandException:
170 try:
171 self.unmount(mountpoint, True)
172 except CephFSException:
173 util.logException('CephFSSR.unmount()')
174 raise CephFSException("Permission denied. Please check user privileges.")
176 def unmount(self, mountpoint, rmmountpoint):
177 try:
178 util.pread(["umount", mountpoint])
179 except util.CommandException as inst:
180 raise CephFSException("umount failed with return code %d" % inst.code)
181 if rmmountpoint:
182 try:
183 os.rmdir(mountpoint)
184 except OSError as inst:
185 raise CephFSException("rmdir failed with error '%s'" % inst.strerror)
187 @override
188 def attach(self, sr_uuid) -> None:
189 if not self.checkmount():
190 try:
191 self.mount()
192 os.symlink(self.linkpath, self.path)
193 except CephFSException as exc:
194 raise xs_errors.SROSError(12, exc.errstr)
195 self.attached = True
197 @override
198 def probe(self) -> str:
199 try:
200 self.mount(PROBE_MOUNTPOINT)
201 sr_list = filter(util.match_uuid, util.listdir(PROBE_MOUNTPOINT))
202 self.unmount(PROBE_MOUNTPOINT, True)
203 except (util.CommandException, xs_errors.XenError):
204 raise
205 # Create a dictionary from the SR uuids to feed SRtoXML()
206 return util.SRtoXML({sr_uuid: {} for sr_uuid in sr_list})
208 @override
209 def detach(self, sr_uuid) -> None:
210 if not self.checkmount():
211 return
212 util.SMlog("Aborting GC/coalesce")
213 cleanup.abort(self.uuid)
214 # Change directory to avoid unmount conflicts
215 os.chdir(SR.MOUNT_BASE)
216 self.unmount(self.mountpoint, True)
217 os.unlink(self.path)
218 self.attached = False
220 @override
221 def create(self, sr_uuid, size) -> None:
222 if self.checkmount():
223 raise xs_errors.SROSError(113, 'CephFS mount point already attached')
225 try:
226 self.mount()
227 except CephFSException as exc:
228 # noinspection PyBroadException
229 try:
230 os.rmdir(self.mountpoint)
231 except:
232 # we have no recovery strategy
233 pass
234 raise xs_errors.SROSError(111, "CephFS mount error [opterr=%s]" % exc.errstr)
236 if util.ioretry(lambda: util.pathexists(self.linkpath)):
237 if len(util.ioretry(lambda: util.listdir(self.linkpath))) != 0:
238 self.detach(sr_uuid)
239 raise xs_errors.XenError('SRExists')
240 else:
241 try:
242 util.ioretry(lambda: util.makedirs(self.linkpath))
243 os.symlink(self.linkpath, self.path)
244 except util.CommandException as inst:
245 if inst.code != errno.EEXIST:
246 try:
247 self.unmount(self.mountpoint, True)
248 except CephFSException:
249 util.logException('CephFSSR.unmount()')
250 raise xs_errors.SROSError(116,
251 "Failed to create CephFS SR. remote directory creation error: {}".format(
252 os.strerror(inst.code)))
253 self.detach(sr_uuid)
255 @override
256 def delete(self, sr_uuid) -> None:
257 # try to remove/delete non VDI contents first
258 super(CephFSSR, self).delete(sr_uuid)
259 try:
260 if self.checkmount():
261 self.detach(sr_uuid)
262 self.mount()
263 if util.ioretry(lambda: util.pathexists(self.linkpath)):
264 util.ioretry(lambda: os.rmdir(self.linkpath))
265 util.SMlog(str(self.unmount(self.mountpoint, True)))
266 except util.CommandException as inst:
267 self.detach(sr_uuid)
268 if inst.code != errno.ENOENT:
269 raise xs_errors.SROSError(114, "Failed to remove CephFS mount point")
271 @override
272 def vdi(self, uuid, loadLocked=False) -> VDI.VDI:
273 return CephFSFileVDI(self, uuid)
275 @staticmethod
276 def _is_ceph_available():
277 return util.find_executable('ceph')
279class CephFSFileVDI(FileSR.FileVDI):
280 @override
281 def attach(self, sr_uuid, vdi_uuid) -> str:
282 if not hasattr(self, 'xenstore_data'):
283 self.xenstore_data = {}
285 self.xenstore_data['storage-type'] = CephFSSR.DRIVER_TYPE
287 return super(CephFSFileVDI, self).attach(sr_uuid, vdi_uuid)
289 @override
290 def generate_config(self, sr_uuid, vdi_uuid) -> str:
291 util.SMlog("SMBFileVDI.generate_config")
292 if not util.pathexists(self.path):
293 raise xs_errors.XenError('VDIUnavailable')
294 resp = {'device_config': self.sr.dconf,
295 'sr_uuid': sr_uuid,
296 'vdi_uuid': vdi_uuid,
297 'sr_sm_config': self.sr.sm_config,
298 'command': 'vdi_attach_from_config'}
299 # Return the 'config' encoded within a normal XMLRPC response so that
300 # we can use the regular response/error parsing code.
301 config = xmlrpc.client.dumps(tuple([resp]), "vdi_attach_from_config")
302 return xmlrpc.client.dumps((config,), "", True)
304 @override
305 def attach_from_config(self, sr_uuid, vdi_uuid) -> str:
306 try:
307 if not util.pathexists(self.sr.path):
308 return self.sr.attach(sr_uuid)
309 except:
310 util.logException("SMBFileVDI.attach_from_config")
311 raise xs_errors.XenError('SRUnavailable',
312 opterr='Unable to attach from config')
313 return ''
315if __name__ == '__main__': 315 ↛ 316line 315 didn't jump to line 316, because the condition on line 315 was never true
316 SRCommand.run(CephFSSR, DRIVER_INFO)
317else:
318 SR.registerSR(CephFSSR)