honeycomb.utils package

Submodules

honeycomb.utils.config_utils module

Honeycomb Config Utilities.

honeycomb.utils.config_utils.config_field_type(field, cls)

Validate a config field against a type.

Similar functionality to validate_field_matches_type() but returns honeycomb.defs.ConfigField

honeycomb.utils.config_utils.get_config_parameters(plugin_path)

Return the parameters section from config.json.

honeycomb.utils.config_utils.get_truetype(value)

Convert a string to a pythonized parameter.

honeycomb.utils.config_utils.is_valid_field_name(value)

Ensure field name is valid.

honeycomb.utils.config_utils.process_config(ctx, configfile)

Process a yaml config with instructions.

This is a heavy method that loads lots of content, so we only run the imports if its called.

honeycomb.utils.config_utils.validate_config(config_json, fields)

Validate a JSON file configuration against list of honeycomb.defs.ConfigField.

honeycomb.utils.config_utils.validate_config_parameters(config_json, allowed_keys, allowed_types)

Validate parameters in config file.

honeycomb.utils.config_utils.validate_field(field, allowed_keys, allowed_types)

Validate field is allowed and valid.

honeycomb.utils.config_utils.validate_field_matches_type(field, value, field_type, select_items=None, _min=None, _max=None)

Validate a config field against a specific type.

honeycomb.utils.daemon module

Honeycomb DaemonRunner utility.

class honeycomb.utils.daemon.myRunner(app, pidfile=None, stdout=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='UTF-8'>, stderr=<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'>, stdin=<_io.TextIOWrapper name='/dev/null' mode='rt' encoding='UTF-8'>)

Bases: daemon.runner.DaemonRunner

Overriding default runner behaviour to be simpler.

Override init to fit honeycomb needs.

We initialize app with default stdout/stderr from sys instead of file path and remove the use of parse_args() since it’s not actually a standalone runner

honeycomb.utils.plugin_utils module

Honeycomb generic plugin install utils.

exception honeycomb.utils.plugin_utils.CTError(errors)

Bases: Exception

Copytree exception class, used to collect errors from the recursive copy_tree function.

Collect errors.

Parameters:errors – Collected errors

honeycomb.utils.plugin_utils.copy_file(src, dst)

Copy a single file.

Parameters:

• src – Source name
• dst – Destination name

honeycomb.utils.plugin_utils.copy_tree(src, dst, symlinks=False, ignore=[])

Copy a full directory structure.

Parameters:

• src – Source path
• dst – Destination path
• symlinks – Copy symlinks
• ignore – Subdirs/filenames to ignore

honeycomb.utils.plugin_utils.get_plugin_path(home, plugin_type, plugin_name, editable=False)

Return path to plugin.

Parameters:

• home – Path to honeycomb home
• plugin_type – Type of plugin (honeycomb.defs.SERVICES pr honeycomb.defs.INTEGRATIONS)
• plugin_name – Name of plugin
• editable – Use plugin_name as direct path instead of loading from honeycomb home folder

honeycomb.utils.plugin_utils.get_select_items(items)

Return list of possible select items.

honeycomb.utils.plugin_utils.install_deps(pkgpath)

Install plugin dependencies using pip.

We import pip here to reduce load time for when its not needed.

honeycomb.utils.plugin_utils.install_dir(pkgpath, install_path, register_func, delete_after_install=False)

Install plugin from specified directory.

install_path and register_func are same as install_plugin(). :param delete_after_install: Delete pkgpath after install (used in install_from_zip()).

honeycomb.utils.plugin_utils.install_from_repo(pkgname, plugin_type, install_path, register_func)

Install plugin from online repo.

honeycomb.utils.plugin_utils.install_from_zip(pkgpath, install_path, register_func, delete_after_install=False)

Install plugin from zipfile.

honeycomb.utils.plugin_utils.install_plugin(pkgpath, plugin_type, install_path, register_func)

Install specified plugin.

Parameters:

• pkgpath – Name of plugin to be downloaded from online repo or path to plugin folder or zip file.
• install_path – Path where plugin will be installed.
• register_func – Method used to register and validate plugin.

honeycomb.utils.plugin_utils.list_local_plugins(plugin_type, plugins_path, plugin_details)

List local plugins with details.

honeycomb.utils.plugin_utils.list_remote_plugins(installed_plugins, plugin_type)

List remote plugins from online repo.

honeycomb.utils.plugin_utils.parse_plugin_args(command_args, config_args)

Parse command line arguments based on the plugin’s parameters config.

Parameters:

• command_args – Command line arguments as provided by the user in key=value format.
• config_args – Plugin parameters parsed from config.json.

Returns:

Validated dictionary of parameters that will be passed to plugin class

honeycomb.utils.plugin_utils.print_plugin_args(plugin_path)

Print plugin parameters table.

honeycomb.utils.plugin_utils.uninstall_plugin(pkgpath, force)

Uninstall a plugin.

Parameters:

• pkgpath – Path to package to uninstall (delete)
• force – Force uninstall without asking

honeycomb.utils.tailer module

Honeycomb service log tailer.

class honeycomb.utils.tailer.Tailer(name: str, filepath: str, color: str = '', nlines: int = 10, follow: bool = False, outfile=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='UTF-8'>, sleeptime: int = 0.5, show_name: bool = True, used_colors: list = [])

Bases: object

Colorized file tailer.

Print lines from a file prefixed with a colored name. Optionally continue to follow file.

follow_file()

Follow a file and send every new line to a callback.

print_log(line)

Print a line from a logfile.

print_named_log(line)

Print a line from a logfile prefixed with service name.

stop()

Stop follow.

honeycomb.utils.wait module

Honeycomb wait utilities.

exception honeycomb.utils.wait.TimeoutException

Bases: Exception

Exception to be raised on timeout.

honeycomb.utils.wait.search_json_log(filepath, key, value)

Search json log file for a key=value pair.

Parameters:

• filepath – Valid path to a json file
• key – key to match
• value – value to match

Returns:

First matching line in json log file, parsed by json.loads()

honeycomb.utils.wait.wait_until(func, check_return_value=True, total_timeout=60, interval=0.5, exc_list=None, error_message='', *args, **kwargs)

Run a command in a loop until desired result or timeout occurs.

Parameters:

• func – Function to call and wait for
• check_return_value (bool) – Examine return value
• total_timeout (int) – Wait timeout,
• interval (float) – Sleep interval between retries
• exc_list (list) – Acceptable exception list
• error_message (str) – Default error messages
• args – args to pass to func
• kwargs – lwargs to pass to fun

Module contents

Honeycomb Utils.