Pause pool rebalance when monitor detects that it can lead to any OSD becoming full

src/client/etcd_state_client.cpp

@@ -785,7 +785,7 @@ void etcd_state_client_t::parse_state(const etcd_kv_t & kv)
         }
         for (auto & pool_item: value.object_items())
         {
-            pool_config_t pc;
+            pool_config_t pc = {};
             // ID
             pool_id_t pool_id;
             char null_byte = 0;
@@ -931,12 +931,28 @@ void etcd_state_client_t::parse_state(const etcd_kv_t & kv)
             // Ignore old key if the new one is present
             return;
         }
+        for (auto & pool_id_json: value["backfillfull_pools"].array_items())
+        {
+            auto pool_id = pool_id_json.uint64_value();
+            auto pool_it = this->pool_config.find(pool_id);
+            if (pool_it != this->pool_config.end())
+            {
+                pool_it->second.backfillfull |= 2;
+            }
+        }
         for (auto & pool_item: this->pool_config)
         {
             for (auto & pg_item: pool_item.second.pg_config)
             {
                 pg_item.second.config_exists = false;
             }
+            // 3 = was 1 and became 1, 0 = was 0 and became 0
+            if (pool_item.second.backfillfull == 2 || pool_item.second.backfillfull == 1)
+            {
+                if (on_change_backfillfull_hook)
+                    on_change_backfillfull_hook(pool_item.first);
+            }
+            pool_item.second.backfillfull = pool_item.second.backfillfull >> 1;
         }
         for (auto & pool_item: value["items"].object_items())
         {

src/client/etcd_state_client.h

@@ -62,6 +62,7 @@ struct pool_config_t
     std::map<pg_num_t, pg_config_t> pg_config;
     uint64_t scrub_interval;
     std::string used_for_fs;
+    int backfillfull;
 };
 
 struct inode_config_t
@@ -131,6 +132,7 @@ public:
     std::function<json11::Json()> load_pgs_checks_hook;
     std::function<void(bool)> on_load_pgs_hook;
     std::function<void()> on_change_pool_config_hook;
+    std::function<void(pool_id_t)> on_change_backfillfull_hook;
     std::function<void(pool_id_t, pg_num_t, osd_num_t)> on_change_pg_state_hook;
     std::function<void(pool_id_t, pg_num_t)> on_change_pg_history_hook;
     std::function<void(osd_num_t)> on_change_osd_state_hook;

src/osd/osd.h

@@ -226,6 +226,7 @@ class osd_t
     void parse_config(bool init);
     void init_cluster();
     void on_change_osd_state_hook(osd_num_t peer_osd);
+    void on_change_backfillfull_hook(pool_id_t pool_id);
     void on_change_pg_history_hook(pool_id_t pool_id, pg_num_t pg_num);
     void on_change_etcd_state_hook(std::map<std::string, etcd_kv_t> & changes);
     void on_load_config_hook(json11::Json::object & changes);

src/osd/osd_cluster.cpp

@@ -65,6 +65,7 @@ void osd_t::init_cluster()
         st_cli.tfd = tfd;
         st_cli.log_level = log_level;
         st_cli.on_change_osd_state_hook = [this](osd_num_t peer_osd) { on_change_osd_state_hook(peer_osd); };
+        st_cli.on_change_backfillfull_hook = [this](pool_id_t pool_id) { on_change_backfillfull_hook(pool_id); };
         st_cli.on_change_pg_history_hook = [this](pool_id_t pool_id, pg_num_t pg_num) { on_change_pg_history_hook(pool_id, pg_num); };
         st_cli.on_change_hook = [this](std::map<std::string, etcd_kv_t> & changes) { on_change_etcd_state_hook(changes); };
         st_cli.on_load_config_hook = [this](json11::Json::object & cfg) { on_load_config_hook(cfg); };
@@ -414,6 +415,14 @@ void osd_t::on_change_osd_state_hook(osd_num_t peer_osd)
     }
 }
 
+void osd_t::on_change_backfillfull_hook(pool_id_t pool_id)
+{
+    if (!(peering_state & (OSD_RECOVERING | OSD_FLUSHING_PGS)))
+    {
+        peering_state = peering_state | OSD_RECOVERING;
+    }
+}
+
 void osd_t::on_change_etcd_state_hook(std::map<std::string, etcd_kv_t> & changes)
 {
     if (changes.find(st_cli.etcd_prefix+"/config/global") != changes.end())

src/osd/osd_flush.cpp

@@ -252,10 +252,18 @@ bool osd_t::pick_next_recovery(osd_recovery_op_t &op)
             auto mask = recovery_last_degraded ? (PG_ACTIVE | PG_HAS_DEGRADED) : (PG_ACTIVE | PG_DEGRADED | PG_HAS_MISPLACED);
             auto check = recovery_last_degraded ? (PG_ACTIVE | PG_HAS_DEGRADED) : (PG_ACTIVE | PG_HAS_MISPLACED);
             // Restart scanning from the same PG as the last time
+        restart:
             for (auto pg_it = pgs.lower_bound(recovery_last_pg); pg_it != pgs.end(); pg_it++)
             {
                 if ((pg_it->second.state & mask) == check)
                 {
+                    auto pool_it = st_cli.pool_config.find(pg_it->first.pool_id);
+                    if (pool_it != st_cli.pool_config.end() && pool_it->second.backfillfull)
+                    {
+                        // Skip the pool
+                        recovery_last_pg.pool_id++;
+                        goto restart;
+                    }
                     auto & src = recovery_last_degraded ? pg_it->second.degraded_objects : pg_it->second.misplaced_objects;
                     assert(src.size() > 0);
                     // Restart scanning from the next object