CLI ดึง Log ส่ง PubSub จัดการทรัพยากรดีๆ

สวัสดีครับเพื่อนๆ โปรแกรมเมอร์

เวลาเราทำงานกับระบบต่างๆ นะครับ บ่อยครั้งเลยที่ต้องมานั่งดู Log ไฟล์ใหญ่ๆ แล้วก็อยากจะดึงข้อมูลบางอย่างออกมาเพื่อเอาไปประมวลผลต่อ หรือส่งไปให้ระบบอื่นใช้งาน บทความนี้ผมจะมาแนะนำวิธีสร้าง Python CLI ง่ายๆ ที่สามารถดึงข้อมูลจากไฟล์ Log แล้วส่งต่อไปยัง Google Pub/Sub ได้แบบสบายๆ แถมยังจัดการเรื่องทรัพยากรดีๆ ด้วย Context Manager แล้วก็มีระบบ Log ที่อ่านง่ายแบบ structured log ด้วยครับ

# ก่อนอื่น ติดตั้ง Library ที่จำเป็นก่อนนะครับ
# pip install google-cloud-pubsub structlog

1. จัดการ File Log ด้วย Context Manager นะครับ

การใช้ with open() นี่มันดีมากเลยนะครับ ช่วยให้เราไม่ต้องกังวลเรื่องการปิดไฟล์เลย มันจะจัดการให้เองอัตโนมัติ สะดวกมากๆ ครับ

# log_processor.py
def read_log_file(file_path):
    events = []
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            for line_num, line in enumerate(f):
                # สมมติว่าแต่ละบรรทัดคือ 1 event ที่เราสนใจ
                # หรืออาจจะมี logic ในการ parse ที่ซับซ้อนกว่านี้ก็ได้นะครับ
                if "ERROR" in line or "WARNING" in line:
                    events.append(f"Line {line_num+1}: {line.strip()}")
        print(f"อ่านไฟล์ '{file_path}' ได้ {len(events)} เหตุการณ์ครับ")
        return events
    except FileNotFoundError:
        print(f"ไม่เจอไฟล์ '{file_path}' ครับ")
        return []
    except Exception as e:
        print(f"เกิดข้อผิดพลาดในการอ่านไฟล์: {e} ครับ")
        return []

# ลองเรียกใช้ดูครับ
# สร้าง dummy_log.log ก่อนนะครับ
# echo "INFO: User logged in" > dummy_log.log
# echo "ERROR: Database connection failed" >> dummy_log.log
# echo "WARNING: Low disk space" >> dummy_log.log
# read_log_file('dummy_log.log')

จากตัวอย่างด้านบนนะครับ เราใช้ with open() จัดการไฟล์ พอทำงานเสร็จมันก็ปิดให้เองเลย ง่ายดีนะครับ

2. ระบบ Log แบบ Structured Log นะครับ

เพื่อให้ Log ของเรามันอ่านง่าย และเอาไปวิเคราะห์ต่อได้ง่ายๆ ผมแนะนำให้ใช้ structlog หรือแนวคิดแบบ uber-go/zap นะครับ มันจะช่วยให้ Log เราเป็น JSON หรือ Key-Value ได้เลย

# logger_setup.py
import logging
import structlog

def setup_logging():
    # ตั้งค่า Python standard logging ก่อนนะครับ
    logging.basicConfig(
        format="%(message)s",
        level=logging.INFO,
    )

    # จากนั้นตั้งค่า structlog ครับ
    structlog.configure(
        processors=[
            structlog.stdlib.add_logger_name,
            structlog.stdlib.add_log_level,
            structlog.dev.ConsoleRenderer() if __debug__ else structlog.processors.JSONRenderer(),
        ],
        wrapper_class=structlog.stdlib.BoundLogger,
        logger_factory=structlog.stdlib.LoggerFactory(),
        cache_logger_on_first_use=True,
    )

    # คืนค่า logger ที่ใช้ได้เลยครับ
    return structlog.get_logger()

# ใน main script เราก็เรียกแบบนี้นะครับ
# logger = setup_logging()
# logger.info("โปรแกรมเริ่มต้น", file_path="my_log.log")
# logger.error("มีปัญหาเกิดขึ้น", error="FileNotFound", details="ไฟล์ไม่เจอ")

ดูดีขึ้นเยอะเลยใช่ไหมครับ Log เราจะมีโครงสร้างแบบนี้ทำให้เอาไปค้นหา หรือ Monitoring ใน CloudWatch หรือที่อื่นๆ ได้ง่ายมากๆ เลย

3. ส่งข้อมูลไป Google Pub/Sub นะครับ

พอเราได้ Event ที่ต้องการแล้ว จากนั้นเราก็เอาไปส่งเข้า Pub/Sub ได้เลยนะครับ เพื่อเอาไปประมวลผลต่อแบบ asynchronous

# pubsub_publisher.py
from google.cloud import pubsub_v1
import json

def publish_message(project_id, topic_id, message_data):
    publisher = pubsub_v1.PublisherClient()
    topic_path = publisher.topic_path(project_id, topic_id)

    # ข้อความต้องเป็น bytes นะครับ
    data = json.dumps(message_data).encode('utf-8')

    future = publisher.publish(topic_path, data)
    try:
        message_id = future.result()
        # print(f"ส่งข้อความ ID: {message_id} ไปยัง Topic: {topic_id} สำเร็จครับ")
        return message_id
    except Exception as e:
        # print(f"เกิดข้อผิดพลาดในการส่งข้อความไป Pub/Sub: {e} ครับ")
        raise e

# ตัวอย่างการใช้งานนะครับ
# config = {
#     "project_id": "your-gcp-project-id",
#     "topic_id": "your-pubsub-topic-id"
# }
# publish_message(config["project_id"], config["topic_id"], {"event": "database_error", "level": "ERROR"})

อย่าลืมตั้งค่า GOOGLE_APPLICATION_CREDENTIALS หรือ Authenticate ให้เรียบร้อยก่อนใช้งานนะครับ

4. เอาทั้งหมดมารวมกันใน CLI Script นะครับ

ทีนี้เราก็เอาส่วนต่างๆ มารวมกันเป็น CLI ง่ายๆ นะครับ

# main_cli.py
import argparse
from log_processor import read_log_file
from logger_setup import setup_logging
from pubsub_publisher import publish_message

def main():
    logger = setup_logging()

    parser = argparse.ArgumentParser(description="CLI tool สำหรับดึง Log และส่งไป Pub/Sub ครับ")
    parser.add_argument("file_path", help="เส้นทางของไฟล์ Log ที่ต้องการประมวลผลครับ")
    parser.add_argument("--project_id", required=True, help="Google Cloud Project ID ครับ")
    parser.add_argument("--topic_id", required=True, help="Google Pub/Sub Topic ID ครับ")

    args = parser.parse_args()

    logger.info("เริ่มต้นประมวลผล", file=args.file_path, project=args.project_id, topic=args.topic_id)

    events = read_log_file(args.file_path)

    if not events:
        logger.warning("ไม่พบเหตุการณ์ที่น่าสนใจ หรือมีปัญหาในการอ่านไฟล์ครับ", file=args.file_path)
        return

    for i, event in enumerate(events):
        try:
            message_id = publish_message(args.project_id, args.topic_id, {"source_file": args.file_path, "event_data": event})
            logger.info("ส่งข้อความสำเร็จ", message_id=message_id, event_num=i+1)
        except Exception as e:
            logger.error("ส่งข้อความล้มเหลว", event_data=event, error=str(e))

    logger.info("ประมวลผลเสร็จสิ้นครับ", total_events=len(events), file=args.file_path)

if __name__ == "__main__":
    main()

วิธีรัน CLI นะครับ:

python main_cli.py dummy_log.log --project_id your-gcp-project-id --topic_id your-pubsub-topic-id

แค่นี้เราก็มี CLI ที่ดึง Log ส่งข้อมูลไป Pub/Sub พร้อม Log แบบมีโครงสร้างใช้แล้วนะครับ สะดวกสบายมากๆ เลยใช่ไหมครับเพื่อนๆ

หวังว่าบทความนี้จะเป็นประโยชน์กับเพื่อนๆ นะครับ ลองเอาไปปรับใช้กับงานของตัวเองดูได้เลยครับ

แหล่งที่มา: Google Cloud Pub/Sub Documentation, structlog library documentation ผู้เขียน: cii3.net