Add one-shot Elasticsearch bulk export with ECS field mapping

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")
 
 

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)

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 [],
+    }

tests/test_cli.py

@@ -1,17 +1,20 @@
 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)
 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
+