JbDesk 3.2편-fabric 으로 로그 메시지 추적

pyhton fabric 을 사용하여 원격 서버의 로그를 추적할 수 있습니다.

예를 들면 api 요청이 들어올 때 다음과 같이 여러 단계를 거치면서 처리가  gateway → api → db 됩니다.

TID(Trace ID) 가 존재하는 경우 각각 서비스에서 TID 로 검색하면 됩니다.

그런데 gateway, api, db 에서 로그를 검색하게 되면 3군데를 모두 검색해야 합니다.

그런데 서비스가 좀더 복잡하여 여러군데를 거치면 추적은 더욱 힘들어집니다.

fabric 을 사용하면 이러한 것을 자동으로 할 수  있습니다.

1) 테스트 환경

우선 코드를 만들 때 여러 단계를 거치는 테스트 환경을 만들어야 합니다.

vagrant 를 사용하야 해당 환경을 만들어보겠습니다.

ChatGPT  에 사용한 프롬프트입니다.

vagrant 로 다음과 같이 구축해주세요.
api 요청이 gateway01 로 오면 api01 의 api 를 호출하고 api01 은 echo01 를 호출합니다.
echo01 은 들어온 요청에 대해 echo message 를 리턴합니다.
python flask 로 구축해주세요.
TID(Trace ID) 도 전달하여 api 호출이 왔을 때 file 로 log 메시지도 출력해주세요.

1. host_name : gateway01 
ip : 192.168.56.40
os : rocky9

2. host_name : api01
ip : 192.168.56.40
os : rocky9

3. host_name : echo01
ip : 192.168.56.40
os : rocky9

 

Flask 기반의 gateway → api → echo 구조를 Vagrant + Rocky 9으로 구성하고, **TID(Trace ID)**를 HTTP 헤더로 전달하여 파일로 로그를 저장하는 구조로 만들어보겠습니다.

---

✅ 시스템 요약

구성 요소	역할	IP	포트
gateway01	외부 요청 수신 및 전달	192.168.56.40	5000
api01	비즈니스 로직 처리 및 전달	192.168.56.41	5001
echo01	메시지 반환 및 로그 기록	192.168.56.42	5002

---

1️⃣ Vagrantfile 설정

Vagrant.configure("2") do |config|
  config.vm.box = "generic/rocky9"

  def setup_vm(vm, hostname, ip, script_name, py_content)
    vm.vm.hostname = hostname
    vm.vm.network "private_network", ip: ip
    vm.vm.provider "virtualbox" do |vb|
      vb.memory = 512
    end

    vm.vm.provision "shell", inline: <<-SHELL
      echo "📦 Setting up #{hostname}..."

      # Install Python and Flask
      sudo dnf install -y python3 python3-pip
      pip3 install flask requests

      # Create Python script
      cat <<EOF > /home/vagrant/#{script_name}
#{py_content}
EOF

      # Make it executable and run in background
      nohup python3 /home/vagrant/#{script_name} > /dev/null 2>&1 &
    SHELL
  end

  # gateway01
  config.vm.define "gateway01" do |gateway|
    setup_vm(gateway, "gateway01", "192.168.56.40", "gateway_server.py", <<-PYTHON
from flask import Flask, request
import requests, uuid, logging

app = Flask(__name__)
logging.basicConfig(filename="/home/vagrant/gateway.log", level=logging.INFO,
                    format="%(asctime)s [%(levelname)s] TID=%(tid)s: %(message)s")

@app.route("/gateway", methods=["GET"])
def gateway():
    tid = str(uuid.uuid4())
    app.logger.info("Incoming request", extra={"tid": tid})
    res = requests.get("http://192.168.56.41:5001/api", headers={"TID": tid})
    return res.json()

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)
    PYTHON
    )
  end

  # api01
  config.vm.define "api01" do |api|
    setup_vm(api, "api01", "192.168.56.41", "api_server.py", <<-PYTHON
from flask import Flask, request
import requests, logging

app = Flask(__name__)
logging.basicConfig(filename="/home/vagrant/api.log", level=logging.INFO,
                    format="%(asctime)s [%(levelname)s] TID=%(tid)s: %(message)s")

@app.route("/api", methods=["GET"])
def api():
    tid = request.headers.get("TID", "no-tid")
    app.logger.info("Forwarding to echo01", extra={"tid": tid})
    res = requests.get("http://192.168.56.42:5002/echo", headers={"TID": tid})
    return res.json()

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5001)
    PYTHON
    )
  end

  # echo01
  config.vm.define "echo01" do |echo|
    setup_vm(echo, "echo01", "192.168.56.42", "echo_server.py", <<-PYTHON
from flask import Flask, request
import logging

app = Flask(__name__)
logging.basicConfig(filename="/home/vagrant/echo.log", level=logging.INFO,
                    format="%(asctime)s [%(levelname)s] TID=%(tid)s: %(message)s")

@app.route("/echo", methods=["GET"])
def echo():
    tid = request.headers.get("TID", "no-tid")
    app.logger.info("Echo received", extra={"tid": tid})
    return {"message": "Echo from echo01", "TID": tid}

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5002)
    PYTHON
    )
  end
end

 

---

🧪 사용 방법

vagrant up

 

→ 모든 VM이 부팅되며 Flask 서버가 자동 실행됩니다.

---

🧪 테스트

 

curl http://192.168.56.40:5000/gateway

 

실행 시 예시입니다.

D:\repo\jbDeskExample\jbdesk\ch3.2\vagrant_msa_vm>curl http://192.168.56.40:5000/gateway
{"TID":"85bb1038-581c-43b1-85c0-98bec6f6e613","message":"Echo from echo01"}

 

로그 확인 (각 VM에서):

cat /home/vagrant/gateway.log
cat /home/vagrant/api.log
cat /home/vagrant/echo.log

 

2) 로그 추적 코드 생성 (커맨드)

ChatGPT 에 사용한 프롬프트입니다.

python fabric 으로 TID 로 로그 메시지를 추적하는 코드를 만들어주세요.

1. host_name : gateway01 
ip : 192.168.56.40
os : rocky9
log : /home/vagrant/gateway.log

2. host_name : api01
ip : 192.168.56.40
os : rocky9
log : /home/vagrant/api.log

3. host_name : echo01
ip : 192.168.56.40
os : rocky9
log : /home/vagrant/echo.log

 

VM에 SSH 접속할 때 password가 vagrant로 동일하다는 가정하에, Fabric의 connect_kwargs를 사용해 비밀번호 기반 SSH 인증을 추가해드릴게요.

---

✅ log_search.py (비밀번호 자동 입력 포함)

from fabric import Connection
import sys

# VM 정보 (각 호스트 이름과 IP)
hosts = {
    "gateway01": "192.168.56.40",
    "api01": "192.168.56.41",
    "echo01": "192.168.56.42"
}

# VM 로그 파일 경로
log_paths = {
    "gateway01": "/home/vagrant/gateway.log",
    "api01": "/home/vagrant/api.log",
    "echo01": "/home/vagrant/echo.log"
}

# 공통 사용자 및 비밀번호
user = "vagrant"
password = "vagrant"

def search_tid(tid):
    for name, ip in hosts.items():
        log_file = log_paths[name]
        print(f"\n🔍 Searching TID={tid} in {name} ({ip})...")
        try:
            conn = Connection(
                host=ip,
                user=user,
                connect_kwargs={"password": password}
            )
            result = conn.run(f"grep {tid} {log_file}", hide=True, warn=True)
            if result.stdout.strip():
                print(result.stdout.strip())
            else:
                print("❌ No log found.")
        except Exception as e:
            print(f"⚠️ Error connecting to {name}: {e}")

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("❗ Usage: python log_search.py <TID>")
        sys.exit(1)

    tid = sys.argv[1]
    search_tid(tid)

---

✅ 사용 방법

 

python log_search.py <TID>

예:

(jbdesk38_64) PS D:\repo\jbDeskExample\jbdesk\ch3.2> python log_search.py 85bb1038-581c-43b1-85c0-98bec6f6e613

🔍 Searching TID=85bb1038-581c-43b1-85c0-98bec6f6e613 in gateway01 (192.168.56.40)...
2025-04-08 11:02:21,136 [INFO] TID=85bb1038-581c-43b1-85c0-98bec6f6e613: Incoming request

🔍 Searching TID=85bb1038-581c-43b1-85c0-98bec6f6e613 in api01 (192.168.56.41)...
2025-04-08 11:02:21,143 [INFO] TID=85bb1038-581c-43b1-85c0-98bec6f6e613: Forwarding to echo01

🔍 Searching TID=85bb1038-581c-43b1-85c0-98bec6f6e613 in echo01 (192.168.56.42)...
2025-04-08 11:02:21,150 [INFO] TID=85bb1038-581c-43b1-85c0-98bec6f6e613: Echo received

 

3) 로그 추적 코드 생성 (GUI)

PyQt5 로 GUI 버전을 만들어보겠습니다.

ChatGPT 프롬프트입니다.

PyQt5 로 TID 입력 후 Search 버튼을 클릭하면
찾은 로그 메시지를 Grid 에 출력하는 예제를 만들어주세요.
user_name 은 vagrant, password 는 vagrant 를 입력해주세요.

1. host_name : gateway01 
ip : 192.168.56.40
log : /home/vagrant/gateway.log

2. host_name : api01
ip : 192.168.56.41
log : /home/vagrant/api.log

3. host_name : echo01
ip : 192.168.56.42
log : /home/vagrant/echo.log

 

다음은 PyQt5를 사용하여 TID를 입력하고 "Search" 버튼을 누르면, 정의된 3개의 로그 파일에서 해당 TID가 포함된 로그 메시지를 검색하여 Grid(QTableWidget) 에 출력하는 완전한 예제입니다.

---

✅ 요구사항 반영 내용

• TID 입력
• "Search" 버튼 클릭 시 원격 서버(192.168.56.40, 192.168.56.41, 192.168.56.42) 접속
• 각각의 로그 파일 검색:
  • /home/vagrant/gateway.log
  • /home/vagrant/api.log
  • /home/vagrant/echo.log
• 결과를 QTableWidget 에 표시
• SSH 사용자 정보: user_name = vagrant, password = vagrant

---

✅ 코드: log_search_gui.py

import sys
from PyQt5.QtWidgets import (
    QApplication, QWidget, QVBoxLayout, QHBoxLayout,
    QLabel, QLineEdit, QPushButton, QTableWidget, QTableWidgetItem
)
from PyQt5.QtCore import Qt
from fabric import Connection

# 서버 및 로그 정보
LOG_SOURCES = [
    {"host_name": "gateway01", "ip": "192.168.56.40", "log_path": "/home/vagrant/gateway.log"},
    {"host_name": "api01",     "ip": "192.168.56.41", "log_path": "/home/vagrant/api.log"},
    {"host_name": "echo01",    "ip": "192.168.56.42", "log_path": "/home/vagrant/echo.log"},
]

# 고정된 SSH 로그인 정보
SSH_USER = "vagrant"
SSH_PASSWORD = "vagrant"

class LogSearchApp(QWidget):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("TID 로그 검색기")
        self.setGeometry(100, 100, 1000, 500)
        self.init_ui()

    def init_ui(self):
        layout = QVBoxLayout()

        # TID 입력 및 버튼
        input_layout = QHBoxLayout()
        self.tid_input = QLineEdit()
        self.tid_input.setPlaceholderText("검색할 TID 입력")
        self.search_button = QPushButton("Search")
        self.search_button.clicked.connect(self.search_logs)

        input_layout.addWidget(QLabel("TID:"))
        input_layout.addWidget(self.tid_input)
        input_layout.addWidget(self.search_button)
        layout.addLayout(input_layout)

        # 결과 테이블
        self.table = QTableWidget()
        self.table.setColumnCount(3)
        self.table.setHorizontalHeaderLabels(["Host", "Log Path", "Message"])
        layout.addWidget(self.table)

        self.setLayout(layout)

    def search_logs(self):
        tid = self.tid_input.text().strip()
        if not tid:
            return

        self.table.setRowCount(0)
        all_results = []

        for source in LOG_SOURCES:
            try:
                conn = Connection(
                    host=source["ip"],
                    user=SSH_USER,
                    connect_kwargs={"password": SSH_PASSWORD},
                )
                result = conn.run(f"grep {tid} {source['log_path']}", hide=True, warn=True)
                lines = result.stdout.strip().splitlines()

                for line in lines:
                    all_results.append((source["host_name"], source["log_path"], line))

            except Exception as e:
                all_results.append((source["host_name"], source["log_path"], f"❌ {str(e)}"))

        self.populate_table(all_results)

    def populate_table(self, results):
        self.table.setRowCount(len(results))
        for row, (host, path, message) in enumerate(results):
            self.table.setItem(row, 0, QTableWidgetItem(host))
            self.table.setItem(row, 1, QTableWidgetItem(path))
            msg_item = QTableWidgetItem(message)
            msg_item.setFlags(msg_item.flags() ^ Qt.ItemIsEditable)
            self.table.setItem(row, 2, msg_item)

if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = LogSearchApp()
    window.show()
    sys.exit(app.exec_())

 

실행 결과 입니다.

만약 오류가 발생하면 ssh 접속이 되는지 확인을 해야합니다.