1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2 // vim: ts=8 sw=2 smarttab
4 * Ceph - scalable distributed file system
6 * Copyright (C) 2015 Red Hat
8 * This is free software; you can redistribute it and/or
9 * modify it under the terms of the GNU Lesser General Public
10 * License version 2.1, as published by the Free Software
11 * Foundation. See file COPYING.
15 #include <string_view>
17 #include "common/debug.h"
18 #include "common/errno.h"
20 #include "messages/MClientRequestForward.h"
21 #include "messages/MMDSLoadTargets.h"
22 #include "messages/MMDSTableRequest.h"
24 #include "MDSDaemon.h"
26 #include "SnapClient.h"
27 #include "SnapServer.h"
28 #include "MDBalancer.h"
32 #include "mon/MonClient.h"
33 #include "common/HeartbeatMap.h"
34 #include "ScrubStack.h"
39 #define dout_context g_ceph_context
40 #define dout_subsys ceph_subsys_mds
42 #define dout_prefix *_dout << "mds." << whoami << '.' << incarnation << ' '
44 class C_Flush_Journal
: public MDSInternalContext
{
46 C_Flush_Journal(MDCache
*mdcache
, MDLog
*mdlog
, MDSRank
*mds
,
47 std::ostream
*ss
, Context
*on_finish
)
48 : MDSInternalContext(mds
),
49 mdcache(mdcache
), mdlog(mdlog
), ss(ss
), on_finish(on_finish
),
50 whoami(mds
->whoami
), incarnation(mds
->incarnation
) {
54 assert(mds
->mds_lock
.is_locked());
56 dout(20) << __func__
<< dendl
;
58 if (mdcache
->is_readonly()) {
59 dout(5) << __func__
<< ": read-only FS" << dendl
;
64 if (!mds
->is_active()) {
65 dout(5) << __func__
<< ": MDS not active, no-op" << dendl
;
76 dout(20) << __func__
<< dendl
;
78 // I need to seal off the current segment, and then mark all
79 // previous segments for expiry
80 mdlog
->start_new_segment();
82 Context
*ctx
= new FunctionContext([this](int r
) {
83 handle_flush_mdlog(r
);
86 // Flush initially so that all the segments older than our new one
87 // will be elegible for expiry
89 mdlog
->wait_for_safe(new MDSInternalContextWrapper(mds
, ctx
));
92 void handle_flush_mdlog(int r
) {
93 dout(20) << __func__
<< ": r=" << r
<< dendl
;
96 *ss
<< "Error " << r
<< " (" << cpp_strerror(r
) << ") while flushing journal";
105 dout(20) << __func__
<< dendl
;
107 Context
*ctx
= new FunctionContext([this](int r
) {
108 handle_clear_mdlog(r
);
111 // Because we may not be the last wait_for_safe context on MDLog,
112 // and subsequent contexts might wake up in the middle of our
113 // later trim_all and interfere with expiry (by e.g. marking
114 // dirs/dentries dirty on previous log segments), we run a second
115 // wait_for_safe here. See #10368
116 mdlog
->wait_for_safe(new MDSInternalContextWrapper(mds
, ctx
));
119 void handle_clear_mdlog(int r
) {
120 dout(20) << __func__
<< ": r=" << r
<< dendl
;
123 *ss
<< "Error " << r
<< " (" << cpp_strerror(r
) << ") while flushing journal";
132 // Put all the old log segments into expiring or expired state
133 dout(5) << __func__
<< ": beginning segment expiry" << dendl
;
135 int ret
= mdlog
->trim_all();
137 *ss
<< "Error " << ret
<< " (" << cpp_strerror(ret
) << ") while trimming log";
145 void expire_segments() {
146 dout(20) << __func__
<< dendl
;
148 // Attach contexts to wait for all expiring segments to expire
149 MDSGatherBuilder
*expiry_gather
= new MDSGatherBuilder(g_ceph_context
);
151 const auto &expiring_segments
= mdlog
->get_expiring_segments();
152 for (auto p
: expiring_segments
) {
153 p
->wait_for_expiry(expiry_gather
->new_sub());
155 dout(5) << __func__
<< ": waiting for " << expiry_gather
->num_subs_created()
156 << " segments to expire" << dendl
;
158 if (!expiry_gather
->has_subs()) {
160 delete expiry_gather
;
164 Context
*ctx
= new FunctionContext([this](int r
) {
165 handle_expire_segments(r
);
167 expiry_gather
->set_finisher(new MDSInternalContextWrapper(mds
, ctx
));
168 expiry_gather
->activate();
171 void handle_expire_segments(int r
) {
172 dout(20) << __func__
<< ": r=" << r
<< dendl
;
174 ceph_assert(r
== 0); // MDLog is not allowed to raise errors via
179 void trim_segments() {
180 dout(20) << __func__
<< dendl
;
182 Context
*ctx
= new C_OnFinisher(new FunctionContext([this](int _
) {
183 std::lock_guard
locker(mds
->mds_lock
);
184 trim_expired_segments();
189 void trim_expired_segments() {
190 dout(5) << __func__
<< ": expiry complete, expire_pos/trim_pos is now "
191 << std::hex
<< mdlog
->get_journaler()->get_expire_pos() << "/"
192 << mdlog
->get_journaler()->get_trimmed_pos() << dendl
;
194 // Now everyone I'm interested in is expired
195 mdlog
->trim_expired_segments();
197 dout(5) << __func__
<< ": trim complete, expire_pos/trim_pos is now "
198 << std::hex
<< mdlog
->get_journaler()->get_expire_pos() << "/"
199 << mdlog
->get_journaler()->get_trimmed_pos() << dendl
;
201 write_journal_head();
204 void write_journal_head() {
205 dout(20) << __func__
<< dendl
;
207 Context
*ctx
= new FunctionContext([this](int r
) {
208 std::lock_guard
locker(mds
->mds_lock
);
209 handle_write_head(r
);
211 // Flush the journal header so that readers will start from after
212 // the flushed region
213 mdlog
->get_journaler()->write_head(ctx
);
216 void handle_write_head(int r
) {
218 *ss
<< "Error " << r
<< " (" << cpp_strerror(r
) << ") while writing header";
220 dout(5) << __func__
<< ": write_head complete, all done!" << dendl
;
226 void finish(int r
) override
{
227 dout(20) << __func__
<< ": r=" << r
<< dendl
;
228 on_finish
->complete(r
);
241 class C_Drop_Cache
: public MDSInternalContext
{
243 C_Drop_Cache(Server
*server
, MDCache
*mdcache
, MDLog
*mdlog
,
244 MDSRank
*mds
, uint64_t recall_timeout
,
245 Formatter
*f
, Context
*on_finish
)
246 : MDSInternalContext(mds
),
247 server(server
), mdcache(mdcache
), mdlog(mdlog
),
248 recall_timeout(recall_timeout
), recall_start(mono_clock::now()),
249 f(f
), on_finish(on_finish
),
250 whoami(mds
->whoami
), incarnation(mds
->incarnation
) {
254 // not really a hard requirement here, but lets ensure this in
255 // case we change the logic here.
256 assert(mds
->mds_lock
.is_locked());
258 dout(20) << __func__
<< dendl
;
259 f
->open_object_section("result");
260 recall_client_state();
264 // context which completes itself (with -ETIMEDOUT) after a specified
265 // timeout or when explicitly completed, whichever comes first. Note
266 // that the context does not detroy itself after completion -- it
267 // needs to be explicitly freed.
268 class C_ContextTimeout
: public MDSInternalContext
{
270 C_ContextTimeout(MDSRank
*mds
, uint64_t timeout
, Context
*on_finish
)
271 : MDSInternalContext(mds
),
273 lock("mds::context::timeout", false, true),
274 on_finish(on_finish
) {
276 ~C_ContextTimeout() {
277 ceph_assert(timer_task
== nullptr);
285 timer_task
= new FunctionContext([this](int _
) {
286 timer_task
= nullptr;
287 complete(-ETIMEDOUT
);
289 mds
->timer
.add_event_after(timeout
, timer_task
);
292 void finish(int r
) override
{
293 Context
*ctx
= nullptr;
295 std::lock_guard
locker(lock
);
296 std::swap(on_finish
, ctx
);
298 if (ctx
!= nullptr) {
302 void complete(int r
) override
{
303 if (timer_task
!= nullptr) {
304 mds
->timer
.cancel_event(timer_task
);
312 Context
*on_finish
= nullptr;
313 Context
*timer_task
= nullptr;
317 auto [throttled
, count
] = mdcache
->trim(UINT64_MAX
);
319 << (throttled
? " (throttled)" : "")
320 << " trimmed " << count
<< " caps" << dendl
;
321 dentries_trimmed
+= count
;
322 return std::make_pair(throttled
, count
);
325 void recall_client_state() {
326 dout(20) << __func__
<< dendl
;
327 auto now
= mono_clock::now();
328 auto duration
= std::chrono::duration
<double>(now
-recall_start
).count();
330 MDSGatherBuilder
*gather
= new MDSGatherBuilder(g_ceph_context
);
331 auto [throttled
, count
] = server
->recall_client_state(gather
, Server::RecallFlags::STEADY
);
333 << (throttled
? " (throttled)" : "")
334 << " recalled " << count
<< " caps" << dendl
;
336 caps_recalled
+= count
;
337 if ((throttled
|| count
> 0) && (recall_timeout
== 0 || duration
< recall_timeout
)) {
338 C_ContextTimeout
*ctx
= new C_ContextTimeout(
339 mds
, 1, new FunctionContext([this](int r
) {
340 recall_client_state();
343 gather
->set_finisher(new MDSInternalContextWrapper(mds
, ctx
));
345 mdlog
->flush(); /* use down-time to incrementally flush log */
346 do_trim(); /* use down-time to incrementally trim cache */
348 if (!gather
->has_subs()) {
350 return handle_recall_client_state(0);
351 } else if (recall_timeout
> 0 && duration
> recall_timeout
) {
352 gather
->set_finisher(new C_MDSInternalNoop
);
354 return handle_recall_client_state(-ETIMEDOUT
);
356 uint64_t remaining
= (recall_timeout
== 0 ? 0 : recall_timeout
-duration
);
357 C_ContextTimeout
*ctx
= new C_ContextTimeout(
358 mds
, remaining
, new FunctionContext([this](int r
) {
359 handle_recall_client_state(r
);
363 gather
->set_finisher(new MDSInternalContextWrapper(mds
, ctx
));
369 void handle_recall_client_state(int r
) {
370 dout(20) << __func__
<< ": r=" << r
<< dendl
;
372 // client recall section
373 f
->open_object_section("client_recall");
374 f
->dump_int("return_code", r
);
375 f
->dump_string("message", cpp_strerror(r
));
376 f
->dump_int("recalled", caps_recalled
);
379 // we can still continue after recall timeout
383 void flush_journal() {
384 dout(20) << __func__
<< dendl
;
386 Context
*ctx
= new FunctionContext([this](int r
) {
387 handle_flush_journal(r
);
390 C_Flush_Journal
*flush_journal
= new C_Flush_Journal(mdcache
, mdlog
, mds
, &ss
, ctx
);
391 flush_journal
->send();
394 void handle_flush_journal(int r
) {
395 dout(20) << __func__
<< ": r=" << r
<< dendl
;
398 cmd_err(f
, ss
.str());
403 // journal flush section
404 f
->open_object_section("flush_journal");
405 f
->dump_int("return_code", r
);
406 f
->dump_string("message", ss
.str());
413 dout(20) << __func__
<< dendl
;
415 auto [throttled
, count
] = do_trim();
416 if (throttled
&& count
> 0) {
417 auto timer
= new FunctionContext([this](int _
) {
420 mds
->timer
.add_event_after(1.0, timer
);
426 void cache_status() {
427 dout(20) << __func__
<< dendl
;
429 f
->open_object_section("trim_cache");
430 f
->dump_int("trimmed", dentries_trimmed
);
433 // cache status section
434 mdcache
->cache_status(f
);
439 void finish(int r
) override
{
440 dout(20) << __func__
<< ": r=" << r
<< dendl
;
442 auto d
= std::chrono::duration
<double>(mono_clock::now()-recall_start
);
443 f
->dump_float("duration", d
.count());
446 on_finish
->complete(r
);
452 uint64_t recall_timeout
;
453 mono_time recall_start
;
458 std::stringstream ss
;
459 uint64_t caps_recalled
= 0;
460 uint64_t dentries_trimmed
= 0;
466 void cmd_err(Formatter
*f
, std::string_view err
) {
468 f
->open_object_section("result");
469 f
->dump_string("error", err
);
477 LogChannelRef
&clog_
,
480 std::unique_ptr
<MDSMap
>& mdsmap_
,
483 Context
*respawn_hook_
,
484 Context
*suicide_hook_
)
486 whoami(whoami_
), incarnation(0),
487 mds_lock(mds_lock_
), cct(msgr
->cct
), clog(clog_
), timer(timer_
),
489 objecter(new Objecter(g_ceph_context
, msgr
, monc_
, nullptr, 0, 0)),
490 server(NULL
), mdcache(NULL
), locker(NULL
), mdlog(NULL
),
491 balancer(NULL
), scrubstack(NULL
),
492 damage_table(whoami_
),
493 inotable(NULL
), snapserver(NULL
), snapclient(NULL
),
494 sessionmap(this), logger(NULL
), mlogger(NULL
),
495 op_tracker(g_ceph_context
, g_conf()->mds_enable_op_tracker
,
496 g_conf()->osd_num_op_tracker_shard
),
497 last_state(MDSMap::STATE_BOOT
),
498 state(MDSMap::STATE_BOOT
),
499 cluster_degraded(false), stopping(false),
500 purge_queue(g_ceph_context
, whoami_
,
501 mdsmap_
->get_metadata_pool(), objecter
,
504 // Purge Queue operates inside mds_lock when we're calling into
505 // it, and outside when in background, so must handle both cases.
506 if (mds_lock
.is_locked_by_me()) {
507 handle_write_error(r
);
509 std::lock_guard
l(mds_lock
);
510 handle_write_error(r
);
515 progress_thread(this), dispatch_depth(0),
516 hb(NULL
), last_tid(0), osd_epoch_barrier(0), beacon(beacon_
),
517 mds_slow_req_count(0),
518 last_client_mdsmap_bcast(0),
519 messenger(msgr
), monc(monc_
),
520 respawn_hook(respawn_hook_
),
521 suicide_hook(suicide_hook_
),
522 standby_replaying(false),
523 starttime(mono_clock::now())
525 hb
= g_ceph_context
->get_heartbeat_map()->add_worker("MDSRank", pthread_self());
527 purge_queue
.update_op_limit(*mdsmap
);
529 objecter
->unset_honor_osdmap_full();
531 finisher
= new Finisher(cct
);
533 mdcache
= new MDCache(this, purge_queue
);
534 mdlog
= new MDLog(this);
535 balancer
= new MDBalancer(this, messenger
, monc
);
537 scrubstack
= new ScrubStack(mdcache
, finisher
);
539 inotable
= new InoTable(this);
540 snapserver
= new SnapServer(this, monc
);
541 snapclient
= new SnapClient(this);
543 server
= new Server(this);
544 locker
= new Locker(this, mdcache
);
546 op_tracker
.set_complaint_and_threshold(cct
->_conf
->mds_op_complaint_time
,
547 cct
->_conf
->mds_op_log_threshold
);
548 op_tracker
.set_history_size_and_duration(cct
->_conf
->mds_op_history_size
,
549 cct
->_conf
->mds_op_history_duration
);
555 g_ceph_context
->get_heartbeat_map()->remove_worker(hb
);
558 if (scrubstack
) { delete scrubstack
; scrubstack
= NULL
; }
559 if (mdcache
) { delete mdcache
; mdcache
= NULL
; }
560 if (mdlog
) { delete mdlog
; mdlog
= NULL
; }
561 if (balancer
) { delete balancer
; balancer
= NULL
; }
562 if (inotable
) { delete inotable
; inotable
= NULL
; }
563 if (snapserver
) { delete snapserver
; snapserver
= NULL
; }
564 if (snapclient
) { delete snapclient
; snapclient
= NULL
; }
566 if (server
) { delete server
; server
= 0; }
567 if (locker
) { delete locker
; locker
= 0; }
570 g_ceph_context
->get_perfcounters_collection()->remove(logger
);
575 g_ceph_context
->get_perfcounters_collection()->remove(mlogger
);
593 void MDSRankDispatcher::init()
596 messenger
->add_dispatcher_head(objecter
);
603 // Expose the OSDMap (already populated during MDS::init) to anyone
604 // who is interested in it.
607 progress_thread
.create("mds_rank_progr");
614 void MDSRank::update_targets()
616 // get MonMap's idea of my export_targets
617 const set
<mds_rank_t
>& map_targets
= mdsmap
->get_mds_info(get_nodeid()).export_targets
;
619 dout(20) << "updating export targets, currently " << map_targets
.size() << " ranks are targets" << dendl
;
622 set
<mds_rank_t
> new_map_targets
;
624 auto it
= export_targets
.begin();
625 while (it
!= export_targets
.end()) {
626 mds_rank_t rank
= it
->first
;
627 auto &counter
= it
->second
;
628 dout(20) << "export target mds." << rank
<< " is " << counter
<< dendl
;
630 double val
= counter
.get();
632 dout(15) << "export target mds." << rank
<< " is no longer an export target" << dendl
;
633 export_targets
.erase(it
++);
637 if (!map_targets
.count(rank
)) {
638 dout(15) << "export target mds." << rank
<< " not in map's export_targets" << dendl
;
641 new_map_targets
.insert(rank
);
644 if (new_map_targets
.size() < map_targets
.size()) {
645 dout(15) << "export target map holds stale targets, sending update" << dendl
;
650 dout(15) << "updating export_targets, now " << new_map_targets
.size() << " ranks are targets" << dendl
;
651 auto m
= MMDSLoadTargets::create(mds_gid_t(monc
->get_global_id()), new_map_targets
);
652 monc
->send_mon_message(m
.detach());
656 void MDSRank::hit_export_target(mds_rank_t rank
, double amount
)
658 double rate
= g_conf()->mds_bal_target_decay
;
660 amount
= 100.0/g_conf()->mds_bal_target_decay
; /* a good default for "i am trying to keep this export_target active" */
662 auto em
= export_targets
.emplace(std::piecewise_construct
, std::forward_as_tuple(rank
), std::forward_as_tuple(DecayRate(rate
)));
663 auto &counter
= em
.first
->second
;
666 dout(15) << "hit export target (new) is " << counter
<< dendl
;
668 dout(15) << "hit export target is " << counter
<< dendl
;
672 class C_MDS_MonCommand
: public MDSInternalContext
{
676 C_MDS_MonCommand(MDSRank
*m
, std::string_view c
)
677 : MDSInternalContext(m
), cmd(c
) {}
678 void finish(int r
) override
{
679 mds
->_mon_command_finish(r
, cmd
, outs
);
683 void MDSRank::_mon_command_finish(int r
, std::string_view cmd
, std::string_view outs
)
686 dout(0) << __func__
<< ": mon command " << cmd
<< " failed with errno " << r
687 << " (" << outs
<< ")" << dendl
;
689 dout(1) << __func__
<< ": mon command " << cmd
<< " succeed" << dendl
;
693 void MDSRank::set_mdsmap_multimds_snaps_allowed()
695 static bool already_sent
= false;
700 ss
<< "{\"prefix\":\"fs set\", \"fs_name\":\"" << mdsmap
->get_fs_name() << "\", ";
701 ss
<< "\"var\":\"allow_multimds_snaps\", \"val\":\"true\", ";
702 ss
<< "\"confirm\":\"--yes-i-am-really-a-mds\"}";
703 std::vector
<std::string
> cmd
= {ss
.str()};
705 dout(0) << __func__
<< ": sending mon command: " << cmd
[0] << dendl
;
707 C_MDS_MonCommand
*fin
= new C_MDS_MonCommand(this, cmd
[0]);
708 monc
->start_mon_command(cmd
, {}, nullptr, &fin
->outs
, new C_IO_Wrapper(this, fin
));
713 void MDSRank::mark_base_recursively_scrubbed(inodeno_t ino
)
715 if (mdsmap
->get_tableserver() == whoami
)
716 snapserver
->mark_base_recursively_scrubbed(ino
);
719 void MDSRankDispatcher::tick()
723 if (beacon
.is_laggy()) {
724 dout(1) << "skipping upkeep work because connection to Monitors appears laggy" << dendl
;
728 check_ops_in_flight();
730 // Wake up thread in case we use to be laggy and have waiting_for_nolaggy
731 // messages to progress.
732 progress_thread
.signal();
734 // make sure mds log flushes, trims periodically
737 // update average session uptime
738 sessionmap
.update_average_session_age();
740 if (is_active() || is_stopping()) {
741 mdlog
->trim(); // NOT during recovery!
745 if (is_cache_trimmable()) {
746 server
->find_idle_sessions();
747 server
->evict_cap_revoke_non_responders();
753 logger
->set(l_mds_subtrees
, mdcache
->num_subtrees());
758 server
->reconnect_tick();
762 mdcache
->find_stale_fragment_freeze();
763 mdcache
->migrator
->find_stale_export_freeze();
765 if (mdsmap
->get_tableserver() == whoami
) {
766 snapserver
->check_osd_map(false);
767 // Filesystem was created by pre-mimic mds. Allow multi-active mds after
768 // all old snapshots are deleted.
769 if (!mdsmap
->allows_multimds_snaps() &&
770 snapserver
->can_allow_multimds_snaps()) {
771 set_mdsmap_multimds_snaps_allowed();
776 if (is_active() || is_stopping()) {
783 if (mdcache
->shutdown_pass()) {
784 uint64_t pq_progress
= 0 ;
785 uint64_t pq_total
= 0;
786 size_t pq_in_flight
= 0;
787 if (!purge_queue
.drain(&pq_progress
, &pq_total
, &pq_in_flight
)) {
788 dout(7) << "shutdown_pass=true, but still waiting for purge queue"
790 // This takes unbounded time, so we must indicate progress
791 // to the administrator: we do it in a slightly imperfect way
792 // by sending periodic (tick frequency) clog messages while
794 clog
->info() << "MDS rank " << whoami
<< " waiting for purge queue ("
795 << std::dec
<< pq_progress
<< "/" << pq_total
<< " " << pq_in_flight
796 << " files purging" << ")";
798 dout(7) << "shutdown_pass=true, finished w/ shutdown, moving to "
799 "down:stopped" << dendl
;
804 dout(7) << "shutdown_pass=false" << dendl
;
808 // Expose ourselves to Beacon to update health indicators
809 beacon
.notify_health(this);
812 void MDSRankDispatcher::shutdown()
814 // It should never be possible for shutdown to get called twice, because
815 // anyone picking up mds_lock checks if stopping is true and drops
817 ceph_assert(stopping
== false);
820 dout(1) << __func__
<< ": shutting down rank " << whoami
<< dendl
;
824 // MDLog has to shut down before the finisher, because some of its
825 // threads block on IOs that require finisher to complete.
831 purge_queue
.shutdown();
834 finisher
->stop(); // no flushing
837 if (objecter
->initialized
)
838 objecter
->shutdown();
842 op_tracker
.on_shutdown();
844 progress_thread
.shutdown();
846 // release mds_lock for finisher/messenger threads (e.g.
847 // MDSDaemon::ms_handle_reset called from Messenger).
850 // shut down messenger
851 messenger
->shutdown();
855 // Workaround unclean shutdown: HeartbeatMap will assert if
856 // worker is not removed (as we do in ~MDS), but ~MDS is not
857 // always called after suicide.
859 g_ceph_context
->get_heartbeat_map()->remove_worker(hb
);
865 * Helper for simple callbacks that call a void fn with no args.
867 class C_MDS_VoidFn
: public MDSInternalContext
869 typedef void (MDSRank::*fn_ptr
)();
873 C_MDS_VoidFn(MDSRank
*mds_
, fn_ptr fn_
)
874 : MDSInternalContext(mds_
), fn(fn_
)
880 void finish(int r
) override
886 int64_t MDSRank::get_metadata_pool()
888 return mdsmap
->get_metadata_pool();
891 MDSTableClient
*MDSRank::get_table_client(int t
)
894 case TABLE_ANCHOR
: return NULL
;
895 case TABLE_SNAP
: return snapclient
;
896 default: ceph_abort();
900 MDSTableServer
*MDSRank::get_table_server(int t
)
903 case TABLE_ANCHOR
: return NULL
;
904 case TABLE_SNAP
: return snapserver
;
905 default: ceph_abort();
909 void MDSRank::suicide()
912 suicide_hook
->complete(0);
917 void MDSRank::respawn()
920 respawn_hook
->complete(0);
925 void MDSRank::damaged()
927 ceph_assert(whoami
!= MDS_RANK_NONE
);
928 ceph_assert(mds_lock
.is_locked_by_me());
930 beacon
.set_want_state(*mdsmap
, MDSMap::STATE_DAMAGED
);
931 monc
->flush_log(); // Flush any clog error from before we were called
932 beacon
.notify_health(this); // Include latest status in our swan song
933 beacon
.send_and_wait(g_conf()->mds_mon_shutdown_timeout
);
935 // It's okay if we timed out and the mon didn't get our beacon, because
936 // another daemon (or ourselves after respawn) will eventually take the
937 // rank and report DAMAGED again when it hits same problem we did.
939 respawn(); // Respawn into standby in case mon has other work for us
942 void MDSRank::damaged_unlocked()
944 std::lock_guard
l(mds_lock
);
948 void MDSRank::handle_write_error(int err
)
950 if (err
== -EBLACKLISTED
) {
951 derr
<< "we have been blacklisted (fenced), respawning..." << dendl
;
956 if (g_conf()->mds_action_on_write_error
>= 2) {
957 derr
<< "unhandled write error " << cpp_strerror(err
) << ", suicide..." << dendl
;
959 } else if (g_conf()->mds_action_on_write_error
== 1) {
960 derr
<< "unhandled write error " << cpp_strerror(err
) << ", force readonly..." << dendl
;
961 mdcache
->force_readonly();
964 derr
<< "unhandled write error " << cpp_strerror(err
) << ", ignore..." << dendl
;
968 void *MDSRank::ProgressThread::entry()
970 std::lock_guard
l(mds
->mds_lock
);
972 while (!mds
->stopping
&&
973 mds
->finished_queue
.empty() &&
974 (mds
->waiting_for_nolaggy
.empty() || mds
->beacon
.is_laggy())) {
975 cond
.Wait(mds
->mds_lock
);
982 mds
->_advance_queues();
989 void MDSRank::ProgressThread::shutdown()
991 ceph_assert(mds
->mds_lock
.is_locked_by_me());
992 ceph_assert(mds
->stopping
);
995 // Stopping is set, we will fall out of our main loop naturally
997 // Kick the thread to notice mds->stopping, and join it
999 mds
->mds_lock
.Unlock();
1002 mds
->mds_lock
.Lock();
1006 bool MDSRankDispatcher::ms_dispatch(const Message::const_ref
&m
)
1008 if (m
->get_source().is_client()) {
1009 Session
*session
= static_cast<Session
*>(m
->get_connection()->get_priv().get());
1011 session
->last_seen
= Session::clock::now();
1014 inc_dispatch_depth();
1015 bool ret
= _dispatch(m
, true);
1016 dec_dispatch_depth();
1020 bool MDSRank::_dispatch(const Message::const_ref
&m
, bool new_msg
)
1022 if (is_stale_message(m
)) {
1026 if (beacon
.is_laggy()) {
1027 dout(5) << " laggy, deferring " << *m
<< dendl
;
1028 waiting_for_nolaggy
.push_back(m
);
1029 } else if (new_msg
&& !waiting_for_nolaggy
.empty()) {
1030 dout(5) << " there are deferred messages, deferring " << *m
<< dendl
;
1031 waiting_for_nolaggy
.push_back(m
);
1033 if (!handle_deferrable_message(m
)) {
1034 dout(0) << "unrecognized message " << *m
<< dendl
;
1041 if (dispatch_depth
> 1)
1044 // finish any triggered contexts
1047 if (beacon
.is_laggy()) {
1048 // We've gone laggy during dispatch, don't do any
1049 // more housekeeping
1053 // hack: thrash exports
1054 static utime_t start
;
1055 utime_t now
= ceph_clock_now();
1056 if (start
== utime_t())
1058 /*double el = now - start;
1061 for (int i
=0; i
<g_conf()->mds_thrash_exports
; i
++) {
1063 if (!is_active()) break;
1064 mdsmap
->get_mds_set(s
, MDSMap::STATE_ACTIVE
);
1065 if (s
.size() < 2 || CInode::count() < 10)
1066 break; // need peers for this to work.
1067 if (mdcache
->migrator
->get_num_exporting() > g_conf()->mds_thrash_exports
* 5 ||
1068 mdcache
->migrator
->get_export_queue_size() > g_conf()->mds_thrash_exports
* 10)
1071 dout(7) << "mds thrashing exports pass " << (i
+1) << "/" << g_conf()->mds_thrash_exports
<< dendl
;
1073 // pick a random dir inode
1074 CInode
*in
= mdcache
->hack_pick_random_inode();
1077 in
->get_dirfrags(ls
);
1078 if (!ls
.empty()) { // must be an open dir.
1079 list
<CDir
*>::iterator p
= ls
.begin();
1080 int n
= rand() % ls
.size();
1084 if (!dir
->get_parent_dir()) continue; // must be linked.
1085 if (!dir
->is_auth()) continue; // must be auth.
1089 int k
= rand() % s
.size();
1090 set
<mds_rank_t
>::iterator p
= s
.begin();
1093 } while (dest
== whoami
);
1094 mdcache
->migrator
->export_dir_nicely(dir
,dest
);
1097 // hack: thrash fragments
1098 for (int i
=0; i
<g_conf()->mds_thrash_fragments
; i
++) {
1099 if (!is_active()) break;
1100 if (mdcache
->get_num_fragmenting_dirs() > 5 * g_conf()->mds_thrash_fragments
) break;
1101 dout(7) << "mds thrashing fragments pass " << (i
+1) << "/" << g_conf()->mds_thrash_fragments
<< dendl
;
1103 // pick a random dir inode
1104 CInode
*in
= mdcache
->hack_pick_random_inode();
1107 in
->get_dirfrags(ls
);
1108 if (ls
.empty()) continue; // must be an open dir.
1109 CDir
*dir
= ls
.front();
1110 if (!dir
->get_parent_dir()) continue; // must be linked.
1111 if (!dir
->is_auth()) continue; // must be auth.
1112 frag_t fg
= dir
->get_frag();
1113 if ((fg
== frag_t() || (rand() % (1 << fg
.bits()) == 0))) {
1114 mdcache
->split_dir(dir
, 1);
1116 balancer
->queue_merge(dir
);
1120 // hack: force hash root?
1123 mdcache->get_root() &&
1124 mdcache->get_root()->dir &&
1125 !(mdcache->get_root()->dir->is_hashed() ||
1126 mdcache->get_root()->dir->is_hashing())) {
1127 dout(0) << "hashing root" << dendl;
1128 mdcache->migrator->hash_dir(mdcache->get_root()->dir);
1136 void MDSRank::update_mlogger()
1139 mlogger
->set(l_mdm_ino
, CInode::count());
1140 mlogger
->set(l_mdm_dir
, CDir::count());
1141 mlogger
->set(l_mdm_dn
, CDentry::count());
1142 mlogger
->set(l_mdm_cap
, Capability::count());
1143 mlogger
->set(l_mdm_inoa
, CInode::increments());
1144 mlogger
->set(l_mdm_inos
, CInode::decrements());
1145 mlogger
->set(l_mdm_dira
, CDir::increments());
1146 mlogger
->set(l_mdm_dirs
, CDir::decrements());
1147 mlogger
->set(l_mdm_dna
, CDentry::increments());
1148 mlogger
->set(l_mdm_dns
, CDentry::decrements());
1149 mlogger
->set(l_mdm_capa
, Capability::increments());
1150 mlogger
->set(l_mdm_caps
, Capability::decrements());
1155 * lower priority messages we defer if we seem laggy
1157 bool MDSRank::handle_deferrable_message(const Message::const_ref
&m
)
1159 int port
= m
->get_type() & 0xff00;
1162 case MDS_PORT_CACHE
:
1163 ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_MDS
);
1164 mdcache
->dispatch(m
);
1167 case MDS_PORT_MIGRATOR
:
1168 ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_MDS
);
1169 mdcache
->migrator
->dispatch(m
);
1173 switch (m
->get_type()) {
1175 case CEPH_MSG_CLIENT_SESSION
:
1176 case CEPH_MSG_CLIENT_RECONNECT
:
1177 case CEPH_MSG_CLIENT_RECLAIM
:
1178 ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_CLIENT
);
1180 case CEPH_MSG_CLIENT_REQUEST
:
1181 server
->dispatch(m
);
1183 case MSG_MDS_SLAVE_REQUEST
:
1184 ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_MDS
);
1185 server
->dispatch(m
);
1188 case MSG_MDS_HEARTBEAT
:
1189 ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_MDS
);
1190 balancer
->proc_message(m
);
1193 case MSG_MDS_TABLE_REQUEST
:
1194 ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_MDS
);
1196 const MMDSTableRequest::const_ref
&req
= MMDSTableRequest::msgref_cast(m
);
1198 MDSTableClient
*client
= get_table_client(req
->table
);
1199 client
->handle_request(req
);
1201 MDSTableServer
*server
= get_table_server(req
->table
);
1202 server
->handle_request(req
);
1208 case MSG_MDS_INODEFILECAPS
:
1209 ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_MDS
);
1210 locker
->dispatch(m
);
1213 case CEPH_MSG_CLIENT_CAPS
:
1214 case CEPH_MSG_CLIENT_CAPRELEASE
:
1215 case CEPH_MSG_CLIENT_LEASE
:
1216 ALLOW_MESSAGES_FROM(CEPH_ENTITY_TYPE_CLIENT
);
1217 locker
->dispatch(m
);
1229 * Advance finished_queue and waiting_for_nolaggy.
1231 * Usually drain both queues, but may not drain waiting_for_nolaggy
1232 * if beacon is currently laggy.
1234 void MDSRank::_advance_queues()
1236 ceph_assert(mds_lock
.is_locked_by_me());
1238 if (!finished_queue
.empty()) {
1239 dout(7) << "mds has " << finished_queue
.size() << " queued contexts" << dendl
;
1240 while (!finished_queue
.empty()) {
1241 auto fin
= finished_queue
.front();
1242 finished_queue
.pop_front();
1244 dout(10) << " finish " << fin
<< dendl
;
1251 while (!waiting_for_nolaggy
.empty()) {
1252 // stop if we're laggy now!
1253 if (beacon
.is_laggy())
1256 Message::const_ref old
= waiting_for_nolaggy
.front();
1257 waiting_for_nolaggy
.pop_front();
1259 if (!is_stale_message(old
)) {
1260 dout(7) << " processing laggy deferred " << *old
<< dendl
;
1261 if (!handle_deferrable_message(old
)) {
1262 dout(0) << "unrecognized message " << *old
<< dendl
;
1271 * Call this when you take mds_lock, or periodically if you're going to
1272 * hold the lock for a long time (e.g. iterating over clients/inodes)
1274 void MDSRank::heartbeat_reset()
1276 // Any thread might jump into mds_lock and call us immediately
1277 // after a call to suicide() completes, in which case MDSRank::hb
1278 // has been freed and we are a no-op.
1280 ceph_assert(stopping
);
1284 // NB not enabling suicide grace, because the mon takes care of killing us
1285 // (by blacklisting us) when we fail to send beacons, and it's simpler to
1286 // only have one way of dying.
1287 auto grace
= g_conf().get_val
<double>("mds_heartbeat_grace");
1288 g_ceph_context
->get_heartbeat_map()->reset_timeout(hb
, grace
, 0);
1291 bool MDSRank::is_stale_message(const Message::const_ref
&m
) const
1294 if (m
->get_source().is_mds()) {
1295 mds_rank_t from
= mds_rank_t(m
->get_source().num());
1297 if (mdsmap
->is_down(from
)) {
1300 // FIXME: this is a convoluted check. we should be maintaining a nice
1301 // clean map of current ConnectionRefs for current mdses!!!
1302 auto c
= messenger
->connect_to(CEPH_ENTITY_TYPE_MDS
,
1303 mdsmap
->get_addrs(from
));
1304 if (c
!= m
->get_connection()) {
1306 dout(5) << " mds." << from
<< " should be " << c
<< " "
1307 << c
->get_peer_addrs() << " but this message is "
1308 << m
->get_connection() << " " << m
->get_source_addrs()
1314 if (m
->get_type() == CEPH_MSG_MDS_MAP
) {
1315 dout(5) << "got " << *m
<< " from old/bad/imposter mds " << m
->get_source()
1316 << ", but it's an mdsmap, looking at it" << dendl
;
1317 } else if (m
->get_type() == MSG_MDS_CACHEEXPIRE
&&
1318 mdsmap
->get_addrs(from
) == m
->get_source_addrs()) {
1319 dout(5) << "got " << *m
<< " from down mds " << m
->get_source()
1320 << ", but it's a cache_expire, looking at it" << dendl
;
1322 dout(5) << "got " << *m
<< " from down/old/bad/imposter mds " << m
->get_source()
1323 << ", dropping" << dendl
;
1331 Session
*MDSRank::get_session(const Message::const_ref
&m
)
1334 auto session
= static_cast<Session
*>(m
->get_connection()->get_priv().get());
1336 dout(20) << "get_session have " << session
<< " " << session
->info
.inst
1337 << " state " << session
->get_state_name() << dendl
;
1338 // Check if we've imported an open session since (new sessions start closed)
1339 if (session
->is_closed()) {
1340 Session
*imported_session
= sessionmap
.get_session(session
->info
.inst
.name
);
1341 if (imported_session
&& imported_session
!= session
) {
1342 dout(10) << __func__
<< " replacing connection bootstrap session "
1343 << session
<< " with imported session " << imported_session
1345 imported_session
->info
.auth_name
= session
->info
.auth_name
;
1346 //assert(session->info.auth_name == imported_session->info.auth_name);
1347 ceph_assert(session
->info
.inst
== imported_session
->info
.inst
);
1348 imported_session
->set_connection(session
->get_connection().get());
1349 // send out any queued messages
1350 while (!session
->preopen_out_queue
.empty()) {
1351 imported_session
->get_connection()->send_message2(std::move(session
->preopen_out_queue
.front()));
1352 session
->preopen_out_queue
.pop_front();
1354 imported_session
->auth_caps
= session
->auth_caps
;
1355 imported_session
->last_seen
= session
->last_seen
;
1356 ceph_assert(session
->get_nref() == 1);
1357 imported_session
->get_connection()->set_priv(imported_session
->get());
1358 session
= imported_session
;
1362 dout(20) << "get_session dne for " << m
->get_source_inst() << dendl
;
1367 void MDSRank::send_message(const Message::ref
& m
, const ConnectionRef
& c
)
1370 c
->send_message2(m
);
1374 void MDSRank::send_message_mds(const Message::ref
& m
, mds_rank_t mds
)
1376 if (!mdsmap
->is_up(mds
)) {
1377 dout(10) << "send_message_mds mds." << mds
<< " not up, dropping " << *m
<< dendl
;
1381 // send mdsmap first?
1382 if (mds
!= whoami
&& peer_mdsmap_epoch
[mds
] < mdsmap
->get_epoch()) {
1383 auto _m
= MMDSMap::create(monc
->get_fsid(), *mdsmap
);
1384 messenger
->send_to_mds(_m
.detach(), mdsmap
->get_addrs(mds
));
1385 peer_mdsmap_epoch
[mds
] = mdsmap
->get_epoch();
1389 messenger
->send_to_mds(Message::ref(m
).detach(), mdsmap
->get_addrs(mds
));
1392 void MDSRank::forward_message_mds(const MClientRequest::const_ref
& m
, mds_rank_t mds
)
1394 ceph_assert(mds
!= whoami
);
1397 * don't actually forward if non-idempotent!
1398 * client has to do it. although the MDS will ignore duplicate requests,
1399 * the affected metadata may migrate, in which case the new authority
1400 * won't have the metareq_id in the completed request map.
1402 // NEW: always make the client resend!
1403 bool client_must_resend
= true; //!creq->can_forward();
1405 // tell the client where it should go
1406 auto session
= get_session(m
);
1407 auto f
= MClientRequestForward::create(m
->get_tid(), mds
, m
->get_num_fwd()+1, client_must_resend
);
1408 send_message_client(f
, session
);
1411 void MDSRank::send_message_client_counted(const Message::ref
& m
, client_t client
)
1413 Session
*session
= sessionmap
.get_session(entity_name_t::CLIENT(client
.v
));
1415 send_message_client_counted(m
, session
);
1417 dout(10) << "send_message_client_counted no session for client." << client
<< " " << *m
<< dendl
;
1421 void MDSRank::send_message_client_counted(const Message::ref
& m
, const ConnectionRef
& connection
)
1424 auto session
= static_cast<Session
*>(connection
->get_priv().get());
1426 send_message_client_counted(m
, session
);
1428 dout(10) << "send_message_client_counted has no session for " << m
->get_source_inst() << dendl
;
1429 // another Connection took over the Session
1433 void MDSRank::send_message_client_counted(const Message::ref
& m
, Session
* session
)
1435 version_t seq
= session
->inc_push_seq();
1436 dout(10) << "send_message_client_counted " << session
->info
.inst
.name
<< " seq "
1437 << seq
<< " " << *m
<< dendl
;
1438 if (session
->get_connection()) {
1439 session
->get_connection()->send_message2(m
);
1441 session
->preopen_out_queue
.push_back(m
);
1445 void MDSRank::send_message_client(const Message::ref
& m
, Session
* session
)
1447 dout(10) << "send_message_client " << session
->info
.inst
<< " " << *m
<< dendl
;
1448 if (session
->get_connection()) {
1449 session
->get_connection()->send_message2(m
);
1451 session
->preopen_out_queue
.push_back(m
);
1456 * This is used whenever a RADOS operation has been cancelled
1457 * or a RADOS client has been blacklisted, to cause the MDS and
1458 * any clients to wait for this OSD epoch before using any new caps.
1460 * See doc/cephfs/eviction
1462 void MDSRank::set_osd_epoch_barrier(epoch_t e
)
1464 dout(4) << __func__
<< ": epoch=" << e
<< dendl
;
1465 osd_epoch_barrier
= e
;
1468 void MDSRank::retry_dispatch(const Message::const_ref
&m
)
1470 inc_dispatch_depth();
1471 _dispatch(m
, false);
1472 dec_dispatch_depth();
1475 double MDSRank::get_dispatch_queue_max_age(utime_t now
) const
1477 return messenger
->get_dispatch_queue_max_age(now
);
1480 bool MDSRank::is_daemon_stopping() const
1485 void MDSRank::request_state(MDSMap::DaemonState s
)
1487 dout(3) << "request_state " << ceph_mds_state_name(s
) << dendl
;
1488 beacon
.set_want_state(*mdsmap
, s
);
1493 class C_MDS_BootStart
: public MDSInternalContext
{
1494 MDSRank::BootStep nextstep
;
1496 C_MDS_BootStart(MDSRank
*m
, MDSRank::BootStep n
)
1497 : MDSInternalContext(m
), nextstep(n
) {}
1498 void finish(int r
) override
{
1499 mds
->boot_start(nextstep
, r
);
1504 void MDSRank::boot_start(BootStep step
, int r
)
1506 // Handle errors from previous step
1508 if (is_standby_replay() && (r
== -EAGAIN
)) {
1509 dout(0) << "boot_start encountered an error EAGAIN"
1510 << ", respawning since we fell behind journal" << dendl
;
1512 } else if (r
== -EINVAL
|| r
== -ENOENT
) {
1513 // Invalid or absent data, indicates damaged on-disk structures
1514 clog
->error() << "Error loading MDS rank " << whoami
<< ": "
1517 ceph_assert(r
== 0); // Unreachable, damaged() calls respawn()
1518 } else if (r
== -EROFS
) {
1519 dout(0) << "boot error forcing transition to read-only; MDS will try to continue" << dendl
;
1521 // Completely unexpected error, give up and die
1522 dout(0) << "boot_start encountered an error, failing" << dendl
;
1528 ceph_assert(is_starting() || is_any_replay());
1531 case MDS_BOOT_INITIAL
:
1533 mdcache
->init_layouts();
1535 MDSGatherBuilder
gather(g_ceph_context
,
1536 new C_MDS_BootStart(this, MDS_BOOT_OPEN_ROOT
));
1537 dout(2) << "Booting: " << step
<< ": opening inotable" << dendl
;
1538 inotable
->set_rank(whoami
);
1539 inotable
->load(gather
.new_sub());
1541 dout(2) << "Booting: " << step
<< ": opening sessionmap" << dendl
;
1542 sessionmap
.set_rank(whoami
);
1543 sessionmap
.load(gather
.new_sub());
1545 dout(2) << "Booting: " << step
<< ": opening mds log" << dendl
;
1546 mdlog
->open(gather
.new_sub());
1548 if (is_starting()) {
1549 dout(2) << "Booting: " << step
<< ": opening purge queue" << dendl
;
1550 purge_queue
.open(new C_IO_Wrapper(this, gather
.new_sub()));
1551 } else if (!standby_replaying
) {
1552 dout(2) << "Booting: " << step
<< ": opening purge queue (async)" << dendl
;
1553 purge_queue
.open(NULL
);
1554 dout(2) << "Booting: " << step
<< ": loading open file table (async)" << dendl
;
1555 mdcache
->open_file_table
.load(nullptr);
1558 if (mdsmap
->get_tableserver() == whoami
) {
1559 dout(2) << "Booting: " << step
<< ": opening snap table" << dendl
;
1560 snapserver
->set_rank(whoami
);
1561 snapserver
->load(gather
.new_sub());
1567 case MDS_BOOT_OPEN_ROOT
:
1569 dout(2) << "Booting: " << step
<< ": loading/discovering base inodes" << dendl
;
1571 MDSGatherBuilder
gather(g_ceph_context
,
1572 new C_MDS_BootStart(this, MDS_BOOT_PREPARE_LOG
));
1574 if (is_starting()) {
1575 // load mydir frag for the first log segment (creating subtree map)
1576 mdcache
->open_mydir_frag(gather
.new_sub());
1578 mdcache
->open_mydir_inode(gather
.new_sub());
1581 mdcache
->create_global_snaprealm();
1583 if (whoami
== mdsmap
->get_root()) { // load root inode off disk if we are auth
1584 mdcache
->open_root_inode(gather
.new_sub());
1585 } else if (is_any_replay()) {
1586 // replay. make up fake root inode to start with
1587 mdcache
->create_root_inode();
1592 case MDS_BOOT_PREPARE_LOG
:
1593 if (is_any_replay()) {
1594 dout(2) << "Booting: " << step
<< ": replaying mds log" << dendl
;
1595 MDSGatherBuilder
gather(g_ceph_context
,
1596 new C_MDS_BootStart(this, MDS_BOOT_REPLAY_DONE
));
1598 if (!standby_replaying
) {
1599 dout(2) << "Booting: " << step
<< ": waiting for purge queue recovered" << dendl
;
1600 purge_queue
.wait_for_recovery(new C_IO_Wrapper(this, gather
.new_sub()));
1603 mdlog
->replay(gather
.new_sub());
1606 dout(2) << "Booting: " << step
<< ": positioning at end of old mds log" << dendl
;
1611 case MDS_BOOT_REPLAY_DONE
:
1612 ceph_assert(is_any_replay());
1614 // Sessiontable and inotable should be in sync after replay, validate
1615 // that they are consistent.
1616 validate_sessions();
1623 void MDSRank::validate_sessions()
1625 ceph_assert(mds_lock
.is_locked_by_me());
1628 // Identify any sessions which have state inconsistent with other,
1629 // after they have been loaded from rados during startup.
1630 // Mitigate bugs like: http://tracker.ceph.com/issues/16842
1631 for (const auto &i
: sessionmap
.get_sessions()) {
1632 Session
*session
= i
.second
;
1633 interval_set
<inodeno_t
> badones
;
1634 if (inotable
->intersects_free(session
->info
.prealloc_inos
, &badones
)) {
1635 clog
->error() << "client " << *session
1636 << "loaded with preallocated inodes that are inconsistent with inotable";
1647 void MDSRank::starting_done()
1649 dout(3) << "starting_done" << dendl
;
1650 ceph_assert(is_starting());
1651 request_state(MDSMap::STATE_ACTIVE
);
1653 mdlog
->start_new_segment();
1655 // sync snaptable cache
1656 snapclient
->sync(new C_MDSInternalNoop
);
1660 void MDSRank::calc_recovery_set()
1662 // initialize gather sets
1664 mdsmap
->get_recovery_mds_set(rs
);
1666 mdcache
->set_recovery_set(rs
);
1668 dout(1) << " recovery set is " << rs
<< dendl
;
1672 void MDSRank::replay_start()
1674 dout(1) << "replay_start" << dendl
;
1676 if (is_standby_replay())
1677 standby_replaying
= true;
1679 calc_recovery_set();
1681 // Check if we need to wait for a newer OSD map before starting
1682 Context
*fin
= new C_IO_Wrapper(this, new C_MDS_BootStart(this, MDS_BOOT_INITIAL
));
1683 bool const ready
= objecter
->wait_for_map(
1684 mdsmap
->get_last_failure_osd_epoch(),
1691 dout(1) << " waiting for osdmap " << mdsmap
->get_last_failure_osd_epoch()
1692 << " (which blacklists prior instance)" << dendl
;
1697 class MDSRank::C_MDS_StandbyReplayRestartFinish
: public MDSIOContext
{
1698 uint64_t old_read_pos
;
1700 C_MDS_StandbyReplayRestartFinish(MDSRank
*mds_
, uint64_t old_read_pos_
) :
1701 MDSIOContext(mds_
), old_read_pos(old_read_pos_
) {}
1702 void finish(int r
) override
{
1703 mds
->_standby_replay_restart_finish(r
, old_read_pos
);
1705 void print(ostream
& out
) const override
{
1706 out
<< "standby_replay_restart";
1710 void MDSRank::_standby_replay_restart_finish(int r
, uint64_t old_read_pos
)
1712 if (old_read_pos
< mdlog
->get_journaler()->get_trimmed_pos()) {
1713 dout(0) << "standby MDS fell behind active MDS journal's expire_pos, restarting" << dendl
;
1714 respawn(); /* we're too far back, and this is easier than
1715 trying to reset everything in the cache, etc */
1717 mdlog
->standby_trim_segments();
1718 boot_start(MDS_BOOT_PREPARE_LOG
, r
);
1722 class MDSRank::C_MDS_StandbyReplayRestart
: public MDSInternalContext
{
1724 explicit C_MDS_StandbyReplayRestart(MDSRank
*m
) : MDSInternalContext(m
) {}
1725 void finish(int r
) override
{
1727 mds
->standby_replay_restart();
1731 void MDSRank::standby_replay_restart()
1733 if (standby_replaying
) {
1734 /* Go around for another pass of replaying in standby */
1735 dout(5) << "Restarting replay as standby-replay" << dendl
;
1736 mdlog
->get_journaler()->reread_head_and_probe(
1737 new C_MDS_StandbyReplayRestartFinish(
1739 mdlog
->get_journaler()->get_read_pos()));
1741 /* We are transitioning out of standby: wait for OSD map update
1742 before making final pass */
1743 dout(1) << "standby_replay_restart (final takeover pass)" << dendl
;
1744 Context
*fin
= new C_IO_Wrapper(this, new C_MDS_StandbyReplayRestart(this));
1745 bool ready
= objecter
->wait_for_map(mdsmap
->get_last_failure_osd_epoch(), fin
);
1748 mdlog
->get_journaler()->reread_head_and_probe(
1749 new C_MDS_StandbyReplayRestartFinish(
1751 mdlog
->get_journaler()->get_read_pos()));
1753 dout(1) << " opening purge_queue (async)" << dendl
;
1754 purge_queue
.open(NULL
);
1755 dout(1) << " opening open_file_table (async)" << dendl
;
1756 mdcache
->open_file_table
.load(nullptr);
1758 dout(1) << " waiting for osdmap " << mdsmap
->get_last_failure_osd_epoch()
1759 << " (which blacklists prior instance)" << dendl
;
1764 void MDSRank::replay_done()
1766 if (!standby_replaying
) {
1767 dout(1) << "Finished replaying journal" << dendl
;
1769 dout(5) << "Finished replaying journal as standby-replay" << dendl
;
1772 if (is_standby_replay()) {
1773 // The replay was done in standby state, and we are still in that state
1774 ceph_assert(standby_replaying
);
1775 dout(10) << "setting replay timer" << dendl
;
1776 timer
.add_event_after(g_conf()->mds_replay_interval
,
1777 new C_MDS_StandbyReplayRestart(this));
1779 } else if (standby_replaying
) {
1780 // The replay was done in standby state, we have now _left_ that state
1781 dout(10) << " last replay pass was as a standby; making final pass" << dendl
;
1782 standby_replaying
= false;
1783 standby_replay_restart();
1786 // Replay is complete, journal read should be up to date
1787 ceph_assert(mdlog
->get_journaler()->get_read_pos() == mdlog
->get_journaler()->get_write_pos());
1788 ceph_assert(!is_standby_replay());
1790 // Reformat and come back here
1791 if (mdlog
->get_journaler()->get_stream_format() < g_conf()->mds_journal_format
) {
1792 dout(4) << "reformatting journal on standby-replay->replay transition" << dendl
;
1793 mdlog
->reopen(new C_MDS_BootStart(this, MDS_BOOT_REPLAY_DONE
));
1798 dout(1) << "making mds journal writeable" << dendl
;
1799 mdlog
->get_journaler()->set_writeable();
1800 mdlog
->get_journaler()->trim_tail();
1802 if (mdsmap
->get_tableserver() == whoami
&&
1803 snapserver
->upgrade_format()) {
1804 dout(1) << "upgrading snaptable format" << dendl
;
1805 snapserver
->save(new C_MDSInternalNoop
);
1808 if (g_conf()->mds_wipe_sessions
) {
1809 dout(1) << "wiping out client sessions" << dendl
;
1811 sessionmap
.save(new C_MDSInternalNoop
);
1813 if (g_conf()->mds_wipe_ino_prealloc
) {
1814 dout(1) << "wiping out ino prealloc from sessions" << dendl
;
1815 sessionmap
.wipe_ino_prealloc();
1816 sessionmap
.save(new C_MDSInternalNoop
);
1818 if (g_conf()->mds_skip_ino
) {
1819 inodeno_t i
= g_conf()->mds_skip_ino
;
1820 dout(1) << "skipping " << i
<< " inodes" << dendl
;
1821 inotable
->skip_inos(i
);
1822 inotable
->save(new C_MDSInternalNoop
);
1825 if (mdsmap
->get_num_in_mds() == 1 &&
1826 mdsmap
->get_num_failed_mds() == 0) { // just me!
1827 dout(2) << "i am alone, moving to state reconnect" << dendl
;
1828 request_state(MDSMap::STATE_RECONNECT
);
1829 // sync snaptable cache
1830 snapclient
->sync(new C_MDSInternalNoop
);
1832 dout(2) << "i am not alone, moving to state resolve" << dendl
;
1833 request_state(MDSMap::STATE_RESOLVE
);
1837 void MDSRank::reopen_log()
1839 dout(1) << "reopen_log" << dendl
;
1840 mdcache
->rollback_uncommitted_fragments();
1843 void MDSRank::resolve_start()
1845 dout(1) << "resolve_start" << dendl
;
1849 mdcache
->resolve_start(new C_MDS_VoidFn(this, &MDSRank::resolve_done
));
1850 finish_contexts(g_ceph_context
, waiting_for_resolve
);
1853 void MDSRank::resolve_done()
1855 dout(1) << "resolve_done" << dendl
;
1856 request_state(MDSMap::STATE_RECONNECT
);
1857 // sync snaptable cache
1858 snapclient
->sync(new C_MDSInternalNoop
);
1861 void MDSRank::reconnect_start()
1863 dout(1) << "reconnect_start" << dendl
;
1865 if (last_state
== MDSMap::STATE_REPLAY
) {
1869 // Drop any blacklisted clients from the SessionMap before going
1870 // into reconnect, so that we don't wait for them.
1871 objecter
->enable_blacklist_events();
1872 std::set
<entity_addr_t
> blacklist
;
1874 objecter
->with_osdmap([&blacklist
, &epoch
](const OSDMap
& o
) {
1875 o
.get_blacklist(&blacklist
);
1876 epoch
= o
.get_epoch();
1878 auto killed
= server
->apply_blacklist(blacklist
);
1879 dout(4) << "reconnect_start: killed " << killed
<< " blacklisted sessions ("
1880 << blacklist
.size() << " blacklist entries, "
1881 << sessionmap
.get_sessions().size() << ")" << dendl
;
1883 set_osd_epoch_barrier(epoch
);
1886 server
->reconnect_clients(new C_MDS_VoidFn(this, &MDSRank::reconnect_done
));
1887 finish_contexts(g_ceph_context
, waiting_for_reconnect
);
1889 void MDSRank::reconnect_done()
1891 dout(1) << "reconnect_done" << dendl
;
1892 request_state(MDSMap::STATE_REJOIN
); // move to rejoin state
1895 void MDSRank::rejoin_joint_start()
1897 dout(1) << "rejoin_joint_start" << dendl
;
1898 mdcache
->rejoin_send_rejoins();
1900 void MDSRank::rejoin_start()
1902 dout(1) << "rejoin_start" << dendl
;
1903 mdcache
->rejoin_start(new C_MDS_VoidFn(this, &MDSRank::rejoin_done
));
1904 finish_contexts(g_ceph_context
, waiting_for_rejoin
);
1906 void MDSRank::rejoin_done()
1908 dout(1) << "rejoin_done" << dendl
;
1909 mdcache
->show_subtrees();
1910 mdcache
->show_cache();
1912 // funny case: is our cache empty? no subtrees?
1913 if (!mdcache
->is_subtrees()) {
1915 // The root should always have a subtree!
1916 clog
->error() << "No subtrees found for root MDS rank!";
1918 ceph_assert(mdcache
->is_subtrees());
1920 dout(1) << " empty cache, no subtrees, leaving cluster" << dendl
;
1921 request_state(MDSMap::STATE_STOPPED
);
1926 if (replay_queue
.empty() && !server
->get_num_pending_reclaim()) {
1927 request_state(MDSMap::STATE_ACTIVE
);
1929 replaying_requests_done
= replay_queue
.empty();
1930 request_state(MDSMap::STATE_CLIENTREPLAY
);
1934 void MDSRank::clientreplay_start()
1936 dout(1) << "clientreplay_start" << dendl
;
1937 finish_contexts(g_ceph_context
, waiting_for_replay
); // kick waiters
1938 mdcache
->start_files_to_recover();
1942 bool MDSRank::queue_one_replay()
1944 if (!replay_queue
.empty()) {
1945 queue_waiter(replay_queue
.front());
1946 replay_queue
.pop_front();
1949 if (!replaying_requests_done
) {
1950 replaying_requests_done
= true;
1953 maybe_clientreplay_done();
1957 void MDSRank::maybe_clientreplay_done()
1959 if (is_clientreplay() && get_want_state() == MDSMap::STATE_CLIENTREPLAY
) {
1961 // don't go to active if there are session waiting for being reclaimed
1962 if (replaying_requests_done
&& !server
->get_num_pending_reclaim()) {
1963 mdlog
->wait_for_safe(new C_MDS_VoidFn(this, &MDSRank::clientreplay_done
));
1967 dout(1) << " still have " << replay_queue
.size() + (int)!replaying_requests_done
1968 << " requests need to be replayed, " << server
->get_num_pending_reclaim()
1969 << " sessions need to be reclaimed" << dendl
;
1973 void MDSRank::clientreplay_done()
1975 dout(1) << "clientreplay_done" << dendl
;
1976 request_state(MDSMap::STATE_ACTIVE
);
1979 void MDSRank::active_start()
1981 dout(1) << "active_start" << dendl
;
1983 if (last_state
== MDSMap::STATE_CREATING
||
1984 last_state
== MDSMap::STATE_STARTING
) {
1985 mdcache
->open_root();
1988 mdcache
->clean_open_file_lists();
1989 mdcache
->export_remaining_imported_caps();
1990 finish_contexts(g_ceph_context
, waiting_for_replay
); // kick waiters
1991 mdcache
->start_files_to_recover();
1993 mdcache
->reissue_all_caps();
1995 finish_contexts(g_ceph_context
, waiting_for_active
); // kick waiters
1998 void MDSRank::recovery_done(int oldstate
)
2000 dout(1) << "recovery_done -- successful recovery!" << dendl
;
2001 ceph_assert(is_clientreplay() || is_active());
2003 if (oldstate
== MDSMap::STATE_CREATING
)
2006 mdcache
->start_recovered_truncates();
2007 mdcache
->do_file_recover();
2009 // tell connected clients
2010 //bcast_mds_map(); // not anymore, they get this from the monitor
2012 mdcache
->populate_mydir();
2015 void MDSRank::creating_done()
2017 dout(1)<< "creating_done" << dendl
;
2018 request_state(MDSMap::STATE_ACTIVE
);
2019 // sync snaptable cache
2020 snapclient
->sync(new C_MDSInternalNoop
);
2023 void MDSRank::boot_create()
2025 dout(3) << "boot_create" << dendl
;
2027 MDSGatherBuilder
fin(g_ceph_context
, new C_MDS_VoidFn(this, &MDSRank::creating_done
));
2029 mdcache
->init_layouts();
2031 inotable
->set_rank(whoami
);
2032 sessionmap
.set_rank(whoami
);
2034 // start with a fresh journal
2035 dout(10) << "boot_create creating fresh journal" << dendl
;
2036 mdlog
->create(fin
.new_sub());
2038 // open new journal segment, but do not journal subtree map (yet)
2039 mdlog
->prepare_new_segment();
2041 if (whoami
== mdsmap
->get_root()) {
2042 dout(3) << "boot_create creating fresh hierarchy" << dendl
;
2043 mdcache
->create_empty_hierarchy(fin
.get());
2046 dout(3) << "boot_create creating mydir hierarchy" << dendl
;
2047 mdcache
->create_mydir_hierarchy(fin
.get());
2049 dout(3) << "boot_create creating global snaprealm" << dendl
;
2050 mdcache
->create_global_snaprealm();
2052 // fixme: fake out inotable (reset, pretend loaded)
2053 dout(10) << "boot_create creating fresh inotable table" << dendl
;
2055 inotable
->save(fin
.new_sub());
2057 // write empty sessionmap
2058 sessionmap
.save(fin
.new_sub());
2060 // Create empty purge queue
2061 purge_queue
.create(new C_IO_Wrapper(this, fin
.new_sub()));
2063 // initialize tables
2064 if (mdsmap
->get_tableserver() == whoami
) {
2065 dout(10) << "boot_create creating fresh snaptable" << dendl
;
2066 snapserver
->set_rank(whoami
);
2067 snapserver
->reset();
2068 snapserver
->save(fin
.new_sub());
2071 ceph_assert(g_conf()->mds_kill_create_at
!= 1);
2073 // ok now journal it
2074 mdlog
->journal_segment_subtree_map(fin
.new_sub());
2077 // Usually we do this during reconnect, but creation skips that.
2078 objecter
->enable_blacklist_events();
2083 void MDSRank::stopping_start()
2085 dout(2) << "Stopping..." << dendl
;
2087 if (mdsmap
->get_num_in_mds() == 1 && !sessionmap
.empty()) {
2088 std::vector
<Session
*> victims
;
2089 const auto& sessions
= sessionmap
.get_sessions();
2090 for (const auto& p
: sessions
) {
2091 if (!p
.first
.is_client()) {
2095 Session
*s
= p
.second
;
2096 victims
.push_back(s
);
2099 dout(20) << __func__
<< " matched " << victims
.size() << " sessions" << dendl
;
2100 ceph_assert(!victims
.empty());
2102 C_GatherBuilder
gather(g_ceph_context
, new C_MDSInternalNoop
);
2103 for (const auto &s
: victims
) {
2104 std::stringstream ss
;
2105 evict_client(s
->get_client().v
, false,
2106 g_conf()->mds_session_blacklist_on_evict
, ss
, gather
.new_sub());
2111 mdcache
->shutdown_start();
2114 void MDSRank::stopping_done()
2116 dout(2) << "Finished stopping..." << dendl
;
2118 // tell monitor we shut down cleanly.
2119 request_state(MDSMap::STATE_STOPPED
);
2122 void MDSRankDispatcher::handle_mds_map(
2123 const MMDSMap::const_ref
&m
,
2124 const MDSMap
&oldmap
)
2126 // I am only to be passed MDSMaps in which I hold a rank
2127 ceph_assert(whoami
!= MDS_RANK_NONE
);
2129 MDSMap::DaemonState oldstate
= state
;
2130 mds_gid_t mds_gid
= mds_gid_t(monc
->get_global_id());
2131 state
= mdsmap
->get_state_gid(mds_gid
);
2132 if (state
!= oldstate
) {
2133 last_state
= oldstate
;
2134 incarnation
= mdsmap
->get_inc_gid(mds_gid
);
2137 version_t epoch
= m
->get_epoch();
2139 // note source's map version
2140 if (m
->get_source().is_mds() &&
2141 peer_mdsmap_epoch
[mds_rank_t(m
->get_source().num())] < epoch
) {
2142 dout(15) << " peer " << m
->get_source()
2143 << " has mdsmap epoch >= " << epoch
2145 peer_mdsmap_epoch
[mds_rank_t(m
->get_source().num())] = epoch
;
2148 // Validate state transitions while I hold a rank
2149 if (!MDSMap::state_transition_valid(oldstate
, state
)) {
2150 derr
<< "Invalid state transition " << ceph_mds_state_name(oldstate
)
2151 << "->" << ceph_mds_state_name(state
) << dendl
;
2155 if (oldstate
!= state
) {
2156 // update messenger.
2157 if (state
== MDSMap::STATE_STANDBY_REPLAY
) {
2158 dout(1) << "handle_mds_map i am now mds." << mds_gid
<< "." << incarnation
2159 << " replaying mds." << whoami
<< "." << incarnation
<< dendl
;
2160 messenger
->set_myname(entity_name_t::MDS(mds_gid
));
2162 dout(1) << "handle_mds_map i am now mds." << whoami
<< "." << incarnation
<< dendl
;
2163 messenger
->set_myname(entity_name_t::MDS(whoami
));
2167 // tell objecter my incarnation
2168 if (objecter
->get_client_incarnation() != incarnation
)
2169 objecter
->set_client_incarnation(incarnation
);
2171 if (oldmap
.get_min_compat_client() != mdsmap
->get_min_compat_client())
2172 server
->update_required_client_features();
2175 if (g_conf()->mds_dump_cache_on_map
)
2176 mdcache
->dump_cache();
2178 cluster_degraded
= mdsmap
->is_degraded();
2180 // mdsmap and oldmap can be discontinuous. failover might happen in the missing mdsmap.
2181 // the 'restart' set tracks ranks that have restarted since the old mdsmap
2182 set
<mds_rank_t
> restart
;
2183 // replaying mds does not communicate with other ranks
2184 if (state
>= MDSMap::STATE_RESOLVE
) {
2185 // did someone fail?
2187 set
<mds_rank_t
> olddown
, down
;
2188 oldmap
.get_down_mds_set(&olddown
);
2189 mdsmap
->get_down_mds_set(&down
);
2190 for (const auto& r
: down
) {
2191 if (oldmap
.have_inst(r
) && olddown
.count(r
) == 0) {
2192 messenger
->mark_down_addrs(oldmap
.get_addrs(r
));
2193 handle_mds_failure(r
);
2197 // did someone fail?
2198 // did their addr/inst change?
2200 mdsmap
->get_up_mds_set(up
);
2201 for (const auto& r
: up
) {
2202 auto& info
= mdsmap
->get_info(r
);
2203 if (oldmap
.have_inst(r
)) {
2204 auto& oldinfo
= oldmap
.get_info(r
);
2205 if (info
.inc
!= oldinfo
.inc
) {
2206 messenger
->mark_down_addrs(oldinfo
.get_addrs());
2207 if (info
.state
== MDSMap::STATE_REPLAY
||
2208 info
.state
== MDSMap::STATE_RESOLVE
) {
2210 handle_mds_failure(r
);
2212 ceph_assert(info
.state
== MDSMap::STATE_STARTING
||
2213 info
.state
== MDSMap::STATE_ACTIVE
);
2214 // -> stopped (missing) -> starting -> active
2216 mdcache
->migrator
->handle_mds_failure_or_stop(r
);
2217 if (mdsmap
->get_tableserver() == whoami
)
2218 snapserver
->handle_mds_failure_or_stop(r
);
2222 if (info
.state
== MDSMap::STATE_REPLAY
||
2223 info
.state
== MDSMap::STATE_RESOLVE
) {
2224 // -> starting/creating (missing) -> active (missing) -> replay -> resolve
2226 handle_mds_failure(r
);
2228 ceph_assert(info
.state
== MDSMap::STATE_CREATING
||
2229 info
.state
== MDSMap::STATE_STARTING
||
2230 info
.state
== MDSMap::STATE_ACTIVE
);
2237 if (oldstate
!= state
) {
2238 dout(1) << "handle_mds_map state change "
2239 << ceph_mds_state_name(oldstate
) << " --> "
2240 << ceph_mds_state_name(state
) << dendl
;
2241 beacon
.set_want_state(*mdsmap
, state
);
2243 if (oldstate
== MDSMap::STATE_STANDBY_REPLAY
) {
2244 dout(10) << "Monitor activated us! Deactivating replay loop" << dendl
;
2245 assert (state
== MDSMap::STATE_REPLAY
);
2247 // did i just recover?
2248 if ((is_active() || is_clientreplay()) &&
2249 (oldstate
== MDSMap::STATE_CREATING
||
2250 oldstate
== MDSMap::STATE_REJOIN
||
2251 oldstate
== MDSMap::STATE_RECONNECT
))
2252 recovery_done(oldstate
);
2256 } else if (is_any_replay()) {
2258 } else if (is_resolve()) {
2260 } else if (is_reconnect()) {
2262 } else if (is_rejoin()) {
2264 } else if (is_clientreplay()) {
2265 clientreplay_start();
2266 } else if (is_creating()) {
2268 } else if (is_starting()) {
2270 } else if (is_stopping()) {
2271 ceph_assert(oldstate
== MDSMap::STATE_ACTIVE
);
2278 // is someone else newly resolving?
2279 if (state
>= MDSMap::STATE_RESOLVE
) {
2280 // recover snaptable
2281 if (mdsmap
->get_tableserver() == whoami
) {
2282 if (oldstate
< MDSMap::STATE_RESOLVE
) {
2284 mdsmap
->get_mds_set_lower_bound(s
, MDSMap::STATE_RESOLVE
);
2285 snapserver
->finish_recovery(s
);
2287 set
<mds_rank_t
> old_set
, new_set
;
2288 oldmap
.get_mds_set_lower_bound(old_set
, MDSMap::STATE_RESOLVE
);
2289 mdsmap
->get_mds_set_lower_bound(new_set
, MDSMap::STATE_RESOLVE
);
2290 for (const auto& r
: new_set
) {
2293 if (!old_set
.count(r
) || restart
.count(r
)) { // newly so?
2294 snapserver
->handle_mds_recovery(r
);
2300 if ((!oldmap
.is_resolving() || !restart
.empty()) && mdsmap
->is_resolving()) {
2301 set
<mds_rank_t
> resolve
;
2302 mdsmap
->get_mds_set(resolve
, MDSMap::STATE_RESOLVE
);
2303 dout(10) << " resolve set is " << resolve
<< dendl
;
2304 calc_recovery_set();
2305 mdcache
->send_resolves();
2310 // is everybody finally rejoining?
2311 if (state
>= MDSMap::STATE_REJOIN
) {
2313 if (!oldmap
.is_rejoining() && mdsmap
->is_rejoining())
2314 rejoin_joint_start();
2317 if (g_conf()->mds_dump_cache_after_rejoin
&&
2318 oldmap
.is_rejoining() && !mdsmap
->is_rejoining())
2319 mdcache
->dump_cache(); // for DEBUG only
2321 if (oldstate
>= MDSMap::STATE_REJOIN
||
2322 oldstate
== MDSMap::STATE_STARTING
) {
2323 // ACTIVE|CLIENTREPLAY|REJOIN => we can discover from them.
2324 set
<mds_rank_t
> olddis
, dis
;
2325 oldmap
.get_mds_set_lower_bound(olddis
, MDSMap::STATE_REJOIN
);
2326 mdsmap
->get_mds_set_lower_bound(dis
, MDSMap::STATE_REJOIN
);
2327 for (const auto& r
: dis
) {
2330 if (!olddis
.count(r
) || restart
.count(r
)) { // newly so?
2331 mdcache
->kick_discovers(r
);
2332 mdcache
->kick_open_ino_peers(r
);
2338 if (oldmap
.is_degraded() && !cluster_degraded
&& state
>= MDSMap::STATE_ACTIVE
) {
2339 dout(1) << "cluster recovered." << dendl
;
2340 auto it
= waiting_for_active_peer
.find(MDS_RANK_NONE
);
2341 if (it
!= waiting_for_active_peer
.end()) {
2342 queue_waiters(it
->second
);
2343 waiting_for_active_peer
.erase(it
);
2347 // did someone go active?
2348 if (state
>= MDSMap::STATE_CLIENTREPLAY
&&
2349 oldstate
>= MDSMap::STATE_CLIENTREPLAY
) {
2350 set
<mds_rank_t
> oldactive
, active
;
2351 oldmap
.get_mds_set_lower_bound(oldactive
, MDSMap::STATE_CLIENTREPLAY
);
2352 mdsmap
->get_mds_set_lower_bound(active
, MDSMap::STATE_CLIENTREPLAY
);
2353 for (const auto& r
: active
) {
2356 if (!oldactive
.count(r
) || restart
.count(r
)) // newly so?
2357 handle_mds_recovery(r
);
2361 if (is_clientreplay() || is_active() || is_stopping()) {
2363 set
<mds_rank_t
> oldstopped
, stopped
;
2364 oldmap
.get_stopped_mds_set(oldstopped
);
2365 mdsmap
->get_stopped_mds_set(stopped
);
2366 for (const auto& r
: stopped
)
2367 if (oldstopped
.count(r
) == 0) { // newly so?
2368 mdcache
->migrator
->handle_mds_failure_or_stop(r
);
2369 if (mdsmap
->get_tableserver() == whoami
)
2370 snapserver
->handle_mds_failure_or_stop(r
);
2375 map
<epoch_t
,MDSContext::vec
>::iterator p
= waiting_for_mdsmap
.begin();
2376 while (p
!= waiting_for_mdsmap
.end() && p
->first
<= mdsmap
->get_epoch()) {
2379 waiting_for_mdsmap
.erase(p
++);
2385 // Before going active, set OSD epoch barrier to latest (so that
2386 // we don't risk handing out caps to clients with old OSD maps that
2387 // might not include barriers from the previous incarnation of this MDS)
2388 set_osd_epoch_barrier(objecter
->with_osdmap(
2389 std::mem_fn(&OSDMap::get_epoch
)));
2391 /* Now check if we should hint to the OSD that a read may follow */
2392 if (mdsmap
->has_standby_replay(whoami
))
2393 mdlog
->set_write_iohint(0);
2395 mdlog
->set_write_iohint(CEPH_OSD_OP_FLAG_FADVISE_DONTNEED
);
2398 if (oldmap
.get_max_mds() != mdsmap
->get_max_mds()) {
2399 purge_queue
.update_op_limit(*mdsmap
);
2402 mdcache
->handle_mdsmap(*mdsmap
);
2405 void MDSRank::handle_mds_recovery(mds_rank_t who
)
2407 dout(5) << "handle_mds_recovery mds." << who
<< dendl
;
2409 mdcache
->handle_mds_recovery(who
);
2411 queue_waiters(waiting_for_active_peer
[who
]);
2412 waiting_for_active_peer
.erase(who
);
2415 void MDSRank::handle_mds_failure(mds_rank_t who
)
2417 if (who
== whoami
) {
2418 dout(5) << "handle_mds_failure for myself; not doing anything" << dendl
;
2421 dout(5) << "handle_mds_failure mds." << who
<< dendl
;
2423 mdcache
->handle_mds_failure(who
);
2425 if (mdsmap
->get_tableserver() == whoami
)
2426 snapserver
->handle_mds_failure_or_stop(who
);
2428 snapclient
->handle_mds_failure(who
);
2431 bool MDSRankDispatcher::handle_asok_command(std::string_view command
,
2432 const cmdmap_t
& cmdmap
,
2436 if (command
== "dump_ops_in_flight" ||
2438 if (!op_tracker
.dump_ops_in_flight(f
)) {
2439 ss
<< "op_tracker tracking is not enabled now, so no ops are tracked currently, even those get stuck. \
2440 please enable \"mds_enable_op_tracker\", and the tracker will start to track new ops received afterwards.";
2442 } else if (command
== "dump_blocked_ops") {
2443 if (!op_tracker
.dump_ops_in_flight(f
, true)) {
2444 ss
<< "op_tracker tracking is not enabled now, so no ops are tracked currently, even those get stuck. \
2445 Please enable \"mds_enable_op_tracker\", and the tracker will start to track new ops received afterwards.";
2447 } else if (command
== "dump_historic_ops") {
2448 if (!op_tracker
.dump_historic_ops(f
)) {
2449 ss
<< "op_tracker tracking is not enabled now, so no ops are tracked currently, even those get stuck. \
2450 please enable \"mds_enable_op_tracker\", and the tracker will start to track new ops received afterwards.";
2452 } else if (command
== "dump_historic_ops_by_duration") {
2453 if (!op_tracker
.dump_historic_ops(f
, true)) {
2454 ss
<< "op_tracker tracking is not enabled now, so no ops are tracked currently, even those get stuck. \
2455 please enable \"mds_enable_op_tracker\", and the tracker will start to track new ops received afterwards.";
2457 } else if (command
== "osdmap barrier") {
2458 int64_t target_epoch
= 0;
2459 bool got_val
= cmd_getval(g_ceph_context
, cmdmap
, "target_epoch", target_epoch
);
2462 ss
<< "no target epoch given";
2467 set_osd_epoch_barrier(target_epoch
);
2471 bool already_got
= objecter
->wait_for_map(target_epoch
, &cond
);
2473 dout(4) << __func__
<< ": waiting for OSD epoch " << target_epoch
<< dendl
;
2476 } else if (command
== "session ls") {
2477 std::lock_guard
l(mds_lock
);
2481 dump_sessions(SessionFilter(), f
);
2482 } else if (command
== "session evict") {
2483 std::string client_id
;
2484 const bool got_arg
= cmd_getval(g_ceph_context
, cmdmap
, "client_id", client_id
);
2486 ss
<< "Invalid client_id specified";
2491 std::stringstream dss
;
2492 bool evicted
= evict_client(strtol(client_id
.c_str(), 0, 10), true,
2493 g_conf()->mds_session_blacklist_on_evict
, dss
);
2495 dout(15) << dss
.str() << dendl
;
2499 } else if (command
== "scrub_path") {
2501 vector
<string
> scrubop_vec
;
2502 cmd_getval(g_ceph_context
, cmdmap
, "scrubops", scrubop_vec
);
2503 cmd_getval(g_ceph_context
, cmdmap
, "path", path
);
2506 command_scrub_start(f
, path
, "", scrubop_vec
, &cond
);
2508 } else if (command
== "tag path") {
2510 cmd_getval(g_ceph_context
, cmdmap
, "path", path
);
2512 cmd_getval(g_ceph_context
, cmdmap
, "tag", tag
);
2513 command_tag_path(f
, path
, tag
);
2514 } else if (command
== "flush_path") {
2516 cmd_getval(g_ceph_context
, cmdmap
, "path", path
);
2517 command_flush_path(f
, path
);
2518 } else if (command
== "flush journal") {
2519 command_flush_journal(f
);
2520 } else if (command
== "get subtrees") {
2521 command_get_subtrees(f
);
2522 } else if (command
== "export dir") {
2524 if(!cmd_getval(g_ceph_context
, cmdmap
, "path", path
)) {
2525 ss
<< "malformed path";
2529 if(!cmd_getval(g_ceph_context
, cmdmap
, "rank", rank
)) {
2530 ss
<< "malformed rank";
2533 command_export_dir(f
, path
, (mds_rank_t
)rank
);
2534 } else if (command
== "dump cache") {
2535 std::lock_guard
l(mds_lock
);
2538 if(!cmd_getval(g_ceph_context
, cmdmap
, "path", path
)) {
2539 r
= mdcache
->dump_cache(f
);
2541 r
= mdcache
->dump_cache(path
);
2545 ss
<< "Failed to dump cache: " << cpp_strerror(r
);
2548 } else if (command
== "cache status") {
2549 std::lock_guard
l(mds_lock
);
2550 mdcache
->cache_status(f
);
2551 } else if (command
== "dump tree") {
2552 command_dump_tree(cmdmap
, ss
, f
);
2553 } else if (command
== "dump loads") {
2554 std::lock_guard
l(mds_lock
);
2555 int r
= balancer
->dump_loads(f
);
2557 ss
<< "Failed to dump loads: " << cpp_strerror(r
);
2560 } else if (command
== "dump snaps") {
2561 std::lock_guard
l(mds_lock
);
2563 cmd_getval(g_ceph_context
, cmdmap
, "server", server
);
2564 if (server
== "--server") {
2565 if (mdsmap
->get_tableserver() == whoami
) {
2566 snapserver
->dump(f
);
2568 ss
<< "Not snapserver";
2571 int r
= snapclient
->dump_cache(f
);
2573 ss
<< "Failed to dump snapclient: " << cpp_strerror(r
);
2577 } else if (command
== "force_readonly") {
2578 std::lock_guard
l(mds_lock
);
2579 mdcache
->force_readonly();
2580 } else if (command
== "dirfrag split") {
2581 command_dirfrag_split(cmdmap
, ss
);
2582 } else if (command
== "dirfrag merge") {
2583 command_dirfrag_merge(cmdmap
, ss
);
2584 } else if (command
== "dirfrag ls") {
2585 command_dirfrag_ls(cmdmap
, ss
, f
);
2586 } else if (command
== "openfiles ls") {
2587 command_openfiles_ls(f
);
2588 } else if (command
== "dump inode") {
2589 command_dump_inode(f
, cmdmap
, ss
);
2597 class C_MDS_Send_Command_Reply
: public MDSInternalContext
{
2599 MCommand::const_ref m
;
2601 C_MDS_Send_Command_Reply(MDSRank
*_mds
, const MCommand::const_ref
&_m
) :
2602 MDSInternalContext(_mds
), m(_m
) {}
2604 void send(int r
, std::string_view ss
) {
2605 std::stringstream ds
;
2609 void send(int r
, std::string_view ss
, std::stringstream
&ds
) {
2612 MDSDaemon::send_command_reply(m
, mds
, r
, bl
, ss
);
2615 void finish(int r
) override
{
2620 class C_ExecAndReply
: public C_MDS_Send_Command_Reply
{
2622 C_ExecAndReply(MDSRank
*mds
, const MCommand::const_ref
&m
)
2623 : C_MDS_Send_Command_Reply(mds
, m
), f(true) {
2626 void finish(int r
) override
{
2627 std::stringstream ds
;
2628 std::stringstream ss
;
2635 send(r
, ss
.str(), ds
);
2638 virtual void exec() = 0;
2644 class C_CacheDropExecAndReply
: public C_ExecAndReply
{
2646 C_CacheDropExecAndReply(MDSRank
*mds
, const MCommand::const_ref
&m
,
2648 : C_ExecAndReply(mds
, m
), timeout(timeout
) {
2651 void exec() override
{
2652 mds
->command_cache_drop(timeout
, &f
, this);
2659 class C_ScrubExecAndReply
: public C_ExecAndReply
{
2661 C_ScrubExecAndReply(MDSRank
*mds
, const MCommand::const_ref
&m
,
2662 const std::string
&path
, const std::string
&tag
,
2663 const std::vector
<std::string
> &scrubop
)
2664 : C_ExecAndReply(mds
, m
), path(path
), tag(tag
), scrubop(scrubop
) {
2667 void exec() override
{
2668 mds
->command_scrub_start(&f
, path
, tag
, scrubop
, this);
2674 std::vector
<std::string
> scrubop
;
2677 class C_ScrubControlExecAndReply
: public C_ExecAndReply
{
2679 C_ScrubControlExecAndReply(MDSRank
*mds
, const MCommand::const_ref
&m
,
2680 const std::string
&command
)
2681 : C_ExecAndReply(mds
, m
), command(command
) {
2684 void exec() override
{
2685 if (command
== "abort") {
2686 mds
->command_scrub_abort(&f
, this);
2687 } else if (command
== "pause") {
2688 mds
->command_scrub_pause(&f
, this);
2694 void finish(int r
) override
{
2695 f
.open_object_section("result");
2696 f
.dump_int("return_code", r
);
2698 C_ExecAndReply::finish(r
);
2702 std::string command
;
2706 * This function drops the mds_lock, so don't do anything with
2707 * MDSRank after calling it (we could have gone into shutdown): just
2708 * send your result back to the calling client and finish.
2710 void MDSRankDispatcher::evict_clients(const SessionFilter
&filter
, const MCommand::const_ref
&m
)
2712 C_MDS_Send_Command_Reply
*reply
= new C_MDS_Send_Command_Reply(this, m
);
2714 if (is_any_replay()) {
2715 reply
->send(-EAGAIN
, "MDS is replaying log");
2720 std::vector
<Session
*> victims
;
2721 const auto& sessions
= sessionmap
.get_sessions();
2722 for (const auto& p
: sessions
) {
2723 if (!p
.first
.is_client()) {
2727 Session
*s
= p
.second
;
2729 if (filter
.match(*s
, std::bind(&Server::waiting_for_reconnect
, server
, std::placeholders::_1
))) {
2730 victims
.push_back(s
);
2734 dout(20) << __func__
<< " matched " << victims
.size() << " sessions" << dendl
;
2736 if (victims
.empty()) {
2742 C_GatherBuilder
gather(g_ceph_context
, reply
);
2743 for (const auto s
: victims
) {
2744 std::stringstream ss
;
2745 evict_client(s
->get_client().v
, false,
2746 g_conf()->mds_session_blacklist_on_evict
, ss
, gather
.new_sub());
2751 void MDSRankDispatcher::dump_sessions(const SessionFilter
&filter
, Formatter
*f
) const
2753 // Dump sessions, decorated with recovery/replay status
2754 f
->open_array_section("sessions");
2755 const ceph::unordered_map
<entity_name_t
, Session
*> session_map
= sessionmap
.get_sessions();
2756 for (auto& p
: session_map
) {
2757 if (!p
.first
.is_client()) {
2761 Session
*s
= p
.second
;
2763 if (!filter
.match(*s
, std::bind(&Server::waiting_for_reconnect
, server
, std::placeholders::_1
))) {
2767 f
->open_object_section("session");
2768 f
->dump_int("id", p
.first
.num());
2770 f
->dump_int("num_leases", s
->leases
.size());
2771 f
->dump_int("num_caps", s
->caps
.size());
2773 f
->dump_string("state", s
->get_state_name());
2774 if (s
->is_open() || s
->is_stale()) {
2775 f
->dump_unsigned("request_load_avg", s
->get_load_avg());
2777 f
->dump_float("uptime", s
->get_session_uptime());
2778 f
->dump_int("replay_requests", is_clientreplay() ? s
->get_request_count() : 0);
2779 f
->dump_unsigned("completed_requests", s
->get_num_completed_requests());
2780 f
->dump_bool("reconnecting", server
->waiting_for_reconnect(p
.first
.num()));
2781 f
->dump_stream("inst") << s
->info
.inst
;
2782 f
->open_object_section("client_metadata");
2783 s
->info
.client_metadata
.dump(f
);
2784 f
->close_section(); // client_metadata
2785 f
->close_section(); //session
2787 f
->close_section(); //sessions
2790 void MDSRank::command_scrub_start(Formatter
*f
,
2791 std::string_view path
, std::string_view tag
,
2792 const vector
<string
>& scrubop_vec
, Context
*on_finish
)
2795 bool recursive
= false;
2796 bool repair
= false;
2797 for (auto &op
: scrubop_vec
) {
2800 else if (op
== "recursive")
2802 else if (op
== "repair")
2806 std::lock_guard
l(mds_lock
);
2807 mdcache
->enqueue_scrub(path
, tag
, force
, recursive
, repair
, f
, on_finish
);
2808 // scrub_dentry() finishers will dump the data for us; we're done!
2811 void MDSRank::command_tag_path(Formatter
*f
,
2812 std::string_view path
, std::string_view tag
)
2816 std::lock_guard
l(mds_lock
);
2817 mdcache
->enqueue_scrub(path
, tag
, true, true, false, f
, &scond
);
2822 void MDSRank::command_scrub_abort(Formatter
*f
, Context
*on_finish
) {
2823 std::lock_guard
l(mds_lock
);
2824 scrubstack
->scrub_abort(on_finish
);
2827 void MDSRank::command_scrub_pause(Formatter
*f
, Context
*on_finish
) {
2828 std::lock_guard
l(mds_lock
);
2829 scrubstack
->scrub_pause(on_finish
);
2832 void MDSRank::command_scrub_resume(Formatter
*f
) {
2833 int r
= scrubstack
->scrub_resume();
2835 f
->open_object_section("result");
2836 f
->dump_int("return_code", r
);
2840 void MDSRank::command_scrub_status(Formatter
*f
) {
2841 scrubstack
->scrub_status(f
);
2844 void MDSRank::command_flush_path(Formatter
*f
, std::string_view path
)
2848 std::lock_guard
l(mds_lock
);
2849 mdcache
->flush_dentry(path
, &scond
);
2851 int r
= scond
.wait();
2852 f
->open_object_section("results");
2853 f
->dump_int("return_code", r
);
2854 f
->close_section(); // results
2857 // synchronous wrapper around "journal flush" asynchronous context
2859 void MDSRank::command_flush_journal(Formatter
*f
) {
2860 ceph_assert(f
!= NULL
);
2863 std::stringstream ss
;
2865 std::lock_guard
locker(mds_lock
);
2866 C_Flush_Journal
*flush_journal
= new C_Flush_Journal(mdcache
, mdlog
, this, &ss
, &cond
);
2867 flush_journal
->send();
2869 int r
= cond
.wait();
2871 f
->open_object_section("result");
2872 f
->dump_string("message", ss
.str());
2873 f
->dump_int("return_code", r
);
2877 void MDSRank::command_get_subtrees(Formatter
*f
)
2879 ceph_assert(f
!= NULL
);
2880 std::lock_guard
l(mds_lock
);
2882 std::vector
<CDir
*> subtrees
;
2883 mdcache
->get_subtrees(subtrees
);
2885 f
->open_array_section("subtrees");
2886 for (const auto& dir
: subtrees
) {
2887 f
->open_object_section("subtree");
2889 f
->dump_bool("is_auth", dir
->is_auth());
2890 f
->dump_int("auth_first", dir
->get_dir_auth().first
);
2891 f
->dump_int("auth_second", dir
->get_dir_auth().second
);
2892 f
->dump_int("export_pin", dir
->inode
->get_export_pin());
2893 f
->open_object_section("dir");
2903 void MDSRank::command_export_dir(Formatter
*f
,
2904 std::string_view path
,
2907 int r
= _command_export_dir(path
, target
);
2908 f
->open_object_section("results");
2909 f
->dump_int("return_code", r
);
2910 f
->close_section(); // results
2913 int MDSRank::_command_export_dir(
2914 std::string_view path
,
2917 std::lock_guard
l(mds_lock
);
2920 if (target
== whoami
|| !mdsmap
->is_up(target
) || !mdsmap
->is_in(target
)) {
2921 derr
<< "bad MDS target " << target
<< dendl
;
2925 CInode
*in
= mdcache
->cache_traverse(fp
);
2927 derr
<< "Bath path '" << path
<< "'" << dendl
;
2930 CDir
*dir
= in
->get_dirfrag(frag_t());
2931 if (!dir
|| !(dir
->is_auth())) {
2932 derr
<< "bad export_dir path dirfrag frag_t() or dir not auth" << dendl
;
2936 mdcache
->migrator
->export_dir(dir
, target
);
2940 void MDSRank::command_dump_tree(const cmdmap_t
&cmdmap
, std::ostream
&ss
, Formatter
*f
)
2944 cmd_getval(g_ceph_context
, cmdmap
, "root", root
);
2945 if (!cmd_getval(g_ceph_context
, cmdmap
, "depth", depth
))
2947 std::lock_guard
l(mds_lock
);
2948 CInode
*in
= mdcache
->cache_traverse(filepath(root
.c_str()));
2950 ss
<< "root inode is not in cache";
2953 f
->open_array_section("inodes");
2954 mdcache
->dump_tree(in
, 0, depth
, f
);
2958 CDir
*MDSRank::_command_dirfrag_get(
2959 const cmdmap_t
&cmdmap
,
2963 bool got
= cmd_getval(g_ceph_context
, cmdmap
, "path", path
);
2965 ss
<< "missing path argument";
2969 std::string frag_str
;
2970 if (!cmd_getval(g_ceph_context
, cmdmap
, "frag", frag_str
)) {
2971 ss
<< "missing frag argument";
2975 CInode
*in
= mdcache
->cache_traverse(filepath(path
.c_str()));
2977 // TODO really we should load something in if it's not in cache,
2978 // but the infrastructure is harder, and we might still be unable
2979 // to act on it if someone else is auth.
2980 ss
<< "directory '" << path
<< "' inode not in cache";
2986 if (!fg
.parse(frag_str
.c_str())) {
2987 ss
<< "frag " << frag_str
<< " failed to parse";
2991 CDir
*dir
= in
->get_dirfrag(fg
);
2993 ss
<< "frag " << in
->ino() << "/" << fg
<< " not in cache ("
2994 "use `dirfrag ls` to see if it should exist)";
2998 if (!dir
->is_auth()) {
2999 ss
<< "frag " << dir
->dirfrag() << " not auth (auth = "
3000 << dir
->authority() << ")";
3007 bool MDSRank::command_dirfrag_split(
3011 std::lock_guard
l(mds_lock
);
3013 if (!cmd_getval(g_ceph_context
, cmdmap
, "bits", by
)) {
3014 ss
<< "missing bits argument";
3019 ss
<< "must split by >0 bits";
3023 CDir
*dir
= _command_dirfrag_get(cmdmap
, ss
);
3028 mdcache
->split_dir(dir
, by
);
3033 bool MDSRank::command_dirfrag_merge(
3037 std::lock_guard
l(mds_lock
);
3039 bool got
= cmd_getval(g_ceph_context
, cmdmap
, "path", path
);
3041 ss
<< "missing path argument";
3045 std::string frag_str
;
3046 if (!cmd_getval(g_ceph_context
, cmdmap
, "frag", frag_str
)) {
3047 ss
<< "missing frag argument";
3051 CInode
*in
= mdcache
->cache_traverse(filepath(path
.c_str()));
3053 ss
<< "directory '" << path
<< "' inode not in cache";
3058 if (!fg
.parse(frag_str
.c_str())) {
3059 ss
<< "frag " << frag_str
<< " failed to parse";
3063 mdcache
->merge_dir(in
, fg
);
3068 bool MDSRank::command_dirfrag_ls(
3073 std::lock_guard
l(mds_lock
);
3075 bool got
= cmd_getval(g_ceph_context
, cmdmap
, "path", path
);
3077 ss
<< "missing path argument";
3081 CInode
*in
= mdcache
->cache_traverse(filepath(path
.c_str()));
3083 ss
<< "directory inode not in cache";
3087 f
->open_array_section("frags");
3089 // NB using get_leaves_under instead of get_dirfrags to give
3090 // you the list of what dirfrags may exist, not which are in cache
3091 in
->dirfragtree
.get_leaves_under(frag_t(), leaves
);
3092 for (const auto& leaf
: leaves
) {
3093 f
->open_object_section("frag");
3094 f
->dump_int("value", leaf
.value());
3095 f
->dump_int("bits", leaf
.bits());
3096 CachedStackStringStream css
;
3097 *css
<< std::hex
<< leaf
.value() << "/" << std::dec
<< leaf
.bits();
3098 f
->dump_string("str", css
->strv());
3106 void MDSRank::command_openfiles_ls(Formatter
*f
)
3108 std::lock_guard
l(mds_lock
);
3109 mdcache
->dump_openfiles(f
);
3112 void MDSRank::command_dump_inode(Formatter
*f
, const cmdmap_t
&cmdmap
, std::ostream
&ss
)
3114 std::lock_guard
l(mds_lock
);
3116 bool got
= cmd_getval(g_ceph_context
, cmdmap
, "number", number
);
3118 ss
<< "missing inode number";
3122 bool success
= mdcache
->dump_inode(f
, number
);
3124 ss
<< "dump inode failed, wrong inode number or the inode is not cached";
3128 void MDSRank::dump_status(Formatter
*f
) const
3130 if (state
== MDSMap::STATE_REPLAY
||
3131 state
== MDSMap::STATE_STANDBY_REPLAY
) {
3132 mdlog
->dump_replay_status(f
);
3133 } else if (state
== MDSMap::STATE_RESOLVE
) {
3134 mdcache
->dump_resolve_status(f
);
3135 } else if (state
== MDSMap::STATE_RECONNECT
) {
3136 server
->dump_reconnect_status(f
);
3137 } else if (state
== MDSMap::STATE_REJOIN
) {
3138 mdcache
->dump_rejoin_status(f
);
3139 } else if (state
== MDSMap::STATE_CLIENTREPLAY
) {
3140 dump_clientreplay_status(f
);
3142 f
->dump_float("rank_uptime", get_uptime().count());
3145 void MDSRank::dump_clientreplay_status(Formatter
*f
) const
3147 f
->open_object_section("clientreplay_status");
3148 f
->dump_unsigned("clientreplay_queue", replay_queue
.size());
3149 f
->dump_unsigned("active_replay", mdcache
->get_num_client_requests());
3153 void MDSRankDispatcher::update_log_config()
3155 map
<string
,string
> log_to_monitors
;
3156 map
<string
,string
> log_to_syslog
;
3157 map
<string
,string
> log_channel
;
3158 map
<string
,string
> log_prio
;
3159 map
<string
,string
> log_to_graylog
;
3160 map
<string
,string
> log_to_graylog_host
;
3161 map
<string
,string
> log_to_graylog_port
;
3165 if (parse_log_client_options(g_ceph_context
, log_to_monitors
, log_to_syslog
,
3166 log_channel
, log_prio
, log_to_graylog
,
3167 log_to_graylog_host
, log_to_graylog_port
,
3169 clog
->update_config(log_to_monitors
, log_to_syslog
,
3170 log_channel
, log_prio
, log_to_graylog
,
3171 log_to_graylog_host
, log_to_graylog_port
,
3173 dout(10) << __func__
<< " log_to_monitors " << log_to_monitors
<< dendl
;
3176 void MDSRank::create_logger()
3178 dout(10) << "create_logger" << dendl
;
3180 PerfCountersBuilder
mds_plb(g_ceph_context
, "mds", l_mds_first
, l_mds_last
);
3182 // super useful (high prio) perf stats
3183 mds_plb
.add_u64_counter(l_mds_request
, "request", "Requests", "req",
3184 PerfCountersBuilder::PRIO_CRITICAL
);
3185 mds_plb
.add_time_avg(l_mds_reply_latency
, "reply_latency", "Reply latency", "rlat",
3186 PerfCountersBuilder::PRIO_CRITICAL
);
3187 mds_plb
.add_u64(l_mds_inodes
, "inodes", "Inodes", "inos",
3188 PerfCountersBuilder::PRIO_CRITICAL
);
3189 mds_plb
.add_u64_counter(l_mds_forward
, "forward", "Forwarding request", "fwd",
3190 PerfCountersBuilder::PRIO_INTERESTING
);
3191 mds_plb
.add_u64(l_mds_caps
, "caps", "Capabilities", "caps",
3192 PerfCountersBuilder::PRIO_INTERESTING
);
3193 mds_plb
.add_u64_counter(l_mds_exported_inodes
, "exported_inodes", "Exported inodes",
3194 "exi", PerfCountersBuilder::PRIO_INTERESTING
);
3195 mds_plb
.add_u64_counter(l_mds_imported_inodes
, "imported_inodes", "Imported inodes",
3196 "imi", PerfCountersBuilder::PRIO_INTERESTING
);
3198 // useful dir/inode/subtree stats
3199 mds_plb
.set_prio_default(PerfCountersBuilder::PRIO_USEFUL
);
3200 mds_plb
.add_u64(l_mds_root_rfiles
, "root_rfiles", "root inode rfiles");
3201 mds_plb
.add_u64(l_mds_root_rbytes
, "root_rbytes", "root inode rbytes");
3202 mds_plb
.add_u64(l_mds_root_rsnaps
, "root_rsnaps", "root inode rsnaps");
3203 mds_plb
.add_u64_counter(l_mds_dir_fetch
, "dir_fetch", "Directory fetch");
3204 mds_plb
.add_u64_counter(l_mds_dir_commit
, "dir_commit", "Directory commit");
3205 mds_plb
.add_u64_counter(l_mds_dir_split
, "dir_split", "Directory split");
3206 mds_plb
.add_u64_counter(l_mds_dir_merge
, "dir_merge", "Directory merge");
3207 mds_plb
.add_u64(l_mds_inode_max
, "inode_max", "Max inodes, cache size");
3208 mds_plb
.add_u64(l_mds_inodes_pinned
, "inodes_pinned", "Inodes pinned");
3209 mds_plb
.add_u64(l_mds_inodes_expired
, "inodes_expired", "Inodes expired");
3210 mds_plb
.add_u64(l_mds_inodes_with_caps
, "inodes_with_caps",
3211 "Inodes with capabilities");
3212 mds_plb
.add_u64(l_mds_subtrees
, "subtrees", "Subtrees");
3213 mds_plb
.add_u64(l_mds_load_cent
, "load_cent", "Load per cent");
3214 mds_plb
.add_u64_counter(l_mds_openino_dir_fetch
, "openino_dir_fetch",
3215 "OpenIno incomplete directory fetchings");
3218 mds_plb
.set_prio_default(PerfCountersBuilder::PRIO_DEBUGONLY
);
3219 mds_plb
.add_u64_counter(l_mds_reply
, "reply", "Replies");
3220 mds_plb
.add_u64(l_mds_inodes_top
, "inodes_top", "Inodes on top");
3221 mds_plb
.add_u64(l_mds_inodes_bottom
, "inodes_bottom", "Inodes on bottom");
3223 l_mds_inodes_pin_tail
, "inodes_pin_tail", "Inodes on pin tail");
3224 mds_plb
.add_u64_counter(l_mds_traverse
, "traverse", "Traverses");
3225 mds_plb
.add_u64_counter(l_mds_traverse_hit
, "traverse_hit", "Traverse hits");
3226 mds_plb
.add_u64_counter(l_mds_traverse_forward
, "traverse_forward",
3227 "Traverse forwards");
3228 mds_plb
.add_u64_counter(l_mds_traverse_discover
, "traverse_discover",
3229 "Traverse directory discovers");
3230 mds_plb
.add_u64_counter(l_mds_traverse_dir_fetch
, "traverse_dir_fetch",
3231 "Traverse incomplete directory content fetchings");
3232 mds_plb
.add_u64_counter(l_mds_traverse_remote_ino
, "traverse_remote_ino",
3233 "Traverse remote dentries");
3234 mds_plb
.add_u64_counter(l_mds_traverse_lock
, "traverse_lock",
3236 mds_plb
.add_u64(l_mds_dispatch_queue_len
, "q", "Dispatch queue length");
3237 mds_plb
.add_u64_counter(l_mds_exported
, "exported", "Exports");
3238 mds_plb
.add_u64_counter(l_mds_imported
, "imported", "Imports");
3239 mds_plb
.add_u64_counter(l_mds_openino_backtrace_fetch
, "openino_backtrace_fetch",
3240 "OpenIno backtrace fetchings");
3241 mds_plb
.add_u64_counter(l_mds_openino_peer_discover
, "openino_peer_discover",
3242 "OpenIno peer inode discovers");
3244 logger
= mds_plb
.create_perf_counters();
3245 g_ceph_context
->get_perfcounters_collection()->add(logger
);
3249 PerfCountersBuilder
mdm_plb(g_ceph_context
, "mds_mem", l_mdm_first
, l_mdm_last
);
3250 mdm_plb
.add_u64(l_mdm_ino
, "ino", "Inodes", "ino",
3251 PerfCountersBuilder::PRIO_INTERESTING
);
3252 mdm_plb
.add_u64(l_mdm_dn
, "dn", "Dentries", "dn",
3253 PerfCountersBuilder::PRIO_INTERESTING
);
3255 mdm_plb
.set_prio_default(PerfCountersBuilder::PRIO_USEFUL
);
3256 mdm_plb
.add_u64_counter(l_mdm_inoa
, "ino+", "Inodes opened");
3257 mdm_plb
.add_u64_counter(l_mdm_inos
, "ino-", "Inodes closed");
3258 mdm_plb
.add_u64(l_mdm_dir
, "dir", "Directories");
3259 mdm_plb
.add_u64_counter(l_mdm_dira
, "dir+", "Directories opened");
3260 mdm_plb
.add_u64_counter(l_mdm_dirs
, "dir-", "Directories closed");
3261 mdm_plb
.add_u64_counter(l_mdm_dna
, "dn+", "Dentries opened");
3262 mdm_plb
.add_u64_counter(l_mdm_dns
, "dn-", "Dentries closed");
3263 mdm_plb
.add_u64(l_mdm_cap
, "cap", "Capabilities");
3264 mdm_plb
.add_u64_counter(l_mdm_capa
, "cap+", "Capabilities added");
3265 mdm_plb
.add_u64_counter(l_mdm_caps
, "cap-", "Capabilities removed");
3266 mdm_plb
.add_u64(l_mdm_heap
, "heap", "Heap size");
3268 mdm_plb
.set_prio_default(PerfCountersBuilder::PRIO_DEBUGONLY
);
3269 mdm_plb
.add_u64(l_mdm_rss
, "rss", "RSS");
3271 mlogger
= mdm_plb
.create_perf_counters();
3272 g_ceph_context
->get_perfcounters_collection()->add(mlogger
);
3275 mdlog
->create_logger();
3276 server
->create_logger();
3277 purge_queue
.create_logger();
3278 sessionmap
.register_perfcounters();
3279 mdcache
->register_perfcounters();
3282 void MDSRank::check_ops_in_flight()
3285 vector
<string
> warnings
;
3287 if (op_tracker
.check_ops_in_flight(&summary
, warnings
, &slow
)) {
3288 clog
->warn() << summary
;
3289 for (const auto& warning
: warnings
) {
3290 clog
->warn() << warning
;
3294 // set mds slow request count
3295 mds_slow_req_count
= slow
;
3299 void MDSRankDispatcher::handle_osd_map()
3302 mdsmap
->get_tableserver() == whoami
) {
3303 snapserver
->check_osd_map(true);
3306 server
->handle_osd_map();
3308 purge_queue
.update_op_limit(*mdsmap
);
3310 std::set
<entity_addr_t
> newly_blacklisted
;
3311 objecter
->consume_blacklist_events(&newly_blacklisted
);
3312 auto epoch
= objecter
->with_osdmap([](const OSDMap
&o
){return o
.get_epoch();});
3313 dout(4) << "handle_osd_map epoch " << epoch
<< ", "
3314 << newly_blacklisted
.size() << " new blacklist entries" << dendl
;
3315 auto victims
= server
->apply_blacklist(newly_blacklisted
);
3317 set_osd_epoch_barrier(epoch
);
3321 // By default the objecter only requests OSDMap updates on use,
3322 // we would like to always receive the latest maps in order to
3323 // apply policy based on the FULL flag.
3324 objecter
->maybe_request_map();
3327 bool MDSRank::evict_client(int64_t session_id
,
3328 bool wait
, bool blacklist
, std::ostream
& err_ss
,
3331 ceph_assert(mds_lock
.is_locked_by_me());
3333 // Mutually exclusive args
3334 ceph_assert(!(wait
&& on_killed
!= nullptr));
3336 if (is_any_replay()) {
3337 err_ss
<< "MDS is replaying log";
3341 Session
*session
= sessionmap
.get_session(
3342 entity_name_t(CEPH_ENTITY_TYPE_CLIENT
, session_id
));
3344 err_ss
<< "session " << session_id
<< " not in sessionmap!";
3348 auto& addr
= session
->info
.inst
.addr
;
3350 CachedStackStringStream css
;
3351 *css
<< "Evicting " << (blacklist
? "(and blacklisting) " : "")
3352 << "client session " << session_id
<< " (" << addr
<< ")";
3353 dout(1) << css
->strv() << dendl
;
3354 clog
->info() << css
->strv();
3357 dout(4) << "Preparing blacklist command... (wait=" << wait
<< ")" << dendl
;
3359 ss
<< "{\"prefix\":\"osd blacklist\", \"blacklistop\":\"add\",";
3360 ss
<< "\"addr\":\"";
3363 std::string tmp
= ss
.str();
3364 std::vector
<std::string
> cmd
= {tmp
};
3366 auto kill_client_session
= [this, session_id
, wait
, on_killed
](){
3367 ceph_assert(mds_lock
.is_locked_by_me());
3368 Session
*session
= sessionmap
.get_session(
3369 entity_name_t(CEPH_ENTITY_TYPE_CLIENT
, session_id
));
3371 if (on_killed
|| !wait
) {
3372 server
->kill_session(session
, on_killed
);
3374 C_SaferCond on_safe
;
3375 server
->kill_session(session
, &on_safe
);
3382 dout(1) << "session " << session_id
<< " was removed while we waited "
3383 "for blacklist" << dendl
;
3385 // Even though it wasn't us that removed it, kick our completion
3386 // as the session has been removed.
3388 on_killed
->complete(0);
3393 auto apply_blacklist
= [this, cmd
](std::function
<void ()> fn
){
3394 ceph_assert(mds_lock
.is_locked_by_me());
3396 Context
*on_blacklist_done
= new FunctionContext([this, fn
](int r
) {
3397 objecter
->wait_for_latest_osdmap(
3399 new FunctionContext([this, fn
](int r
) {
3400 std::lock_guard
l(mds_lock
);
3401 auto epoch
= objecter
->with_osdmap([](const OSDMap
&o
){
3402 return o
.get_epoch();
3405 set_osd_epoch_barrier(epoch
);
3412 dout(4) << "Sending mon blacklist command: " << cmd
[0] << dendl
;
3413 monc
->start_mon_command(cmd
, {}, nullptr, nullptr, on_blacklist_done
);
3418 C_SaferCond inline_ctx
;
3419 apply_blacklist([&inline_ctx
](){inline_ctx
.complete(0);});
3425 // We dropped mds_lock, so check that session still exists
3426 session
= sessionmap
.get_session(entity_name_t(CEPH_ENTITY_TYPE_CLIENT
,
3429 dout(1) << "session " << session_id
<< " was removed while we waited "
3430 "for blacklist" << dendl
;
3433 kill_client_session();
3436 apply_blacklist(kill_client_session
);
3438 kill_client_session();
3445 void MDSRank::bcast_mds_map()
3447 dout(7) << "bcast_mds_map " << mdsmap
->get_epoch() << dendl
;
3449 // share the map with mounted clients
3450 set
<Session
*> clients
;
3451 sessionmap
.get_client_session_set(clients
);
3452 for (const auto &session
: clients
) {
3453 auto m
= MMDSMap::create(monc
->get_fsid(), *mdsmap
);
3454 session
->get_connection()->send_message2(std::move(m
));
3456 last_client_mdsmap_bcast
= mdsmap
->get_epoch();
3459 Context
*MDSRank::create_async_exec_context(C_ExecAndReply
*ctx
) {
3460 return new C_OnFinisher(new FunctionContext([ctx
](int _
) {
3465 MDSRankDispatcher::MDSRankDispatcher(
3468 LogChannelRef
&clog_
,
3471 std::unique_ptr
<MDSMap
> &mdsmap_
,
3474 Context
*respawn_hook_
,
3475 Context
*suicide_hook_
)
3476 : MDSRank(whoami_
, mds_lock_
, clog_
, timer_
, beacon_
, mdsmap_
,
3477 msgr
, monc_
, respawn_hook_
, suicide_hook_
)
3480 bool MDSRankDispatcher::handle_command(
3481 const cmdmap_t
&cmdmap
,
3482 const MCommand::const_ref
&m
,
3484 std::stringstream
*ds
,
3485 std::stringstream
*ss
,
3486 Context
**run_later
,
3489 ceph_assert(r
!= nullptr);
3490 ceph_assert(ds
!= nullptr);
3491 ceph_assert(ss
!= nullptr);
3496 cmd_getval(g_ceph_context
, cmdmap
, "prefix", prefix
);
3498 if (prefix
== "session ls" || prefix
== "client ls") {
3499 std::vector
<std::string
> filter_args
;
3500 cmd_getval(g_ceph_context
, cmdmap
, "filters", filter_args
);
3502 SessionFilter filter
;
3503 *r
= filter
.parse(filter_args
, ss
);
3508 JSONFormatter
f(true);
3509 dump_sessions(filter
, &f
);
3512 } else if (prefix
== "session evict" || prefix
== "client evict") {
3513 std::vector
<std::string
> filter_args
;
3514 cmd_getval(g_ceph_context
, cmdmap
, "filters", filter_args
);
3516 SessionFilter filter
;
3517 *r
= filter
.parse(filter_args
, ss
);
3522 evict_clients(filter
, m
);
3524 *need_reply
= false;
3526 } else if (prefix
== "damage ls") {
3527 JSONFormatter
f(true);
3528 damage_table
.dump(&f
);
3531 } else if (prefix
== "damage rm") {
3532 damage_entry_id_t id
= 0;
3533 bool got
= cmd_getval(g_ceph_context
, cmdmap
, "damage_id", (int64_t&)id
);
3539 damage_table
.erase(id
);
3541 } else if (prefix
== "cache drop") {
3543 if (!cmd_getval(g_ceph_context
, cmdmap
, "timeout", timeout
)) {
3547 *need_reply
= false;
3548 *run_later
= create_async_exec_context(new C_CacheDropExecAndReply
3549 (this, m
, (uint64_t)timeout
));
3551 } else if (prefix
== "scrub start") {
3554 vector
<string
> scrubop_vec
;
3555 cmd_getval(g_ceph_context
, cmdmap
, "scrubops", scrubop_vec
);
3556 cmd_getval(g_ceph_context
, cmdmap
, "path", path
);
3557 cmd_getval(g_ceph_context
, cmdmap
, "tag", tag
);
3559 *need_reply
= false;
3560 *run_later
= create_async_exec_context(new C_ScrubExecAndReply
3561 (this, m
, path
, tag
, scrubop_vec
));
3563 } else if (prefix
== "scrub abort") {
3564 *need_reply
= false;
3565 *run_later
= create_async_exec_context(new C_ScrubControlExecAndReply
3566 (this, m
, "abort"));
3568 } else if (prefix
== "scrub pause") {
3569 *need_reply
= false;
3570 *run_later
= create_async_exec_context(new C_ScrubControlExecAndReply
3571 (this, m
, "pause"));
3573 } else if (prefix
== "scrub resume") {
3574 JSONFormatter
f(true);
3575 command_scrub_resume(&f
);
3578 } else if (prefix
== "scrub status") {
3579 JSONFormatter
f(true);
3580 command_scrub_status(&f
);
3588 void MDSRank::command_cache_drop(uint64_t timeout
, Formatter
*f
, Context
*on_finish
) {
3589 dout(20) << __func__
<< dendl
;
3591 std::lock_guard
locker(mds_lock
);
3592 C_Drop_Cache
*request
= new C_Drop_Cache(server
, mdcache
, mdlog
, this,
3593 timeout
, f
, on_finish
);
3597 epoch_t
MDSRank::get_osd_epoch() const
3599 return objecter
->with_osdmap(std::mem_fn(&OSDMap::get_epoch
));