#!/usr/bin/env python3 """Generate versioned index mappings and portable OpenSearch saved objects.""" import json from pathlib import Path ROOT = Path(__file__).resolve().parents[1] INDEX = 'hm-index-v1' OBJECTS = [] FIELDS = {'@timestamp': 'date', 'ingested_at': 'date', 'timestamp': 'number', 'name': 'string'} TAGS = ['host', 'environment', 'role', 'series', 'cpu', 'path', 'device', 'fstype', 'mode', 'name', 'interface'] FIELDS.update({'tags.' + t: 'string' for t in TAGS}) FORMATS = {} def dumps(value): return json.dumps(value, separators=(',', ':')) def write_json(path, value): (ROOT / path).write_text(json.dumps(value, indent=2) + '\n') def agg(field, label=None, op='avg'): FIELDS.setdefault(field, 'number') if any(s in field for s in ['bytes', 'fields.available', 'fields.used', 'fields.free', 'fields.total', 'fields.cached', 'fields.buffered']): if 'percent' not in field: FORMATS[field] = {'id': 'bytes', 'params': {'pattern': '0.0 b'}} return {'type': op, 'schema': 'metric', 'params': {'field': field, 'customLabel': label or field.split('.')[-1]}} def vis(key, title, query, metrics, kind='line', split='tags.series', description=''): aggs = [dict(a, id=str(i + 1), enabled=True) for i, a in enumerate(metrics)] if kind == 'line': aggs.append({'id': 'time', 'enabled': True, 'type': 'date_histogram', 'schema': 'segment', 'params': {'field': '@timestamp', 'interval': 'auto', 'min_doc_count': 1, 'extended_bounds': {}}}) if split and kind in ('line', 'table'): aggs.append({'id': 'series', 'enabled': True, 'type': 'terms', 'schema': 'group' if kind == 'line' else 'bucket', 'params': {'field': split, 'size': 30, 'order': 'desc', 'orderBy': '_count', 'otherBucket': False, 'missingBucket': False}}) if kind == 'metric': params = {'addTooltip': True, 'addLegend': False, 'type': 'metric', 'metric': {'percentageMode': False, 'useRanges': False, 'colorSchema': 'Green to Red', 'metricColorMode': 'None', 'colorsRange': [{'from': 0, 'to': 100}], 'labels': {'show': True}, 'invertColors': False, 'style': {'bgFill': '#000', 'bgColor': False, 'labelColor': False, 'subText': '', 'fontSize': 42}}} elif kind == 'table': params = {'perPage': 15, 'showPartialRows': False, 'showMetricsAtAllLevels': False, 'sort': {'columnIndex': None, 'direction': None}, 'showTotal': False, 'totalFunc': 'sum'} else: params = {'type': 'line', 'addTooltip': True, 'addLegend': True, 'legendPosition': 'bottom', 'grid': {'categoryLines': False}, 'times': [], 'addTimeMarker': False, 'categoryAxes': [{'id': 'CategoryAxis-1', 'type': 'category', 'position': 'bottom', 'show': True, 'style': {}, 'scale': {'type': 'linear'}, 'labels': {'show': True, 'truncate': 100}, 'title': {}}], 'valueAxes': [{'id': 'ValueAxis-1', 'type': 'value', 'position': 'left', 'show': True, 'style': {}, 'scale': {'type': 'linear', 'mode': 'normal'}, 'labels': {'show': True}, 'title': {'text': ''}}], 'seriesParams': [{'show': True, 'type': 'line', 'mode': 'normal', 'data': {'id': a['id'], 'label': a['params'].get('customLabel', '')}, 'valueAxis': 'ValueAxis-1', 'drawLinesBetweenPoints': True, 'showCircles': False, 'lineWidth': 2, 'interpolate': 'linear'} for a in aggs if a['schema'] == 'metric']} OBJECTS.append({'type': 'visualization', 'id': key, 'attributes': { 'title': title, 'description': description, 'version': 1, 'visState': dumps({'title': title, 'type': kind, 'params': params, 'aggs': aggs}), 'uiStateJSON': '{}', 'kibanaSavedObjectMeta': {'searchSourceJSON': dumps({'query': {'language': 'lucene', 'query': query}, 'filter': [], 'indexRefName': 'kibanaSavedObjectMeta.searchSourceJSON.index'})}}, 'references': [{'name': 'kibanaSavedObjectMeta.searchSourceJSON.index', 'type': 'index-pattern', 'id': INDEX}]}) return key def markdown(key, title, body): OBJECTS.append({'type': 'visualization', 'id': key, 'attributes': { 'title': title, 'description': '', 'version': 1, 'uiStateJSON': '{}', 'visState': dumps({'title': title, 'type': 'markdown', 'params': {'markdown': body, 'openLinksInNewTab': False}, 'aggs': []}), 'kibanaSavedObjectMeta': {'searchSourceJSON': dumps({'query': {'query': '', 'language': 'lucene'}, 'filter': []})}}, 'references': []}) return key def dashboard(key, title, subtitle, cards, charts): nav = ' • '.join(f'[{label}](/app/dashboards#/view/hm-{target})' for target, label in [('fleet', 'Fleet'), ('compute', 'CPU & memory'), ('storage', 'Storage'), ('network', 'Network')]) banner = markdown(key + '-intro', title + ' / guide', f'# {title}\n{subtitle}\n\n{nav}\n\n' '**Scope:** use **Add filter → tags.host** to isolate a server; `tags.environment` and `tags.role` select a fleet. ' 'Cards summarize the selected time range. Charts show the top 30 series; narrow the filter for larger fleets. ' 'Empty charts indicate missing data, never zero. Times are UTC.') placements = [(banner, 0, 0, 48, 8)] for i, card in enumerate(cards): placements.append((card, i * 12, 8, 12, 8)) for i, chart in enumerate(charts): placements.append((chart, (i % 2) * 24, 16 + (i // 2) * 15, 24, 15)) panels, refs = [], [] for i, (obj, x, y, w, h) in enumerate(placements): ref = f'panel_{i}' panels.append({'version': '3.8.0', 'type': 'visualization', 'panelIndex': str(i + 1), 'panelRefName': ref, 'embeddableConfig': {}, 'gridData': {'x': x, 'y': y, 'w': w, 'h': h, 'i': str(i + 1)}}) refs.append({'name': ref, 'type': 'visualization', 'id': obj}) OBJECTS.append({'type': 'dashboard', 'id': key, 'attributes': { 'title': 'Host Metrics / ' + title, 'description': subtitle, 'version': 1, 'panelsJSON': dumps(panels), 'optionsJSON': dumps({'useMargins': True, 'hidePanelTitles': False}), 'timeRestore': True, 'timeFrom': 'now-1h', 'timeTo': 'now', 'refreshInterval': {'pause': False, 'value': 15000}, 'kibanaSavedObjectMeta': {'searchSourceJSON': dumps({'query': {'language': 'lucene', 'query': ''}, 'filter': []})}}, 'references': refs}) def m(key, title, query, field, op='avg'): return vis('hm-' + key, title, query, [agg(field, title, op)], 'metric', None) def line(key, title, query, *fields, split='tags.series'): return vis('hm-' + key, title, query, [agg(f, label) for f, label in fields], split=split) CPU = 'name:cpu AND tags.cpu:"cpu-total"' active = vis('hm-hosts', 'Hosts seen in selected range', 'name:system', [agg('tags.host', 'Hosts', 'cardinality')], 'metric', None) cpu_peak = m('cpu-peak', 'Peak CPU busy (%)', CPU, 'fields.usage_active', 'max') ram_peak = m('ram-peak', 'Peak RAM used (%)', 'name:mem', 'fields.used_percent', 'max') disk_peak = m('disk-peak', 'Peak filesystem used (%)', 'name:disk', 'fields.used_percent', 'max') cpu = line('cpu', 'CPU busy (%) · by host', CPU, ('fields.usage_active', 'Busy %'), split='tags.host') ram = line('ram', 'RAM used (%) · by host', 'name:mem', ('fields.used_percent', 'Used %'), split='tags.host') load = line('load-normalized', 'Load / logical CPU · 1.0 means one task per CPU', 'name:system', ('fields.load1_per_cpu', 'Load / CPU'), split='tags.host') fs = line('fs-used', 'Filesystem used (%) · by mount', 'name:disk', ('fields.used_percent', 'Used %')) fresh = vis('hm-freshness', 'Latest sample per host · compare with current UTC time', 'name:system', [dict(type='top_hits', schema='metric', params={'field': '@timestamp', 'aggregate': 'concat', 'size': 1, 'sortField': '@timestamp', 'sortOrder': 'desc', 'customLabel': 'Latest sample (UTC)'})], 'table', 'tags.host') uptime = vis('hm-uptime', 'Host inventory · range statistics', 'name:system', [agg('fields.uptime', 'Max uptime (s)', 'max'), agg('fields.n_cpus', 'Logical CPUs', 'max'), agg('fields.load1', 'Peak load', 'max')], 'table', 'tags.host') dashboard('hm-fleet', 'Fleet overview', 'Health, capacity and collection freshness across your servers.', [active, cpu_peak, ram_peak, disk_peak], [cpu, ram, load, fs, fresh, uptime]) dashboard('hm-compute', 'CPU & memory', 'CPU saturation, scheduler pressure, memory headroom and process health.', [cpu_peak, ram_peak, m('iowait-peak', 'Peak I/O wait (%)', CPU, 'fields.usage_iowait', 'max'), m('swap-peak', 'Peak swap used (%)', 'name:swap', 'fields.used_percent', 'max')], [cpu, line('cpu-modes', 'CPU modes (%) · user / system / I/O wait / steal', CPU, ('fields.usage_user', 'User'), ('fields.usage_system', 'System'), ('fields.usage_iowait', 'I/O wait'), ('fields.usage_steal', 'Steal'), split='tags.host'), line('cpu-core', 'Per-core CPU busy (%)', 'name:cpu AND NOT tags.cpu:"cpu-total"', ('fields.usage_active', 'Busy %')), line('load', 'Load average · 1 / 5 / 15 minutes', 'name:system', ('fields.load1', '1m'), ('fields.load5', '5m'), ('fields.load15', '15m'), split='tags.host'), ram, line('memory', 'RAM headroom · available / cached / buffered', 'name:mem', ('fields.available', 'Available'), ('fields.cached', 'Cached'), ('fields.buffered', 'Buffered'), split='tags.host'), line('swap-io', 'Swap activity · bytes/s', 'name:swap', ('fields.in_per_sec', 'In'), ('fields.out_per_sec', 'Out'), split='tags.host'), line('processes', 'Processes · running / blocked / zombie', 'name:processes', ('fields.running', 'Running'), ('fields.blocked', 'Blocked'), ('fields.zombie', 'Zombie'), split='tags.host'), line('context', 'Context switches / second', 'name:kernel', ('fields.context_switches_per_sec', 'Switches/s'), split='tags.host'), line('interrupts', 'Interrupts / second', 'name:kernel', ('fields.interrupts_per_sec', 'Interrupts/s'), split='tags.host')]) dashboard('hm-storage', 'Storage', 'Filesystem and inode capacity, per-device throughput, operations and busy time.', [disk_peak, m('inode-peak', 'Peak inode use (%)', 'name:disk', 'fields.inodes_used_percent', 'max'), m('disk-busy-peak', 'Peak device busy (%)', 'name:diskio', 'fields.io_busy_percent', 'max'), m('disk-free-min', 'Minimum mount free bytes', 'name:disk', 'fields.free', 'min')], [fs, line('fs-free', 'Free bytes · by mount', 'name:disk', ('fields.free', 'Free bytes')), line('inodes', 'Inode use (%) · by mount', 'name:disk', ('fields.inodes_used_percent', 'Inodes used %')), vis('hm-fs-table', 'Mount capacity · extrema over selected range', 'name:disk', [agg('fields.total', 'Capacity', 'max'), agg('fields.free', 'Minimum free', 'min'), agg('fields.used_percent', 'Peak used %', 'max'), agg('fields.inodes_used_percent', 'Peak inode %', 'max')], 'table'), line('disk-throughput', 'Disk throughput · bytes/s', 'name:diskio', ('fields.read_bytes_per_sec', 'Read'), ('fields.write_bytes_per_sec', 'Write')), line('disk-iops', 'Disk operations / second', 'name:diskio', ('fields.reads_per_sec', 'Read IOPS'), ('fields.writes_per_sec', 'Write IOPS')), line('disk-busy', 'Device busy (%) · not an NVMe saturation measurement', 'name:diskio', ('fields.io_busy_percent', 'Busy %')), line('disk-queue', 'I/O requests in progress', 'name:diskio', ('fields.iops_in_progress', 'In progress'))]) dashboard('hm-network', 'Network', 'Per-interface throughput, packet rates, errors and drops. Loopback and Docker bridges are excluded.', [m('rx-peak', 'Peak interface RX bytes/s', 'name:net', 'fields.bytes_recv_per_sec', 'max'), m('tx-peak', 'Peak interface TX bytes/s', 'name:net', 'fields.bytes_sent_per_sec', 'max'), m('rx-errors', 'Peak RX errors/s', 'name:net', 'fields.err_in_per_sec', 'max'), m('rx-drops', 'Peak RX drops/s', 'name:net', 'fields.drop_in_per_sec', 'max')], [line('network-bytes', 'Network throughput · bytes/s', 'name:net', ('fields.bytes_recv_per_sec', 'RX'), ('fields.bytes_sent_per_sec', 'TX')), line('network-packets', 'Packets / second', 'name:net', ('fields.packets_recv_per_sec', 'RX'), ('fields.packets_sent_per_sec', 'TX')), line('network-errors', 'Errors / second', 'name:net', ('fields.err_in_per_sec', 'RX errors'), ('fields.err_out_per_sec', 'TX errors')), line('network-drops', 'Drops / second', 'name:net', ('fields.drop_in_per_sec', 'RX drops'), ('fields.drop_out_per_sec', 'TX drops')), vis('hm-net-table', 'Interface inventory · peaks over selected range', 'name:net', [agg('fields.bytes_recv_per_sec', 'Peak RX', 'max'), agg('fields.bytes_sent_per_sec', 'Peak TX', 'max'), agg('fields.err_in_per_sec', 'Peak errors/s', 'max')], 'table'), fresh]) field_list = [{'name': k, 'type': v, 'count': 0, 'scripted': False, 'searchable': True, 'aggregatable': True, 'readFromDocValues': True} for k, v in sorted(FIELDS.items())] OBJECTS.insert(0, {'type': 'index-pattern', 'id': INDEX, 'attributes': { 'title': 'host-metrics-v1-*', 'timeFieldName': '@timestamp', 'fields': dumps(field_list), 'fieldFormatMap': dumps(FORMATS)}, 'references': []}) (ROOT / 'dashboards/host-metrics.ndjson').write_text('\n'.join(dumps(o) for o in OBJECTS) + '\n') write_json('dashboards/host-metrics.json', OBJECTS) properties = {'@timestamp': {'type': 'date'}, 'ingested_at': {'type': 'date'}, 'timestamp': {'type': 'long'}, 'name': {'type': 'keyword'}, 'tags': {'type': 'object', 'dynamic': True}, 'fields': {'type': 'object', 'dynamic': True, 'properties': {k.removeprefix('fields.'): {'type': 'double'} for k in FIELDS if k.startswith('fields.')}}} write_json('configs/opensearch/index-template.json', { 'index_patterns': ['host-metrics-v1-*'], 'priority': 200, 'version': 1, 'template': {'settings': {'number_of_shards': 1, 'number_of_replicas': 0, 'refresh_interval': '5s', 'index.mapping.total_fields.limit': 500}, 'mappings': {'dynamic': True, 'properties': properties, 'dynamic_templates': [ {'tags': {'path_match': 'tags.*', 'mapping': {'type': 'keyword', 'ignore_above': 1024}}}, {'numeric_fields': {'path_match': 'fields.*', 'mapping': {'type': 'double'}}}]}}}) write_json('configs/opensearch/retention-policy.json', {'policy': { 'description': 'Delete host metric daily indices after 14 days; applies only to host-metrics-v1-*.', 'default_state': 'hot', 'states': [ {'name': 'hot', 'actions': [], 'transitions': [{'state_name': 'delete', 'conditions': {'min_index_age': '14d'}}]}, {'name': 'delete', 'actions': [{'delete': {}}], 'transitions': []}], 'ism_template': [{'index_patterns': ['host-metrics-v1-*'], 'priority': 200}]}}) bridge = '''# Kafka-to-Prometheus adapter only; no local host inputs. [agent] flush_interval = "5s" metric_batch_size = 500 metric_buffer_limit = 10000 omit_hostname = true [[inputs.kafka_consumer]] brokers = ["kafka:29092"] topics = ["host-metrics-v1"] consumer_group = "prometheus-host-metrics-v1" kafka_version = "3.6.0" offset = "newest" max_undelivered_messages = 1000 data_format = "json_v2" [[inputs.kafka_consumer.json_v2]] measurement_name_path = "name" timestamp_path = "timestamp" timestamp_format = "unix_ms" timestamp_timezone = "UTC" [[inputs.kafka_consumer.json_v2.object]] path = "fields" ''' for tag in TAGS: bridge += f''' [[inputs.kafka_consumer.json_v2.tag]] path = "tags.{tag}" rename = "{tag}" optional = true ''' bridge += ''' [[outputs.prometheus_client]] listen = ":9273" metric_version = 1 expiration_interval = "60s" string_as_label = false export_timestamp = true collectors_exclude = ["gocollector", "process"] ''' (ROOT / 'configs/metrics-bridge/telegraf.conf').write_text(bridge) print(f'Generated {len(OBJECTS)} saved objects, 4 dashboards, mappings, retention policy and metrics bridge.')