mongo perftest

mongo_perf.py

import pymongo
import time
import random
import string
import socks
import socket

PROXY_HOST = "1.2.3.4" 
PROXY_PORT = 1080

socks.set_default_proxy(socks.SOCKS5, PROXY_HOST, PROXY_PORT)
socket.socket = socks.socksocket  
socket.socket = lambda *args, **kwargs: socks.socksocket(socket.AF_INET, socket.SOCK_STREAM)

uri = "mongodb://adathor:[email protected]/admin?replicaSet=replicaset&tls=true"
client = pymongo.MongoClient(uri, tls=True)
db = client["testdb"]
collection = db["testcollection"]

def random_string(size=1024):
    return ''.join(random.choices(string.ascii_letters + string.digits, k=size))

def generate_large_document(i):
    return {
        "id": i,
        "name": f"User_{i}",
        "email": f"user{i}@example.com",
        "bio": random_string(3 * 1024 * 1024),  # 3MB of random text
        "tags": [random_string(128) for _ in range(500)],  # 500 small random strings (~64KB)
        "metadata": {
            "age": random.randint(18, 99),
            "country": random.choice(["USA", "Canada", "UK", "Germany", "France"]),
            "preferences": {
                "theme": random.choice(["dark", "light"]),
                "notifications": bool(random.getrandbits(1))
            },
            "extra_data": [random_string(1024) for _ in range(1000)],  # 1000 strings of 1KB (~1MB)
        },
        "history": [{"timestamp": random.randint(1609459200, 1700000000), "action": random.choice(["login", "purchase", "logout"])} for _ in range(5000)],  # Large history list (~500KB)
        "created_at": time.time()
    }

write_times = []
for i in range(100):
    start = time.time()
    collection.insert_one(generate_large_document(i))
    end = time.time()
    write_times.append(end - start)

# Measure read latency
read_times = []
for i in range(100):
    start = time.time()
    collection.find_one({"id": i})
    end = time.time()
    read_times.append(end - start)

# Calculate and print results
def print_stats(times, operation):
    total_time = sum(times)
    min_time = min(times)
    max_time = max(times)
    avg_time = total_time / len(times)

    print(f"\n{operation} Stats:")
    print(f"  Total Time: {total_time:.6f} seconds")
    print(f"  Fastest {operation}: {min_time:.6f} seconds")
    print(f"  Slowest {operation}: {max_time:.6f} seconds")
    print(f"  Average {operation} Time: {avg_time:.6f} seconds")

print_stats(write_times, "Write")
print_stats(read_times, "Read")