Skip to content
This repository was archived by the owner on Apr 19, 2022. It is now read-only.

Commit

Permalink
Fix release 2.3.0
Browse files Browse the repository at this point in the history 
Including:
 * make test relative to snakebite package [test]
 * add default port for namenode [feature]
 * test config support for trash [test]
 * add --usetrash, more tests, support for --skiptrash [feature] [fix]
 * fix bug in parser usetrash parameter [fix]
 * cosmetic changes - better print info/error [fix]
 * CL port has the highest priority

Change-Id: I828182f8e4b012d38373f6ff1e686cab9398bcf6
  • Loading branch information
Rafal Wojdyla committed Mar 31, 2014
1 parent b4768bf commit 468b70d
Show file tree
Hide file tree
Showing 13 changed files with 526 additions and 68 deletions.
3 changes: 1 addition & 2 deletions bin/snakebite
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,7 @@ class SnakebiteCli(object):
clparser = CommandLineParser()
self.args = clparser.parse()
self._setup_logging()
clparser.read_config()
clparser.setup_client()
clparser.init()
clparser.execute()

def _setup_logging(self):
Expand Down
142 changes: 104 additions & 38 deletions snakebite/commandlineparser.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,13 @@
from snakebite.namenode import Namenode


def print_error_exit(msg, fd=sys.stderr):
print >> fd, "Error: %s" % msg
sys.exit(-1)

def print_info(msg, fd=sys.stderr):
print >> fd, "Info: %s" % msg

def exitError(error):
if isinstance(error, FileNotFoundException) or \
isinstance(error, DirectoryException) or \
Expand Down Expand Up @@ -100,7 +107,7 @@ class CommandLineParser(object):
"type": float},
'p': {"short": '-p',
"long": '--port',
"help": 'namenode RPC port',
"help": 'namenode RPC port (default: %d)' % Namenode.DEFAULT_PORT,
"type": int},
'h': {"short": '-h',
"long": '--help',
Expand Down Expand Up @@ -129,6 +136,10 @@ class CommandLineParser(object):
"help": 'skip the trash (when trash is enabled)',
"default": False,
"action": 'store_true'},
'T': {"short": '-T',
"long": "--usetrash",
"help": "enable the trash",
"action": 'store_true'},
'z': {"short": '-z',
"long": '--zero',
"help": 'check for zero length',
Expand Down Expand Up @@ -230,6 +241,28 @@ def _add_subparsers(self):
command_parser = subparsers.add_parser(cmd_name, add_help=False, parents=parents)
command_parser.set_defaults(command=cmd_name)

def init(self):
self.read_config()
self._clean_args()
self.setup_client()

def _clean_args(self):
for path in self.__get_all_directories():
if path.startswith('hdfs://'):
parse_result = urlparse(path)
if path in self.args.dir:
self.args.dir.remove(path)
self.args.dir.append(parse_result.path)
else:
self.args.single_arg = parse_result.path

def __usetrash_unset(self):
return not 'usetrash' in self.args or self.args.usetrash == False

def __use_cl_port_first(self, alt):
# Port provided from CL has the highest priority:
return self.args.port if self.args.port else alt

def read_config(self):

# Try to retrieve namenode config from within CL arguments
Expand All @@ -252,10 +285,10 @@ def read_config(self):
if configs:
for config in configs:
nn = Namenode(config['namenode'],
config['port'])
self.__use_cl_port_first(config['port']))
self.namenodes.append(nn)

self.args.skiptrash = not any([c['use_trash'] for c in configs])
if self.__usetrash_unset():
self.args.usetrash = HDFSConfig.use_trash

if len(self.namenodes):
return
Expand All @@ -272,14 +305,15 @@ def read_config(self):
print ' "config_version": 2,'
print ' "use_trash": true,'
print ' "namenodes": ['
print ' {"host": "namenode-ha1", "port": 54310, "version": %d},' % Namenode.DEFAULT_VERSION
print ' {"host": "namenode-ha2", "port": 54310, "version": %d}' % Namenode.DEFAULT_VERSION
print ' {"host": "namenode-ha1", "port": %d, "version": %d},' % (Namenode.DEFAULT_PORT, Namenode.DEFAULT_VERSION)
print ' {"host": "namenode-ha2", "port": %d, "version": %d}' % (Namenode.DEFAULT_PORT, Namenode.DEFAULT_VERSION)
print ' ]'
print '}'

sys.exit(1)

def _read_config_snakebiterc(self):
old_version_info = "You're are using snakebite %s with Trash support together with old snakebiterc, please update/remove your ~/.snakebiterc file. By default Trash is %s." % (version(), 'disabled' if not HDFSConfig.use_trash else 'enabled')
with open(os.path.join(os.path.expanduser('~'), '.snakebiterc')) as config_file:
configs = json.load(config_file)

Expand All @@ -288,52 +322,81 @@ def _read_config_snakebiterc(self):
# config is a list of namenode(s) - possibly HA
for config in configs:
nn = Namenode(config['namenode'],
config['port'],
self.__use_cl_port_first(config.get('port', Namenode.DEFAULT_PORT)),
config.get('version', Namenode.DEFAULT_VERSION))
self.namenodes.append(nn)
if self.__usetrash_unset():
# commandline setting has higher priority
print_info(old_version_info)
# There's no info about Trash in version 1, use default policy:
self.args.usetrash = HDFSConfig.use_trash
elif isinstance(configs, dict):
if configs.get("config_version"):
# Config ersion > 2
# Version 2: {}
# Can be either new configuration or just one namenode
# which was the very first configuration syntax
if 'config_version' in configs:
# Config version => 2
for nn_config in configs['namenodes']:
nn = Namenode(nn_config['host'], nn_config['port'], nn_config.get('version', Namenode.DEFAULT_VERSION))
nn = Namenode(nn_config['host'],
self.__use_cl_port_first(nn_config.get('port', Namenode.DEFAULT_PORT)),
nn_config.get('version', Namenode.DEFAULT_VERSION))
self.namenodes.append(nn)

self.args.skiptrash = configs.get("skiptrash")
if self.__usetrash_unset():
# commandline setting has higher priority
self.args.usetrash = configs.get("use_trash", HDFSConfig.use_trash)
else:
# config is a single namenode - no HA
# config is a single namenode - no HA
self.namenodes.append(Namenode(configs['namenode'],
configs['port'],
self.__use_cl_port_first(configs.get('port', Namenode.DEFAULT_PORT)),
configs.get('version', Namenode.DEFAULT_VERSION)))
if self.__usetrash_unset():
# commandline setting has higher priority
print_info(old_version_info)
self.args.usetrash = HDFSConfig.use_trash
else:
print "Config retrieved from ~/.snakebiterc is corrupted! Remove it!"
sys.exit(1)
print_error_exit("Config retrieved from ~/.snakebiterc is corrupted! Remove it!")

def _read_config_cl(self):
''' Check if any directory arguments contain hdfs://'''
def __get_all_directories(self):
if self.args and 'dir' in self.args:
dirs_to_check = list(self.args.dir)
if self.args.command == 'mv':
dirs_to_check.append(self.args.single_arg)
for directory in dirs_to_check:
if 'hdfs://' in directory:
parse_result = urlparse(directory)

if not self.args.namenode is None and not self.args.port is None and (self.args.port != parse_result.port or self.args.namenode != parse_result.hostname):
print "error: conflicting nodenames or ports"
sys.exit(-1)
else:
self.args.namenode = parse_result.hostname
self.args.port = parse_result.port

if directory in self.args.dir:
self.args.dir.remove(directory)
self.args.dir.append(parse_result.path)
else:
self.args.single_arg = parse_result.path

if self.args.namenode and self.args.port:
# If namenode config found based on arguments, save namenode
return dirs_to_check
else:
return ()

def _read_config_cl(self):
''' Check if any directory arguments contain hdfs://'''
dirs_to_check = self.__get_all_directories()
hosts, ports = [], []
for path in dirs_to_check:
if path.startswith('hdfs://'):
parse_result = urlparse(path)
hosts.append(parse_result.hostname)
ports.append(parse_result.port)

# remove duplicates and None from (hosts + self.args.namenode)
hosts = filter(lambda x: x != None, set(hosts + [self.args.namenode]))
if len(hosts) > 1:
print_error_exit('Conficiting namenode hosts in commandline arguments, hosts: %s' % str(hosts))

ports = filter(lambda x: x != None, set(ports + [self.args.port]))
if len(ports) > 1:
print_error_exit('Conflicting namenode ports in commandline arguments, ports: %s' % str(ports))

# Store port from CL in arguments - CL port has the highest priority
if len(ports) == 1:
self.args.port = ports[0]

# do we agree on one namenode?
if len(hosts) == 1 and len(ports) <= 1:
self.args.namenode = hosts[0]
self.args.port = ports[0] if len(ports) == 1 else Namenode.DEFAULT_PORT
self.namenodes.append(Namenode(self.args.namenode, self.args.port))
# we got the info from CL -> check if use_trash is set - if not use default policy:
if self.__usetrash_unset():
self.args.usetrash = HDFSConfig.use_trash
return True
else:
return False
Expand Down Expand Up @@ -366,7 +429,10 @@ def parse(self, non_cli_input=None): # Allow input for testing purposes
return self.args

def setup_client(self):
use_trash = not self.args.skiptrash
if 'skiptrash' in self.args:
use_trash = self.args.usetrash and not self.args.skiptrash
else:
use_trash = self.args.usetrash
self.client = HAClient(self.namenodes, use_trash)

def execute(self):
Expand Down Expand Up @@ -504,7 +570,7 @@ def mv(self):
for line in format_results(result, json_output=self.args.json):
print line

@command(args="[paths]", descr="remove paths", allowed_opts=["R", "S"], req_args=['dir [dirs]'])
@command(args="[paths]", descr="remove paths", allowed_opts=["R", "S", "T"], req_args=['dir [dirs]'])
def rm(self):
result = self.client.delete(self.args.dir, recurse=self.args.recurse)
for line in format_results(result, json_output=self.args.json):
Expand Down
14 changes: 6 additions & 8 deletions snakebite/config.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import xml.etree.ElementTree as ET

from urlparse import urlparse
from namenode import Namenode

log = logging.getLogger(__name__)

Expand Down Expand Up @@ -48,15 +49,14 @@ def read_core_config(cls, core_site_path):
if property.findall('name')[0].text == 'fs.defaultFS':
parse_result = urlparse(property.findall('value')[0].text)
log.debug("Got namenode '%s' from %s" % (parse_result.geturl(), core_site_path))

config.append({"namenode": parse_result.hostname,
"port": parse_result.port})
"port": parse_result.port if parse_result.port
else Namenode.DEFAULT_PORT})

if property.findall('name')[0].text == 'fs.trash.interval':
cls.use_trash = True

for c in config:
c["use_trash"] = cls.use_trash

return config

@classmethod
Expand All @@ -67,14 +67,12 @@ def read_hdfs_config(cls, hdfs_site_path):
parse_result = urlparse("//" + property.findall('value')[0].text)
log.debug("Got namenode '%s' from %s" % (parse_result.geturl(), hdfs_site_path))
configs.append({"namenode": parse_result.hostname,
"port": parse_result.port})
"port": parse_result.port if parse_result.port
else Namenode.DEFAULT_PORT})

if property.findall('name')[0].text == 'fs.trash.interval':
cls.use_trash = True

for c in configs:
c["use_trash"] = cls.use_trash

return configs

core_try_paths = ('/etc/hadoop/conf/core-site.xml',
Expand Down
6 changes: 3 additions & 3 deletions snakebite/namenode.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,10 @@

class Namenode(object):
'''Namenode class - represents HDFS namenode'''

DEFAULT_PORT = 8020
DEFAULT_VERSION = 9

def __init__(self, host, port, version=DEFAULT_VERSION):
def __init__(self, host, port=DEFAULT_PORT, version=DEFAULT_VERSION):
self.host = host
self.port = port
self.version = version
Expand All @@ -29,4 +29,4 @@ def is_active(self):
def toDict(self):
return {"namenode": self.host,
"port": self.port,
"version": self.version}
"version": self.version}
Empty file added test/__init__.py
View file
Empty file.
Loading

0 comments on commit 468b70d

Please sign in to comment.