ceph/src/mgr/DaemonServer.h

   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 #ifndef DAEMON_SERVER_H_
  15 #define DAEMON_SERVER_H_
  16
  17 #include "PyModuleRegistry.h"
  18
  19 #include <set>
  20 #include <string>
  21 #include <boost/variant.hpp>
  22
  23 #include "common/ceph_mutex.h"
  24 #include "common/LogClient.h"
  25 #include "common/Timer.h"
  26
  27 #include <msg/Messenger.h>
  28 #include <mon/MonClient.h>
  29
  30 #include "ServiceMap.h"
  31 #include "MgrSession.h"
  32 #include "DaemonState.h"
  33 #include "MetricCollector.h"
  34 #include "OSDPerfMetricCollector.h"
  35
  36 class MMgrReport;
  37 class MMgrOpen;
  38 class MMgrClose;
  39 class MMonMgrReport;
  40 class MCommand;
  41 class MMgrCommand;
  42 struct MonCommand;
  43 class CommandContext;
  44 struct OSDPerfMetricQuery;
  45
  46
  47 /**
  48  * Server used in ceph-mgr to communicate with Ceph daemons like
  49  * MDSs and OSDs.
  50  */
  51 class DaemonServer : public Dispatcher, public md_config_obs_t
  52 {
  53 protected:
  54   boost::scoped_ptr<Throttle> client_byte_throttler;
  55   boost::scoped_ptr<Throttle> client_msg_throttler;
  56   boost::scoped_ptr<Throttle> osd_byte_throttler;
  57   boost::scoped_ptr<Throttle> osd_msg_throttler;
  58   boost::scoped_ptr<Throttle> mds_byte_throttler;
  59   boost::scoped_ptr<Throttle> mds_msg_throttler;
  60   boost::scoped_ptr<Throttle> mon_byte_throttler;
  61   boost::scoped_ptr<Throttle> mon_msg_throttler;
  62
  63   Messenger *msgr;
  64   MonClient *monc;
  65   Finisher  &finisher;
  66   DaemonStateIndex &daemon_state;
  67   ClusterState &cluster_state;
  68   PyModuleRegistry &py_modules;
  69   LogChannelRef clog, audit_clog;
  70
  71   // Connections for daemons, and clients with service names set
  72   // (i.e. those MgrClients that are allowed to send MMgrReports)
  73   std::set<ConnectionRef> daemon_connections;
  74
  75   /// connections for osds
  76   ceph::unordered_map<int,set<ConnectionRef>> osd_cons;
  77
  78   ServiceMap pending_service_map;  // uncommitted
  79
  80   epoch_t pending_service_map_dirty = 0;
  81
  82   ceph::mutex lock = ceph::make_mutex("DaemonServer");
  83
  84   static void _generate_command_map(cmdmap_t& cmdmap,
  85                                     map<string,string> &param_str_map);
  86   static const MonCommand *_get_mgrcommand(const string &cmd_prefix,
  87                                            const std::vector<MonCommand> &commands);
  88   bool _allowed_command(
  89     MgrSession *s, const string &service, const string &module,
  90     const string &prefix, const cmdmap_t& cmdmap,
  91     const map<string,string>& param_str_map,
  92     const MonCommand *this_cmd);
  93
  94 private:
  95   friend class ReplyOnFinish;
  96   bool _reply(MCommand* m,
  97               int ret, const std::string& s, const bufferlist& payload);
  98
  99   void _prune_pending_service_map();
 100
 101   utime_t started_at;
 102   std::atomic<bool> pgmap_ready;
 103   std::set<int32_t> reported_osds;
 104   void maybe_ready(int32_t osd_id);
 105
 106   SafeTimer timer;
 107   bool shutting_down;
 108   Context *tick_event;
 109   void tick();
 110   void schedule_tick_locked(double delay_sec);
 111
 112   class OSDPerfMetricCollectorListener : public MetricListener {
 113   public:
 114     OSDPerfMetricCollectorListener(DaemonServer *server)
 115       : server(server) {
 116     }
 117     void handle_query_updated() override {
 118       server->handle_osd_perf_metric_query_updated();
 119     }
 120   private:
 121     DaemonServer *server;
 122   };
 123   OSDPerfMetricCollectorListener osd_perf_metric_collector_listener;
 124   OSDPerfMetricCollector osd_perf_metric_collector;
 125   void handle_osd_perf_metric_query_updated();
 126
 127   void handle_metric_payload(const OSDMetricPayload &payload) {
 128     osd_perf_metric_collector.process_reports(payload);
 129   }
 130
 131   void handle_metric_payload(const UnknownMetricPayload &payload) {
 132     ceph_abort();
 133   }
 134
 135   struct HandlePayloadVisitor : public boost::static_visitor<void> {
 136     DaemonServer *server;
 137
 138     HandlePayloadVisitor(DaemonServer *server)
 139       : server(server) {
 140     }
 141
 142     template <typename MetricPayload>
 143     inline void operator()(const MetricPayload &payload) const {
 144       server->handle_metric_payload(payload);
 145     }
 146   };
 147
 148   void update_task_status(DaemonKey key,
 149                           const std::map<std::string,std::string>& task_status);
 150
 151 public:
 152   int init(uint64_t gid, entity_addrvec_t client_addrs);
 153   void shutdown();
 154
 155   entity_addrvec_t get_myaddrs() const;
 156
 157   DaemonServer(MonClient *monc_,
 158                Finisher &finisher_,
 159                DaemonStateIndex &daemon_state_,
 160                ClusterState &cluster_state_,
 161                PyModuleRegistry &py_modules_,
 162                LogChannelRef cl,
 163                LogChannelRef auditcl);
 164   ~DaemonServer() override;
 165
 166   bool ms_dispatch2(const ceph::ref_t<Message>& m) override;
 167   int ms_handle_authentication(Connection *con) override;
 168   bool ms_handle_reset(Connection *con) override;
 169   void ms_handle_remote_reset(Connection *con) override {}
 170   bool ms_handle_refused(Connection *con) override;
 171
 172   void fetch_missing_metadata(const DaemonKey& key, const entity_addr_t& addr);
 173   bool handle_open(const ceph::ref_t<MMgrOpen>& m);
 174   bool handle_close(const ceph::ref_t<MMgrClose>& m);
 175   bool handle_report(const ceph::ref_t<MMgrReport>& m);
 176   bool handle_command(const ceph::ref_t<MCommand>& m);
 177   bool handle_command(const ceph::ref_t<MMgrCommand>& m);
 178   bool _handle_command(std::shared_ptr<CommandContext>& cmdctx);
 179   void send_report();
 180   void got_service_map();
 181   void got_mgr_map();
 182   void adjust_pgs();
 183
 184   void _send_configure(ConnectionRef c);
 185
 186   MetricQueryID add_osd_perf_query(
 187       const OSDPerfMetricQuery &query,
 188       const std::optional<OSDPerfMetricLimit> &limit);
 189   int remove_osd_perf_query(MetricQueryID query_id);
 190   int get_osd_perf_counters(MetricQueryID query_id,
 191                             std::map<OSDPerfMetricKey, PerformanceCounters> *c);
 192
 193   virtual const char** get_tracked_conf_keys() const override;
 194   virtual void handle_conf_change(const ConfigProxy& conf,
 195                           const std::set <std::string> &changed) override;
 196
 197   void schedule_tick(double delay_sec);
 198
 199   void log_access_denied(std::shared_ptr<CommandContext>& cmdctx,
 200                          MgrSession* session, std::stringstream& ss);
 201   void dump_pg_ready(ceph::Formatter *f);
 202 };
 203
 204 #endif
 205