text: /* Copyright (c) 2019 vesoft inc. All rights reserved.
   *
   * This source code is licensed under Apache 2.0 License,
   * attached with Common Clause Condition 1.0, found in the LICENSES directory.
   */
  
  #include "meta/processors/admin/SnapShot.h"
  #include <cstdlib>
  #include "meta/processors/Common.h"
  #include "meta/ActiveHostsMan.h"
  #include "meta/MetaServiceUtils.h"
  #include "network/NetworkUtils.h"
  
  namespace nebula {
  namespace meta {
  cpp2::ErrorCode Snapshot::createSnapshot(const std::string& name) {
      auto retSpacesHosts = getSpacesHosts();
      if (!retSpacesHosts.ok()) {
          return cpp2::ErrorCode::E_STORE_FAILURE;
      }
      auto spacesHosts = retSpacesHosts.value();
      for (const auto& spaceHosts : spacesHosts) {
          for (const auto& host : spaceHosts.second) {
              auto status = client_->createSnapshot(spaceHosts.first, name, host).get();
              if (!status.ok()) {
                  return cpp2::ErrorCode::E_RPC_FAILURE;
              }
          }
      }
      return cpp2::ErrorCode::SUCCEEDED;
  }
  
  cpp2::ErrorCode Snapshot::dropSnapshot(const std::string& name,
                                         const std::vector<HostAddr>& hosts) {
      auto retSpacesHosts = getSpacesHosts();
      if (!retSpacesHosts.ok()) {
          return cpp2::ErrorCode::E_STORE_FAILURE;
      }
      auto spacesHosts = retSpacesHosts.value();
      for (const auto& spaceHosts : spacesHosts) {
          for (const auto& host : spaceHosts.second) {
              if (std::find(hosts.begin(), hosts.end(), host) != hosts.end()) {
                  auto status = client_->dropSnapshot(spaceHosts.first, name, host).get();
                  if (!status.ok()) {
                      auto msg = "failed drop checkpoint : \"%s\". on host %s. error %s";
                      auto error = folly::stringPrintf(msg,
                                                       name.c_str(),
                                                       network::NetworkUtils::toHosts({host}).c_str(),
                                                       status.toString().c_str());
                      LOG(ERROR) << error;
                  }
              }
          }
      }
      return cpp2::ErrorCode::SUCCEEDED;
  }
  
  cpp2::ErrorCode Snapshot::blockingWrites(storage::cpp2::EngineSignType sign) {
      auto retSpacesHosts = getSpacesHosts();
      if (!retSpacesHosts.ok()) {
          return cpp2::ErrorCode::E_STORE_FAILURE;
      }
      auto spacesHosts = retSpacesHosts.value();
      for (const auto& spaceHosts : spacesHosts) {
          for (const auto& host : spaceHosts.second) {
          auto status = client_->blockingWrites(spaceHosts.first, sign, host).get();
          if (!status.ok()) {
              LOG(ERROR) << " Send blocking sign error on host : "
                         << network::NetworkUtils::toHosts({host});
          }
          }
      }
      return cpp2::ErrorCode::SUCCEEDED;
  }
  
  StatusOr<std::map<GraphSpaceID, std::set<HostAddr>>> Snapshot::getSpacesHosts() {
      folly::SharedMutex::ReadHolder rHolder(LockUtils::spaceLock());
      std::map<GraphSpaceID, std::set<HostAddr>> hostsByspaces;
      auto prefix = MetaServiceUtils::partPrefix();
      std::unique_ptr<kvstore::KVIterator> iter;
      auto kvRet = kv_->prefix(kDefaultSpaceId, kDefaultPartId, prefix, &iter);
      if (kvRet != kvstore::ResultCode::SUCCEEDED) {
          LOG(ERROR) << "Get hosts meta data error";
          return Status::Error("Get hosts meta data error");
      }
      while (iter->valid()) {
          auto partHosts = MetaServiceUtils::parsePartVal(iter->val());
          auto space = MetaServiceUtils::parsePartKeySpaceId(iter->key());
          for (auto& ph : partHosts) {
              hostsByspaces[space].emplace(HostAddr(ph.get_ip(), ph.get_port()));
          }
          iter->next();
      }
      return hostsByspaces;
  }
  
  }  // namespace meta
  }  // namespace nebula
  
  