Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# AMP name (= module name without glances_)
if name is None:
self.amp_name = self.__class__.__module__[len('glances_'):]
else:
self.amp_name = name
# Init the args
self.args = args
# Init the configs
self.configs = {}
# A timer is needed to only update every refresh seconds
# Init to 0 in order to update the AMP on startup
self.timer = Timer(0)
def __init__(self,
config=None,
args=None):
# Init stats
self.stats = GlancesStatsServer(config=config, args=args)
# Initial update
self.stats.update()
# cached_time is the minimum time interval between stats updates
# i.e. XML/RPC calls will not retrieve updated info until the time
# since last update is passed (will retrieve old cached info instead)
self.timer = Timer(0)
self.cached_time = args.cached_time
def __init__(self, cached_time=1):
self.cpu_percent = 0
self.percpu_percent = []
# cached_time is the minimum time interval between stats updates
# since last update is passed (will retrieve old cached info instead)
self.timer_cpu = Timer(0)
self.timer_percpu = Timer(0)
self.cached_time = cached_time
def __init__(self, cache_timeout=60):
"""Init the class to collect stats about processes."""
# Add internals caches because PSUtil do not cache all the stats
# See: https://code.google.com/p/psutil/issues/detail?id=462
self.username_cache = {}
self.cmdline_cache = {}
# The internals caches will be cleaned each 'cache_timeout' seconds
self.cache_timeout = cache_timeout
self.cache_timer = Timer(self.cache_timeout)
# Init the io dict
# key = pid
# value = [ read_bytes_old, write_bytes_old ]
self.io_old = {}
# Wether or not to enable process tree
self._enable_tree = False
self.process_tree = None
# Init stats
self.auto_sort = True
self._sort_key = 'cpu_percent'
self.allprocesslist = []
self.processlist = []
self.processcount = {'total': 0, 'running': 0, 'sleeping': 0, 'thread': 0}
def __update__(self):
# Never update more than 1 time per cached_time
if self.timer.finished():
self.stats.update()
self.timer = Timer(self.args.cached_time)
if hasattr(cputimes, 'iowait'):
cpu['iowait'] = cputimes.iowait
if hasattr(cputimes, 'irq'):
cpu['irq'] = cputimes.irq
if hasattr(cputimes, 'softirq'):
cpu['softirq'] = cputimes.softirq
if hasattr(cputimes, 'steal'):
cpu['steal'] = cputimes.steal
if hasattr(cputimes, 'guest'):
cpu['guest'] = cputimes.guest
if hasattr(cputimes, 'guest_nice'):
cpu['guest_nice'] = cputimes.guest_nice
# Append new CPU to the list
self.percpu_percent.append(cpu)
# Reset timer for cache
self.timer_percpu = Timer(self.cached_time)
return self.percpu_percent
def __update__(self):
# Never update more than 1 time per cached_time
if self.timer.finished():
self.stats.update()
self.timer = Timer(self.cached_time)
def __init__(self, config=None, args=None):
# Init config
self.config = config
# Init args
self.args = args
# Init stats
# Will be updated within Bottle route
self.stats = None
# cached_time is the minimum time interval between stats updates
# i.e. HTTP/RESTful calls will not retrieve updated info until the time
# since last update is passed (will retrieve old cached info instead)
self.timer = Timer(0)
# Load configuration file
self.load_config(config)
# Set the bind URL
self.bind_url = 'http://{}:{}/'.format(self.args.bind_address,
self.args.port)
# Init Bottle
self._app = Bottle()
# Enable CORS (issue #479)
self._app.install(EnableCors())
# Password
if args.password != '':
self._app.install(auth_basic(self.check_auth))
# Define routes
"""Update the ports list."""
if self.input_method == 'local':
# Only refresh:
# * if there is not other scanning thread
# * every refresh seconds (define in the configuration file)
if self._thread is None:
thread_is_running = False
else:
thread_is_running = self._thread.isAlive()
if self.timer_ports.finished() and not thread_is_running:
# Run ports scanner
self._thread = ThreadScanner(self.stats)
self._thread.start()
# Restart timer
if len(self.stats) > 0:
self.timer_ports = Timer(self.stats[0]['refresh'])
else:
self.timer_ports = Timer(0)
else:
# Not available in SNMP mode
pass
return self.stats
def __init__(self, cache_timeout=60):
"""Init the class to collect stats about processes."""
# Add internals caches because psutil do not cache all the stats
# See: https://code.google.com/p/psutil/issues/detail?id=462
self.username_cache = {}
self.cmdline_cache = {}
# The internals caches will be cleaned each 'cache_timeout' seconds
self.cache_timeout = cache_timeout
self.cache_timer = Timer(self.cache_timeout)
# Init the io dict
# key = pid
# value = [ read_bytes_old, write_bytes_old ]
self.io_old = {}
# Init stats
self.auto_sort = True
self._sort_key = 'cpu_percent'
self.processlist = []
self.reset_processcount()
# Tag to enable/disable the processes stats (to reduce the Glances CPU consumption)
# Default is to enable the processes stats
self.disable_tag = False