zk_nn_client.h

// Copyright 2017 Rice University, COMP 413 2017

#ifndef ZOOKEEPER_INCLUDE_ZK_NN_CLIENT_H_
#define ZOOKEEPER_INCLUDE_ZK_NN_CLIENT_H_

#define MIN_XMITS 'x'
#define MAX_FREE_SPACE 'f'

#include "zk_client_common.h"
#include <google/protobuf/message.h>
#include <queue>
#include <string>
#include <vector>
#include <utility>
#include "hdfs.pb.h"
#include "ClientNamenodeProtocol.pb.h"
#include "erasurecoding.pb.h"
#include <ConfigReader.h>
#include "util.h"
#include "LRUCache.h"

#define MAX_USERNAME_LEN 256

namespace zkclient {

typedef enum class FileStatus : int {
  UnderConstruction,
  FileComplete,
  UnderDestruction
} FileStatus;

typedef struct {
  uint32_t replication;  // the block replication factor.
  bool isEC;  // 1 if EC file. 0 if replication based.
  uint64_t blocksize;
  // 1 for under construction, 0 for complete
  zkclient::FileStatus under_construction;
  int filetype;  // 0 or 1 for dir, 2 for file, 3 for symlinks (not supported)
  std::uint64_t length;
// https://hadoop.apache.org/docs/r2.4.1/api/org/apache/hadoop/fs/
// FileSystem.html#setOwner(org.apache.hadoop.fs.Path,
// java.lang.String,
// java.lang.String)
  std::uint64_t access_time;
  std::uint64_t modification_time;
  char owner[MAX_USERNAME_LEN];  // the client who created the file
  char group[MAX_USERNAME_LEN];
  char permissions[20][MAX_USERNAME_LEN];  // max 20 users can view the file.
  int perm_length;  // number of slots filled in permissions
  int permission_number;
  std::uint64_t last_block_id;
} FileZNode;

typedef struct {
  std::string clientName;   // Same as the
                            // clientName passed
                            // from RenewLeaseRequestProto
} LeaseInfo;

typedef struct {
  uint64_t timestamp;    // std::time()
} ClientInfo;

struct TargetDN {
  char policy;
  std::string dn_id;
  uint64_t free_bytes;    // free space on disk
  uint32_t num_xmits;        // current number of xmits

  TargetDN(std::string id, int bytes, int xmits, char policy) : policy(policy),
                                  dn_id(id),
                                  free_bytes(bytes),
                                  num_xmits(xmits) {
  }

  bool operator<(const struct TargetDN &other) const {
    // If storage policy is 'x' for xmits, choose the min xmits node
    if (policy == MIN_XMITS) {
    if (num_xmits == other.num_xmits) {
      return free_bytes < other.free_bytes;
    }
    return num_xmits > other.num_xmits;

      // Default policy is choose the node with the most free space
    } else {
    if (free_bytes == other.free_bytes) {
      return num_xmits > other.num_xmits;
    }
    return free_bytes < other.free_bytes;
    }
  }
};

using hadoop::hdfs::AddBlockRequestProto;
using hadoop::hdfs::AddBlockResponseProto;
using hadoop::hdfs::AbandonBlockRequestProto;
using hadoop::hdfs::AbandonBlockResponseProto;
using hadoop::hdfs::ExtendedBlockProto;
using hadoop::hdfs::GetFileInfoRequestProto;
using hadoop::hdfs::GetFileInfoResponseProto;
using hadoop::hdfs::HdfsFileStatusProto;
using hadoop::hdfs::DeleteRequestProto;
using hadoop::hdfs::DeleteResponseProto;
using hadoop::hdfs::CreateRequestProto;
using hadoop::hdfs::CreateResponseProto;
using hadoop::hdfs::CompleteRequestProto;
using hadoop::hdfs::CompleteResponseProto;
using hadoop::hdfs::RenameRequestProto;
using hadoop::hdfs::RenameResponseProto;
using hadoop::hdfs::MkdirsRequestProto;
using hadoop::hdfs::MkdirsResponseProto;
using hadoop::hdfs::GetListingRequestProto;
using hadoop::hdfs::GetListingResponseProto;
using hadoop::hdfs::LocatedBlocksProto;
using hadoop::hdfs::GetBlockLocationsRequestProto;
using hadoop::hdfs::GetBlockLocationsResponseProto;
using hadoop::hdfs::LocatedBlockProto;
using hadoop::hdfs::GetContentSummaryRequestProto;
using hadoop::hdfs::GetContentSummaryResponseProto;
using hadoop::hdfs::ContentSummaryProto;
using hadoop::hdfs::DatanodeInfoProto;
using hadoop::hdfs::SetOwnerRequestProto;
using hadoop::hdfs::SetOwnerResponseProto;
using hadoop::hdfs::RenameResponseProto;
using hadoop::hdfs::SetPermissionRequestProto;
using hadoop::hdfs::SetPermissionResponseProto;
using hadoop::hdfs::AppendRequestProto;
using hadoop::hdfs::AppendResponseProto;
using hadoop::hdfs::RenewLeaseRequestProto;
using hadoop::hdfs::RenewLeaseResponseProto;
using hadoop::hdfs::RecoverLeaseRequestProto;
using hadoop::hdfs::RecoverLeaseResponseProto;
using hadoop::hdfs::ErasureCodingPolicyProto;
using hadoop::hdfs::ECSchemaProto;
using hadoop::hdfs::GetErasureCodingPoliciesRequestProto;
using hadoop::hdfs::GetErasureCodingPoliciesResponseProto;
using hadoop::hdfs::GetErasureCodingPolicyRequestProto;
using hadoop::hdfs::GetErasureCodingPolicyResponseProto;
using hadoop::hdfs::SetErasureCodingPolicyResponseProto;
using hadoop::hdfs::SetErasureCodingPolicyRequestProto;
using hadoop::hdfs::UpdateBlockForPipelineRequestProto;
using hadoop::hdfs::UpdateBlockForPipelineResponseProto;

class ZkNnClient : public ZkClientCommon {
 public:
  char policy;
  const char* EC_REPLICATION = "replication";
  const char* DEFAULT_EC_POLICY = "RS-6-3-1024k";  // the default policy.
  uint32_t DEFAULT_EC_CELLSIZE = 1024*1024;  // the default cell size is 64kb.
  uint32_t DEFAULT_EC_ID = 1;
  uint32_t REPLICATION_EC_ID = 63;
  const uint32_t DEFAULT_DATA_UNITS = 6;
  const uint32_t DEFAULT_PARITY_UNITS = 3;
  const char* DEFAULT_EC_CODEC_NAME = "rs";
  std::string DEFAULT_STORAGE_ID = "1";  // the default storage id.
  std::string REPLICATION_STORAGE_ID = "63";
  /*
   * TODO(2017) For some reason, these fields can't be properly
   * reclaimed/destructed in some cases.
   */
  ECSchemaProto DEFAULT_EC_SCHEMA;
  ErasureCodingPolicyProto RS_SOLOMON_PROTO;
  ErasureCodingPolicyProto REPLICATION_PROTO;
  ECSchemaProto REPLICATION_1_2_SCHEMA;


  enum class ListingResponse {
    Ok,                    // 0
    FileDoesNotExist,      // 1
    FailedChildRetrieval,  // 2
    FileAccessRestricted   // 3
  };

  enum class DeleteResponse {
    Ok,
    FileDoesNotExist,
    FileUnderConstruction,
    FileIsDirectoryMismatch,
    FailedChildRetrieval,
    FailedBlockRetrieval,
    FailedDataNodeRetrieval,
    FailedZookeeperOp,
    FileAccessRestricted
  };

  enum class GetFileInfoResponse {
    Ok,
    FileDoesNotExist,
    FailedReadZnode,
    FileAccessRestricted
  };

  enum class MkdirResponse {
    Ok,
    FailedZnodeCreation
  };

  enum class CreateResponse {
    Ok,
    FileAlreadyExists,
    FailedMkdir,
    FailedCreateZnode
  };

  enum class RenameResponse {
    Ok,
    FileDoesNotExist,
    RenameOpsFailed,
    InvalidType,
    MultiOpFailed,
    FileAccessRestricted
  };

  enum class ErasureCodingPoliciesResponse {
      Ok
  };

  enum class ErasureCodingPolicyResponse {
      Ok,
      FileDoesNotExist
  };

  enum class SetErasureCodingPolicyResponse {
      Ok,
      FileDoesNotExist,
      FailedZookeeperOp
  };

  explicit ZkNnClient(std::string zkIpAndAddress)
                        : ZkClientCommon(zkIpAndAddress),
                          cache(new lru::Cache<std::string,
                          std::shared_ptr<GetListingResponseProto>>(64, 10)) {
    mkdir_helper("/", false);
    populateDefaultECProto();
  }

  explicit ZkNnClient(std::shared_ptr<ZKWrapper> zk_in,
                      bool secureMode = false)
                        : ZkClientCommon(zk_in),
                          cache(new lru::Cache<std::string,
                          std::shared_ptr<GetListingResponseProto>>(64, 10)) {
    mkdir_helper("/", false);
    isSecureMode = secureMode;
    populateDefaultECProto();
  }
  void register_watches();


  uint64_t current_time_ms();

  uint64_t get_client_lease_timestamp(std::string client_name);

  void renew_lease(RenewLeaseRequestProto &req,
                   RenewLeaseResponseProto &res);

  void recover_lease(RecoverLeaseRequestProto &req,
                     RecoverLeaseResponseProto &res);
  GetFileInfoResponse get_info(GetFileInfoRequestProto &req,
                               GetFileInfoResponseProto &res,
                               std::string client_name = "default");

  CreateResponse create_file(CreateRequestProto &request,
                             CreateResponseProto &response);

  void get_block_locations(GetBlockLocationsRequestProto &req,
                           GetBlockLocationsResponseProto &res,
                           std::string client_name = "default");
  MkdirResponse mkdir(MkdirsRequestProto &req,
                      MkdirsResponseProto &res);

  DeleteResponse destroy(DeleteRequestProto &req,
                         DeleteResponseProto &res,
                         std::string client_name = "default");

  void complete(CompleteRequestProto &req,
                CompleteResponseProto &res,
                std::string client_name = "default");

  RenameResponse rename(RenameRequestProto &req,
                        RenameResponseProto &res,
                        std::string client_name = "default");

  ListingResponse get_listing(GetListingRequestProto &req,
                              GetListingResponseProto &res,
                              std::string client_name = "default");
  void get_content(GetContentSummaryRequestProto &req,
                   GetContentSummaryResponseProto &res,
                   std::string client_name = "default");

  void set_file_info_content(ContentSummaryProto *status,
                             const std::string &path,
                             FileZNode &znode_data);

  void set_node_policy(char policy);

  char get_node_policy();

  ErasureCodingPoliciesResponse get_erasure_coding_policies(
      GetErasureCodingPoliciesRequestProto &req,
      GetErasureCodingPoliciesResponseProto &res);

  ErasureCodingPolicyResponse get_erasure_coding_policy_of_path(
      GetErasureCodingPolicyRequestProto &req,
      GetErasureCodingPolicyResponseProto &res);

  SetErasureCodingPolicyResponse set_erasure_coding_policy_of_path(
      SetErasureCodingPolicyRequestProto &req,
      SetErasureCodingPolicyResponseProto &res);

//  bool modifyAclEntries(ModifyAclEntriesRequestProto req,
//                        ModifyAclEntriesResponseProto res);
  bool set_permission(SetPermissionRequestProto &req,
                      SetPermissionResponseProto &res);

  bool set_owner(SetOwnerRequestProto &req,
                 SetOwnerResponseProto &res,
                 std::string client_name = "default");
  bool add_block(AddBlockRequestProto &req,
                 AddBlockResponseProto &res,
                 std::string client_name = "default");
  /*
   * Sets the owner of the file.
   * @param req SetOwnerRequestProto
   * @param res SetOwnerResponseProto
   * @return boolean indicating whether operation succeeded or not
   */
  bool set_owner(SetOwnerRequestProto &req, SetOwnerResponseProto &res);

  bool add_block(const std::string &fileName,
                 u_int64_t &block_id,
                 std::vector<std::string> &dataNodes,
                 uint32_t replication_factor);

  bool add_block_group(const std::string &filePath,
                       u_int64_t &block_group_id,
                       std::vector<std::string> &dataNodes,
                       std::vector<char> &blockIndices,
                       uint32_t total_num_storage_blocks);

  uint32_t get_total_num_storage_blocks(
          const std::string &fileName,
          u_int64_t &block_group_id);

  u_int64_t generate_storage_block_id(
          uint64_t block_group_id,
          uint64_t index_within_group);
  u_int64_t generate_block_group_id();

  u_int64_t get_block_group_id(u_int64_t storage_block_id);

  u_int64_t get_index_within_block_group(u_int64_t storage_block_id);

  bool abandon_block(AbandonBlockRequestProto &req,
                     AbandonBlockResponseProto &res,
                     std::string client_name = "default");

  bool previousBlockComplete(uint64_t prev_id);
  bool file_exists(const std::string &path);

  bool get_block_size(const u_int64_t &block_id, uint64_t &blocksize);

  // this is public because we have not member functions in this file
  static const std::string CLASS_NAME;

  bool find_live_datanodes(const uint64_t blockId, int error_code,
                           std::vector<std::string> &live_data_nodes);

  bool find_datanode_for_block(std::vector<std::string> &datanodes,
                                        std::vector<std::string> &excluded_dns,
                                         const u_int64_t blockId,
                                         uint32_t replication_factor,
                                         uint64_t blocksize);

  bool find_all_datanodes_with_block(const uint64_t &block_uuid,
                                     std::vector<std::string> &rdatanodes,
                                     int &error_code);

  bool rename_ops_for_file(const std::string &src, const std::string &dst,
                           std::vector<std::shared_ptr<ZooOp>> &ops);
  bool rename_ops_for_dir(const std::string &src, const std::string &dst,
                          std::vector<std::shared_ptr<ZooOp>> &ops);

  bool check_acks();

// get locations given src, offset, and length
  void get_block_locations(const std::string &src,
                           google::protobuf::uint64 offset,
                           google::protobuf::uint64 length,
                           LocatedBlocksProto *blocks,
                           std::string client_name = "default");

  bool append_file(AppendRequestProto &req, AppendResponseProto &res);
  void update_block_for_pipeline(UpdateBlockForPipelineRequestProto &req,
                                 UpdateBlockForPipelineResponseProto &res);
  bool process_request(std::string client_name, std::string file_path,
                       AppendRequestProto &req);
  bool get_primary_block_info(std::string file_path,
                                     AppendRequestProto &req,
                                     AppendResponseProto &res);
  bool check_lease(std::string client_name, std::string file_path);

  void read_file_znode(FileZNode &znode_data, const std::string &path);

  bool cache_contains(const std::string &path);

  int cache_size();

  std::string find_parent(const std::string &path);

 private:
  bool set_located_block(LocatedBlockProto* locatedBlockProto,
                         uint64_t block_id, uint64_t block_size);

  bool set_file_status(std::string file_path,
                        HdfsFileStatusProto* hdfsFileStatusProto);

  void recover_lease_helper(RecoverLeaseRequestProto &req,
                     RecoverLeaseResponseProto &res);

  void renew_lease_helper(RenewLeaseRequestProto &req,
                          RenewLeaseResponseProto &res);

  bool sort_by_xmits(const std::vector<std::string> &unsorted_dn_ids,
                       std::vector<std::string> &sorted_dn_ids);

  void set_file_info(HdfsFileStatusProto *fs,
                     const std::string &path,
                     FileZNode &node);
  std::string ClientZookeeperPath(const std::string & clientname);
  std::string LeaseZookeeperPath(const std::string & hadoopPath);

  std::string ZookeeperBlocksPath(const std::string &hadoopPath);

  std::string ZookeeperFilePath(const std::string &hadoopPath);

  config_reader::ConfigReader config;

  bool create_file_znode(const std::string &path, FileZNode *znode_data);

  void set_mkdir_znode(FileZNode *znode_data);
  MkdirResponse mkdir_helper(const std::string &path, bool create_parent);

  void file_znode_struct_to_vec(FileZNode *znode_data,
                                std::vector<std::uint8_t> &data);
  template <class T>
  void znode_data_to_vec(T *znode_data, std::vector<std::uint8_t> &data);
  template <class T>
  void read_znode_data(T &znode_data, const std::string &path);

  void delete_node_wrapper(std::string &path,
                           DeleteResponseProto &response);

  DeleteResponse destroy_helper(const std::string &path,
                      std::vector<std::shared_ptr<ZooOp>> &ops);
  bool recover_ec_blocks(const std::vector<std::string> &to_ec_recover,
                        int error_code);


  bool replicate_blocks(const std::vector<std::string> &to_replicate,
                        int error_code);

  int ms_since_creation(std::string &path);

// bool updateLocatedBlockProto(LocatedBlockProto* location,
//                              uint64_t block_id);

  bool buildDatanodeInfoProto(DatanodeInfoProto *dn_info,
                              const std::string &data_node);

  bool buildTokenProto(hadoop::common::TokenProto *token);

  bool buildExtendedBlockProto(ExtendedBlockProto *eb,
                               const std::uint64_t &block_id,
                               const uint64_t &block_size);

  static void watcher_health(zhandle_t *zzh, int type, int state,
                             const char *path, void *watcherCtx);

  static void watcher_health_child(zhandle_t *zzh, int type, int state,
                                   const char *path, void *watcherCtx);

  static void watcher_listing(zhandle_t *zzh, int type, int state,
                              const char *path, void *watcherCtx);

  bool lease_expired(std::string lease_holder_client);

  bool blockDeleted(uint64_t uuid, std::string id);

  void populateDefaultECProto();

  bool checkAccess(std::string username, FileZNode &znode_data);

  const int UNDER_CONSTRUCTION = 1;
  const int FILE_COMPLETE = 0;
  const int UNDER_DESTRUCTION = 2;

  const int IS_FILE = 2;
  const int IS_DIR = 1;
// TODO(2016): Should eventually be read from a conf file
// in millisecons, 10 minute timeout when waiting for
// replication acknowledgements
  const int ACK_TIMEOUT = 600000;

// Boolean indicating whether zk_nn is in secure mode
  bool isSecureMode = false;
  const uint64_t EXPIRATION_TIME =
    2 * 60 * 60 * 1000;  // 2 hours in milliseconds.

  /* TODO(2017) This needs to be reclaimed properly. */
  lru::Cache<std::string, std::shared_ptr<GetListingResponseProto>> *cache;
};

}  // namespace zkclient

#endif  // ZOOKEEPER_INCLUDE_ZK_NN_CLIENT_H_