logfile: Improve source code

Author: markuslf

check-plugins/logfile/logfile

@@ -26,9 +26,10 @@ from lib.globals import (STATE_CRIT, STATE_OK,  # pylint: disable=C0413
                           STATE_UNKNOWN, STATE_WARN)
 
 __author__ = 'Linuxfabrik GmbH, Zurich/Switzerland'
-__version__ = '2023071203'
+__version__ = '2023112801'
 
-DESCRIPTION = "Scans a logfile for set of pattern or regex and alarms on the number of findings."
+DESCRIPTION = """Scans a logfile for a set of patterns or regex and alerts
+                 on the number of matches."""
 
 DEFAULT_ALARM_DURATION = 60 # minutes (1 hour)
 DEFAULT_CRIT = 1
@@ -51,7 +52,9 @@ def parse_args():
 
     parser.add_argument(
         '--alarm-duration',
-        help='How long should this check return an alarm on new matches (in minutes)? This is overwritten by --icinga-callback. Default: %(default)s',
+        help='How long should this check return an alert on new matches (in minutes)? '
+             'This is overwritten by --icinga-callback. '
+             'Default: %(default)s',
         dest='ALARM_DURATION',
         type=int,
         default=DEFAULT_ALARM_DURATION,
@@ -67,7 +70,8 @@ def parse_args():
 
     parser.add_argument(
         '-c', '--critical',
-        help='Set the critical threshold for the number of found critical matches. Default: %(default)s',
+        help='Set the critical threshold for the number of found critical matches. '
+             'Default: %(default)s',
         dest='CRIT',
         default=DEFAULT_CRIT,
     )
@@ -82,23 +86,24 @@ def parse_args():
 
     parser.add_argument(
         '--critical-regex',
-        help='Any line matching this python regex will count as a critical.',
+        help='Any line matching this Python regex will count as a critical.',
         action='append',
         dest='CRIT_REGEX',
         default=[],
     )
 
     parser.add_argument(
         '--filename',
-        help='Set the path of the logfile.',
+        help='Set the path to the logfile.',
         dest='FILENAME',
         required=True,
         type=str,
     )
 
     parser.add_argument(
         '--icinga-callback',
-        help='Get the service acknowledgement from Icinga. This overwrites --alarm-duration. Default: %(default)s',
+        help='Get the service acknowledgement from Icinga. This overwrites `--alarm-duration`. '
+             'Default: %(default)s',
         dest='ICINGA_CALLBACK',
         action='store_true',
         default=DEFAULT_ICINGA_CALLBACK,
@@ -112,7 +117,9 @@ def parse_args():
 
     parser.add_argument(
         '--icinga-service-name',
-        help='Unique name of the service using this check within Icinga. Take it from the `__name` service attribute, for example `icinga-server!my-service-name`.',
+        help='Unique name of the service using this check within Icinga. '
+             'Take it from the `__name` service attribute, for example '
+             '`icinga-server!my-service-name`.',
         dest='ICINGA_SERVICE_NAME',
     )
 
@@ -138,7 +145,7 @@ def parse_args():
 
     parser.add_argument(
         '--ignore-regex',
-        help='Any line matching this python regex will be ignored.',
+        help='Any line matching this Python regex will be ignored.',
         action='append',
         default=[],
         dest='IGNORE_REGEX',
@@ -154,7 +161,8 @@ def parse_args():
 
     parser.add_argument(
         '-w', '--warning',
-        help='Set the warning threshold for the number of found warning matches. Default: %(default)s',
+        help='Set the warning threshold for the number of found warning matches. '
+             'Default: %(default)s',
         dest='WARN',
         default=DEFAULT_WARN,
     )
@@ -169,7 +177,7 @@ def parse_args():
 
     parser.add_argument(
         '--warning-regex',
-        help='Any line matching this python regex will count as a warning.',
+        help='Any line matching this Python regex will count as a warning.',
         action='append',
         dest='WARN_REGEX',
         default=[],
@@ -185,17 +193,26 @@ def main():
     except SystemExit:
         sys.exit(STATE_UNKNOWN)
 
+    try:
+        file_stat = os.stat(args.FILENAME)
+    except FileNotFoundError as e:
+        # no traceback wanted, so using oao() instead of cu()
+        lib.base.oao('{}: "{}"'.format(e.strerror, args.FILENAME), STATE_UNKNOWN)
+
     if not any((args.WARN_PATTERN, args.WARN_REGEX, args.CRIT_PATTERN, args.CRIT_REGEX)):
         lib.base.cu('At least one pattern or regex is required.')
 
-    if args.ICINGA_CALLBACK and not all((args.ICINGA_URL, args.ICINGA_PASSWORD, args.ICINGA_USERNAME, args.ICINGA_SERVICE_NAME)):
-        lib.base.cu('--icinga-callback requires --icinga-url, --icinga-password, --icinga-username and --icinga-service-name.')
+    if args.ICINGA_CALLBACK \
+    and not all((args.ICINGA_URL, args.ICINGA_PASSWORD, args.ICINGA_USERNAME, args.ICINGA_SERVICE_NAME)): # pylint: disable=C0301
+        lib.base.cu('`--icinga-callback` requires `--icinga-url`, `--icinga-password`, `--icinga-username` and `--icinga-service-name`.') # pylint: disable=C0301
 
     # create the db table
     conn = lib.base.coe(
-        lib.db_sqlite.connect(filename='linuxfabrik-monitoring-plugins-logfile-{}.db'.format(
-            os.path.basename(args.FILENAME)
-        ))
+        lib.db_sqlite.connect(
+            filename='linuxfabrik-monitoring-plugins-logfile-{}.db'.format(
+                os.path.basename(args.FILENAME),
+            )
+        )
     )
     definition = '''
         filename TEXT NOT NULL PRIMARY KEY,
@@ -211,7 +228,6 @@ def main():
     '''
     lib.base.coe(lib.db_sqlite.create_table(conn, definition, table='matching_lines'))
 
-    file_stat = os.stat(args.FILENAME)
     current_inode = file_stat.st_ino
     current_size = file_stat.st_size
 
@@ -229,8 +245,7 @@ def main():
     if old_file_stats:
         # this is not the first time we are scanning this logfile, try to continue where we left off
         offset = old_file_stats.get('offset', 0)
-
-        if (current_inode != old_file_stats.get('inode')
+        if (current_inode != old_file_stats.get('inode') \
         or current_size < offset):
             # this means the file has been rotated
             offset = 0
@@ -249,25 +264,21 @@ def main():
             logfile.seek(offset)
             for line in logfile:
                 line_counter += 1
-                # due to lazy evaluation, the regex will only be executed if the pattern does not match
+                # due to lazy evaluation, the regex will only be executed
+                # if the pattern does not match
                 # see https://docs.python.org/3/reference/expressions.html#boolean-operations
-                if any(warn_pattern in line for warn_pattern in args.WARN_PATTERN)\
+                if any(warn_pattern in line for warn_pattern in args.WARN_PATTERN) \
                 or any(item.search(line) for item in compiled_warn_regex):
-                    if not any(ignore_pattern in line for ignore_pattern in args.IGNORE_PATTERN)\
+                    if not any(ignore_pattern in line for ignore_pattern in args.IGNORE_PATTERN) \
                     and not any(item.search(line) for item in compiled_ignore_regex):
                         warn_matches.append(line.strip())
 
-                if any(crit_pattern in line for crit_pattern in args.CRIT_PATTERN)\
+                if any(crit_pattern in line for crit_pattern in args.CRIT_PATTERN) \
                 or any(item.search(line) for item in compiled_crit_regex):
-                    if not any(ignore_pattern in line for ignore_pattern in args.IGNORE_PATTERN)\
+                    if not any(ignore_pattern in line for ignore_pattern in args.IGNORE_PATTERN) \
                     and not any(item.search(line) for item in compiled_ignore_regex):
                         crit_matches.append(line.strip())
-
             offset = logfile.tell()
-    except FileNotFoundError:
-        lib.base.cu("Could not find logfile '{}'.".format(args.FILENAME))
-        # make sure conn is always closed
-        lib.db_sqlite.close(conn)
     except PermissionError:
         lib.base.cu("Permission denied opening '{}'.".format(args.FILENAME))
         # make sure conn is always closed
@@ -281,7 +292,8 @@ def main():
     }
     lib.base.coe(lib.db_sqlite.replace(conn, new_file_stats, table='file_stats'))
 
-    # if we are using the alarm duration (aka not using the icinga callback), remove all outdated matches from the db now
+    # if we are using the alarm duration (aka not using the icinga callback),
+    # remove all outdated matches from the db now
     now = lib.time.now(as_type='datetime')
     if not args.ICINGA_CALLBACK:
         outdated = now - datetime.timedelta(minutes=args.ALARM_DURATION)
@@ -296,7 +308,8 @@ def main():
         ))
 
     # get the old matches and take them into consideration for the state
-    # for example, if there are no new lines we still want to alarm the old ones until the service is acknowledged
+    # for example, if there are no new lines we still want to alarm the old ones
+    # until the service is acknowledged
     old_warn_matches = lib.base.coe(lib.db_sqlite.select(
         conn,
         '''
@@ -352,9 +365,9 @@ def main():
                 else:
                     msg_addendum += 'Note: Acknowledge this service to reset the state to OK.'
             except IndexError:
-                msg_addendum += 'Note: Could not determine the acknowledgement from the Icinga API, this could be due to an incorrect service name.'
+                msg_addendum += 'Note: Could not determine the acknowledgement from the Icinga API, this could be due to an incorrect service name.' # pylint: disable=C0301
         else:
-            msg_addendum += 'Note: Could not determine the acknowledgement from the Icinga API:\n{}.'.format(icinga)
+            msg_addendum += 'Note: Could not determine the acknowledgement from the Icinga API:\n{}.'.format(icinga) # pylint: disable=C0301
 
     # save to db, these lines will be alarmed until the service is acknowledged in icinga2
     for match in warn_matches:
@@ -416,14 +429,14 @@ def main():
             msg += '\n\nCritical matches:\n* ' + '\n* '.join(crit_matches)
 
         if old_warn_matches:
-            msg += '\n\nOld warning matches:\n* ' + '\n* '.join([match['line'] for match in old_warn_matches])
+            msg += '\n\nOld warning matches:\n* ' + '\n* '.join([match['line'] for match in old_warn_matches]) # pylint: disable=C0301
 
         if old_crit_matches:
-            msg += '\n\nOld critical matches:\n* ' + '\n* '.join([match['line'] for match in old_crit_matches])
+            msg += '\n\nOld critical matches:\n* ' + '\n* '.join([match['line'] for match in old_crit_matches]) # pylint: disable=C0301
 
     perfdata = lib.base.get_perfdata('scanned_lines', line_counter, None, None, None, None, None)
-    perfdata += lib.base.get_perfdata('warn_matches', len(warn_matches), None, args.WARN, None, None, None)
-    perfdata += lib.base.get_perfdata('crit_matches', len(crit_matches), None, args.WARN, None, None, None)
+    perfdata += lib.base.get_perfdata('warn_matches', len(warn_matches), None, args.WARN, None, None, None) # pylint: disable=C0301
+    perfdata += lib.base.get_perfdata('crit_matches', len(crit_matches), None, args.WARN, None, None, None) # pylint: disable=C0301
 
     lib.base.oao(msg + '\n\n' + msg_addendum, state, perfdata, always_ok=args.ALWAYS_OK)