Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def init(self):
"""Init the connection to the Riemann server."""
if not self.export_enable:
return None
try:
client = bernhard.Client(host=self.host, port=self.port)
return client
except Exception as e:
logger.critical("Connection to Riemann failed : %s " % e)
return None
def init(self):
"""Init the connection to the RESTful server."""
if not self.export_enable:
return None
# Build the RESTful URL where the stats will be posted
url = '{}://{}:{}{}'.format(self.protocol,
self.host,
self.port,
self.path)
logger.info(
"Stats will be exported to the RESTful endpoint {}".format(url))
return url
def init(self):
"""Init the connection to the CouchDB server."""
if not self.export_enable:
return None
server_uri = 'tcp://{}:{}'.format(self.host, self.port)
try:
self.context = zmq.Context()
publisher = self.context.socket(zmq.PUB)
publisher.bind(server_uri)
except Exception as e:
logger.critical("Cannot connect to ZeroMQ server %s (%s)" % (server_uri, e))
sys.exit(2)
else:
logger.info("Connected to the ZeroMQ server %s" % server_uri)
return publisher
def load(self, config):
"""Load the password from the configuration file."""
password_dict = {}
if config is None:
logger.warning("No configuration file available. Cannot load password list.")
elif not config.has_section(self._section):
logger.warning("No [%s] section in the configuration file. Cannot load password list." % self._section)
else:
logger.info("Start reading the [%s] section in the configuration file" % self._section)
password_dict = dict(config.items(self._section))
# Password list loaded
logger.info("%s password(s) loaded from the configuration file" % len(password_dict))
logger.debug("Password dictionary: %s" % password_dict)
return password_dict
def _web_scan(self, web):
"""Scan the Web/URL (dict) and update the status key."""
try:
req = requests.head(web['url'],
allow_redirects=True,
verify=web['ssl_verify'],
proxies=web['proxies'],
timeout=web['timeout'])
except Exception as e:
logger.debug(e)
web['status'] = 'Error'
web['elapsed'] = 0
else:
web['status'] = req.status_code
web['elapsed'] = req.elapsed.total_seconds()
return web
def export(self, name, columns, points):
"""Write the points to the ZeroMQ server."""
logger.debug("Export {} stats to ZeroMQ".format(name))
# Create DB input
data = dict(zip(columns, points))
# Do not publish empty stats
if data == {}:
return False
# Glances envelopes the stats in a publish message with two frames:
# - First frame containing the following prefix (STRING)
# - Second frame with the Glances plugin name (STRING)
# - Third frame with the Glances plugin stats (JSON)
message = [b(self.prefix),
b(name),
asbytes(json.dumps(data))]
def wrapper(*args, **kw):
counter = Counter()
ret = fct(*args, **kw)
duration = counter.get()
logger.debug("%s %s %s return %s in %s seconds" % (
args[0].__class__.__name__,
args[0].__class__.__module__[len('glances_'):],
fct.__name__, ret,
duration))
return ret
return wrapper
def __display_server(self, server):
"""
Connect and display the given server
"""
# Display the Glances client for the selected server
logger.debug("Selected server: {}".format(server))
# Connection can take time
# Display a popup
self.screen.display_popup(
'Connect to {}:{}'.format(server['name'], server['port']), duration=1)
# A password is needed to access to the server's stats
if server['password'] is None:
# First of all, check if a password is available in the [passwords] section
clear_password = self.password.get_password(server['name'])
if (clear_password is None or self.get_servers_list()
[self.screen.active_server]['status'] == 'PROTECTED'):
# Else, the password should be enter by the user
# Display a popup to enter password
clear_password = self.screen.display_popup(
'Password needed for {}: '.format(server['name']), is_input=True)