SonarQube Analysis Machine Learning Pipeline — ตรวจคุณภาพโค้ด ML Pipeline | SiamCafe Blog

SonarQube ML Pipeline

SonarQube Machine Learning Pipeline Code Quality Bug Vulnerability Coverage Data Validation Feature Engineering Model Training Serving CI/CD

ML Code Quality Rules

# === ML-specific Quality Rules ===

from dataclasses import dataclass

@dataclass
class MLQualityRule:
 rule_id: str
 name: str
 severity: str
 category: str
 bad_example: str
 good_example: str

rules = [
 MLQualityRule("ML001",
 "ห้าม Hardcode File Path",
 "MAJOR",
 "Data Ingestion",
 "df = pd.read_csv('/home/user/data/train.csv')",
 "df = pd.read_csv(config.DATA_PATH / 'train.csv')"),
 MLQualityRule("ML002",
 "ต้องตั้ง Random Seed",
 "CRITICAL",
 "Model Training",
 "model = RandomForestClassifier()",
 "model = RandomForestClassifier(random_state=config.SEED)"),
 MLQualityRule("ML003",
 "ห้าม print() ใน Production Code",
 "MINOR",
 "All",
 "print(f'Training loss: {loss}')",
 "logger.info(f'Training loss: {loss}')"),
 MLQualityRule("ML004",
 "ต้อง Validate Input ใน API",
 "CRITICAL",
 "Model Serving",
 "prediction = model.predict(request.json['data'])",
 "validated = InputSchema(**request.json); prediction = model.predict(validated.data)"),
 MLQualityRule("ML005",
 "ห้าม Data Leak (ใช้ Test Data ก่อน Split)",
 "BLOCKER",
 "Feature Engineering",
 "scaler.fit(all_data); X_train = scaler.transform(X_train)",
 "scaler.fit(X_train); X_test = scaler.transform(X_test)"),
 MLQualityRule("ML006",
 "ต้อง Handle Missing Values",
 "MAJOR",
 "Feature Engineering",
 "features = df[['col1', 'col2']].values",
 "features = df[['col1', 'col2']].fillna(strategy).values"),
]

print("=== ML Quality Rules ===")
for r in rules:
 print(f"\n [{r.rule_id}] {r.name} ({r.severity})")
 print(f" Category: {r.category}")
 print(f" Bad: {r.bad_example}")
 print(f" Good: {r.good_example}")

CI/CD Integration

# === ML Pipeline CI/CD with SonarQube ===

# GitHub Actions for ML Pipeline
# name: ML Pipeline Quality
# on: [pull_request]
# jobs:
# quality:
# runs-on: ubuntu-latest
# steps:
# - uses: actions/checkout@v4
# - name: Setup Python
# uses: actions/setup-python@v5
# with: { python-version: '3.10' }
# - name: Install Dependencies
# run: pip install -r requirements.txt
# - name: Run Tests with Coverage
# run: pytest tests/ --cov=src --cov-report=xml
# - name: Data Validation Tests
# run: great_expectations checkpoint run ml_data_check
# - name: SonarQube Scan
# uses: SonarSource/sonarqube-scan-action@v2
# env:
# SONAR_TOKEN: }
# - name: Quality Gate
# uses: SonarSource/sonarqube-quality-gate-action@v1

@dataclass
class PipelineStage:
 stage: str
 tools: str
 checks: str
 gate: str

stages = [
 PipelineStage("1. PR Quality Check",
 "SonarQube + pytest + Great Expectations",
 "Code Quality, Unit Tests, Data Schema Validation",
 "Quality Gate Pass + Tests Pass + Data Valid"),
 PipelineStage("2. Integration Test",
 "pytest + MLflow + Docker",
 "End-to-end Pipeline Test, Model Training Smoke Test",
 "All Tests Pass, Model Metrics > Baseline"),
 PipelineStage("3. Model Training",
 "MLflow + DVC + SonarQube",
 "Data Quality, Training Metrics, Model Validation",
 "Accuracy > Threshold, No Data Drift"),
 PipelineStage("4. Model Registry",
 "MLflow Model Registry",
 "Model Artifact, Metrics, Lineage, Version",
 "Model Approved by Reviewer"),
 PipelineStage("5. Staging Deploy",
 "Kubernetes + Seldon/BentoML",
 "API Tests, Load Test, Security Scan (DAST)",
 "Response Time < 100ms, No Vulnerabilities"),
 PipelineStage("6. Production Deploy",
 "Kubernetes + Canary/Blue-Green",
 "Canary Metrics, A/B Test, Model Monitor",
 "No Degradation in Business Metrics"),
]

print("=== ML CI/CD Pipeline ===")
for s in stages:
 print(f"\n [{s.stage}]")
 print(f" Tools: {s.tools}")
 print(f" Checks: {s.checks}")
 print(f" Gate: {s.gate}")

Testing Strategy

# === ML Testing Strategy ===

@dataclass
class TestCategory:
 category: str
 what_to_test: str
 tools: str
 coverage_target: str

test_categories = [
 TestCategory("Unit Tests (Data)",
 "Data Loading, Transformation, Validation, Encoding",
 "pytest + pandas testing + Great Expectations",
 "> 80% Coverage"),
 TestCategory("Unit Tests (Feature)",
 "Feature Engineering, Scaling, Selection, Encoding",
 "pytest + numpy testing",
 "> 70% Coverage"),
 TestCategory("Unit Tests (Model)",
 "Model Init, Predict Shape, Save/Load, Config",
 "pytest + sklearn metrics",
 "> 50% Coverage (Training ยากทดสอบ)"),
 TestCategory("Integration Tests",
 "End-to-end Pipeline, Data→Feature→Model→Predict",
 "pytest + Docker + Sample Data",
 "> 60% Coverage"),
 TestCategory("API Tests",
 "Endpoint, Input Validation, Error Handling, Auth",
 "pytest + httpx + FastAPI TestClient",
 "> 80% Coverage"),
 TestCategory("Data Quality Tests",
 "Schema, Null, Range, Distribution, Freshness",
 "Great Expectations + Monte Carlo",
 "100% Critical Checks"),
]

print("=== ML Testing Strategy ===")
for t in test_categories:
 print(f" [{t.category}]")
 print(f" Test: {t.what_to_test}")
 print(f" Tools: {t.tools}")
 print(f" Target: {t.coverage_target}")

เคล็ดลับ

• Coverage: ML Training Code Coverage ต่ำกว่า Web App เป็นปกติ ตั้ง 50-70%
• Custom Rules: สร้าง Custom Rules สำหรับ ML เช่น Data Leak Random Seed
• Notebook: แปลง Notebook เป็น .py ก่อน Scan ด้วย nbconvert
• Profile: แยก Quality Profile สำหรับ ML Code กับ App Code
• Data Test: ใช้ Great Expectations ตรวจ Data Quality ควบคู่ Code Quality

การนำไปใช้งานจริงในองค์กร

สำหรับองค์กรขนาดกลางถึงใหญ่ แนะนำให้ใช้หลัก Three-Tier Architecture คือ Core Layer ที่เป็นแกนกลางของระบบ Distribution Layer ที่ทำหน้าที่กระจาย Traffic และ Access Layer ที่เชื่อมต่อกับผู้ใช้โดยตรง การแบ่ง Layer ชัดเจนช่วยให้การ Troubleshoot ง่ายขึ้นและสามารถ Scale ระบบได้ตามความต้องการ

เรื่อง Network Security ก็สำคัญไม่แพ้กัน ควรติดตั้ง Next-Generation Firewall ที่สามารถ Deep Packet Inspection ได้ ใช้ Network Segmentation แยก VLAN สำหรับแต่ละแผนก ติดตั้ง IDS/IPS เพื่อตรวจจับการโจมตี และทำ Regular Security Audit อย่างน้อยปีละ 2 ครั้ง

สรุป

SonarQube ML Pipeline Code Quality Custom Rules Data Leak Random Seed Coverage API Security Great Expectations MLflow CI/CD Production