Ray 分散式計算框架的雲端除錯實務技術

import ray
import pandas as pd
from typing import List, Dict

# 初始化 Ray 執行環境
# local_mode=True 表示在本地模式執行,便於除錯
ray.init(local_mode=False)

@ray.remote
def load_data_chunk(file_path: str, chunk_id: int) -> pd.DataFrame:
    """
    載入資料檔案的特定區塊
    
    參數說明:
        file_path: 資料檔案的完整路徑
        chunk_id: 要載入的資料區塊編號
        
    回傳值:
        載入的 DataFrame 資料區塊
        
    設計考量:
        將資料載入獨立成函式,便於追蹤 IO 相關錯誤
        當檔案讀取失敗時,錯誤訊息會明確指出是哪個檔案與區塊
    """
    try:
        # 讀取指定區塊的資料
        # skiprows 與 nrows 參數用於實現分塊讀取
        chunk_size = 10000
        skip_rows = chunk_id * chunk_size
        
        df = pd.read_csv(
            file_path,
            skiprows=range(1, skip_rows + 1),  # 跳過前面的區塊
            nrows=chunk_size,  # 只讀取當前區塊
            encoding='utf-8'
        )
        
        print(f"成功載入區塊 {chunk_id},資料筆數: {len(df)}")
        return df
        
    except FileNotFoundError as e:
        # 檔案不存在的錯誤處理
        print(f"錯誤: 找不到檔案 {file_path}")
        raise
    except pd.errors.EmptyDataError:
        # 空資料檔案的錯誤處理
        print(f"警告: 區塊 {chunk_id} 為空資料")
        return pd.DataFrame()
    except Exception as e:
        # 其他非預期錯誤的處理
        print(f"載入區塊 {chunk_id} 時發生錯誤: {str(e)}")
        raise

@ray.remote
def clean_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    清洗資料,移除無效記錄與異常值
    
    參數說明:
        df: 原始資料 DataFrame
        
    回傳值:
        清洗後的 DataFrame
        
    設計考量:
        資料清洗邏輯獨立,便於追蹤資料品質問題
        可以輕鬆替換不同的清洗策略
    """
    # 移除空值記錄
    initial_count = len(df)
    df_cleaned = df.dropna()
    
    # 移除重複記錄
    df_cleaned = df_cleaned.drop_duplicates()
    
    # 記錄清洗統計資訊
    removed_count = initial_count - len(df_cleaned)
    if removed_count > 0:
        print(f"資料清洗: 移除 {removed_count} 筆無效記錄")
    
    return df_cleaned

@ray.remote
def transform_data(df: pd.DataFrame, config: Dict) -> pd.DataFrame:
    """
    根據組態進行資料轉換
    
    參數說明:
        df: 清洗後的資料 DataFrame
        config: 轉換組態字典
        
    回傳值:
        轉換後的 DataFrame
        
    設計考量:
        轉換邏輯參數化,便於測試不同的轉換策略
        錯誤訊息會包含具體的轉換步驟資訊
    """
    try:
        # 執行資料型態轉換
        if 'type_conversions' in config:
            for col, dtype in config['type_conversions'].items():
                if col in df.columns:
                    df[col] = df[col].astype(dtype)
        
        # 執行數值正規化
        if 'normalize_columns' in config:
            for col in config['normalize_columns']:
                if col in df.columns:
                    # Min-Max 正規化到 [0, 1] 區間
                    min_val = df[col].min()
                    max_val = df[col].max()
                    df[col] = (df[col] - min_val) / (max_val - min_val + 1e-8)
        
        return df
        
    except KeyError as e:
        print(f"轉換錯誤: 找不到欄位 {e}")
        raise
    except Exception as e:
        print(f"資料轉換時發生錯誤: {str(e)}")
        raise

@ray.remote
def aggregate_results(dataframes: List[pd.DataFrame]) -> pd.DataFrame:
    """
    彙總多個資料區塊的處理結果
    
    參數說明:
        dataframes: DataFrame 列表
        
    回傳值:
        彙總後的 DataFrame
        
    設計考量:
        彙總邏輯獨立,便於處理分散式運算結果
        包含資料一致性檢查
    """
    if not dataframes:
        print("警告: 沒有資料需要彙總")
        return pd.DataFrame()
    
    # 合併所有 DataFrame
    result = pd.concat(dataframes, ignore_index=True)
    
    # 確保索引連續性
    result = result.reset_index(drop=True)
    
    print(f"彙總完成: 總計 {len(result)} 筆記錄")
    return result

def process_pipeline(file_paths: List[str], transform_config: Dict):
    """
    執行完整的資料處理管線
    展示如何組合多個 Ray 遠端函式
    
    參數說明:
        file_paths: 資料檔案路徑列表
        transform_config: 轉換組態
        
    設計考量:
        每個步驟都可以獨立除錯
        錯誤發生時能夠精確定位到特定步驟
    """
    # 第一階段: 平行載入所有資料區塊
    load_tasks = []
    for file_path in file_paths:
        # 假設每個檔案切分成 3 個區塊處理
        for chunk_id in range(3):
            task = load_data_chunk.remote(file_path, chunk_id)
            load_tasks.append(task)
    
    # 等待所有載入任務完成
    loaded_chunks = ray.get(load_tasks)
    print(f"完成載入 {len(loaded_chunks)} 個資料區塊")
    
    # 第二階段: 平行清洗所有區塊
    clean_tasks = [clean_data.remote(chunk) for chunk in loaded_chunks]
    cleaned_chunks = ray.get(clean_tasks)
    print("完成資料清洗階段")
    
    # 第三階段: 平行轉換所有區塊
    transform_tasks = [
        transform_data.remote(chunk, transform_config) 
        for chunk in cleaned_chunks
    ]
    transformed_chunks = ray.get(transform_tasks)
    print("完成資料轉換階段")
    
    # 第四階段: 彙總結果
    final_result = ray.get(aggregate_results.remote(transformed_chunks))
    
    return final_result

# 使用範例
if __name__ == "__main__":
    # 組態轉換參數
    config = {
        'type_conversions': {
            'price': 'float64',
            'quantity': 'int64'
        },
        'normalize_columns': ['price']
    }
    
    # 執行資料處理管線
    files = ['data_part1.csv', 'data_part2.csv']
    
    try:
        result = process_pipeline(files, config)
        print(f"處理完成,最終資料筆數: {len(result)}")
    except Exception as e:
        print(f"管線執行失敗: {str(e)}")
    finally:
        # 清理 Ray 資源
        ray.shutdown()

import ray
from multiprocessing import Pool
import numpy as np

# 這是一個常見的錯誤範例
# 展示作用域捕獲可能導致的問題

# 全域變數 - 可能被意外捕獲
global_cache = {"data": np.random.rand(1000000)}

# 無法序列化的物件
process_pool = Pool(4)

def problematic_function(x):
    """
    這個函式示範了幾種常見的作用域捕獲問題
    """
    # 問題 1: 捕獲全域變數
    # Ray 會嘗試序列化 global_cache,造成不必要的資料傳輸
    cached_value = global_cache.get("data")
    
    # 問題 2: 捕獲無法序列化的物件
    # process_pool 無法被 Ray 序列化
    def nested_function(y):
        return process_pool.map(lambda z: z * 2, range(y))
    
    return nested_function(x)

# 改進後的版本
@ray.remote
def improved_function(x, cache_data=None):
    """
    改進後的函式,明確傳遞需要的參數
    
    參數說明:
        x: 運算輸入值
        cache_data: 需要使用的快取資料,預設為 None
        
    回傳值:
        運算結果
        
    改進要點:
        1. 不依賴全域變數,而是透過參數明確傳遞
        2. 避免捕獲無法序列化的物件
        3. 所有依賴都透過參數介面暴露
    """
    # 只使用明確傳入的參數
    if cache_data is not None:
        result = np.sum(cache_data) + x
    else:
        result = x * 2
    
    return result

# 正確使用方式
if __name__ == "__main__":
    ray.init(local_mode=False)
    
    # 明確傳遞需要的資料
    cache_subset = global_cache["data"][:1000]  # 只傳遞必要的子集
    
    # 啟動遠端任務
    future = improved_function.remote(10, cache_data=cache_subset)
    result = ray.get(future)
    
    print(f"運算結果: {result}")
    ray.shutdown()

import ray
from multiprocessing import Pool
import threading

# 建立一些測試用的物件
thread_pool = Pool(5)  # 無法序列化
lock_object = threading.Lock()  # 無法序列化

class ComplexClass:
    """
    包含多種物件型態的複雜類別
    用於測試序列化檢查工具
    """
    def __init__(self):
        self.normal_data = {"key": "value"}  # 可序列化
        self.pool = thread_pool  # 無法序列化
        self.lock = lock_object  # 無法序列化
        
    def process(self, x):
        return x * 2

def test_serialization():
    """
    測試並展示序列化檢查工具的使用
    """
    # 測試簡單函式
    def simple_function(x):
        """這個函式可以正常序列化"""
        return x + 1
    
    print("=== 測試簡單函式 ===")
    ray.util.inspect_serializability(simple_function)
    
    # 測試捕獲了無法序列化物件的函式
    def problematic_function(x):
        """
        這個函式捕獲了無法序列化的 thread_pool
        inspect_serializability 會指出問題所在
        """
        def inner_function(y):
            # 嘗試使用 thread_pool - 這會導致序列化失敗
            return thread_pool.map(lambda z: z + y, range(x))
        
        return inner_function(x)
    
    print("\n=== 測試有問題的函式 ===")
    try:
        # 檢查序列化狀態
        # 這會輸出詳細的診斷資訊,指出 thread_pool 無法序列化
        ray.util.inspect_serializability(
            problematic_function, 
            name="problematic_function"
        )
    except Exception as e:
        print(f"序列化檢查發現問題: {e}")
    
    # 測試類別實例
    print("\n=== 測試類別實例 ===")
    instance = ComplexClass()
    
    try:
        # 檢查類別實例的序列化狀態
        # 會列出所有無法序列化的屬性
        ray.util.inspect_serializability(instance, name="ComplexClass_instance")
    except Exception as e:
        print(f"類別實例序列化檢查: {e}")
    
    # 展示如何修正序列化問題
    class FixedClass:
        """
        修正後的類別,移除了無法序列化的屬性
        """
        def __init__(self):
            self.normal_data = {"key": "value"}
            # 不儲存 Pool 或 Lock 物件
            # 如需這些物件,在方法內部建立
            
        def process(self, x):
            # 在需要時才建立 Pool
            with Pool(5) as local_pool:
                result = local_pool.map(lambda y: y * 2, range(x))
            return sum(result)
    
    print("\n=== 測試修正後的類別 ===")
    fixed_instance = FixedClass()
    ray.util.inspect_serializability(fixed_instance, name="FixedClass_instance")
    print("修正後的類別可以正常序列化")

def demonstrate_serialization_best_practices():
    """
    展示序列化的最佳實踐
    """
    # 最佳實踐 1: 只傳遞可序列化的簡單資料型態
    @ray.remote
    def good_practice_1(data: dict) -> dict:
        """
        只接受與回傳標準 Python 資料型態
        dict, list, tuple, str, int, float, bool 等都可以正常序列化
        """
        result = {
            "input_count": len(data),
            "processed": True
        }
        return result
    
    # 最佳實踐 2: 將無法序列化的物件在 Actor 內部建立
    @ray.remote
    class GoodPracticeActor:
        """
        在 Actor 內部建立與管理無法序列化的資源
        """
        def __init__(self):
            # Pool 在 Actor 初始化時建立
            # 不需要透過網路傳遞
            self.pool = Pool(4)
            
        def process(self, data):
            """使用內部的 Pool 進行處理"""
            return self.pool.map(lambda x: x * 2, data)
        
        def __del__(self):
            """清理資源"""
            if hasattr(self, 'pool'):
                self.pool.close()
                self.pool.join()
    
    # 最佳實踐 3: 使用 Ray 的 put/get 機制傳遞大型物件
    import numpy as np
    
    # 建立大型陣列
    large_array = np.random.rand(1000, 1000)
    
    # 將大型物件放入 Ray 的物件儲存
    # 這樣只需傳遞物件參考,而非實際資料
    array_ref = ray.put(large_array)
    
    @ray.remote
    def process_large_data(data_ref):
        """
        接收物件參考而非實際資料
        Ray 會自動處理資料的傳遞
        """
        # 從物件儲存中取得資料
        data = ray.get(data_ref)
        return np.sum(data)
    
    # 使用範例
    result_ref = process_large_data.remote(array_ref)
    result = ray.get(result_ref)
    print(f"大型資料處理結果: {result}")

if __name__ == "__main__":
    ray.init(local_mode=True)
    
    print("開始序列化診斷測試\n")
    test_serialization()
    
    print("\n\n開始最佳實踐示範\n")
    demonstrate_serialization_best_practices()
    
    ray.shutdown()

import ray
import pdb

# 啟用本地模式
# 所有任務都在主程序執行,可以使用一般的除錯工具
ray.init(local_mode=True)

@ray.remote
def calculate_statistics(data_list):
    """
    計算資料的統計量
    在本地模式下可以直接使用 pdb 設定中斷點
    """
    # 設定除錯中斷點
    # 在本地模式下會正常觸發
    pdb.set_trace()
    
    total = sum(data_list)
    count = len(data_list)
    average = total / count if count > 0 else 0
    
    return {
        'total': total,
        'count': count,
        'average': average
    }

@ray.remote
class DataProcessor:
    """
    資料處理器 Actor
    展示如何在 Actor 中進行本地除錯
    """
    def __init__(self):
        self.processed_count = 0
        
    def process_batch(self, batch_data):
        """
        批次處理資料
        """
        # 在關鍵點設定中斷
        if self.processed_count == 0:
            # 第一批資料時觸發除錯
            pdb.set_trace()
        
        results = []
        for item in batch_data:
            # 執行資料處理
            processed = item * 2
            results.append(processed)
        
        self.processed_count += len(batch_data)
        return results

def local_debugging_workflow():
    """
    本地除錯工作流程示範
    """
    # 準備測試資料
    test_data = [1, 2, 3, 4, 5]
    
    # 方法1: 直接執行遠端函式
    # 在本地模式下,這會同步執行
    print("=== 測試遠端函式 ===")
    result = ray.get(calculate_statistics.remote(test_data))
    print(f"統計結果: {result}")
    
    # 方法2: 使用 Actor
    print("\n=== 測試 Actor ===")
    processor = DataProcessor.remote()
    batch_result = ray.get(processor.process_batch.remote(test_data))
    print(f"批次處理結果: {batch_result}")

if __name__ == "__main__":
    try:
        local_debugging_workflow()
    finally:
        ray.shutdown()

"""
遠端除錯設定與使用指南

環境準備:
1. 在 Ray 叢集啟動時加入除錯器支援
   ray start --head --ray-debugger-external

2. 在程式碼中設定除錯中斷點
   使用 ray.util.pdb.set_trace() 替代標準的 pdb.set_trace()

3. 連線到除錯會話
   在另一個終端執行: ray debug
"""

import ray

# 連線到支援除錯的 Ray 叢集
# 這個設定會啟用遠端除錯功能
ray.init(address='auto')  # 在實際環境中使用叢集位址

@ray.remote
def complex_computation(data, threshold):
    """
    複雜運算函式
    展示如何在遠端執行時進行除錯
    """
    results = []
    
    for i, value in enumerate(data):
        # 當處理到特定條件時觸發除錯
        if value > threshold:
            # 使用 Ray 的除錯中斷點
            # 這會暫停遠端執行,等待除錯連線
            ray.util.pdb.set_trace()
        
        # 執行運算
        processed = value ** 2
        results.append(processed)
    
    return results

@ray.remote
class RemoteService:
    """
    遠端服務 Actor
    展示在長期執行的服務中如何除錯
    """
    def __init__(self):
        self.request_count = 0
        self.error_count = 0
        
    def handle_request(self, request_data):
        """
        處理請求
        """
        self.request_count += 1
        
        try:
            # 處理邏輯
            result = self._process(request_data)
            return {"status": "success", "data": result}
            
        except Exception as e:
            self.error_count += 1
            
            # 當錯誤累積到一定次數時觸發除錯
            if self.error_count >= 3:
                # 進入除錯模式檢查問題
                ray.util.pdb.set_trace()
            
            return {"status": "error", "message": str(e)}
    
    def _process(self, data):
        """內部處理方法"""
        # 模擬可能出錯的處理邏輯
        if not isinstance(data, dict):
            raise ValueError("資料格式錯誤")
        
        return {"processed": True, "value": data.get("value", 0) * 2}

def remote_debugging_example():
    """
    遠端除錯使用範例
    
    操作步驟:
    1. 執行此程式碼
    2. 當中斷點觸發時,程式會暫停
    3. 在另一個終端執行 'ray debug' 連線到除錯會話
    4. 可以檢查變數、執行命令、單步執行等
    """
    # 測試資料
    test_data = [1, 5, 3, 8, 2, 10, 4]
    threshold = 6
    
    # 啟動遠端運算
    # 當資料值大於 threshold 時會觸發除錯
    future = complex_computation.remote(test_data, threshold)
    
    print("遠端運算已啟動")
    print("如觸發中斷點,請在另一終端執行: ray debug")
    
    # 等待結果
    result = ray.get(future)
    print(f"運算結果: {result}")
    
    # 測試 Actor 的遠端除錯
    service = RemoteService.remote()
    
    # 發送幾個會觸發錯誤的請求
    for i in range(5):
        response = ray.get(service.handle_request.remote(i))
        print(f"請求 {i} 回應: {response}")

if __name__ == "__main__":
    # 注意: 這個範例需要在實際的 Ray 叢集上執行
    # 且叢集需要以 --ray-debugger-external 參數啟動
    try:
        remote_debugging_example()
    finally:
        ray.shutdown()

"""
PyCharm 遠端除錯整合設定

安裝需求:
pip install pydevd-pycharm

PyCharm 設定步驟:
1. 開啟 Run > Edit Configurations
2. 新增 Python Remote Debug 組態
3. 設定 Local host name: localhost
4. 設定 Port: 7779 (或自訂埠號)
5. 啟動除錯伺服器 (Run > Debug)
"""

import ray
import pydevd_pycharm
import os

@ray.remote
class DataAnalyzer:
    """
    資料分析 Actor
    整合 PyCharm 遠端除錯功能
    """
    def __init__(self, enable_debug=False, debug_host='localhost', debug_port=7779):
        """
        初始化 Actor
        
        參數說明:
            enable_debug: 是否啟用 PyCharm 除錯
            debug_host: PyCharm 除錯伺服器的主機位址
            debug_port: PyCharm 除錯伺服器的埠號
            
        設計考量:
            在生產環境中 enable_debug 應設為 False
            除錯功能僅在開發與測試環境啟用
        """
        self.data_cache = {}
        
        # 嘗試連接 PyCharm 除錯器
        if enable_debug:
            try:
                # 連接到 PyCharm 的除錯伺服器
                # stdoutToServer 與 stderrToServer 將輸出導向 PyCharm
                pydevd_pycharm.settrace(
                    debug_host,
                    port=debug_port,
                    stdoutToServer=True,
                    stderrToServer=True,
                    suspend=False  # 不立即暫停,繼續執行
                )
                print(f"成功連接 PyCharm 除錯器 ({debug_host}:{debug_port})")
                
            except ConnectionRefusedError:
                # 無法連接時優雅降級
                print("警告: 無法連接 PyCharm 除錯器,繼續正常執行")
            except Exception as e:
                print(f"除錯器連接錯誤: {str(e)}")
    
    def analyze_dataset(self, dataset_id, data):
        """
        分析資料集
        當連接到 PyCharm 時,可以在此設定中斷點
        
        參數說明:
            dataset_id: 資料集識別碼
            data: 要分析的資料
        """
        # 在 PyCharm 中可以在這裡設定中斷點
        # 程式執行到此處時會暫停,可以檢查變數
        
        # 快取資料
        self.data_cache[dataset_id] = data
        
        # 執行分析
        analysis_result = {
            'dataset_id': dataset_id,
            'record_count': len(data),
            'data_types': self._detect_types(data)
        }
        
        return analysis_result
    
    def _detect_types(self, data):
        """
        偵測資料型態分佈
        展示如何在私有方法中除錯
        """
        type_counts = {}
        
        for item in data:
            item_type = type(item).__name__
            type_counts[item_type] = type_counts.get(item_type, 0) + 1
        
        return type_counts

def pycharm_debugging_workflow():
    """
    PyCharm 除錯工作流程
    
    使用步驟:
    1. 在 PyCharm 中啟動 Remote Debug Server
    2. 執行此程式碼
    3. Actor 會自動連接到 PyCharm
    4. 在 PyCharm 中設定中斷點
    5. 呼叫 Actor 方法時會觸發中斷點
    """
    # 初始化 Ray
    ray.init(local_mode=False)
    
    # 建立啟用除錯的 Actor
    # 在開發環境設定環境變數 DEBUG_MODE=1
    enable_debug = os.getenv('DEBUG_MODE', '0') == '1'
    
    analyzer = DataAnalyzer.remote(
        enable_debug=enable_debug,
        debug_host='localhost',  # PyCharm 執行的機器
        debug_port=7779
    )
    
    # 準備測試資料
    test_data = [1, 2, 3, 'a', 'b', 4.5, 6.7]
    
    # 呼叫分析方法
    # 如果在 PyCharm 中設定了中斷點,執行會在此暫停
    result = ray.get(analyzer.analyze_dataset.remote('test_001', test_data))
    
    print(f"分析結果: {result}")
    
    ray.shutdown()

if __name__ == "__main__":
    print("PyCharm 除錯整合範例")
    print("請先在 PyCharm 中啟動 Remote Debug Server")
    print("然後設定環境變數 DEBUG_MODE=1 再執行此程式")
    
    pycharm_debugging_workflow()

"""
Ray 任務的效能分析

使用 cProfile 進行 CPU profiling:
python -m cProfile -o output.prof your_script.py

使用 memory_profiler 進行記憶體 profiling:
mprof run -E --include-children -o memory.dat python your_script.py
mprof plot memory.dat
"""

import ray
import time
import numpy as np
from functools import wraps

def profile_function(func):
    """
    函式執行時間分析裝飾器
    用於快速識別慢速函式
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        
        execution_time = end_time - start_time
        print(f"函式 {func.__name__} 執行時間: {execution_time:.4f} 秒")
        
        return result
    return wrapper

@ray.remote
class PerformanceAnalyzer:
    """
    效能分析 Actor
    展示如何在 Ray 任務中進行效能分析
    """
    def __init__(self):
        self.execution_stats = []
        
    @profile_function
    def cpu_intensive_task(self, size):
        """
        CPU 密集型任務
        用於測試 CPU profiling
        
        參數說明:
            size: 運算規模
        """
        # 建立大型矩陣
        matrix_a = np.random.rand(size, size)
        matrix_b = np.random.rand(size, size)
        
        # 矩陣乘法 - CPU 密集
        result = np.dot(matrix_a, matrix_b)
        
        # 統計運算
        stats = {
            'mean': np.mean(result),
            'std': np.std(result),
            'max': np.max(result),
            'min': np.min(result)
        }
        
        return stats
    
    @profile_function  
    def memory_intensive_task(self, iterations):
        """
        記憶體密集型任務
        用於測試記憶體 profiling
        
        參數說明:
            iterations: 迭代次數
        """
        data_accumulator = []
        
        for i in range(iterations):
            # 每次迭代建立新的大型陣列
            large_array = np.random.rand(10000, 100)
            
            # 累積資料(可能導致記憶體洩漏)
            data_accumulator.append(large_array)
            
            # 模擬處理延遲
            time.sleep(0.01)
        
        # 計算總記憶體使用量(概估)
        total_memory_mb = sum(arr.nbytes for arr in data_accumulator) / (1024 * 1024)
        
        return {
            'iterations': iterations,
            'arrays_created': len(data_accumulator),
            'total_memory_mb': total_memory_mb
        }
    
    def get_execution_stats(self):
        """取得執行統計資訊"""
        return self.execution_stats

def profiling_example():
    """
    效能分析範例
    展示如何使用不同的 profiling 工具
    """
    ray.init(local_mode=False)
    
    # 建立效能分析 Actor
    analyzer = PerformanceAnalyzer.remote()
    
    print("=== CPU 密集型任務分析 ===")
    cpu_result = ray.get(analyzer.cpu_intensive_task.remote(1000))
    print(f"運算統計: {cpu_result}")
    
    print("\n=== 記憶體密集型任務分析 ===")
    memory_result = ray.get(analyzer.memory_intensive_task.remote(100))
    print(f"記憶體使用: {memory_result}")
    
    ray.shutdown()

if __name__ == "__main__":
    profiling_example()

"""
Ray 容器退出碼處理

常見退出碼:
- 0: 正常結束
- 1: 一般錯誤
- 127: 命令找不到
- 130: 使用者中斷(Ctrl+C)
- 137: 記憶體不足被強制終止(OOM Kill)
- 139: 段錯誤(Segmentation Fault)
"""

import ray
import sys
import os
import signal

@ray.remote
def risky_operation(operation_type):
    """
    模擬各種可能導致不同退出碼的操作
    
    參數說明:
        operation_type: 操作類型
            'normal' - 正常執行
            'error' - 一般錯誤
            'oom' - 模擬記憶體不足
            'segfault' - 模擬段錯誤
    """
    if operation_type == 'normal':
        # 正常執行,退出碼 0
        return {"status": "success", "exit_code": 0}
    
    elif operation_type == 'error':
        # 一般錯誤,退出碼 1
        raise RuntimeError("模擬一般執行錯誤")
    
    elif operation_type == 'oom':
        # 模擬記憶體不足
        # 實際環境中會被 OOM Killer 終止,退出碼 137
        try:
            # 嘗試配置過大的記憶體
            import numpy as np
            huge_array = np.zeros((100000, 100000))  # 可能導致 OOM
        except MemoryError:
            print("記憶體配置失敗")
            sys.exit(137)
    
    elif operation_type == 'signal':
        # 模擬接收終止訊號
        # 退出碼 128 + 訊號編號
        os.kill(os.getpid(), signal.SIGTERM)

def handle_ray_task_errors():
    """
    處理 Ray 任務的各種錯誤情況
    """
    ray.init(local_mode=False)
    
    test_cases = ['normal', 'error']
    
    for test_case in test_cases:
        print(f"\n測試案例: {test_case}")
        
        try:
            # 提交任務
            future = risky_operation.remote(test_case)
            result = ray.get(future)
            print(f"任務成功完成: {result}")
            
        except ray.exceptions.RayTaskError as e:
            # Ray 任務執行錯誤
            print(f"任務執行失敗: {e}")
            
        except Exception as e:
            # 其他非預期錯誤
            print(f"發生非預期錯誤: {e}")
    
    ray.shutdown()

if __name__ == "__main__":
    handle_ray_task_errors()

"""
Ray 日誌系統使用指南

日誌位置:
- 主節點: /tmp/ray/session_latest/logs/
- 工作節點: 同樣路徑,但需要 SSH 或 kubectl exec 存取

重要日誌檔案:
- dashboard.log: Ray Dashboard 日誌
- gcs_server.log: Global Control Store 日誌
- raylet.log: Raylet 程序日誌
- worker-*.log: 工作程序日誌
"""

import ray
import logging
import os

# 設定 Python logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

@ray.remote
class LoggingActor:
    """
    展示如何在 Ray Actor 中正確使用日誌
    """
    def __init__(self, actor_id):
        """
        初始化 Actor
        
        參數說明:
            actor_id: Actor 識別碼,用於日誌追蹤
        """
        self.actor_id = actor_id
        
        # 在 Actor 內部建立 logger
        # 這些日誌會寫入對應的 worker 日誌檔
        self.logger = logging.getLogger(f"Actor-{actor_id}")
        self.logger.setLevel(logging.INFO)
        
        self.logger.info(f"Actor {actor_id} 初始化完成")
    
    def process_task(self, task_id, data):
        """
        處理任務並記錄詳細日誌
        
        參數說明:
            task_id: 任務識別碼
            data: 任務資料
        """
        self.logger.info(f"開始處理任務 {task_id}")
        
        try:
            # 模擬資料處理
            result = self._process_data(data)
            
            self.logger.info(
                f"任務 {task_id} 處理成功, "
                f"輸入筆數: {len(data)}, 輸出筆數: {len(result)}"
            )
            
            return {"status": "success", "result": result}
            
        except Exception as e:
            # 記錄錯誤詳情
            self.logger.error(
                f"任務 {task_id} 處理失敗: {str(e)}",
                exc_info=True  # 包含完整堆疊追蹤
            )
            raise
    
    def _process_data(self, data):
        """
        內部資料處理方法
        """
        # 記錄處理細節
        self.logger.debug(f"處理 {len(data)} 筆資料")
        
        # 模擬處理邏輯
        processed = [item * 2 for item in data]
        
        return processed

def access_ray_logs():
    """
    示範如何存取 Ray 日誌
    """
    ray.init(local_mode=False)
    
    # 取得 Ray 日誌目錄
    # 這會回傳當前會話的日誌目錄路徑
    log_dir = ray.get_runtime_context().get_logs_dir()
    print(f"Ray 日誌目錄: {log_dir}")
    
    # 列出日誌檔案
    if os.path.exists(log_dir):
        log_files = os.listdir(log_dir)
        print(f"日誌檔案列表: {log_files}")
    
    # 建立使用日誌的 Actor
    actor = LoggingActor.remote("actor-001")
    
    # 執行任務
    test_data = [1, 2, 3, 4, 5]
    result = ray.get(actor.process_task.remote("task-001", test_data))
    
    print(f"任務結果: {result}")
    print(f"\n請檢查以下位置的日誌檔案:")
    print(f"{log_dir}/worker-*.log")
    
    ray.shutdown()

if __name__ == "__main__":
    access_ray_logs()

#!/bin/bash
# Ray 容器啟動指令碼範例
# 包含錯誤處理與核心轉儲保存邏輯

# 設定核心轉儲檔案大小限制
ulimit -c unlimited

# 設定核心轉儲檔案路徑
echo "/tmp/cores/core.%e.%p.%t" > /proc/sys/kernel/core_pattern

# 定義錯誤處理函式
handle_error() {
    local exit_code=$?
    echo "Ray 任務失敗,退出碼: ${exit_code}"
    
    # 檢查是否產生核心轉儲檔案
    if [ -d "/tmp/cores" ]; then
        # 將核心轉儲檔案複製到持久化儲存
        # 例如 S3、NFS 或其他共享儲存
        aws s3 cp /tmp/cores/ s3://my-debug-bucket/cores/ --recursive
        echo "核心轉儲檔案已上傳到 S3"
    fi
    
    # 暫停一段時間以便進行除錯
    # 在 Kubernetes 環境中可以使用 kubectl exec 連線到容器
    sleep 300
    
    exit ${exit_code}
}

# 設定錯誤陷阱
trap handle_error ERR

# 啟動 Ray
ray start --head --port=6379 || handle_error

# 保持容器執行
tail -f /dev/null

"""
Kubernetes 環境中的 Ray 除錯組態
"""

# kubernetes_pod_spec.yaml
pod_spec_example = """
apiVersion: v1
kind: Pod
metadata:
  name: ray-head
spec:
  containers:
  - name: ray
    image: rayproject/ray:latest
    command: ["/bin/bash"]
    args: ["-c", "/scripts/ray-start.sh || sleep 100000"]
    
    # 設定終止訊息路徑
    # Kubernetes 會將此檔案內容作為 Pod 的終止訊息
    terminationMessagePath: /dev/termination-log
    terminationMessagePolicy: FallbackToLogsOnError
    
    # 掛載持久化儲存用於核心轉儲
    volumeMounts:
    - name: core-dumps
      mountPath: /tmp/cores
    - name: scripts
      mountPath: /scripts
      
  volumes:
  - name: core-dumps
    persistentVolumeClaim:
      claimName: ray-core-dumps-pvc
  - name: scripts
    configMap:
      name: ray-scripts
      defaultMode: 0755
"""

# 在 Python 程式碼中記錄終止資訊
import sys
import traceback

def write_termination_message(message):
    """
    寫入終止訊息到 Kubernetes 指定的路徑
    這個訊息會在 Pod 失敗時顯示在 kubectl describe 輸出中
    """
    try:
        with open('/dev/termination-log', 'w') as f:
            f.write(message)
    except Exception as e:
        print(f"無法寫入終止訊息: {e}")

@ray.remote
def critical_task():
    """
    關鍵任務,失敗時記錄詳細資訊
    """
    try:
        # 執行任務邏輯
        result = perform_complex_operation()
        return result
        
    except Exception as e:
        # 建立詳細的錯誤訊息
        error_details = {
            'error_type': type(e).__name__,
            'error_message': str(e),
            'traceback': traceback.format_exc()
        }
        
        # 寫入終止訊息
        termination_msg = f"任務失敗: {error_details}"
        write_termination_message(termination_msg)
        
        # 重新拋出例外
        raise

def perform_complex_operation():
    """模擬複雜操作"""
    # 實際的業務邏輯
    pass