flow_stats_manager.h

Go to the documentation of this file.

/*
 * Copyright (c) 2015 Juniper Networks, Inc. All rights reserved.
 */
 
#ifndef vnsw_agent_flow_stats_maanger_h
#define vnsw_agent_flow_stats_maanger_h
 
#include <atomic>
 
 
#include <cmn/agent_cmn.h>
#include <cmn/index_vector.h>
#include <uve/stats_collector.h>
#include <uve/interface_uve_stats_table.h>
#include <pkt/flow_table.h>
#include <vrouter/ksync/flowtable_ksync.h>
#include <sandesh/common/flow_types.h>
 
extern SandeshTraceBufferPtr FlowExportStatsTraceBuf;
 
#define FLOW_EXPORT_STATS_TRACE(...)\
do {\
    FlowExportStatsTrace::TraceMsg(FlowExportStatsTraceBuf, __FILE__, __LINE__, __VA_ARGS__);\
} while (false)
 
class FlowStatsCollector;
class FlowStatsCollectorObject;
class SessionStatsCollector;
class SessionStatsCollectorObject;
 
struct FlowAgingTableKey {
    FlowAgingTableKey(const uint8_t &protocol, const uint16_t &dst_port):
        proto(protocol), port(dst_port) {}
 
    bool operator==(const FlowAgingTableKey &rhs) const {
        return (proto == rhs.proto && port == rhs.port);
    }
 
    bool operator<(const FlowAgingTableKey &rhs) const {
        if (proto != rhs.proto) {
            return proto < rhs.proto;
        }
 
        return port < rhs.port;
    }
 
    uint8_t proto;
    uint16_t port;
};
 
struct FlowStatsCollectorReq {
    enum Event {
        INVALID,
        ADD_FLOW_STATS_COLLECTOR,
        DELETE_FLOW_STATS_COLLECTOR,
        FREE_FLOW_STATS_COLLECTOR,
    };
 
    FlowStatsCollectorReq(Event ev, const FlowAgingTableKey &k,
                          uint64_t interval, uint64_t timeout) :
        event(ev), key(k), flow_stats_interval(interval),
        flow_cache_timeout(timeout) {}
 
    FlowStatsCollectorReq(Event ev, const FlowAgingTableKey &k):
        event(ev), key(k) {}
 
    Event event;
    FlowAgingTableKey key;
    uint64_t flow_stats_interval;
    uint64_t flow_cache_timeout;
};
 
class FlowStatsManager {
public:
    static const uint8_t kCatchAllProto = 0x0;
    static const uint64_t FlowThresoldUpdateTime = 1000 * 2;
    static const uint32_t kDefaultFlowSamplingThreshold = 500;
    static const uint32_t kMinFlowSamplingThreshold = 20;
 
    typedef boost::shared_ptr<FlowStatsCollectorObject> FlowAgingTablePtr;
    typedef boost::shared_ptr<SessionStatsCollectorObject> SessionStatsCollectorPtr;
 
    typedef std::map<const FlowAgingTableKey, FlowAgingTablePtr>
                     FlowAgingTableMap;
    typedef std::pair<const FlowAgingTableKey, FlowAgingTablePtr>
                     FlowAgingTableEntry;
 
    FlowStatsManager(Agent *agent);
    ~FlowStatsManager();
 
    Agent* agent() { return agent_; }
    FlowStatsCollectorObject* default_flow_stats_collector_obj() {
        return default_flow_stats_collector_obj_.get();
    }
    SessionStatsCollectorObject* session_stats_collector_obj() {
        return session_stats_collector_obj_.get();
    }
 
    //Add protocol + port based flow aging table
    void Add(const FlowAgingTableKey &key,
             uint64_t flow_stats_interval,
             uint64_t flow_cache_timeout);
    void Delete(const FlowAgingTableKey &key);
    void Free(const FlowAgingTableKey &key);
 
    //Add flow entry to particular aging table
    void AddEvent(FlowEntryPtr &flow);
    void DeleteEvent(const FlowEntryPtr &flow, const RevFlowDepParams &params);
    void UpdateStatsEvent(const FlowEntryPtr &flow, uint32_t bytes,
                          uint32_t packets, uint32_t oflow_bytes,
                          const boost::uuids::uuid &u);
 
    void Init(uint64_t flow_stats_interval, uint64_t flow_cache_timeout);
    void InitDone();
    void Shutdown();
 
    FlowAgingTableMap::iterator begin() {
        return flow_aging_table_map_.begin();
    }
 
    FlowAgingTableMap::iterator end() {
        return flow_aging_table_map_.end();
    }
 
    FlowStatsCollector* GetFlowStatsCollector(const FlowEntry *p) const;
    const FlowStatsCollectorObject* Find(uint32_t proto, uint32_t port) const;
 
    bool RequestHandler(boost::shared_ptr<FlowStatsCollectorReq> req);
    void AddReqHandler(boost::shared_ptr<FlowStatsCollectorReq> req);
    void DeleteReqHandler(boost::shared_ptr<FlowStatsCollectorReq> req);
    void FreeReqHandler(boost::shared_ptr<FlowStatsCollectorReq> req);
 
    FlowStatsCollectorObject* proto(int protocol) {
        return protocol_list_[protocol];
    }
    uint32_t session_export_rate() const {
        return session_export_rate_;
    }
 
    uint32_t session_export_count() const {
        return session_export_count_;
    }
 
    void set_session_export_count(uint32_t count) {
        session_export_count_ = count;
    }
 
    uint32_t session_export_count_reset() {
        return session_export_count_.exchange(0);
    }
 
    uint32_t session_export_without_sampling_reset() {
        return session_export_without_sampling_.exchange(0);
    }
 
    uint32_t session_export_drops() const { return session_export_drops_; }
 
    uint64_t session_sample_exports() const { return session_sample_exports_; }
    uint64_t session_msg_exports() const { return session_msg_exports_; }
    uint64_t session_exports() const { return session_exports_; }
 
    uint64_t session_export_disable_drops() const {
        return session_export_disable_drops_;
    }
    uint32_t session_export_sampling_drops() const {
        return session_export_sampling_drops_;
    }
    uint64_t session_global_slo_logging_drops() const {
        return session_global_slo_logging_drops_;
    }
    uint64_t session_slo_logging_drops() const {
        return session_slo_logging_drops_;
    }
    void set_sessions_sampled_atleast_once() {
        sessions_sampled_atleast_once_ = true;
    }
 
    uint64_t threshold() const { return threshold_;}
    bool delete_short_flow() const {
        return delete_short_flow_;
    }
 
    void set_delete_short_flow(bool val) {
        delete_short_flow_ = val;
    }
    static void FlowStatsReqHandler(Agent *agent, uint32_t proto,
                                    uint32_t port,
                                    uint64_t protocol);
    void FreeIndex(uint32_t idx);
    uint32_t AllocateIndex();
    void UpdateSessionSampleExportStats(uint32_t count);
    void UpdateSessionMsgExportStats(uint32_t count);
    void UpdateSessionExportStats(uint32_t count, bool first_export,
                                  bool sampled);
 
    void SetProfileData(ProfileData *data);
    void RegisterDBClients();
    friend class AgentUtXmlFlowThreshold;
    friend class AgentUtXmlFlowThresholdValidate;
private:
    friend struct FlowStatsCollectorReq;
    friend class FlowStatsRecordsReq;
    friend class FlowStatsCollector;
    friend class SessionStatsCollector;
    bool UpdateSessionThreshold(void);
    void UpdateThreshold(uint64_t new_value, bool check_oflow);
    FlowStatsCollectorObject* GetFlowStatsCollectorObject(const FlowEntry *flow)
        const;
    Agent *agent_;
    WorkQueue<boost::shared_ptr<FlowStatsCollectorReq> > request_queue_;
    FlowAgingTableMap flow_aging_table_map_;
    FlowAgingTablePtr default_flow_stats_collector_obj_;
    SessionStatsCollectorPtr session_stats_collector_obj_;
    uint64_t prev_flow_export_rate_compute_time_;
    uint64_t threshold_;
    uint32_t prev_cfg_flow_export_rate_;
    uint32_t session_export_rate_;
    std::atomic<uint32_t> session_export_count_;
    std::atomic<uint64_t> session_sample_exports_;
    std::atomic<uint64_t> session_msg_exports_;
    std::atomic<uint64_t> session_exports_;
    std::atomic<uint64_t> session_export_disable_drops_;
    std::atomic<uint64_t> session_export_sampling_drops_;
    std::atomic<uint32_t> session_export_without_sampling_;
    std::atomic<uint64_t> session_export_drops_;
    std::atomic<bool> sessions_sampled_atleast_once_;
    std::atomic<uint64_t> session_global_slo_logging_drops_;
    std::atomic<uint64_t> session_slo_logging_drops_;
    Timer* timer_;
    bool delete_short_flow_;
    //Protocol based array for minimal tree comparision
    FlowStatsCollectorObject* protocol_list_[256];
    IndexVector<FlowStatsCollector *> instance_table_;
};
#endif //vnsw_agent_flow_stats_manager_h