Cấu hình proxy cho Locust load testing và tránh chặn IP

Giới thiệu về Locust và tại sao cần proxy

Locust là công cụ kiểm tra hiệu suất mã nguồn mở, cho phép mô phifflng hàng nghìn người dùng ảo bằng Python. Khi chạy thử nghiệm tải trọng lớn, việc gửi nhiều request từ cùng một địa chỉ IP dễ dẫn đến bị chặn, giới hạn tốc độ hoặc nhận được phản hồi sai do cơ chế bảo vệ của máy chủ đích. Việc sử dụng proxy giúp

• Phân tán lưu lượng ra nhiều IP khác nhau.
• Giảm nguy cơ bị chặn do Rate Limiting hoặc anti‑bot.
• Mô phỏng truy cập từ các vị trí địa lý khác nhau (nếu proxy có thể chọn theo quốc gia).
• Bảo vệ IP thực của máy kiểm tra khỏi bị ghi log hoặc tấn công násled.

Chuẩn bị môi trường

Trước khi bắt đầu, cần cài đặt Locust và một số thư viện hỗ trợ

pip install locust requests

Tạo thư mục dự án và file locustfile.py chứa kịch bản kiểm tra cơ bản.

Kịch bản kiểm tra đơn giản

Dưới đây là ví dụ một HttpUser truy cập trang chủ của một website mục tiêu

from locust import HttpUser, task, between

class BrowseUser(HttpUser):
    wait_time = between(1, 3)

    @task
    def load_homepage(self):
        self.client.get('/')

Truyền proxy qua tham số dòng lệnh

Locust cho phép chỉ định proxy cho toàn bộ quá trình chạy bằng cách sử dụng biến môi trường HTTP_PROXY và HTTPS_PROXY hoặc tham số --proxy. Ví dụ

locust -f locustfile.py --headless -u 200 -r 20 --run-time 5m --proxy http://user:[email protected]:3128

Trong lệnh trên

• --headless chạy không giao diện.
• -u 200 mô phỏng 200 người dùng ảo.
• -r 20 tăng 20 người dùng mỗi giây.
• --run-time 5m chạy trong 5 phút.
• --proxy chỉ định địa chỉ proxy mà tất cả request sẽ đi qua.

Cấu hình proxy trong code (tùy chỉnh hơn)

Nếu muốn mỗi người dùng hoặc mỗi request sử dụng proxy khác nhau, chúng ta có thể overriding session của HttpUser. Dưới đây là cách tạo danh sách proxy và chọn ngẫu nhiên cho mỗi user

import random
from locust import HttpUser, task, between

PROXIES = [
    'http://user1:[email protected]:3128',
    'http://user2:[email protected]:3128',
    'http://user3:[email protected]:3128',
]

class ProxyUser(HttpUser):
    wait_time = between(1, 2)

    def on_start(self):
        # chọn một proxy ngẫu nhiên khi user khởi động
        proxy = random.choice(PROXIES)
        self.client.proxies = {
            'http': proxy,
            'https': proxy,
        }

    @task
    def load_api(self):
        self.client.get('/api/items')

Luân phiên proxy theo mỗi request (rotating proxy)

Một số trường hợp cần thay đổi proxy cho mỗi request để tránh việc một IP được sử dụng quá nhiều. Ta có thể override phương thức request của session hoặc sử dụng hook events. Dưới đây là cách dùng request_event để thay đổi proxy trước mỗi request

from locust import HttpUser, task, between, events
import random

PROXIES = [
    'http://user1:[email protected]:3128',
    'http://user2:[email protected]:3128',
    'http://user3:[email protected]:3128',
]

class RotatingProxyUser(HttpUser):
    wait_time = between(1, 2)

    @events.init.add_listener
    def on_locust_init(environment, **kwargs):
        # lưu danh sách proxy vào environment để mọi user truy cập
        environment.parsed_options.proxies = PROXIES

    def setup_proxy(self):
        proxy = random.choice(self.environment.parsed_options.proxies)
        self.client.proxies = {'http': proxy, 'https': proxy}

    @events.request.add_listener
    def _on_request(self, request_type, name, response, **kwargs):
        # thay đổi proxy sau mỗi request thành công
        self.setup_proxy()

    @task
    def call_endpoint(self):
        self.setup_proxy()   # đảm bảo première request dùng proxy
        self.client.get('/endpoint')

Lưu ý Cách trên thay đổi proxy sau mỗi request nhưng có thể làm tăng overhead do tạo kết nối TCP mới. Nếu mục tiêu là mô phỏng nhiều người dùng thực tế với IP-fixed trong thời gian ngắn hãy cân đối giữa rotating và sticky.

Kiểm tra và debug proxy

Để đảm bảo proxy đang hoạt động đúng chúng ta có thể ghi log địa chỉ IP mà request thấy được từ phía server. Giả sử endpoint /ip trả về IP công cộng của requester

    @task
    def check_ip(self):
        with self.client.get('/ip', catch_response=True) as resp:
            if resp.status_code == 200:
                ip = resp.text.strip()
                self.environment.events.request.fire(
                    request_type='GET',
                    name='check_ip',
                    response_time=resp.elapsed.total_seconds()*1000,
                    response_length=len(resp.text),
                    exception=None,
                    context={},
                    extra={'detected_ip': ip}
                )
            else:
                resp.failure(f'Status code {resp.status_code}')

Trong log Locust sẽ xuất hiện trường detected_ip giúp chúng ta xác nhận proxy đang được sử dụng.

Tối ưu hiệu suất khi dùng proxy

1. Chọn proxy gần đích – latency thấp giúp tăng throughput và giảm thời gian chờ.
2. Giới hạn số kết nối đồng thời – một số proxy service có giới hạn kết nối; cấu hình -w (số worker) trong Locust để không vượt quá.
3. Sử dụng keep‑alive – đảm bảo requests session reuse kết nối để giảm tải handshake TLS.
4. Giám sát lỗi proxy – theo dõi sự kiện request_failure để phát hiện proxy die và tự động loại bỏ khỏi danh sách.
5. Throttle tốc độ – nếu mục tiêu có Rate Limiting nghiêm ngặt điều chỉnh spawn rate (-r) và wait time giữa các task.

Lưu ý pháp lý và ethics

• Chỉ chạy thử nghiệm trên hệ thống mà bạn có quyền hoặc đã nhận được permiso explícito từ chủ sở hữu.
• Tránh gửi lượng request quá lớn có thể làm ngừng dịch vụ (DoS) mà không có sự đồng ý.
• Tuân thủ các quy định về bảo vệ dữ liệu và riêng tư khi thu thập thông qua proxy.
• Chọn nhà cung cấp proxy uy tín như RoProxy để đảm bảo IP sạch, không bị blacklist và có chính sách sử dụng rõ ràng.

Kết luận

Việc tích hợp proxy vào quy trình load testing với Locust giúp bạn mô phỏng truy cập thực tế từ nhiều nguồn IP tránh bị chặn và thu được kết quả đo lường chính xác hơn. Bắt đầu bằng cấu hình đơn giản qua tham số --proxy sau đó nâng cao tới rotating proxy hoặc sticky session tùy thuộc vào mục tiêu kiểm tra. Hãy luôn giám sát tối ưu latency và tuân thủ nguyên tắc ethics để kiểm tra hiệu suất thành công và bền vững.