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) 2014 John Spray <john.spray@inktank.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 this first to get python headers earlier
15 #include "BaseMgrModule.h"
18 #include "common/errno.h"
19 #include "include/stringify.h"
21 #include "PyFormatter.h"
23 #include "osd/OSDMap.h"
24 #include "mon/MonMap.h"
26 #include "mgr/MgrContext.h"
28 // For ::config_prefix
29 #include "PyModuleRegistry.h"
31 #include "ActivePyModules.h"
33 #define dout_context g_ceph_context
34 #define dout_subsys ceph_subsys_mgr
36 #define dout_prefix *_dout << "mgr " << __func__ << " "
39 ActivePyModules::ActivePyModules(PyModuleConfig
const &config_
,
40 DaemonStateIndex
&ds
, ClusterState
&cs
,
41 MonClient
&mc
, LogChannelRef clog_
, Objecter
&objecter_
,
42 Client
&client_
, Finisher
&f
)
43 : config_cache(config_
), daemon_state(ds
), cluster_state(cs
),
44 monc(mc
), clog(clog_
), objecter(objecter_
), client(client_
), finisher(f
),
45 lock("ActivePyModules")
48 ActivePyModules::~ActivePyModules() = default;
50 void ActivePyModules::dump_server(const std::string
&hostname
,
51 const DaemonStateCollection
&dmc
,
54 f
->dump_string("hostname", hostname
);
55 f
->open_array_section("services");
56 std::string ceph_version
;
58 for (const auto &i
: dmc
) {
59 Mutex::Locker
l(i
.second
->lock
);
60 const auto &key
= i
.first
;
61 const std::string
&str_type
= key
.first
;
62 const std::string
&svc_name
= key
.second
;
64 // TODO: pick the highest version, and make sure that
65 // somewhere else (during health reporting?) we are
66 // indicating to the user if we see mixed versions
67 auto ver_iter
= i
.second
->metadata
.find("ceph_version");
68 if (ver_iter
!= i
.second
->metadata
.end()) {
69 ceph_version
= i
.second
->metadata
.at("ceph_version");
72 f
->open_object_section("service");
73 f
->dump_string("type", str_type
);
74 f
->dump_string("id", svc_name
);
79 f
->dump_string("ceph_version", ceph_version
);
84 PyObject
*ActivePyModules::get_server_python(const std::string
&hostname
)
86 PyThreadState
*tstate
= PyEval_SaveThread();
87 Mutex::Locker
l(lock
);
88 PyEval_RestoreThread(tstate
);
89 dout(10) << " (" << hostname
<< ")" << dendl
;
91 auto dmc
= daemon_state
.get_by_server(hostname
);
94 dump_server(hostname
, dmc
, &f
);
99 PyObject
*ActivePyModules::list_servers_python()
101 PyThreadState
*tstate
= PyEval_SaveThread();
102 Mutex::Locker
l(lock
);
103 PyEval_RestoreThread(tstate
);
104 dout(10) << " >" << dendl
;
106 PyFormatter
f(false, true);
107 daemon_state
.with_daemons_by_server([this, &f
]
108 (const std::map
<std::string
, DaemonStateCollection
> &all
) {
109 for (const auto &i
: all
) {
110 const auto &hostname
= i
.first
;
112 f
.open_object_section("server");
113 dump_server(hostname
, i
.second
, &f
);
121 PyObject
*ActivePyModules::get_metadata_python(
122 const std::string
&svc_type
,
123 const std::string
&svc_id
)
125 auto metadata
= daemon_state
.get(DaemonKey(svc_type
, svc_id
));
126 if (metadata
== nullptr) {
127 derr
<< "Requested missing service " << svc_type
<< "." << svc_id
<< dendl
;
131 Mutex::Locker
l(metadata
->lock
);
133 f
.dump_string("hostname", metadata
->hostname
);
134 for (const auto &i
: metadata
->metadata
) {
135 f
.dump_string(i
.first
.c_str(), i
.second
);
141 PyObject
*ActivePyModules::get_daemon_status_python(
142 const std::string
&svc_type
,
143 const std::string
&svc_id
)
145 auto metadata
= daemon_state
.get(DaemonKey(svc_type
, svc_id
));
146 if (metadata
== nullptr) {
147 derr
<< "Requested missing service " << svc_type
<< "." << svc_id
<< dendl
;
151 Mutex::Locker
l(metadata
->lock
);
153 for (const auto &i
: metadata
->service_status
) {
154 f
.dump_string(i
.first
.c_str(), i
.second
);
159 PyObject
*ActivePyModules::get_python(const std::string
&what
)
161 PyThreadState
*tstate
= PyEval_SaveThread();
162 Mutex::Locker
l(lock
);
163 PyEval_RestoreThread(tstate
);
165 if (what
== "fs_map") {
167 cluster_state
.with_fsmap([&f
](const FSMap
&fsmap
) {
171 } else if (what
== "osdmap_crush_map_text") {
173 cluster_state
.with_osdmap([&rdata
](const OSDMap
&osd_map
){
174 osd_map
.crush
->encode(rdata
, CEPH_FEATURES_SUPPORTED_DEFAULT
);
176 std::string crush_text
= rdata
.to_str();
177 return PyString_FromString(crush_text
.c_str());
178 } else if (what
.substr(0, 7) == "osd_map") {
180 cluster_state
.with_osdmap([&f
, &what
](const OSDMap
&osd_map
){
181 if (what
== "osd_map") {
183 } else if (what
== "osd_map_tree") {
184 osd_map
.print_tree(&f
, nullptr);
185 } else if (what
== "osd_map_crush") {
186 osd_map
.crush
->dump(&f
);
190 } else if (what
== "config") {
192 g_conf
->show_config(&f
);
194 } else if (what
== "mon_map") {
196 cluster_state
.with_monmap(
197 [&f
](const MonMap
&monmap
) {
202 } else if (what
== "service_map") {
204 cluster_state
.with_servicemap(
205 [&f
](const ServiceMap
&service_map
) {
206 service_map
.dump(&f
);
210 } else if (what
== "osd_metadata") {
212 auto dmc
= daemon_state
.get_by_service("osd");
213 for (const auto &i
: dmc
) {
214 Mutex::Locker
l(i
.second
->lock
);
215 f
.open_object_section(i
.first
.second
.c_str());
216 f
.dump_string("hostname", i
.second
->hostname
);
217 for (const auto &j
: i
.second
->metadata
) {
218 f
.dump_string(j
.first
.c_str(), j
.second
);
223 } else if (what
== "pg_summary") {
225 cluster_state
.with_pgmap(
226 [&f
](const PGMap
&pg_map
) {
227 std::map
<std::string
, std::map
<std::string
, uint32_t> > osds
;
228 std::map
<std::string
, std::map
<std::string
, uint32_t> > pools
;
229 std::map
<std::string
, uint32_t> all
;
230 for (const auto &i
: pg_map
.pg_stat
) {
231 const auto pool
= i
.first
.m_pool
;
232 const std::string state
= pg_state_string(i
.second
.state
);
233 // Insert to per-pool map
234 pools
[stringify(pool
)][state
]++;
235 for (const auto &osd_id
: i
.second
.acting
) {
236 osds
[stringify(osd_id
)][state
]++;
240 f
.open_object_section("by_osd");
241 for (const auto &i
: osds
) {
242 f
.open_object_section(i
.first
.c_str());
243 for (const auto &j
: i
.second
) {
244 f
.dump_int(j
.first
.c_str(), j
.second
);
249 f
.open_object_section("by_pool");
250 for (const auto &i
: pools
) {
251 f
.open_object_section(i
.first
.c_str());
252 for (const auto &j
: i
.second
) {
253 f
.dump_int(j
.first
.c_str(), j
.second
);
258 f
.open_object_section("all");
259 for (const auto &i
: all
) {
260 f
.dump_int(i
.first
.c_str(), i
.second
);
263 f
.open_object_section("pg_stats_sum");
264 pg_map
.pg_sum
.dump(&f
);
269 } else if (what
== "pg_status") {
271 cluster_state
.with_pgmap(
272 [&f
](const PGMap
&pg_map
) {
273 pg_map
.print_summary(&f
, nullptr);
277 } else if (what
== "pg_dump") {
279 cluster_state
.with_pgmap(
280 [&f
](const PGMap
&pg_map
) {
285 } else if (what
== "df") {
288 cluster_state
.with_osdmap([this, &f
](const OSDMap
&osd_map
){
289 cluster_state
.with_pgmap(
290 [&osd_map
, &f
](const PGMap
&pg_map
) {
291 pg_map
.dump_fs_stats(nullptr, &f
, true);
292 pg_map
.dump_pool_stats_full(osd_map
, nullptr, &f
, true);
296 } else if (what
== "osd_stats") {
298 cluster_state
.with_pgmap(
299 [&f
](const PGMap
&pg_map
) {
300 pg_map
.dump_osd_stats(&f
);
303 } else if (what
== "health" || what
== "mon_status") {
306 if (what
== "health") {
307 json
= cluster_state
.get_health();
308 } else if (what
== "mon_status") {
309 json
= cluster_state
.get_mon_status();
313 f
.dump_string("json", json
.to_str());
315 } else if (what
== "mgr_map") {
317 cluster_state
.with_mgrmap([&f
](const MgrMap
&mgr_map
) {
322 derr
<< "Python module requested unknown data '" << what
<< "'" << dendl
;
327 int ActivePyModules::start_one(std::string
const &module_name
,
328 PyObject
*pClass
, const SafeThreadState
&pMyThreadState
)
330 Mutex::Locker
l(lock
);
332 assert(modules
.count(module_name
) == 0);
334 modules
[module_name
].reset(new ActivePyModule(
336 pMyThreadState
, clog
));
338 int r
= modules
[module_name
]->load(this);
342 dout(4) << "Starting thread for " << module_name
<< dendl
;
343 // Giving Thread the module's module_name member as its
344 // char* thread name: thread must not outlive module class lifetime.
345 modules
[module_name
]->thread
.create(
346 modules
[module_name
]->get_name().c_str());
352 void ActivePyModules::shutdown()
354 Mutex::Locker
locker(lock
);
356 // Signal modules to drop out of serve() and/or tear down resources
357 for (auto &i
: modules
) {
358 auto module
= i
.second
.get();
359 const auto& name
= i
.first
;
362 dout(10) << "calling module " << name
<< " shutdown()" << dendl
;
364 dout(10) << "module " << name
<< " shutdown() returned" << dendl
;
368 // For modules implementing serve(), finish the threads where we
369 // were running that.
370 for (auto &i
: modules
) {
372 dout(10) << "joining module " << i
.first
<< dendl
;
373 i
.second
->thread
.join();
374 dout(10) << "joined module " << i
.first
<< dendl
;
381 void ActivePyModules::notify_all(const std::string
¬ify_type
,
382 const std::string
¬ify_id
)
384 Mutex::Locker
l(lock
);
386 dout(10) << __func__
<< ": notify_all " << notify_type
<< dendl
;
387 for (auto& i
: modules
) {
388 auto module
= i
.second
.get();
389 // Send all python calls down a Finisher to avoid blocking
390 // C++ code, and avoid any potential lock cycles.
391 finisher
.queue(new FunctionContext([module
, notify_type
, notify_id
](int r
){
392 module
->notify(notify_type
, notify_id
);
397 void ActivePyModules::notify_all(const LogEntry
&log_entry
)
399 Mutex::Locker
l(lock
);
401 dout(10) << __func__
<< ": notify_all (clog)" << dendl
;
402 for (auto& i
: modules
) {
403 auto module
= i
.second
.get();
404 // Send all python calls down a Finisher to avoid blocking
405 // C++ code, and avoid any potential lock cycles.
407 // Note intentional use of non-reference lambda binding on
408 // log_entry: we take a copy because caller's instance is
409 // probably ephemeral.
410 finisher
.queue(new FunctionContext([module
, log_entry
](int r
){
411 module
->notify_clog(log_entry
);
416 bool ActivePyModules::get_config(const std::string
&module_name
,
417 const std::string
&key
, std::string
*val
) const
419 PyThreadState
*tstate
= PyEval_SaveThread();
420 Mutex::Locker
l(lock
);
421 PyEval_RestoreThread(tstate
);
423 const std::string global_key
= PyModuleRegistry::config_prefix
424 + module_name
+ "/" + key
;
426 dout(4) << __func__
<< "key: " << global_key
<< dendl
;
428 if (config_cache
.count(global_key
)) {
429 *val
= config_cache
.at(global_key
);
436 PyObject
*ActivePyModules::get_config_prefix(const std::string
&module_name
,
437 const std::string
&prefix
) const
439 PyThreadState
*tstate
= PyEval_SaveThread();
440 Mutex::Locker
l(lock
);
441 PyEval_RestoreThread(tstate
);
443 const std::string base_prefix
= PyModuleRegistry::config_prefix
445 const std::string global_prefix
= base_prefix
+ prefix
;
446 dout(4) << __func__
<< "prefix: " << global_prefix
<< dendl
;
449 for (auto p
= config_cache
.lower_bound(global_prefix
);
450 p
!= config_cache
.end() && p
->first
.find(global_prefix
) == 0;
452 f
.dump_string(p
->first
.c_str() + base_prefix
.size(), p
->second
);
457 void ActivePyModules::set_config(const std::string
&module_name
,
458 const std::string
&key
, const boost::optional
<std::string
>& val
)
460 const std::string global_key
= PyModuleRegistry::config_prefix
461 + module_name
+ "/" + key
;
465 PyThreadState
*tstate
= PyEval_SaveThread();
466 Mutex::Locker
l(lock
);
467 PyEval_RestoreThread(tstate
);
469 config_cache
[global_key
] = *val
;
471 config_cache
.erase(global_key
);
474 std::ostringstream cmd_json
;
476 jf
.open_object_section("cmd");
478 jf
.dump_string("prefix", "config-key set");
479 jf
.dump_string("key", global_key
);
480 jf
.dump_string("val", *val
);
482 jf
.dump_string("prefix", "config-key del");
483 jf
.dump_string("key", global_key
);
487 set_cmd
.run(&monc
, cmd_json
.str());
491 if (set_cmd
.r
!= 0) {
492 // config-key set will fail if mgr's auth key has insufficient
493 // permission to set config keys
494 // FIXME: should this somehow raise an exception back into Python land?
495 dout(0) << "`config-key set " << global_key
<< " " << val
<< "` failed: "
496 << cpp_strerror(set_cmd
.r
) << dendl
;
497 dout(0) << "mon returned " << set_cmd
.r
<< ": " << set_cmd
.outs
<< dendl
;
501 std::vector
<ModuleCommand
> ActivePyModules::get_py_commands() const
503 Mutex::Locker
l(lock
);
505 std::vector
<ModuleCommand
> result
;
506 for (const auto& i
: modules
) {
507 auto module
= i
.second
.get();
508 auto mod_commands
= module
->get_commands();
509 for (auto j
: mod_commands
) {
517 std::vector
<MonCommand
> ActivePyModules::get_commands() const
519 std::vector
<ModuleCommand
> commands
= get_py_commands();
520 std::vector
<MonCommand
> result
;
521 for (auto &pyc
: commands
) {
522 result
.push_back({pyc
.cmdstring
, pyc
.helpstring
, "mgr",
523 pyc
.perm
, "cli", MonCommand::FLAG_MGR
});
529 std::map
<std::string
, std::string
> ActivePyModules::get_services() const
531 std::map
<std::string
, std::string
> result
;
532 Mutex::Locker
l(lock
);
533 for (const auto& i
: modules
) {
534 const auto &module
= i
.second
.get();
535 std::string svc_str
= module
->get_uri();
536 if (!svc_str
.empty()) {
537 result
[module
->get_name()] = svc_str
;
544 PyObject
* ActivePyModules::get_counter_python(
545 const std::string
&svc_name
,
546 const std::string
&svc_id
,
547 const std::string
&path
)
549 PyThreadState
*tstate
= PyEval_SaveThread();
550 Mutex::Locker
l(lock
);
551 PyEval_RestoreThread(tstate
);
554 f
.open_array_section(path
.c_str());
556 auto metadata
= daemon_state
.get(DaemonKey(svc_name
, svc_id
));
558 Mutex::Locker
l2(metadata
->lock
);
559 if (metadata
->perf_counters
.instances
.count(path
)) {
560 auto counter_instance
= metadata
->perf_counters
.instances
.at(path
);
561 auto counter_type
= metadata
->perf_counters
.types
.at(path
);
562 if (counter_type
.type
& PERFCOUNTER_LONGRUNAVG
) {
563 const auto &avg_data
= counter_instance
.get_data_avg();
564 for (const auto &datapoint
: avg_data
) {
565 f
.open_array_section("datapoint");
566 f
.dump_unsigned("t", datapoint
.t
.sec());
567 f
.dump_unsigned("s", datapoint
.s
);
568 f
.dump_unsigned("c", datapoint
.c
);
572 const auto &data
= counter_instance
.get_data();
573 for (const auto &datapoint
: data
) {
574 f
.open_array_section("datapoint");
575 f
.dump_unsigned("t", datapoint
.t
.sec());
576 f
.dump_unsigned("v", datapoint
.v
);
581 dout(4) << "Missing counter: '" << path
<< "' ("
582 << svc_name
<< "." << svc_id
<< ")" << dendl
;
583 dout(20) << "Paths are:" << dendl
;
584 for (const auto &i
: metadata
->perf_counters
.instances
) {
585 dout(20) << i
.first
<< dendl
;
589 dout(4) << "No daemon state for "
590 << svc_name
<< "." << svc_id
<< ")" << dendl
;
596 PyObject
* ActivePyModules::get_perf_schema_python(
597 const std::string svc_type
,
598 const std::string
&svc_id
)
600 PyThreadState
*tstate
= PyEval_SaveThread();
601 Mutex::Locker
l(lock
);
602 PyEval_RestoreThread(tstate
);
604 DaemonStateCollection daemons
;
606 if (svc_type
== "") {
607 daemons
= std::move(daemon_state
.get_all());
608 } else if (svc_id
.empty()) {
609 daemons
= std::move(daemon_state
.get_by_service(svc_type
));
611 auto key
= DaemonKey(svc_type
, svc_id
);
612 // so that the below can be a loop in all cases
613 auto got
= daemon_state
.get(key
);
614 if (got
!= nullptr) {
620 if (!daemons
.empty()) {
621 for (auto statepair
: daemons
) {
622 auto key
= statepair
.first
;
623 auto state
= statepair
.second
;
625 std::ostringstream daemon_name
;
626 daemon_name
<< key
.first
<< "." << key
.second
;
627 f
.open_object_section(daemon_name
.str().c_str());
629 Mutex::Locker
l(state
->lock
);
630 for (auto ctr_inst_iter
: state
->perf_counters
.instances
) {
631 const auto &counter_name
= ctr_inst_iter
.first
;
632 f
.open_object_section(counter_name
.c_str());
633 auto type
= state
->perf_counters
.types
[counter_name
];
634 f
.dump_string("description", type
.description
);
635 if (!type
.nick
.empty()) {
636 f
.dump_string("nick", type
.nick
);
638 f
.dump_unsigned("type", type
.type
);
639 f
.dump_unsigned("priority", type
.priority
);
645 dout(4) << __func__
<< ": No daemon state found for "
646 << svc_type
<< "." << svc_id
<< ")" << dendl
;
651 PyObject
*ActivePyModules::get_context()
653 PyThreadState
*tstate
= PyEval_SaveThread();
654 Mutex::Locker
l(lock
);
655 PyEval_RestoreThread(tstate
);
657 // Construct a capsule containing ceph context.
658 // Not incrementing/decrementing ref count on the context because
659 // it's the global one and it has process lifetime.
660 auto capsule
= PyCapsule_New(g_ceph_context
, nullptr, nullptr);
665 * Helper for our wrapped types that take a capsule in their constructor.
667 PyObject
*construct_with_capsule(
668 const std::string
&module_name
,
669 const std::string
&clsname
,
672 // Look up the OSDMap type which we will construct
673 PyObject
*module
= PyImport_ImportModule(module_name
.c_str());
675 derr
<< "Failed to import python module:" << dendl
;
676 derr
<< handle_pyerror() << dendl
;
680 PyObject
*wrapper_type
= PyObject_GetAttrString(
681 module
, (const char*)clsname
.c_str());
683 derr
<< "Failed to get python type:" << dendl
;
684 derr
<< handle_pyerror() << dendl
;
686 assert(wrapper_type
);
688 // Construct a capsule containing an OSDMap.
689 auto wrapped_capsule
= PyCapsule_New(wrapped
, nullptr, nullptr);
690 assert(wrapped_capsule
);
692 // Construct the python OSDMap
693 auto pArgs
= PyTuple_Pack(1, wrapped_capsule
);
694 auto wrapper_instance
= PyObject_CallObject(wrapper_type
, pArgs
);
695 if (wrapper_instance
== nullptr) {
696 derr
<< "Failed to construct python OSDMap:" << dendl
;
697 derr
<< handle_pyerror() << dendl
;
699 assert(wrapper_instance
!= nullptr);
701 Py_DECREF(wrapped_capsule
);
703 Py_DECREF(wrapper_type
);
706 return wrapper_instance
;
709 PyObject
*ActivePyModules::get_osdmap()
711 PyThreadState
*tstate
= PyEval_SaveThread();
712 Mutex::Locker
l(lock
);
713 PyEval_RestoreThread(tstate
);
715 OSDMap
*newmap
= new OSDMap
;
717 cluster_state
.with_osdmap([&](const OSDMap
& o
) {
718 newmap
->deepish_copy_from(o
);
721 return construct_with_capsule("mgr_module", "OSDMap", (void*)newmap
);
724 void ActivePyModules::set_health_checks(const std::string
& module_name
,
725 health_check_map_t
&& checks
)
727 Mutex::Locker
l(lock
);
728 auto p
= modules
.find(module_name
);
729 if (p
!= modules
.end()) {
730 p
->second
->set_health_checks(std::move(checks
));
734 void ActivePyModules::get_health_checks(health_check_map_t
*checks
)
736 Mutex::Locker
l(lock
);
737 for (auto& p
: modules
) {
738 p
.second
->get_health_checks(checks
);
742 void ActivePyModules::set_uri(const std::string
& module_name
,
743 const std::string
&uri
)
745 Mutex::Locker
l(lock
);
747 dout(4) << " module " << module_name
<< " set URI '" << uri
<< "'" << dendl
;
749 modules
[module_name
]->set_uri(uri
);