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 ((g_conf
&& !g_conf
->name
.is_client()) ||
154 // resend any pending commands
155 for (const auto &p
: command_table
.get_commands()) {
156 MCommand
*m
= p
.second
.get_message({});
158 assert(session
->con
);
159 session
->con
->send_message(m
);
163 void MgrClient::_send_open()
165 if (session
&& session
->con
) {
166 auto open
= new MMgrOpen();
167 if (!service_name
.empty()) {
168 open
->service_name
= service_name
;
169 open
->daemon_name
= daemon_name
;
171 open
->daemon_name
= g_conf
->name
.get_id();
173 if (service_daemon
) {
174 open
->service_daemon
= service_daemon
;
175 open
->daemon_metadata
= daemon_metadata
;
177 session
->con
->send_message(open
);
181 bool MgrClient::handle_mgr_map(MMgrMap
*m
)
183 assert(lock
.is_locked_by_me());
185 ldout(cct
, 20) << *m
<< dendl
;
188 ldout(cct
, 4) << "Got map version " << map
.epoch
<< dendl
;
191 ldout(cct
, 4) << "Active mgr is now " << map
.get_active_addr() << dendl
;
195 session
->con
->get_peer_addr() != map
.get_active_addr()) {
202 bool MgrClient::ms_handle_reset(Connection
*con
)
204 Mutex::Locker
l(lock
);
205 if (session
&& con
== session
->con
) {
206 ldout(cct
, 4) << __func__
<< " con " << con
<< dendl
;
213 bool MgrClient::ms_handle_refused(Connection
*con
)
215 // do nothing for now
219 void MgrClient::send_report()
221 assert(lock
.is_locked_by_me());
223 report_callback
= nullptr;
225 auto report
= new MMgrReport();
226 auto pcc
= cct
->get_perfcounters_collection();
228 pcc
->with_counters([this, report
](
229 const PerfCountersCollection::CounterMap
&by_path
)
231 ENCODE_START(1, 1, report
->packed
);
232 for (auto p
= session
->declared
.begin(); p
!= session
->declared
.end(); ) {
233 if (by_path
.count(*p
) == 0) {
234 report
->undeclare_types
.push_back(*p
);
235 ldout(cct
,20) << __func__
<< " undeclare " << *p
<< dendl
;
236 p
= session
->declared
.erase(p
);
241 for (const auto &i
: by_path
) {
242 auto& path
= i
.first
;
243 auto& data
= *(i
.second
);
245 if (session
->declared
.count(path
) == 0) {
246 ldout(cct
,20) << __func__
<< " declare " << path
<< dendl
;
247 PerfCounterType type
;
249 if (data
.description
) {
250 type
.description
= data
.description
;
253 type
.nick
= data
.nick
;
255 type
.type
= data
.type
;
256 report
->declare_types
.push_back(std::move(type
));
257 session
->declared
.insert(path
);
260 ::encode(static_cast<uint64_t>(data
.u64
), report
->packed
);
261 if (data
.type
& PERFCOUNTER_LONGRUNAVG
) {
262 ::encode(static_cast<uint64_t>(data
.avgcount
), report
->packed
);
263 ::encode(static_cast<uint64_t>(data
.avgcount2
), report
->packed
);
266 ENCODE_FINISH(report
->packed
);
268 ldout(cct
, 20) << by_path
.size() << " counters, of which "
269 << report
->declare_types
.size() << " new" << dendl
;
272 ldout(cct
, 20) << "encoded " << report
->packed
.length() << " bytes" << dendl
;
274 if (daemon_name
.size()) {
275 report
->daemon_name
= daemon_name
;
277 report
->daemon_name
= g_conf
->name
.get_id();
279 report
->service_name
= service_name
;
281 if (daemon_dirty_status
) {
282 report
->daemon_status
= daemon_status
;
283 daemon_dirty_status
= false;
286 session
->con
->send_message(report
);
288 if (stats_period
!= 0) {
289 report_callback
= new FunctionContext([this](int r
){send_report();});
290 timer
.add_event_after(stats_period
, report_callback
);
296 void MgrClient::send_pgstats()
298 if (pgstats_cb
&& session
) {
299 session
->con
->send_message(pgstats_cb());
303 bool MgrClient::handle_mgr_configure(MMgrConfigure
*m
)
305 assert(lock
.is_locked_by_me());
307 ldout(cct
, 20) << *m
<< dendl
;
310 lderr(cct
) << "dropping unexpected configure message" << dendl
;
315 ldout(cct
, 4) << "stats_period=" << m
->stats_period
<< dendl
;
317 bool starting
= (stats_period
== 0) && (m
->stats_period
!= 0);
318 stats_period
= m
->stats_period
;
327 int MgrClient::start_command(const vector
<string
>& cmd
, const bufferlist
& inbl
,
328 bufferlist
*outbl
, string
*outs
,
331 Mutex::Locker
l(lock
);
333 ldout(cct
, 20) << "cmd: " << cmd
<< dendl
;
335 if (map
.epoch
== 0) {
336 ldout(cct
,20) << " no MgrMap, assuming EACCES" << dendl
;
340 auto &op
= command_table
.start_command();
345 op
.on_finish
= onfinish
;
347 if (session
&& session
->con
) {
348 // Leaving fsid argument null because it isn't used.
349 MCommand
*m
= op
.get_message({});
350 session
->con
->send_message(m
);
355 bool MgrClient::handle_command_reply(MCommandReply
*m
)
357 assert(lock
.is_locked_by_me());
359 ldout(cct
, 20) << *m
<< dendl
;
361 const auto tid
= m
->get_tid();
362 if (!command_table
.exists(tid
)) {
363 ldout(cct
, 4) << "handle_command_reply tid " << m
->get_tid()
364 << " not found" << dendl
;
369 auto &op
= command_table
.get_command(tid
);
371 op
.outbl
->claim(m
->get_data());
379 op
.on_finish
->complete(m
->r
);
382 command_table
.erase(tid
);
388 int MgrClient::service_daemon_register(
389 const std::string
& service
,
390 const std::string
& name
,
391 const std::map
<std::string
,std::string
>& metadata
)
393 Mutex::Locker
l(lock
);
399 // normal ceph entity types are not allowed!
402 if (service_daemon
) {
405 ldout(cct
,1) << service
<< "." << name
<< " metadata " << metadata
<< dendl
;
406 service_daemon
= true;
407 service_name
= service
;
409 daemon_metadata
= metadata
;
410 daemon_dirty_status
= true;
413 if (g_conf
->name
.is_client() && session
&& session
->con
) {
420 int MgrClient::service_daemon_update_status(
421 const std::map
<std::string
,std::string
>& status
)
423 Mutex::Locker
l(lock
);
424 ldout(cct
,10) << status
<< dendl
;
425 daemon_status
= status
;
426 daemon_dirty_status
= true;