top × psutil × systemctl で徹底監視！サーバー上の全プロセス情報をInfluxDBに一括記録する

# -*- coding: utf-8 -*-
__group__ = '情報記録'           # Grafanaでのグループ名
__description__ = 'センサーデータを記録'  # Grafanaでの説明

# ... 以降に通常の処理 ...

pip install psutil

top -b -n 1 | head -20

RECORD_TARGET_COMMANDS = ['python', 'influxd', 'telegraf', 'grafana', 'nginx', 'dnsmasq']

import datetime

def parse_cputime(s: str) -> datetime.timedelta:
    """topのTIME+をtimedeltaに変換（複数フォーマット対応）"""
    if s is None:
        raise ValueError("Empty time string")
    s = s.strip().replace(",", ".")
    if not s:
        raise ValueError("Empty time string")

    # H:MM:SS（2つのコロン）
    if s.count(":") == 2:
        h, m, sec = s.split(":")
        return datetime.timedelta(hours=int(h), minutes=int(m), seconds=int(float(sec)))

    # M:SS(.ms)（1つのコロン）
    if s.count(":") == 1:
        minutes, sec_part = s.split(":", 1)
        if "." in sec_part:
            seconds_str, frac = sec_part.split(".", 1)
            ms = int(round(float("0." + frac) * 1000))
            return datetime.timedelta(minutes=int(minutes), seconds=int(seconds_str), milliseconds=ms)
        else:
            return datetime.timedelta(minutes=int(minutes), seconds=int(sec_part))

    # 小数秒（例: "6.17"）
    if "." in s:
        total_seconds = float(s)
        seconds = int(total_seconds)
        ms = int(round((total_seconds - seconds) * 1000))
        return datetime.timedelta(seconds=seconds, milliseconds=ms)

    # 整数秒（例: "123"）
    if s.isdigit():
        return datetime.timedelta(seconds=int(s))

    raise ValueError(f"Unknown time format: {s}")


def parse_mem_value(value: str) -> int:
    """topのVIRT/RES/SHR表記をバイトに変換（単位なしはKB扱い）"""
    value = value.strip().lower()
    if value[-1].isdigit():
        return int(float(value) * 1024)    # 単位なし → KB → バイト
    num = float(value[:-1])
    unit = value[-1]
    if unit == 'g':
        return int(num * 1024 ** 3)
    elif unit == 'm':
        return int(num * 1024 ** 2)
    elif unit == 'k':
        return int(num * 1024)
    else:
        raise ValueError(f"Unknown memory unit: {value}")

import subprocess
import re

def parse_top_output():
    """top の1回分の出力をパースしてプロセスリストを返す"""
    top_output = subprocess.check_output(["top", "-b", "-n", "1"]).decode()
    lines = top_output.splitlines()

    processes = []
    found = False
    for line in lines:
        if re.match(r"\s*PID\s+USER", line):
            found = True
            continue
        if not found:
            continue

        cols = re.split(r"\s+", line.strip(), maxsplit=11)
        if len(cols) < 12:
            continue

        pid = int(cols[0])
        processes.append({
            "pid": pid,
            "user": cols[1],
            "pr": cols[2],
            "ni": cols[3],
            "virt": parse_mem_value(cols[4]),
            "res": parse_mem_value(cols[5]),
            "shr": parse_mem_value(cols[6]),
            "state": cols[7],
            "cpu": float(cols[8]),
            "mem": float(cols[9]),
            "time": parse_cputime(cols[10]),
            "command": cols[11] if len(cols) > 11 else None
        })
    return processes

def get_service_map():
    """systemdのサービス情報を一括取得してPID→サービス情報の辞書を返す"""
    cmd = [
        "systemctl", "show", "--type=service",
        "-p", "Id", "-p", "MainPID", "-p", "LoadState",
        "-p", "ActiveState", "-p", "SubState", "-p", "Description"
    ]
    result = subprocess.run(cmd, capture_output=True, text=True)
    lines = result.stdout.strip().splitlines()

    pid_info_map = {}
    current_info = {}
    pid = None

    for line in lines:
        if not line.strip() or "=" not in line:
            continue
        key, value = line.split("=", 1)
        current_info[key] = value

        if key == 'MainPID':
            pid = int(value)
        elif key == "SubState":
            if "Id" in current_info and pid is not None:
                info_copy = current_info.copy()
                info_copy["ServiceName"] = info_copy.pop("Id")
                pid_info_map[pid] = info_copy
            current_info = {}
            pid = None

    return pid_info_map

import os
import psutil
import common
from typing import Optional

RECORD_TARGET_COMMANDS = ['python', 'influxd', 'telegraf', 'grafana', 'nginx', 'dnsmasq']

def build_process_info():
    top_list = parse_top_output()
    service_map = get_service_map()
    top_dict = {p["pid"]: p for p in top_list}

    tags_list = []
    fields_list = []

    for pid, info in top_dict.items():
        command = info["command"]
        if command not in RECORD_TARGET_COMMANDS:
            continue

        try:
            p = psutil.Process(pid)
        except:
            continue

        # 子プロセスを取得
        try:
            children = p.children(recursive=True)
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            children = []

        # 親子判定
        try:
            ppid = p.ppid()
        except:
            ppid = None
        is_parent = (len(children) > 0)
        is_child  = (ppid != 1 and len(children) == 0)

        # 子プロセスのリソースを親に合算
        total_cpu  = info["cpu"]
        total_mem  = info["mem"]
        total_virt = info["virt"]
        total_res  = info["res"]
        total_shr  = info["shr"]
        child_pids = [c.pid for c in children]

        for child in children:
            try:
                cpid = child.pid
                if cpid in top_dict:
                    total_virt += top_dict[cpid]["virt"]
                    total_res  += top_dict[cpid]["res"]
                    total_shr  += top_dict[cpid]["shr"]
                    total_cpu  += top_dict[cpid]["cpu"]
                    total_mem  += top_dict[cpid]["mem"]
            except psutil.NoSuchProcess:
                continue

        # CPU正規化（コア数で割って全体100%基準に）
        cpu_cores = psutil.cpu_count()
        normalized_cpu = total_cpu / cpu_cores

        # コマンドライン・プログラムパスの取得（Pythonスクリプト判定）
        try:
            cmdline = p.cmdline()
            application_path = cmdline[0] if cmdline else None
            program_path = None
            if len(cmdline) > 2 and cmdline[1] == "-E":
                program_path = cmdline[2]
            elif len(cmdline) > 1:
                program_path = cmdline[1]
            program_name = os.path.basename(program_path) if program_path else None
            # プログラムのメタ情報取得（自作ライブラリのget_program_metadata関数）
            program_description, program_group = common.get_program_metadata(program_path)
        except:
            application_path = program_path = program_name = None
            program_description = program_group = None

        # Python以外はコマンド名をプログラム名として使用
        if 'python' not in command:
            program_name = command

        # systemdサービス情報の取得（service_mapで高速引き、なければ個別取得）
        if pid in service_map:
            svc = service_map[pid]
            service_name        = svc['ServiceName']
            service_loadstate   = svc['LoadState']
            service_activestate = svc['ActiveState']
            service_substate    = svc['SubState']
            service_description = svc['Description']
        else:
            service_info = get_service_info_from_pid(pid)
            if service_info:
                service_name        = service_info['ServiceName']
                service_loadstate   = service_info['LoadState']
                service_activestate = service_info['ActiveState']
                service_substate    = service_info['SubState']
                service_description = service_info['Description']
            else:
                service_name = service_loadstate = service_activestate = None
                service_substate = service_description = None

        if not program_description:
            program_description = service_name
            program_group       = service_name

        tags_list.append({
            "user":                info["user"],
            "command":             command,
            "program_name":        program_name,
            "program_description": program_description,
            "program_group":       program_group,
            "service_name":        service_name
        })

        fields_list.append({
            "pid":                  pid,
            "priority":             info["pr"],
            "nice_value":           info["ni"],
            "state":                info["state"],
            "children_pids":        ",".join(map(str, child_pids)),
            "parent_pid":           ppid,
            "is_parent":            is_parent,
            "is_child":             is_child,
            "application_path":     application_path,
            "program_path":         program_path,
            "service_loadstate":    service_loadstate,
            "service_activestate":  service_activestate,
            "service_substate":     service_substate,
            "service_description":  service_description,
            "virtual_memory":       total_virt,
            "resident_memory":      total_res,
            "shared_memory":        total_shr,
            "cpu_percent_raw":      total_cpu,
            "cpu_percent":          normalized_cpu,
            "memory_percent":       total_mem,
            "cpu_time_total":       info["time"].total_seconds()
        })

    return tags_list, fields_list

import time
from influx.v2 import writer as influxwriter
import smarthomelog

influx = influxwriter.Writer()
success_log = smarthomelog.Log(__file__, smarthomelog.LogType.SUCCESS)
cpu_cores = psutil.cpu_count()

while True:
    tags, fields = build_process_info()
    influx.insert_process_status_log(tags, fields)
    success_log.write('処理が正常に終了しました。', __file__)
    time.sleep(5)

def insert_process_status_log(self, tags: List[Dict], fields: List[Dict]):
    measurement = 'process_status_log'
    try:
        # タグリスト・フィールドリストを一括でInfluxDBに書き込む
        self.conn.write_dict_batch(measurement, tags, fields)
    except Exception as e:
        print(e)

if re.match(r"\s*PID\s+USER", line):
    found = True
    continue

cols = re.split(r"\s+", line.strip(), maxsplit=11)

elif key == "SubState":
    if "Id" in current_info and pid is not None:
        info_copy = current_info.copy()
        info_copy["ServiceName"] = info_copy.pop("Id")
        pid_info_map[pid] = info_copy
    current_info = {}
    pid = None

[Unit]
Description=Process Status Record to InfluxDB
After=network.target influxd.service

[Service]
ExecStart=/usr/bin/python3 /home/<username>/projects/smarthome/core/record/raspi/process_status_record.py
Restart=always
RestartSec=10
User=<username>

[Install]
WantedBy=multi-user.target

sudo cp process_status_record.service /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl enable process_status_record
sudo systemctl start process_status_record

python3 process_status_record.py

from(bucket: "your-bucket")
  |> range(start: -1m)
  |> filter(fn: (r) => r._measurement == "process_status_log")
  |> filter(fn: (r) => r._field == "cpu_percent")