db_handler.h

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

#ifndef DB_HANDLER_H_
#define DB_HANDLER_H_

#include <boost/scoped_ptr.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/ptr_container/ptr_map.hpp>
#include <boost/uuid/uuid.hpp>
#include <boost/uuid/name_generator.hpp>

#if __GNUC_PREREQ(4, 6)
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wunused-result"
#endif
#include <boost/uuid/uuid_generators.hpp>
#if __GNUC_PREREQ(4, 6)
#pragma GCC diagnostic pop
#endif

#include <boost/tuple/tuple.hpp>

#include "base/parse_object.h"
#include "io/event_manager.h"
#include "base/random_generator.h"
#include "gendb_if.h"
#include "gendb_statistics.h"
#include "sandesh/sandesh.h"
#include "viz_message.h"
#include "uflow_types.h"
#include "viz_constants.h"
#include <database/cassandra/cql/cql_types.h>
#include "usrdef_counters.h"

class Options;
class DiscoveryServiceClient;

class DbHandler {
public:
    static const int DefaultDbTTL = 0;
    static boost::uuids::uuid seed_uuid;

    typedef enum {
        INVALID = 0,
        UINT64 = 1,
        STRING = 2,
        DOUBLE = 3,
        MAXVAL 
    } VarType;

    struct Var {
        Var() : type(INVALID), str(""), num(0), dbl(0) {}
        Var(const std::string &s) : type(STRING), str(s), num(0), dbl(0) {}
        Var(uint64_t v) : type(UINT64), str(""), num(v), dbl(0) {}
        Var(double d) : type(DOUBLE), str(""), num(0), dbl(d) {}
        VarType type;
        std::string str;
        uint64_t num;
        double dbl;
        bool operator==(const Var &other) const {
            if (type!=other.type) return false;
            switch (type) {
                case STRING:
                    if (str!=other.str) return false;
                    break;
                case UINT64:
                    if (num!=other.num) return false;
                    break;
                case DOUBLE:
                    if (dbl!=other.dbl) return false;
                    break;
                default:
                    break; 
            }
            return true;
        }
        friend inline std::ostream& operator<<(std::ostream& out,
            const Var& value);
    };

    typedef std::map<std::string, std::string> RuleMap;

    typedef std::map<std::string, Var > AttribMap;
    typedef std::multimap<std::string, std::pair<Var, AttribMap> > TagMap;

    DbHandler(EventManager *evm, GenDb::GenDbIf::DbErrorHandler err_handler,
        const std::vector<std::string> &cassandra_ips,
        const std::vector<int> &cassandra_ports,
        std::string name, const TtlMap& ttl_map,
        const std::string& cassandra_user,
        const std::string& cassandra_password,
        const std::string& cassandra_compaction_strategy,
        const std::string& cassandra_flow_tables_compaction_strategy,
        const std::string &zookeeper_server_list,
        bool use_zookeeper, bool disable_all_writes, bool disable_stats_writes,
        bool disable_messages_writes, bool disable_messages_keyword_writes);
    DbHandler(GenDb::GenDbIf *dbif, const TtlMap& ttl_map);
    virtual ~DbHandler();

    static uint64_t GetTtlInHourFromMap(const TtlMap& ttl_map,
            TtlType::type type);
    static uint64_t GetTtlFromMap(const TtlMap& ttl_map,
            TtlType::type type);
    bool DropMessage(const SandeshHeader &header, const VizMsg *vmsg);
    bool Init(bool initial, int instance);
    void UnInit(int instance);
    void UnInitUnlocked(int instance);
    void GetRuleMap(RuleMap& rulemap);

    virtual void MessageTableInsert(const VizMsg *vmsgp,
        GenDb::GenDbIf::DbAddColumnCb db_cb);
    void ObjectTableInsert(const std::string &table, const std::string &rowkey,
        uint64_t &timestamp, const boost::uuids::uuid& unm,
        const VizMsg *vmsgp, GenDb::GenDbIf::DbAddColumnCb db_cb);
    void StatTableInsert(uint64_t ts, 
            const std::string& statName,
            const std::string& statAttr,
            const TagMap & attribs_tag,
            const AttribMap & attribs_all,
            GenDb::GenDbIf::DbAddColumnCb db_cb);
    bool FlowTableInsert(const pugi::xml_node& parent,
        const SandeshHeader &header, GenDb::GenDbIf::DbAddColumnCb db_cb);
    bool UnderlayFlowSampleInsert(const UFlowData& flow_data,
        uint64_t timestamp, GenDb::GenDbIf::DbAddColumnCb db_cb);

    bool GetStats(uint64_t *queue_count, uint64_t *enqueues) const;

    bool GetStats(std::vector<GenDb::DbTableInfo> *vdbti,
        GenDb::DbErrors *dbe, std::vector<GenDb::DbTableInfo> *vstats_dbti);
    bool GetCumulativeStats(std::vector<GenDb::DbTableInfo> *vdbti,
        GenDb::DbErrors *dbe, std::vector<GenDb::DbTableInfo> *vstats_dbti)
        const;
    void GetSandeshStats(std::string *drop_level,
        std::vector<SandeshStats> *vdropmstats) const;
    bool GetCqlMetrics(cass::cql::Metrics *metrics) const;
    bool GetCqlStats(cass::cql::DbStats *stats) const;
    void SetDbQueueWaterMarkInfo(Sandesh::QueueWaterMarkInfo &wm,
        boost::function<void (void)> defer_undefer_cb);
    void ResetDbQueueWaterMarkInfo();
    std::vector<boost::asio::ip::tcp::endpoint> GetEndpoints() const;
    std::string GetName() const;
    void UpdateUdc(Options *o, DiscoveryServiceClient *c) {
        udc_->Update(o, c);
    }
    bool IsAllWritesDisabled() const;
    bool IsStatisticsWritesDisabled() const;
    bool IsMessagesWritesDisabled() const;
    bool IsMessagesKeywordWritesDisabled() const;
    void DisableAllWrites(bool disable);
    void DisableStatisticsWrites(bool disable);
    void DisableMessagesWrites(bool disable);
    void DisableMessagesKeywordWrites(bool disable);

private:
    void MessageTableKeywordInsert(const VizMsg *vmsgp,
        GenDb::GenDbIf::DbAddColumnCb db_cb);
    void StatTableInsertTtl(uint64_t ts,
        const std::string& statName,
        const std::string& statAttr,
        const TagMap & attribs_tag,
        const AttribMap & attribs_all, int ttl,
        GenDb::GenDbIf::DbAddColumnCb db_cb);
    void FieldNamesTableInsert(uint64_t timestamp,
        const std::string& table_name, const std::string& field_name,
        const std::string& field_val, int ttl,
        GenDb::GenDbIf::DbAddColumnCb db_cb);
    void MessageTableOnlyInsert(const VizMsg *vmsgp,
        GenDb::GenDbIf::DbAddColumnCb db_cb);
    bool MessageIndexTableInsert(const std::string& cfname,
        const SandeshHeader& header, const std::string& message_type,
        const boost::uuids::uuid& unm, const std::string keyword,
        GenDb::GenDbIf::DbAddColumnCb db_cb);
    bool AllowMessageTableInsert(const SandeshHeader &header);
    bool CreateTables();
    void SetDropLevel(size_t queue_count, SandeshLevel::type level,
        boost::function<void (void)> cb);
    bool Setup(int instance);
    bool Initialize(int instance);
    bool InitializeInternal(int instance);
    bool InitializeInternalLocked(int instance);
    bool StatTableWrite(uint32_t t2,
        const std::string& statName, const std::string& statAttr,
        const std::pair<std::string,DbHandler::Var>& ptag,
        const std::pair<std::string,DbHandler::Var>& stag,
        uint32_t t1, const boost::uuids::uuid& unm,
        const std::string& jsonline, int ttl,
        GenDb::GenDbIf::DbAddColumnCb db_cb);
    bool FlowSampleAdd(const pugi::xml_node& flowdata,
        const SandeshHeader& header,
        GenDb::GenDbIf::DbAddColumnCb db_cb);
    uint64_t GetTtl(TtlType::type type) {
        return GetTtlFromMap(ttl_map_, type);
    }
    bool InsertIntoDb(std::auto_ptr<GenDb::ColList> col_list,
        GenDb::GenDbIf::DbAddColumnCb db_cb);

    boost::scoped_ptr<GenDb::GenDbIf> dbif_;
    // Random generator for UUIDs
    ThreadSafeUuidGenerator umn_gen_;
    std::string name_;
    std::string col_name_;
    SandeshLevel::type drop_level_;
    VizMsgStatistics dropped_msg_stats_;
    GenDb::DbTableStatistics stable_stats_;
    mutable tbb::mutex smutex_;
    TtlMap ttl_map_;
    static uint32_t field_cache_t2_;
    static std::set<std::string> field_cache_set_[2];
    static uint32_t field_cache_old_t2_;
    static uint8_t old_t2_index_;
    static uint8_t new_t2_index_;
    static tbb::mutex fmutex_;
    std::string tablespace_;
    std::string compaction_strategy_;
    std::string flow_tables_compaction_strategy_;
    UniformInt8RandomGenerator gen_partition_no_;
    std::string zookeeper_server_list_;
    bool use_zookeeper_;
    bool disable_all_writes_;
    bool disable_statistics_writes_;
    bool disable_messages_writes_;
    bool disable_messages_keyword_writes_;
    bool CanRecordDataForT2(uint32_t, std::string);
    boost::scoped_ptr<UserDefinedCounters> udc_;
    Timer *udc_cfg_poll_timer_;
    static const int kUDCPollInterval = 120 * 1000; // in ms
    bool PollUDCCfg() { if(udc_) udc_->PollCfg(); return true; }
    void PollUDCCfgErrorHandler(std::string err_name, std::string err_message);
    friend class DbHandlerTest;
    DISALLOW_COPY_AND_ASSIGN(DbHandler);
};

typedef boost::shared_ptr<DbHandler> DbHandlerPtr;

inline std::ostream& operator<<(std::ostream& out, const DbHandler::Var& value) {
    switch (value.type) {
      case DbHandler::STRING:
	out << value.str;
	break;
      case DbHandler::UINT64:
	out << value.num;
	break;
      case DbHandler::DOUBLE:
	out << value.dbl;
	break;
      default:
	out << "Invalid type: " << value.type;
	break;
    }
    return out;
}

//
// DbHandlerInitializer - Wrapper to perform DbHandler initialization
//
class DbHandlerInitializer {
 public:
    typedef boost::function<void(void)> InitializeDoneCb;
    DbHandlerInitializer(EventManager *evm,
        const std::string &db_name, int db_task_instance,
        const std::string &timer_task_name, InitializeDoneCb callback,
        const std::vector<std::string> &cassandra_ips,
        const std::vector<int> &cassandra_ports,
        const TtlMap& ttl_map,
        const std::string& cassandra_user,
        const std::string& cassandra_password,
        const std::string &cassandra_compaction_strategy,
        const std::string &cassandra_flow_tables_compaction_strategy,
        const std::string &zookeeper_server_list,
        bool use_zookeeper, bool disable_all_db_writes,
        bool disable_db_stats_writes, bool disable_db_messages_writes,
        bool disable_db_messages_keyword_writes);
    DbHandlerInitializer(EventManager *evm,
        const std::string &db_name, int db_task_instance,
        const std::string &timer_task_name, InitializeDoneCb callback,
        DbHandlerPtr db_handler);
    virtual ~DbHandlerInitializer();
    bool Initialize();
    void Shutdown();
    DbHandlerPtr GetDbHandler() const;

 private:
    bool InitTimerExpired();
    void InitTimerErrorHandler(std::string error_name,
        std::string error_message);
    void StartInitTimer();
    void ScheduleInit();

    static const int kInitRetryInterval = 10 * 1000; // in ms
    const std::string db_name_;
    const int db_task_instance_;
    DbHandlerPtr db_handler_;
    InitializeDoneCb callback_;
    Timer *db_init_timer_;
};

/*
 * pugi walker to process flow message
 */
template <typename T>
class FlowLogDataObjectWalker : public pugi::xml_tree_walker {
public:
    FlowLogDataObjectWalker(T &values) :
        values_(values) {
    }
    ~FlowLogDataObjectWalker() {}

    // Callback that is called when traversal begins
    virtual bool begin(pugi::xml_node& node) {
        return true;
    }

    // Callback that is called for each node traversed
    virtual bool for_each(pugi::xml_node& node);

    // Callback that is called when traversal ends
    virtual bool end(pugi::xml_node& node) {
        return true;
    }

private:
    T &values_;
};

#endif /* DB_HANDLER_H_ */