Coverage for drivers/LUNperVDI.py : 22%

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/python3
2#
3# Copyright (C) Citrix Systems Inc.
4#
5# This program is free software; you can redistribute it and/or modify
6# it under the terms of the GNU Lesser General Public License as published
7# by the Free Software Foundation; version 2.1 only.
8#
9# This program is distributed in the hope that it will be useful,
10# but WITHOUT ANY WARRANTY; without even the implied warranty of
11# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12# GNU Lesser General Public License for more details.
13#
14# You should have received a copy of the GNU Lesser General Public License
15# along with this program; if not, write to the Free Software Foundation, Inc.,
16# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17#
18# LUNperVDI: Generic Raw LUN handler, used by HBASR and ISCSISR
19#
21from sm_typing import override
23import os
24import VDI
25import util
26import scsiutil
27import xs_errors
29MAX_TIMEOUT = 15
32class RAWVDI(VDI.VDI):
33 @override
34 def load(self, vdi_uuid) -> None:
35 if not self.sr.attached: 35 ↛ 36line 35 didn't jump to line 36, because the condition on line 35 was never true
36 raise xs_errors.XenError('SRUnavailable')
38 self.uuid = vdi_uuid
39 self.location = vdi_uuid
40 self.managed = False
41 try:
42 vdi_ref = self.sr.session.xenapi.VDI.get_by_uuid(vdi_uuid)
43 self.managed = self.sr.session.xenapi.VDI.get_managed(vdi_ref)
44 self.sm_config = self.sr.session.xenapi.VDI.get_sm_config(vdi_ref)
45 self.path = self.sr.mpathmodule.path(self.sm_config['SCSIid'])
46 except:
47 pass
48 if self.sr.cmd == "vdi_introduce": 48 ↛ 49line 48 didn't jump to line 49, because the condition on line 48 was never true
49 self.managed = True
51 def _query(self, path, id):
52 self.uuid = scsiutil.gen_uuid_from_string(scsiutil.getuniqueserial(path))
53 self.location = self.uuid
54 self.vendor = scsiutil.getmanufacturer(path)
55 self.serial = scsiutil.getserial(path)
56 self.LUNid = id
57 self.size = scsiutil.getsize(path)
58 self.SCSIid = scsiutil.getSCSIid(path)
59 self.path = path
60 sm_config = util.default(self, "sm_config", lambda: {})
61 sm_config['LUNid'] = str(self.LUNid)
62 sm_config['SCSIid'] = self.SCSIid
63 sm_config['backend-kind'] = 'vbd'
64 self.sm_config = sm_config
66 @override
67 def introduce(self, sr_uuid, vdi_uuid) -> str:
68 self.sm_config = self.sr.srcmd.params['vdi_sm_config']
69 vdi_path = self.sr._getLUNbySMconfig(self.sm_config)
70 self._query(vdi_path, self.sm_config['LUNid'])
71 vdi_uuid = self.uuid
72 self.sr.vdis[vdi_uuid] = self
74 try:
75 util._getVDI(self.sr, vdi_uuid)
76 self.sr.vdis[vdi_uuid]._db_update()
77 # For reasons I don't understand, VDI._db_update() doesn't set the
78 # managed flag, so we do that ourselves here.
79 vdi_ref = self.sr.session.xenapi.VDI.get_by_uuid(vdi_uuid)
80 self.sr.session.xenapi.VDI.set_managed(vdi_ref, self.managed)
81 except:
82 self.sr.vdis[vdi_uuid]._db_introduce()
83 return super(RAWVDI, self).get_params()
85 @override
86 def create(self, sr_uuid, vdi_uuid, size) -> str:
87 VDIs = util._getVDIs(self.sr)
88 self.sr._loadvdis()
89 smallest = 0
90 v = {}
91 for vdi in VDIs:
92 if not vdi['managed'] \
93 and int(vdi['virtual_size']) >= int(size) \
94 and vdi['uuid'] in self.sr.vdis:
95 if not smallest:
96 smallest = int(vdi['virtual_size'])
97 v = vdi
98 elif int(vdi['virtual_size']) < smallest:
99 smallest = int(vdi['virtual_size'])
100 v = vdi
101 if smallest > 0:
102 self.managed = True
103 self.sr.session.xenapi.VDI.set_managed(v['vdi_ref'], self.managed)
104 return super(RAWVDI, self.sr.vdis[v['uuid']]).get_params()
105 raise xs_errors.XenError('SRNoSpace')
107 @override
108 def delete(self, sr_uuid, vdi_uuid, data_only=False) -> None:
109 try:
110 vdi = util._getVDI(self.sr, vdi_uuid)
111 if not vdi['managed']:
112 return
113 sm_config = vdi['sm_config']
114 self.sr.session.xenapi.VDI.set_managed(vdi['vdi_ref'], False)
115 except:
116 pass
118 @override
119 def attach(self, sr_uuid, vdi_uuid) -> str:
120 self.sr._loadvdis()
121 if vdi_uuid not in self.sr.vdis:
122 raise xs_errors.XenError('VDIUnavailable')
123 if not util.pathexists(self.path):
124 self.sr.refresh()
125 if 'SCSIid' in self.sm_config:
126 if self.sr.mpath == 'true':
127 self.sr.mpathmodule.refresh(self.sm_config['SCSIid'], 0)
128 devs = os.listdir("/dev/disk/by-scsid/%s" % self.sm_config['SCSIid'])
129 for dev in devs:
130 realdev = os.path.realpath("/dev/disk/by-scsid/%s/%s" % (self.sm_config['SCSIid'], dev))
131 util.set_scheduler(realdev.split("/")[-1])
132 if not util.wait_for_path(self.path, MAX_TIMEOUT):
133 util.SMlog("Unable to detect LUN attached to host [%s]" % self.sr.path)
134 raise xs_errors.XenError('VDIUnavailable')
135 return super(RAWVDI, self).attach(sr_uuid, vdi_uuid)
137 @override
138 def detach(self, sr_uuid, vdi_uuid) -> None:
139 self.sr._loadvdis()
140 if 'SCSIid' in self.sm_config:
141 self.sr.mpathmodule.reset(self.sm_config['SCSIid'], True) # explicitly unmap
142 if vdi_uuid not in self.sr.vdis:
143 raise xs_errors.XenError('VDIUnavailable')
145 def _set_managed(self, vdi_uuid, managed):
146 try:
147 vdi = util._getVDI(self.sr, vdi_uuid)
148 self.sr.session.xenapi.VDI.set_managed(vdi['vdi_ref'], managed)
149 except:
150 raise xs_errors.XenError('VDIUnavailable')