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.
15 #include "MgrClient.h"
17 #include "mgr/MgrContext.h"
19 #include "msg/Messenger.h"
20 #include "messages/MMgrMap.h"
21 #include "messages/MMgrReport.h"
22 #include "messages/MMgrOpen.h"
23 #include "messages/MMgrConfigure.h"
24 #include "messages/MCommand.h"
25 #include "messages/MCommandReply.h"
26 #include "messages/MPGStats.h"
28 #define dout_subsys ceph_subsys_mgrc
30 #define dout_prefix *_dout << "mgrc " << __func__ << " "
32 MgrClient::MgrClient(CephContext
*cct_
, Messenger
*msgr_
)
33 : Dispatcher(cct_
), cct(cct_
), msgr(msgr_
),
36 assert(cct
!= nullptr);
39 void MgrClient::init()
41 Mutex::Locker
l(lock
);
43 assert(msgr
!= nullptr);
48 void MgrClient::shutdown()
50 Mutex::Locker
l(lock
);
52 if (connect_retry_callback
) {
53 timer
.cancel_event(connect_retry_callback
);
54 connect_retry_callback
= nullptr;
57 // forget about in-flight commands if we are prematurely shut down
58 // (e.g., by control-C)
59 command_table
.clear();
63 session
->con
->mark_down();
68 bool MgrClient::ms_dispatch(Message
*m
)
70 Mutex::Locker
l(lock
);
72 switch(m
->get_type()) {
74 return handle_mgr_map(static_cast<MMgrMap
*>(m
));
75 case MSG_MGR_CONFIGURE
:
76 return handle_mgr_configure(static_cast<MMgrConfigure
*>(m
));
77 case MSG_COMMAND_REPLY
:
78 if (m
->get_source().type() == CEPH_ENTITY_TYPE_MGR
) {
79 handle_command_reply(static_cast<MCommandReply
*>(m
));
85 ldout(cct
, 30) << "Not handling " << *m
<< dendl
;
90 void MgrClient::reconnect()
92 assert(lock
.is_locked_by_me());
95 ldout(cct
, 4) << "Terminating session with "
96 << session
->con
->get_peer_addr() << dendl
;
97 session
->con
->mark_down();
100 if (report_callback
!= nullptr) {
101 timer
.cancel_event(report_callback
);
102 report_callback
= nullptr;
106 if (!map
.get_available()) {
107 ldout(cct
, 4) << "No active mgr available yet" << dendl
;
111 if (last_connect_attempt
!= utime_t()) {
112 utime_t now
= ceph_clock_now();
113 utime_t when
= last_connect_attempt
;
114 when
+= cct
->_conf
->mgr_connect_retry_interval
;
116 if (!connect_retry_callback
) {
117 connect_retry_callback
= new FunctionContext([this](int r
){
118 connect_retry_callback
= nullptr;
121 timer
.add_event_at(when
, connect_retry_callback
);
123 ldout(cct
, 4) << "waiting to retry connect until " << when
<< dendl
;
128 if (connect_retry_callback
) {
129 timer
.cancel_event(connect_retry_callback
);
130 connect_retry_callback
= nullptr;
133 ldout(cct
, 4) << "Starting new session with " << map
.get_active_addr()
136 inst
.addr
= map
.get_active_addr();
137 inst
.name
= entity_name_t::MGR(map
.get_active_gid());
138 last_connect_attempt
= ceph_clock_now();
140 session
.reset(new MgrSessionState());
141 session
->con
= msgr
->get_connection(inst
);
143 if (service_daemon
) {
144 daemon_dirty_status
= true;
147 // Don't send an open if we're just a client (i.e. doing
148 // command-sending, not stats etc)
149 if (!cct
->_conf
->name
.is_client() || service_daemon
) {
153 // resend any pending commands
154 for (const auto &p
: command_table
.get_commands()) {
155 MCommand
*m
= p
.second
.get_message({});
157 assert(session
->con
);
158 session
->con
->send_message(m
);
162 void MgrClient::_send_open()
164 if (session
&& session
->con
) {
165 auto open
= new MMgrOpen();
166 if (!service_name
.empty()) {
167 open
->service_name
= service_name
;
168 open
->daemon_name
= daemon_name
;
170 open
->daemon_name
= cct
->_conf
->name
.get_id();
172 if (service_daemon
) {
173 open
->service_daemon
= service_daemon
;
174 open
->daemon_metadata
= daemon_metadata
;
176 session
->con
->send_message(open
);
180 bool MgrClient::handle_mgr_map(MMgrMap
*m
)
182 assert(lock
.is_locked_by_me());
184 ldout(cct
, 20) << *m
<< dendl
;
187 ldout(cct
, 4) << "Got map version " << map
.epoch
<< dendl
;
190 ldout(cct
, 4) << "Active mgr is now " << map
.get_active_addr() << dendl
;
194 session
->con
->get_peer_addr() != map
.get_active_addr()) {
201 bool MgrClient::ms_handle_reset(Connection
*con
)
203 Mutex::Locker
l(lock
);
204 if (session
&& con
== session
->con
) {
205 ldout(cct
, 4) << __func__
<< " con " << con
<< dendl
;
212 bool MgrClient::ms_handle_refused(Connection
*con
)
214 // do nothing for now
218 void MgrClient::send_report()
220 assert(lock
.is_locked_by_me());
222 report_callback
= nullptr;
224 auto report
= new MMgrReport();
225 auto pcc
= cct
->get_perfcounters_collection();
227 pcc
->with_counters([this, report
](
228 const PerfCountersCollection::CounterMap
&by_path
)
230 ENCODE_START(1, 1, report
->packed
);
231 for (auto p
= session
->declared
.begin(); p
!= session
->declared
.end(); ) {
232 if (by_path
.count(*p
) == 0) {
233 report
->undeclare_types
.push_back(*p
);
234 ldout(cct
,20) << __func__
<< " undeclare " << *p
<< dendl
;
235 p
= session
->declared
.erase(p
);
240 for (const auto &i
: by_path
) {
241 auto& path
= i
.first
;
242 auto& data
= *(i
.second
);
244 if (session
->declared
.count(path
) == 0) {
245 ldout(cct
,20) << __func__
<< " declare " << path
<< dendl
;
246 PerfCounterType type
;
248 if (data
.description
) {
249 type
.description
= data
.description
;
252 type
.nick
= data
.nick
;
254 type
.type
= data
.type
;
255 report
->declare_types
.push_back(std::move(type
));
256 session
->declared
.insert(path
);
259 ::encode(static_cast<uint64_t>(data
.u64
), report
->packed
);
260 if (data
.type
& PERFCOUNTER_LONGRUNAVG
) {
261 ::encode(static_cast<uint64_t>(data
.avgcount
), report
->packed
);
262 ::encode(static_cast<uint64_t>(data
.avgcount2
), report
->packed
);
265 ENCODE_FINISH(report
->packed
);
267 ldout(cct
, 20) << by_path
.size() << " counters, of which "
268 << report
->declare_types
.size() << " new" << dendl
;
271 ldout(cct
, 20) << "encoded " << report
->packed
.length() << " bytes" << dendl
;
273 if (daemon_name
.size()) {
274 report
->daemon_name
= daemon_name
;
276 report
->daemon_name
= cct
->_conf
->name
.get_id();
278 report
->service_name
= service_name
;
280 if (daemon_dirty_status
) {
281 report
->daemon_status
= daemon_status
;
282 daemon_dirty_status
= false;
285 session
->con
->send_message(report
);
287 if (stats_period
!= 0) {
288 report_callback
= new FunctionContext([this](int r
){send_report();});
289 timer
.add_event_after(stats_period
, report_callback
);
295 void MgrClient::send_pgstats()
297 if (pgstats_cb
&& session
) {
298 session
->con
->send_message(pgstats_cb());
302 bool MgrClient::handle_mgr_configure(MMgrConfigure
*m
)
304 assert(lock
.is_locked_by_me());
306 ldout(cct
, 20) << *m
<< dendl
;
309 lderr(cct
) << "dropping unexpected configure message" << dendl
;
314 ldout(cct
, 4) << "stats_period=" << m
->stats_period
<< dendl
;
316 bool starting
= (stats_period
== 0) && (m
->stats_period
!= 0);
317 stats_period
= m
->stats_period
;
326 int MgrClient::start_command(const vector
<string
>& cmd
, const bufferlist
& inbl
,
327 bufferlist
*outbl
, string
*outs
,
330 Mutex::Locker
l(lock
);
332 ldout(cct
, 20) << "cmd: " << cmd
<< dendl
;
334 if (map
.epoch
== 0) {
335 ldout(cct
,20) << " no MgrMap, assuming EACCES" << dendl
;
339 auto &op
= command_table
.start_command();
344 op
.on_finish
= onfinish
;
346 if (session
&& session
->con
) {
347 // Leaving fsid argument null because it isn't used.
348 MCommand
*m
= op
.get_message({});
349 session
->con
->send_message(m
);
354 bool MgrClient::handle_command_reply(MCommandReply
*m
)
356 assert(lock
.is_locked_by_me());
358 ldout(cct
, 20) << *m
<< dendl
;
360 const auto tid
= m
->get_tid();
361 if (!command_table
.exists(tid
)) {
362 ldout(cct
, 4) << "handle_command_reply tid " << m
->get_tid()
363 << " not found" << dendl
;
368 auto &op
= command_table
.get_command(tid
);
370 op
.outbl
->claim(m
->get_data());
378 op
.on_finish
->complete(m
->r
);
381 command_table
.erase(tid
);
387 int MgrClient::service_daemon_register(
388 const std::string
& service
,
389 const std::string
& name
,
390 const std::map
<std::string
,std::string
>& metadata
)
392 Mutex::Locker
l(lock
);
398 // normal ceph entity types are not allowed!
401 if (service_daemon
) {
404 ldout(cct
,1) << service
<< "." << name
<< " metadata " << metadata
<< dendl
;
405 service_daemon
= true;
406 service_name
= service
;
408 daemon_metadata
= metadata
;
409 daemon_dirty_status
= true;
412 if (cct
->_conf
->name
.is_client() && session
&& session
->con
) {
419 int MgrClient::service_daemon_update_status(
420 const std::map
<std::string
,std::string
>& status
)
422 Mutex::Locker
l(lock
);
423 ldout(cct
,10) << status
<< dendl
;
424 daemon_status
= status
;
425 daemon_dirty_status
= true;