Bookeeper: A scalable, fault-tolerant, and low-latency storage service optimized for real-time workloads

Introducation

• HDFS：Namenode用于WAL

• Twitter Manhattan: 一个通用的KV存储系统，用于WAL的存储

• Herddb：一个提供SQL语义的分布式数据库，用于WAL的存储

• Apache Pulsar: 一个分布式云原生的流式消息队列，用于消息和Offset的存储

• Object/Blob Storage: 比如可以利用bk的多副本和一致性存储快照

Movation

• 高效写入

• 通过将数据写入一组存储节点提供容错能力

• 通过将数据条带化可以将数据写入到多组bookies，从而提高了写入的吞吐

Basic Terms

Architecture

message LedgerMetadataFormat {
    required int32 quorumSize = 1;      // 写入的成功需要的bookie个数
    required int32 ensembleSize = 2;    // ensemble包含的bookie个数
    required int64 length = 3;          // entry个数？
    optional int64 lastEntryId = 4;     // 最后一个entry的id
    enum State {
        OPEN = 1;
        IN_RECOVERY = 2;
        CLOSED = 3;
    }
    // 当前ledger的状态
    // Open: 正在写入
    // IN_RECOVERY: 修复中，可能是数据损坏也可能是fencing过程中
    // Closed: 改ledger已经被关闭，不能写入
    required State state = 5 [default = OPEN];  
    // 文档里说segment保存Bookie列表或者firstEntryId
    // 只保存firstEntry怎么找到需要写入的Bookies呢？
    // 或者意思是对于第一个Segment，则只填充列表
    message Segment {
        repeated string ensembleMember = 1;  // bookie的地址
        required int64 firstEntryId = 2;     // 第一个entry的id
    }
    // 当前ledger的所有segment
    repeated Segment segment = 6;
    enum DigestType {
        CRC32 = 1;
        HMAC = 2;
        CRC32C = 3;
        DUMMY = 4;
    }
    optional DigestType digestType = 7;
    optional bytes password = 8;
    // 对于上面的quorumSize，写入ack成功的bookie即认为写入成功
    optional int32 ackQuorumSize = 9;
    optional int64 ctime = 10;
    message cMetadataMapEntry {
        optional string key = 1;
        optional bytes value = 2;
    }
    repeated cMetadataMapEntry customMetadata = 11;
    optional int64 cToken = 12;
}

{level1 (2 digits)}{level2 (4 digits)}{level3 (4 digits)}

{ledgers_root_path}/{level1}/{level2}/L{level3}

{ledgers_root_path}/00/0000/L0001

• 该Fragment的第一个EntryId

• 或改Fragment保存的Bookies列表，改列表的size等于Ensemble Size

• 对Ledger里的Entry的读取顺序和写入顺序一致

• 所有client读到的顺序也都是一致的

1. Writer Client去Open一个新的Ledger，实际上是在zk创建一个Ledger的元数据并且为其分配一个Ensemble，作为其第一个Fragment，此时Ledger的状态是OPEN

2. 写者按照前面说的轮询规则从Ensemble中选择Qw个Bookie节点发送写入Entry的请求，并且允许并发的发送多个Entry的写入请求，每个Entry会携带Ledger Id, Entry Id, LastConfirmed Entry ID和用户的数据

3. 在收到Qw中Qa个节点的写入成功响应后，client向用户返回写入成功，但是一定会确保小于该Entry ID 的Entry先响应用户，响应给用户的Entry称之为Committed Entry

4. Writer Client按照2、3步骤继续写入新的数据，并且会将当前认为的已经Commited的Entryid携带过去，这样读者以及恢复流程就可以尽快知道已经Commited的Entry id是多少

5. 一旦写者在写入该Ensemble出错的时候，会切换一个先的Ensemble写入，具体是创建一个新的Fragement，替换故障的Bookie生成一个新的Ensemble并一并写入到Ledger的元数据，然后继续写入Entry。创建Fragment需要确保没有其他写者在修改元数据，所以使用CAS进行修改，如果有其他人已经修改过了元数据，那么就会写入失败，可能是有修复或者Recovery产生的。这个时候会重新读取元数据，如果ledger的状态不再是OPEN ，那么就像客户端发送写入失败的错误

1. 通过CAS操作修改Ledger的状态为CLOSED状态，并且将最新Commited的EntryID写入元数据

2. 如果CAS失败，则说明可能有其他人修改过元数据。那么会重新读取元数据，然后根据状态执行下面的操作：
1. 如果Ledger的状态依然是OPEN状态那么就会一直尝试关闭
2. 如果是IN_RECOVERY，则直接返回用户错误
3. 如果是CLOSED并且元数据里的last entry id和其认为的一样，那么返回用户close成功，如果不同则返回用户close失败

1. CAS更新Ledger的状态为IN_RECOVERY

2. 然后向该Ledger最后一个Fragment所有的Bookies发送Fence请求，要求其Fence该Ledger，Bookies在收到请求后会将对应Ledger的Fence状态持久化到磁盘

3. 因为一个Fragment里的Entry可能写入到Ensemble里任意Qw个Bookies节点，共有C(E,Qw)个Quorum组合，当Client收到每个组合里至少（Qw -Qa）+1个fence成功的响应后就认为Ledger Fence成功

1. 向最后一个fragment的Ensemble里的所有Bookies发送请求获得其最后一个Commit的Entry(LastAddConfirmed)，前面我们说过在写入Entry的时候会携带写者认为的已经Commit的Entry，但是这个信息毕竟是滞后的，新的写者是无法确定真正的最后一个Commit的Entry的

2. Client确保收到每一个Quorum组合至少（Qw-Qa）+1个Booki响应后，最大的一个LAC以及之前的Entry可以确定一定是已经复制到了至少Qa个节点，并且已经返回用户写入成功

3. 从LAC+1的Entry Id开始向所有的Bookie发送读取Eentry请求，将读取到的Entry全部写入到其所在的Quorum里

4. 然后增加Entry ID 继续读取直到读不到未知，到此EntryID作为该Ledger的最后一个Entry ID，携带该信息去更新Ledger的状态为CLOSED

5. 这个过程中可能有多个Reader去尝试Recovery流程，但是因为第1步通过CAS操作将Ledger状态修改为了IN_RECOVERY，所以其他Reader一定会失败

1. 新的Reader首先读取Ledger的Meta，然后CAS修改其状态为IN_RECOVERY ，然后向所有Bookie发送Fence请求，确保Bookie Fence成功

2. 向所有Bookie请求他们看到的LAC(Last Add Confirmed)Entry ID，这里Looger1返回9，logger2返回10，logger3返回11，选择最大的11作为当前Ledger的LAC

3. 从12开始去读取所有的Bookie，这里Logger1和Logger3返回了，并且是2个，不需要重新写入到其他Bookie，可能12已经响应过client写入成功

4. 继续读取Entry 13，没有Bookie返回存在，说明没有写入过Entry 13，将12作为Ledger的Last Entry去更新元数据的状态为CLOSED

5. 这个过程中如果老的写者来Add Entry 12，那么因为改Ledger已经被Fence，则会写入失败，老的写者就知道自己已经不是权威的写者了

• Auditor：通过zk进行选主，只有一个节点可以成为Auditor，主要负责监控已经注册在zk上的Bookie是否有故障，然后如果发现故障则会扫描所有的Ledger找出存储在其上面的ledger，然后为每个Ledger生成修复任务放在zk的/underreplicated节点下

• Replication Worker：Replication worker负责watch zk节点/underreplicated下的task，如果发现有新的task生成，那么会尝试对该task加锁，如果不能抢到锁，那么就继续下一个task。抢到锁之后，扫描改task里的Legder的Fragment查看其Ensemble里是否包含他自己，如果不包含的话可以将该Fragement里的Entry复制到其所在的Bookie上。复制完成后则将其该Task从/underreplicated下删除，然后释放锁

1. The client goes through all ledger fragments in the ledger, selecting those that contain the failed bookie. 扫描Ledger所有的Fragment，然后找到包含故障bookie的Fragment

2. 然后开始修复改Fragment：
1. 选择一个Bookie将该Fragement的所有Entry复制到上面；如果AutoRecovery进程和Bookie同机部署，通常情况下就是本节点
2. 读取所有属于该Fragement的Entry，然后写入到选择的bookie节点
3. 一旦所有的Entry复制完成，就会更新该Fragment的meta以及其Ensemble
4. 标记该Fragment已经修复完成

3. 重复1、2将该Ledger所有的Fragment都修复完成后，这个ledger就可以认为修复完成从zk的/underreplicated下移除并释放锁

Reference

1. https://bookkeeper.apache.org/docs/latest/overview/overview/

2. Durability with BookKeeper

3. DistributedLog: A High Performance Replicated Log Service

4. Fencing - Apache BookKeeper - Apache Software Foundation