Skip to content

Commit

Permalink
pylintrc: re-enable too-many-arguments
Browse files Browse the repository at this point in the history 
started using dataclass types for grouping long list of arguments.

Suppressed the warning for test functions.

Signed-off-by: Mustafa Kemal Gilor <[email protected]>
  • Loading branch information
@xmkg
xmkg committed Jul 12, 2024
1 parent 1a84705 commit bcb845d
Show file tree
Hide file tree
Showing 17 changed files with 448 additions and 242 deletions.
123 changes: 88 additions & 35 deletions hotsos/cli.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import sys
import threading
from importlib import metadata, resources
from dataclasses import dataclass, fields

import click
import distro
Expand All @@ -15,7 +16,7 @@
from hotsos.core.log import log, LoggingManager
from hotsos.client import (
HotSOSClient,
OutputManager,
OutputManagerParameters
)

SNAP_ERROR_MSG = """ERROR: hotsos is installed as a snap which only supports
Expand Down Expand Up @@ -116,7 +117,7 @@ def set_plugin_options(f):


def get_defs_path():
""" Get patch to HotSOS defs. """
""" Get path to HotSOS defs. """

# source
defs = os.path.join(get_hotsos_root(), 'defs')
Expand Down Expand Up @@ -188,7 +189,52 @@ def progress_spinner(show_spinner, path):
thread.join()


def main(): # pylint: disable=R0915
def filter_unexpected_fields(cls):
original_init = cls.__init__

def new_init(self, *args, **kwargs):
def_fields = {field.name for field in fields(cls)}
cleaned_kwargs = {
key: value for key, value in kwargs.items() if key in def_fields
}
original_init(self, *args, **cleaned_kwargs)

cls.__init__ = new_init
return cls


@filter_unexpected_fields
@dataclass
class CLIArgs:
"""Command line arguments."""

data_root: click.Path
version: bool
defs_path: str
templates_path: str
all_logs: bool
debug: bool
quiet: bool
save: bool
output_format: str
html_escape: bool
short: bool
very_short: bool
force: bool
event_tally_granularity: str
max_logrotate_depth: int
max_parallel_tasks: int
list_plugins: bool
machine_readable: bool
output_path: str
command_timeout: int
sos_unpack_dir: str
scenario: str
event: str


# pylint: disable=R0915
def main():
@click.command(name='hotsos')
@click.option('--event', default='',
help=('Filter a particular event name. Useful for '
Expand Down Expand Up @@ -248,7 +294,8 @@ def main(): # pylint: disable=R0915
help=('Apply html escaping to the output so that it is safe '
'to display in html.'))
@click.option('--format', '--output-format', 'output_format',
type=click.Choice(OutputManager.SUMMARY_FORMATS),
type=click.Choice(
OutputManagerParameters.SUPPORTED_SUMMARY_FORMATS),
default='yaml',
show_default=True,
help='Summary output format.')
Expand All @@ -275,11 +322,7 @@ def main(): # pylint: disable=R0915
help='Show the version.')
@set_plugin_options
@click.argument('data_root', required=False, type=click.Path(exists=True))
def cli(data_root, version, defs_path, templates_path, all_logs, debug,
quiet, save, output_format, html_escape, short, very_short,
force, event_tally_granularity, max_logrotate_depth,
max_parallel_tasks, list_plugins, machine_readable, output_path,
command_timeout, sos_unpack_dir, scenario, event, **kwargs):
def cli(*args, **kwargs):
"""
Run this tool on a host or against a sosreport to perform
analysis of specific applications and the host itself. A summary of
Expand All @@ -303,49 +346,53 @@ def cli(data_root, version, defs_path, templates_path, all_logs, debug,
host.
""" # noqa

arguments = CLIArgs(*args, **kwargs)

_version = get_version()
if version:
if arguments.version:
print(_version)
return

config = {'repo_info': get_repo_info(),
'force_mode': force,
'force_mode': arguments.force,
'hotsos_version': _version,
'command_timeout': command_timeout,
'use_all_logs': all_logs,
'plugin_yaml_defs': defs_path,
'templates_path': templates_path,
'event_tally_granularity': event_tally_granularity,
'max_logrotate_depth': max_logrotate_depth,
'max_parallel_tasks': max_parallel_tasks,
'machine_readable': machine_readable,
'debug_mode': debug,
'scenario_filter': scenario,
'event_filter': event}
'command_timeout': arguments.command_timeout,
'use_all_logs': arguments.all_logs,
'plugin_yaml_defs': arguments.defs_path,
'templates_path': arguments.templates_path,
'event_tally_granularity': arguments.event_tally_granularity,
'max_logrotate_depth': arguments.max_logrotate_depth,
'max_parallel_tasks': arguments.max_parallel_tasks,
'machine_readable': arguments.machine_readable,
'debug_mode': arguments.debug,
'scenario_filter': arguments.scenario,
'event_filter': arguments.event}
HotSOSConfig.set(**config)

with LoggingManager() as logmanager:
with DataRootManager(data_root,
sos_unpack_dir=sos_unpack_dir) as drm:
with DataRootManager(
arguments.data_root, sos_unpack_dir=arguments.sos_unpack_dir
) as drm:
HotSOSConfig.data_root = drm.data_root
if is_snap() and drm.data_root == '/':
print(SNAP_ERROR_MSG)
sys.exit(1)

if debug and quiet:
if arguments.debug and arguments.quiet:
sys.stderr.write('ERROR: cannot use both --debug and '
'--quiet\n')
return

# Set a name so that logs have this until real plugins are run.
log.name = 'hotsos.cli'

if list_plugins:
if arguments.list_plugins:
sys.stdout.write('\n'.join(plugintools.PLUGINS.keys()))
sys.stdout.write('\n')
return

with progress_spinner(not quiet and not debug, drm.name):
with progress_spinner(
not arguments.quiet and not arguments.debug, drm.name):
plugins = []
for k, v in kwargs.items():
if v is True:
Expand All @@ -371,21 +418,27 @@ def cli(data_root, version, defs_path, templates_path, all_logs, debug,

summary = client.summary

if save:
path = summary.save(drm.basename, html_escape=html_escape,
output_path=output_path)
if arguments.save:
path = summary.save(
drm.basename,
html_escape=arguments.html_escape,
output_path=arguments.output_path,
)
sys.stdout.write(f"INFO: output saved to {path}\n")
else:
if short:
if arguments.short:
minimal_mode = 'short'
elif very_short:
elif arguments.very_short:
minimal_mode = 'very-short'
else:
minimal_mode = None

out = summary.get(fmt=output_format,
html_escape=html_escape,
minimal_mode=minimal_mode)
params = OutputManagerParameters(
fmt=arguments.output_format,
html_escape=arguments.html_escape,
minimal_mode=minimal_mode
)
out = summary.get(params)
if out:
sys.stdout.write(f"{out}\n")

Expand Down
78 changes: 48 additions & 30 deletions hotsos/client.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
import os
import shutil
import tempfile
from dataclasses import dataclass
from typing import Literal

# load all plugins
import hotsos.plugin_extensions # noqa: F401, pylint: disable=W0611
Expand All @@ -13,7 +15,6 @@
from hotsos.core.issues import IssuesManager
from hotsos.core.log import log
from hotsos.core import plugintools
from hotsos.core.exceptions import UnsupportedFormatError


class HotSOSSummary(plugintools.PluginPartBase):
Expand All @@ -39,11 +40,23 @@ def summary(self):
return out


@dataclass
class OutputManagerParameters:
"""Parameters for OutputManager."""

SUPPORTED_SUMMARY_FORMATS = ['yaml', 'json', 'markdown', 'html']
fmt: Literal[SUPPORTED_SUMMARY_FORMATS] = "yaml"
html_escape: bool = False
minimal_mode: str = None
plugin: str = None
max_level: int = 2


class OutputManager():
""" Handle conversion of plugin output into summary format. """

FILTER_SCHEMA = [IssuesManager.SUMMARY_OUT_ISSUES_ROOT,
IssuesManager.SUMMARY_OUT_BUGS_ROOT]
SUMMARY_FORMATS = ['yaml', 'json', 'markdown', 'html']

def __init__(self, initial=None):
self._summary = initial or {}
Expand Down Expand Up @@ -116,46 +129,41 @@ def minimise(self, summary, mode):
log.warning("Unknown minimalmode '%s'", mode)
return summary

def get(self, fmt='yaml', html_escape=False, minimal_mode=None,
plugin=None, max_level=2):
if plugin:
filtered = {plugin: self._summary[plugin]}
def get(self, params: OutputManagerParameters = OutputManagerParameters()):
if params.plugin:
filtered = {params.plugin: self._summary[params.plugin]}
else:
filtered = self._summary

if minimal_mode:
filtered = self.minimise(filtered, minimal_mode)

if fmt not in self.SUMMARY_FORMATS:
raise UnsupportedFormatError(
f"unsupported summary format '{fmt}'")
if params.minimal_mode:
filtered = self.minimise(filtered, params.minimal_mode)

hostname = CLIHelper().hostname() or ""
log.debug('Saving summary as %s', fmt)
if fmt == 'yaml':
log.debug('Saving summary as %s', params.fmt)
if params.fmt == 'yaml':
filtered = plugintools.yaml_dump(filtered)
elif fmt == 'json':
elif params.fmt == 'json':
filtered = json.dumps(filtered, indent=2, sort_keys=True)
elif fmt == 'markdown':
elif params.fmt == 'markdown':
filtered = plugintools.MarkdownFormatter().dump(filtered)
elif fmt == 'html':
elif params.fmt == 'html':
filtered = plugintools.HTMLFormatter(
hostname=hostname,
max_level=max_level).dump(filtered)
max_level=params.max_level).dump(
filtered)

if html_escape:
if params.html_escape:
log.debug('Applying html escaping to summary')
filtered = html.escape(filtered)

return filtered

def _save(self, path, fmt, html_escape=None, minimal_mode=None,
plugin=None):
content = self.get(fmt=fmt, html_escape=html_escape,
minimal_mode=minimal_mode, plugin=plugin)
with open(path, 'w', encoding='utf-8') as fd:
def _save(self, path,
params: OutputManagerParameters = OutputManagerParameters()):
content = self.get(params)
with open(path, "w", encoding="utf-8") as fd:
fd.write(content)
fd.write('\n')
fd.write("\n")

def save(self, name, html_escape=False, output_path=None):
"""
Expand All @@ -171,7 +179,7 @@ def save(self, name, html_escape=False, output_path=None):

for minimal_mode in ['full', 'short', 'very-short']:
_minimal_mode = minimal_mode.replace('-', '_')
for fmt in self.SUMMARY_FORMATS:
for fmt in OutputManagerParameters.SUPPORTED_SUMMARY_FORMATS:
output_path = os.path.join(output_root, name, 'summary',
_minimal_mode, fmt)
if minimal_mode == 'full':
Expand All @@ -183,12 +191,22 @@ def save(self, name, html_escape=False, output_path=None):
for plugin in self._summary:
path = os.path.join(output_path,
f"hotsos-summary.{plugin}.{fmt}")
self._save(path, fmt, html_escape=html_escape,
minimal_mode=minimal_mode, plugin=plugin)
self._save(
path,
OutputManagerParameters(
fmt=fmt,
html_escape=html_escape,
minimal_mode=minimal_mode,
plugin=plugin,
),
)

path = os.path.join(output_path, f"hotsos-summary.all.{fmt}")
self._save(path, fmt, html_escape=html_escape,
minimal_mode=minimal_mode)
self._save(path, OutputManagerParameters(
fmt=fmt,
html_escape=html_escape,
minimal_mode=minimal_mode
))

if not minimal_mode:
dst = os.path.join(output_root, f'{name}.summary.{fmt}')
Expand Down
26 changes: 13 additions & 13 deletions hotsos/core/analytics.py
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import statistics
from datetime import datetime
from dataclasses import dataclass


class EventCollection():
Expand Down Expand Up @@ -151,24 +152,23 @@ def calculate_event_deltas(self):
start_item["end"] = end_ts


@dataclass
class SearchResultIndices():
"""
Used to know where to find required information within a SearchResult.
The indexes refer to python.re groups.
The minimum required information that a result must contain is day,
secs and event_id. Results will be referred to using whatever event_id
is set to.
"""
def __init__(self, day_idx=1, secs_idx=2, event_id_idx=3,
metadata_idx=None, metadata_key=None):
"""
The indexes refer to python.re groups.

The minimum required information that a result must contain is day,
secs and event_id. Results will be referred to using whatever event_id
is set to.
"""
self.day = day_idx
self.secs = secs_idx
self.event_id = event_id_idx
self.metadata = metadata_idx
self.metadata_key = metadata_key
day: int = 1
secs: int = 2
event_id: int = 3
metadata: int = None
metadata_key: str = None


class LogEventStats():
Expand Down
Loading

0 comments on commit bcb845d

Please sign in to comment.