ceph/src/mgr/BaseMgrModule.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2016 John Spray <john.spray@redhat.com>
   7  *
   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.
  12  */
  13
  14 /**
  15  * The interface we present to python code that runs within
  16  * ceph-mgr.  This is implemented as a Python class from which
  17  * all modules must inherit -- access to the Ceph state is then
  18  * available as methods on that object.
  19  */
  20
  21 #include "Python.h"
  22
  23 #include "Mgr.h"
  24
  25 #include "mon/MonClient.h"
  26 #include "common/errno.h"
  27 #include "common/version.h"
  28
  29 #include "BaseMgrModule.h"
  30 #include "Gil.h"
  31
  32 #include <algorithm>
  33
  34 #define dout_context g_ceph_context
  35 #define dout_subsys ceph_subsys_mgr
  36
  37 #define PLACEHOLDER ""
  38
  39
  40 typedef struct {
  41   PyObject_HEAD
  42   ActivePyModules *py_modules;
  43   ActivePyModule *this_module;
  44 } BaseMgrModule;
  45
  46 class MonCommandCompletion : public Context
  47 {
  48   ActivePyModules *py_modules;
  49   PyObject *python_completion;
  50   const std::string tag;
  51   SafeThreadState pThreadState;
  52
  53 public:
  54   std::string outs;
  55   bufferlist outbl;
  56
  57   MonCommandCompletion(
  58       ActivePyModules *py_modules_, PyObject* ev,
  59       const std::string &tag_, PyThreadState *ts_)
  60     : py_modules(py_modules_), python_completion(ev),
  61       tag(tag_), pThreadState(ts_)
  62   {
  63     ceph_assert(python_completion != nullptr);
  64     Py_INCREF(python_completion);
  65   }
  66
  67   ~MonCommandCompletion() override
  68   {
  69     if (python_completion) {
  70       // Usually do this in finish(): this path is only for if we're
  71       // being destroyed without completing.
  72       Gil gil(pThreadState, true);
  73       Py_DECREF(python_completion);
  74       python_completion = nullptr;
  75     }
  76   }
  77
  78   void finish(int r) override
  79   {
  80     ceph_assert(python_completion != nullptr);
  81
  82     dout(10) << "MonCommandCompletion::finish()" << dendl;
  83     {
  84       // Scoped so the Gil is released before calling notify_all()
  85       // Create new thread state because this is called via the MonClient
  86       // Finisher, not the PyModules finisher.
  87       Gil gil(pThreadState, true);
  88
  89       auto set_fn = PyObject_GetAttrString(python_completion, "complete");
  90       ceph_assert(set_fn != nullptr);
  91
  92       auto pyR = PyInt_FromLong(r);
  93       auto pyOutBl = PyString_FromString(outbl.to_str().c_str());
  94       auto pyOutS = PyString_FromString(outs.c_str());
  95       auto args = PyTuple_Pack(3, pyR, pyOutBl, pyOutS);
  96       Py_DECREF(pyR);
  97       Py_DECREF(pyOutBl);
  98       Py_DECREF(pyOutS);
  99
 100       auto rtn = PyObject_CallObject(set_fn, args);
 101       if (rtn != nullptr) {
 102         Py_DECREF(rtn);
 103       }
 104       Py_DECREF(args);
 105       Py_DECREF(set_fn);
 106
 107       Py_DECREF(python_completion);
 108       python_completion = nullptr;
 109     }
 110     py_modules->notify_all("command", tag);
 111   }
 112 };
 113
 114
 115 static PyObject*
 116 ceph_send_command(BaseMgrModule *self, PyObject *args)
 117 {
 118   // Like mon, osd, mds
 119   char *type = nullptr;
 120
 121   // Like "23" for an OSD or "myid" for an MDS
 122   char *name = nullptr;
 123
 124   char *cmd_json = nullptr;
 125   char *tag = nullptr;
 126   PyObject *completion = nullptr;
 127   if (!PyArg_ParseTuple(args, "Ossss:ceph_send_command",
 128         &completion, &type, &name, &cmd_json, &tag)) {
 129     return nullptr;
 130   }
 131
 132   auto set_fn = PyObject_GetAttrString(completion, "complete");
 133   if (set_fn == nullptr) {
 134     ceph_abort();  // TODO raise python exception instead
 135   } else {
 136     ceph_assert(PyCallable_Check(set_fn));
 137   }
 138   Py_DECREF(set_fn);
 139
 140   MonCommandCompletion *command_c = new MonCommandCompletion(self->py_modules,
 141       completion, tag, PyThreadState_Get());
 142
 143   PyThreadState *tstate = PyEval_SaveThread();
 144
 145   if (std::string(type) == "mon") {
 146
 147     // Wait for the latest OSDMap after each command we send to
 148     // the mons.  This is a heavy-handed hack to make life simpler
 149     // for python module authors, so that they know whenever they
 150     // run a command they've gt a fresh OSDMap afterwards.
 151     // TODO: enhance MCommand interface so that it returns
 152     // latest cluster map versions on completion, and callers
 153     // can wait for those.
 154     auto c = new FunctionContext([command_c, self](int command_r){
 155       self->py_modules->get_objecter().wait_for_latest_osdmap(
 156           new FunctionContext([command_c, command_r](int wait_r){
 157             command_c->finish(command_r);
 158           })
 159       );
 160     });
 161
 162     self->py_modules->get_monc().start_mon_command(
 163         name,
 164         {cmd_json},
 165         {},
 166         &command_c->outbl,
 167         &command_c->outs,
 168         c);
 169   } else if (std::string(type) == "osd") {
 170     std::string err;
 171     uint64_t osd_id = strict_strtoll(name, 10, &err);
 172     if (!err.empty()) {
 173       delete command_c;
 174       string msg("invalid osd_id: ");
 175       msg.append("\"").append(name).append("\"");
 176       PyEval_RestoreThread(tstate);
 177       PyErr_SetString(PyExc_ValueError, msg.c_str());
 178       return nullptr;
 179     }
 180
 181     ceph_tid_t tid;
 182     self->py_modules->get_objecter().osd_command(
 183         osd_id,
 184         {cmd_json},
 185         {},
 186         &tid,
 187         &command_c->outbl,
 188         &command_c->outs,
 189         command_c);
 190   } else if (std::string(type) == "mds") {
 191     int r = self->py_modules->get_client().mds_command(
 192         name,
 193         {cmd_json},
 194         {},
 195         &command_c->outbl,
 196         &command_c->outs,
 197         command_c);
 198     if (r != 0) {
 199       string msg("failed to send command to mds: ");
 200       msg.append(cpp_strerror(r));
 201       PyEval_RestoreThread(tstate);
 202       PyErr_SetString(PyExc_RuntimeError, msg.c_str());
 203       return nullptr;
 204     }
 205   } else if (std::string(type) == "pg") {
 206     pg_t pgid;
 207     if (!pgid.parse(name)) {
 208       delete command_c;
 209       string msg("invalid pgid: ");
 210       msg.append("\"").append(name).append("\"");
 211       PyEval_RestoreThread(tstate);
 212       PyErr_SetString(PyExc_ValueError, msg.c_str());
 213       return nullptr;
 214     }
 215
 216     ceph_tid_t tid;
 217     self->py_modules->get_objecter().pg_command(
 218         pgid,
 219         {cmd_json},
 220         {},
 221         &tid,
 222         &command_c->outbl,
 223         &command_c->outs,
 224         command_c);
 225     PyEval_RestoreThread(tstate);
 226     return nullptr;
 227   } else {
 228     delete command_c;
 229     string msg("unknown service type: ");
 230     msg.append(type);
 231     PyEval_RestoreThread(tstate);
 232     PyErr_SetString(PyExc_ValueError, msg.c_str());
 233     return nullptr;
 234   }
 235
 236   PyEval_RestoreThread(tstate);
 237   Py_RETURN_NONE;
 238 }
 239
 240 static PyObject*
 241 ceph_set_health_checks(BaseMgrModule *self, PyObject *args)
 242 {
 243   PyObject *checks = NULL;
 244   if (!PyArg_ParseTuple(args, "O:ceph_set_health_checks", &checks)) {
 245     return NULL;
 246   }
 247   if (!PyDict_Check(checks)) {
 248     derr << __func__ << " arg not a dict" << dendl;
 249     Py_RETURN_NONE;
 250   }
 251   PyObject *checksls = PyDict_Items(checks);
 252   health_check_map_t out_checks;
 253   for (int i = 0; i < PyList_Size(checksls); ++i) {
 254     PyObject *kv = PyList_GET_ITEM(checksls, i);
 255     char *check_name = nullptr;
 256     PyObject *check_info = nullptr;
 257     if (!PyArg_ParseTuple(kv, "sO:pair", &check_name, &check_info)) {
 258       derr << __func__ << " dict item " << i
 259            << " not a size 2 tuple" << dendl;
 260       continue;
 261     }
 262     if (!PyDict_Check(check_info)) {
 263       derr << __func__ << " item " << i << " " << check_name
 264            << " value not a dict" << dendl;
 265       continue;
 266     }
 267     health_status_t severity = HEALTH_OK;
 268     string summary;
 269     list<string> detail;
 270     PyObject *infols = PyDict_Items(check_info);
 271     for (int j = 0; j < PyList_Size(infols); ++j) {
 272       PyObject *pair = PyList_GET_ITEM(infols, j);
 273       if (!PyTuple_Check(pair)) {
 274         derr << __func__ << " item " << i << " pair " << j
 275              << " not a tuple" << dendl;
 276         continue;
 277       }
 278       char *k = nullptr;
 279       PyObject *v = nullptr;
 280       if (!PyArg_ParseTuple(pair, "sO:pair", &k, &v)) {
 281         derr << __func__ << " item " << i << " pair " << j
 282              << " not a size 2 tuple" << dendl;
 283         continue;
 284       }
 285       string ks(k);
 286       if (ks == "severity") {
 287         if (!PyString_Check(v)) {
 288           derr << __func__ << " check " << check_name
 289                << " severity value not string" << dendl;
 290           continue;
 291         }
 292         string vs(PyString_AsString(v));
 293         if (vs == "warning") {
 294           severity = HEALTH_WARN;
 295         } else if (vs == "error") {
 296           severity = HEALTH_ERR;
 297         }
 298       } else if (ks == "summary") {
 299         if (!PyString_Check(v)) {
 300           derr << __func__ << " check " << check_name
 301                << " summary value not string" << dendl;
 302           continue;
 303         }
 304         summary = PyString_AsString(v);
 305       } else if (ks == "detail") {
 306         if (!PyList_Check(v)) {
 307           derr << __func__ << " check " << check_name
 308                << " detail value not list" << dendl;
 309           continue;
 310         }
 311         for (int k = 0; k < PyList_Size(v); ++k) {
 312           PyObject *di = PyList_GET_ITEM(v, k);
 313           if (!PyString_Check(di)) {
 314             derr << __func__ << " check " << check_name
 315                  << " detail item " << k << " not a string" << dendl;
 316             continue;
 317           }
 318           detail.push_back(PyString_AsString(di));
 319         }
 320       } else {
 321         derr << __func__ << " check " << check_name
 322              << " unexpected key " << k << dendl;
 323       }
 324     }
 325     auto& d = out_checks.add(check_name, severity, summary);
 326     d.detail.swap(detail);
 327   }
 328
 329   JSONFormatter jf(true);
 330   dout(10) << "module " << self->this_module->get_name()
 331           << " health checks:\n";
 332   out_checks.dump(&jf);
 333   jf.flush(*_dout);
 334   *_dout << dendl;
 335
 336   PyThreadState *tstate = PyEval_SaveThread();
 337   self->py_modules->set_health_checks(self->this_module->get_name(),
 338                                       std::move(out_checks));
 339   PyEval_RestoreThread(tstate);
 340
 341   Py_RETURN_NONE;
 342 }
 343
 344
 345 static PyObject*
 346 ceph_state_get(BaseMgrModule *self, PyObject *args)
 347 {
 348   char *what = NULL;
 349   if (!PyArg_ParseTuple(args, "s:ceph_state_get", &what)) {
 350     return NULL;
 351   }
 352
 353   return self->py_modules->get_python(what);
 354 }
 355
 356
 357 static PyObject*
 358 ceph_get_server(BaseMgrModule *self, PyObject *args)
 359 {
 360   char *hostname = NULL;
 361   if (!PyArg_ParseTuple(args, "z:ceph_get_server", &hostname)) {
 362     return NULL;
 363   }
 364
 365   if (hostname) {
 366     return self->py_modules->get_server_python(hostname);
 367   } else {
 368     return self->py_modules->list_servers_python();
 369   }
 370 }
 371
 372 static PyObject*
 373 ceph_get_mgr_id(BaseMgrModule *self, PyObject *args)
 374 {
 375   return PyString_FromString(g_conf()->name.get_id().c_str());
 376 }
 377
 378 static PyObject*
 379 ceph_option_get(BaseMgrModule *self, PyObject *args)
 380 {
 381   char *what = nullptr;
 382   if (!PyArg_ParseTuple(args, "s:ceph_option_get", &what)) {
 383     derr << "Invalid args!" << dendl;
 384     return nullptr;
 385   }
 386
 387   std::string value;
 388   int r = g_conf().get_val(string(what), &value);
 389   if (r >= 0) {
 390     dout(10) << "ceph_option_get " << what << " found: " << value << dendl;
 391     return PyString_FromString(value.c_str());
 392   } else {
 393     dout(4) << "ceph_option_get " << what << " not found " << dendl;
 394     Py_RETURN_NONE;
 395   }
 396 }
 397
 398 static PyObject*
 399 ceph_get_module_option(BaseMgrModule *self, PyObject *args)
 400 {
 401   char *module = nullptr;
 402   char *key = nullptr;
 403   char *prefix = nullptr;
 404   if (!PyArg_ParseTuple(args, "ss|s:ceph_get_module_option", &module, &key,
 405                         &prefix)) {
 406     derr << "Invalid args!" << dendl;
 407     return nullptr;
 408   }
 409   std::string str_prefix;
 410   if (prefix) {
 411     str_prefix = prefix;
 412   }
 413   assert(self->this_module->py_module);
 414   auto pResult = self->py_modules->get_typed_config(module, key, str_prefix);
 415   return pResult;
 416 }
 417
 418 static PyObject*
 419 ceph_store_get_prefix(BaseMgrModule *self, PyObject *args)
 420 {
 421   char *prefix = nullptr;
 422   if (!PyArg_ParseTuple(args, "s:ceph_store_get_prefix", &prefix)) {
 423     derr << "Invalid args!" << dendl;
 424     return nullptr;
 425   }
 426
 427   return self->py_modules->get_store_prefix(self->this_module->get_name(),
 428       prefix);
 429 }
 430
 431 static PyObject*
 432 ceph_set_module_option(BaseMgrModule *self, PyObject *args)
 433 {
 434   char *module = nullptr;
 435   char *key = nullptr;
 436   char *value = nullptr;
 437   if (!PyArg_ParseTuple(args, "ssz:ceph_set_module_option",
 438         &module, &key, &value)) {
 439     derr << "Invalid args!" << dendl;
 440     return nullptr;
 441   }
 442   boost::optional<string> val;
 443   if (value) {
 444     val = value;
 445   }
 446   self->py_modules->set_config(module, key, val);
 447
 448   Py_RETURN_NONE;
 449 }
 450
 451 static PyObject*
 452 ceph_store_get(BaseMgrModule *self, PyObject *args)
 453 {
 454   char *what = nullptr;
 455   if (!PyArg_ParseTuple(args, "s:ceph_store_get", &what)) {
 456     derr << "Invalid args!" << dendl;
 457     return nullptr;
 458   }
 459
 460   std::string value;
 461   bool found = self->py_modules->get_store(self->this_module->get_name(),
 462       what, &value);
 463   if (found) {
 464     dout(10) << "ceph_store_get " << what << " found: " << value.c_str() << dendl;
 465     return PyString_FromString(value.c_str());
 466   } else {
 467     dout(4) << "ceph_store_get " << what << " not found " << dendl;
 468     Py_RETURN_NONE;
 469   }
 470 }
 471
 472 static PyObject*
 473 ceph_store_set(BaseMgrModule *self, PyObject *args)
 474 {
 475   char *key = nullptr;
 476   char *value = nullptr;
 477   if (!PyArg_ParseTuple(args, "sz:ceph_store_set", &key, &value)) {
 478     return nullptr;
 479   }
 480   boost::optional<string> val;
 481   if (value) {
 482     val = value;
 483   }
 484   self->py_modules->set_store(self->this_module->get_name(), key, val);
 485
 486   Py_RETURN_NONE;
 487 }
 488
 489 static PyObject*
 490 get_metadata(BaseMgrModule *self, PyObject *args)
 491 {
 492   char *svc_name = NULL;
 493   char *svc_id = NULL;
 494   if (!PyArg_ParseTuple(args, "ss:get_metadata", &svc_name, &svc_id)) {
 495     return nullptr;
 496   }
 497   return self->py_modules->get_metadata_python(svc_name, svc_id);
 498 }
 499
 500 static PyObject*
 501 get_daemon_status(BaseMgrModule *self, PyObject *args)
 502 {
 503   char *svc_name = NULL;
 504   char *svc_id = NULL;
 505   if (!PyArg_ParseTuple(args, "ss:get_daemon_status", &svc_name,
 506                         &svc_id)) {
 507     return nullptr;
 508   }
 509   return self->py_modules->get_daemon_status_python(svc_name, svc_id);
 510 }
 511
 512 static PyObject*
 513 ceph_log(BaseMgrModule *self, PyObject *args)
 514 {
 515   int level = 0;
 516   char *record = nullptr;
 517   if (!PyArg_ParseTuple(args, "is:log", &level, &record)) {
 518     return nullptr;
 519   }
 520
 521   ceph_assert(self->this_module);
 522
 523   self->this_module->log(level, record);
 524
 525   Py_RETURN_NONE;
 526 }
 527
 528 static PyObject*
 529 ceph_cluster_log(BaseMgrModule *self, PyObject *args)
 530 {
 531   int prio = 0;
 532   char *channel = nullptr;
 533   char *message = nullptr;
 534   std::vector<std::string> channels = { "audit", "cluster" };
 535
 536   if (!PyArg_ParseTuple(args, "sis:ceph_cluster_log", &channel, &prio, &message)) {
 537     return nullptr;
 538   }
 539
 540   if (std::find(channels.begin(), channels.end(), std::string(channel)) == channels.end()) {
 541     std::string msg("Unknown channel: ");
 542     msg.append(channel);
 543     PyErr_SetString(PyExc_ValueError, msg.c_str());
 544     return nullptr;
 545   }
 546
 547   PyThreadState *tstate = PyEval_SaveThread();
 548   self->py_modules->cluster_log(channel, (clog_type)prio, message);
 549   PyEval_RestoreThread(tstate);
 550
 551   Py_RETURN_NONE;
 552 }
 553
 554 static PyObject *
 555 ceph_get_version(BaseMgrModule *self, PyObject *args)
 556 {
 557   return PyString_FromString(pretty_version_to_str().c_str());
 558 }
 559
 560 static PyObject *
 561 ceph_get_context(BaseMgrModule *self, PyObject *args)
 562 {
 563   return self->py_modules->get_context();
 564 }
 565
 566 static PyObject*
 567 get_counter(BaseMgrModule *self, PyObject *args)
 568 {
 569   char *svc_name = nullptr;
 570   char *svc_id = nullptr;
 571   char *counter_path = nullptr;
 572   if (!PyArg_ParseTuple(args, "sss:get_counter", &svc_name,
 573                                                   &svc_id, &counter_path)) {
 574     return nullptr;
 575   }
 576   return self->py_modules->get_counter_python(
 577       svc_name, svc_id, counter_path);
 578 }
 579
 580 static PyObject*
 581 get_latest_counter(BaseMgrModule *self, PyObject *args)
 582 {
 583   char *svc_name = nullptr;
 584   char *svc_id = nullptr;
 585   char *counter_path = nullptr;
 586   if (!PyArg_ParseTuple(args, "sss:get_counter", &svc_name,
 587                                                   &svc_id, &counter_path)) {
 588     return nullptr;
 589   }
 590   return self->py_modules->get_latest_counter_python(
 591       svc_name, svc_id, counter_path);
 592 }
 593
 594 static PyObject*
 595 get_perf_schema(BaseMgrModule *self, PyObject *args)
 596 {
 597   char *type_str = nullptr;
 598   char *svc_id = nullptr;
 599   if (!PyArg_ParseTuple(args, "ss:get_perf_schema", &type_str,
 600                                                     &svc_id)) {
 601     return nullptr;
 602   }
 603
 604   return self->py_modules->get_perf_schema_python(type_str, svc_id);
 605 }
 606
 607 static PyObject *
 608 ceph_get_osdmap(BaseMgrModule *self, PyObject *args)
 609 {
 610   return self->py_modules->get_osdmap();
 611 }
 612
 613 static PyObject*
 614 ceph_set_uri(BaseMgrModule *self, PyObject *args)
 615 {
 616   char *svc_str = nullptr;
 617   if (!PyArg_ParseTuple(args, "s:ceph_advertize_service",
 618         &svc_str)) {
 619     return nullptr;
 620   }
 621
 622   // We call down into PyModules even though we have a MgrPyModule
 623   // reference here, because MgrPyModule's fields are protected
 624   // by PyModules' lock.
 625   PyThreadState *tstate = PyEval_SaveThread();
 626   self->py_modules->set_uri(self->this_module->get_name(), svc_str);
 627   PyEval_RestoreThread(tstate);
 628
 629   Py_RETURN_NONE;
 630 }
 631
 632 static PyObject*
 633 ceph_have_mon_connection(BaseMgrModule *self, PyObject *args)
 634 {
 635   if (self->py_modules->get_monc().is_connected()) {
 636     Py_RETURN_TRUE;
 637   } else {
 638     Py_RETURN_FALSE;
 639   }
 640 }
 641
 642 static PyObject*
 643 ceph_update_progress_event(BaseMgrModule *self, PyObject *args)
 644 {
 645   char *evid = nullptr;
 646   char *desc = nullptr;
 647   float progress = 0.0;
 648   if (!PyArg_ParseTuple(args, "ssf:ceph_update_progress_event",
 649                         &evid, &desc, &progress)) {
 650     return nullptr;
 651   }
 652
 653   PyThreadState *tstate = PyEval_SaveThread();
 654   self->py_modules->update_progress_event(evid, desc, progress);
 655   PyEval_RestoreThread(tstate);
 656
 657   Py_RETURN_NONE;
 658 }
 659
 660 static PyObject*
 661 ceph_complete_progress_event(BaseMgrModule *self, PyObject *args)
 662 {
 663   char *evid = nullptr;
 664   if (!PyArg_ParseTuple(args, "s:ceph_complete_progress_event",
 665                         &evid)) {
 666     return nullptr;
 667   }
 668
 669   PyThreadState *tstate = PyEval_SaveThread();
 670   self->py_modules->complete_progress_event(evid);
 671   PyEval_RestoreThread(tstate);
 672
 673   Py_RETURN_NONE;
 674 }
 675
 676 static PyObject*
 677 ceph_clear_all_progress_events(BaseMgrModule *self, PyObject *args)
 678 {
 679   PyThreadState *tstate = PyEval_SaveThread();
 680   self->py_modules->clear_all_progress_events();
 681   PyEval_RestoreThread(tstate);
 682
 683   Py_RETURN_NONE;
 684 }
 685
 686
 687
 688 static PyObject *
 689 ceph_dispatch_remote(BaseMgrModule *self, PyObject *args)
 690 {
 691   char *other_module = nullptr;
 692   char *method = nullptr;
 693   PyObject *remote_args = nullptr;
 694   PyObject *remote_kwargs = nullptr;
 695   if (!PyArg_ParseTuple(args, "ssOO:ceph_dispatch_remote",
 696         &other_module, &method, &remote_args, &remote_kwargs)) {
 697     return nullptr;
 698   }
 699
 700   // Early error handling, because if the module doesn't exist then we
 701   // won't be able to use its thread state to set python error state
 702   // inside dispatch_remote().
 703   if (!self->py_modules->module_exists(other_module)) {
 704     derr << "no module '" << other_module << "'" << dendl;
 705     PyErr_SetString(PyExc_ImportError, "Module not found");
 706     return nullptr;
 707   }
 708
 709   // Drop GIL from calling python thread state, it will be taken
 710   // both for checking for method existence and for executing method.
 711   PyThreadState *tstate = PyEval_SaveThread();
 712
 713   if (!self->py_modules->method_exists(other_module, method)) {
 714     PyEval_RestoreThread(tstate);
 715     PyErr_SetString(PyExc_NameError, "Method not found");
 716     return nullptr;
 717   }
 718
 719   std::string err;
 720   auto result = self->py_modules->dispatch_remote(other_module, method,
 721       remote_args, remote_kwargs, &err);
 722
 723   PyEval_RestoreThread(tstate);
 724
 725   if (result == nullptr) {
 726     std::stringstream ss;
 727     ss << "Remote method threw exception: " << err;
 728     PyErr_SetString(PyExc_RuntimeError, ss.str().c_str());
 729     derr << ss.str() << dendl;
 730   }
 731
 732   return result;
 733 }
 734
 735 static PyObject*
 736 ceph_add_osd_perf_query(BaseMgrModule *self, PyObject *args)
 737 {
 738   static const std::string NAME_KEY_DESCRIPTOR = "key_descriptor";
 739   static const std::string NAME_COUNTERS_DESCRIPTORS =
 740       "performance_counter_descriptors";
 741   static const std::string NAME_LIMIT = "limit";
 742   static const std::string NAME_SUB_KEY_TYPE = "type";
 743   static const std::string NAME_SUB_KEY_REGEX = "regex";
 744   static const std::string NAME_LIMIT_ORDER_BY = "order_by";
 745   static const std::string NAME_LIMIT_MAX_COUNT = "max_count";
 746   static const std::map<std::string, OSDPerfMetricSubKeyType> sub_key_types = {
 747     {"client_id", OSDPerfMetricSubKeyType::CLIENT_ID},
 748     {"client_address", OSDPerfMetricSubKeyType::CLIENT_ADDRESS},
 749     {"pool_id", OSDPerfMetricSubKeyType::POOL_ID},
 750     {"namespace", OSDPerfMetricSubKeyType::NAMESPACE},
 751     {"osd_id", OSDPerfMetricSubKeyType::OSD_ID},
 752     {"pg_id", OSDPerfMetricSubKeyType::PG_ID},
 753     {"object_name", OSDPerfMetricSubKeyType::OBJECT_NAME},
 754     {"snap_id", OSDPerfMetricSubKeyType::SNAP_ID},
 755   };
 756   static const std::map<std::string, PerformanceCounterType> counter_types = {
 757     {"ops", PerformanceCounterType::OPS},
 758     {"write_ops", PerformanceCounterType::WRITE_OPS},
 759     {"read_ops", PerformanceCounterType::READ_OPS},
 760     {"bytes", PerformanceCounterType::BYTES},
 761     {"write_bytes", PerformanceCounterType::WRITE_BYTES},
 762     {"read_bytes", PerformanceCounterType::READ_BYTES},
 763     {"latency", PerformanceCounterType::LATENCY},
 764     {"write_latency", PerformanceCounterType::WRITE_LATENCY},
 765     {"read_latency", PerformanceCounterType::READ_LATENCY},
 766   };
 767
 768   PyObject *py_query = nullptr;
 769   if (!PyArg_ParseTuple(args, "O:ceph_add_osd_perf_query", &py_query)) {
 770     derr << "Invalid args!" << dendl;
 771     return nullptr;
 772   }
 773   if (!PyDict_Check(py_query)) {
 774     derr << __func__ << " arg not a dict" << dendl;
 775     Py_RETURN_NONE;
 776   }
 777
 778   PyObject *query_params = PyDict_Items(py_query);
 779   OSDPerfMetricQuery query;
 780   std::optional<OSDPerfMetricLimit> limit;
 781
 782   // {
 783   //   'key_descriptor': [
 784   //     {'type': subkey_type, 'regex': regex_pattern},
 785   //     ...
 786   //   ],
 787   //   'performance_counter_descriptors': [
 788   //     list, of, descriptor, types
 789   //   ],
 790   //   'limit': {'order_by': performance_counter_type, 'max_count': n},
 791   // }
 792
 793   for (int i = 0; i < PyList_Size(query_params); ++i) {
 794     PyObject *kv = PyList_GET_ITEM(query_params, i);
 795     char *query_param_name = nullptr;
 796     PyObject *query_param_val = nullptr;
 797     if (!PyArg_ParseTuple(kv, "sO:pair", &query_param_name, &query_param_val)) {
 798       derr << __func__ << " dict item " << i << " not a size 2 tuple" << dendl;
 799       Py_RETURN_NONE;
 800     }
 801     if (query_param_name == NAME_KEY_DESCRIPTOR) {
 802       if (!PyList_Check(query_param_val)) {
 803         derr << __func__ << " " << query_param_name << " not a list" << dendl;
 804         Py_RETURN_NONE;
 805       }
 806       for (int j = 0; j < PyList_Size(query_param_val); j++) {
 807         PyObject *sub_key = PyList_GET_ITEM(query_param_val, j);
 808         if (!PyDict_Check(sub_key)) {
 809           derr << __func__ << " query " << query_param_name << " item " << j
 810                << " not a dict" << dendl;
 811           Py_RETURN_NONE;
 812         }
 813         OSDPerfMetricSubKeyDescriptor d;
 814         PyObject *sub_key_params = PyDict_Items(sub_key);
 815         for (int k = 0; k < PyList_Size(sub_key_params); ++k) {
 816           PyObject *pair = PyList_GET_ITEM(sub_key_params, k);
 817           if (!PyTuple_Check(pair)) {
 818             derr << __func__ << " query " << query_param_name << " item " << j
 819                  << " pair " << k << " not a tuple" << dendl;
 820             Py_RETURN_NONE;
 821           }
 822           char *param_name = nullptr;
 823           PyObject *param_value = nullptr;
 824           if (!PyArg_ParseTuple(pair, "sO:pair", &param_name, &param_value)) {
 825             derr << __func__ << " query " << query_param_name << " item " << j
 826                  << " pair " << k << " not a size 2 tuple" << dendl;
 827             Py_RETURN_NONE;
 828           }
 829           if (param_name == NAME_SUB_KEY_TYPE) {
 830             if (!PyString_Check(param_value)) {
 831               derr << __func__ << " query " << query_param_name << " item " << j
 832                    << " contains invalid param " << param_name << dendl;
 833               Py_RETURN_NONE;
 834             }
 835             auto type = PyString_AsString(param_value);
 836             auto it = sub_key_types.find(type);
 837             if (it == sub_key_types.end()) {
 838               derr << __func__ << " query " << query_param_name << " item " << j
 839                    << " contains invalid type " << dendl;
 840               Py_RETURN_NONE;
 841             }
 842             d.type = it->second;
 843           } else if (param_name == NAME_SUB_KEY_REGEX) {
 844             if (!PyString_Check(param_value)) {
 845               derr << __func__ << " query " << query_param_name << " item " << j
 846                    << " contains invalid param " << param_name << dendl;
 847               Py_RETURN_NONE;
 848             }
 849             d.regex_str = PyString_AsString(param_value);
 850             try {
 851               d.regex = {d.regex_str.c_str()};
 852             } catch (const std::regex_error& e) {
 853               derr << __func__ << " query " << query_param_name << " item " << j
 854                    << " contains invalid regex " << d.regex_str << dendl;
 855               Py_RETURN_NONE;
 856             }
 857             if (d.regex.mark_count() == 0) {
 858               derr << __func__ << " query " << query_param_name << " item " << j
 859                    << " regex " << d.regex_str << ": no capturing groups"
 860                    << dendl;
 861               Py_RETURN_NONE;
 862             }
 863           } else {
 864             derr << __func__ << " query " << query_param_name << " item " << j
 865                  << " contains invalid param " << param_name << dendl;
 866             Py_RETURN_NONE;
 867           }
 868         }
 869         if (d.type == static_cast<OSDPerfMetricSubKeyType>(-1) ||
 870             d.regex_str.empty()) {
 871           derr << __func__ << " query " << query_param_name << " item " << i
 872                << " invalid" << dendl;
 873           Py_RETURN_NONE;
 874         }
 875         query.key_descriptor.push_back(d);
 876       }
 877     } else if (query_param_name == NAME_COUNTERS_DESCRIPTORS) {
 878       if (!PyList_Check(query_param_val)) {
 879         derr << __func__ << " " << query_param_name << " not a list" << dendl;
 880         Py_RETURN_NONE;
 881       }
 882       for (int j = 0; j < PyList_Size(query_param_val); j++) {
 883         PyObject *py_type = PyList_GET_ITEM(query_param_val, j);
 884         if (!PyString_Check(py_type)) {
 885           derr << __func__ << " query " << query_param_name << " item " << j
 886                << " not a string" << dendl;
 887           Py_RETURN_NONE;
 888         }
 889         auto type = PyString_AsString(py_type);
 890         auto it = counter_types.find(type);
 891         if (it == counter_types.end()) {
 892           derr << __func__ << " query " << query_param_name << " item " << type
 893                << " is not valid type" << dendl;
 894           Py_RETURN_NONE;
 895         }
 896         query.performance_counter_descriptors.push_back(it->second);
 897       }
 898     } else if (query_param_name == NAME_LIMIT) {
 899       if (!PyDict_Check(query_param_val)) {
 900         derr << __func__ << " query " << query_param_name << " not a dict"
 901              << dendl;
 902         Py_RETURN_NONE;
 903       }
 904
 905       limit = OSDPerfMetricLimit();
 906       PyObject *limit_params = PyDict_Items(query_param_val);
 907
 908       for (int j = 0; j < PyList_Size(limit_params); ++j) {
 909         PyObject *kv = PyList_GET_ITEM(limit_params, j);
 910         char *limit_param_name = nullptr;
 911         PyObject *limit_param_val = nullptr;
 912         if (!PyArg_ParseTuple(kv, "sO:pair", &limit_param_name,
 913                               &limit_param_val)) {
 914           derr << __func__ << " limit item " << j << " not a size 2 tuple"
 915                << dendl;
 916           Py_RETURN_NONE;
 917         }
 918
 919         if (limit_param_name == NAME_LIMIT_ORDER_BY) {
 920           if (!PyString_Check(limit_param_val)) {
 921             derr << __func__ << " " << limit_param_name << " not a string"
 922                  << dendl;
 923             Py_RETURN_NONE;
 924           }
 925           auto order_by = PyString_AsString(limit_param_val);
 926           auto it = counter_types.find(order_by);
 927           if (it == counter_types.end()) {
 928             derr << __func__ << " limit " << limit_param_name
 929                  << " not a valid counter type" << dendl;
 930             Py_RETURN_NONE;
 931           }
 932           limit->order_by = it->second;
 933         } else if (limit_param_name == NAME_LIMIT_MAX_COUNT) {
 934 #if PY_MAJOR_VERSION <= 2
 935           if (!PyInt_Check(limit_param_val) && !PyLong_Check(limit_param_val)) {
 936 #else
 937           if (!PyLong_Check(limit_param_val)) {
 938 #endif
 939             derr << __func__ << " " << limit_param_name << " not an int"
 940                  << dendl;
 941             Py_RETURN_NONE;
 942           }
 943           limit->max_count = PyLong_AsLong(limit_param_val);
 944         } else {
 945           derr << __func__ << " unknown limit param: " << limit_param_name
 946                << dendl;
 947           Py_RETURN_NONE;
 948         }
 949       }
 950     } else {
 951       derr << __func__ << " unknown query param: " << query_param_name << dendl;
 952       Py_RETURN_NONE;
 953     }
 954   }
 955
 956   if (query.key_descriptor.empty() ||
 957       query.performance_counter_descriptors.empty()) {
 958     derr << __func__ << " invalid query" << dendl;
 959     Py_RETURN_NONE;
 960   }
 961
 962   if (limit) {
 963     auto &ds = query.performance_counter_descriptors;
 964     if (std::find(ds.begin(), ds.end(), limit->order_by) == ds.end()) {
 965       derr << __func__ << " limit order_by " << limit->order_by
 966            << " not in performance_counter_descriptors" << dendl;
 967       Py_RETURN_NONE;
 968     }
 969   }
 970
 971   auto query_id = self->py_modules->add_osd_perf_query(query, limit);
 972   return PyLong_FromLong(query_id);
 973 }
 974
 975 static PyObject*
 976 ceph_remove_osd_perf_query(BaseMgrModule *self, PyObject *args)
 977 {
 978   OSDPerfMetricQueryID query_id;
 979   if (!PyArg_ParseTuple(args, "i:ceph_remove_osd_perf_query", &query_id)) {
 980     derr << "Invalid args!" << dendl;
 981     return nullptr;
 982   }
 983
 984   self->py_modules->remove_osd_perf_query(query_id);
 985   Py_RETURN_NONE;
 986 }
 987
 988 static PyObject*
 989 ceph_get_osd_perf_counters(BaseMgrModule *self, PyObject *args)
 990 {
 991   OSDPerfMetricQueryID query_id;
 992   if (!PyArg_ParseTuple(args, "i:ceph_get_osd_perf_counters", &query_id)) {
 993     derr << "Invalid args!" << dendl;
 994     return nullptr;
 995   }
 996
 997   return self->py_modules->get_osd_perf_counters(query_id);
 998 }
 999
1000 PyMethodDef BaseMgrModule_methods[] = {
1001   {"_ceph_get", (PyCFunction)ceph_state_get, METH_VARARGS,
1002    "Get a cluster object"},
1003
1004   {"_ceph_get_server", (PyCFunction)ceph_get_server, METH_VARARGS,
1005    "Get a server object"},
1006
1007   {"_ceph_get_metadata", (PyCFunction)get_metadata, METH_VARARGS,
1008    "Get a service's metadata"},
1009
1010   {"_ceph_get_daemon_status", (PyCFunction)get_daemon_status, METH_VARARGS,
1011    "Get a service's status"},
1012
1013   {"_ceph_send_command", (PyCFunction)ceph_send_command, METH_VARARGS,
1014    "Send a mon command"},
1015
1016   {"_ceph_set_health_checks", (PyCFunction)ceph_set_health_checks, METH_VARARGS,
1017    "Set health checks for this module"},
1018
1019   {"_ceph_get_mgr_id", (PyCFunction)ceph_get_mgr_id, METH_NOARGS,
1020    "Get the name of the Mgr daemon where we are running"},
1021
1022   {"_ceph_get_option", (PyCFunction)ceph_option_get, METH_VARARGS,
1023    "Get a native configuration option value"},
1024
1025   {"_ceph_get_module_option", (PyCFunction)ceph_get_module_option, METH_VARARGS,
1026    "Get a module configuration option value"},
1027
1028   {"_ceph_get_store_prefix", (PyCFunction)ceph_store_get_prefix, METH_VARARGS,
1029    "Get all KV store values with a given prefix"},
1030
1031   {"_ceph_set_module_option", (PyCFunction)ceph_set_module_option, METH_VARARGS,
1032    "Set a module configuration option value"},
1033
1034   {"_ceph_get_store", (PyCFunction)ceph_store_get, METH_VARARGS,
1035    "Get a stored field"},
1036
1037   {"_ceph_set_store", (PyCFunction)ceph_store_set, METH_VARARGS,
1038    "Set a stored field"},
1039
1040   {"_ceph_get_counter", (PyCFunction)get_counter, METH_VARARGS,
1041     "Get a performance counter"},
1042
1043   {"_ceph_get_latest_counter", (PyCFunction)get_latest_counter, METH_VARARGS,
1044     "Get the latest performance counter"},
1045
1046   {"_ceph_get_perf_schema", (PyCFunction)get_perf_schema, METH_VARARGS,
1047     "Get the performance counter schema"},
1048
1049   {"_ceph_log", (PyCFunction)ceph_log, METH_VARARGS,
1050    "Emit a (local) log message"},
1051
1052   {"_ceph_cluster_log", (PyCFunction)ceph_cluster_log, METH_VARARGS,
1053    "Emit a cluster log message"},
1054
1055   {"_ceph_get_version", (PyCFunction)ceph_get_version, METH_VARARGS,
1056    "Get the ceph version of this process"},
1057
1058   {"_ceph_get_context", (PyCFunction)ceph_get_context, METH_NOARGS,
1059     "Get a CephContext* in a python capsule"},
1060
1061   {"_ceph_get_osdmap", (PyCFunction)ceph_get_osdmap, METH_NOARGS,
1062     "Get an OSDMap* in a python capsule"},
1063
1064   {"_ceph_set_uri", (PyCFunction)ceph_set_uri, METH_VARARGS,
1065     "Advertize a service URI served by this module"},
1066
1067   {"_ceph_have_mon_connection", (PyCFunction)ceph_have_mon_connection,
1068     METH_NOARGS, "Find out whether this mgr daemon currently has "
1069                  "a connection to a monitor"},
1070
1071   {"_ceph_update_progress_event", (PyCFunction)ceph_update_progress_event,
1072    METH_VARARGS, "Update status of a progress event"},
1073   {"_ceph_complete_progress_event", (PyCFunction)ceph_complete_progress_event,
1074    METH_VARARGS, "Complete a progress event"},
1075   {"_ceph_clear_all_progress_events", (PyCFunction)ceph_clear_all_progress_events,
1076    METH_NOARGS, "Clear all progress events"},
1077
1078   {"_ceph_dispatch_remote", (PyCFunction)ceph_dispatch_remote,
1079     METH_VARARGS, "Dispatch a call to another module"},
1080
1081   {"_ceph_add_osd_perf_query", (PyCFunction)ceph_add_osd_perf_query,
1082     METH_VARARGS, "Add an osd perf query"},
1083
1084   {"_ceph_remove_osd_perf_query", (PyCFunction)ceph_remove_osd_perf_query,
1085     METH_VARARGS, "Remove an osd perf query"},
1086
1087   {"_ceph_get_osd_perf_counters", (PyCFunction)ceph_get_osd_perf_counters,
1088     METH_VARARGS, "Get osd perf counters"},
1089
1090   {NULL, NULL, 0, NULL}
1091 };
1092
1093
1094 static PyObject *
1095 BaseMgrModule_new(PyTypeObject *type, PyObject *args, PyObject *kwds)
1096 {
1097     BaseMgrModule *self;
1098
1099     self = (BaseMgrModule *)type->tp_alloc(type, 0);
1100
1101     return (PyObject *)self;
1102 }
1103
1104 static int
1105 BaseMgrModule_init(BaseMgrModule *self, PyObject *args, PyObject *kwds)
1106 {
1107     PyObject *py_modules_capsule = nullptr;
1108     PyObject *this_module_capsule = nullptr;
1109     static const char *kwlist[] = {"py_modules", "this_module", NULL};
1110
1111     if (! PyArg_ParseTupleAndKeywords(args, kwds, "OO",
1112                                       const_cast<char**>(kwlist),
1113                                       &py_modules_capsule,
1114                                       &this_module_capsule)) {
1115         return -1;
1116     }
1117
1118     self->py_modules = static_cast<ActivePyModules*>(PyCapsule_GetPointer(
1119         py_modules_capsule, nullptr));
1120     ceph_assert(self->py_modules);
1121     self->this_module = static_cast<ActivePyModule*>(PyCapsule_GetPointer(
1122         this_module_capsule, nullptr));
1123     ceph_assert(self->this_module);
1124
1125     return 0;
1126 }
1127
1128 PyTypeObject BaseMgrModuleType = {
1129   PyVarObject_HEAD_INIT(NULL, 0)
1130   "ceph_module.BaseMgrModule", /* tp_name */
1131   sizeof(BaseMgrModule),     /* tp_basicsize */
1132   0,                         /* tp_itemsize */
1133   0,                         /* tp_dealloc */
1134   0,                         /* tp_print */
1135   0,                         /* tp_getattr */
1136   0,                         /* tp_setattr */
1137   0,                         /* tp_compare */
1138   0,                         /* tp_repr */
1139   0,                         /* tp_as_number */
1140   0,                         /* tp_as_sequence */
1141   0,                         /* tp_as_mapping */
1142   0,                         /* tp_hash */
1143   0,                         /* tp_call */
1144   0,                         /* tp_str */
1145   0,                         /* tp_getattro */
1146   0,                         /* tp_setattro */
1147   0,                         /* tp_as_buffer */
1148   Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE,        /* tp_flags */
1149   "ceph-mgr Python Plugin", /* tp_doc */
1150   0,                         /* tp_traverse */
1151   0,                         /* tp_clear */
1152   0,                         /* tp_richcompare */
1153   0,                         /* tp_weaklistoffset */
1154   0,                         /* tp_iter */
1155   0,                         /* tp_iternext */
1156   BaseMgrModule_methods,     /* tp_methods */
1157   0,                         /* tp_members */
1158   0,                         /* tp_getset */
1159   0,                         /* tp_base */
1160   0,                         /* tp_dict */
1161   0,                         /* tp_descr_get */
1162   0,                         /* tp_descr_set */
1163   0,                         /* tp_dictoffset */
1164   (initproc)BaseMgrModule_init,                         /* tp_init */
1165   0,                         /* tp_alloc */
1166   BaseMgrModule_new,     /* tp_new */
1167 };
1168