[fabric] fabirc 사용한 업무 자동화 구축

1. 개요

프로세스를 개발하고, 서버에 배포한 뒤 정상적으로 동작하는지 확인이 필요하다. 나는 매일 로컬환경에서 다수의 배포서버에 SSH로 접속해서 로그파일들을 확인해야 했다.

 

반복 작업을 줄이기 위해 자동화를 구축 하기로 했다.  

 

 

2. 설계 및 사용 기술

설계한 내용은 다음과 같다.

 

1. 프로그램이 다수의 서버에 ssh로 접속한다.

2. 내가 지정해놓은 shell 명령어를 각각 실행한 다음, 결과를 가져온다.   

3. 출근하면 자동으로 1, 2번 동작을 실행할 수 있게 스케줄러를 사용하고, 결과를 파일로 저장하고, 메일로 전송하도록 한다.

 

 

이해를 돕기 위해서 내가 처한 서버 환경에 대해서 설명하겠다.

개발 서버는 로컬에서 직접 접속이 가능하다. 하지만 운영서버는 보안상의 이유로 로컬에서 직접 접속을 할 수 없다. 운영서버는 개발서버를 통해 SSH로 접속해야 한다.

 

사용한 기술은 다음과 같다.

• Python 3.12.0
• SMTP (메일)
• fabric (SSH)
• schedule (스케줄러)
• PyYAML (외부 프로퍼티)

 

3. fabric 이란?

이번 자동화 구축에 사용되는 핵심 라이브러리는 fabric이다. 

 

• fabric은 ssh 연결을 통해 원격으로 shell 명령어를 실행하여 python 개체를 생성하는 라이브러리이다.
• shell 명령어뿐만 아니라 파일 전송도 가능해서 배포 자동화 등에도 활용할 수 있다.

fabric 참고 문서

https://www.fabfile.org/upgrading.html

Upgrading from 1.x — Fabric documentation

 

 

 

4. 주요 코드

라이브러리 설치

• python 3.12.0 기준으로 smtp는 내장되어 있다.

pip install fabric
pip install schedule
pip install PyYAML

 

 

application.yml(설정 파일) 읽어오기

• 변경이 빈번하게 일어나는 설정들은 외부 프로퍼티에서 읽어어도록 하였다.
• PyYAML을 사용하여 application.yml에서 세팅해놓은 설정정보를 읽어오도록 하는 코드이다. 

import yaml

application_yml_path = './conf/application.yml'

def read_ssh_targets():
    with open(application_yml_path, 'r', encoding=encoding) as file:
        config_data = yaml.safe_load(file)

    ssh_data = config_data.get('ssh')
    targets = ssh_data.get('targets', [])

    return targets

 

fabric 라이브러리를 사용하여 ssh 연결 및 명령어 처리

• fabric을 사용하여 ssh 연결 및 전달받은 shell 명령어를 실행하는 클래스이다.
• 운영서버는 직접 접속이 안되기 때문에 시행착오를 겪었다. fabric에서 Connection을 생성할 때 gateway를 제공해줘서 이 문제를 해결할 수 있었다. 
• 운영서버는 개발서버를 통해 접속이 가능하기 때문에 gateway를 프로퍼티에 세팅해주었다. gateway가 있는 서버라면 게이트웨이를 통해 연결을 하도록 설정했다.
• run() 메서드는 ssh 연결을 가져와 전달받은 명령어를 실행하는 메서드이다. 
• 예를 df -h(디스크 사용량 체크) 등 쉘 명령어를 실행한 후 결과는 stdout에 담겨있다. 

from socket import error as socket_error
from fabric import Connection
from invoke import CommandTimedOut
from paramiko.ssh_exception import AuthenticationException

class Host:
    def __init__(self, ip, port, username, password, encoding, gateway=None):
        self.ip = ip
        self.port = port
        self.username = username
        self.password = password
        self.encoding = encoding
        self.gateway = gateway

    def get_connection(self):
        if self.gateway:
            return Connection(host=self.ip, port=self.port, user=self.username,
                              connect_kwargs={"password": self.password}, gateway=self.gateway.get_connection())
        else:
            return Connection(host=self.ip, port=self.port, user=self.username,
                              connect_kwargs={"password": self.password})

    def run(self, command):
        try:
            with self.get_connection() as connection:
                logger.info(f'Running {command} on {self.ip}')
                result = connection.run(command, warn=True, hide=True, encoding=self.encoding, timeout=run_timeout)
                return result.stdout
        except (socket_error, AuthenticationException, CommandTimedOut) as exc:
            logger.info(f"SSH: run failed. {self.ip} by {self.username} : {exc}")
            return 'error'
            
            
 # 서버 정보
ssh:
  targets:
    - server_name: 개발서버
      ip: 개발서버 ip
      port: 개발서버 ssh 포트
      username: 개발서버 계정
      password: 개발서버 PW
      encoding: utf-8

    - server_name: 운영서버
      ip: 운영서버 ip
      port: 운영서버 ssh 포트
      username: 운영서버 계정
      password: 운영서버 PW
      encoding: utf-8
      gateway:
        ip: 개발서버 ip
        port: 개발서버 ssh 포트
        username: 개발서버 계정
        password: 개발서버 PW
        encoding: utf-8
        
# 점검해야 할 목록(shell 명령어)
ssh_monitor_list:
  - process_name: hardware_check
    cmd: free -h   # 메모리 사용률 명령어

  - process_name: hardware_check
    cmd: df -h    # 디스크 사용률 명령어

 

SMTP 메일 전송

• shell 명령어들의 결과인 report가 담긴 리스트를 전달받아, 메일로 전송하는 부분이다.
• SMTP 서버는 daum 서버를 사용하였다. 
• 제목, 발신자, 메일 본문을 설정하고, 모든 수신자들한테 전송하도록 구현하였다.

from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from smtplib import SMTP_SSL
from datetime import datetime
from util import fileread

def send_mail(reports: list):
    try:
        server, login_id, login_pw, send_email = fileread.read_smtp_info()

        conn = SMTP_SSL(server)
        conn.ehlo()
        conn.login(login_id, login_pw)

        title_email = server_name + '_report' + datetime.now().strftime("%Y%m%d")

        msg = MIMEMultipart()
        msg['Subject'] = title_email
        msg['From'] = send_email

        # 메일 본문
        for report in reports:
            content = MIMEText(report.report, 'plain', 'utf-8')
            msg.attach(content)

        email_recv_list = fileread.read_email_recv_list()

        # 이메일 수신자
        for recipient in email_recv_list:
            msg['To'] = recipient
            conn.sendmail(send_email, recipient, msg.as_string())
            logger.info(f"Success send to {recipient} emails")

        conn.close()

    except Exception as e:
        logger.error("Failed to send mail. error: ", e)
        
       
# smtp
smtp:
  server: smtp.daum.net:465
  login_id: 로그인 ID
  login_pw: 로그인 PW
  send_email: 발신자 이메일
  email_recv_list:
    - 수신자 이메일1
    - 수신자 이메일2

 

파일 저장

• ssh 명령어 결과를 파일로 저장하는 코드이다. 
• file 이름 형식을 지정한 다음 파일을 열어서 shell 결과를 파일에 write 하도록 구현하였다. 

from util import fileread
import os
from datetime import datetime


def file_save(reports: list):
    save_path = fileread.read_save_path()

    if not os.path.isdir(save_path):
        os.makedirs(save_path)

    file_path = save_path + '/' + reports[0].target_name + '_report_' + datetime.now().strftime("%Y%m%d") + '.log'

    with open(file_path, 'w', encoding='utf-8') as report_file:
        for report in reports:
            report_file.write(report.report)

 

스케줄러 설정

• application.yml에서 스케줄러 정보를 읽어서 스케줄러를 등록하는 코드이다.
• 크론식이나 다른 방식으로 설정할 수 있는데, 누가봐도 이해할 수 있도록 스케줄러를 구성해보고 싶었다. (오히려 불편하긴 했음..)

def main():
    days, time_str = fileread.read_schedule_day_and_time()

    # [monday, tuesday, wednesday, thursday, friday]
    for day in days:
        if day == 'monday':
            schedule.every().monday.at(time_str).do(process)
        elif day == 'tuesday':
            schedule.every().tuesday.at(time_str).do(process)
        elif day == 'wednesday':
            schedule.every().wednesday.at(time_str).do(process)
        elif day == 'thursday':
            schedule.every().thursday.at(time_str).do(process)
        elif day == 'friday':
            schedule.every().friday.at(time_str).do(process)
        elif day == 'saturday':
            schedule.every().saturday.at(time_str).do(process)
        elif day == 'sunday':
            schedule.every().sunday.at(time_str).do(process)

    while True:
        schedule.run_pending()  # 현재시간에 처리할 스케줄러를 모두 처리
        time.sleep(1)
        
# 스케줄
schedule:
  day: # monday tuesday wednesday thursday friday saturday sunday
    - monday
    - tuesday
    - wednesday
    - thursday
    - friday
  time: '09:00'       # 00:00 ~ 23:59

 

 

5. 마무리

위에서는 2개의 서버만 점검하는 예시를 들었지만, 원래는 5개 정도의 서버에 접속해서 프로세스 상태를 확인해야 했다. 이번 자동화를 구축한 뒤로 출근만 하면 메일로 확인만 하면 되기 때문에 업무 효율성은 매우 좋아졌다. 

 

그리고, 파이썬으로 개발 하는 것은 처음이라 쉽지는 않았다. 그래도 파이썬 기본적인 문법과 파이썬 환경, 라이브러리 관리 시스템인 pip에 대해서 이해할 수 있었다. 무엇보다 java에 비해 가벼운 프로그램을 만들기 편리했다. 

 

추후 사용하면서 불편하고, 개선할 점들을 찾아서 수정해볼 예정이다.