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) 2016 John Spray <john.spray@redhat.com>
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.
14 #include "DaemonServer.h"
17 #include "include/stringify.h"
18 #include "include/str_list.h"
19 #include "auth/RotatingKeyRing.h"
20 #include "json_spirit/json_spirit_writer.h"
22 #include "mgr/mgr_commands.h"
23 #include "mgr/OSDHealthMetricCollector.h"
24 #include "mon/MonCommand.h"
26 #include "messages/MMgrOpen.h"
27 #include "messages/MMgrConfigure.h"
28 #include "messages/MMonMgrReport.h"
29 #include "messages/MCommand.h"
30 #include "messages/MCommandReply.h"
31 #include "messages/MPGStats.h"
32 #include "messages/MOSDScrub.h"
33 #include "messages/MOSDForceRecovery.h"
34 #include "common/errno.h"
36 #define dout_context g_ceph_context
37 #define dout_subsys ceph_subsys_mgr
39 #define dout_prefix *_dout << "mgr.server " << __func__ << " "
43 DaemonServer::DaemonServer(MonClient
*monc_
,
45 DaemonStateIndex
&daemon_state_
,
46 ClusterState
&cluster_state_
,
47 PyModuleRegistry
&py_modules_
,
49 LogChannelRef audit_clog_
)
50 : Dispatcher(g_ceph_context
),
51 client_byte_throttler(new Throttle(g_ceph_context
, "mgr_client_bytes",
52 g_conf
->get_val
<uint64_t>("mgr_client_bytes"))),
53 client_msg_throttler(new Throttle(g_ceph_context
, "mgr_client_messages",
54 g_conf
->get_val
<uint64_t>("mgr_client_messages"))),
55 osd_byte_throttler(new Throttle(g_ceph_context
, "mgr_osd_bytes",
56 g_conf
->get_val
<uint64_t>("mgr_osd_bytes"))),
57 osd_msg_throttler(new Throttle(g_ceph_context
, "mgr_osd_messsages",
58 g_conf
->get_val
<uint64_t>("mgr_osd_messages"))),
59 mds_byte_throttler(new Throttle(g_ceph_context
, "mgr_mds_bytes",
60 g_conf
->get_val
<uint64_t>("mgr_mds_bytes"))),
61 mds_msg_throttler(new Throttle(g_ceph_context
, "mgr_mds_messsages",
62 g_conf
->get_val
<uint64_t>("mgr_mds_messages"))),
63 mon_byte_throttler(new Throttle(g_ceph_context
, "mgr_mon_bytes",
64 g_conf
->get_val
<uint64_t>("mgr_mon_bytes"))),
65 mon_msg_throttler(new Throttle(g_ceph_context
, "mgr_mon_messsages",
66 g_conf
->get_val
<uint64_t>("mgr_mon_messages"))),
70 daemon_state(daemon_state_
),
71 cluster_state(cluster_state_
),
72 py_modules(py_modules_
),
74 audit_clog(audit_clog_
),
75 auth_cluster_registry(g_ceph_context
,
76 g_conf
->auth_supported
.empty() ?
77 g_conf
->auth_cluster_required
:
78 g_conf
->auth_supported
),
79 auth_service_registry(g_ceph_context
,
80 g_conf
->auth_supported
.empty() ?
81 g_conf
->auth_service_required
:
82 g_conf
->auth_supported
),
86 g_conf
->add_observer(this);
89 DaemonServer::~DaemonServer() {
91 g_conf
->remove_observer(this);
94 int DaemonServer::init(uint64_t gid
, entity_addr_t client_addr
)
96 // Initialize Messenger
97 std::string public_msgr_type
= g_conf
->ms_public_type
.empty() ?
98 g_conf
->get_val
<std::string
>("ms_type") : g_conf
->ms_public_type
;
99 msgr
= Messenger::create(g_ceph_context
, public_msgr_type
,
100 entity_name_t::MGR(gid
),
103 msgr
->set_default_policy(Messenger::Policy::stateless_server(0));
106 msgr
->set_policy_throttlers(entity_name_t::TYPE_CLIENT
,
107 client_byte_throttler
.get(),
108 client_msg_throttler
.get());
111 msgr
->set_policy_throttlers(entity_name_t::TYPE_OSD
,
112 osd_byte_throttler
.get(),
113 osd_msg_throttler
.get());
114 msgr
->set_policy_throttlers(entity_name_t::TYPE_MDS
,
115 mds_byte_throttler
.get(),
116 mds_msg_throttler
.get());
117 msgr
->set_policy_throttlers(entity_name_t::TYPE_MON
,
118 mon_byte_throttler
.get(),
119 mon_msg_throttler
.get());
121 int r
= msgr
->bind(g_conf
->public_addr
);
123 derr
<< "unable to bind mgr to " << g_conf
->public_addr
<< dendl
;
127 msgr
->set_myname(entity_name_t::MGR(gid
));
128 msgr
->set_addr_unknowns(client_addr
);
131 msgr
->add_dispatcher_tail(this);
133 started_at
= ceph_clock_now();
138 entity_addr_t
DaemonServer::get_myaddr() const
140 return msgr
->get_myaddr();
144 bool DaemonServer::ms_verify_authorizer(Connection
*con
,
147 ceph::bufferlist
& authorizer_data
,
148 ceph::bufferlist
& authorizer_reply
,
150 CryptoKey
& session_key
)
152 AuthAuthorizeHandler
*handler
= nullptr;
153 if (peer_type
== CEPH_ENTITY_TYPE_OSD
||
154 peer_type
== CEPH_ENTITY_TYPE_MON
||
155 peer_type
== CEPH_ENTITY_TYPE_MDS
||
156 peer_type
== CEPH_ENTITY_TYPE_MGR
) {
157 handler
= auth_cluster_registry
.get_handler(protocol
);
159 handler
= auth_service_registry
.get_handler(protocol
);
162 dout(0) << "No AuthAuthorizeHandler found for protocol " << protocol
<< dendl
;
167 MgrSessionRef
s(new MgrSession(cct
));
168 s
->inst
.addr
= con
->get_peer_addr();
169 AuthCapsInfo caps_info
;
171 RotatingKeyRing
*keys
= monc
->rotating_secrets
.get();
173 is_valid
= handler
->verify_authorizer(
176 authorizer_reply
, s
->entity_name
,
177 s
->global_id
, caps_info
,
180 dout(10) << __func__
<< " no rotating_keys (yet), denied" << dendl
;
185 if (caps_info
.allow_all
) {
186 dout(10) << " session " << s
<< " " << s
->entity_name
187 << " allow_all" << dendl
;
188 s
->caps
.set_allow_all();
190 if (caps_info
.caps
.length() > 0) {
191 bufferlist::iterator p
= caps_info
.caps
.begin();
196 catch (buffer::error
& e
) {
198 bool success
= s
->caps
.parse(str
);
200 dout(10) << " session " << s
<< " " << s
->entity_name
201 << " has caps " << s
->caps
<< " '" << str
<< "'" << dendl
;
203 dout(10) << " session " << s
<< " " << s
->entity_name
204 << " failed to parse caps '" << str
<< "'" << dendl
;
208 con
->set_priv(s
->get());
210 if (peer_type
== CEPH_ENTITY_TYPE_OSD
) {
211 Mutex::Locker
l(lock
);
212 s
->osd_id
= atoi(s
->entity_name
.get_id().c_str());
213 dout(10) << "registering osd." << s
->osd_id
<< " session "
214 << s
<< " con " << con
<< dendl
;
215 osd_cons
[s
->osd_id
].insert(con
);
223 bool DaemonServer::ms_get_authorizer(int dest_type
,
224 AuthAuthorizer
**authorizer
, bool force_new
)
226 dout(10) << "type=" << ceph_entity_type_name(dest_type
) << dendl
;
228 if (dest_type
== CEPH_ENTITY_TYPE_MON
) {
233 if (monc
->wait_auth_rotating(10) < 0)
237 *authorizer
= monc
->build_authorizer(dest_type
);
238 dout(20) << "got authorizer " << *authorizer
<< dendl
;
239 return *authorizer
!= NULL
;
242 bool DaemonServer::ms_handle_reset(Connection
*con
)
244 if (con
->get_peer_type() == CEPH_ENTITY_TYPE_OSD
) {
245 MgrSessionRef
session(static_cast<MgrSession
*>(con
->get_priv()));
249 session
->put(); // SessionRef takes a ref
250 Mutex::Locker
l(lock
);
251 dout(10) << "unregistering osd." << session
->osd_id
252 << " session " << session
<< " con " << con
<< dendl
;
253 osd_cons
[session
->osd_id
].erase(con
);
255 auto iter
= daemon_connections
.find(con
);
256 if (iter
!= daemon_connections
.end()) {
257 daemon_connections
.erase(iter
);
263 bool DaemonServer::ms_handle_refused(Connection
*con
)
265 // do nothing for now
269 bool DaemonServer::ms_dispatch(Message
*m
)
271 // Note that we do *not* take ::lock here, in order to avoid
272 // serializing all message handling. It's up to each handler
273 // to take whatever locks it needs.
274 switch (m
->get_type()) {
276 cluster_state
.ingest_pgstats(static_cast<MPGStats
*>(m
));
277 maybe_ready(m
->get_source().num());
281 return handle_report(static_cast<MMgrReport
*>(m
));
283 return handle_open(static_cast<MMgrOpen
*>(m
));
285 return handle_command(static_cast<MCommand
*>(m
));
287 dout(1) << "Unhandled message type " << m
->get_type() << dendl
;
292 void DaemonServer::maybe_ready(int32_t osd_id
)
294 if (pgmap_ready
.load()) {
295 // Fast path: we don't need to take lock because pgmap_ready
298 Mutex::Locker
l(lock
);
300 if (reported_osds
.find(osd_id
) == reported_osds
.end()) {
301 dout(4) << "initial report from osd " << osd_id
<< dendl
;
302 reported_osds
.insert(osd_id
);
303 std::set
<int32_t> up_osds
;
305 cluster_state
.with_osdmap([&](const OSDMap
& osdmap
) {
306 osdmap
.get_up_osds(up_osds
);
309 std::set
<int32_t> unreported_osds
;
310 std::set_difference(up_osds
.begin(), up_osds
.end(),
311 reported_osds
.begin(), reported_osds
.end(),
312 std::inserter(unreported_osds
, unreported_osds
.begin()));
314 if (unreported_osds
.size() == 0) {
315 dout(4) << "all osds have reported, sending PG state to mon" << dendl
;
317 reported_osds
.clear();
318 // Avoid waiting for next tick
321 dout(4) << "still waiting for " << unreported_osds
.size() << " osds"
322 " to report in before PGMap is ready" << dendl
;
328 void DaemonServer::shutdown()
330 dout(10) << "begin" << dendl
;
333 dout(10) << "done" << dendl
;
338 bool DaemonServer::handle_open(MMgrOpen
*m
)
340 Mutex::Locker
l(lock
);
343 if (!m
->service_name
.empty()) {
344 key
.first
= m
->service_name
;
346 key
.first
= ceph_entity_type_name(m
->get_connection()->get_peer_type());
348 key
.second
= m
->daemon_name
;
350 dout(4) << "from " << m
->get_connection() << " " << key
<< dendl
;
352 _send_configure(m
->get_connection());
354 DaemonStatePtr daemon
;
355 if (daemon_state
.exists(key
)) {
356 daemon
= daemon_state
.get(key
);
359 dout(20) << "updating existing DaemonState for " << m
->daemon_name
<< dendl
;
360 Mutex::Locker
l(daemon
->lock
);
361 daemon
->perf_counters
.clear();
364 if (m
->service_daemon
) {
366 dout(4) << "constructing new DaemonState for " << key
<< dendl
;
367 daemon
= std::make_shared
<DaemonState
>(daemon_state
.types
);
369 if (m
->daemon_metadata
.count("hostname")) {
370 daemon
->hostname
= m
->daemon_metadata
["hostname"];
372 daemon_state
.insert(daemon
);
374 Mutex::Locker
l(daemon
->lock
);
375 daemon
->service_daemon
= true;
376 daemon
->metadata
= m
->daemon_metadata
;
377 daemon
->service_status
= m
->daemon_status
;
379 utime_t now
= ceph_clock_now();
380 auto d
= pending_service_map
.get_daemon(m
->service_name
,
382 if (d
->gid
!= (uint64_t)m
->get_source().num()) {
383 dout(10) << "registering " << key
<< " in pending_service_map" << dendl
;
384 d
->gid
= m
->get_source().num();
385 d
->addr
= m
->get_source_addr();
386 d
->start_epoch
= pending_service_map
.epoch
;
387 d
->start_stamp
= now
;
388 d
->metadata
= m
->daemon_metadata
;
389 pending_service_map_dirty
= pending_service_map
.epoch
;
393 if (m
->get_connection()->get_peer_type() != entity_name_t::TYPE_CLIENT
&&
394 m
->service_name
.empty())
396 // Store in set of the daemon/service connections, i.e. those
397 // connections that require an update in the event of stats
398 // configuration changes.
399 daemon_connections
.insert(m
->get_connection());
406 bool DaemonServer::handle_report(MMgrReport
*m
)
409 if (!m
->service_name
.empty()) {
410 key
.first
= m
->service_name
;
412 key
.first
= ceph_entity_type_name(m
->get_connection()->get_peer_type());
414 key
.second
= m
->daemon_name
;
416 dout(4) << "from " << m
->get_connection() << " " << key
<< dendl
;
418 if (m
->get_connection()->get_peer_type() == entity_name_t::TYPE_CLIENT
&&
419 m
->service_name
.empty()) {
420 // Clients should not be sending us stats unless they are declaring
421 // themselves to be a daemon for some service.
422 dout(4) << "rejecting report from non-daemon client " << m
->daemon_name
424 m
->get_connection()->mark_down();
429 // Look up the DaemonState
430 DaemonStatePtr daemon
;
431 if (daemon_state
.exists(key
)) {
432 dout(20) << "updating existing DaemonState for " << key
<< dendl
;
433 daemon
= daemon_state
.get(key
);
435 // we don't know the hostname at this stage, reject MMgrReport here.
436 dout(5) << "rejecting report from " << key
<< ", since we do not have its metadata now."
439 // issue metadata request in background
440 if (!daemon_state
.is_updating(key
) &&
441 (key
.first
== "osd" || key
.first
== "mds")) {
443 std::ostringstream oss
;
444 auto c
= new MetadataUpdate(daemon_state
, key
);
445 if (key
.first
== "osd") {
446 oss
<< "{\"prefix\": \"osd metadata\", \"id\": "
449 } else if (key
.first
== "mds") {
450 c
->set_default("addr", stringify(m
->get_source_addr()));
451 oss
<< "{\"prefix\": \"mds metadata\", \"who\": \""
452 << key
.second
<< "\"}";
458 monc
->start_mon_command({oss
.str()}, {}, &c
->outbl
, &c
->outs
, c
);
462 Mutex::Locker
l(lock
);
464 MgrSessionRef
session(static_cast<MgrSession
*>(m
->get_connection()->get_priv()));
468 m
->get_connection()->mark_down();
471 dout(10) << "unregistering osd." << session
->osd_id
472 << " session " << session
<< " con " << m
->get_connection() << dendl
;
474 if (osd_cons
.find(session
->osd_id
) != osd_cons
.end()) {
475 osd_cons
[session
->osd_id
].erase(m
->get_connection());
478 auto iter
= daemon_connections
.find(m
->get_connection());
479 if (iter
!= daemon_connections
.end()) {
480 daemon_connections
.erase(iter
);
487 // Update the DaemonState
488 assert(daemon
!= nullptr);
490 Mutex::Locker
l(daemon
->lock
);
491 auto &daemon_counters
= daemon
->perf_counters
;
492 daemon_counters
.update(m
);
494 if (daemon
->service_daemon
) {
495 utime_t now
= ceph_clock_now();
496 if (m
->daemon_status
) {
497 daemon
->service_status
= *m
->daemon_status
;
498 daemon
->service_status_stamp
= now
;
500 daemon
->last_service_beacon
= now
;
501 } else if (m
->daemon_status
) {
502 derr
<< "got status from non-daemon " << key
<< dendl
;
504 if (m
->get_connection()->peer_is_osd()) {
505 // only OSD sends health_checks to me now
506 daemon
->osd_health_metrics
= std::move(m
->osd_health_metrics
);
510 // if there are any schema updates, notify the python modules
511 if (!m
->declare_types
.empty() || !m
->undeclare_types
.empty()) {
513 oss
<< key
.first
<< '.' << key
.second
;
514 py_modules
.notify_all("perf_schema_update", oss
.str());
522 void DaemonServer::_generate_command_map(
523 map
<string
,cmd_vartype
>& cmdmap
,
524 map
<string
,string
> ¶m_str_map
)
526 for (map
<string
,cmd_vartype
>::const_iterator p
= cmdmap
.begin();
527 p
!= cmdmap
.end(); ++p
) {
528 if (p
->first
== "prefix")
530 if (p
->first
== "caps") {
532 if (cmd_getval(g_ceph_context
, cmdmap
, "caps", cv
) &&
533 cv
.size() % 2 == 0) {
534 for (unsigned i
= 0; i
< cv
.size(); i
+= 2) {
535 string k
= string("caps_") + cv
[i
];
536 param_str_map
[k
] = cv
[i
+ 1];
541 param_str_map
[p
->first
] = cmd_vartype_stringify(p
->second
);
545 const MonCommand
*DaemonServer::_get_mgrcommand(
546 const string
&cmd_prefix
,
547 const std::vector
<MonCommand
> &cmds
)
549 const MonCommand
*this_cmd
= nullptr;
550 for (const auto &cmd
: cmds
) {
551 if (cmd
.cmdstring
.compare(0, cmd_prefix
.size(), cmd_prefix
) == 0) {
559 bool DaemonServer::_allowed_command(
561 const string
&module
,
562 const string
&prefix
,
563 const map
<string
,cmd_vartype
>& cmdmap
,
564 const map
<string
,string
>& param_str_map
,
565 const MonCommand
*this_cmd
) {
567 if (s
->entity_name
.is_mon()) {
568 // mon is all-powerful. even when it is forwarding commands on behalf of
569 // old clients; we expect the mon is validating commands before proxying!
573 bool cmd_r
= this_cmd
->requires_perm('r');
574 bool cmd_w
= this_cmd
->requires_perm('w');
575 bool cmd_x
= this_cmd
->requires_perm('x');
577 bool capable
= s
->caps
.is_capable(
579 CEPH_ENTITY_TYPE_MGR
,
581 module
, prefix
, param_str_map
,
582 cmd_r
, cmd_w
, cmd_x
);
584 dout(10) << " " << s
->entity_name
<< " "
585 << (capable
? "" : "not ") << "capable" << dendl
;
589 bool DaemonServer::handle_command(MCommand
*m
)
591 Mutex::Locker
l(lock
);
593 std::stringstream ss
;
596 assert(lock
.is_locked_by_me());
599 * The working data for processing an MCommand. This lives in
600 * a class to enable passing it into other threads for processing
601 * outside of the thread/locks that called handle_command.
610 CommandContext(MCommand
*m_
)
620 void reply(int r
, const std::stringstream
&ss
)
625 void reply(int r
, const std::string
&rs
)
627 // Let the connection drop as soon as we've sent our response
628 ConnectionRef con
= m
->get_connection();
630 con
->mark_disposable();
633 dout(1) << "handle_command " << cpp_strerror(r
) << " " << rs
<< dendl
;
635 MCommandReply
*reply
= new MCommandReply(r
, rs
);
636 reply
->set_tid(m
->get_tid());
637 reply
->set_data(odata
);
638 con
->send_message(reply
);
644 * A context for receiving a bufferlist/error string from a background
645 * function and then calling back to a CommandContext when it's done
647 class ReplyOnFinish
: public Context
{
648 std::shared_ptr
<CommandContext
> cmdctx
;
654 ReplyOnFinish(std::shared_ptr
<CommandContext
> cmdctx_
)
657 void finish(int r
) override
{
658 cmdctx
->odata
.claim_append(from_mon
);
659 cmdctx
->reply(r
, outs
);
663 std::shared_ptr
<CommandContext
> cmdctx
= std::make_shared
<CommandContext
>(m
);
665 MgrSessionRef
session(static_cast<MgrSession
*>(m
->get_connection()->get_priv()));
669 session
->put(); // SessionRef takes a ref
670 if (session
->inst
.name
== entity_name_t())
671 session
->inst
.name
= m
->get_source();
674 boost::scoped_ptr
<Formatter
> f
;
675 map
<string
,string
> param_str_map
;
677 if (!cmdmap_from_json(m
->cmd
, &(cmdctx
->cmdmap
), ss
)) {
678 cmdctx
->reply(-EINVAL
, ss
);
683 cmd_getval(g_ceph_context
, cmdctx
->cmdmap
, "format", format
, string("plain"));
684 f
.reset(Formatter::create(format
));
687 cmd_getval(cct
, cmdctx
->cmdmap
, "prefix", prefix
);
689 dout(4) << "decoded " << cmdctx
->cmdmap
.size() << dendl
;
690 dout(4) << "prefix=" << prefix
<< dendl
;
692 if (prefix
== "get_command_descriptions") {
693 dout(10) << "reading commands from python modules" << dendl
;
694 const auto py_commands
= py_modules
.get_commands();
698 f
.open_object_section("command_descriptions");
700 auto dump_cmd
= [&cmdnum
, &f
](const MonCommand
&mc
){
701 ostringstream secname
;
702 secname
<< "cmd" << setfill('0') << std::setw(3) << cmdnum
;
703 dump_cmddesc_to_json(&f
, secname
.str(), mc
.cmdstring
, mc
.helpstring
,
704 mc
.module
, mc
.req_perms
, mc
.availability
, 0);
708 for (const auto &pyc
: py_commands
) {
712 for (const auto &mgr_cmd
: mgr_commands
) {
716 f
.close_section(); // command_descriptions
717 f
.flush(cmdctx
->odata
);
718 cmdctx
->reply(0, ss
);
723 const MonCommand
*mgr_cmd
= _get_mgrcommand(prefix
, mgr_commands
);
724 _generate_command_map(cmdctx
->cmdmap
, param_str_map
);
726 MonCommand py_command
= {"", "", "py", "rw", "cli"};
727 if (!_allowed_command(session
.get(), py_command
.module
, prefix
, cmdctx
->cmdmap
,
728 param_str_map
, &py_command
)) {
729 dout(1) << " access denied" << dendl
;
730 ss
<< "access denied; does your client key have mgr caps?"
731 " See http://docs.ceph.com/docs/master/mgr/administrator/#client-authentication";
732 cmdctx
->reply(-EACCES
, ss
);
736 // validate user's permissions for requested command
737 if (!_allowed_command(session
.get(), mgr_cmd
->module
, prefix
, cmdctx
->cmdmap
,
738 param_str_map
, mgr_cmd
)) {
739 dout(1) << " access denied" << dendl
;
740 audit_clog
->info() << "from='" << session
->inst
<< "' "
741 << "entity='" << session
->entity_name
<< "' "
742 << "cmd=" << m
->cmd
<< ": access denied";
743 ss
<< "access denied' does your client key have mgr caps?"
744 " See http://docs.ceph.com/docs/master/mgr/administrator/#client-authentication";
745 cmdctx
->reply(-EACCES
, ss
);
751 << "from='" << session
->inst
<< "' "
752 << "entity='" << session
->entity_name
<< "' "
753 << "cmd=" << m
->cmd
<< ": dispatch";
756 // service map commands
757 if (prefix
== "service dump") {
759 f
.reset(Formatter::create("json-pretty"));
760 cluster_state
.with_servicemap([&](const ServiceMap
&service_map
) {
761 f
->dump_object("service_map", service_map
);
763 f
->flush(cmdctx
->odata
);
764 cmdctx
->reply(0, ss
);
767 if (prefix
== "service status") {
769 f
.reset(Formatter::create("json-pretty"));
770 // only include state from services that are in the persisted service map
771 f
->open_object_section("service_status");
773 cluster_state
.with_servicemap([&](const ServiceMap
& service_map
) {
776 for (auto& p
: s
.services
) {
777 f
->open_object_section(p
.first
.c_str());
778 for (auto& q
: p
.second
.daemons
) {
779 f
->open_object_section(q
.first
.c_str());
780 DaemonKey
key(p
.first
, q
.first
);
781 assert(daemon_state
.exists(key
));
782 auto daemon
= daemon_state
.get(key
);
783 Mutex::Locker
l(daemon
->lock
);
784 f
->dump_stream("status_stamp") << daemon
->service_status_stamp
;
785 f
->dump_stream("last_beacon") << daemon
->last_service_beacon
;
786 f
->open_object_section("status");
787 for (auto& r
: daemon
->service_status
) {
788 f
->dump_string(r
.first
.c_str(), r
.second
);
796 f
->flush(cmdctx
->odata
);
797 cmdctx
->reply(0, ss
);
801 if (prefix
== "config set") {
804 cmd_getval(cct
, cmdctx
->cmdmap
, "key", key
);
805 cmd_getval(cct
, cmdctx
->cmdmap
, "value", val
);
806 r
= cct
->_conf
->set_val(key
, val
, true, &ss
);
808 cct
->_conf
->apply_changes(nullptr);
810 cmdctx
->reply(0, ss
);
817 if (prefix
== "pg scrub" ||
818 prefix
== "pg repair" ||
819 prefix
== "pg deep-scrub") {
820 string scrubop
= prefix
.substr(3, string::npos
);
823 cmd_getval(g_ceph_context
, cmdctx
->cmdmap
, "pgid", pgidstr
);
824 if (!pgid
.parse(pgidstr
.c_str())) {
825 ss
<< "invalid pgid '" << pgidstr
<< "'";
826 cmdctx
->reply(-EINVAL
, ss
);
829 bool pg_exists
= false;
830 cluster_state
.with_osdmap([&](const OSDMap
& osdmap
) {
831 pg_exists
= osdmap
.pg_exists(pgid
);
834 ss
<< "pg " << pgid
<< " dne";
835 cmdctx
->reply(-ENOENT
, ss
);
838 int acting_primary
= -1;
839 cluster_state
.with_osdmap([&](const OSDMap
& osdmap
) {
840 acting_primary
= osdmap
.get_pg_acting_primary(pgid
);
842 if (acting_primary
== -1) {
843 ss
<< "pg " << pgid
<< " has no primary osd";
844 cmdctx
->reply(-EAGAIN
, ss
);
847 auto p
= osd_cons
.find(acting_primary
);
848 if (p
== osd_cons
.end()) {
849 ss
<< "pg " << pgid
<< " primary osd." << acting_primary
850 << " is not currently connected";
851 cmdctx
->reply(-EAGAIN
, ss
);
853 vector
<pg_t
> pgs
= { pgid
};
854 for (auto& con
: p
->second
) {
855 con
->send_message(new MOSDScrub(monc
->get_fsid(),
858 scrubop
== "deep-scrub"));
860 ss
<< "instructing pg " << pgid
<< " on osd." << acting_primary
861 << " to " << scrubop
;
862 cmdctx
->reply(0, ss
);
864 } else if (prefix
== "osd scrub" ||
865 prefix
== "osd deep-scrub" ||
866 prefix
== "osd repair") {
868 cmd_getval(g_ceph_context
, cmdctx
->cmdmap
, "who", whostr
);
870 get_str_vec(prefix
, pvec
);
873 if (whostr
== "*" || whostr
== "all" || whostr
== "any") {
874 cluster_state
.with_osdmap([&](const OSDMap
& osdmap
) {
875 for (int i
= 0; i
< osdmap
.get_max_osd(); i
++)
876 if (osdmap
.is_up(i
)) {
881 long osd
= parse_osd_id(whostr
.c_str(), &ss
);
883 ss
<< "invalid osd '" << whostr
<< "'";
884 cmdctx
->reply(-EINVAL
, ss
);
887 cluster_state
.with_osdmap([&](const OSDMap
& osdmap
) {
888 if (osdmap
.is_up(osd
)) {
893 ss
<< "osd." << osd
<< " is not up";
894 cmdctx
->reply(-EAGAIN
, ss
);
898 set
<int> sent_osds
, failed_osds
;
899 for (auto osd
: osds
) {
900 auto p
= osd_cons
.find(osd
);
901 if (p
== osd_cons
.end()) {
902 failed_osds
.insert(osd
);
904 sent_osds
.insert(osd
);
905 for (auto& con
: p
->second
) {
906 con
->send_message(new MOSDScrub(monc
->get_fsid(),
907 pvec
.back() == "repair",
908 pvec
.back() == "deep-scrub"));
912 if (failed_osds
.size() == osds
.size()) {
913 ss
<< "failed to instruct osd(s) " << osds
<< " to " << pvec
.back()
914 << " (not connected)";
917 ss
<< "instructed osd(s) " << sent_osds
<< " to " << pvec
.back();
918 if (!failed_osds
.empty()) {
919 ss
<< "; osd(s) " << failed_osds
<< " were not connected";
923 cmdctx
->reply(0, ss
);
925 } else if (prefix
== "osd reweight-by-pg" ||
926 prefix
== "osd reweight-by-utilization" ||
927 prefix
== "osd test-reweight-by-pg" ||
928 prefix
== "osd test-reweight-by-utilization") {
930 prefix
== "osd reweight-by-pg" || prefix
== "osd test-reweight-by-pg";
932 prefix
== "osd test-reweight-by-pg" ||
933 prefix
== "osd test-reweight-by-utilization";
935 cmd_getval(g_ceph_context
, cmdctx
->cmdmap
, "oload", oload
, int64_t(120));
937 vector
<string
> poolnames
;
938 cmd_getval(g_ceph_context
, cmdctx
->cmdmap
, "pools", poolnames
);
939 cluster_state
.with_osdmap([&](const OSDMap
& osdmap
) {
940 for (const auto& poolname
: poolnames
) {
941 int64_t pool
= osdmap
.lookup_pg_pool_name(poolname
);
943 ss
<< "pool '" << poolname
<< "' does not exist";
950 cmdctx
->reply(r
, ss
);
953 double max_change
= g_conf
->mon_reweight_max_change
;
954 cmd_getval(g_ceph_context
, cmdctx
->cmdmap
, "max_change", max_change
);
955 if (max_change
<= 0.0) {
956 ss
<< "max_change " << max_change
<< " must be positive";
957 cmdctx
->reply(-EINVAL
, ss
);
960 int64_t max_osds
= g_conf
->mon_reweight_max_osds
;
961 cmd_getval(g_ceph_context
, cmdctx
->cmdmap
, "max_osds", max_osds
);
963 ss
<< "max_osds " << max_osds
<< " must be positive";
964 cmdctx
->reply(-EINVAL
, ss
);
967 string no_increasing
;
968 cmd_getval(g_ceph_context
, cmdctx
->cmdmap
, "no_increasing", no_increasing
);
970 mempool::osdmap::map
<int32_t, uint32_t> new_weights
;
971 r
= cluster_state
.with_pgmap([&](const PGMap
& pgmap
) {
972 return cluster_state
.with_osdmap([&](const OSDMap
& osdmap
) {
973 return reweight::by_utilization(osdmap
, pgmap
,
978 pools
.empty() ? NULL
: &pools
,
979 no_increasing
== "--no-increasing",
981 &ss
, &out_str
, f
.get());
985 dout(10) << "reweight::by_utilization: finished with " << out_str
<< dendl
;
988 f
->flush(cmdctx
->odata
);
990 cmdctx
->odata
.append(out_str
);
993 ss
<< "FAILED reweight-by-pg";
994 cmdctx
->reply(r
, ss
);
996 } else if (r
== 0 || dry_run
) {
998 cmdctx
->reply(r
, ss
);
1001 json_spirit::Object json_object
;
1002 for (const auto& osd_weight
: new_weights
) {
1003 json_spirit::Config::add(json_object
,
1004 std::to_string(osd_weight
.first
),
1005 std::to_string(osd_weight
.second
));
1007 string s
= json_spirit::write(json_object
);
1008 std::replace(begin(s
), end(s
), '\"', '\'');
1011 "\"prefix\": \"osd reweightn\", "
1012 "\"weights\": \"" + s
+ "\""
1014 auto on_finish
= new ReplyOnFinish(cmdctx
);
1015 monc
->start_mon_command({cmd
}, {},
1016 &on_finish
->from_mon
, &on_finish
->outs
, on_finish
);
1019 } else if (prefix
== "osd df") {
1021 cmd_getval(g_ceph_context
, cmdctx
->cmdmap
, "output_method", method
);
1022 r
= cluster_state
.with_pgservice([&](const PGMapStatService
& pgservice
) {
1023 return cluster_state
.with_osdmap([&](const OSDMap
& osdmap
) {
1024 print_osd_utilization(osdmap
, &pgservice
, ss
,
1025 f
.get(), method
== "tree");
1027 cmdctx
->odata
.append(ss
);
1031 cmdctx
->reply(r
, "");
1033 } else if (prefix
== "osd safe-to-destroy") {
1035 cmd_getval(g_ceph_context
, cmdctx
->cmdmap
, "ids", ids
);
1038 cluster_state
.with_osdmap([&](const OSDMap
& osdmap
) {
1039 r
= osdmap
.parse_osd_id_list(ids
, &osds
, &ss
);
1041 if (!r
&& osds
.empty()) {
1042 ss
<< "must specify one or more OSDs";
1046 cmdctx
->reply(r
, ss
);
1049 set
<int> active_osds
, missing_stats
, stored_pgs
;
1050 int affected_pgs
= 0;
1051 cluster_state
.with_pgmap([&](const PGMap
& pg_map
) {
1052 if (pg_map
.num_pg_unknown
> 0) {
1053 ss
<< pg_map
.num_pg_unknown
<< " pgs have unknown state; cannot draw"
1054 << " any conclusions";
1058 int num_active_clean
= 0;
1059 for (auto& p
: pg_map
.num_pg_by_state
) {
1060 unsigned want
= PG_STATE_ACTIVE
|PG_STATE_CLEAN
;
1061 if ((p
.first
& want
) == want
) {
1062 num_active_clean
+= p
.second
;
1065 cluster_state
.with_osdmap([&](const OSDMap
& osdmap
) {
1066 for (auto osd
: osds
) {
1067 if (!osdmap
.exists(osd
)) {
1068 continue; // clearly safe to destroy
1070 auto q
= pg_map
.num_pg_by_osd
.find(osd
);
1071 if (q
!= pg_map
.num_pg_by_osd
.end()) {
1072 if (q
->second
.acting
> 0 || q
->second
.up
> 0) {
1073 active_osds
.insert(osd
);
1074 affected_pgs
+= q
->second
.acting
+ q
->second
.up
;
1078 if (num_active_clean
< pg_map
.num_pg
) {
1079 // all pgs aren't active+clean; we need to be careful.
1080 auto p
= pg_map
.osd_stat
.find(osd
);
1081 if (p
== pg_map
.osd_stat
.end()) {
1082 missing_stats
.insert(osd
);
1084 if (p
->second
.num_pgs
> 0) {
1085 stored_pgs
.insert(osd
);
1091 if (!r
&& !active_osds
.empty()) {
1092 ss
<< "OSD(s) " << active_osds
<< " have " << affected_pgs
1093 << " pgs currently mapped to them";
1095 } else if (!missing_stats
.empty()) {
1096 ss
<< "OSD(s) " << missing_stats
<< " have no reported stats, and not all"
1097 << " PGs are active+clean; we cannot draw any conclusions";
1099 } else if (!stored_pgs
.empty()) {
1100 ss
<< "OSD(s) " << stored_pgs
<< " last reported they still store some PG"
1101 << " data, and not all PGs are active+clean; we cannot be sure they"
1102 << " aren't still needed.";
1106 cmdctx
->reply(r
, ss
);
1109 ss
<< "OSD(s) " << osds
<< " are safe to destroy without reducing data"
1111 cmdctx
->reply(0, ss
);
1113 } else if (prefix
== "osd ok-to-stop") {
1115 cmd_getval(g_ceph_context
, cmdctx
->cmdmap
, "ids", ids
);
1118 cluster_state
.with_osdmap([&](const OSDMap
& osdmap
) {
1119 r
= osdmap
.parse_osd_id_list(ids
, &osds
, &ss
);
1121 if (!r
&& osds
.empty()) {
1122 ss
<< "must specify one or more OSDs";
1126 cmdctx
->reply(r
, ss
);
1129 map
<pg_t
,int> pg_delta
; // pgid -> net acting set size change
1130 int dangerous_pgs
= 0;
1131 cluster_state
.with_pgmap([&](const PGMap
& pg_map
) {
1132 return cluster_state
.with_osdmap([&](const OSDMap
& osdmap
) {
1133 if (pg_map
.num_pg_unknown
> 0) {
1134 ss
<< pg_map
.num_pg_unknown
<< " pgs have unknown state; "
1135 << "cannot draw any conclusions";
1139 for (auto osd
: osds
) {
1140 auto p
= pg_map
.pg_by_osd
.find(osd
);
1141 if (p
!= pg_map
.pg_by_osd
.end()) {
1142 for (auto& pgid
: p
->second
) {
1147 for (auto& p
: pg_delta
) {
1148 auto q
= pg_map
.pg_stat
.find(p
.first
);
1149 if (q
== pg_map
.pg_stat
.end()) {
1150 ss
<< "missing information about " << p
.first
<< "; cannot draw"
1151 << " any conclusions";
1155 if (!(q
->second
.state
& PG_STATE_ACTIVE
) ||
1156 (q
->second
.state
& PG_STATE_DEGRADED
)) {
1157 // we don't currently have a good way to tell *how* degraded
1158 // a degraded PG is, so we have to assume we cannot remove
1159 // any more replicas/shards.
1163 const pg_pool_t
*pi
= osdmap
.get_pg_pool(p
.first
.pool());
1165 ++dangerous_pgs
; // pool is creating or deleting
1167 if (q
->second
.acting
.size() + p
.second
< pi
->min_size
) {
1175 cmdctx
->reply(r
, ss
);
1178 if (dangerous_pgs
) {
1179 ss
<< dangerous_pgs
<< " PGs are already degraded or might become "
1181 cmdctx
->reply(-EBUSY
, ss
);
1184 ss
<< "OSD(s) " << osds
<< " are ok to stop without reducing"
1185 << " availability, provided there are no other concurrent failures"
1186 << " or interventions. " << pg_delta
.size() << " PGs are likely to be"
1187 << " degraded (but remain available) as a result.";
1188 cmdctx
->reply(0, ss
);
1190 } else if (prefix
== "pg force-recovery" ||
1191 prefix
== "pg force-backfill" ||
1192 prefix
== "pg cancel-force-recovery" ||
1193 prefix
== "pg cancel-force-backfill") {
1194 string forceop
= prefix
.substr(3, string::npos
);
1195 list
<pg_t
> parsed_pgs
;
1196 map
<int, list
<pg_t
> > osdpgs
;
1198 // figure out actual op just once
1200 if (forceop
== "force-recovery") {
1201 actual_op
= OFR_RECOVERY
;
1202 } else if (forceop
== "force-backfill") {
1203 actual_op
= OFR_BACKFILL
;
1204 } else if (forceop
== "cancel-force-backfill") {
1205 actual_op
= OFR_BACKFILL
| OFR_CANCEL
;
1206 } else if (forceop
== "cancel-force-recovery") {
1207 actual_op
= OFR_RECOVERY
| OFR_CANCEL
;
1210 // covnert pg names to pgs, discard any invalid ones while at it
1212 // we don't want to keep pgidstr and pgidstr_nodup forever
1213 vector
<string
> pgidstr
;
1214 // get pgids to process and prune duplicates
1215 cmd_getval(g_ceph_context
, cmdctx
->cmdmap
, "pgid", pgidstr
);
1216 set
<string
> pgidstr_nodup(pgidstr
.begin(), pgidstr
.end());
1217 if (pgidstr
.size() != pgidstr_nodup
.size()) {
1218 // move elements only when there were duplicates, as this
1220 pgidstr
.resize(pgidstr_nodup
.size());
1221 auto it
= pgidstr_nodup
.begin();
1222 for (size_t i
= 0 ; i
< pgidstr_nodup
.size(); i
++) {
1223 pgidstr
[i
] = std::move(*it
++);
1227 cluster_state
.with_pgmap([&](const PGMap
& pg_map
) {
1228 for (auto& pstr
: pgidstr
) {
1230 if (!parsed_pg
.parse(pstr
.c_str())) {
1231 ss
<< "invalid pgid '" << pstr
<< "'; ";
1234 auto workit
= pg_map
.pg_stat
.find(parsed_pg
);
1235 if (workit
== pg_map
.pg_stat
.end()) {
1236 ss
<< "pg " << pstr
<< " does not exist; ";
1239 pg_stat_t workpg
= workit
->second
;
1241 // discard pgs for which user requests are pointless
1245 if ((workpg
.state
& (PG_STATE_DEGRADED
| PG_STATE_RECOVERY_WAIT
| PG_STATE_RECOVERING
)) == 0) {
1246 // don't return error, user script may be racing with cluster. not fatal.
1247 ss
<< "pg " << pstr
<< " doesn't require recovery; ";
1249 } else if (workpg
.state
& PG_STATE_FORCED_RECOVERY
) {
1250 ss
<< "pg " << pstr
<< " recovery already forced; ";
1251 // return error, as it may be a bug in user script
1257 if ((workpg
.state
& (PG_STATE_DEGRADED
| PG_STATE_BACKFILL_WAIT
| PG_STATE_BACKFILLING
)) == 0) {
1258 ss
<< "pg " << pstr
<< " doesn't require backfilling; ";
1260 } else if (workpg
.state
& PG_STATE_FORCED_BACKFILL
) {
1261 ss
<< "pg " << pstr
<< " backfill already forced; ";
1266 case OFR_BACKFILL
| OFR_CANCEL
:
1267 if ((workpg
.state
& PG_STATE_FORCED_BACKFILL
) == 0) {
1268 ss
<< "pg " << pstr
<< " backfill not forced; ";
1272 case OFR_RECOVERY
| OFR_CANCEL
:
1273 if ((workpg
.state
& PG_STATE_FORCED_RECOVERY
) == 0) {
1274 ss
<< "pg " << pstr
<< " recovery not forced; ";
1279 assert(0 == "actual_op value is not supported");
1282 parsed_pgs
.push_back(std::move(parsed_pg
));
1287 // group pgs to process by osd
1288 for (auto& pgid
: parsed_pgs
) {
1289 auto workit
= pg_map
.pg_stat
.find(pgid
);
1290 if (workit
!= pg_map
.pg_stat
.end()) {
1291 pg_stat_t workpg
= workit
->second
;
1292 set
<int32_t> osds(workpg
.up
.begin(), workpg
.up
.end());
1293 osds
.insert(workpg
.acting
.begin(), workpg
.acting
.end());
1294 for (auto i
: osds
) {
1295 osdpgs
[i
].push_back(pgid
);
1303 // respond with error only when no pgs are correct
1304 // yes, in case of mixed errors, only the last one will be emitted,
1305 // but the message presented will be fine
1306 if (parsed_pgs
.size() != 0) {
1307 // clear error to not confuse users/scripts
1311 // optimize the command -> messages conversion, use only one message per distinct OSD
1312 cluster_state
.with_osdmap([&](const OSDMap
& osdmap
) {
1313 for (auto& i
: osdpgs
) {
1314 if (osdmap
.is_up(i
.first
)) {
1315 vector
<pg_t
> pgvec(make_move_iterator(i
.second
.begin()), make_move_iterator(i
.second
.end()));
1316 auto p
= osd_cons
.find(i
.first
);
1317 if (p
== osd_cons
.end()) {
1318 ss
<< "osd." << i
.first
<< " is not currently connected";
1322 for (auto& con
: p
->second
) {
1323 con
->send_message(new MOSDForceRecovery(monc
->get_fsid(), pgvec
, actual_op
));
1325 ss
<< "instructing pg(s) " << i
.second
<< " on osd." << i
.first
<< " to " << forceop
<< "; ";
1330 cmdctx
->reply(r
, ss
);
1333 r
= cluster_state
.with_pgmap([&](const PGMap
& pg_map
) {
1334 return cluster_state
.with_osdmap([&](const OSDMap
& osdmap
) {
1335 return process_pg_map_command(prefix
, cmdctx
->cmdmap
, pg_map
, osdmap
,
1336 f
.get(), &ss
, &cmdctx
->odata
);
1340 if (r
!= -EOPNOTSUPP
) {
1341 cmdctx
->reply(r
, ss
);
1346 // None of the special native commands,
1347 ActivePyModule
*handler
= nullptr;
1348 auto py_commands
= py_modules
.get_py_commands();
1349 for (const auto &pyc
: py_commands
) {
1350 auto pyc_prefix
= cmddesc_get_prefix(pyc
.cmdstring
);
1351 dout(1) << "pyc_prefix: '" << pyc_prefix
<< "'" << dendl
;
1352 if (pyc_prefix
== prefix
) {
1353 handler
= pyc
.handler
;
1358 if (handler
== nullptr) {
1359 ss
<< "No handler found for '" << prefix
<< "'";
1360 dout(4) << "No handler found for '" << prefix
<< "'" << dendl
;
1361 cmdctx
->reply(-EINVAL
, ss
);
1364 // Okay, now we have a handler to call, but we must not call it
1365 // in this thread, because the python handlers can do anything,
1366 // including blocking, and including calling back into mgr.
1367 dout(4) << "passing through " << cmdctx
->cmdmap
.size() << dendl
;
1368 finisher
.queue(new FunctionContext([cmdctx
, handler
](int r_
) {
1369 std::stringstream ds
;
1370 std::stringstream ss
;
1371 int r
= handler
->handle_command(cmdctx
->cmdmap
, &ds
, &ss
);
1372 cmdctx
->odata
.append(ds
);
1373 cmdctx
->reply(r
, ss
);
1379 void DaemonServer::_prune_pending_service_map()
1381 utime_t cutoff
= ceph_clock_now();
1382 cutoff
-= g_conf
->get_val
<double>("mgr_service_beacon_grace");
1383 auto p
= pending_service_map
.services
.begin();
1384 while (p
!= pending_service_map
.services
.end()) {
1385 auto q
= p
->second
.daemons
.begin();
1386 while (q
!= p
->second
.daemons
.end()) {
1387 DaemonKey
key(p
->first
, q
->first
);
1388 if (!daemon_state
.exists(key
)) {
1389 derr
<< "missing key " << key
<< dendl
;
1393 auto daemon
= daemon_state
.get(key
);
1394 Mutex::Locker
l(daemon
->lock
);
1395 if (daemon
->last_service_beacon
== utime_t()) {
1396 // we must have just restarted; assume they are alive now.
1397 daemon
->last_service_beacon
= ceph_clock_now();
1401 if (daemon
->last_service_beacon
< cutoff
) {
1402 dout(10) << "pruning stale " << p
->first
<< "." << q
->first
1403 << " last_beacon " << daemon
->last_service_beacon
<< dendl
;
1404 q
= p
->second
.daemons
.erase(q
);
1405 pending_service_map_dirty
= pending_service_map
.epoch
;
1410 if (p
->second
.daemons
.empty()) {
1411 p
= pending_service_map
.services
.erase(p
);
1412 pending_service_map_dirty
= pending_service_map
.epoch
;
1419 void DaemonServer::send_report()
1422 if (ceph_clock_now() - started_at
> g_conf
->get_val
<int64_t>("mgr_stats_period") * 4.0) {
1424 reported_osds
.clear();
1425 dout(1) << "Giving up on OSDs that haven't reported yet, sending "
1426 << "potentially incomplete PG state to mon" << dendl
;
1428 dout(1) << "Not sending PG status to monitor yet, waiting for OSDs"
1434 auto m
= new MMonMgrReport();
1435 py_modules
.get_health_checks(&m
->health_checks
);
1437 cluster_state
.with_pgmap([&](const PGMap
& pg_map
) {
1438 cluster_state
.update_delta_stats();
1440 if (pending_service_map
.epoch
) {
1441 _prune_pending_service_map();
1442 if (pending_service_map_dirty
>= pending_service_map
.epoch
) {
1443 pending_service_map
.modified
= ceph_clock_now();
1444 ::encode(pending_service_map
, m
->service_map_bl
, CEPH_FEATURES_ALL
);
1445 dout(10) << "sending service_map e" << pending_service_map
.epoch
1447 pending_service_map
.epoch
++;
1451 cluster_state
.with_osdmap([&](const OSDMap
& osdmap
) {
1452 // FIXME: no easy way to get mon features here. this will do for
1453 // now, though, as long as we don't make a backward-incompat change.
1454 pg_map
.encode_digest(osdmap
, m
->get_data(), CEPH_FEATURES_ALL
);
1455 dout(10) << pg_map
<< dendl
;
1457 pg_map
.get_health_checks(g_ceph_context
, osdmap
,
1460 dout(10) << m
->health_checks
.checks
.size() << " health checks"
1462 dout(20) << "health checks:\n";
1463 JSONFormatter
jf(true);
1464 jf
.dump_object("health_checks", m
->health_checks
);
1470 auto osds
= daemon_state
.get_by_service("osd");
1471 map
<osd_metric
, unique_ptr
<OSDHealthMetricCollector
>> accumulated
;
1472 for (const auto& osd
: osds
) {
1473 Mutex::Locker
l(osd
.second
->lock
);
1474 for (const auto& metric
: osd
.second
->osd_health_metrics
) {
1475 auto acc
= accumulated
.find(metric
.get_type());
1476 if (acc
== accumulated
.end()) {
1477 auto collector
= OSDHealthMetricCollector::create(metric
.get_type());
1479 derr
<< __func__
<< " " << osd
.first
<< "." << osd
.second
1480 << " sent me an unknown health metric: "
1481 << static_cast<uint8_t>(metric
.get_type()) << dendl
;
1484 tie(acc
, std::ignore
) = accumulated
.emplace(metric
.get_type(),
1485 std::move(collector
));
1487 acc
->second
->update(osd
.first
, metric
);
1490 for (const auto& acc
: accumulated
) {
1491 acc
.second
->summarize(m
->health_checks
);
1493 // TODO? We currently do not notify the PyModules
1494 // TODO: respect needs_send, so we send the report only if we are asked to do
1495 // so, or the state is updated.
1496 monc
->send_mon_message(m
);
1499 void DaemonServer::got_service_map()
1501 Mutex::Locker
l(lock
);
1503 cluster_state
.with_servicemap([&](const ServiceMap
& service_map
) {
1504 if (pending_service_map
.epoch
== 0) {
1505 // we just started up
1506 dout(10) << "got initial map e" << service_map
.epoch
<< dendl
;
1507 pending_service_map
= service_map
;
1509 // we we already active and therefore must have persisted it,
1510 // which means ours is the same or newer.
1511 dout(10) << "got updated map e" << service_map
.epoch
<< dendl
;
1513 pending_service_map
.epoch
= service_map
.epoch
+ 1;
1516 // cull missing daemons, populate new ones
1517 for (auto& p
: pending_service_map
.services
) {
1518 std::set
<std::string
> names
;
1519 for (auto& q
: p
.second
.daemons
) {
1520 names
.insert(q
.first
);
1521 DaemonKey
key(p
.first
, q
.first
);
1522 if (!daemon_state
.exists(key
)) {
1523 auto daemon
= std::make_shared
<DaemonState
>(daemon_state
.types
);
1525 daemon
->metadata
= q
.second
.metadata
;
1526 if (q
.second
.metadata
.count("hostname")) {
1527 daemon
->hostname
= q
.second
.metadata
["hostname"];
1529 daemon
->service_daemon
= true;
1530 daemon_state
.insert(daemon
);
1531 dout(10) << "added missing " << key
<< dendl
;
1534 daemon_state
.cull(p
.first
, names
);
1539 const char** DaemonServer::get_tracked_conf_keys() const
1541 static const char *KEYS
[] = {
1542 "mgr_stats_threshold",
1550 void DaemonServer::handle_conf_change(const struct md_config_t
*conf
,
1551 const std::set
<std::string
> &changed
)
1553 dout(4) << "ohai" << dendl
;
1554 // We may be called within lock (via MCommand `config set`) or outwith the
1555 // lock (via admin socket `config set`), so handle either case.
1556 const bool initially_locked
= lock
.is_locked_by_me();
1557 if (!initially_locked
) {
1561 if (changed
.count("mgr_stats_threshold") || changed
.count("mgr_stats_period")) {
1562 dout(4) << "Updating stats threshold/period on "
1563 << daemon_connections
.size() << " clients" << dendl
;
1564 // Send a fresh MMgrConfigure to all clients, so that they can follow
1565 // the new policy for transmitting stats
1566 for (auto &c
: daemon_connections
) {
1572 void DaemonServer::_send_configure(ConnectionRef c
)
1574 assert(lock
.is_locked_by_me());
1576 auto configure
= new MMgrConfigure();
1577 configure
->stats_period
= g_conf
->get_val
<int64_t>("mgr_stats_period");
1578 configure
->stats_threshold
= g_conf
->get_val
<int64_t>("mgr_stats_threshold");
1579 c
->send_message(configure
);