config_etcd_client.cc

Go to the documentation of this file.

/*
 * Copyright (c) 2018 Juniper Networks, Inc. All rights reserved.
 */

#include "config-client-mgr/config_etcd_client.h"

#include <sandesh/request_pipeline.h>

#include "rapidjson/writer.h"
#include "rapidjson/stringbuffer.h"

#include "config_client_log.h"
#include "config_client_log_types.h"
#include "config_client_show_types.h"

#include <boost/algorithm/string/join.hpp>
#include <boost/algorithm/string/predicate.hpp>
#include <boost/foreach.hpp>
#include <boost/functional/hash.hpp>
#include <boost/ptr_container/ptr_map.hpp>
#include <boost/uuid/uuid.hpp>
#include <map>
#include <set>
#include <string>
#include <utility>

#include "base/connection_info.h"
#include "base/logging.h"
#include "base/regex.h"
#include "base/task.h"
#include "base/task_annotations.h"
#include "base/task_trigger.h"
#include "config_cass2json_adapter.h"
#include "io/event_manager.h"
#include "config_factory.h"
#include "config_client_log.h"
#include "config_client_log_types.h"
#include "config_client_show_types.h"
#include "sandesh/common/vns_constants.h"

using contrail::regex;
using contrail::regex_match;
using contrail::regex_search;
using namespace std;
using etcd::etcdql::EtcdIf;
using etcd::etcdql::EtcdResponse;
using contrail_rapidjson::Value;
using contrail_rapidjson::Document;
using contrail_rapidjson::SizeType;
using contrail_rapidjson::StringBuffer;
using contrail_rapidjson::Writer;

bool ConfigEtcdClient::disable_watch_;

class ConfigEtcdClient::EtcdWatcher : public Task {
public:
    EtcdWatcher(ConfigEtcdClient *etcd_client) :
        Task(TaskScheduler::GetInstance()->GetTaskId("etcd::EtcdWatcher")),
        etcd_client_(etcd_client) {
    }

    virtual bool Run();

    ConfigEtcdClient *client() const {
        return etcd_client_;
    }
    string Description() const {
        return "ConfigEtcdClient::EtcdWatcher";
    }

private:
    ConfigEtcdClient *etcd_client_;
    void ProcessResponse(EtcdResponse resp);
};

ConfigEtcdClient::ConfigEtcdClient(ConfigClientManager *mgr,
                                   EventManager *evm,
                                   const ConfigClientOptions &options,
                                   int num_workers)
                 : ConfigDbClient(mgr, evm, options),
                   num_workers_(num_workers)
{
    eqlif_.reset(ConfigStaticObjectFactory::Create<EtcdIf>(config_db_ips(),
                                              GetFirstConfigDbPort(),
                                              false));

    InitConnectionInfo();
    bulk_sync_status_ = 0;

    for (int i = 0; i < num_workers_; i++) {
        partitions_.push_back(
            ConfigStaticObjectFactory::Create<ConfigEtcdPartition>(this, i));
    }

    uuid_reader_.reset(new
        TaskTrigger(boost::bind(&ConfigEtcdClient::UUIDReader, this),
        TaskScheduler::GetInstance()->GetTaskId("config_client::DBReader"),
        0));
}

ConfigEtcdClient::~ConfigEtcdClient() {
    STLDeleteValues(&partitions_);
}

void ConfigEtcdClient::StartWatcher() {
    if (disable_watch_)  {
        CONFIG_CLIENT_DEBUG(
            ConfigClientMgrDebug,
            "ETCD Watcher SM: StartWatcher: ETCD watch disabled");
        return;
    }

    if (mgr()->is_reinit_triggered()) {
        CONFIG_CLIENT_DEBUG(
            ConfigClientMgrDebug,
            "ETCD Watcher SM: StartWatcher: re init triggered,"
            " don't enqueue ETCD Watcher Task.");
        return;
    }

    TaskScheduler *scheduler = TaskScheduler::GetInstance();
    Task *task = new EtcdWatcher(this);
    scheduler->Enqueue(task);
}

void ConfigEtcdClient::EtcdWatcher::ProcessResponse(
                                         EtcdResponse resp) {
    client()->ProcessResponse(resp);
}

bool ConfigEtcdClient::EtcdWatcher::Run() {
    if (etcd_client_->mgr()->is_reinit_triggered()) {
        CONFIG_CLIENT_DEBUG(
            ConfigClientMgrDebug,
            "ETCD Watcher SM: Run: re init triggered,"
            " don't wait for end of config");
        return true;
    }

    client()->eqlif_->Watch("/contrail/",
            boost::bind(&ConfigEtcdClient::EtcdWatcher::ProcessResponse,
                        this, _1));

    return true;
}

void ConfigEtcdClient::ProcessResponse(EtcdResponse resp) {
    if (mgr()->is_reinit_triggered()) {
        CONFIG_CLIENT_DEBUG(
            ConfigClientMgrDebug,
            "ETCD Watcher SM: ProcessResponse: re init triggered,"
            " stop watching");
        eqlif_->StopWatch();
        return;
    }

    mgr()->WaitForEndOfConfig();

    assert(resp.err_code() == 0);

    if (resp.action() == 0) {
        EnqueueUUIDRequest("CREATE", resp.key(), resp.value());
    } else if (resp.action() == 1) {
        EnqueueUUIDRequest("UPDATE", resp.key(), resp.value());
    } else if (resp.action() == 2) {
        EnqueueUUIDRequest("DELETE", resp.key(), resp.value());
    }
}

void ConfigEtcdClient::InitDatabase() {
    HandleEtcdConnectionStatus(false, true);
    while (true) {
        CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug, "ETCD SM: Db Init");
        if (!eqlif_->Connect()) {
            CONFIG_CLIENT_DEBUG(ConfigEtcdInitErrorMessage,
                                "Database initialization failed");
            if (!InitRetry()) return;
            continue;
        }
        break;
    }
    HandleEtcdConnectionStatus(true);
    BulkDataSync();
}

void ConfigEtcdClient::HandleEtcdConnectionStatus(bool success,
                                                  bool force_update) {
    UpdateConnectionInfo(success, force_update);

    if (success) {
        // Update connection info
        process::ConnectionState::GetInstance()->Update(
            process::ConnectionType::DATABASE, "Etcd",
            process::ConnectionStatus::UP,
            eqlif_->endpoints(), "Established ETCD connection");
       CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
                           "ETCD SM: Established ETCD connection");
    } else {
        process::ConnectionState::GetInstance()->Update(
            process::ConnectionType::DATABASE, "Etcd",
            process::ConnectionStatus::DOWN,
            eqlif_->endpoints(), "Lost ETCD connection");
       CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
                           "ETCD SM: Lost ETCD connection");
    }
}

bool ConfigEtcdClient::InitRetry() {
    CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug, "ETCD SM: DB Init Retry");
    // If reinit is triggered, return false to abort connection attempt
    if (mgr()->is_reinit_triggered()) return false;
    usleep(GetInitRetryTimeUSec());
    return true;
}

bool ConfigEtcdClient::BulkDataSync() {
    CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
                        "ETCD SM: BulkDataSync Started");
    bulk_sync_status_ = num_workers_;
    uuid_reader_->Set();
    return true;
}

void ConfigEtcdClient::BulkSyncDone() {
    long num_config_readers_still_processing =
        bulk_sync_status_.fetch_and_decrement();
    if (num_config_readers_still_processing == 1) {
        CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
                            "Etcd SM: BulkSyncDone by all readers");
        mgr()->EndOfConfig();
    } else {
        CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
                            "Etcd SM: One reader finished BulkSync");
    }
}

void ConfigEtcdClient::PostShutdown() {
    CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
                        "ETCD SM: Post shutdown during re-init");
    STLDeleteValues(&partitions_);
    ClearFQNameCache();
}

ConfigEtcdPartition *
ConfigEtcdClient::GetPartition(const string &uuid) {
    int worker_id = HashUUID(uuid);
    return partitions_[worker_id];
}

const ConfigEtcdPartition *
ConfigEtcdClient::GetPartition(const string &uuid) const {
    int worker_id = HashUUID(uuid);
    return partitions_[worker_id];
}

const ConfigEtcdPartition *
ConfigEtcdClient::GetPartition(int worker_id) const {
    assert(worker_id < num_workers_);
    return partitions_[worker_id];
}

int ConfigEtcdClient::HashUUID(const string &uuid_str) const {
    boost::hash<string> string_hash;
    return string_hash(uuid_str) % num_workers_;
}

void ConfigEtcdClient::EnqueueUUIDRequest(string oper,
                                          string uuid,
                                          string value) {
    // Get the trimmed uuid
    size_t front_pos = uuid.rfind('/');
    string uuid_key = uuid.substr(front_pos + 1);

    // Cache uses the trimmed uuid
    if (oper == "CREATE" || oper == "UPDATE") {

        Document d;
        d.Parse<0>(value.c_str());
        Document::AllocatorType &a = d.GetAllocator();

        // If non-object JSON is received, log a warning and return.
        if (!d.IsObject()) {
            CONFIG_CLIENT_WARN(ConfigClientMgrWarning, "ETCD SM: Received "
                  "non-object json. uuid: "
                  + uuid_key + " value: "
                  + value + " .Skipping");
            return;
        }

        // ETCD does not provide obj-type since it is encoded in the
        // UUID key. Since config_json_parser and IFMap need type to be
        // present in the document, fix up by adding obj-type.
        if (!d.HasMember("type")) {
            string type = uuid.substr(10, front_pos - 10);
            Value v;
            Value va;
            d.AddMember(v.SetString("type", a),
                        va.SetString(type.c_str(), a), a);
            StringBuffer sb;
            Writer<StringBuffer> writer(sb);
            d.Accept(writer);
            value = sb.GetString();
        }

        // Add to FQName cache if not present
        if (d.HasMember("type") &&
            d.HasMember("fq_name")) {
            string obj_type = d["type"].GetString();
            string fq_name;
            const Value &name = d["fq_name"];
            for (Value::ConstValueIterator name_itr = name.Begin();
                 name_itr != name.End(); ++name_itr) {
                fq_name += name_itr->GetString();
                fq_name += ":";
            }
            fq_name.erase(fq_name.end()-1);
            if (FindFQName(uuid_key) == "ERROR") {
                AddFQNameCache(uuid_key, obj_type, fq_name);
            }
        }
    } else if (oper == "DELETE") {
        // Invalidate cache
        InvalidateFQNameCache(uuid_key);
    }

    // Request has the uuid with entire path
    ObjectProcessReq *req = new ObjectProcessReq(oper, uuid, value);

    // GetPartition uses the trimmed uuid so that the same
    // partition is returned for different requests on the
    // same UUID
    GetPartition(uuid_key)->Enqueue(req);
}

void ConfigEtcdClient::EnqueueDBSyncRequest(
         const UUIDValueList &uuid_list) {
    for (UUIDValueList::const_iterator it = uuid_list.begin();
        it != uuid_list.end(); it++) {
        EnqueueUUIDRequest("CREATE", it->first, it->second);
    }
}

bool ConfigEtcdClient::UUIDReader() {

    string next_key;
    string prefix = "/contrail/";
    bool read_done = false;
    ostringstream os;

    for (ConfigClientManager::ObjectTypeList::const_iterator it =
             mgr()->config_json_parser()->ObjectTypeListToRead().begin();
         it != mgr()->config_json_parser()->ObjectTypeListToRead().end();
         it++) {

        /* Form the key for the object type to lookup */
        next_key = prefix + it->c_str();
        os.str("");
        os << next_key << 1;

        while (true) {
            unsigned int num_entries;

            if (mgr()->is_reinit_triggered()) {
                CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
                "ETCD SM: Abort UUID reader on reinit trigger");
                return true;
            }

            num_entries = GetNumReadRequestToBunch();

            EtcdResponse resp = eqlif_->Get(next_key,
                                            os.str(),
                                            num_entries);
            EtcdResponse::kv_map kvs = resp.kvmap();

            if (resp.err_code() == 0) {
                UUIDValueList uuid_list;

                for (multimap<string, string>::const_iterator iter = kvs.begin();
                     iter != kvs.end();
                     ++iter) {
                     next_key = iter->first;
                     if (!boost::starts_with(next_key, "/contrail/")) {
                         CONFIG_CLIENT_WARN(ConfigClientMgrWarning,
                              "ETCD SM: Non-contrail uuid: "
                              + next_key + " received");
                     } else {
                         uuid_list.push_back(make_pair(iter->first, iter->second));
                     }
                }

                EnqueueDBSyncRequest(uuid_list);

                next_key += "00";

                if (kvs.size() < num_entries) {
                    break;
                }
            } else if (resp.err_code() == 100)  {
                break;
            } else if (resp.err_code() == -1) {
                /* Test ONLY */
                read_done = true;
                break;
            } else {
                HandleEtcdConnectionStatus(false);
                usleep(GetInitRetryTimeUSec());
            }
        } //while
        if (read_done) {
            break;
        }
    } //for

    // At the end of task trigger
    BOOST_FOREACH(ConfigEtcdPartition *partition, partitions_) {
        ObjectProcessReq *req = new ObjectProcessReq("EndOfConfig", "", "");
        partition->Enqueue(req);
    }

    return true;
}

bool ConfigEtcdClient::IsTaskTriggered() const {
    // If UUIDReader task has been triggered return true.
    if (uuid_reader_->IsSet()) {
        return true;
    }

    BOOST_FOREACH(ConfigEtcdPartition *partition, partitions_) {
        if (partition->IsTaskTriggered()) {
            return true;
        }
    }
    return false;
}

bool ConfigEtcdClient::UUIDToObjCacheShow(
                           const string &search_string,
                           int inst_num,
                           const string &last_uuid,
                           uint32_t num_entries,
                           vector<ConfigDBUUIDCacheEntry> *entries) const {
   return GetPartition(inst_num)->UUIDToObjCacheShow(search_string, last_uuid,
                                                     num_entries, entries);
}

bool ConfigEtcdClient::IsListOrMapPropEmpty(const string &uuid_key,
                                            const string &lookup_key) {
    return GetPartition(uuid_key)->IsListOrMapPropEmpty(uuid_key, lookup_key);
}

ConfigEtcdPartition::ConfigEtcdPartition(
                   ConfigEtcdClient *client, size_t idx)
    : config_client_(client), worker_id_(idx) {
    int task_id = TaskScheduler::GetInstance()->GetTaskId("config_client::Reader");
    config_reader_.reset(new
     TaskTrigger(boost::bind(&ConfigEtcdPartition::ConfigReader, this),
     task_id, idx));
    task_id =
        TaskScheduler::GetInstance()->GetTaskId("config_client::ObjectProcessor");
    obj_process_queue_.reset(new WorkQueue<ObjectProcessReq *>(
        task_id, idx, bind(&ConfigEtcdPartition::RequestHandler, this, _1),
        WorkQueue<ObjectProcessReq *>::kMaxSize, 512));
}

ConfigEtcdPartition::~ConfigEtcdPartition() {
    obj_process_queue_->Shutdown();
}

void ConfigEtcdPartition::Enqueue(ObjectProcessReq *req) {
    obj_process_queue_->Enqueue(req);
}

bool ConfigEtcdPartition::RequestHandler(ObjectProcessReq *req) {
    AddUUIDToProcessList(req->oper_, req->uuid_str_, req->value_);
    delete req;
    return true;
}

void ConfigEtcdPartition::AddUUIDToProcessList(const string &oper,
                                               const string &uuid_key,
                                               const string &value_str) {
    pair<UUIDProcessSet::iterator, bool> ret;
    bool trigger = uuid_process_set_.empty();

    size_t front_pos = uuid_key.rfind('/');
    string uuid = uuid_key.substr(front_pos + 1);
    UUIDProcessRequestType *req =
        new UUIDProcessRequestType(oper, uuid, value_str);
    ret = uuid_process_set_.insert(make_pair(client()->GetUUID(uuid), req));
    if (ret.second) {
        if (trigger) {
            config_reader_->Set();
        }
    } else {
        if ((oper == "DELETE") &&
            (ret.first->second->oper == "CREATE")) {
            uuid_process_set_.erase(ret.first);
            client()->PurgeFQNameCache(uuid);
        } else {
            delete req;
            ret.first->second->oper = oper;
            ret.first->second->uuid = uuid;
            ret.first->second->value = value_str;
        }
    }
}

ConfigEtcdPartition::UUIDCacheEntry::~UUIDCacheEntry() {
}

void ConfigEtcdPartition::FillUUIDToObjCacheInfo(const string &uuid,
                                  UUIDCacheMap::const_iterator uuid_iter,
                                  ConfigDBUUIDCacheEntry *entry) const {
    entry->set_uuid(uuid);
    entry->set_timestamp(
            UTCUsecToString(uuid_iter->second->GetLastReadTimeStamp()));
    entry->set_fq_name(uuid_iter->second->GetFQName());
    entry->set_obj_type(uuid_iter->second->GetObjType());
    entry->set_json_str(uuid_iter->second->GetJsonString());
}

bool ConfigEtcdPartition::UUIDToObjCacheShow(
                      const string &search_string,
                      const string &last_uuid,
                      uint32_t num_entries,
                      vector<ConfigDBUUIDCacheEntry> *entries) const {
    uint32_t count = 0;
    regex search_expr(search_string);
    for (UUIDCacheMap::const_iterator it =
        uuid_cache_map_.upper_bound(last_uuid);
        count < num_entries && it != uuid_cache_map_.end(); it++) {
        if (regex_search(it->first, search_expr) ||
                regex_search(it->second->GetObjType(), search_expr) ||
                regex_search(it->second->GetFQName(), search_expr)) {
            count++;
            ConfigDBUUIDCacheEntry entry;
            FillUUIDToObjCacheInfo(it->first, it, &entry);
            entries->push_back(entry);
        }
    }
    return true;
}

ConfigEtcdPartition::UUIDCacheEntry *
ConfigEtcdPartition::GetUUIDCacheEntry(const string &uuid) {
    UUIDCacheMap::iterator uuid_iter = uuid_cache_map_.find(uuid);
    if (uuid_iter == uuid_cache_map_.end()) {
        return NULL;
    }
    return uuid_iter->second;
}

ConfigEtcdPartition::UUIDCacheEntry *
ConfigEtcdPartition::GetUUIDCacheEntry(const string &uuid,
                                       const string &value,
                                       bool &is_new) {
    UUIDCacheMap::iterator uuid_iter = uuid_cache_map_.find(uuid);
    if (uuid_iter == uuid_cache_map_.end()) {
        UUIDCacheEntry *obj;
        obj = new UUIDCacheEntry(this,
                                 value,
                                 UTCTimestampUsec());
        string tmp_uuid = uuid;
        pair<UUIDCacheMap::iterator, bool> ret_uuid =
            uuid_cache_map_.insert(tmp_uuid, obj);
        assert(ret_uuid.second);
        uuid_iter = ret_uuid.first;

        is_new = true;
    } else {
        uuid_iter->second->SetLastReadTimeStamp(UTCTimestampUsec());
    }

    return uuid_iter->second;
}

int ConfigEtcdPartition::UUIDRetryTimeInMSec(
        const UUIDCacheEntry *obj) const {
    uint32_t retry_time_pow_of_two =
        obj->GetRetryCount() > kMaxUUIDRetryTimePowOfTwo ?
        kMaxUUIDRetryTimePowOfTwo : obj->GetRetryCount();
   return ((1 << retry_time_pow_of_two) * kMinUUIDRetryTimeMSec);
}

void ConfigEtcdPartition::UUIDCacheEntry::EnableEtcdReadRetry(
        const string uuid,
        const string value) {
    if (!retry_timer_) {
        retry_timer_ = TimerManager::CreateTimer(
                *parent_->client()->event_manager()->io_service(),
                "UUID retry timer for " + uuid,
                TaskScheduler::GetInstance()->GetTaskId(
                                "config_client::Reader"),
                parent_->worker_id_);
        CONFIG_CLIENT_DEBUG(ConfigClientReadRetry,
                "Created UUID read retry timer ", uuid);
    }
    retry_timer_->Cancel();
    retry_timer_->Start(parent_->UUIDRetryTimeInMSec(this),
            boost::bind(
                &ConfigEtcdPartition::UUIDCacheEntry::EtcdReadRetryTimerExpired,
                this, uuid, value),
            boost::bind(
                &ConfigEtcdPartition::UUIDCacheEntry::EtcdReadRetryTimerErrorHandler,
                this));
    CONFIG_CLIENT_DEBUG(ConfigClientReadRetry,
            "Start/restart UUID Read Retry timer due to configuration", uuid);
}

void ConfigEtcdPartition::UUIDCacheEntry::DisableEtcdReadRetry(
        const string uuid) {
    CHECK_CONCURRENCY("config_client::Reader");
    if (retry_timer_) {
        retry_timer_->Cancel();
        TimerManager::DeleteTimer(retry_timer_);
        retry_timer_ = NULL;
        retry_count_ = 0;
        CONFIG_CLIENT_DEBUG(ConfigClientReadRetry,
                "UUID Read retry timer - deleted timer due to configuration",
                uuid);
    }
}

bool ConfigEtcdPartition::UUIDCacheEntry::IsRetryTimerRunning() const {
    if (retry_timer_)
        return (retry_timer_->running());
    return false;
}

bool ConfigEtcdPartition::UUIDCacheEntry::EtcdReadRetryTimerExpired(
        const string uuid,
        const string value) {
    CHECK_CONCURRENCY("config_client::Reader");
    parent_->client()->EnqueueUUIDRequest(
            "UPDATE", parent_->client()->uuid_str(uuid), value);
    retry_count_++;
    CONFIG_CLIENT_DEBUG(ConfigClientReadRetry, "timer expired ", uuid);
    return false;
}

void
ConfigEtcdPartition::UUIDCacheEntry::EtcdReadRetryTimerErrorHandler() {
     std::string message = "Timer";
     CONFIG_CLIENT_WARN(ConfigClientGetRowError,
            "UUID Read Retry Timer error ", message, message);
}

bool ConfigEtcdPartition::UUIDCacheEntry::ListOrMapPropEmpty(
         const string &prop) const {
    ListMapSet::const_iterator it = list_map_set_.find(prop);
    if (it == list_map_set_.end()) {
        return true;
    }
    return (it->second == false);
}

bool ConfigEtcdPartition::GenerateAndPushJson(const string &uuid,
                                              Document &doc,
                                              bool add_change,
                                              UUIDCacheEntry *cache) {

    // Get obj_type from cache.
    const string &obj_type = cache->GetObjType();

    // string to get the type field.
    string type_str;

    // bool to indicate if an update to ifmap_server is
    // necessary.
    bool notify_update = false;

    // Walk the document, remove unwanted properties and do
    // needed fixup for the others.
    Value::ConstMemberIterator itr = doc.MemberBegin();
    Document::AllocatorType &a = doc.GetAllocator();

    while (itr != doc.MemberEnd()) {

        string key = itr->name.GetString();

        /*
         * Indicate need for update since document has
         * at least one field other than fq_name or
         * obj_type to be updated.
         */
        if (!notify_update &&
            key.compare("type") != 0 &&
            key.compare("fq_name") != 0) {
            notify_update = true;
        }

        if (ConfigClientManager::skip_properties.find(key) !=
            ConfigClientManager::skip_properties.end()) {
            itr = doc.EraseMember(itr);
            continue;
        }

        if (key.compare("type") == 0) {
            type_str = itr->value.GetString();
            itr = doc.EraseMember(itr);
            continue;
        }

        string wrapper = client()->mgr()->config_json_parser()-> \
                           GetWrapperFieldName(obj_type, key.c_str());
        if (!wrapper.empty()) {

            // Get the propl/propm json Value
            Value &map_value = doc[key.c_str()];

            // Indicate in cache if propm/propl is empty
            cache->SetListOrMapPropEmpty(key,
                                         map_value.IsNull());

        } else if (key.compare("parent_type") == 0) {

            string parent_type = doc[key.c_str()].GetString();
            replace(parent_type.begin(), parent_type.end(),
                   '-', '_');
            doc[key.c_str()].SetString(parent_type.c_str(), a);

        } else if (key.compare("parent_uuid") == 0) {

            if (add_change) {
                string parent_uuid = doc[key.c_str()].GetString();
                string parent_fq_name = client()->FindFQName(parent_uuid);
                if (parent_fq_name == "ERROR") {
                    CONFIG_CLIENT_DEBUG(ConfigClientReadRetry,
                        "Parent fq_name not available for ", uuid);
                    return false;
                }
            }

        } else if (key.compare("bgpaas_session_attributes") == 0) {

            doc[key.c_str()].SetString("", a);

        } else if (key.find("_refs") != string::npos && add_change) {

            // Determine if NULL attr needs to be processed
            string ref_type = key.substr(0, key.length() - 5);
            bool link_with_attr =
                client()->mgr()->config_json_parser()-> \
                    IsLinkWithAttr(obj_type, ref_type);

            // Get a pointer to the _refs json Value
            Value *v = &doc[key.c_str()];

            assert(v->IsArray());
            for (SizeType i = 0; i < v->Size(); i++) {

                // Process NULL attr
                Value &va = (*v)[i];
                if (link_with_attr) {
                    if (va["attr"].IsNull()) {
                        (*v)[i].RemoveMember("attr");
                        Value vm;
                        (*v)[i].AddMember("attr", vm.SetObject(), a);
                    }
                }

                Value &uuidVal = va["uuid"];
                const string ref_uuid = uuidVal.GetString();
                string ref_fq_name = client()->FindFQName(ref_uuid);

                if (ref_fq_name == "ERROR") {
                    // ref_fq_name not in FQNameCache
                    // If we cannot find ref_fq_name in the doc
                    // as well, return false to enable retry.
                    if (!va.HasMember("to")) {
                        CONFIG_CLIENT_DEBUG(ConfigClientReadRetry,
                                   "Ref fq_name not available for ", uuid);
                        return false;
                    }

                    ref_fq_name.clear();
                    const Value &name = va["to"];
                    for (Value::ConstValueIterator itr = name.Begin();
                         itr != name.End(); ++itr) {
                        ref_fq_name += itr->GetString();
                        ref_fq_name += ":";
                    }
                    ref_fq_name.erase(ref_fq_name.end()-1);
                }

                // Remove ref_fq_name from doc and re-add the
                // string formatted fq_name.
                (*v)[i].RemoveMember("to");
                Value vs1(ref_fq_name.c_str(), a);
                (*v)[i].AddMember("to", vs1, a);
            }

            // For creates/updates, need to update cache json_str
            // with the new fixed ref_fq_names for _refs.
            // Remove existing reference in cache and create a new
            // ref with updated ref_fq_names.
            Document cacheDoc;
            string cache_json_str = cache->GetJsonString();
            cacheDoc.Parse<0>(cache_json_str.c_str());
            cacheDoc.RemoveMember(key.c_str());
            Value vr;
            Value vra;
            Value refVal;
            refVal.CopyFrom(*v, a);
            cacheDoc.AddMember(vr.SetString(key.c_str(), a),
                               refVal, a);
            StringBuffer sb;
            Writer<StringBuffer> writer(sb);
            cacheDoc.Accept(writer);
            string cache_str = sb.GetString();
            cache->SetJsonString(cache_str);
        }

        if (itr != doc.MemberEnd()) itr++;
    }

    if (!notify_update) {
        CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
             "ETCD SM: Nothing to update");
        return true;
    }

    StringBuffer sb1;
    Writer<StringBuffer> writer1(sb1);
    Document refDoc;
    refDoc.CopyFrom(doc, refDoc.GetAllocator());
    refDoc.Accept(writer1);
    string refString = sb1.GetString();
    CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
        "ETCD SM: JSON Doc fed to CJP: " + refString);

    ConfigCass2JsonAdapter ccja(uuid, type_str, doc);
    client()->mgr()->config_json_parser()->Receive(ccja, add_change);

    return true;
}

void ConfigEtcdPartition::ProcessUUIDDelete(
                                       const string &uuid_key) {

    if (client()->FindFQName(uuid_key) == "ERROR") {
        CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
             "ETCD SM: Nothing to delete");
       return;
    }

    UUIDCacheMap::iterator uuid_iter = uuid_cache_map_.find(uuid_key);
    if (uuid_iter == uuid_cache_map_.end()) {
        return;
    }
    UUIDCacheEntry *cache = uuid_iter->second;

    if (cache->IsRetryTimerRunning()) {
        cache->DisableEtcdReadRetry(uuid_key);
        client()->PurgeFQNameCache(uuid_key);
        return;
    }

    const string cache_json_str = cache->GetJsonString();
    Document delDoc;
    delDoc.Parse<0>(cache_json_str.c_str());

    GenerateAndPushJson(uuid_key,
                        delDoc,
                        false,
                        cache);

    uuid_cache_map_.erase(uuid_iter);

    client()->PurgeFQNameCache(uuid_key);
}

void ConfigEtcdPartition::ProcessUUIDUpdate(const string &uuid_key,
                                            const string &value_str) {
    bool is_new = false;
    UUIDCacheEntry *cache = GetUUIDCacheEntry(uuid_key,
                                              value_str,
                                              is_new);

    string cache_json_str = cache->GetJsonString();
    if (cache_json_str.compare("retry") == 0) {
        // If we are retrying due to ref or parent
        // fq_name not available previously, cache
        // json_str would have been cleared and set
        // to retry. Process now like a new create.
        cache_json_str = value_str;
        is_new = true;
    }
    Document cacheDoc;
    cacheDoc.Parse<0>(cache_json_str.c_str());

    Document updDoc;
    updDoc.Parse<0>(value_str.c_str());
    string key;

    if (!updDoc.HasMember("fq_name") ||
        !updDoc.HasMember("type")) {
        CONFIG_CLIENT_WARN(ConfigClientGetRowError,
             "fq_name or type not present for ",
             "obj_uuid_table with uuid: ", uuid_key);
        cache->DisableEtcdReadRetry(uuid_key);
        ProcessUUIDDelete(uuid_key);
        return;
    }

    Value::ConstMemberIterator itr = updDoc.MemberBegin();
    while (itr != updDoc.MemberEnd()) {

        key = itr->name.GetString();

        if (key.compare("draft_mode_state") == 0) {
            string mode = itr->value.GetString();
            if (!mode.empty()) {
                client()->PurgeFQNameCache(uuid_key);
                DeleteCacheMap(uuid_key);
                return;
            }
            itr = updDoc.EraseMember(itr);
            continue;
        }

        if (is_new) {
            if (key.compare("type") == 0) {
                string type = itr->value.GetString();
                cache->SetObjType(type);
            } else if (key.compare("fq_name") == 0) {
                string fq_name;
                const Value &name = updDoc[key.c_str()];
                for (Value::ConstValueIterator name_itr = name.Begin();
                     name_itr != name.End(); ++name_itr) {
                    fq_name += name_itr->GetString();
                    fq_name += ":";
                }
                fq_name.erase(fq_name.end()-1);
                cache->SetFQName(fq_name);
            }
        }

        if (!is_new && cacheDoc.HasMember(key.c_str()) &&
            key.compare("type") != 0 &&
            key.compare("fq_name") != 0) {
            if (cacheDoc[key.c_str()] == updDoc[key.c_str()]) {
                itr = updDoc.EraseMember(itr);
            }
            assert(cacheDoc.RemoveMember(key.c_str()));
        } else {
            if (itr != updDoc.MemberEnd()) itr++;
        }
    }

    cache->SetJsonString(value_str);

    if (!GenerateAndPushJson(uuid_key,
                        updDoc,
                        true,
                        cache)) {
        cache->EnableEtcdReadRetry(uuid_key, value_str);
        cache->SetJsonString("retry");
    } else {
        cache->DisableEtcdReadRetry(uuid_key);
    }
    if (!is_new) {
        GenerateAndPushJson(uuid_key,
                            cacheDoc,
                            false,
                            cache);
    }
}

bool ConfigEtcdPartition::IsListOrMapPropEmpty(const string &uuid_key,
                                          const string &lookup_key) {
    UUIDCacheMap::iterator uuid_iter = uuid_cache_map_.find(uuid_key);
    if (uuid_iter == uuid_cache_map_.end()) {
        return true;
    }
    UUIDCacheEntry *cache = uuid_iter->second;

    return cache->ListOrMapPropEmpty(lookup_key);
}

bool ConfigEtcdPartition::IsTaskTriggered() const {
    return (config_reader_->IsSet());
}

bool ConfigEtcdPartition::ConfigReader() {
    CHECK_CONCURRENCY("config_client::Reader");

    int num_req_handled = 0;

    for (UUIDProcessSet::iterator it = uuid_process_set_.begin(), itnext;
         it != uuid_process_set_.end() &&
                 !client()->mgr()->is_reinit_triggered();
         it = itnext) {

        itnext = it;
        ++itnext;

        UUIDProcessRequestType *obj_req = it->second;

        if (obj_req->oper == "CREATE" || obj_req->oper == "UPDATE") {
            ProcessUUIDUpdate(obj_req->uuid, obj_req->value);
        } else if (obj_req->oper == "DELETE") {
            ProcessUUIDDelete(obj_req->uuid);
        } else if (obj_req->oper == "EndOfConfig") {
            client()->BulkSyncDone();
        }
        RemoveObjReqEntry(obj_req->uuid);

        if (++num_req_handled == client()->GetMaxRequestsToYield()) {
            return false;
        }
    }

    // Clear the UUID read set if we are currently processing reinit request
    if (client()->mgr()->is_reinit_triggered()) {
        CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
            "ETCD SM: Clear UUID process set due to reinit");
        uuid_process_set_.clear();
    }
    assert(uuid_process_set_.empty());
    return true;
}

void ConfigEtcdPartition::RemoveObjReqEntry(string &uuid) {
    UUIDProcessSet::iterator req_it =
        uuid_process_set_.find(client()->GetUUID(uuid));
    delete req_it->second;
    uuid_process_set_.erase(req_it);
}