analisis_logs / generador_logs.py
spjasper's picture
version inicial
cf8a984
raw
history blame
4.9 kB
import random
import datetime
import argparse
# --- Configuración de los Mensajes de Log ---
# Puedes personalizar estos mensajes para que se parezcan más a tus logs reales.
LOG_MESSAGES = {
"INFO": [
"User {user_id} logged in successfully from IP {ip_address}.",
"Request received for endpoint /api/v1/data.",
"Processing batch job #{job_id} started.",
"Data synchronization with external service completed.",
"Configuration reloaded successfully.",
"Cache cleared for key: '{cache_key}'.",
],
"WARNING": [
"API response time is high: {response_time}ms for endpoint /api/v1/status.",
"Disk space is running low on /dev/sda1 ({disk_space}% used).",
"A deprecated function 'old_function()' was called.",
"Could not resolve hostname 'temp-service.local', using fallback.",
"User session for {user_id} is about to expire.",
],
"ERROR": [
"Database connection failed: Timeout expired.",
"NullPointerException in module 'PaymentProcessor' at line {line_number}.",
"Failed to write to file /var/logs/app.log: Permission denied.",
"Critical component 'MessageQueue' is not responding.",
"Authentication failed for user '{user_name}'. Invalid credentials provided.",
],
"DEBUG": [
"Executing SQL query: SELECT * FROM users WHERE id={user_id}.",
"Variable 'x' has value: {random_value}.",
"Entering function 'calculate_metrics()'.",
"HTTP request headers: {header_info}.",
]
}
def generate_random_ip():
"""Genera una dirección IP aleatoria."""
return ".".join(str(random.randint(1, 255)) for _ in range(4))
def generate_log_line(current_time):
"""
Genera una única línea de log con un nivel y mensaje aleatorio.
Los niveles tienen diferentes probabilidades para ser más realistas.
"""
# Probabilidades: INFO (65%), DEBUG (20%), WARNING (10%), ERROR (5%)
log_level = random.choices(
population=["INFO", "DEBUG", "WARNING", "ERROR"],
weights=[0.65, 0.20, 0.10, 0.05],
k=1
)[0]
# Elige un mensaje aleatorio para el nivel seleccionado
message_template = random.choice(LOG_MESSAGES[log_level])
# Rellena el template con datos aleatorios para hacerlo más dinámico
message = message_template.format(
user_id=random.randint(100, 999),
ip_address=generate_random_ip(),
job_id=random.randint(1000, 9999),
cache_key=f"user_data_{random.randint(1, 100)}",
response_time=random.randint(200, 2000),
disk_space=random.randint(85, 99),
line_number=random.randint(50, 500),
user_name=random.choice(["admin", "guest", "testuser", "api_service"]),
random_value=round(random.uniform(0, 1), 4),
header_info="{...}"
)
# Formatea el timestamp: [YYYY-MM-DD HH:MM:SS,ms]
timestamp = current_time.strftime('%Y-%m-%d %H:%M:%S') + f",{current_time.microsecond // 1000:03d}"
return f"[{timestamp}] {log_level}: {message}"
def create_log_file(filename, num_lines):
"""Crea un archivo de log con un número específico de líneas."""
print(f"Generando {num_lines} líneas de log en el archivo '{filename}'...")
# Empezamos desde hace unas horas para que los logs no sean todos del mismo segundo
current_time = datetime.datetime.now() - datetime.timedelta(hours=3)
try:
with open(filename, 'w', encoding='utf-8') as f:
for i in range(num_lines):
# Incrementa el tiempo aleatoriamente para cada línea (entre 50ms y 3s)
time_increment = random.uniform(0.05, 3.0)
current_time += datetime.timedelta(seconds=time_increment)
log_line = generate_log_line(current_time)
f.write(log_line + "\n")
# Imprime un progreso para archivos grandes
if (i + 1) % 1000 == 0:
print(f" ... {i + 1} líneas generadas")
print(f"¡Éxito! Archivo '{filename}' creado con {num_lines} líneas.")
except IOError as e:
print(f"Error al escribir en el archivo '{filename}': {e}")
if __name__ == "__main__":
# Configura el parser de argumentos para usar el script desde la terminal
parser = argparse.ArgumentParser(
description="Generador de Archivos de Log de Muestra.",
formatter_class=argparse.RawTextHelpFormatter
)
parser.add_argument(
"num_lines",
type=int,
help="El número de líneas de log a generar."
)
parser.add_argument(
"-o", "--output",
type=str,
default="sample.log",
help="El nombre del archivo de log de salida.\n(default: sample.log)"
)
args = parser.parse_args()
create_log_file(args.output, args.num_lines)