Dump environment variables, system information and other useful information

dump.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Dump environment variables, system information and other useful information"""

import os
import contextlib
import platform

import psutil
import requests


def dump(request=None):
    """Dump environment variables, system information and other useful information"""
    self_process = psutil.Process()
    with self_process.oneshot():
        ps_util_ = {
            # cpu
            'cpu_times': psutil.cpu_times()._asdict(),
            'cpu_times.per_cpu': [
                cpu._asdict() for cpu in psutil.cpu_times(percpu=True)
            ],
            'cpu_count': psutil.cpu_count(),
            'cpu_count.logical': psutil.cpu_count(logical=True),
            'cpu_count.physical': psutil.cpu_count(logical=False),
            'cpu_stats': psutil.cpu_stats()._asdict(),
            # memory
            'virtual_memory': psutil.virtual_memory()._asdict(),
            'swap_memory': psutil.swap_memory()._asdict(),
            # disk
            'disk_usage': psutil.disk_usage('/')._asdict(),
            # network
            'net_io_counters': psutil.net_io_counters()._asdict(),
            'net_io_counters.per_nic': [
                nic._asdict() for nic in psutil.net_io_counters(pernic=True).values()
            ],
            'net_connections': [c._asdict() for c in psutil.net_connections()],
            'net_if_addrs': {
                nic: [addr._asdict() for addr in addrs]
                for nic, addrs in psutil.net_if_addrs().items()
            },
            'net_if_stats': {
                nic: stats._asdict() for nic, stats in psutil.net_if_stats().items()
            },
            # system
            'users': [user._asdict() for user in psutil.users()],
            'boot_time': psutil.boot_time(),
            'pids': psutil.pids(),
            # constants
            'constants': {
                'POSIX': psutil.POSIX,
                'WINDOWS': psutil.WINDOWS,
                'LINUX': psutil.LINUX,
                'FREEBSD': psutil.FREEBSD,
                'OPENBSD': psutil.OPENBSD,
                'NETBSD': psutil.NETBSD,
                'SUNOS': psutil.SUNOS,
                'OSX': psutil.OSX,
                'version_info': psutil.version_info,
                'version': psutil.__version__,
            },
        }
        processes = {}
        self_ = self_process.as_dict()
        self_.update(
            {
                'memory_full_info': self_process.memory_full_info()._asdict(),
                'connections': [c._asdict() for c in self_process.connections()],
                'threads': [t._asdict() for t in self_process.threads()],
            }
        )
        if 'memory_maps' in self_:
            del self_['memory_maps']
        ad_value = "<psutil.AccessDenied>"
        processes['self'] = self_
        for process in psutil.process_iter():
            with contextlib.suppress(psutil.NoSuchProcess), process.oneshot():
                process_dict = process.as_dict(attrs=list(self_.keys()), ad_value=ad_value)
                if process_dict['memory_full_info'] != ad_value:
                    process_dict['memory_full_info'] = process.memory_full_info()._asdict()
                if process_dict['connections'] != ad_value:
                    process_dict['connections'] = [c._asdict() for c in process.connections()]
                if process_dict['threads'] != ad_value:
                    process_dict['threads'] = [t._asdict() for t in process.threads()]
                processes[str(process.pid)] = process_dict

        ps_util_['processes.self'] = self_
        ps_util_['processes'] = processes
    platform_ = {
        'platform': platform.platform(),
        'system': platform.system(),
        'machine': platform.machine(),
        'processor': platform.processor(),
        'node': platform.node(),
        'release': platform.release(),
        'version': platform.version(),
        'uname': platform.uname()._asdict(),
        'python_version': platform.python_version(),
        'python_implementation': platform.python_implementation(),
        'python_compiler': platform.python_compiler(),
        'python_build': platform.python_build(),
        'python_branch': platform.python_branch(),
        'python_revision': platform.python_revision(),
        'python_version_tuple': platform.python_version_tuple(),
    }

    if os.environ.get('VIRTUAL_ENV'):
        platform_['virtual_env'] = os.environ.get('VIRTUAL_ENV')
    ecs_task_stats = {}
    if os.environ.get('ECS_CONTAINER_METADATA_URI_V4'):
        url = os.environ.get('ECS_CONTAINER_METADATA_URI_V4') + '/task/stats'
        try:
            resp = requests.get(url, timeout=1)
            if resp.status_code == 200:
                ecs_task_stats = resp.json()
        except requests.RequestException:
            pass

    request_ = {}
    if request:
        request_ = {
            'url': request.url,
            'environ': request.environ,
            'identity': request.identity,
            'headers': dict(request.headers),
            'host': request.host,
            'host_port': request.host_port,
            'host_url': request.host_url,
            'http_version': request.http_version,
            'method': request.method,
            'query_string': request.query_string,
            'remote_addr': request.remote_addr,
            'remote_user': request.remote_user,
            'scheme': request.scheme,
            'name': request.name,
            'server_name': request.server_name,
            'server_port': request.server_port,
            'path': request.path,
        }

    return {
        'request': request_,
        'os.environ': dict(os.environ),
        'psutil': ps_util_,
        'platform': platform_,
        'ecs_task_stats': ecs_task_stats,
    }