Ceph OSD写操作失效如何处理

很多人对Ceph写操作的异常处理的过程还不是特别的清楚。本文就介绍Ceph如何处理异常处理的。

首先需要明确的是，Ceph的读写操作并没有超时机制。 rbd_write并没有超时机制。所有经常看到，用ceph -s 命令查看，有些 slow request请求会显示延迟 30s甚至更长的时间。

当然，osd有一个机制，会检查某个线程执行时间，如果执行时间过长，osd就会触发自杀机制（osd suicide timeout）。这个机制默认关闭。

下面就研究一下ceph如何处理写操作的异常。

正常的写流程

在OSD端正常的写操作流程中，在函数ReplicatedBackend::submit_transaction把请求加入到in_progress_ops 队列中

map<ceph_tid_t, InProgressOp> in_progress_ops;

该map保存了所有正在处理的请求。一个请求必须等到所有的up的osd的请求都返回才能返回给客户端。

例如

pg 1.1（osd1,osd2,osd4）

client的请求先发起给osd1，osd1把请求发送给osd2和osd4，等到osd2和osd4上的请求返回给osd1，osd1才能给客户端应答。

那么，当osd2或者osd4出现crash的情况下，该请求如何处理？ 甚至当osd1出现crash的情况该如何处理呢？ 实际上，客户端和OSD端都根据接收到的OSDMap的变化来做相应的处理。

OSD端的处理

当一个osd发送crash时（无论何种原因），通过Heartbeat机制，Monitor会检查到该osd的状态，并把该状态通过OSDMap推送给集群中其它的节点。相应的PG在peering的过程中，会首先丢弃正在处理请求。

例如上述的例子：

当一个osd2或者osd4发生crash时，通过hearbeat等机制，该osd的状态会上报给monitor，当monitor判断该osd处于down状态，会把最新的osdmap推送给各个osd。当osd1收到该osdmap后，pg1.1发现自己的osd 列表发生了变化，就会重新发起peering过程。其会调用函数：PG::start_peering_interval， 该函数调用了ReplicatedBackend::on_change()， 该函数会把正在处理的请求从in_progress_ops中删除。

void PG::start_peering_interval(
{
on_change(t);
}

客户端的处理

当客户端接收到osdmap的变化时，会检查所有正在进行的请求。如果该请求受osdmap的变化的影响，就会在客户端重发请求。

void Objecter::handle_osd_map(MOSDMap *m)
{
...
// 扫描需要重发的请求
_scan_requests(s, false, false, NULL, need_resend,
need_resend_linger, need_resend_command, sul);
...
// resend requests 重发请求
for (map<ceph_tid_t, Op*>::iterator p = need_resend.begin();
p != need_resend.end(); ++p) {
Op *op = p->second;
OSDSession *s = op->session;
bool mapped_session = false;
if (!s) {
int r = _map_session(&op->target, &s, sul);
assert(r == 0);
mapped_session = true;
} else {
get_session(s);
}
OSDSession::unique_lock sl(s->lock);
if (mapped_session) {
_session_op_assign(s, op);
}
if (op->should_resend) {
if (!op->session->is_homeless() && !op->target.paused) {
logger->inc(l_osdc_op_resend);
_send_op(op);
}
} else {
_op_cancel_map_check(op);
_cancel_linger_op(op);
}
sl.unlock();
put_session(s);
...
}
void Objecter::_scan_requests(OSDSession *s,
bool force_resend,
bool cluster_full,
map<int64_t, bool> *pool_full_map,
map<ceph_tid_t, Op*>& need_resend,
list<LingerOp*>& need_resend_linger,
map<ceph_tid_t, CommandOp*>& need_resend_command,
shunique_lock& sul)
{
...

// check for changed request mappings
map<ceph_tid_t,Op*>::iterator p = s->ops.begin();
while (p != s->ops.end()) {
Op *op = p->second;
++p;   // check_op_pool_dne() may touch ops; prevent iterator invalidation
ldout(cct, 10) << " checking op " << op->tid << dendl;
bool force_resend_writes = cluster_full;
if (pool_full_map)
force_resend_writes = force_resend_writes ||
(*pool_full_map)[op->target.base_oloc.pool];
int r = _calc_target(&op->target, &op->last_force_resend);
switch (r) {
case RECALC_OP_TARGET_NO_ACTION:
if (!force_resend &&
(!force_resend_writes || !(op->target.flags & CEPH_OSD_FLAG_WRITE)))
break;
// -- fall-thru --
case RECALC_OP_TARGET_NEED_RESEND:
if (op->session) {
_session_op_remove(op->session, op);
}
need_resend[op->tid] = op;
_op_cancel_map_check(op);
break;
case RECALC_OP_TARGET_POOL_DNE:
_check_op_pool_dne(op, sl);
break;
}
}
...
}