Ceph分布式存储系统源码剖析 - API使用指南与实战经验

graph TB
    subgraph "Ceph API生态系统"
        subgraph "应用层"
            WebApp[Web应用]
            MobileApp[移动应用]
            BigData[大数据平台]
            Backup[备份系统]
        end
        
        subgraph "SDK和框架层"
            PythonSDK[Python SDK]
            JavaSDK[Java SDK]
            GoSDK[Go SDK]
            S3SDK[AWS S3 SDK]
        end
        
        subgraph "协议接口层"
            S3API[S3 RESTful API]
            SwiftAPI[Swift API]
            POSIXAPI[POSIX API]
            RADOSAPI[RADOS API]
        end
        
        subgraph "网关和服务层"
            RGW[RGW Gateway<br/>对象存储网关]
            CephFS[CephFS Client<br/>文件系统客户端]
            RBDClient[RBD Client<br/>块设备客户端]
        end
        
        subgraph "核心库层"
            LibRADOS[librados<br/>原生客户端库]
            LibRBD[librbd<br/>块设备库]
            LibCephFS[libcephfs<br/>文件系统库]
        end
        
        subgraph "集群核心"
            RADOS[RADOS集群<br/>可靠分布式对象存储]
        end
    end
    
    WebApp --> S3SDK
    MobileApp --> S3SDK
    BigData --> PythonSDK
    Backup --> S3SDK
    
    PythonSDK --> RADOSAPI
    JavaSDK --> S3API
    GoSDK --> RADOSAPI
    S3SDK --> S3API
    
    S3API --> RGW
    SwiftAPI --> RGW
    POSIXAPI --> CephFS
    RADOSAPI --> LibRADOS
    
    RGW --> LibRADOS
    CephFS --> LibCephFS
    RBDClient --> LibRBD
    
    LibRADOS --> RADOS
    LibRBD --> RADOS
    LibCephFS --> RADOS

/**
 * LibRADOS核心API - 提供对RADOS集群的原生访问
 * 文件: src/librados/librados_cxx.cc
 * 
 * LibRADOS是Ceph的核心客户端库，提供：
 * 1. 集群连接和认证
 * 2. 存储池操作
 * 3. 对象CRUD操作
 * 4. 异步操作支持
 * 5. 集群管理功能
 */

/**
 * Rados类 - 集群连接和管理的主入口
 */
class Rados {
public:
    // ===================== 初始化和连接 =====================
    
    /**
     * 初始化Rados客户端
     * @param id 客户端ID，通常是"admin"
     * @return 0成功，负数为错误码
     * 
     * 示例：
     * librados::Rados cluster;
     * int ret = cluster.init("admin");
     * if (ret < 0) {
     *     std::cerr << "Failed to initialize cluster: " << ret << std::endl;
     *     return ret;
     * }
     */
    int init(const char * const id);
    
    /**
     * 连接到Ceph集群
     * @return 0成功，负数为错误码
     * 
     * 连接过程包括：
     * 1. 读取配置文件(/etc/ceph/ceph.conf)
     * 2. 连接到Monitor节点
     * 3. 获取集群映射信息
     * 4. 建立与OSD的连接
     */
    int connect();
    
    /**
     * 断开集群连接并清理资源
     */
    void shutdown();
    
    // ===================== 存储池操作 =====================
    
    /**
     * 创建存储池
     * @param name 存储池名称
     * @param auid 用户ID（可选）
     * @return 0成功，负数为错误码
     */
    int pool_create(const char *name, uint64_t auid = 0);
    
    /**
     * 删除存储池
     * @param name 存储池名称
     * @return 0成功，负数为错误码
     */
    int pool_delete(const char *name);
    
    /**
     * 列出所有存储池
     * @param pools 输出存储池列表
     * @return 0成功，负数为错误码
     */
    int pool_list(std::list<std::string>& pools);
    
    /**
     * 创建IO上下文
     * @param pool_name 存储池名称
     * @param ioctx 输出IO上下文
     * @return 0成功，负数为错误码
     */
    int ioctx_create(const char *pool_name, IoCtx& ioctx);

private:
    RadosClient *client;                // 客户端实现
};

/**
 * IoCtx类 - 存储池IO操作上下文
 */
class IoCtx {
public:
    // ===================== 基本对象操作 =====================
    
    /**
     * 写入对象（覆盖）
     * @param oid 对象ID
     * @param bl 要写入的数据
     * @param len 数据长度
     * @param off 写入偏移量
     * @return 0成功，负数为错误码
     * 
     * 示例：
     * bufferlist bl;
     * bl.append("Hello, Ceph!");
     * int ret = ioctx.write_full("my_object", bl);
     */
    int write(const std::string& oid, bufferlist& bl, size_t len, uint64_t off);
    int write_full(const std::string& oid, bufferlist& bl);
    
    /**
     * 读取对象
     * @param oid 对象ID  
     * @param bl 输出数据缓冲
     * @param len 读取长度
     * @param off 读取偏移量
     * @return 实际读取字节数，负数为错误码
     */
    int read(const std::string& oid, bufferlist& bl, size_t len, uint64_t off);
    
    /**
     * 删除对象
     * @param oid 对象ID
     * @return 0成功，负数为错误码
     */
    int remove(const std::string& oid);
    
    /**
     * 获取对象状态
     * @param oid 对象ID
     * @param psize 输出对象大小
     * @param pmtime 输出修改时间
     * @return 0成功，负数为错误码
     */
    int stat(const std::string& oid, uint64_t *psize, time_t *pmtime);

    // ===================== 异步操作 =====================
    
    /**
     * 异步写入对象
     * @param oid 对象ID
     * @param completion 完成回调
     * @param bl 要写入的数据
     * @param len 数据长度
     * @param off 写入偏移量
     * @return 0成功提交，负数为错误码
     */
    int aio_write(const std::string& oid, AioCompletion *c, 
                 const bufferlist& bl, size_t len, uint64_t off);
    
    /**
     * 异步读取对象
     * @param oid 对象ID
     * @param completion 完成回调  
     * @param bl 输出数据缓冲
     * @param len 读取长度
     * @param off 读取偏移量
     * @return 0成功提交，负数为错误码
     */
    int aio_read(const std::string& oid, AioCompletion *c,
                bufferlist *bl, size_t len, uint64_t off);

    // ===================== 扩展属性操作 =====================
    
    /**
     * 设置对象扩展属性
     * @param oid 对象ID
     * @param name 属性名
     * @param bl 属性值
     * @return 0成功，负数为错误码
     */
    int setxattr(const std::string& oid, const char *name, bufferlist& bl);
    
    /**
     * 获取对象扩展属性
     * @param oid 对象ID
     * @param name 属性名
     * @param bl 输出属性值
     * @return 属性值长度，负数为错误码
     */
    int getxattr(const std::string& oid, const char *name, bufferlist& bl);
    
    /**
     * 列出对象所有扩展属性
     * @param oid 对象ID
     * @param attrset 输出属性集合
     * @return 0成功，负数为错误码
     */
    int getxattrs(const std::string& oid, std::map<std::string, bufferlist>& attrset);

    // ===================== 对象列举操作 =====================
    
    /**
     * 开始对象列举
     * @param ctx 输出列举上下文
     * @return 0成功，负数为错误码
     */
    int objects_list_open(ObjectCursor& ctx);
    
    /**
     * 列举下一批对象
     * @param ctx 列举上下文
     * @param result 输出对象列表
     * @param finished 输出是否结束标志
     * @return 0成功，负数为错误码
     */
    int objects_list_next(ObjectCursor ctx, std::vector<ObjectItem> *result, bool *finished);

private:
    IoCtxImpl *io_ctx_impl;             // IO上下文实现
};

/**
 * AioCompletion类 - 异步操作完成回调
 */
class AioCompletion {
public:
    /**
     * 创建异步完成对象
     * @param cb_arg 回调参数
     * @param cb_complete 完成回调函数
     * @param cb_safe 安全回调函数（可选）
     * @return 异步完成对象
     */
    static AioCompletion *create(void *cb_arg, 
                                callback_t cb_complete, 
                                callback_t cb_safe = nullptr);
    
    /**
     * 等待操作完成
     */
    void wait_for_complete();
    
    /**
     * 等待操作安全完成（数据持久化）
     */
    void wait_for_safe();
    
    /**
     * 获取操作返回值
     * @return 操作结果
     */
    int get_return_value();
    
    /**
     * 释放完成对象
     */
    void release();

private:
    AioCompletionImpl *pc;              // 完成实现
};

/**
 * LibRADOS完整使用示例
 * 演示连接集群、创建存储池、对象操作等功能
 */

#include <rados/librados.hpp>
#include <iostream>
#include <string>

class RadosExample {
private:
    librados::Rados cluster;            // 集群连接
    librados::IoCtx io_ctx;             // IO上下文
    
public:
    /**
     * 初始化连接到Ceph集群
     * @param pool_name 存储池名称
     * @return 0成功，负数为错误码
     */
    int initialize(const std::string& pool_name) {
        int ret = 0;
        
        // 1. 初始化客户端
        ret = cluster.init("admin");    // 使用admin用户
        if (ret < 0) {
            std::cerr << "Failed to initialize cluster: " << strerror(-ret) << std::endl;
            return ret;
        }
        
        // 2. 读取配置文件
        ret = cluster.conf_read_file("/etc/ceph/ceph.conf");
        if (ret < 0) {
            std::cerr << "Failed to read config file: " << strerror(-ret) << std::endl;
            return ret;
        }
        
        // 3. 连接到集群
        ret = cluster.connect();
        if (ret < 0) {
            std::cerr << "Failed to connect to cluster: " << strerror(-ret) << std::endl;
            return ret;
        }
        
        std::cout << "Connected to Ceph cluster successfully!" << std::endl;
        
        // 4. 创建存储池（如果不存在）
        ret = cluster.pool_create(pool_name.c_str());
        if (ret < 0 && ret != -EEXIST) {
            std::cerr << "Failed to create pool: " << strerror(-ret) << std::endl;
            return ret;
        }
        
        // 5. 创建IO上下文
        ret = cluster.ioctx_create(pool_name.c_str(), io_ctx);
        if (ret < 0) {
            std::cerr << "Failed to create IO context: " << strerror(-ret) << std::endl;
            return ret;
        }
        
        std::cout << "IO context created for pool: " << pool_name << std::endl;
        return 0;
    }
    
    /**
     * 写入对象示例
     * @param object_name 对象名称
     * @param data 要写入的数据
     * @return 0成功，负数为错误码
     */
    int write_object(const std::string& object_name, const std::string& data) {
        // 准备数据缓冲
        librados::bufferlist bl;
        bl.append(data);
        
        // 写入对象（覆盖写入）
        int ret = io_ctx.write_full(object_name, bl);
        if (ret < 0) {
            std::cerr << "Failed to write object " << object_name 
                      << ": " << strerror(-ret) << std::endl;
            return ret;
        }
        
        std::cout << "Object " << object_name << " written successfully, "
                  << data.length() << " bytes" << std::endl;
        return 0;
    }
    
    /**
     * 读取对象示例
     * @param object_name 对象名称
     * @param data 输出读取的数据
     * @return 0成功，负数为错误码
     */
    int read_object(const std::string& object_name, std::string& data) {
        librados::bufferlist bl;
        
        // 读取整个对象
        int ret = io_ctx.read(object_name, bl, 0, 0);  // len=0表示读取全部
        if (ret < 0) {
            std::cerr << "Failed to read object " << object_name 
                      << ": " << strerror(-ret) << std::endl;
            return ret;
        }
        
        // 转换为字符串
        data = bl.to_str();
        
        std::cout << "Object " << object_name << " read successfully, "
                  << ret << " bytes" << std::endl;
        return 0;
    }
    
    /**
     * 异步写入示例
     * @param object_name 对象名称
     * @param data 要写入的数据
     * @return 0成功，负数为错误码
     */
    int async_write_object(const std::string& object_name, const std::string& data) {
        // 创建异步完成回调
        librados::AioCompletion *completion = librados::Rados::aio_create_completion(
            nullptr,  // callback argument
            [](librados::completion_t cb, void *arg) {
                // 完成回调函数
                librados::AioCompletion *comp = (librados::AioCompletion*)cb;
                int ret = comp->get_return_value();
                if (ret == 0) {
                    std::cout << "Async write completed successfully" << std::endl;
                } else {
                    std::cerr << "Async write failed: " << strerror(-ret) << std::endl;
                }
                comp->release();  // 释放资源
            },
            nullptr   // safe callback (可选)
        );
        
        // 准备数据
        librados::bufferlist bl;
        bl.append(data);
        
        // 提交异步写入
        int ret = io_ctx.aio_write_full(object_name, completion, bl);
        if (ret < 0) {
            std::cerr << "Failed to submit async write: " << strerror(-ret) << std::endl;
            completion->release();
            return ret;
        }
        
        std::cout << "Async write submitted for object: " << object_name << std::endl;
        return 0;
    }
    
    /**
     * 对象属性操作示例
     * @param object_name 对象名称
     * @return 0成功，负数为错误码
     */
    int manage_object_attrs(const std::string& object_name) {
        // 设置扩展属性
        librados::bufferlist attr_bl;
        attr_bl.append("example_value");
        
        int ret = io_ctx.setxattr(object_name, "user.example_attr", attr_bl);
        if (ret < 0) {
            std::cerr << "Failed to set xattr: " << strerror(-ret) << std::endl;
            return ret;
        }
        
        std::cout << "Set extended attribute for " << object_name << std::endl;
        
        // 获取扩展属性
        librados::bufferlist read_attr_bl;
        ret = io_ctx.getxattr(object_name, "user.example_attr", read_attr_bl);
        if (ret < 0) {
            std::cerr << "Failed to get xattr: " << strerror(-ret) << std::endl;
            return ret;
        }
        
        std::cout << "Extended attribute value: " << read_attr_bl.to_str() << std::endl;
        
        // 列出所有扩展属性
        std::map<std::string, librados::bufferlist> attr_map;
        ret = io_ctx.getxattrs(object_name, attr_map);
        if (ret < 0) {
            std::cerr << "Failed to list xattrs: " << strerror(-ret) << std::endl;
            return ret;
        }
        
        std::cout << "All extended attributes:" << std::endl;
        for (const auto& pair : attr_map) {
            std::cout << "  " << pair.first << " = " << pair.second.to_str() << std::endl;
        }
        
        return 0;
    }
    
    /**
     * 列举对象示例
     * @return 0成功，负数为错误码
     */
    int list_objects() {
        librados::ObjectCursor cur = io_ctx.object_list_begin();
        librados::ObjectCursor end = io_ctx.object_list_end();
        
        std::cout << "Objects in pool:" << std::endl;
        
        while (cur != end) {
            std::vector<librados::ObjectItem> results;
            int ret = io_ctx.object_list(cur, end, 100, {}, &results, &cur);
            if (ret < 0) {
                std::cerr << "Failed to list objects: " << strerror(-ret) << std::endl;
                return ret;
            }
            
            for (const auto& obj : results) {
                std::cout << "  Object: " << obj.oid 
                         << ", Size: " << obj.size 
                         << ", Modified: " << ctime(&obj.mtime) << std::endl;
            }
        }
        
        return 0;
    }
    
    /**
     * 清理资源
     */
    void cleanup() {
        io_ctx.close();
        cluster.shutdown();
        std::cout << "Connection closed" << std::endl;
    }
};

/**
 * 主函数 - 演示完整的使用流程
 */
int main() {
    RadosExample example;
    
    // 初始化连接
    if (example.initialize("test_pool") != 0) {
        return 1;
    }
    
    // 写入对象
    example.write_object("test_object", "Hello, Ceph World!");
    
    // 读取对象
    std::string data;
    example.read_object("test_object", data);
    std::cout << "Read data: " << data << std::endl;
    
    // 异步写入
    example.async_write_object("async_object", "Async data");
    
    // 管理对象属性
    example.manage_object_attrs("test_object");
    
    // 列举对象
    example.list_objects();
    
    // 清理资源
    example.cleanup();
    
    return 0;
}

graph TB
    subgraph "RGW S3 API架构"
        subgraph "HTTP层"
            HTTPRequest[HTTP Request]
            S3Parser[S3 Request Parser]
            AuthHandler[Authentication Handler]
        end
        
        subgraph "操作路由"
            ServiceHandler[Service Handler<br/>账户级操作]
            BucketHandler[Bucket Handler<br/>桶级操作]
            ObjectHandler[Object Handler<br/>对象级操作]
        end
        
        subgraph "操作实现"
            ListBuckets[List Buckets]
            CreateBucket[Create Bucket]
            DeleteBucket[Delete Bucket]
            PutObject[Put Object]
            GetObject[Get Object]
            DeleteObject[Delete Object]
            ListObjects[List Objects]
        end
        
        subgraph "存储抽象层"
            RGWStore[RGW Store<br/>存储抽象]
            RADOSStore[RADOS Store<br/>RADOS实现]
        end
        
        subgraph "底层存储"
            RADOS[RADOS集群]
        end
    end
    
    HTTPRequest --> S3Parser
    S3Parser --> AuthHandler
    AuthHandler --> ServiceHandler
    AuthHandler --> BucketHandler  
    AuthHandler --> ObjectHandler
    
    ServiceHandler --> ListBuckets
    BucketHandler --> CreateBucket
    BucketHandler --> DeleteBucket
    BucketHandler --> ListObjects
    ObjectHandler --> PutObject
    ObjectHandler --> GetObject
    ObjectHandler --> DeleteObject
    
    ListBuckets --> RGWStore
    CreateBucket --> RGWStore
    PutObject --> RGWStore
    GetObject --> RGWStore
    
    RGWStore --> RADOSStore
    RADOSStore --> RADOS

/**
 * RGW S3 API核心实现
 * 文件: src/rgw/rgw_rest_s3.h:698-764
 * 
 * RGW提供完整的S3兼容API，包括桶操作、对象操作等
 */

/**
 * RGWHandler_REST_Bucket_S3 - S3桶操作处理器
 */
class RGWHandler_REST_Bucket_S3 : public RGWHandler_REST_S3 {
public:
    // ===================== 桶操作检测方法 =====================
    
    /**
     * 检查是否为ACL操作
     * @return true 如果请求包含acl参数
     */
    bool is_acl_op() const {
        return s->info.args.exists("acl");
    }
    
    /**
     * 检查是否为CORS操作
     * @return true 如果请求包含cors参数
     */
    bool is_cors_op() const {
        return s->info.args.exists("cors");
    }
    
    /**
     * 检查是否为生命周期操作
     * @return true 如果请求包含lifecycle参数
     */
    bool is_lc_op() const {
        return s->info.args.exists("lifecycle");
    }
    
    /**
     * 检查是否为标签操作
     * @return true 如果请求包含tagging参数
     */
    bool is_tagging_op() const {
        return s->info.args.exists("tagging");
    }
    
    /**
     * 检查是否为策略操作
     * @return true 如果请求包含policy参数
     */
    bool is_policy_op() const {
        return s->info.args.exists("policy");
    }

    // ===================== HTTP方法处理 =====================
    
    /**
     * 处理GET请求
     * @return 对应的RGWOp操作对象
     */
    RGWOp *op_get() override {
        if (is_acl_op()) {
            return new RGWGetACLs_ObjStore_S3;
        } else if (s->info.args.exists("uploads")) {
            return new RGWListBucketMultiparts_ObjStore_S3;
        } else if (is_cors_op()) {
            return new RGWGetCORS_ObjStore_S3;
        } else if (is_tagging_op()) {
            return new RGWGetBucketTags_ObjStore_S3;
        } else if (is_lc_op()) {
            return new RGWGetLC_ObjStore_S3;
        } else if (is_policy_op()) {
            return new RGWGetBucketPolicy;
        } else {
            return new RGWListBucket_ObjStore_S3;  // 默认列出桶内容
        }
    }
    
    /**
     * 处理PUT请求
     * @return 对应的RGWOp操作对象
     */
    RGWOp *op_put() override {
        if (is_tagging_op()) {
            return new RGWPutBucketTags_ObjStore_S3;
        } else if (is_acl_op()) {
            return new RGWPutACLs_ObjStore_S3;
        } else if (is_cors_op()) {
            return new RGWPutCORS_ObjStore_S3;
        } else if (is_lc_op()) {
            return new RGWPutLC_ObjStore_S3;
        } else if (is_policy_op()) {
            return new RGWPutBucketPolicy;
        } else {
            return new RGWCreateBucket_ObjStore_S3;  // 默认创建桶
        }
    }
    
    /**
     * 处理DELETE请求
     * @return 对应的RGWOp操作对象
     */
    RGWOp *op_delete() override {
        if (is_tagging_op()) {
            return new RGWDeleteBucketTags_ObjStore_S3;
        } else if (is_cors_op()) {
            return new RGWDeleteCORS_ObjStore_S3;
        } else if (is_lc_op()) {
            return new RGWDeleteLC_ObjStore_S3;
        } else if (is_policy_op()) {
            return new RGWDeleteBucketPolicy;
        } else {
            return new RGWDeleteBucket_ObjStore_S3;  // 默认删除桶
        }
    }
};

/**
 * RGWCreateBucket_ObjStore_S3 - S3创建桶操作
 */
class RGWCreateBucket_ObjStore_S3 : public RGWCreateBucket_ObjStore {
public:
    /**
     * 执行创建桶操作
     * @return 操作结果
     */
    int execute() override {
        // 1. 解析请求参数
        parse_location_constraint();
        parse_canned_acl();
        
        // 2. 检查桶名称合法性
        if (!validate_bucket_name(s->bucket_name)) {
            return -EINVAL;
        }
        
        // 3. 检查权限
        if (!verify_bucket_permission(s, RGW_PERM_WRITE)) {
            return -EACCES;
        }
        
        // 4. 调用基类执行创建
        int ret = RGWCreateBucket_ObjStore::execute();
        if (ret < 0) {
            return ret;
        }
        
        // 5. 设置响应头
        if (ret == 0) {
            dump_location_header();
        }
        
        return ret;
    }
    
    /**
     * 发送HTTP响应
     */
    void send_response() override {
        if (op_ret == 0) {
            dump_location_header();
        }
        
        set_req_state_err(s, op_ret);
        dump_errno(s);
        end_header(s, this, "application/xml");
        
        if (op_ret == 0) {
            rgw_flush_formatter_and_reset(s, s->formatter);
        }
    }

private:
    /**
     * 解析位置约束
     */
    void parse_location_constraint() {
        // 解析请求体中的位置约束XML
        RGWXMLDecoder::XMLParser parser;
        if (!parser.init()) {
            return;
        }
        
        char* buf = nullptr;
        int len = 0;
        int ret = rgw_rest_read_all_input(s, &buf, &len);
        if (ret < 0) {
            return;
        }
        
        // 解析XML获取region信息
        if (!parser.parse(buf, len, 1)) {
            free(buf);
            return;
        }
        
        free(buf);
    }
    
    /**
     * 转储位置头信息
     */
    void dump_location_header() {
        std::string location = "/" + s->bucket_name;
        dump_header(s, "Location", location);
    }
};

/**
 * RGWListBucket_ObjStore_S3 - S3列出桶内容操作
 */
class RGWListBucket_ObjStore_S3 : public RGWListBucket_ObjStore {
public:
    /**
     * 执行列出操作
     * @return 操作结果
     */
    int execute() override {
        // 1. 解析查询参数
        parse_list_params();
        
        // 2. 执行列出操作
        int ret = RGWListBucket_ObjStore::execute();
        if (ret < 0) {
            return ret;
        }
        
        return 0;
    }
    
    /**
     * 发送响应
     */
    void send_response() override {
        if (op_ret < 0) {
            set_req_state_err(s, op_ret);
        } else {
            dump_start(s);
        }
        
        dump_errno(s);
        end_header(s, this, "application/xml");
        
        if (op_ret < 0) {
            return;
        }
        
        // 输出XML响应
        s->formatter->open_object_section_in_ns("ListBucketResult", 
                                               XMLNS_AWS_S3);
        
        s->formatter->dump_string("Name", s->bucket_name);
        s->formatter->dump_string("Prefix", prefix);
        s->formatter->dump_string("Marker", marker.name);
        s->formatter->dump_int("MaxKeys", max);
        s->formatter->dump_bool("IsTruncated", is_truncated);
        
        // 输出对象列表
        for (const auto& entry : objs) {
            s->formatter->open_object_section("Contents");
            s->formatter->dump_string("Key", entry.key.name);
            s->formatter->dump_string("LastModified", 
                                    entry.meta.mtime.to_iso_8601());
            s->formatter->dump_string("ETag", entry.meta.etag);
            s->formatter->dump_int("Size", entry.meta.accounted_size);
            s->formatter->dump_string("StorageClass", "STANDARD");
            s->formatter->close_section();
        }
        
        // 输出通用前缀
        for (const auto& prefix : common_prefixes) {
            s->formatter->open_object_section("CommonPrefixes");
            s->formatter->dump_string("Prefix", prefix.first);
            s->formatter->close_section();
        }
        
        s->formatter->close_section();
        rgw_flush_formatter_and_reset(s, s->formatter);
    }

private:
    /**
     * 解析列出参数
     */
    void parse_list_params() {
        // 解析查询参数
        prefix = s->info.args.get("prefix");
        delimiter = s->info.args.get("delimiter");
        marker.name = s->info.args.get("marker");
        
        string max_keys_str = s->info.args.get("max-keys");
        if (!max_keys_str.empty()) {
            max = atoi(max_keys_str.c_str());
            if (max > 1000) {
                max = 1000;  // S3限制
            }
        }
        
        // 解析list-type参数（支持v2 API）
        string list_type = s->info.args.get("list-type");
        if (list_type == "2") {
            // S3 List Objects v2
            fetch_owner = false;
            continuation_token = s->info.args.get("continuation-token");
            start_after = s->info.args.get("start-after");
        }
    }
};

"""
S3 API使用示例 - Python版本
使用boto3库访问Ceph RGW的S3兼容API
"""

import boto3
import botocore.config
from botocore.exceptions import ClientError
import json

class CephS3Client:
    def __init__(self, endpoint_url, access_key, secret_key, region='default'):
        """
        初始化S3客户端连接到Ceph RGW
        
        Args:
            endpoint_url: RGW的访问端点
            access_key: S3访问密钥ID
            secret_key: S3秘密访问密钥
            region: 区域名称
        """
        # 配置连接参数
        config = botocore.config.Config(
            region_name=region,
            signature_version='s3v4',  # 使用签名版本4
            s3={'addressing_style': 'path'}  # 使用路径风格
        )
        
        # 创建S3客户端
        self.s3_client = boto3.client(
            's3',
            endpoint_url=endpoint_url,
            aws_access_key_id=access_key,
            aws_secret_access_key=secret_key,
            config=config
        )
        
        # 创建S3资源（高级接口）
        self.s3_resource = boto3.resource(
            's3',
            endpoint_url=endpoint_url,
            aws_access_key_id=access_key,
            aws_secret_access_key=secret_key,
            config=config
        )
    
    def create_bucket(self, bucket_name, location=None):
        """
        创建存储桶
        
        Args:
            bucket_name: 桶名称
            location: 位置约束（可选）
            
        Returns:
            bool: 成功返回True，失败返回False
        """
        try:
            # 配置创建参数
            create_params = {'Bucket': bucket_name}
            
            if location:
                create_params['CreateBucketConfiguration'] = {
                    'LocationConstraint': location
                }
            
            # 创建桶
            response = self.s3_client.create_bucket(**create_params)
            print(f"Bucket '{bucket_name}' created successfully")
            print(f"Location: {response.get('Location', 'N/A')}")
            return True
            
        except ClientError as e:
            error_code = e.response['Error']['Code']
            if error_code == 'BucketAlreadyExists':
                print(f"Bucket '{bucket_name}' already exists")
                return True
            else:
                print(f"Error creating bucket: {e}")
                return False
    
    def list_buckets(self):
        """
        列出所有存储桶
        
        Returns:
            list: 桶信息列表
        """
        try:
            response = self.s3_client.list_buckets()
            buckets = response.get('Buckets', [])
            
            print("Available buckets:")
            for bucket in buckets:
                print(f"  - {bucket['Name']} (Created: {bucket['CreationDate']})")
            
            return buckets
            
        except ClientError as e:
            print(f"Error listing buckets: {e}")
            return []
    
    def put_object(self, bucket_name, object_key, data, metadata=None):
        """
        上传对象
        
        Args:
            bucket_name: 桶名称
            object_key: 对象键
            data: 对象数据（字符串或字节）
            metadata: 用户元数据字典（可选）
            
        Returns:
            bool: 成功返回True，失败返回False
        """
        try:
            # 准备上传参数
            put_params = {
                'Bucket': bucket_name,
                'Key': object_key,
                'Body': data
            }
            
            # 添加用户元数据
            if metadata:
                put_params['Metadata'] = metadata
            
            # 上传对象
            response = self.s3_client.put_object(**put_params)
            
            print(f"Object '{object_key}' uploaded to bucket '{bucket_name}'")
            print(f"ETag: {response.get('ETag', 'N/A')}")
            return True
            
        except ClientError as e:
            print(f"Error uploading object: {e}")
            return False
    
    def get_object(self, bucket_name, object_key):
        """
        下载对象
        
        Args:
            bucket_name: 桶名称
            object_key: 对象键
            
        Returns:
            tuple: (对象内容, 元数据) 或 (None, None)
        """
        try:
            response = self.s3_client.get_object(
                Bucket=bucket_name,
                Key=object_key
            )
            
            # 读取对象内容
            content = response['Body'].read()
            
            # 获取元数据
            metadata = {
                'ContentLength': response.get('ContentLength'),
                'ContentType': response.get('ContentType'),
                'LastModified': response.get('LastModified'),
                'ETag': response.get('ETag'),
                'Metadata': response.get('Metadata', {})
            }
            
            print(f"Object '{object_key}' downloaded from bucket '{bucket_name}'")
            print(f"Size: {metadata['ContentLength']} bytes")
            
            return content, metadata
            
        except ClientError as e:
            if e.response['Error']['Code'] == 'NoSuchKey':
                print(f"Object '{object_key}' not found")
            else:
                print(f"Error downloading object: {e}")
            return None, None
    
    def list_objects(self, bucket_name, prefix='', delimiter='', max_keys=1000):
        """
        列出桶中的对象
        
        Args:
            bucket_name: 桶名称
            prefix: 对象键前缀过滤
            delimiter: 分隔符
            max_keys: 最大返回对象数
            
        Returns:
            dict: 包含对象列表和通用前缀的字典
        """
        try:
            # 构建列出参数
            list_params = {
                'Bucket': bucket_name,
                'MaxKeys': max_keys
            }
            
            if prefix:
                list_params['Prefix'] = prefix
            if delimiter:
                list_params['Delimiter'] = delimiter
            
            # 列出对象
            response = self.s3_client.list_objects_v2(**list_params)
            
            # 解析响应
            objects = response.get('Contents', [])
            common_prefixes = response.get('CommonPrefixes', [])
            
            print(f"Objects in bucket '{bucket_name}':")
            for obj in objects:
                print(f"  - {obj['Key']} ({obj['Size']} bytes, {obj['LastModified']})")
            
            if common_prefixes:
                print("Common prefixes:")
                for cp in common_prefixes:
                    print(f"  - {cp['Prefix']}")
            
            return {
                'objects': objects,
                'common_prefixes': common_prefixes,
                'is_truncated': response.get('IsTruncated', False)
            }
            
        except ClientError as e:
            print(f"Error listing objects: {e}")
            return {'objects': [], 'common_prefixes': [], 'is_truncated': False}
    
    def delete_object(self, bucket_name, object_key):
        """
        删除对象
        
        Args:
            bucket_name: 桶名称
            object_key: 对象键
            
        Returns:
            bool: 成功返回True，失败返回False
        """
        try:
            self.s3_client.delete_object(
                Bucket=bucket_name,
                Key=object_key
            )
            
            print(f"Object '{object_key}' deleted from bucket '{bucket_name}'")
            return True
            
        except ClientError as e:
            print(f"Error deleting object: {e}")
            return False
    
    def put_bucket_policy(self, bucket_name, policy):
        """
        设置桶策略
        
        Args:
            bucket_name: 桶名称
            policy: 策略文档（字典格式）
            
        Returns:
            bool: 成功返回True，失败返回False
        """
        try:
            policy_json = json.dumps(policy)
            
            self.s3_client.put_bucket_policy(
                Bucket=bucket_name,
                Policy=policy_json
            )
            
            print(f"Bucket policy set for '{bucket_name}'")
            return True
            
        except ClientError as e:
            print(f"Error setting bucket policy: {e}")
            return False
    
    def get_bucket_policy(self, bucket_name):
        """
        获取桶策略
        
        Args:
            bucket_name: 桶名称
            
        Returns:
            dict: 策略文档或None
        """
        try:
            response = self.s3_client.get_bucket_policy(Bucket=bucket_name)
            policy = json.loads(response['Policy'])
            
            print(f"Bucket policy for '{bucket_name}':")
            print(json.dumps(policy, indent=2))
            
            return policy
            
        except ClientError as e:
            if e.response['Error']['Code'] == 'NoSuchBucketPolicy':
                print(f"No policy found for bucket '{bucket_name}'")
            else:
                print(f"Error getting bucket policy: {e}")
            return None

def main():
    """
    主函数 - 演示S3 API的完整使用流程
    """
    # 配置连接参数
    endpoint_url = 'http://rgw-endpoint:7480'  # RGW端点
    access_key = 'your-access-key'
    secret_key = 'your-secret-key'
    
    # 创建客户端
    client = CephS3Client(endpoint_url, access_key, secret_key)
    
    # 测试桶名称
    bucket_name = 'test-bucket'
    
    # 1. 创建桶
    print("=== Creating Bucket ===")
    client.create_bucket(bucket_name)
    
    # 2. 列出所有桶
    print("\n=== Listing Buckets ===")
    client.list_buckets()
    
    # 3. 上传对象
    print("\n=== Uploading Objects ===")
    test_data = "Hello, Ceph S3!"
    metadata = {
        'author': 'ceph-example',
        'content-type': 'text/plain'
    }
    client.put_object(bucket_name, 'test-file.txt', test_data, metadata)
    
    # 上传更多测试对象
    for i in range(5):
        client.put_object(bucket_name, f'files/test-{i}.txt', f'Content {i}')
    
    # 4. 列出对象
    print("\n=== Listing Objects ===")
    client.list_objects(bucket_name)
    
    print("\n=== Listing Objects with Prefix ===")
    client.list_objects(bucket_name, prefix='files/')
    
    # 5. 下载对象
    print("\n=== Downloading Object ===")
    content, metadata = client.get_object(bucket_name, 'test-file.txt')
    if content:
        print(f"Downloaded content: {content.decode()}")
        print(f"Metadata: {metadata}")
    
    # 6. 设置桶策略
    print("\n=== Setting Bucket Policy ===")
    policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "AllowPublicRead",
                "Effect": "Allow",
                "Principal": "*",
                "Action": "s3:GetObject",
                "Resource": f"arn:aws:s3:::{bucket_name}/*"
            }
        ]
    }
    client.put_bucket_policy(bucket_name, policy)
    
    # 7. 获取桶策略
    print("\n=== Getting Bucket Policy ===")
    client.get_bucket_policy(bucket_name)
    
    # 8. 清理 - 删除对象
    print("\n=== Cleanup - Deleting Objects ===")
    objects_to_delete = ['test-file.txt'] + [f'files/test-{i}.txt' for i in range(5)]
    for obj_key in objects_to_delete:
        client.delete_object(bucket_name, obj_key)
    
    print("\nS3 API demonstration completed!")

if __name__ == '__main__':
    main()

graph TB
    subgraph "CephFS API层次"
        subgraph "应用接口"
            POSIXApp[POSIX应用]
            CustomApp[自定义应用]
        end
        
        subgraph "文件系统接口"
            KernelClient[内核客户端<br/>ceph.ko]
            FUSEClient[FUSE客户端<br/>ceph-fuse]
            LibCephFS[libcephfs<br/>用户态库]
        end
        
        subgraph "网络协议层"
            MDSProtocol[MDS协议]
            OSDProtocol[OSD协议]
        end
        
        subgraph "服务层"
            MDS[MDS集群<br/>元数据服务]
            OSDs[OSD集群<br/>数据存储]
        end
    end
    
    POSIXApp --> KernelClient
    POSIXApp --> FUSEClient
    CustomApp --> LibCephFS
    
    KernelClient --> MDSProtocol
    FUSEClient --> MDSProtocol
    LibCephFS --> MDSProtocol
    
    MDSProtocol --> MDS
    OSDProtocol --> OSDs
    
    KernelClient --> OSDProtocol
    FUSEClient --> OSDProtocol
    LibCephFS --> OSDProtocol

/**
 * LibCephFS C API使用示例
 * 演示如何使用libcephfs进行文件系统操作
 */

#include <cephfs/libcephfs.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <errno.h>

/**
 * CephFS操作示例类
 */
typedef struct {
    struct ceph_mount_info *cmount;     // CephFS挂载句柄
} cephfs_example_t;

/**
 * 初始化CephFS连接
 * @param example 示例结构体
 * @param user_id 用户ID（如 "admin"）
 * @return 0成功，负数为错误码
 */
int cephfs_init(cephfs_example_t *example, const char *user_id) {
    int ret;
    
    // 1. 创建CephFS句柄
    ret = ceph_create(&example->cmount, user_id);
    if (ret) {
        fprintf(stderr, "Failed to create ceph mount: %s\n", strerror(-ret));
        return ret;
    }
    
    // 2. 读取配置文件
    ret = ceph_conf_read_file(example->cmount, "/etc/ceph/ceph.conf");
    if (ret) {
        fprintf(stderr, "Failed to read config file: %s\n", strerror(-ret));
        ceph_shutdown(example->cmount);
        return ret;
    }
    
    // 3. 挂载文件系统
    ret = ceph_mount(example->cmount, "/");  // 挂载到根目录
    if (ret) {
        fprintf(stderr, "Failed to mount cephfs: %s\n", strerror(-ret));
        ceph_shutdown(example->cmount);
        return ret;
    }
    
    printf("CephFS mounted successfully\n");
    return 0;
}

/**
 * 文件操作示例
 * @param example 示例结构体
 * @param filename 文件名
 * @param content 要写入的内容
 * @return 0成功，负数为错误码
 */
int cephfs_file_operations(cephfs_example_t *example, 
                          const char *filename, 
                          const char *content) {
    int ret;
    int fd;
    char buffer[1024];
    struct ceph_statx stx;
    
    printf("=== File Operations Example ===\n");
    
    // 1. 创建并写入文件
    printf("Creating file: %s\n", filename);
    fd = ceph_open(example->cmount, filename, O_CREAT | O_WRONLY | O_TRUNC, 0644);
    if (fd < 0) {
        fprintf(stderr, "Failed to create file %s: %s\n", 
                filename, strerror(-fd));
        return fd;
    }
    
    // 写入内容
    ret = ceph_write(example->cmount, fd, content, strlen(content), 0);
    if (ret < 0) {
        fprintf(stderr, "Failed to write to file: %s\n", strerror(-ret));
        ceph_close(example->cmount, fd);
        return ret;
    }
    
    printf("Wrote %d bytes to file\n", ret);
    ceph_close(example->cmount, fd);
    
    // 2. 读取文件
    printf("Reading file: %s\n", filename);
    fd = ceph_open(example->cmount, filename, O_RDONLY, 0);
    if (fd < 0) {
        fprintf(stderr, "Failed to open file for read: %s\n", strerror(-fd));
        return fd;
    }
    
    memset(buffer, 0, sizeof(buffer));
    ret = ceph_read(example->cmount, fd, buffer, sizeof(buffer) - 1, 0);
    if (ret < 0) {
        fprintf(stderr, "Failed to read file: %s\n", strerror(-ret));
        ceph_close(example->cmount, fd);
        return ret;
    }
    
    printf("Read %d bytes: %s\n", ret, buffer);
    ceph_close(example->cmount, fd);
    
    // 3. 获取文件状态
    printf("Getting file stats: %s\n", filename);
    ret = ceph_statx(example->cmount, filename, &stx, 
                     CEPH_STATX_BASIC_STATS, AT_STATX_SYNC_AS_STAT);
    if (ret) {
        fprintf(stderr, "Failed to stat file: %s\n", strerror(-ret));
        return ret;
    }
    
    printf("File stats:\n");
    printf("  Size: %lu bytes\n", stx.stx_size);
    printf("  Mode: 0%o\n", stx.stx_mode);
    printf("  UID: %u, GID: %u\n", stx.stx_uid, stx.stx_gid);
    printf("  Access time: %lu.%09lu\n", stx.stx_atime.tv_sec, stx.stx_atime.tv_nsec);
    printf("  Modify time: %lu.%09lu\n", stx.stx_mtime.tv_sec, stx.stx_mtime.tv_nsec);
    
    return 0;
}

/**
 * 目录操作示例
 * @param example 示例结构体
 * @param dirname 目录名
 * @return 0成功，负数为错误码
 */
int cephfs_directory_operations(cephfs_example_t *example, const char *dirname) {
    int ret;
    struct ceph_dir_result *dirp;
    struct dirent *entry;
    
    printf("=== Directory Operations Example ===\n");
    
    // 1. 创建目录
    printf("Creating directory: %s\n", dirname);
    ret = ceph_mkdir(example->cmount, dirname, 0755);
    if (ret && ret != -EEXIST) {
        fprintf(stderr, "Failed to create directory %s: %s\n", 
                dirname, strerror(-ret));
        return ret;
    }
    
    // 2. 在目录中创建一些文件
    char filepath[256];
    for (int i = 0; i < 3; i++) {
        snprintf(filepath, sizeof(filepath), "%s/file%d.txt", dirname, i);
        
        int fd = ceph_open(example->cmount, filepath, O_CREAT | O_WRONLY | O_TRUNC, 0644);
        if (fd < 0) {
            fprintf(stderr, "Failed to create %s: %s\n", filepath, strerror(-fd));
            continue;
        }
        
        char content[64];
        snprintf(content, sizeof(content), "Content of file %d", i);
        ceph_write(example->cmount, fd, content, strlen(content), 0);
        ceph_close(example->cmount, fd);
    }
    
    // 3. 列出目录内容
    printf("Listing directory contents: %s\n", dirname);
    ret = ceph_opendir(example->cmount, dirname, &dirp);
    if (ret) {
        fprintf(stderr, "Failed to open directory %s: %s\n", 
                dirname, strerror(-ret));
        return ret;
    }
    
    printf("Directory contents:\n");
    while ((entry = ceph_readdir(example->cmount, dirp)) != NULL) {
        if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0) {
            continue;
        }
        printf("  %s (type: %d)\n", entry->d_name, entry->d_type);
    }
    
    ceph_closedir(example->cmount, dirp);
    
    return 0;
}

/**
 * 扩展属性操作示例
 * @param example 示例结构体
 * @param filepath 文件路径
 * @return 0成功，负数为错误码
 */
int cephfs_xattr_operations(cephfs_example_t *example, const char *filepath) {
    int ret;
    char value[256];
    char list[1024];
    
    printf("=== Extended Attributes Example ===\n");
    
    // 1. 设置扩展属性
    const char *attr_name = "user.example_attr";
    const char *attr_value = "example_value";
    
    printf("Setting extended attribute: %s = %s\n", attr_name, attr_value);
    ret = ceph_setxattr(example->cmount, filepath, attr_name, 
                       attr_value, strlen(attr_value), 0);
    if (ret) {
        fprintf(stderr, "Failed to set xattr: %s\n", strerror(-ret));
        return ret;
    }
    
    // 2. 获取扩展属性
    printf("Getting extended attribute: %s\n", attr_name);
    ret = ceph_getxattr(example->cmount, filepath, attr_name, 
                       value, sizeof(value) - 1);
    if (ret < 0) {
        fprintf(stderr, "Failed to get xattr: %s\n", strerror(-ret));
        return ret;
    }
    
    value[ret] = '\0';
    printf("Extended attribute value: %s\n", value);
    
    // 3. 列出所有扩展属性
    printf("Listing all extended attributes:\n");
    ret = ceph_listxattr(example->cmount, filepath, list, sizeof(list) - 1);
    if (ret < 0) {
        fprintf(stderr, "Failed to list xattrs: %s\n", strerror(-ret));
        return ret;
    }
    
    // 解析属性列表（以null分隔）
    char *attr = list;
    while (attr < list + ret) {
        printf("  Attribute: %s\n", attr);
        attr += strlen(attr) + 1;
    }
    
    return 0;
}

/**
 * 清理资源
 * @param example 示例结构体
 */
void cephfs_cleanup(cephfs_example_t *example) {
    if (example->cmount) {
        printf("Unmounting CephFS\n");
        ceph_unmount(example->cmount);
        ceph_shutdown(example->cmount);
        example->cmount = NULL;
    }
}

/**
 * 主函数 - 演示CephFS API的完整使用
 */
int main(int argc, char **argv) {
    cephfs_example_t example = {0};
    int ret;
    
    const char *user_id = (argc > 1) ? argv[1] : "admin";
    
    printf("CephFS libcephfs API Example\n");
    printf("Using user ID: %s\n", user_id);
    
    // 初始化CephFS连接
    ret = cephfs_init(&example, user_id);
    if (ret) {
        return 1;
    }
    
    // 文件操作示例
    ret = cephfs_file_operations(&example, "/test_file.txt", "Hello, CephFS!");
    if (ret) {
        cephfs_cleanup(&example);
        return 1;
    }
    
    // 目录操作示例
    ret = cephfs_directory_operations(&example, "/test_directory");
    if (ret) {
        cephfs_cleanup(&example);
        return 1;
    }
    
    // 扩展属性操作示例
    ret = cephfs_xattr_operations(&example, "/test_file.txt");
    if (ret) {
        cephfs_cleanup(&example);
        return 1;
    }
    
    printf("\nAll operations completed successfully!\n");
    
    // 清理资源
    cephfs_cleanup(&example);
    
    return 0;
}