Python/AI/generate_k8s_data.py - main - Demek/DSam621

Python/AI/generate_k8s_data.py
307 строк10 KB
d_e_m_e_k
добавил учебную ai для обучения
27 июл 2026, 23:34
27 июл 2026, 23:342dfa071
import json
import os
import random
import string

# ============================================
# НАСТРОЙКИ
# ============================================
NUM_EXAMPLES = 1000  # желаемое количество примеров
OUTPUT_PATH = "data/expanded_dataset.jsonl"

# ============================================
# ГЕНЕРАТОРЫ
# ============================================

DEPLOYMENT_NAMES = [
    "nginx", "redis", "postgres", "mysql", "api-server", "worker", "frontend",
    "backend", "auth-service", "cache", "db-migration", "scheduler", "cronjob",
    "monitoring", "logger", "proxy", "gateway", "ingress-nginx", "cert-manager",
    "vault", "consul", "envoy", "traefik", "prometheus", "grafana", "alertmanager",
    "kube-state-metrics", "node-exporter", "coredns", "kubernetes-dashboard"
]

REPLICAS = [1, 2, 3, 5, 7, 10, 15, 20]
CONTAINER_IMAGES = [
    "nginx:1.24", "nginx:1.23-alpine", "nginx:latest",
    "redis:7-alpine", "redis:latest",
    "postgres:15", "postgres:14-alpine",
    "mysql:8.0", "mysql:5.7",
    "golang:1.21-alpine", "python:3.11-slim", "node:20-alpine",
    "java:17-slim", "openjdk:17-jdk-slim"
]
PORTS = [80, 443, 8080, 8443, 3000, 5000, 6379, 5432, 27017, 9090, 9091]
SERVICE_TYPES = ["ClusterIP", "NodePort", "LoadBalancer"]

CONFIGMAP_KEYS = ["config.yaml", "app.properties", "nginx.conf", "my.cnf", "settings.json"]
SECRET_KEYS = ["password", "api-key", "token", "credentials.json", "private-key.pem"]

# ============================================
# ГЕНЕРАТОРЫ YAML
# ============================================

def random_name(prefix="app"):
    suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=6))
    return f"{prefix}-{suffix}"

def generate_deployment():
    name = random.choice(DEPLOYMENT_NAMES) + "-deploy-" + ''.join(random.choices(string.ascii_lowercase, k=4))
    replicas = random.choice(REPLICAS)
    image = random.choice(CONTAINER_IMAGES)
    container_name = name.replace("-", "")
    container_port = random.choice(PORTS)

    return f"""apiVersion: apps/v1
kind: Deployment
metadata:
  name: {name}
  labels:
    app: {name}
spec:
  replicas: {replicas}
  selector:
    matchLabels:
      app: {name}
  template:
    metadata:
      labels:
        app: {name}
    spec:
      containers:
      - name: {container_name}
        image: {image}
        ports:
        - containerPort: {container_port}
        env:
        - name: LOG_LEVEL
          value: "{random.choice(['debug', 'info', 'warn', 'error'])}"
        resources:
          requests:
            memory: "{random.choice(['64Mi','128Mi','256Mi'])}"
            cpu: "100m"
          limits:
            memory: "{random.choice(['128Mi','256Mi','512Mi'])}"
            cpu: "500m"
"""

def generate_service(selector_app=None):
    name = random_name("svc")
    selector_app = selector_app or random.choice(DEPLOYMENT_NAMES)
    port = random.choice(PORTS)
    target_port = port if random.random() > 0.3 else port + 1000
    svc_type = random.choice(SERVICE_TYPES)

    return f"""apiVersion: v1
kind: Service
metadata:
  name: {name}
spec:
  selector:
    app: {selector_app}
  ports:
  - port: {port}
    targetPort: {target_port}
  type: {svc_type}
"""

def generate_configmap():
    name = random_name("config")
    key = random.choice(CONFIGMAP_KEYS)

    configs = {
        "config.yaml": f"""logging:
  level: {random.choice(['debug', 'info', 'warn'])}
  format: json
service:
  timeout: 30s
""",
        "app.properties": f"""app.name={name}
app.version=1.0.0
db.host=postgres
""",
        "nginx.conf": f"""server {{
    listen {random.choice([80, 443])};
    server_name {name}.local;
    location / {{ proxy_pass http://localhost:8080; }}
}}
""",
        "my.cnf": f"""[mysqld]
max_connections = {random.choice([50, 100, 200])}
""",
        "settings.json": json.dumps({
            "debug": random.choice([True, False]),
            "version": "1.0.0"
        }, indent=2)
    }

    return f"""apiVersion: v1
kind: ConfigMap
metadata:
  name: {name}
data:
  {key}: |
{configs[key].strip().replace(chr(10), chr(10) + "    ")}
"""

def generate_secret():
    name = random_name("secret")
    key = random.choice(SECRET_KEYS)
    value = ''.join(random.choices(string.ascii_letters + string.digits, k=24))

    return f"""apiVersion: v1
kind: Secret
metadata:
  name: {name}
type: Opaque
data:
  {key}: {value.encode('utf-8').hex()}
"""

def generate_ingress():
    name = random_name("ingress")
    host = random.choice(INGRESS_HOSTS) if 'INGRESS_HOSTS' in dir() else random_name("example") + ".com"
    path = random.choice(["/", "/api", "/admin", "/health"])

    return f"""apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: {name}
  annotations:
    nginx.ingress.kubernetes.io/rewrite-target: /
spec:
  rules:
  - host: {host}
    http:
      paths:
      - path: {path}
        pathType: Prefix
        backend:
          service:
            name: {random_name("svc")}
            port:
              number: {random.choice(PORTS)}
"""

# ============================================
# ГЕНЕРАТОР ИНСТРУКЦИЙ
# ============================================

INSTRUCTION_TEMPLATES = {
    "deployment": [
        "Создай Deployment для {app} с {replicas} репликами",
        "Опиши Deployment {app}, который должен иметь {replicas} экземпляра",
        "Сделай Deployment для {app} с числом реплик {replicas}",
        "Создай Deployment с именем {app} и {replicas} подами",
        "Напиши манифест Deployment для {app}, используя {replicas} копии",
    ],
    "service": [
        "Создай Service для {app} типа {type}",
        "Опиши Service {name} для {app} с типом {type}",
        "Сделай Service {name} для подов {app} типа {type}",
        "Напиши Service {name} типа {type} для {app}",
        "Опиши Service для {app} с именем {name} и типом {type}",
    ],
    "configmap": [
        "Создай ConfigMap {name} с данными {key}",
        "Опиши ConfigMap {name}, добавив ключ {key}",
        "Сделай ConfigMap {name} с содержимым {key}",
        "Создай ConfigMap {name} и добавь туда {key}",
        "Опиши ConfigMap {name} с данными: {key}",
    ],
    "secret": [
        "Создай Secret {name} с ключом {key}",
        "Опиши Secret {name}, где будет {key}",
        "Сделай Secret {name} и добавь туда {key}",
        "Создай Secret {name} с данными: {key}",
        "Опиши Secret {name}, содержащий {key}",
    ],
    "ingress": [
        "Создай Ingress {name} для хоста {host}",
        "Опиши Ingress {name} с правилом для {host}",
        "Сделай Ingress {name}, который направляет трафик на {host}",
        "Создай Ingress {name} для маршрутизации на {host}",
        "Опиши Ingress {name} с правилом host={host}",
    ],
}

def generate_instruction(template, **kwargs):
    return template.format(**kwargs)

# ============================================
# ОСНОВНАЯ ФУНКЦИЯ
# ============================================

def main():
    os.makedirs(os.path.dirname(OUTPUT_PATH) or ".", exist_ok=True)
    examples = []

    for i in range(NUM_EXAMPLES):
        # Выбираем тип манифеста
        resource_type = random.choice(["deployment", "service", "configmap", "secret", "ingress"])

        if resource_type == "deployment":
            yaml = generate_deployment()
            # Извлекаем имя из YAML
            name = yaml.split("name: ")[1].split("\n")[0].strip()
            replicas = int(yaml.split("replicas: ")[1].split("\n")[0].strip())
            instruction = generate_instruction(
                random.choice(INSTRUCTION_TEMPLATES["deployment"]),
                app=name, replicas=replicas
            )

        elif resource_type == "service":
            yaml = generate_service()
            name = yaml.split("name: ")[1].split("\n")[0].strip()
            selector = yaml.split("app: ")[1].split("\n")[0].strip()
            svc_type = yaml.split("type: ")[1].split("\n")[0].strip()
            instruction = generate_instruction(
                random.choice(INSTRUCTION_TEMPLATES["service"]),
                name=name, app=selector, type=svc_type
            )

        elif resource_type == "configmap":
            yaml = generate_configmap()
            name = yaml.split("name: ")[1].split("\n")[0].strip()
            key = yaml.split("data:")[1].split("\n")[1].strip().split(":")[0]
            instruction = generate_instruction(
                random.choice(INSTRUCTION_TEMPLATES["configmap"]),
                name=name, key=key
            )

        elif resource_type == "secret":
            yaml = generate_secret()
            name = yaml.split("name: ")[1].split("\n")[0].strip()
            key = yaml.split("data:")[1].split("\n")[1].strip().split(":")[0]
            instruction = generate_instruction(
                random.choice(INSTRUCTION_TEMPLATES["secret"]),
                name=name, key=key
            )

        else:  # ingress
            yaml = generate_ingress()
            name = yaml.split("name: ")[1].split("\n")[0].strip()
            host = yaml.split("host: ")[1].split("\n")[0].strip()
            instruction = generate_instruction(
                random.choice(INSTRUCTION_TEMPLATES["ingress"]),
                name=name, host=host
            )

        examples.append({
            "instruction": instruction,
            "response": yaml.strip()
        })

        if (i + 1) % 100 == 0:
            print(f"✅ Сгенерировано {i+1}/{NUM_EXAMPLES} примеров")

    # Сохраняем
    with open(OUTPUT_PATH, 'w', encoding='utf-8') as f:
        for ex in examples:
            f.write(json.dumps(ex, ensure_ascii=False) + '\n')

    print(f"\n🎉 Готово! {NUM_EXAMPLES} примеров сохранено в {OUTPUT_PATH}")
    print(f"📊 Размер: {os.path.getsize(OUTPUT_PATH) // (1024*1024)} МБ")


if __name__ == "__main__":
    main()