Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_config_ssh_user(self):
c = config.Config(self.real_config)
expected = 'carbonate'
self.assertEqual(c.ssh_user('standalone'), expected)
def test_config_destinations(self):
c = config.Config(self.simple_config)
expected = ['1.1.1.1:2003:0', '2.2.2.2:2003:0']
self.assertEqual(c.destinations(), expected)
def test_config_hashing_type_default(self):
c = config.Config(self.simple_config)
expected = 'carbon_ch'
self.assertEqual(c.hashing_type(), expected)
parser.add_argument(
'--rename-separator',
default=',',
help='Character used to separate old and new metric ' +
'names when using "--rename"')
args = parser.parse_args()
logging.basicConfig(level=logging.INFO)
if args.metrics_file and args.metrics_file[0] != '-':
fi = args.metrics_file
else:
fi = []
config = Config(args.config_file)
user = config.ssh_user()
remote_ip = args.source_node
remote = "%s@%s:%s/" % (user, remote_ip, args.source_storage_dir)
metrics_to_sync = []
start = time()
total_metrics = 0
batch_size = int(args.batch_size)
for metric in fileinput.input(fi):
total_metrics += 1
if args.rename:
(old_metric, new_metric) = metric.split(args.rename_separator)
else:
parser.add_argument(
'--rename-separator',
default=',',
help='Character used to separate old and new metric ' +
'names when using "--rename"')
args = parser.parse_args()
logging.basicConfig(level=logging.INFO)
if args.metrics_file and args.metrics_file[0] != '-':
fi = args.metrics_file
else:
fi = []
config = Config(args.config_file)
user = config.ssh_user()
remote_ip = args.source_node
remote = "%s@%s:%s/" % (user, remote_ip, args.source_storage_dir)
metrics_to_sync = []
start = time()
total_metrics = 0
batch_size = int(args.batch_size)
for metric in fileinput.input(fi):
total_metrics += 1
if args.rename:
(old_metric, new_metric) = metric.split(args.rename_separator)
else:
def carbon_lookup():
parser = common_parser('Lookup where a metric lives in a carbon cluster')
parser.add_argument(
'metric', metavar='METRIC', nargs=1,
type=str,
help='Full metric name to search for')
parser.add_argument(
'-s', '--short',
action='store_true',
help='Only display the address, without port and cluster name')
args = parser.parse_args()
config = Config(args.config_file)
cluster = Cluster(config, args.cluster)
results = lookup(str(args.metric[0]), cluster)
if args.short:
for i, _ in enumerate(results):
results[i] = results[i].split(':')[0]
print "\n".join(results)
def carbon_hosts():
parser = common_parser('Return the addresses for all nodes in a cluster')
args = parser.parse_args()
config = Config(args.config_file)
cluster = Cluster(config, args.cluster)
cluster_hosts = [d[0] for d in cluster.destinations]
print "\n".join(cluster_hosts)