diff --git a/src/vmstat_visualizer/cli/__main__.py b/src/vmstat_visualizer/cli/__main__.py index 11f7094..8b12c7c 100644 --- a/src/vmstat_visualizer/cli/__main__.py +++ b/src/vmstat_visualizer/cli/__main__.py @@ -4,6 +4,7 @@ import click import vmstat_visualizer.cli.visualizer as viz +from vmstat_visualizer.cli.es_export import export_es @click.group(no_args_is_help=True) @@ -17,6 +18,7 @@ def main(): "main is the entrypoint of vmstat-visualizer" cli.add_command(viz.visualize) cli.add_command(viz.compare) + cli.add_command(export_es) cli(prog_name="vmstat-visualizer") diff --git a/src/vmstat_visualizer/cli/es_export.py b/src/vmstat_visualizer/cli/es_export.py new file mode 100644 index 0000000..c02cce3 --- /dev/null +++ b/src/vmstat_visualizer/cli/es_export.py @@ -0,0 +1,85 @@ +"""CLI command for one-shot Elasticsearch export.""" + +import click +from vmstat_visualizer.parser.parser import Parser + + +@click.command("export-es", no_args_is_help=True) +@click.argument("file", type=click.Path(exists=True)) +@click.option( + "--es-url", + default=None, + help="Elasticsearch URL (e.g. https://localhost:9200). Not needed with --cloud-id.", +) +@click.option( + "--es-api-key", + default=None, + envvar="ES_API_KEY", + help="Elasticsearch API key. Can also be set via ES_API_KEY env var.", +) +@click.option( + "--cloud-id", + default=None, + envvar="ES_CLOUD_ID", + help="Elastic Cloud ID. Can also be set via ES_CLOUD_ID env var.", +) +@click.option( + "--index", + default="vmstat", + help="Target Elasticsearch index name.", +) +@click.option( + "--dry-run", + is_flag=True, + default=False, + help="Print the bulk request body to stdout instead of sending.", +) +@click.option( + "--no-verify-certs", + is_flag=True, + default=False, + help="Disable TLS certificate verification.", +) +def export_es(file, es_url, es_api_key, cloud_id, index, dry_run, + no_verify_certs): + """Export parsed vmstat data to Elasticsearch in one shot. + + Requires: pip install elasticsearch + """ + parser = Parser(file) + parser.parse() + + if not parser.timeseries: + click.echo("No data found in file.", err=True) + raise SystemExit(1) + + click.echo(f">>> Parsed {len(parser.timeseries)} entries from {file}") + + if dry_run: + from vmstat_visualizer.es_exporter import build_bulk_body + body = build_bulk_body(parser.timeseries, index=index, + source_file=file) + click.echo(body) + return + + if not es_url and not cloud_id: + raise click.UsageError( + "Either --es-url or --cloud-id is required (unless --dry-run)." + ) + + from vmstat_visualizer.es_exporter import export_to_elasticsearch + result = export_to_elasticsearch( + parser.timeseries, + es_url=es_url, + api_key=es_api_key, + cloud_id=cloud_id, + index=index, + source_file=file, + verify_certs=not no_verify_certs, + ) + + click.echo(f">>> Indexed {result['indexed']} documents to '{index}'") + if result['errors']: + click.echo(f">>> {len(result['errors'])} errors occurred", err=True) + for err in result['errors'][:5]: + click.echo(f" {err}", err=True) diff --git a/src/vmstat_visualizer/es_exporter.py b/src/vmstat_visualizer/es_exporter.py new file mode 100644 index 0000000..b399974 --- /dev/null +++ b/src/vmstat_visualizer/es_exporter.py @@ -0,0 +1,137 @@ +"""One-shot bulk export of parsed vmstat data to Elasticsearch. + +Requires: pip install elasticsearch +(or: pip install vmstat-visualizer[elasticsearch]) + +Uses ECS-compatible field names so data works with standard Kibana +dashboards and can be correlated with other observability data. +""" + +import json +import sys + + +ECS_FIELD_MAP = { + 'time': '@timestamp', + 'run_queue': 'system.load.1', + 'blocked_processes': 'system.process.blocked', + 'swapped_memory_kb': 'system.memory.swap.used.bytes', + 'free_memory_kb': 'system.memory.free.bytes', + 'inactive_memory_kb': 'system.memory.inactive.bytes', + 'active_memory_kb': 'system.memory.active.bytes', + 'swap_in_kb': 'system.memory.swap.in.bytes', + 'swap_out_kb': 'system.memory.swap.out.bytes', + 'blocks_in': 'system.diskio.read.bytes', + 'blocks_out': 'system.diskio.write.bytes', + 'interrupts': 'system.cpu.interrupts', + 'context_switches': 'system.cpu.context_switches', + 'user_cpu_percent': 'system.cpu.user.pct', + 'system_cpu_percent': 'system.cpu.system.pct', + 'idle_cpu_percent': 'system.cpu.idle.pct', + 'wait_cpu_percent': 'system.cpu.iowait.pct', + 'steal_cpu_percent': 'system.cpu.steal.pct', + 'guest_cpu_percent': 'system.cpu.guest.pct', +} + +KB_FIELDS = { + 'swapped_memory_kb', 'free_memory_kb', 'inactive_memory_kb', + 'active_memory_kb', 'swap_in_kb', 'swap_out_kb', + 'blocks_in', 'blocks_out', +} + +PCT_FIELDS = { + 'user_cpu_percent', 'system_cpu_percent', 'idle_cpu_percent', + 'wait_cpu_percent', 'steal_cpu_percent', 'guest_cpu_percent', +} + + +def _to_ecs_doc(ts_entry, source_file=None): + """Convert a Timeseries entry to an ECS-compatible dict.""" + doc = {} + for attr, ecs_field in ECS_FIELD_MAP.items(): + val = getattr(ts_entry, attr, None) + if val is None: + continue + + if attr == 'time': + doc[ecs_field] = val.replace(' ', 'T') + elif attr in KB_FIELDS: + doc[ecs_field] = int(val) * 1024 + elif attr in PCT_FIELDS: + doc[ecs_field] = int(val) / 100.0 + else: + doc[ecs_field] = int(val) + + doc['event.module'] = 'vmstat' + doc['event.dataset'] = 'system.vmstat' + if source_file: + doc['file.path'] = source_file + + return doc + + +def build_bulk_body(timeseries, index, source_file=None): + """Build an Elasticsearch bulk request body (NDJSON).""" + lines = [] + for entry in timeseries: + action = json.dumps({"index": {"_index": index}}) + doc = json.dumps(_to_ecs_doc(entry, source_file=source_file)) + lines.append(action) + lines.append(doc) + return "\n".join(lines) + "\n" + + +def export_to_elasticsearch(timeseries, es_url, api_key=None, + cloud_id=None, index='vmstat', + source_file=None, verify_certs=True): + """Bulk-index vmstat data into Elasticsearch. + + Args: + timeseries: list of Timeseries entries + es_url: Elasticsearch URL (ignored if cloud_id is set) + api_key: API key string + cloud_id: Elastic Cloud ID + index: target index name + source_file: original vmstat log filename + verify_certs: whether to verify TLS certificates + + Returns: + dict with 'indexed' count and 'errors' list + """ + try: + from elasticsearch import Elasticsearch, helpers + except ImportError: + print( + "ERROR: elasticsearch-py is required. " + "Install with: pip install elasticsearch", + file=sys.stderr, + ) + raise SystemExit(1) + + kwargs = {} + if cloud_id: + kwargs['cloud_id'] = cloud_id + else: + kwargs['hosts'] = [es_url] + + if api_key: + kwargs['api_key'] = api_key + + kwargs['verify_certs'] = verify_certs + + es = Elasticsearch(**kwargs) + + actions = [] + for entry in timeseries: + doc = _to_ecs_doc(entry, source_file=source_file) + actions.append({ + "_index": index, + "_source": doc, + }) + + success, errors = helpers.bulk(es, actions, raise_on_error=False) + + return { + 'indexed': success, + 'errors': errors if errors else [], + } diff --git a/tests/test_cli.py b/tests/test_cli.py index 073361e..8b0c118 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,10 +1,12 @@ import os import glob +import json import pytest from unittest.mock import patch from click.testing import CliRunner from vmstat_visualizer.cli.__main__ import cli import vmstat_visualizer.cli.visualizer as viz +from vmstat_visualizer.cli.es_export import export_es @pytest.fixture(autouse=True) @@ -12,6 +14,7 @@ def _register_commands(): """Ensure commands are registered before each test.""" cli.add_command(viz.visualize) cli.add_command(viz.compare) + cli.add_command(export_es) class TestCLIHelp: @@ -21,6 +24,7 @@ def test_main_help(self): assert result.exit_code == 0 assert 'visualize' in result.output assert 'compare' in result.output + assert 'export-es' in result.output def test_visualize_help(self): runner = CliRunner() @@ -112,3 +116,58 @@ def test_compare_invalid_metric(self, active_log): 'compare', active_log, active_log, '-m', 'bogus' ]) assert result.exit_code != 0 + + +class TestExportEsCommand: + @patch("vmstat_visualizer.parser.parser.check_vmstat_columns", + return_value=(True, True)) + def test_export_es_help_shows_expected_options(self, mock_check): + runner = CliRunner() + result = runner.invoke(cli, ['export-es', '--help']) + assert result.exit_code == 0 + for needle in ( + '--es-url', + '--es-api-key', + '--cloud-id', + '--index', + '--dry-run', + '--no-verify-certs', + ): + assert needle in result.output + + @patch("vmstat_visualizer.parser.parser.check_vmstat_columns", + return_value=(True, True)) + def test_export_es_dry_run_prints_bulk_to_stdout(self, mock_check, + active_log): + runner = CliRunner() + result = runner.invoke(cli, ['export-es', '--dry-run', active_log]) + + assert result.exit_code == 0 + out = result.output + json_lines = [ln for ln in out.splitlines() if ln.startswith('{')] + assert len(json_lines) == 10 + assert json.loads(json_lines[0]) == {"index": {"_index": "vmstat"}} + + doc = json.loads(json_lines[1]) + assert doc.get("event.module") == "vmstat" + assert doc.get("file.path") == active_log + + assert '"index"' in out and '"system.' in out + + @patch("vmstat_visualizer.parser.parser.check_vmstat_columns", + return_value=(True, True)) + def test_export_es_without_endpoint_errors(self, mock_check, active_log): + runner = CliRunner() + result = runner.invoke(cli, ['export-es', active_log]) + assert result.exit_code != 0 + low = result.output.lower() + assert 'es-url' in low or 'cloud-id' in low + + @patch("vmstat_visualizer.parser.parser.check_vmstat_columns", + return_value=(True, True)) + def test_export_es_empty_file_errors(self, mock_check, empty_log): + runner = CliRunner() + result = runner.invoke(cli, ['export-es', '--dry-run', empty_log]) + assert result.exit_code == 1 + assert 'No data found' in result.output + diff --git a/tests/test_es_exporter.py b/tests/test_es_exporter.py new file mode 100644 index 0000000..4347da9 --- /dev/null +++ b/tests/test_es_exporter.py @@ -0,0 +1,202 @@ +"""Tests for Elasticsearch ECS export helpers.""" + +import json +import sys +from types import ModuleType +from unittest.mock import MagicMock, patch + +import pytest + +from vmstat_visualizer.parser.timeseries import Timeseries + + +def _timeseries_metric_attr_names(): + """Attributes on Timeseries that represent exported metrics.""" + exclude = frozenset({"raw_data"}) + names = [] + ts = Timeseries() + for name in vars(ts): + if name.startswith("_") or name in exclude: + continue + names.append(name) + return frozenset(names) + + +@pytest.fixture +def elasticsearch_fake(): + """Inject a stub ``elasticsearch`` package so exporter can import locally.""" + es_pkg = ModuleType("elasticsearch") + helpers_pkg = ModuleType("elasticsearch.helpers") + + es_pkg.Elasticsearch = MagicMock(return_value=MagicMock(name="client")) + helpers_pkg.bulk = MagicMock(return_value=(0, [])) + es_pkg.helpers = helpers_pkg + + with patch.dict( + sys.modules, + { + "elasticsearch": es_pkg, + "elasticsearch.helpers": helpers_pkg, + }, + ): + yield es_pkg + + +def test_ecs_field_map_covers_timeseries_metric_attrs(): + from vmstat_visualizer.es_exporter import ECS_FIELD_MAP + + expected = _timeseries_metric_attr_names() + assert set(ECS_FIELD_MAP.keys()) == expected + + +def test_kb_fields_and_pct_fields_are_subsets(): + from vmstat_visualizer.es_exporter import ECS_FIELD_MAP, KB_FIELDS, PCT_FIELDS + + keys = set(ECS_FIELD_MAP) + assert KB_FIELDS.issubset(keys) + assert PCT_FIELDS.issubset(keys) + + +def test_to_ecs_doc_timestamp_and_transforms_and_event_fields(): + from vmstat_visualizer.es_exporter import _to_ecs_doc + + e = Timeseries() + e.time = "2025-07-31 23:52:52" + e.free_memory_kb = 100 + e.user_cpu_percent = 50 + e.interrupts = 12345 + e.context_switches = 987654 + + doc = _to_ecs_doc(e, source_file=None) + + assert doc["@timestamp"] == "2025-07-31T23:52:52" + assert doc["system.memory.free.bytes"] == 102400 + assert doc["system.cpu.user.pct"] == 0.5 + assert doc["system.cpu.interrupts"] == 12345 + assert doc["system.cpu.context_switches"] == 987654 + assert doc["event.module"] == "vmstat" + assert doc["event.dataset"] == "system.vmstat" + assert "file.path" not in doc + + +def test_to_ecs_doc_omits_none_and_includes_file_path(): + from vmstat_visualizer.es_exporter import _to_ecs_doc + + e = Timeseries() + e.time = "2025-01-01 00:00:00" + e.user_cpu_percent = 10 + + doc = _to_ecs_doc(e, source_file="/logs/vm.csv") + + assert "@timestamp" in doc + assert "system.cpu.user.pct" in doc + assert doc["file.path"] == "/logs/vm.csv" + + +def test_build_bulk_body_ndjson_structure(): + from vmstat_visualizer.es_exporter import build_bulk_body + + e = Timeseries() + e.time = "2025-07-31 01:02:03" + e.run_queue = 1 + + body = build_bulk_body([e], index="vmstat", source_file="/a.log") + lines = body.splitlines() + + assert len(lines) == 2 + action = json.loads(lines[0]) + doc = json.loads(lines[1]) + + assert action == {"index": {"_index": "vmstat"}} + assert doc["event.module"] == "vmstat" + assert doc["file.path"] == "/a.log" + assert body.endswith("\n") + + +def test_build_bulk_body_empty_returns_single_newline(): + from vmstat_visualizer.es_exporter import build_bulk_body + + body = build_bulk_body([], index="vmstat") + assert body == "\n" + + +def test_export_to_elasticsearch_hosts_and_bulk(elasticsearch_fake): + from vmstat_visualizer.es_exporter import export_to_elasticsearch + + es_cls = elasticsearch_fake.Elasticsearch + bulk = elasticsearch_fake.helpers.bulk + bulk.return_value = (2, ["err1"]) + + e1 = Timeseries() + e1.time = "2025-07-31 01:02:03" + e1.idle_cpu_percent = 90 + e2 = Timeseries() + e2.time = "2025-07-31 01:03:03" + e2.idle_cpu_percent = 91 + + result = export_to_elasticsearch( + [e1, e2], + es_url="https://localhost:9200", + api_key="sekret", + cloud_id=None, + index="myidx", + source_file="/tmp/f.log", + verify_certs=False, + ) + + es_cls.assert_called_once_with( + hosts=["https://localhost:9200"], + api_key="sekret", + verify_certs=False, + ) + client = es_cls.return_value + + bulk.assert_called_once() + call_args = bulk.call_args + assert call_args.args[0] is client + + actions = call_args.args[1] + assert len(actions) == 2 + assert actions[0]["_index"] == "myidx" + assert actions[0]["_source"]["file.path"] == "/tmp/f.log" + + assert result == {"indexed": 2, "errors": ["err1"]} + + +def test_export_to_elasticsearch_cloud_id(elasticsearch_fake): + from vmstat_visualizer.es_exporter import export_to_elasticsearch + + es_cls = elasticsearch_fake.Elasticsearch + bulk = elasticsearch_fake.helpers.bulk + bulk.return_value = (1, []) + + e = Timeseries() + e.time = "2025-07-31 01:02:03" + + export_to_elasticsearch( + [e], + es_url="ignored", + api_key=None, + cloud_id="cloud:ZWxhc3RpYw==", + index="cloudidx", + source_file=None, + verify_certs=True, + ) + + es_cls.assert_called_once_with( + cloud_id="cloud:ZWxhc3RpYw==", + verify_certs=True, + ) + + +def test_export_helpers_bulk_empty_errors_normalized(elasticsearch_fake): + from vmstat_visualizer.es_exporter import export_to_elasticsearch + + bulk = elasticsearch_fake.helpers.bulk + bulk.return_value = (3, []) # falsy errors -> [] + + e = Timeseries() + e.time = "2025-07-31 01:02:03" + + result = export_to_elasticsearch([e], es_url="http://es", index="x") + assert result == {"indexed": 3, "errors": []}