diff options
| -rw-r--r-- | README.md | 31 | ||||
| -rw-r--r-- | pylogsentinel/core.py | 193 | ||||
| -rw-r--r-- | pyproject.toml | 2 |
3 files changed, 140 insertions, 86 deletions
@@ -12,7 +12,7 @@ previous run (tracked per-file, per-inode) and then exits. ## Key Features - INI-style configuration (`pylogsentinel.conf`). -- Two log discovery modes: static paths or shell command +- Multiple named log sets (e.g. [logs.default], [logs.errors]) with per-set discovery via static paths or shell command - Per-rule regular expressions with optional flags (`/pattern/i` style). - Per-rule or actions executed as shell commands with rich contextual environment variables. - Reliable state tracking via inode-based files (`<inode>` in `state_dir`). @@ -65,9 +65,10 @@ The configuration is an INI-style file. An example: state_dir = /var/run/pylogsentinel max_block_size = 10M -[logs] -# Choose exactly one, `paths` or `cmd`: +[logs.default] paths = /var/log /custom/app/logs/app.log + +[logs.other] cmd = find /var/log -type f -name '*.log' [action.default] @@ -79,7 +80,8 @@ cmd = echo "Another action" [rule.error] description = Error-like conditions pattern = /(error|fatal|exception|kill|crash)/i -action = another # if omitted, use default action +action = another +logs = default other ``` ### `[system]` Section @@ -89,22 +91,23 @@ action = another # if omitted, use default action | `state_dir` | Yes | Directory storing lock + per-inode state files. Must be writable. (Defined under `[system]`.) | (none) | | `max_block_size` | No | Maximum newly appended bytes to read per file per run. Supports suffixes `K`, `M`, `G` (in `[system]`). | `10M` | -### `[logs]` Section +### Log Sets (`[logs.<id>]`) -Supply **either**: +Define one or more log sets; each `[logs.<id>]` specifies exactly one of `paths` or `cmd`. The default set must be named `[logs.default]`. A bare `[logs]` section is not allowed. -- `paths`: Space-separated list of file and/or directory paths. Directories are traversed recursively; only files whose `file -b` output contains the substring "text" are monitored (others are skipped). Explicit file paths are always processed. -- `cmd`: A shell command producing **one path per line** on stdout. +- `paths`: Space-separated file and/or directory paths. Directories are traversed recursively; only files whose `file -b` output contains 'text' are monitored. Explicit file paths are always processed. +- `cmd`: Shell command producing one path per line on stdout. -Exactly one must be present. If using `cmd`, ensure it returns relatively quickly (recommended under a second) as it executes on every run. +Rules may reference multiple log sets using a whitespace-separated `logs` field. ### `[rule.<rule_id>]` Sections -| Field | Required | Description | -| ------------- | -------- | -------------------------------------------------------------------------------------------------------------- | -| `pattern` | Yes | Regular expression in the mandatory form `/pattern/flags` (flags optional). Supported flags: `i` (IGNORECASE). | -| `description` | No | Human-readable description for environment variable `RULE_DESCRIPTION`. | -| `action` | No | The `action_id` to invoke. Defaults to `default` if omitted. | +| Field | Required | Description | +| ------------- | -------- | ---------------------------------------------------------------------------------------------------------- | +| `pattern` | Yes | Regular expression in `/pattern/flags` form. Supported flag: `i` (IGNORECASE). | +| `description` | No | Human-readable description for environment variable `RULE_DESCRIPTION`. | +| `action` | No | Action id to invoke; defaults to `default` if omitted. | +| `logs` | No | Whitespace-separated list of log set ids (e.g. `default errors access`). Defaults to `default` if omitted. | At least one rule is required. diff --git a/pylogsentinel/core.py b/pylogsentinel/core.py index cea0a5a..1b6ce13 100644 --- a/pylogsentinel/core.py +++ b/pylogsentinel/core.py @@ -35,8 +35,8 @@ import time DEFAULT_CONFIG_PATH = "pylogsentinel.conf" LOCK_FILENAME = "LOCK" -DEFAULT_MAX_BLOCK_SIZE = 10 * 1024 * 1024 # 10 MiB -CONTEXT_RADIUS = 2 # lines before and after a match to include in CONTEXT +DEFAULT_MAX_BLOCK_SIZE = 10 * 1024 * 1024 +CONTEXT_RADIUS = 5 class ConfigError(Exception): @@ -54,8 +54,10 @@ class Rule: rule_id: str pattern: str description: str - action_id: str # may reference "default" + action_id: str compiled: re.Pattern[str] + log_set_ids: list[str] + log_base_paths: list[str] @dataclass @@ -97,7 +99,7 @@ def parse_size(value: str) -> int: _FLAG_MAP = { - "i": re.IGNORECASE, # only supported flag + "i": re.IGNORECASE, } @@ -135,10 +137,10 @@ def load_config( path: str, ) -> tuple[list[str], dict[str, Rule], dict[str, Action], int, str]: """ - Load and validate configuration. + Load and validate configuration supporting multiple log sets. Returns: - (log_paths (may contain directories), + (all_log_paths (union of all log set base paths), rules mapping, actions mapping, max_block_size, @@ -162,50 +164,58 @@ def load_config( if max_block_size <= 0: raise ConfigError("max_block_size must be positive") - # Logs section - if not parser.has_section("logs"): - raise ConfigError("Missing [logs] section") - log_cmd = parser.get("logs", "cmd", fallback=None) - log_paths_raw = parser.get("logs", "paths", fallback=None) - if log_cmd and log_paths_raw: - raise ConfigError("Specify only one of [logs].cmd or [logs].paths") - if not log_cmd and not log_paths_raw: - raise ConfigError("Must specify one of [logs].cmd or [logs].paths") + log_sets: dict[str, list[str]] = {} - log_paths: list[str] - if log_paths_raw: - log_paths = [ - os.path.expanduser(os.path.expandvars(p)) - for p in log_paths_raw.split() - if p - ] - else: - # Run discovery command (log_cmd validated above) - assert log_cmd is not None and log_cmd.strip(), ( - "internal: log_cmd unexpectedly None/empty" + for section in parser.sections(): + if section.startswith("logs."): + log_set_id = section[len("logs.") :] + else: + continue + + cmd = parser.get(section, "cmd", fallback=None) + paths_raw = parser.get(section, "paths", fallback=None) + if cmd and paths_raw: + raise ConfigError(f"Specify only one of cmd or paths in [{section}]") + if not cmd and not paths_raw: + raise ConfigError(f"Must specify cmd or paths in [{section}]") + + if paths_raw: + paths = [ + os.path.expanduser(os.path.expandvars(p)) + for p in paths_raw.split() + if p + ] + else: + assert cmd is not None + try: + proc = subprocess.run( + cmd, + shell=True, + check=False, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + except OSError as e: + raise ConfigError( + f"Failed to run log discovery cmd in [{section}]: {e}" + ) from e + if proc.returncode != 0: + raise ConfigError( + f"Log discovery command failed ({proc.returncode}) in [{section}]: {cmd}\n{proc.stderr}" + ) + paths = [ + os.path.expanduser(os.path.expandvars(line.strip())) + for line in proc.stdout.splitlines() + if line.strip() + ] + log_sets[log_set_id] = paths + + if not log_sets: + raise ConfigError( + "No log sets defined (need at least one [logs.<id>] section, e.g. [logs.default])" ) - try: - proc = subprocess.run( - log_cmd, # type: ignore[arg-type] # narrowed by assert for type checkers - shell=True, - check=False, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - ) - except OSError as e: - raise ConfigError(f"Failed to run logs cmd: {e}") from e - if proc.returncode != 0: - raise ConfigError( - f"Log discovery command failed ({proc.returncode}): {log_cmd}\n{proc.stderr}" - ) - log_paths = [ - os.path.expanduser(os.path.expandvars(line.strip())) - for line in proc.stdout.splitlines() - if line.strip() - ] - # Parse actions actions: dict[str, Action] = {} for section in parser.sections(): if section.startswith("action."): @@ -218,7 +228,6 @@ def load_config( if "default" not in actions: raise ConfigError("Missing [action.default] section (mandatory default action)") - # Parse rules rules: dict[str, Rule] = {} for section in parser.sections(): if section.startswith("rule."): @@ -234,19 +243,48 @@ def load_config( raise ConfigError( f"Rule {rule_id!r} references unknown action {action_id!r}" ) + logs_field = parser.get(section, "logs", fallback=None) + if not logs_field: + logs_ids = ["default"] + else: + logs_ids = [tok for tok in logs_field.split() if tok] + if not logs_ids: + raise ConfigError(f"Rule {rule_id!r} has empty logs list") + for lid in logs_ids: + if lid not in log_sets: + raise ConfigError( + f"Rule {rule_id!r} references unknown log set {lid!r}" + ) normalized, compiled = _compile_pattern(pattern_raw) + combined_paths: list[str] = [] + _seen_paths: set[str] = set() + for lid in logs_ids: + for p in log_sets[lid]: + if p not in _seen_paths: + _seen_paths.add(p) + combined_paths.append(p) rules[rule_id] = Rule( rule_id=rule_id, pattern=normalized, description=description, action_id=action_id, compiled=compiled, + log_set_ids=logs_ids, + log_base_paths=combined_paths, ) if not rules: raise ConfigError("No rules defined (need at least one [rule.*] section)") - return log_paths, rules, actions, max_block_size, state_dir + all_log_paths: list[str] = [] + seen: set[str] = set() + for paths in log_sets.values(): + for p in paths: + if p not in seen: + seen.add(p) + all_log_paths.append(p) + + return all_log_paths, rules, actions, max_block_size, state_dir class LockFile: @@ -394,24 +432,23 @@ def discover_files(paths: list[str]) -> Iterator[str]: def _is_text(path: str) -> bool: try: proc = subprocess.run( - ["file", "-b", path], # type: ignore[arg-type] # pyright: acceptable sequence of str + ["file", "-b", path], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, ) if proc.returncode != 0: - return True # permissive fallback + return True desc = proc.stdout.lower() return "text" in desc except Exception: - return True # permissive fallback on unexpected errors + return True for p in paths: if not p: continue if os.path.isfile(p): - # Explicit file path: yield directly yield p elif os.path.isdir(p): for root, _, files in os.walk(p): @@ -445,7 +482,6 @@ def read_new_lines( fh.seek(0, os.SEEK_END) file_size = fh.tell() if file_size < start_offset: - # Truncated start_offset = 0 start_line_number = 0 fh.seek(start_offset) @@ -456,7 +492,6 @@ def read_new_lines( break bytes_read += len(line_bytes) if bytes_read > max_block_size: - # Do not include this partial line; rewind to before line. bytes_read -= len(line_bytes) break try: @@ -490,18 +525,26 @@ def generate_matches( ) -> Iterator[MatchEvent]: """ Iterate over lines and yield MatchEvent objects for each rule match. + A rule is only applied if the file belongs to its configured log set + (file path equal to or under one of the set's base paths). """ total_lines = len(lines) for idx, line in enumerate(lines): absolute_line_number = initial_line_number + idx + 1 for rule in rules.values(): + applies = False + for base in rule.log_base_paths: + b = base.rstrip(os.sep) + if file_path == b or file_path.startswith(b + os.sep): + applies = True + break + if not applies: + continue m = rule.compiled.search(line) if not m: continue - # Build context start_ctx = max(0, idx - context_radius) - end_ctx = min(total_lines, idx + context_radius + 1) - context = "".join(lines[start_ctx:end_ctx]) + context = "".join(lines[start_ctx : idx + 1]) yield MatchEvent( rule=rule, file_path=file_path, @@ -545,7 +588,7 @@ def execute_action( return 0 try: - proc = subprocess.run( # type: ignore + proc = subprocess.run( action.cmd, shell=True, env=env, @@ -599,7 +642,6 @@ class Sentinel: try: os.makedirs(self.state_dir, exist_ok=True) except OSError as e: - # Explicit error output before failing; re-raise original exception print( f"ERROR: Failed to create state_dir '{self.state_dir}': {e}", file=sys.stderr, @@ -625,7 +667,6 @@ class Sentinel: file_path, prev_offset, prev_line_no, self.max_block_size ) if not lines: - # No new content OR failed to read return for event in generate_matches( @@ -685,7 +726,6 @@ def parse_args(argv: Sequence[str]) -> tuple[str, bool]: elif arg == "--skip-actions": skip_actions = True else: - # Unrecognized argument raise SystemExit(f"ERROR: Unrecognized argument: {arg}") return config_path, skip_actions @@ -698,18 +738,26 @@ def main(argv: Sequence[str] | None = None) -> int: argv = sys.argv[1:] try: config_path, skip_actions = parse_args(list(argv)) - # If the provided (or default) config path does not exist, attempt standard locations. + candidates = [ + config_path, + os.path.expanduser("~/.pylogsentinel.conf"), + os.path.expanduser("~/.config/pylogsentinel.conf"), + "/etc/pylogsentinel.conf", + "/usr/local/etc/pylogsentinel.conf", + ] if not os.path.isfile(config_path): - for cand in ( - os.path.expanduser("~/.pylogsentinel.conf"), - os.path.expanduser("~/.config/pylogsentinel.conf"), - "/etc/pylogsentinel.conf", - "/usr/local/etc/pylogsentinel.conf", - config_path, # fallback original (may still not exist) - ): + for cand in candidates[1:]: if os.path.isfile(cand): config_path = cand break + if not os.path.isfile(config_path): + print( + "CONFIG ERROR: No configuration file found (searched: " + + ", ".join(candidates) + + ")", + file=sys.stderr, + ) + return 2 ( log_paths, rules, @@ -727,7 +775,10 @@ def main(argv: Sequence[str] | None = None) -> int: ) if skip_actions: print(f"[skip-actions] Using config: {config_path}", file=sys.stderr) - print(f"[skip-actions] Log path entries: {len(log_paths)}", file=sys.stderr) + print( + f"[skip-actions] Log path entries (union): {len(log_paths)}", + file=sys.stderr, + ) print( f"[skip-actions] Rules: {', '.join(sorted(rules.keys()))}", file=sys.stderr, diff --git a/pyproject.toml b/pyproject.toml index 5133c3c..ea6a84f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "pylogsentinel" -version = "0.2.1" +version = "0.3.0" description = "Lightweight cron-friendly log monitoring utility with regex rules and per-rule shell actions" readme = "README.md" requires-python = ">=3.9" |