/*
   Copyright (c) 2014 Red Hat, Inc. <http://www.redhat.com>
   This file is part of GlusterFS.

   This file is licensed to you under your choice of the GNU Lesser
   General Public License, version 3 or any later version (LGPLv3 or
   later), or the GNU General Public License, version 2 (GPLv2), in all
   cases as published by the Free Software Foundation.
*/

#include "glusterd-peer-utils.h"
#include "glusterd-store.h"
#include "glusterd-server-quorum.h"
#include "glusterd-messages.h"
#include <glusterfs/common-utils.h>
#include "glusterd-utils.h"

void
glusterd_peerinfo_destroy(struct rcu_head *head)
{
    int32_t ret = -1;
    glusterd_peerinfo_t *peerinfo = NULL;
    glusterd_peer_hostname_t *hostname = NULL;
    glusterd_peer_hostname_t *tmp = NULL;

    /* This works as rcu_head is the first member of gd_rcu_head */
    peerinfo = caa_container_of((gd_rcu_head *)head, glusterd_peerinfo_t,
                                rcu_head);

    /* Set THIS to the saved this. Needed by some functions below */
    THIS = peerinfo->rcu_head.this;

    CDS_INIT_LIST_HEAD(&peerinfo->uuid_list);

    ret = glusterd_store_delete_peerinfo(peerinfo);
    if (ret) {
        gf_msg("glusterd", GF_LOG_ERROR, errno, GD_MSG_PEERINFO_DELETE_FAIL,
               "Deleting peer info failed");
    }

    GF_FREE(peerinfo->hostname);
    peerinfo->hostname = NULL;

    cds_list_for_each_entry_safe(hostname, tmp, &peerinfo->hostnames,
                                 hostname_list)
    {
        glusterd_peer_hostname_free(hostname);
    }

    glusterd_sm_tr_log_delete(&peerinfo->sm_log);
    pthread_mutex_unlock(&peerinfo->delete_lock);
    pthread_mutex_destroy(&peerinfo->delete_lock);
    GF_FREE(peerinfo);

    peerinfo = NULL;

    return;
}

int32_t
glusterd_peerinfo_cleanup(glusterd_peerinfo_t *peerinfo)
{
    GF_ASSERT(peerinfo);
    gf_boolean_t quorum_action = _gf_false;
    glusterd_conf_t *priv = THIS->private;

    if (pthread_mutex_trylock(&peerinfo->delete_lock)) {
        /* Someone else is already deleting the peer, so give up */
        return 0;
    }

    if (peerinfo->quorum_contrib != QUORUM_NONE)
        quorum_action = _gf_true;
    if (peerinfo->rpc) {
        peerinfo->rpc = glusterd_rpc_clnt_unref(priv, peerinfo->rpc);
        peerinfo->rpc = NULL;
    }

    cds_list_del_rcu(&peerinfo->uuid_list);
    /* Saving THIS, as it is needed by the callback function */
    peerinfo->rcu_head.this = THIS;
    call_rcu(&peerinfo->rcu_head.head, glusterd_peerinfo_destroy);

    if (quorum_action)
        /* coverity[SLEEP] */
        glusterd_do_quorum_action();
    return 0;
}

/* gd_peerinfo_find_from_hostname iterates over all the addresses saved for each
 * peer and matches it to @hoststr.
 * Returns the matched peer if found else returns NULL
 */
static glusterd_peerinfo_t *
gd_peerinfo_find_from_hostname(const char *hoststr)
{
    xlator_t *this = THIS;
    glusterd_conf_t *priv = NULL;
    glusterd_peerinfo_t *peer = NULL;
    glusterd_peerinfo_t *found = NULL;
    glusterd_peer_hostname_t *tmphost = NULL;

    GF_ASSERT(this != NULL);
    priv = this->private;
    GF_VALIDATE_OR_GOTO(this->name, (priv != NULL), out);

    GF_VALIDATE_OR_GOTO(this->name, (hoststr != NULL), out);

    RCU_READ_LOCK;
    cds_list_for_each_entry_rcu(peer, &priv->peers, uuid_list)
    {
        cds_list_for_each_entry_rcu(tmphost, &peer->hostnames, hostname_list)
        {
            if (!strncasecmp(tmphost->hostname, hoststr, 1024)) {
                gf_msg_debug(this->name, 0, "Friend %s found.. state: %d",
                             tmphost->hostname, peer->state.state);
                found = peer; /* Probably needs to be
                                 dereferenced*/
                goto unlock;
            }
        }
    }
unlock:
    RCU_READ_UNLOCK;
out:
    return found;
}

/* gd_peerinfo_find_from_addrinfo iterates over all the addresses saved for each
 * peer, resolves them and compares them to @addr.
 *
 *
 * NOTE: As getaddrinfo is a blocking call and is being performed multiple times
 * in this function, it could lead to the calling thread to be blocked for
 * significant amounts of time.
 *
 * Returns the matched peer if found else returns NULL
 */
static glusterd_peerinfo_t *
gd_peerinfo_find_from_addrinfo(const struct addrinfo *addr)
{
    xlator_t *this = THIS;
    glusterd_conf_t *conf = NULL;
    glusterd_peerinfo_t *peer = NULL;
    glusterd_peerinfo_t *found = NULL;
    glusterd_peer_hostname_t *address = NULL;
    int ret = 0;
    struct addrinfo *paddr = NULL;
    struct addrinfo *tmp = NULL;

    GF_ASSERT(this != NULL);
    conf = this->private;
    GF_VALIDATE_OR_GOTO(this->name, (conf != NULL), out);

    RCU_READ_LOCK;
    cds_list_for_each_entry_rcu(peer, &conf->peers, uuid_list)
    {
        cds_list_for_each_entry_rcu(address, &peer->hostnames, hostname_list)
        {
            /* TODO: Cache the resolved addrinfos to improve
             * performance
             */
            ret = getaddrinfo(address->hostname, NULL, NULL, &paddr);
            if (ret) {
                /* Don't fail if getaddrinfo fails, continue
                 * onto the next address
                 */
                gf_msg_trace(this->name, 0, "getaddrinfo for %s failed (%s)",
                             address->hostname, gai_strerror(ret));
                continue;
            }

            for (tmp = paddr; tmp != NULL; tmp = tmp->ai_next) {
                if (gf_compare_sockaddr(addr->ai_addr, tmp->ai_addr)) {
                    found = peer; /* (de)referenced? */
                    break;
                }
            }

            freeaddrinfo(paddr);
            if (found)
                goto unlock;
        }
    }
unlock:
    RCU_READ_UNLOCK;
out:
    return found;
}

/* glusterd_peerinfo_find_by_hostname searches for a peer which matches the
 * hostname @hoststr and if found returns the pointer to peerinfo object.
 * Returns NULL otherwise.
 *
 * It first attempts a quick search by string matching @hoststr. If that fails,
 * it'll attempt a more thorough match by resolving the addresses and matching
 * the resolved addrinfos.
 */
glusterd_peerinfo_t *
glusterd_peerinfo_find_by_hostname(const char *hoststr)
{
    int ret = -1;
    struct addrinfo *addr = NULL;
    struct addrinfo *p = NULL;
    xlator_t *this = THIS;
    glusterd_peerinfo_t *peerinfo = NULL;

    GF_ASSERT(hoststr);

    peerinfo = gd_peerinfo_find_from_hostname(hoststr);
    if (peerinfo)
        return peerinfo;

    ret = getaddrinfo(hoststr, NULL, NULL, &addr);
    if (ret != 0) {
        gf_msg(this->name, GF_LOG_ERROR, ret, GD_MSG_GETADDRINFO_FAIL,
               "error in getaddrinfo: %s\n", gai_strerror(ret));
        goto out;
    }

    for (p = addr; p != NULL; p = p->ai_next) {
        peerinfo = gd_peerinfo_find_from_addrinfo(p);
        if (peerinfo) {
            freeaddrinfo(addr);
            return peerinfo;
        }
    }

out:
    gf_msg_debug(this->name, 0, "Unable to find friend: %s", hoststr);
    if (addr)
        freeaddrinfo(addr);
    return NULL;
}

int
glusterd_hostname_to_uuid(char *hostname, uuid_t uuid)
{
    GF_ASSERT(hostname);
    GF_ASSERT(uuid);

    glusterd_peerinfo_t *peerinfo = NULL;
    glusterd_conf_t *priv = NULL;
    int ret = -1;
    xlator_t *this = NULL;

    this = THIS;
    GF_ASSERT(this);
    priv = this->private;
    GF_ASSERT(priv);

    peerinfo = glusterd_peerinfo_find_by_hostname(hostname);
    if (peerinfo) {
        ret = 0;
        gf_uuid_copy(uuid, peerinfo->uuid);
    } else {
        if (gf_is_local_addr(hostname)) {
            gf_uuid_copy(uuid, MY_UUID);
            ret = 0;
        } else {
            ret = -1;
        }
    }

    gf_msg_debug(this->name, 0, "returning %d", ret);
    return ret;
}

/* glusterd_peerinfo_find_by_uuid searches for a peer which matches the
 * uuid @uuid and if found returns the pointer to peerinfo object.
 * Returns NULL otherwise.
 */
glusterd_peerinfo_t *
glusterd_peerinfo_find_by_uuid(uuid_t uuid)
{
    glusterd_conf_t *priv = NULL;
    glusterd_peerinfo_t *entry = NULL;
    glusterd_peerinfo_t *found = NULL;
    xlator_t *this = THIS;
    glusterd_friend_sm_state_t state;

    GF_ASSERT(this);

    if (gf_uuid_is_null(uuid))
        return NULL;

    priv = this->private;

    GF_ASSERT(priv);

    RCU_READ_LOCK;
    cds_list_for_each_entry_rcu(entry, &priv->peers, uuid_list)
    {
        if (!gf_uuid_compare(entry->uuid, uuid)) {
            found = entry; /* Probably should be rcu_dereferenced */
            state = found->state.state;
            break;
        }
    }
    RCU_READ_UNLOCK;

    if (found)
        gf_msg_debug(this->name, 0, "Friend found... state: %s",
                     glusterd_friend_sm_state_name_get(state));
    else
        gf_msg_debug(this->name, 0, "Friend with uuid: %s, not found",
                     uuid_utoa(uuid));
    return found;
}

/* glusterd_peerinfo_find will search for a peer matching either @uuid or
 * @hostname and return a pointer to the peerinfo object
 * Returns NULL otherwise.
 */
glusterd_peerinfo_t *
glusterd_peerinfo_find(uuid_t uuid, const char *hostname)
{
    glusterd_peerinfo_t *peerinfo = NULL;
    xlator_t *this = THIS;

    GF_ASSERT(this);

    if (uuid) {
        peerinfo = glusterd_peerinfo_find_by_uuid(uuid);

        if (peerinfo) {
            return peerinfo;
        } else {
            gf_msg_debug(this->name, 0, "Unable to find peer by uuid: %s",
                         uuid_utoa(uuid));
        }
    }

    if (hostname) {
        peerinfo = glusterd_peerinfo_find_by_hostname(hostname);

        if (peerinfo) {
            return peerinfo;
        } else {
            gf_msg_debug(this->name, 0, "Unable to find hostname: %s",
                         hostname);
        }
    }
    return NULL;
}

/* glusterd_peerinfo_new will create a new peerinfo object and set it's members
 * values using the passed parameters.
 * @hostname is added as the first entry in peerinfo->hostnames list and also
 * set to peerinfo->hostname.
 * It returns a pointer to peerinfo object if successful and returns NULL
 * otherwise. The caller should take care of freeing the created peerinfo
 * object.
 */
glusterd_peerinfo_t *
glusterd_peerinfo_new(glusterd_friend_sm_state_t state, uuid_t *uuid,
                      const char *hostname, int port)
{
    glusterd_peerinfo_t *new_peer = NULL;
    int ret = -1;
    xlator_t *this = NULL;
    glusterd_conf_t *conf = NULL;

    this = THIS;
    GF_ASSERT(this);
    conf = this->private;
    GF_ASSERT(conf);

    new_peer = GF_CALLOC(1, sizeof(*new_peer), gf_gld_mt_peerinfo_t);
    if (!new_peer)
        goto out;

    CDS_INIT_LIST_HEAD(&new_peer->uuid_list);

    new_peer->state.state = state;

    CDS_INIT_LIST_HEAD(&new_peer->hostnames);
    if (hostname) {
        ret = gd_add_address_to_peer(new_peer, hostname);
        if (ret)
            goto out;
        /* Also set it to peerinfo->hostname. Doing this as we use
         * peerinfo->hostname in a lot of places and is really hard to
         * get everything right
         */
        new_peer->hostname = gf_strdup(hostname);
    }

    if (uuid) {
        gf_uuid_copy(new_peer->uuid, *uuid);
    }

    ret = glusterd_sm_tr_log_init(
        &new_peer->sm_log, glusterd_friend_sm_state_name_get,
        glusterd_friend_sm_event_name_get, GLUSTERD_TR_LOG_SIZE);
    if (ret)
        goto out;

    if (new_peer->state.state == GD_FRIEND_STATE_BEFRIENDED)
        new_peer->quorum_contrib = QUORUM_WAITING;
    new_peer->port = port;

    pthread_mutex_init(&new_peer->delete_lock, NULL);

    new_peer->generation = uatomic_add_return(&conf->generation, 1);
out:
    if (ret && new_peer) {
        glusterd_peerinfo_cleanup(new_peer);
        new_peer = NULL;
    }
    return new_peer;
}

/* Check if the all peers are connected and befriended, except the peer
 * specified (the peer being detached)
 */
gf_boolean_t
glusterd_chk_peers_connected_befriended(uuid_t skip_uuid)
{
    gf_boolean_t ret = _gf_true;
    glusterd_peerinfo_t *peerinfo = NULL;
    glusterd_conf_t *priv = NULL;

    priv = THIS->private;
    GF_ASSERT(priv);

    RCU_READ_LOCK;
    cds_list_for_each_entry_rcu(peerinfo, &priv->peers, uuid_list)
    {
        if (!gf_uuid_is_null(skip_uuid) &&
            !gf_uuid_compare(skip_uuid, peerinfo->uuid))
            continue;

        if ((GD_FRIEND_STATE_BEFRIENDED != peerinfo->state.state) ||
            !(peerinfo->connected)) {
            ret = _gf_false;
            break;
        }
    }
    RCU_READ_UNLOCK;

    gf_msg_debug(THIS->name, 0, "Returning %s", (ret ? "TRUE" : "FALSE"));
    return ret;
}

/* Return hostname for given uuid if it exists
 * else return NULL
 */
char *
glusterd_uuid_to_hostname(uuid_t uuid)
{
    char *hostname = NULL;
    glusterd_conf_t *priv = NULL;
    glusterd_peerinfo_t *entry = NULL;

    priv = THIS->private;
    GF_ASSERT(priv);

    if (!gf_uuid_compare(MY_UUID, uuid)) {
        hostname = gf_strdup("localhost");
        return hostname;
    }
    RCU_READ_LOCK;
    if (!cds_list_empty(&priv->peers)) {
        cds_list_for_each_entry_rcu(entry, &priv->peers, uuid_list)
        {
            if (!gf_uuid_compare(entry->uuid, uuid)) {
                hostname = gf_strdup(entry->hostname);
                break;
            }
        }
    }
    RCU_READ_UNLOCK;

    return hostname;
}

char *
gd_peer_uuid_str(glusterd_peerinfo_t *peerinfo)
{
    if ((peerinfo == NULL) || gf_uuid_is_null(peerinfo->uuid))
        return NULL;

    if (peerinfo->uuid_str[0] == '\0')
        uuid_utoa_r(peerinfo->uuid, peerinfo->uuid_str);

    return peerinfo->uuid_str;
}

gf_boolean_t
glusterd_are_all_peers_up()
{
    glusterd_peerinfo_t *peerinfo = NULL;
    xlator_t *this = NULL;
    glusterd_conf_t *conf = NULL;
    gf_boolean_t peers_up = _gf_false;

    this = THIS;
    GF_VALIDATE_OR_GOTO("glusterd", this, out);

    conf = this->private;
    GF_VALIDATE_OR_GOTO(this->name, conf, out);

    RCU_READ_LOCK;
    cds_list_for_each_entry_rcu(peerinfo, &conf->peers, uuid_list)
    {
        if (!peerinfo->connected) {
            RCU_READ_UNLOCK;
            goto out;
        }
    }
    RCU_READ_UNLOCK;

    peers_up = _gf_true;

out:
    return peers_up;
}

gf_boolean_t
glusterd_are_vol_all_peers_up(glusterd_volinfo_t *volinfo,
                              struct cds_list_head *peers, char **down_peerstr)
{
    glusterd_peerinfo_t *peerinfo = NULL;
    glusterd_brickinfo_t *brickinfo = NULL;
    gf_boolean_t ret = _gf_false;

    cds_list_for_each_entry(brickinfo, &volinfo->bricks, brick_list)
    {
        if (!gf_uuid_compare(brickinfo->uuid, MY_UUID))
            continue;

        RCU_READ_LOCK;
        cds_list_for_each_entry_rcu(peerinfo, peers, uuid_list)
        {
            if (gf_uuid_compare(peerinfo->uuid, brickinfo->uuid))
                continue;

            /*Found peer who owns the brick, return false
             * if peer is not connected or not friend */
            if (!(peerinfo->connected) ||
                (peerinfo->state.state != GD_FRIEND_STATE_BEFRIENDED)) {
                *down_peerstr = gf_strdup(peerinfo->hostname);
                RCU_READ_UNLOCK;
                gf_msg_debug(THIS->name, 0, "Peer %s is down. ", *down_peerstr);
                goto out;
            }
        }
        RCU_READ_UNLOCK;
    }

    ret = _gf_true;
out:
    gf_msg_debug("glusterd", 0, "Returning %d", ret);
    return ret;
}

int32_t
glusterd_peer_hostname_new(const char *hostname,
                           glusterd_peer_hostname_t **name)
{
    glusterd_peer_hostname_t *peer_hostname = NULL;
    int32_t ret = -1;

    GF_ASSERT(hostname);
    GF_ASSERT(name);

    peer_hostname = GF_CALLOC(1, sizeof(*peer_hostname),
                              gf_gld_mt_peer_hostname_t);

    if (!peer_hostname)
        goto out;

    peer_hostname->hostname = gf_strdup(hostname);
    CDS_INIT_LIST_HEAD(&peer_hostname->hostname_list);

    *name = peer_hostname;
    ret = 0;

out:
    gf_msg_debug("glusterd", 0, "Returning %d", ret);
    return ret;
}

void
glusterd_peer_hostname_free(glusterd_peer_hostname_t *name)
{
    if (!name)
        return;

    cds_list_del_init(&name->hostname_list);

    GF_FREE(name->hostname);
    name->hostname = NULL;

    GF_FREE(name);

    return;
}

gf_boolean_t
gd_peer_has_address(glusterd_peerinfo_t *peerinfo, const char *address)
{
    glusterd_peer_hostname_t *hostname = NULL;

    GF_VALIDATE_OR_GOTO("glusterd", (peerinfo != NULL), out);
    GF_VALIDATE_OR_GOTO("glusterd", (address != NULL), out);

    cds_list_for_each_entry(hostname, &peerinfo->hostnames, hostname_list)
    {
        if (strcmp(hostname->hostname, address) == 0) {
            return _gf_true;
        }
    }

out:
    return _gf_false;
}

int
gd_add_address_to_peer(glusterd_peerinfo_t *peerinfo, const char *address)
{
    int ret = -1;
    glusterd_peer_hostname_t *hostname = NULL;

    GF_VALIDATE_OR_GOTO("glusterd", (peerinfo != NULL), out);
    GF_VALIDATE_OR_GOTO("glusterd", (address != NULL), out);

    if (gd_peer_has_address(peerinfo, address)) {
        ret = 0;
        goto out;
    }

    ret = glusterd_peer_hostname_new(address, &hostname);
    if (ret)
        goto out;

    cds_list_add_tail_rcu(&hostname->hostname_list, &peerinfo->hostnames);

    ret = 0;
out:
    return ret;
}

/* gd_add_friend_to_dict() adds details of @friend into @dict with the given
 * @prefix. All the parameters are compulsory.
 *
 * The complete address list is added to the dict only if the cluster op-version
 * is >= GD_OP_VERSION_3_6_0
 */
int
gd_add_friend_to_dict(glusterd_peerinfo_t *friend, dict_t *dict,
                      const char *prefix)
{
    int ret = -1;
    xlator_t *this = NULL;
    glusterd_conf_t *conf = NULL;
    char key[100] = {
        0,
    };
    glusterd_peer_hostname_t *address = NULL;
    int count = 0;

    this = THIS;
    GF_VALIDATE_OR_GOTO("glusterd", (this != NULL), out);

    conf = this->private;
    GF_VALIDATE_OR_GOTO(this->name, (conf != NULL), out);

    GF_VALIDATE_OR_GOTO(this->name, (friend != NULL), out);
    GF_VALIDATE_OR_GOTO(this->name, (dict != NULL), out);
    GF_VALIDATE_OR_GOTO(this->name, (prefix != NULL), out);

    snprintf(key, sizeof(key), "%s.uuid", prefix);
    ret = dict_set_dynstr_with_alloc(dict, key, uuid_utoa(friend->uuid));
    if (ret) {
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_SET_FAILED,
               "Failed to set key %s in dict", key);
        goto out;
    }

    /* Setting the first hostname from the list with this key for backward
     * compatibility
     */
    snprintf(key, sizeof(key), "%s.hostname", prefix);
    address = cds_list_entry(&friend->hostnames, glusterd_peer_hostname_t,
                             hostname_list);
    ret = dict_set_dynstr_with_alloc(dict, key, address->hostname);
    if (ret) {
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_SET_FAILED,
               "Failed to set key %s in dict", key);
        goto out;
    }

    if (conf->op_version < GD_OP_VERSION_3_6_0) {
        ret = 0;
        goto out;
    }

    address = NULL;
    count = 0;
    cds_list_for_each_entry(address, &friend->hostnames, hostname_list)
    {
        GF_VALIDATE_OR_GOTO(this->name, (address != NULL), out);

        snprintf(key, sizeof(key), "%s.hostname%d", prefix, count);
        ret = dict_set_dynstr_with_alloc(dict, key, address->hostname);
        if (ret) {
            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_SET_FAILED,
                   "Failed to set key %s in dict", key);
            goto out;
        }
        count++;
    }
    ret = snprintf(key, sizeof(key), "%s.address-count", prefix);
    ret = dict_set_int32n(dict, key, ret, count);
    if (ret)
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_SET_FAILED,
               "Failed to set key %s in dict", key);

out:
    gf_msg_debug(this ? this->name : "glusterd", 0, "Returning %d", ret);
    return ret;
}

/* gd_update_peerinfo_from_dict will update the hostnames for @peerinfo from
 * peer details with @prefix in @dict.
 * Returns 0 on success and -1 on failure.
 */
int
gd_update_peerinfo_from_dict(glusterd_peerinfo_t *peerinfo, dict_t *dict,
                             const char *prefix)
{
    int ret = -1;
    xlator_t *this = NULL;
    glusterd_conf_t *conf = NULL;
    char key[100] = {
        0,
    };
    char *hostname = NULL;
    int count = 0;
    int i = 0;

    this = THIS;
    GF_ASSERT(this != NULL);

    conf = this->private;
    GF_VALIDATE_OR_GOTO(this->name, (conf != NULL), out);

    GF_VALIDATE_OR_GOTO(this->name, (peerinfo != NULL), out);
    GF_VALIDATE_OR_GOTO(this->name, (dict != NULL), out);
    GF_VALIDATE_OR_GOTO(this->name, (prefix != NULL), out);

    ret = snprintf(key, sizeof(key), "%s.hostname", prefix);
    ret = dict_get_strn(dict, key, ret, &hostname);
    if (ret) {
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_GET_FAILED,
               "Key %s not present in "
               "dictionary",
               key);
        goto out;
    }
    ret = gd_add_address_to_peer(peerinfo, hostname);
    if (ret) {
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_ADD_ADDRESS_TO_PEER_FAIL,
               "Could not add address to peer");
        goto out;
    }
    /* Also set peerinfo->hostname to the first address */
    if (peerinfo->hostname != NULL)
        GF_FREE(peerinfo->hostname);
    peerinfo->hostname = gf_strdup(hostname);

    if (conf->op_version < GD_OP_VERSION_3_6_0) {
        ret = 0;
        goto out;
    }

    ret = snprintf(key, sizeof(key), "%s.address-count", prefix);
    ret = dict_get_int32n(dict, key, ret, &count);
    if (ret) {
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_GET_FAILED,
               "Key %s not present in "
               "dictionary",
               key);
        goto out;
    }
    hostname = NULL;
    for (i = 0; i < count; i++) {
        ret = snprintf(key, sizeof(key), "%s.hostname%d", prefix, i);
        ret = dict_get_strn(dict, key, ret, &hostname);
        if (ret) {
            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_GET_FAILED,
                   "Key %s not present "
                   "in dictionary",
                   key);
            goto out;
        }
        ret = gd_add_address_to_peer(peerinfo, hostname);
        if (ret) {
            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_ADD_ADDRESS_TO_PEER_FAIL,
                   "Could not add address to peer");
            goto out;
        }

        hostname = NULL;
    }

out:
    gf_msg_debug(this->name, 0, "Returning %d", ret);
    return ret;
}

/* gd_peerinfo_from_dict creates a peerinfo object from details of peer with
 * @prefix in @dict.
 * Returns a pointer to the created peerinfo object on success, and NULL on
 * failure.
 */
glusterd_peerinfo_t *
gd_peerinfo_from_dict(dict_t *dict, const char *prefix)
{
    int ret = -1;
    xlator_t *this = NULL;
    glusterd_conf_t *conf = NULL;
    glusterd_peerinfo_t *new_peer = NULL;
    char key[64] = {
        0,
    };
    char *uuid_str = NULL;

    this = THIS;
    GF_VALIDATE_OR_GOTO("glusterd", (this != NULL), out);

    conf = this->private;
    GF_VALIDATE_OR_GOTO(this->name, (conf != NULL), out);

    GF_VALIDATE_OR_GOTO(this->name, (dict != NULL), out);
    GF_VALIDATE_OR_GOTO(this->name, (prefix != NULL), out);

    new_peer = glusterd_peerinfo_new(GD_FRIEND_STATE_DEFAULT, NULL, NULL, 0);
    if (new_peer == NULL) {
        ret = -1;
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_PEERINFO_CREATE_FAIL,
               "Could not create peerinfo "
               "object");
        goto out;
    }

    ret = snprintf(key, sizeof(key), "%s.uuid", prefix);
    ret = dict_get_strn(dict, key, ret, &uuid_str);
    if (ret) {
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_GET_FAILED,
               "Key %s not present in "
               "dictionary",
               key);
        goto out;
    }
    gf_uuid_parse(uuid_str, new_peer->uuid);

    ret = gd_update_peerinfo_from_dict(new_peer, dict, prefix);

out:
    if ((ret != 0) && (new_peer != NULL)) {
        glusterd_peerinfo_cleanup(new_peer);
        new_peer = NULL;
    }

    return new_peer;
}

static int
gd_add_peer_hostnames_to_dict(glusterd_peerinfo_t *peerinfo, dict_t *dict,
                              const char *prefix)
{
    int ret = -1;
    xlator_t *this = NULL;
    glusterd_conf_t *conf = NULL;
    char key[64] = {
        0,
    };
    glusterd_peer_hostname_t *addr = NULL;
    int count = 0;

    this = THIS;
    GF_ASSERT(this != NULL);

    conf = this->private;
    GF_VALIDATE_OR_GOTO(this->name, (conf != NULL), out);

    if (conf->op_version < GD_OP_VERSION_3_6_0) {
        ret = 0;
        goto out;
    }

    GF_VALIDATE_OR_GOTO(this->name, (peerinfo != NULL), out);
    GF_VALIDATE_OR_GOTO(this->name, (dict != NULL), out);
    GF_VALIDATE_OR_GOTO(this->name, (prefix != NULL), out);

    cds_list_for_each_entry(addr, &peerinfo->hostnames, hostname_list)
    {
        snprintf(key, sizeof(key), "%s.hostname%d", prefix, count);
        ret = dict_set_dynstr_with_alloc(dict, key, addr->hostname);
        if (ret)
            goto out;
        count++;
    }

    ret = snprintf(key, sizeof(key), "%s.hostname_count", prefix);
    ret = dict_set_int32n(dict, key, ret, count);

out:
    return ret;
}

int
gd_add_peer_detail_to_dict(glusterd_peerinfo_t *peerinfo, dict_t *friends,
                           int count)
{
    int ret = -1;
    char key[32] = {
        0,
    };
    int keylen;
    char *peer_uuid_str = NULL;

    GF_ASSERT(peerinfo);
    GF_ASSERT(friends);

    peer_uuid_str = gd_peer_uuid_str(peerinfo);
    keylen = snprintf(key, sizeof(key), "friend%d.uuid", count);
    ret = dict_set_strn(friends, key, keylen, peer_uuid_str);
    if (ret)
        goto out;

    keylen = snprintf(key, sizeof(key), "friend%d.hostname", count);
    ret = dict_set_strn(friends, key, keylen, peerinfo->hostname);
    if (ret)
        goto out;

    keylen = snprintf(key, sizeof(key), "friend%d.port", count);
    ret = dict_set_int32n(friends, key, keylen, peerinfo->port);
    if (ret)
        goto out;

    keylen = snprintf(key, sizeof(key), "friend%d.stateId", count);
    ret = dict_set_int32n(friends, key, keylen, peerinfo->state.state);
    if (ret)
        goto out;

    keylen = snprintf(key, sizeof(key), "friend%d.state", count);
    ret = dict_set_strn(
        friends, key, keylen,
        glusterd_friend_sm_state_name_get(peerinfo->state.state));
    if (ret)
        goto out;

    keylen = snprintf(key, sizeof(key), "friend%d.connected", count);
    ret = dict_set_int32n(friends, key, keylen, (int32_t)peerinfo->connected);
    if (ret)
        goto out;

    snprintf(key, sizeof(key), "friend%d", count);
    ret = gd_add_peer_hostnames_to_dict(peerinfo, friends, key);

out:
    return ret;
}

/* glusterd_peerinfo_find_by_generation searches for a peer which has the
 * generation number @generation and if found returns the pointer to peerinfo
 * object. Returns NULL otherwise.
 */
glusterd_peerinfo_t *
glusterd_peerinfo_find_by_generation(uint32_t generation)
{
    glusterd_conf_t *priv = NULL;
    glusterd_peerinfo_t *entry = NULL;
    glusterd_peerinfo_t *found = NULL;
    xlator_t *this = THIS;
    glusterd_friend_sm_state_t state;

    GF_ASSERT(this);

    priv = this->private;

    GF_ASSERT(priv);

    RCU_READ_LOCK;
    cds_list_for_each_entry_rcu(entry, &priv->peers, uuid_list)
    {
        if (entry->generation == generation) {
            found = entry; /* Probably should be rcu_dereferenced */
            state = found->state.state;
            break;
        }
    }
    RCU_READ_UNLOCK;

    if (found)
        gf_msg_debug(this->name, 0, "Friend found... state: %s",
                     glusterd_friend_sm_state_name_get(state));
    else
        gf_msg_debug(this->name, 0,
                     "Friend with generation: %" PRIu32 ", not found",
                     generation);
    return found;
}

int
glusterd_get_peers_count()
{
    int count = 0;
    xlator_t *this = NULL;
    glusterd_conf_t *conf = NULL;
    glusterd_peerinfo_t *peer = NULL;

    this = THIS;
    GF_VALIDATE_OR_GOTO("glusterd", this, out);

    conf = this->private;
    GF_VALIDATE_OR_GOTO(this->name, conf, out);

    RCU_READ_LOCK;
    cds_list_for_each_entry_rcu(peer, &conf->peers, uuid_list) count++;
    RCU_READ_UNLOCK;

out:
    return count;
}