Skip to content

log.py

This module provides log functionalities.

__all__ = ['log_enter_exit', 'LogException', 'Logs'] module-attribute

log_authentication_error = partial(_base_error_log, exe_label='Authentication Error') module-attribute

log_configuration_error = partial(_base_error_log, exe_label='Configuration Error') module-attribute

log_connection_error = partial(_base_error_log, exe_label='Connection Error') module-attribute

log_permission_error = partial(_base_error_log, exe_label='Permission Error') module-attribute

log_server_error = partial(_base_error_log, exe_label='Server Error') module-attribute

LogException

Bases: Exception

Exception raised by Logs class.

Source code in solnlib/log.py 
60
61
62
63
class LogException(Exception):
    """Exception raised by Logs class."""

    pass

Logs

A singleton class that manage all kinds of logger.

Examples:

>>> from solnlib import log
>>> log.Logs.set_context(directory='/var/log/test',
                         namespace='test')
>>> logger = log.Logs().get_logger('mymodule')
>>> logger.set_level(logging.DEBUG)
>>> logger.debug('a debug log')
Source code in solnlib/log.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
class Logs(metaclass=Singleton):
    """A singleton class that manage all kinds of logger.

    Examples:
      >>> from solnlib import log
      >>> log.Logs.set_context(directory='/var/log/test',
                               namespace='test')
      >>> logger = log.Logs().get_logger('mymodule')
      >>> logger.set_level(logging.DEBUG)
      >>> logger.debug('a debug log')
    """

    # Normal logger settings
    _default_directory = None
    _default_namespace = None
    _default_log_format = (
        "%(asctime)s log_level=%(levelname)s pid=%(process)d tid=%(threadName)s "
        "file=%(filename)s:%(funcName)s:%(lineno)d | %(message)s"
    )
    _default_log_level = logging.INFO
    _default_max_bytes = 25000000
    _default_backup_count = 5

    # Default root logger settings
    _default_root_logger_log_file = "solnlib"

    @classmethod
    def set_context(cls, **context: dict):
        """Set log context.

        List of keyword arguments:

            directory: Log directory, default is splunk log root directory.
            namespace: Logger namespace, default is None.
            log_format: Log format, default is `_default_log_format`.
            log_level: Log level, default is logging.INFO.
            max_bytes: The maximum log file size before rollover, default is 25000000.
            backup_count: The number of log files to retain,default is 5.
            root_logger_log_file: Root logger log file name, default is 'solnlib'   .

        Arguments:
            context: Keyword arguments. See list of arguments above.
        """
        if "directory" in context:
            cls._default_directory = context["directory"]
        if "namespace" in context:
            cls._default_namespace = context["namespace"]
        if "log_format" in context:
            cls._default_log_format = context["log_format"]
        if "log_level" in context:
            cls._default_log_level = context["log_level"]
        if "max_bytes" in context:
            cls._default_max_bytes = context["max_bytes"]
        if "backup_count" in context:
            cls._default_backup_count = context["backup_count"]
        if "root_logger_log_file" in context:
            cls._default_root_logger_log_file = context["root_logger_log_file"]
            cls._reset_root_logger()

    @classmethod
    def _reset_root_logger(cls):
        logger = logging.getLogger()
        log_file = cls._get_log_file(cls._default_root_logger_log_file)
        file_handler = logging.handlers.RotatingFileHandler(
            log_file,
            mode="a",
            maxBytes=cls._default_max_bytes,
            backupCount=cls._default_backup_count,
        )
        file_handler.setFormatter(logging.Formatter(cls._default_log_format))
        logger.addHandler(file_handler)
        logger.setLevel(cls._default_log_level)

    @classmethod
    def _get_log_file(cls, name):
        if cls._default_namespace:
            name = f"{cls._default_namespace}_{name}.log"
        else:
            name = f"{name}.log"

        if cls._default_directory:
            directory = cls._default_directory
        else:
            try:
                directory = make_splunkhome_path(["var", "log", "splunk"])
            except KeyError:
                raise LogException(
                    "Log directory is empty, please set log directory "
                    'by calling Logs.set_context(directory="/var/log/...").'
                )
        log_file = op.sep.join([directory, name])

        return log_file

    def __init__(self):
        self._lock = Lock()
        self._loggers = {}

    def get_logger(self, name: str) -> logging.Logger:
        """Get logger with the name of `name`.

        If logger with the name of `name` exists just return else create a new
        logger with the name of `name`.

        Arguments:
            name: Logger name, it will be used as log file name too.

        Returns:
            A named logger.
        """

        with self._lock:
            log_file = self._get_log_file(name)
            if log_file in self._loggers:
                return self._loggers[log_file]

            logger = logging.getLogger(log_file)
            handler_exists = any(
                [True for h in logger.handlers if h.baseFilename == log_file]
            )
            if not handler_exists:
                file_handler = logging.handlers.RotatingFileHandler(
                    log_file,
                    mode="a",
                    maxBytes=self._default_max_bytes,
                    backupCount=self._default_backup_count,
                )
                file_handler.setFormatter(logging.Formatter(self._default_log_format))
                logger.addHandler(file_handler)
                logger.setLevel(self._default_log_level)
                logger.propagate = False

            self._loggers[log_file] = logger
            return logger

    def set_level(self, level: int, name: str = None):
        """Set log level of logger.

        Set log level of all logger if `name` is None else of
        logger with the name of `name`.

        Arguments:
            level: Log level to set.
            name: The name of logger, default is None.
        """

        with self._lock:
            if name:
                log_file = self._get_log_file(name)
                logger = self._loggers.get(log_file)
                if logger:
                    logger.setLevel(level)
            else:
                self._default_log_level = level
                for logger in list(self._loggers.values()):
                    logger.setLevel(level)
                logging.getLogger().setLevel(level)

__init__()

Source code in solnlib/log.py 
160
161
162
def __init__(self):
    self._lock = Lock()
    self._loggers = {}

get_logger(name)

Get logger with the name of name.

If logger with the name of name exists just return else create a new logger with the name of name.

Parameters:

Name Type Description Default
name str

Logger name, it will be used as log file name too.

 required

Returns:

Type Description
logging.Logger

A named logger.
Source code in solnlib/log.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def get_logger(self, name: str) -> logging.Logger:
    """Get logger with the name of `name`.

    If logger with the name of `name` exists just return else create a new
    logger with the name of `name`.

    Arguments:
        name: Logger name, it will be used as log file name too.

    Returns:
        A named logger.
    """

    with self._lock:
        log_file = self._get_log_file(name)
        if log_file in self._loggers:
            return self._loggers[log_file]

        logger = logging.getLogger(log_file)
        handler_exists = any(
            [True for h in logger.handlers if h.baseFilename == log_file]
        )
        if not handler_exists:
            file_handler = logging.handlers.RotatingFileHandler(
                log_file,
                mode="a",
                maxBytes=self._default_max_bytes,
                backupCount=self._default_backup_count,
            )
            file_handler.setFormatter(logging.Formatter(self._default_log_format))
            logger.addHandler(file_handler)
            logger.setLevel(self._default_log_level)
            logger.propagate = False

        self._loggers[log_file] = logger
        return logger

set_context(**context) classmethod

Set log context.

List of keyword arguments

directory: Log directory, default is splunk log root directory. namespace: Logger namespace, default is None. log_format: Log format, default is _default_log_format. log_level: Log level, default is logging.INFO. max_bytes: The maximum log file size before rollover, default is 25000000. backup_count: The number of log files to retain,default is 5. root_logger_log_file: Root logger log file name, default is ‘solnlib’ .

Parameters:

Name Type Description Default
context dict

Keyword arguments. See list of arguments above.

 {}
Source code in solnlib/log.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
@classmethod
def set_context(cls, **context: dict):
    """Set log context.

    List of keyword arguments:

        directory: Log directory, default is splunk log root directory.
        namespace: Logger namespace, default is None.
        log_format: Log format, default is `_default_log_format`.
        log_level: Log level, default is logging.INFO.
        max_bytes: The maximum log file size before rollover, default is 25000000.
        backup_count: The number of log files to retain,default is 5.
        root_logger_log_file: Root logger log file name, default is 'solnlib'   .

    Arguments:
        context: Keyword arguments. See list of arguments above.
    """
    if "directory" in context:
        cls._default_directory = context["directory"]
    if "namespace" in context:
        cls._default_namespace = context["namespace"]
    if "log_format" in context:
        cls._default_log_format = context["log_format"]
    if "log_level" in context:
        cls._default_log_level = context["log_level"]
    if "max_bytes" in context:
        cls._default_max_bytes = context["max_bytes"]
    if "backup_count" in context:
        cls._default_backup_count = context["backup_count"]
    if "root_logger_log_file" in context:
        cls._default_root_logger_log_file = context["root_logger_log_file"]
        cls._reset_root_logger()

set_level(level, name=None)

Set log level of logger.

Set log level of all logger if name is None else of logger with the name of name.

Parameters:

Name Type Description Default
level int

Log level to set.

 required
name str

The name of logger, default is None.

 None
Source code in solnlib/log.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def set_level(self, level: int, name: str = None):
    """Set log level of logger.

    Set log level of all logger if `name` is None else of
    logger with the name of `name`.

    Arguments:
        level: Log level to set.
        name: The name of logger, default is None.
    """

    with self._lock:
        if name:
            log_file = self._get_log_file(name)
            logger = self._loggers.get(log_file)
            if logger:
                logger.setLevel(level)
        else:
            self._default_log_level = level
            for logger in list(self._loggers.values()):
                logger.setLevel(level)
            logging.getLogger().setLevel(level)

events_ingested(logger, modular_input_name, sourcetype, n_events, index, account=None, host=None, license_usage_source=None)

Specific function to log the basic information of events ingested for the monitoring dashboard.

Parameters:

Name Type Description Default
logger logging.Logger

Add-on logger.

 required
modular_input_name str

Full name of the modular input. It needs to be in a format ://.

 required
sourcetype str

Source type used to write event.

 required
n_events int

Number of ingested events.

 required
index str

Index used to write event.

 required
license_usage_source str

source used to match data with license_usage.log.

 None
account str

Account used to write event. (optional)

 None
host str

Host used to write event. (optional)

 None
Source code in solnlib/log.py 
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
def events_ingested(
    logger: logging.Logger,
    modular_input_name: str,
    sourcetype: str,
    n_events: int,
    index: str,
    account: str = None,
    host: str = None,
    license_usage_source: str = None,
):
    """Specific function to log the basic information of events ingested for
    the monitoring dashboard.

    Arguments:
        logger: Add-on logger.
        modular_input_name: Full name of the modular input. It needs to be in a format <input_type>://<input_name>.
        In case of invalid format ValueError is raised.
        sourcetype: Source type used to write event.
        n_events: Number of ingested events.
        index: Index used to write event.
        license_usage_source: source used to match data with license_usage.log.
        account: Account used to write event. (optional)
        host: Host used to write event. (optional)
    """

    if "://" in modular_input_name:
        input_name = modular_input_name.split("/")[-1]
    else:
        raise ValueError(
            f"Invalid modular input name: {modular_input_name}. "
            f"It should be in format <input_type>://<input_name>"
        )

    result = {
        "action": "events_ingested",
        "modular_input_name": license_usage_source
        if license_usage_source
        else modular_input_name,
        "sourcetype_ingested": sourcetype,
        "n_events": n_events,
        "event_input": input_name,
        "event_index": index,
    }

    if account:
        result["event_account"] = account

    if host:
        result["event_host"] = host

    log_event(logger, result)

log_enter_exit(logger)

Decorator for logger to log function enter and exit.

This decorator will generate a lot of debug log, please add this only when it is required.

Parameters:

Name Type Description Default
logger logging.Logger

Logger to decorate.

 required

Examples:

>>> @log_enter_exit
>>> def myfunc():
>>>     doSomething()
Source code in solnlib/log.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def log_enter_exit(logger: logging.Logger):
    """Decorator for logger to log function enter and exit.

    This decorator will generate a lot of debug log, please add this
    only when it is required.

    Arguments:
        logger: Logger to decorate.

    Examples:
        >>> @log_enter_exit
        >>> def myfunc():
        >>>     doSomething()
    """

    def log_decorator(func):
        def wrapper(*args, **kwargs):
            logger.debug("%s entered", func.__name__)
            result = func(*args, **kwargs)
            logger.debug("%s exited", func.__name__)
            return result

        return wrapper

    return log_decorator

log_event(logger, key_values, log_level=logging.INFO)

General function to log any event in key-value format.

Source code in solnlib/log.py 
225
226
227
228
229
230
def log_event(
    logger: logging.Logger, key_values: Dict[str, Any], log_level: int = logging.INFO
):
    """General function to log any event in key-value format."""
    message = " ".join([f"{k}={v}" for k, v in key_values.items()])
    logger.log(log_level, message)

log_exception(logger, e, exc_label, full_msg=True, msg_before=None, msg_after=None, log_level=logging.ERROR)

General function to log exceptions.

Parameters:

Name Type Description Default
logger logging.Logger

Add-on logger.

 required
e Exception

Exception to log.

 required
exc_label str

label for the error to categorize it.

 required
full_msg bool

if set to True, full traceback will be logged. Default: True

 True
msg_before str

custom message before exception traceback. Default: None

 None
msg_after str

custom message after exception traceback. Default: None

 None
log_level int

Log level to log exception. Default: ERROR.

 logging.ERROR
Source code in solnlib/log.py 
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
def log_exception(
    logger: logging.Logger,
    e: Exception,
    exc_label: str,
    full_msg: bool = True,
    msg_before: str = None,
    msg_after: str = None,
    log_level: int = logging.ERROR,
):
    """General function to log exceptions.

    Arguments:
        logger: Add-on logger.
        e: Exception to log.
        exc_label: label for the error to categorize it.
        full_msg: if set to True, full traceback will be logged. Default: True
        msg_before: custom message before exception traceback. Default: None
        msg_after: custom message after exception traceback. Default: None
        log_level: Log level to log exception. Default: ERROR.
    """

    msg = _get_exception_message(e, full_msg, msg_before, msg_after)
    logger.log(log_level, f'exc_l="{exc_label}" {msg}')

modular_input_end(logger, modular_input_name)

Specific function to log the end of the modular input.

Source code in solnlib/log.py 
244
245
246
247
248
249
250
251
252
def modular_input_end(logger: logging.Logger, modular_input_name: str):
    """Specific function to log the end of the modular input."""
    log_event(
        logger,
        {
            "action": "ended",
            "modular_input_name": modular_input_name,
        },
    )

modular_input_start(logger, modular_input_name)

Specific function to log the start of the modular input.

Source code in solnlib/log.py 
233
234
235
236
237
238
239
240
241
def modular_input_start(logger: logging.Logger, modular_input_name: str):
    """Specific function to log the start of the modular input."""
    log_event(
        logger,
        {
            "action": "started",
            "modular_input_name": modular_input_name,
        },
    )