【Python】Paramikoを使ってssh接続でサーバーのリソース監視をやってみよう

  • 2024/8/22
  • ブログ
  • , ,
  • 【Python】Paramikoを使ってssh接続でサーバーのリソース監視をやってみよう はコメントを受け付けていません

はじめに

今回はPythonのSSH接続ライブラリであるParamikoを使ってサーバーの簡易的な死活監視プログラムを作ってみます。監視する項目としては

  • RAM使用率
  • ストレージ容量

について監視を行うプログラムをPythonで作成してみます。

環境構築

Dockerを用いてSSH serverとSSH clientのコンテナを作成して開発を行います。

VMや外部のレンタルサーバーなどの選択肢もありますが、ローカルでサクッとテストしたいのと、Dockerのネットワーク周りの勉強がてらこの方法でいきます。

アーキテクチャ

各コンテナにIPアドレスを付与するためにDockerのブリッジネットワークという機能を使います。この機能により、Dockerホスト内で独立したネットワークを構築でき、同じブリッジネットワークに接続されたコンテナ間で通信が可能になります。各コンテナにはプライベートなIPアドレスが割り当てられます。

今回の設定では、サブネット10.10.0.0/24を持つssh_explorer_test_networkというブリッジネットワークに以下の二つのコンテナを接続します。

  1. ssh-serverコンテナ
    • IPアドレス、10.10.0.5を付与。
    • 死活監視対象のサーバーを模したコンテナ。ユーザー名はserver、パスワードはpasswordでssh接続が可能。
    • 図には書いてませんが、22番ポートをブリッジネットワーク間でのみ開放しています
  2. ssh-clientコンテナ
    • IPアドレス、10.10.0.6を付与。
    • 死活監視プログラムを実行するためのコンテナ。

docker-compose.ymlとその他

インフラの説明がそのまま反映されているだけなので、ここはサラッといきます。

ディレクトリ構造

ssh_explorer
├── docker-compose.yml
├── ssh-client
│   ├── Dockerfile
│   ├── requirements.txt
│   └── src
│       ├── main.py
│ 	   ├── get_ram_status.sh
│       └── get_strage_status.sh
└── ssh-server
    └── Dockerfile

docker-compose.yml

version: '3.8'

services:
  ssh-server:
    image: ssh-server
    build: ./ssh-server
    tty: true
    networks:
      ssh_explorer_test_network:
        ipv4_address: 10.10.0.5

  ssh-client:
    image: ssh-client
    build: ./ssh-client
    tty: true
    volumes:
      - ./ssh-client/src:/root/src
    networks:
      ssh_explorer_test_network:
        ipv4_address: 10.10.0.6

networks:
  ssh_explorer_test_network:
    driver: bridge
    ipam:
      config:
        - subnet: 10.10.0.0/24

ssh-clientのDockerfile

# 基本イメージ
FROM python:3.11.9-bookworm

RUN apt-get update && \
    apt-get upgrade -y &&\
    apt-get install -y openssh-client python3-pip iputils-ping
 
WORKDIR /root

COPY ./requirements.txt /root
RUN pip install --upgrade pip
RUN pip install -r requirements.txt

sshおよび、paramikoを実行するためのパッケージがインストールされる構造になっています。iputils-ping はpingを実行するためのパッケージです。デフォルトのベースイメージにはインストールされていなかったので、別途インストールしています。
requirements.txtはparamikoしか記述されていないので省略。

ssh-serverのDockerfile

# 基本イメージ
FROM ubuntu:22.04

# 必要なパッケージのインストール
RUN apt-get update && apt-get install -y openssh-server sudo

# SSH用のユーザーを追加
RUN useradd -rm -d /home/ubuntu -s /bin/bash -g root -G sudo -u 1000 server 
RUN  echo 'server:password' | chpasswd

# SSH接続用の設定
RUN mkdir /var/run/sshd
RUN sed -i 's/#PermitRootLogin prohibit-password/PermitRootLogin yes/' /etc/ssh/sshd_config
# パスワード認証を有効にする
RUN sed -i 's/#PasswordAuthentication yes/PasswordAuthentication yes/' /etc/ssh/sshd_config

# SSH接続のためのポートを開放
EXPOSE 22

# SSHサーバーを起動
CMD ["/usr/sbin/sshd", "-D"]

ssh-serverをインストールし、指定のユーザー名とパスワード（server:password） を自動で設定できる仕組みになっています。

ChatGPTの出力を見て、「sedってそうやって使うのか、、、、」と勉強になりました。

これにてコンテナのセットアップは完了です。

sshおよび監視を行うためのPythonコード 

ここからpythonコードを用いてsshを行い、レスポンスを受け取るコードを書きます。

受け取る出力はシェルスクリプトを一工夫して、JSON形式で出力できるようにします。

ソースコードのディレクトリ構造

root
    └ src
        ├── main.py
        ├── get_ram_status.sh
        └── get_strage_status.sh

main.py

import paramiko
import ast
from pprint import pprint

all_shells_info = [
    {"value_name": "ram_status","file_name": "get_ram_status.sh"},
    {"value_name": "strage_status","file_name": "get_strage_status.sh"},
]

def join_ssh_commands(all_shells_info: list):
    command = "echo '['"
    for shell in all_shells_info:
        value_name = shell["value_name"]
        file_name = shell["file_name"]
        command += \
            "&& echo " + \
            "\"{'name': " + "'" +value_name+ "'" + \
            "," + " 'value': \" " + \
            "&& "
        
        with open(file_name, "r", encoding='utf-8') as file:
            for line in file:
                if (not line[0] == "#") and  (not line == ""):
                    command += line
        command += "&& echo  \"},\" "
    command += "&& echo ']'"
    return command.replace("\n","")

def ssh_command(hostname, port, username, password, command):
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(hostname, port=port, username=username, password=password)

    # コマンドの実行
    stdin, stdout, stderr = client.exec_command(command)
    
    # 出力結果の取得
    output = stdout.read().decode('utf-8')
    error = stderr.read().decode('utf-8')

    # 接続のクローズ
    client.close()

    return output, error

def main():
    command = join_ssh_commands(all_shells_info)
    output_str, error_str = ssh_command("10.10.0.5", 22, "server" ,"password", command)
    output = ast.literal_eval(output_str)
    pprint(output)
    
if __name__ == "__main__":
    main()

get_ram_status.sh

#!/bin/bash

free -m | awk 'NR==2 {print "{\"total(MB)\": " $2 ", \"used(MB)\": " $3 ", \"free(MB)\": " $4 ", \"shared(MB)\": " $5 ", \"buff/cache(MB)\": " $6 ", \"available(MB)\": " $7 "}"}'

get_strage_status.sh

#!/bin/bash

echo "[" && df -BM | awk 'NR>1 {print "{\"Filesystem\": \"" $1 "\", \"Size(MB)\": " substr($2, 1, length($2)-1) ", \"Used(MB)\": " substr($3, 1, length($3)-1) ", \"Available(MB)\": " substr($4, 1, length($4)-1) ", \"Use(%)\": " substr($5, 1, length($5)-1) ", \"MountedOn\": \"" $6 "\"},"}' | sed ':a;N;$!ba;s/\\n/, /g' && echo "]"

解説

上記のmain.pyを実行すると以下の結果が得られます。

json形式で取得できるのがこのコードの推しポイントです。

[
    {
        'name': 'ram_status',
        'value': {
            'available(MB)': 2262,
            'buff/cache(MB)': 2333,
            'free(MB)': 277,
            'shared(MB)': 113,
            'total(MB)': 3933,
            'used(MB)': 1322
        }
    },{
        'name': 'strage_status',
        'value': [
            {
                'Available(MB)': 31482,
                'Filesystem': 'overlay',
                'MountedOn': '/',
                'Size(MB)': 59768,
                'Use(%)': 45,
                'Used(MB)': 25219
            },{
                'Available(MB)': 64,
                'Filesystem': 'tmpfs',
                'MountedOn': '/dev',
                'Size(MB)': 64,
                'Use(%)': 0,
                'Used(MB)': 0
            }
        ]
    }
]

main.pyでssh接続とサーバー上でのデータの収集をおこないます。

基本的な動作は以下の流れになっております。

１：シェルスクリプト（ get_ram_status.sh、get_strage_status.sh）を読み込む。

２：シェルスクリプトを&&で連結して、１度の接続で同時に実行できるコマンドとして読み込む

３：連結したコマンドを実行する&出力を得る

ここから詳しく解説していきます。

まずライブラリを読み込みます。

import paramiko
import ast
from pprint import pprint

今回はsshにはparamiko、取得した文字列をPythonの値として読み込むライブラリとしてastを使います。

ssh時に実行するシェルスクリプトを辞書型で定義します。

ここに記載されているシェルスクリプトを連結して実行します。

all_shells_info = [
    {"value_name": "ram_status","file_name": "get_ram_status.sh"},
    {"value_name": "strage_status","file_name": "get_strage_status.sh"},
]

シェルスクリプトを連結する関数です。

echoで[]を出力して、pythonのリストになるように出力を整形しています。

また&&と改行を消す処理を使って、最終的にシェルスクリプトがワンライナーになるように整形します。

def join_ssh_commands(all_shells_info: list):
    command = "echo '['"
    for shell in all_shells_info:
        value_name = shell["value_name"]
        file_name = shell["file_name"]
        command += \
            "&& echo " + \
            "\"{'name': " + "'" +value_name+ "'" + \
            "," + " 'value': \" " + \
            "&& "
        
        with open(file_name, "r", encoding='utf-8') as file:
            for line in file:
                if (not line[0] == "#") and  (not line == ""):
                    command += line
        command += "&& echo  \"},\" "
    command += "&& echo ']'"
    return command.replace("\n","")

実際にssh接続を行いコマンドを実行する関数です。

注意点としてはparamiko.AutoAddPolicy()の部分です。ここでは接続先のホストが未知のホストであっても自動的に接続を許可する設定を行なっています。

def ssh_command(hostname, port, username, password, command):
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(hostname, port=port, username=username, password=password)

    # コマンドの実行
    stdin, stdout, stderr = client.exec_command(command)
    
    # 出力結果の取得
    output = stdout.read().decode('utf-8')
    error = stderr.read().decode('utf-8')

    # 接続のクローズ
    client.close()

    return output, error

最後にmain.pyで全ての関数を実行します。繰り返しにはなりますが、以下の流れを体現できるような流れになっています。

１：シェルスクリプト（ get_ram_status.sh、get_strage_status.sh）を読み込む。

２：シェルスクリプトを&&で連結して、１度の接続で同時に実行できるコマンドとして読み込む

３：連結したコマンドを実行する&出力を得る

def main():
    command = join_ssh_commands(all_shells_info)
    output_str, error_str = ssh_command("10.10.0.5", 22, "server" ,"password", command)
    output = ast.literal_eval(output_str)
    pprint(output)
    
if __name__ == "__main__":
    main()

まとめ

今回は簡単ですが、sshを使った簡単なリソース監視をしてみました。

今ではあまりない状況かもしれませんが、どうしてもrestAPIを立てられない監視側に、何も追加せずに監視だけしたいケースなどで役にたつ知識だと思います。

それでは良いエンジニアライフを！！！

＜類似記事＞

  1. Automating Docker Image Deployment to AWS ECR using GitHub Actions
  2. Easy data validation with Yup
  3. Making a simple game with Rust
  4. Essential Golang Knowledge for Beginners
  5. Building a Simple Web Server in Rust Using Actix-web

関連記事

トップページに戻る

メールマガジンで一週間のおまとめニュースを配信中！登録無料、登録はコチラから

カテゴリー：

ブログ

SHARE

情シス求人

  1. チームメンバーで作字やってみた#1
ページ上部へ戻る