abfs.py 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630
  1. #!/usr/bin/env python
  2. # Licensed to Cloudera, Inc. under one
  3. # or more contributor license agreements. See the NOTICE file
  4. # distributed with this work for additional information
  5. # regarding copyright ownership. Cloudera, Inc. licenses this file
  6. # to you under the Apache License, Version 2.0 (the
  7. # "License"); you may not use this file except in compliance
  8. # with the License. You may obtain a copy of the License at
  9. #
  10. # http://www.apache.org/licenses/LICENSE-2.0
  11. #
  12. # Unless required by applicable law or agreed to in writing, software
  13. # distributed under the License is distributed on an "AS IS" BASIS,
  14. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. # See the License for the specific language governing permissions and
  16. # limitations under the License.
  17. """
  18. Interfaces for ABFS
  19. """
  20. from future import standard_library
  21. standard_library.install_aliases()
  22. from builtins import object
  23. import logging
  24. import os
  25. import threading
  26. from math import ceil
  27. from posixpath import join
  28. from urllib.parse import urlparse
  29. from hadoop.hdfs_site import get_umask_mode
  30. from hadoop.fs.exceptions import WebHdfsException
  31. from desktop.lib.rest import http_client, resource
  32. import azure.abfs.__init__ as Init_ABFS
  33. from azure.abfs.abfsfile import ABFSFile
  34. from azure.abfs.abfsstats import ABFSStat
  35. from azure.conf import PERMISSION_ACTION_ABFS
  36. LOG = logging.getLogger(__name__)
  37. #Azure has a 30MB block limit on upload.
  38. UPLOAD_CHUCK_SIZE = 30 * 1000 * 1000
  39. class ABFSFileSystemException(IOError):
  40. def __init__(self, *args, **kwargs):
  41. super(ABFSFileSystemException, self).__init__(*args, **kwargs)
  42. class ABFS(object):
  43. def __init__(self, url,
  44. fs_defaultfs,
  45. logical_name=None,
  46. hdfs_superuser=None,
  47. security_enabled=False,
  48. ssl_cert_ca_verify=True,
  49. temp_dir="/tmp",
  50. umask=0o1022,
  51. hdfs_supergroup=None,
  52. auth_provider=None):
  53. self._url = url
  54. self._superuser = hdfs_superuser
  55. self._security_enabled = security_enabled
  56. self._ssl_cert_ca_verify = ssl_cert_ca_verify
  57. self._temp_dir = temp_dir
  58. self._umask = umask
  59. self._fs_defaultfs = fs_defaultfs
  60. self._logical_name = logical_name
  61. self._supergroup = hdfs_supergroup
  62. self._auth_provider = auth_provider
  63. split = urlparse(fs_defaultfs)
  64. self._scheme = split.scheme
  65. self._netloc = split.netloc
  66. self._is_remote = True
  67. self._has_trash_support = False
  68. self._filebrowser_action = PERMISSION_ACTION_ABFS
  69. self._client = http_client.HttpClient(url, exc_class=WebHdfsException, logger=LOG)
  70. self._root = resource.Resource(self._client)
  71. # To store user info
  72. self._thread_local = threading.local()
  73. LOG.debug("Initializing ABFS : %s (security: %s, superuser: %s)" % (self._url, self._security_enabled, self._superuser))
  74. @classmethod
  75. def from_config(cls, hdfs_config, auth_provider):
  76. return cls(url=hdfs_config.WEBHDFS_URL.get(),
  77. fs_defaultfs=hdfs_config.FS_DEFAULTFS.get(),
  78. logical_name=None,
  79. security_enabled=False,
  80. ssl_cert_ca_verify=False,
  81. temp_dir=None,
  82. umask=get_umask_mode(),
  83. hdfs_supergroup=None,
  84. auth_provider=auth_provider)
  85. def _getheaders(self):
  86. return {
  87. "Authorization": self._auth_provider.get_token(),
  88. }
  89. # Parse info about filesystems, directories, and files
  90. # --------------------------------
  91. def isdir(self, path):
  92. """
  93. Checks if the path is a directory (note diabled because filebrowser/views is bugged)
  94. """
  95. resp = self.stats(path)
  96. #LOG.debug("checking directoty or not")
  97. return resp.isDir
  98. def isfile(self, path):
  99. """
  100. Checks if the path is a file
  101. """
  102. return not self.isdir(path)
  103. def exists(self, path):
  104. """
  105. Test if a path exists
  106. """
  107. try:
  108. #LOG.debug("checking existence")
  109. if ABFS.isroot(path):
  110. return True
  111. self.stats(path)
  112. except WebHdfsException as e:
  113. if e.code == 404:
  114. return False
  115. raise WebHdfsException
  116. return True
  117. def stats(self, path, params = None, **kwargs):
  118. """
  119. List the stat of the actual file/directory
  120. Returns the ABFFStat object
  121. """
  122. if ABFS.isroot(path):
  123. return ABFSStat.for_root()
  124. file_system, dir_name = Init_ABFS.parse_uri(path)[:2]
  125. if dir_name == '':
  126. LOG.debug("Path being called is a Filesystem")
  127. return ABFSStat.for_filesystem(self._statsf(file_system, params, **kwargs), path)
  128. return ABFSStat.for_single(self._stats(file_system + '/' +dir_name, params, **kwargs), path)
  129. def listdir_stats(self,path, params = None, **kwargs):
  130. """
  131. List the stats for the directories inside the specified path
  132. Returns the Multiple ABFFStat object #note change later for recursive cases
  133. """
  134. if ABFS.isroot(path):
  135. LOG.warn("Path: %s is a Filesystem" %path)
  136. return self.listfilesystems_stats(params = None, **kwargs)
  137. dir_stats = []
  138. file_system, directory_name = Init_ABFS.parse_uri(path)[:2]
  139. if params is None:
  140. params = {}
  141. if 'recursive' not in params:
  142. params['recursive'] = 'false'
  143. params['resource'] = 'filesystem'
  144. if directory_name != "":
  145. params['directory'] = directory_name
  146. res = self._root._invoke("GET",file_system, params, headers= self._getheaders(), **kwargs)
  147. resp = self._root._format_response(res)
  148. for x in resp['paths']:
  149. dir_stats.append(ABFSStat.for_directory(res.headers, x, Init_ABFS.ABFS_ROOT +file_system + "/" + x['name']))
  150. return dir_stats
  151. def listfilesystems_stats(self, params = None, **kwargs):
  152. """
  153. Lists the stats inside the File Systems, No functionality for params
  154. """
  155. stats = []
  156. if params is None:
  157. params = {}
  158. params["resource"] = "account"
  159. res = self._root._invoke("GET", params = params, headers = self._getheaders() )
  160. resp = self._root._format_response(res)
  161. for x in resp['filesystems']:
  162. stats.append(ABFSStat.for_filesystems(res.headers, x))
  163. return stats
  164. def _stats(self, schemeless_path, params = None, **kwargs):
  165. """
  166. Container function for both stats,
  167. Returns the header of the result
  168. """
  169. if params is None:
  170. params = {}
  171. params['action'] = 'getStatus'
  172. res = self._root._invoke('HEAD', schemeless_path, params, headers = self._getheaders(), **kwargs)
  173. return res.headers
  174. def _statsf(self, schemeless_path, params = None, **kwargs):
  175. """
  176. Continer function for both stats but if it's a file system
  177. Returns the header of the result
  178. """
  179. if params is None:
  180. params = {}
  181. params['resource'] = 'filesystem'
  182. res = self._root._invoke('HEAD', schemeless_path, params, headers = self._getheaders(), **kwargs)
  183. return res.headers
  184. def listdir(self, path, params = None, glob=None, **kwargs):
  185. """
  186. Lists the names inside the current directories
  187. """
  188. if ABFS.isroot(path):
  189. LOG.warn("Path being called is a Filesystem")
  190. return self.listfilesystems(params, **kwargs)
  191. listofDir = self.listdir_stats(path, params)
  192. return [x.name for x in listofDir]
  193. def listfilesystems(self, params=None,**kwargs):
  194. """
  195. Lists the names of the File Systems, limited arguements
  196. """
  197. listofFileSystems = self.listfilesystems_stats(params)
  198. return [x.name for x in listofFileSystems]
  199. # Find or alter information about the URI path
  200. # --------------------------------
  201. @staticmethod
  202. def isroot(path):
  203. """
  204. Checks if the path is the root path
  205. """
  206. return Init_ABFS.is_root(path)
  207. @staticmethod
  208. def normpath(path):
  209. """
  210. Normalizes a path
  211. """
  212. resp = Init_ABFS.normpath(path)
  213. return resp
  214. @staticmethod
  215. def netnormpath(path):
  216. """
  217. Normalizes a path
  218. """
  219. return Init_ABFS.normpath(path)
  220. def open(self, path, option = 'r', *args, **kwargs):
  221. return ABFSFile(self,path, option )
  222. @staticmethod
  223. def parent_path(path):
  224. """
  225. Returns the Parent Path
  226. """
  227. return Init_ABFS.parent_path(path)
  228. @staticmethod
  229. def join(first, *comp_list):
  230. """
  231. Joins two paths together
  232. """
  233. return Init_ABFS.join(first,*comp_list)
  234. # Create Files,directories, or File Systems
  235. # --------------------------------
  236. def mkdir(self, path, params = None, headers = None, *args, **kwargs):
  237. """
  238. Makes a directory
  239. """
  240. if params is None:
  241. params = {}
  242. params['resource'] = 'directory'
  243. self._create_path(path, params = params, headers = params, overwrite = False)
  244. def create(self, path, overwrite= False, data = None, headers = None, *args, **kwargs):
  245. """
  246. Makes a File (Put text in data if adding data)
  247. """
  248. params = {'resource' : 'file'}
  249. self._create_path(path, params = params, headers =headers, overwrite = overwrite)
  250. if data:
  251. self._writedata(path, data, len(data))
  252. def create_home_dir(self, home_path=None):
  253. raise NotImplementedError("File System not named")
  254. def _create_path(self,path, params = None, headers = None, overwrite = False):
  255. """
  256. Container method for Create
  257. """
  258. file_system, dir_name = Init_ABFS.parse_uri(path)[:2]
  259. if dir_name == '':
  260. return self._create_fs(file_system)
  261. no_scheme = file_system + '/' + dir_name
  262. additional_header = self._getheaders()
  263. if headers is not None:
  264. additional_header.update(headers)
  265. if not overwrite:
  266. additional_header['If-None-Match'] = '*'
  267. self._root.put(no_scheme,params, headers= additional_header)
  268. def _create_fs(self, file_system):
  269. """
  270. Creates a File System
  271. """
  272. self._root.put(file_system,{'resource': 'filesystem'}, headers= self._getheaders())
  273. # Read Files
  274. # --------------------------------
  275. def read(self, path, offset = '0', length = 0, *args, **kwargs):
  276. """
  277. Read data from a file
  278. """
  279. path = Init_ABFS.strip_scheme(path)
  280. headers = self._getheaders()
  281. if length != 0 and length != '0':
  282. headers['range']= 'bytes=%s-%s' %(str(offset), str(int(offset) + int(length)))
  283. return self._root.get(path, headers = headers)
  284. # Alter Files
  285. # --------------------------------
  286. def append(self, path, data, size = 0, offset =0 ,params = None, **kwargs):
  287. """
  288. Appends the data to a file
  289. """
  290. path = Init_ABFS.strip_scheme(path)
  291. if params is None:
  292. LOG.warn("Params not specified, Append will take longer")
  293. resp = self._stats(path)
  294. params = {'position' : int(resp['Content-Length']) + offset, 'action' : 'append'}
  295. LOG.debug("%s" %params)
  296. else:
  297. params['action'] = 'append'
  298. headers = {}
  299. if size == 0:
  300. headers['Content-Length'] = str(len(data))
  301. else:
  302. headers['Content-Length'] = str(size)
  303. LOG.debug("%s" %headers['Content-Length'])
  304. return self._patching_sl( path, params, data, headers, **kwargs)
  305. def flush(self, path, params = None, headers = None, **kwargs):
  306. """
  307. Flushes the data(i.e. writes appended data to File)
  308. """
  309. path = Init_ABFS.strip_scheme(path)
  310. if params is None:
  311. LOG.warn("Params not specified")
  312. params = {'position' : 0}
  313. if 'position' not in params:
  314. LOG.warn("Position is not specified")
  315. params['position'] = 0
  316. params['action'] = 'flush'
  317. if headers is None:
  318. headers = {}
  319. headers['Content-Length'] = '0'
  320. self._patching_sl( path, params, header = headers, **kwargs)
  321. # Remove Filesystems, directories. or Files
  322. # --------------------------------
  323. def remove(self, path, skip_trash=True):
  324. """
  325. Removes an item indicated in the path
  326. Also removes empty directories
  327. """
  328. self._delete(path, recursive = 'false', skip_trash = skip_trash)
  329. def rmtree(self, path, skip_trash = True):
  330. """
  331. Remove everything in a given directory
  332. """
  333. self._delete(path, recursive = 'true', skip_trash = skip_trash)
  334. def _delete(self, path, recursive = 'false', skip_trash=True):
  335. """
  336. Wrapper function for calling delete, no support for trash or
  337. """
  338. if not skip_trash:
  339. raise NotImplementedError("Trash not implemented for ABFS")
  340. if ABFS.isroot(path):
  341. raise RuntimeError("Cannot Remove Root")
  342. file_system, dir_name = Init_ABFS.parse_uri(path)[:2]
  343. if dir_name == '':
  344. return self._root.delete(file_system,{'resource': 'filesystem'}, headers= self._getheaders())
  345. new_path = file_system + '/' + dir_name
  346. param = None
  347. if self.isdir(path):
  348. param = {'recursive' : recursive}
  349. self._root.delete(new_path,param , headers= self._getheaders())
  350. def restore(self, path):
  351. raise NotImplementedError("")
  352. # Edit permissions of Filesystems, directories. or Files
  353. # --------------------------------
  354. def chown(self, path, user = None, group = None, *args, **kwargs):
  355. """
  356. Changes ownership (not implemented)
  357. """
  358. headers = {}
  359. if user is not None:
  360. headers['x-ms-owner'] = user
  361. if group is not None:
  362. headers['x-ms-group'] = group
  363. self.setAccessControl(path, headers = headers, **kwargs)
  364. def chmod(self, path, permissionNumber = None, *args, **kwargs):
  365. """
  366. Set File Permissions (not implemented)
  367. """
  368. header = {}
  369. if permissionNumber is not None:
  370. header['x-ms-permissions'] = str(permissionNumber)
  371. self.setAccessControl(path, headers = header)
  372. def setAccessControl(self, path, headers, **kwargs):
  373. """
  374. Set Access Controls (Can do both chmod and chown) (not implemented)
  375. """
  376. path = Init_ABFS.strip_scheme(path)
  377. params= {'action': 'setAccessControl'}
  378. if headers is None:
  379. headers ={}
  380. self._patching_sl( path, params, header = headers, **kwargs)
  381. def mktemp(self, subdir='', prefix='tmp', basedir=None):
  382. raise NotImplementedError("")
  383. def purge_trash(self):
  384. raise NotImplementedError("")
  385. # Handle file systems interactions
  386. # --------------------------------
  387. def copy(self, src, dst, *args, **kwargs):
  388. """
  389. General Copying
  390. """
  391. if self.isfile(src):
  392. return self.copyfile(src ,dst)
  393. self.copy_remote_dir(src, dst)
  394. def copyfile(self, src, dst, *args, **kwargs):
  395. """
  396. Copies a File to another location
  397. """
  398. new_path = dst + '/' + Init_ABFS.strip_path(src)
  399. self.create(new_path)
  400. chunk_size = self.get_upload_chuck_size()
  401. file = self.read(src)
  402. size = len(file)
  403. self._writedata(new_path, file, size)
  404. def copy_remote_dir(self, src, dst, *args, **kwargs):
  405. """
  406. Copies the entire contents of a directory to another location
  407. """
  408. dst = dst + '/' + Init_ABFS.strip_path(src)
  409. LOG.debug("%s" %dst)
  410. self.mkdir(dst)
  411. other_files = self.listdir(src)
  412. for x in other_files:
  413. x = src + '/' + Init_ABFS.strip_path(x)
  414. LOG.debug("%s" %x)
  415. self.copy(x, dst)
  416. def rename(self, old, new):
  417. """
  418. Renames a file
  419. """
  420. LOG.debug("%s\n%s" %(old, new))
  421. headers = {'x-ms-rename-source' : '/' + Init_ABFS.strip_scheme(old) }
  422. try:
  423. self._create_path(new, headers = headers, overwrite= True)
  424. except WebHdfsException as e:
  425. if e.code == 409:
  426. self.copy(old, new)
  427. self.rmtree(old)
  428. else:
  429. raise e
  430. def rename_star(self, old_dir, new_dir):
  431. """
  432. Renames a directory
  433. """
  434. self.rename(old_dir, new_dir)
  435. def upload(self, file, path, *args, **kwargs):
  436. """
  437. Upload is done by the client
  438. """
  439. pass
  440. def copyFromLocal(self, local_src, remote_dst, *args, **kwargs):
  441. """
  442. Copy a directory or file from Local (Testing)
  443. """
  444. local_src = local_src.endswith('/') and local_src[:-1] or local_src
  445. remote_dst = remote_dst.endswith('/') and remote_dst[:-1] or remote_dst
  446. if os.path.isdir(local_src):
  447. self._local_copy_dir(local_src,remote_dst)
  448. else:
  449. (basename, filename) = os.path.split(local_src)
  450. self._local_copy_file(local_src, self.isdir(remote_dst) and self.join(remote_dst, filename) or remote_dst)
  451. def _local_copy_dir(self,local_src,remote_dst):
  452. """
  453. A wraper function for copying local directories
  454. """
  455. self.mkdir(remote_dir)
  456. for f in os.listdir(local_dir):
  457. local_src = os.path.join(local_dir, f)
  458. remote_dst = self.join(remote_dir, f)
  459. if os.path.isdir(local_src):
  460. self._copy_dir(local_src, remote_dst, mode)
  461. else:
  462. self._copy_file(local_src, remote_dst)
  463. def _local_copy_file(self,local_src,remote_dst, chunk_size = UPLOAD_CHUCK_SIZE ):
  464. """
  465. A wraper function for copying local Files
  466. """
  467. if os.path.isfile(local_src):
  468. if self.exists(remote_dst):
  469. LOG.info('%s already exists. Skipping.' %remote_dst)
  470. return
  471. else:
  472. LOG.info('%s does not exist. Trying to copy.' % remote_dst)
  473. src = file(local_src)
  474. try:
  475. try:
  476. self.create(remote_dst)
  477. chunk = src.read(chunk_size)
  478. offset = 0
  479. while chunk:
  480. size = len(chunk)
  481. self.append(remote_dst, chunk, size = size, params = {'position' : offset})
  482. offset += size
  483. chunk = src.read(chunk_size)
  484. self.flush(remote_dst, params = {'position' : offset})
  485. LOG.info(_('Copied %s -> %s.') % (local_src, remote_dst))
  486. except:
  487. LOG.exception(_('Copying %s -> %s failed.') % (local_src, remote_dst))
  488. raise
  489. finally:
  490. src.close()
  491. else:
  492. LOG.info(_('Skipping %s (not a file).') % local_src)
  493. def check_access(self, path, *args, **kwargs):
  494. """
  495. Check access of a file/directory (Work in Progress/Not Ready)
  496. """
  497. raise NotImplementedError("")
  498. try:
  499. status = self.stats(path)
  500. if 'x-ms-permissions' not in status.keys():
  501. raise b
  502. except b:
  503. LOG.debug("Permisions have not been set")
  504. except:
  505. Exception
  506. def mkswap(self, filename, subdir='', suffix='swp', basedir=None):
  507. """
  508. Makes a directory and returns a potential filename for that directory
  509. """
  510. base = self.join(basedir or self._temp_dir, subdir)
  511. if not self.isdir(base):
  512. self.mkdir(base)
  513. candidate = self.join(base, "%s.%s" % (filename, suffix))
  514. return candidate
  515. def setuser(self, user):
  516. """
  517. Changes the User
  518. """
  519. self._user = user
  520. def get_upload_chuck_size(self):
  521. """
  522. Gets the maximum size allowed to upload
  523. """
  524. return UPLOAD_CHUCK_SIZE
  525. def filebrowser_action(self):
  526. return self._filebrowser_action
  527. #Other Methods to condense stuff
  528. #----------------------------
  529. # Write Files on creation
  530. #----------------------------
  531. def _writedata(self, path, data, size):
  532. """
  533. Adds text to a given file
  534. """
  535. chunk_size = self.get_upload_chuck_size()
  536. cycles = ceil(float(size) / chunk_size)
  537. for i in range(0,cycles):
  538. chunk = size % chunk_size
  539. if i != cycles or chunk == 0:
  540. length = chunk_size
  541. else:
  542. length = chunk
  543. self.append(path, data[i*chunk_size:i*chunk_size + length], length)
  544. LOG.debug("%s" %data)
  545. self.flush(path, {'position' : int(size) })
  546. # Use Patch HTTP request
  547. #----------------------------
  548. def _patching_sl(self, schemeless_path, param, data = None, header = None, **kwargs):
  549. """
  550. A wraper function for patch
  551. """
  552. if header is None:
  553. header = {}
  554. header.update(self._getheaders())
  555. LOG.debug("%s" %kwargs)
  556. return self._root.invoke('PATCH', schemeless_path, param, data, headers = header, **kwargs)