Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
ret.append(self.curse_add_line(msg))
msg = '{:>7}'.format('W/s')
ret.append(self.curse_add_line(msg))
# Disk list (sorted by name)
for i in self.sorted_stats():
# Is there an alias for the disk name ?
disk_real_name = i['disk_name']
disk_name = self.has_alias(i['disk_name'])
if disk_name is None:
disk_name = disk_real_name
# New line
ret.append(self.curse_new_line())
if len(disk_name) > name_max_width:
# Cut disk name if it is too long
disk_name = '_' + disk_name[-name_max_width:]
msg = '{:{width}}'.format(nativestr(disk_name),
width=name_max_width)
ret.append(self.curse_add_line(msg))
if args.diskio_iops:
# count
txps = self.auto_unit(
int(i['read_count'] // i['time_since_update']))
rxps = self.auto_unit(
int(i['write_count'] // i['time_since_update']))
msg = '{:>7}'.format(txps)
ret.append(self.curse_add_line(msg,
self.get_views(item=i[self.get_key()],
key='read_count',
option='decoration')))
msg = '{:>7}'.format(rxps)
ret.append(self.curse_add_line(msg,
self.get_views(item=i[self.get_key()],
# Exit if no data
if data == "":
return
# Safety check to avoid malformed data
# Considering the size of "|/dev/sda||0||" as the minimum
if len(data) < 14:
data = self.cache if len(self.cache) > 0 else self.fetch()
self.cache = data
try:
fields = data.split(b'|')
except TypeError:
fields = ""
devices = (len(fields) - 1) // 5
for item in range(devices):
offset = item * 5
hddtemp_current = {}
device = os.path.basename(nativestr(fields[offset + 1]))
temperature = fields[offset + 3]
unit = nativestr(fields[offset + 4])
hddtemp_current['label'] = device
try:
hddtemp_current['value'] = float(temperature)
except ValueError:
# Temperature could be 'ERR', 'SLP' or 'UNK' (see issue #824)
# Improper bytes/unicode in glances_hddtemp.py (see issue #887)
hddtemp_current['value'] = nativestr(temperature)
hddtemp_current['unit'] = unit
self.hddtemp_list.append(hddtemp_current)
def exit(self):
"""Just log an event when Glances exit."""
logger.debug("Stop the {} plugin".format(self.plugin_name))
def exit(self):
"""Close the CSV file."""
logger.debug("Finalise export interface %s" % self.export_name)
self.csv_file.close()
def init(self):
"""Init the Prometheus Exporter"""
try:
start_http_server(port=int(self.port), addr=self.host)
except Exception as e:
logger.critical("Can not start Prometheus exporter on {}:{} ({})".format(self.host, self.port, e))
sys.exit(2)
else:
logger.info("Start Prometheus exporter on {}:{}".format(self.host, self.port))
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
# 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
# The users dict
# username / password couple
# By default, no auth is needed
self.server.user_dict = {}
self.server.isAuth = False
# Register functions
self.server.register_introspection_functions()
self.server.register_instance(GlancesInstance(config, args))
if not self.args.disable_autodiscover:
# Note: The Zeroconf service name will be based on the hostname
# Correct issue: Zeroconf problem with zeroconf service name #889
self.autodiscover_client = GlancesAutoDiscoverClient(socket.gethostname().split('.', 1)[0], args)
else:
logger.info("Glances autodiscover announce is disabled")