Coverage for drivers/udevSR.py : 42%

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# udevSR: represents VDIs which are hotplugged into dom0 via udev e.g.
19# USB CDROM/disk devices
21from sm_typing import override
23import SR
24import VDI
25import SRCommand
26import util
27import os
28import time
29import stat
30import xs_errors
31import sysdevice
33CAPABILITIES = ["VDI_INTRODUCE", "VDI_ATTACH", "VDI_DETACH", "VDI_UPDATE", "SR_UPDATE"]
35CONFIGURATION = \
36 [['location', 'path to mount (required) (e.g. server:/path)']]
38DRIVER_INFO = {
39 'name': 'udev',
40 'description': 'SR plugin which represents devices plugged in via udev as VDIs',
41 'vendor': 'Citrix Systems Inc',
42 'copyright': '(C) 2008 Citrix Systems Inc',
43 'driver_version': '1.0',
44 'required_api_version': '1.1',
45 'capabilities': CAPABILITIES,
46 'configuration': CONFIGURATION
47 }
49TYPE = 'udev'
52class udevSR(SR.SR):
53 """udev-driven storage repository"""
55 @override
56 @staticmethod
57 def handles(type) -> bool:
58 if type == TYPE:
59 return True
60 return False
62 @override
63 def content_type(self, sr_uuid) -> str:
64 return super(udevSR, self).content_type(sr_uuid)
66 @override
67 def vdi(self, uuid) -> VDI.VDI:
68 util.SMlog("params = %s" % (self.srcmd.params.keys()))
70 if 'vdi_location' in self.srcmd.params: 70 ↛ 71line 70 didn't jump to line 71, because the condition on line 70 was never true
71 vdi_location = self.srcmd.params['vdi_location']
72 else:
73 vdi_location = self.get_vdi_location(uuid)
75 return udevVDI(self, vdi_location)
77 def get_vdi_location(self, uuid):
78 vdi = self.session.xenapi.VDI
79 vdi_ref = vdi.get_by_uuid(uuid)
80 return vdi.get_location(vdi_ref)
82 @override
83 def load(self, sr_uuid) -> None:
84 # First of all, check we've got the correct keys in dconf
85 if 'location' not in self.dconf:
86 raise xs_errors.XenError('ConfigLocationMissing')
87 self.sr_vditype = 'phy'
88 # Cache the sm_config
89 self.sm_config = self.session.xenapi.SR.get_sm_config(self.sr_ref)
91 @override
92 def update(self, sr_uuid) -> None:
93 # Return as much information as we have
94 sr_root = self.dconf['location']
96 if util.pathexists(sr_root):
97 for filename in os.listdir(sr_root):
98 path = os.path.join(sr_root, filename)
99 x = udevVDI(self, path)
100 self.vdis[path] = x
102 the_sum = 0
103 for vdi in self.vdis.values():
104 the_sum = the_sum + vdi.size
106 self.physical_size = the_sum
107 self.physical_utilisation = the_sum
108 self.virtual_allocation = the_sum
110 self._db_update()
112 @override
113 def scan(self, sr_uuid) -> None:
114 self.update(sr_uuid)
116 # base class scan does all the work:
117 super(udevSR, self).scan(sr_uuid)
119 @override
120 def create(self, sr_uuid, size) -> None:
121 pass
123 @override
124 def delete(self, sr_uuid) -> None:
125 pass
127 @override
128 def attach(self, sr_uuid) -> None:
129 pass
131 @override
132 def detach(self, sr_uuid) -> None:
133 pass
136def read_whole_file(filename):
137 f = open(filename, 'r')
138 try:
139 return f.readlines()
140 finally:
141 f.close()
144class udevVDI(VDI.VDI):
145 def __init__(self, sr, location):
146 self.location = location
147 VDI.VDI.__init__(self, sr, None)
149 @override
150 def load(self, location) -> None:
151 self.path = self.location
152 self.size = 0
153 self.utilisation = 0
154 self.label = self.path
155 self.sm_config = {}
156 try:
157 s = os.stat(self.path)
158 self.deleted = False
160 # Use the CTIME of the symlink to mean "time it was hotplugged"
161 iso8601 = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(s[stat.ST_CTIME]))
162 self.sm_config['hotplugged_at'] = iso8601
164 self.path = os.path.realpath(self.path)
166 dev = os.path.basename(self.path)
167 info = sysdevice.stat(dev)
168 if "size" in info.keys():
169 self.size = info["size"]
170 self.utilisation = self.size
172 self.label = "%s %s" % (info["bus"], info["bus_path"])
173 self.description = info["hwinfo"]
175 # XXX: what other information can we recover?
176 if 'type' in self.sr.sm_config:
177 self.read_only = self.sr.sm_config['type'] == "cd"
179 usb_path = info.get("usb_path")
180 if usb_path:
181 self.sm_config["usb_path"] = info["usb_path"]
182 pusbs = self.session.xenapi.PUSB.get_all_records()
183 for pusb in pusbs.values():
184 if usb_path == pusb.get("path"):
185 if pusb.get("passthrough_enabled"):
186 raise xs_errors.XenError('VDIUnavailable')
187 break
189 except OSError as e:
190 self.deleted = True
192 @override
193 def introduce(self, sr_uuid, vdi_uuid) -> str:
194 self.uuid = vdi_uuid
195 self.location = self.sr.srcmd.params['vdi_location']
196 self._db_introduce()
197 # Update the physical_utilisation etc
198 self.sr.update(sr_uuid)
199 return super(udevVDI, self).get_params()
201 @override
202 def update(self, sr_uuid, vdi_location) -> None:
203 self.load(vdi_location)
204 # _db_update requires self.uuid to be set
205 self.uuid = self.sr.srcmd.params['vdi_uuid']
206 self._db_update()
207 # also reset the name-label and description since we're a bit of
208 # a special SR
209 # this would lead to an infinite loop as VDI.set_name_label now
210 # calls VDI.update
211 # temporarily commenting this to pass quicktest
212 #vdi = self.sr.session.xenapi.VDI.get_by_uuid(self.uuid)
213 #self.sr.session.xenapi.VDI.set_name_label(vdi, self.label)
214 #self.sr.session.xenapi.VDI.set_name_description(vdi, self.description)
216 @override
217 def attach(self, sr_uuid, vdi_uuid) -> str:
218 if self.deleted:
219 raise xs_errors.XenError('VDIUnavailable')
221 return super(udevVDI, self).attach(sr_uuid, vdi_uuid)
223 @override
224 def detach(self, sr_uuid, vdi_uuid) -> None:
225 pass
227if __name__ == '__main__': 227 ↛ 228line 227 didn't jump to line 228, because the condition on line 227 was never true
228 SRCommand.run(udevSR, DRIVER_INFO)
229else:
230 SR.registerSR(udevSR)