Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see .
import sys
from glances.logger import logger
# Import mandatory PySNMP lib
try:
from pysnmp.entity.rfc3413.oneliner import cmdgen
except ImportError:
logger.critical("PySNMP library not found. To install it: pip install pysnmp")
sys.exit(2)
class GlancesSNMPClient(object):
"""SNMP client class (based on pysnmp library)."""
def __init__(self, host='localhost', port=161, version='2c',
community='public', user='private', auth=''):
super(GlancesSNMPClient, self).__init__()
self.cmdGen = cmdgen.CommandGenerator()
self.version = version
self.host = host
def init(self):
"""Init the connection to the Kafka server."""
if not self.export_enable:
return None
# Build the server URI with host and port
server_uri = '{}:{}'.format(self.host, self.port)
try:
s = KafkaProducer(bootstrap_servers=server_uri,
value_serializer=lambda v: json.dumps(v).encode('utf-8'),
compression_type=self.compression)
except Exception as e:
logger.critical("Cannot connect to Kafka server %s (%s)" % (server_uri, e))
sys.exit(2)
else:
logger.info("Connected to the Kafka server %s" % server_uri)
return s
def init(self):
"""Init the connection to the ES server."""
if not self.export_enable:
return None
try:
es = Elasticsearch(hosts=['{}:{}'.format(self.host, self.port)])
except Exception as e:
logger.critical("Cannot connect to ElasticSearch server %s:%s (%s)" % (self.host, self.port, e))
sys.exit(2)
else:
logger.info("Connected to the ElasticSearch server %s:%s" % (self.host, self.port))
try:
index_count = es.count(index=self.index)['count']
except Exception as e:
# Index did not exist, it will be created at the first write
# Create it...
es.indices.create(self.index)
else:
logger.info("There is already %s entries in the ElasticSearch %s index" % (index_count, self.index))
return es
from glances.compat import to_ascii, nativestr, b, u, itervalues
from glances.globals import MACOS, WINDOWS
from glances.logger import logger
from glances.events import glances_events
from glances.processes import glances_processes
from glances.timer import Timer
# Import curses library for "normal" operating system
if not WINDOWS:
try:
import curses
import curses.panel
from curses.textpad import Textbox
except ImportError:
logger.critical("Curses module not found. Glances cannot start in standalone mode.")
sys.exit(1)
class _GlancesCurses(object):
"""This class manages the curses display (and key pressed).
Note: It is a private class, use GlancesCursesClient or GlancesCursesBrowser.
"""
_hotkeys = {
'0': {'switch': 'disable_irix'},
'1': {'switch': 'percpu'},
'2': {'switch': 'disable_left_sidebar'},
'3': {'switch': 'disable_quicklook'},
'6': {'switch': 'meangpu'},
# Init
self.config = config
self.args = args
# Init windows positions
self.term_w = 80
self.term_h = 24
# Space between stats
self.space_between_column = 3
self.space_between_line = 2
# Init the curses screen
self.screen = curses.initscr()
if not self.screen:
logger.critical("Cannot init the curses library.\n")
sys.exit(1)
# Load the 'outputs' section of the configuration file
# - Init the theme (default is black)
self.theme = {'name': 'black'}
# Load configuration file
self.load_config(config)
# Init cursor
self._init_cursor()
# Init the colors
self._init_colors()
# Init main window
def init(self):
"""Init the connection to the InfluxDB server."""
if not self.export_enable:
return None
# Cluster
try:
cluster = Cluster([self.host],
port=int(self.port),
protocol_version=int(self.protocol_version))
session = cluster.connect()
except Exception as e:
logger.critical("Cannot connect to Cassandra cluster '%s:%s' (%s)" % (self.host, self.port, e))
sys.exit(2)
# Keyspace
try:
session.set_keyspace(self.keyspace)
except InvalidRequest as e:
logger.info("Create keyspace {} on the Cassandra cluster".format(self.keyspace))
c = "CREATE KEYSPACE %s WITH replication = { 'class': 'SimpleStrategy', 'replication_factor': '%s' }" % (self.keyspace, self.replication_factor)
session.execute(c)
session.set_keyspace(self.keyspace)
logger.info(
"Stats will be exported to Cassandra cluster {} ({}) in keyspace {}".format(
cluster.metadata.cluster_name, cluster.metadata.all_hosts(), self.keyspace))
# Table
def update(self):
"""Update stats from Glances/SNMP server."""
if self.client_mode == 'glances':
return self.update_glances()
elif self.client_mode == 'snmp':
return self.update_snmp()
else:
self.end()
logger.critical("Unknown server mode: {}".format(self.client_mode))
sys.exit(2)
# Should be set to True to generate graphs
args.generate_graph = False
# Control parameter and exit if it is not OK
self.args = args
# Export is only available in standalone or client mode (issue #614)
export_tag = self.args.export is not None and any(self.args.export)
if WINDOWS and export_tag:
# On Windows, export is possible but only in quiet mode
# See issue #1038
logger.info("On Windows OS, export disable the Web interface")
self.args.quiet = True
self.args.webserver = False
elif not (self.is_standalone() or self.is_client()) and export_tag:
logger.critical("Export is only available in standalone or client mode")
sys.exit(2)
# Filter is only available in standalone mode
if args.process_filter is not None and not self.is_standalone():
logger.critical(
"Process filter is only available in standalone mode")
sys.exit(2)
# Disable HDDTemp if sensors are disabled
if getattr(args, 'disable_sensors', False):
disable(args, 'hddtemp')
logger.debug("Sensors and HDDTemp are disabled")
return args
# The key is the plugin name
# for example, the file glances_xxx.py
# generate self._plugins_list["xxx"] = ...
name = plugin_script[len(self.header):-3].lower()
try:
# Import the plugin
plugin = __import__(plugin_script[:-3])
# Init and add the plugin to the dictionary
if name in ('help', 'amps', 'ports'):
self._plugins[name] = plugin.Plugin(args=args, config=config)
else:
self._plugins[name] = plugin.Plugin(args=args)
except Exception as e:
# If a plugin can not be log, display a critical message
# on the console but do not crash
logger.critical("Error while initializing the {} plugin ({})".format(name, e))
logger.error(traceback.format_exc())
"""Load the plugin (script), init it and add to the _plugin dict."""
# The key is the plugin name
# for example, the file glances_xxx.py
# generate self._plugins_list["xxx"] = ...
name = plugin_script[len(self.header):-3].lower()
# Loaf the plugin class
try:
# Import the plugin
plugin = __import__(plugin_script[:-3])
# Init and add the plugin to the dictionary
self._plugins[name] = plugin.Plugin(args=args, config=config)
except Exception as e:
# If a plugin can not be loaded, display a critical message
# on the console but do not crash
logger.critical("Error while initializing the {} plugin ({})".format(name, e))
logger.error(traceback.format_exc())
# Disable the plugin
if args is not None:
setattr(args,
'disable_' + name,
False)
else:
# Set the disable_ to False by default
if args is not None:
setattr(args,
'disable_' + name,
getattr(args, 'disable_' + name, False))