Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
       3             :  * Copyright Contributors to the OpenSDN project
       4             :  */
       5             : 
       6             : #ifndef SRC_BGP_BGP_XMPP_CHANNEL_H_
       7             : #define SRC_BGP_BGP_XMPP_CHANNEL_H_
       8             : 
       9             : #include <map>
      10             : #include <set>
      11             : #include <string>
      12             : #include <utility>
      13             : #include <mutex>
      14             : #include <atomic>
      15             : 
      16             : #include <boost/function.hpp>
      17             : #include <boost/system/error_code.hpp>
      18             : #include <boost/scoped_ptr.hpp>
      19             : 
      20             : #include "base/queue_task.h"
      21             : #include "bgp/bgp_rib_policy.h"
      22             : #include "bgp/routing-instance/routing_instance.h"
      23             : #include "io/tcp_session.h"
      24             : #include "net/rd.h"
      25             : #include "schema/xmpp_mvpn_types.h"
      26             : #include "xmpp/xmpp_channel.h"
      27             : 
      28             : namespace pugi {
      29             : class xml_node;
      30             : }
      31             : 
      32             : class BgpGlobalSystemConfig;
      33             : class BgpRouterState;
      34             : class BgpServer;
      35             : class BgpXmppRTargetManager;
      36             : struct DBRequest;
      37             : class IPeer;
      38             : class PeerCloseManager;
      39             : class XmppServer;
      40             : class BgpXmppChannelMock;
      41             : class BgpXmppChannelManager;
      42             : class BgpXmppChannelManagerMock;
      43             : class BgpXmppPeerClose;
      44             : class Timer;
      45             : class XmppConfigUpdater;
      46             : class XmppPeerInfoData;
      47             : class XmppSession;
      48             : 
      49             : class BgpXmppChannel {
      50             : public:
      51             :     static const int kEndOfRibSendRetryTime = 1; // Seconds
      52             :     enum StatsIndex {
      53             :         RX,
      54             :         TX,
      55             :     };
      56             :     struct Stats {
      57       20084 :         Stats() {
      58       20084 :             rt_updates = 0;
      59       20084 :             reach = 0;
      60       20084 :             unreach = 0;
      61       20083 :             end_of_rib = 0;
      62       20084 :         }
      63             :         std::atomic<uint64_t> rt_updates;
      64             :         std::atomic<uint64_t> reach;
      65             :         std::atomic<uint64_t> unreach;
      66             :         std::atomic<uint64_t> end_of_rib;
      67             :     };
      68             :     struct ChannelStats {
      69       10042 :         ChannelStats() {
      70       10042 :             instance_subscribe = 0;
      71       10042 :             instance_unsubscribe = 0;
      72       10042 :             table_subscribe = 0;
      73       10041 :             table_subscribe_complete = 0;
      74       10042 :             table_unsubscribe = 0;
      75       10042 :             table_unsubscribe_complete = 0;
      76       10042 :         }
      77             :         std::atomic<uint64_t> instance_subscribe;
      78             :         std::atomic<uint64_t> instance_unsubscribe;
      79             :         std::atomic<uint64_t> table_subscribe;
      80             :         std::atomic<uint64_t> table_subscribe_complete;
      81             :         std::atomic<uint64_t> table_unsubscribe;
      82             :         std::atomic<uint64_t> table_unsubscribe_complete;
      83             :     };
      84             : 
      85             :     struct ErrorStats {
      86       10042 :         ErrorStats() {
      87       10042 :             inet6_rx_bad_xml_token_count = 0;
      88       10042 :             inet6_rx_bad_prefix_count = 0;
      89       10042 :             inet6_rx_bad_nexthop_count = 0;
      90       10042 :             inet6_rx_bad_afi_safi_count = 0;
      91       10042 :         }
      92             :         void incr_inet6_rx_bad_xml_token_count();
      93             :         void incr_inet6_rx_bad_prefix_count();
      94             :         void incr_inet6_rx_bad_nexthop_count();
      95             :         void incr_inet6_rx_bad_afi_safi_count();
      96             :         uint64_t get_inet6_rx_bad_xml_token_count() const;
      97             :         uint64_t get_inet6_rx_bad_prefix_count() const;
      98             :         uint64_t get_inet6_rx_bad_nexthop_count() const;
      99             :         uint64_t get_inet6_rx_bad_afi_safi_count() const;
     100             : 
     101             :         std::atomic<uint64_t> inet6_rx_bad_xml_token_count;
     102             :         std::atomic<uint64_t> inet6_rx_bad_prefix_count;
     103             :         std::atomic<uint64_t> inet6_rx_bad_nexthop_count;
     104             :         std::atomic<uint64_t> inet6_rx_bad_afi_safi_count;
     105             :     };
     106             : 
     107             :     explicit BgpXmppChannel(XmppChannel *channel, BgpServer *bgp_server = NULL,
     108             :                             BgpXmppChannelManager *manager = NULL);
     109             :     virtual ~BgpXmppChannel();
     110             : 
     111             :     void Close();
     112             :     IPeer *Peer();
     113             :     const IPeer *Peer() const;
     114             :     virtual TcpSession::Endpoint endpoint() const;
     115             : 
     116             :     const std::string &ToString() const;
     117             :     const std::string &ToUVEKey() const;
     118             :     std::string StateName() const;
     119             :     TcpSession::Endpoint remote_endpoint() const;
     120             :     TcpSession::Endpoint local_endpoint() const;
     121             :     std::string transport_address_string() const;
     122             : 
     123             :     void set_peer_closed(bool flag);
     124             :     bool peer_deleted() const;
     125             :     uint64_t peer_closed_at() const;
     126             :     bool table_membership_request_map_empty() const;
     127             :     size_t GetMembershipRequestQueueSize() const;
     128             : 
     129             :     const XmppSession *GetSession() const;
     130             :     const Stats &rx_stats() const { return stats_[RX]; }
     131             :     const Stats &tx_stats() const { return stats_[TX]; }
     132          88 :     ErrorStats &error_stats() { return error_stats_; }
     133             :     const ErrorStats &error_stats() const { return error_stats_; }
     134        9993 :     void set_deleted(bool deleted) { deleted_ = deleted; }
     135        6178 :     bool deleted() { return deleted_; }
     136             :     void RoutingInstanceCallback(std::string vrf_name, int op);
     137             :     void ASNUpdateCallback(as_t old_asn, as_t old_local_asn);
     138             :     void FillInstanceMembershipInfo(BgpNeighborResp *resp) const;
     139             :     void FillTableMembershipInfo(BgpNeighborResp *resp) const;
     140             :     void FillCloseInfo(BgpNeighborResp *resp) const;
     141             :     void StaleCurrentSubscriptions();
     142             :     void LlgrStaleCurrentSubscriptions();
     143             :     void SweepCurrentSubscriptions();
     144             :     void XMPPPeerInfoSend(const XmppPeerInfoData &peer_info) const;
     145         974 :     const XmppChannel *channel() const { return channel_; }
     146       25348 :     XmppChannel *channel() { return channel_; }
     147             :     void StartEndOfRibReceiveTimer();
     148             :     void ResetEndOfRibSendState();
     149             :     bool EndOfRibSendTimerExpired();
     150             :     bool MembershipResponseHandler(std::string table_name);
     151          18 :     Timer *eor_send_timer() const { return eor_send_timer_; }
     152           9 :     bool eor_sent() const { return eor_sent_; }
     153             :     size_t table_membership_requests() const;
     154             :     void ClearEndOfRibState();
     155       72567 :     PeerCloseManager *close_manager() { return close_manager_.get(); }
     156             : 
     157           7 :     uint64_t get_rx_route_reach() const { return stats_[RX].reach; }
     158           7 :     uint64_t get_rx_route_unreach() const { return stats_[RX].unreach; }
     159             :     uint64_t get_rx_update() const { return stats_[RX].rt_updates; }
     160             : 
     161           7 :     uint64_t get_tx_route_reach() const { return stats_[TX].reach; }
     162           7 :     uint64_t get_tx_route_unreach() const { return stats_[TX].unreach; }
     163           7 :     uint64_t get_tx_update() const { return stats_[TX].rt_updates; }
     164             :     bool SkipUpdateSend();
     165       42219 :     bool delete_in_progress() const { return delete_in_progress_; }
     166        6178 :     void set_delete_in_progress(bool flag) { delete_in_progress_ = flag; }
     167             : 
     168        7237 :     BgpXmppRTargetManager *rtarget_manager() {
     169        7237 :         return rtarget_manager_.get();
     170             :     }
     171             :     bool IsSubscriptionGrStale(RoutingInstance *instance) const;
     172             :     bool IsSubscriptionLlgrStale(RoutingInstance *instance) const;
     173             :     bool IsSubscriptionEmpty() const;
     174             :     const RoutingInstance::RouteTargetList &GetSubscribedRTargets(
     175             :             RoutingInstance *instance) const;
     176        6178 :     void ClearSubscriptions() { routing_instances_.clear(); }
     177      273987 :     BgpServer *bgp_server() { return bgp_server_; }
     178       28796 :     const BgpXmppChannelManager *manager() const { return manager_; }
     179       38278 :     BgpXmppChannelManager *manager() { return manager_; }
     180             :     XmppChannel *xmpp_channel() const { return channel_; }
     181             :     void ReceiveEndOfRIB(Address::Family family);
     182             :     void ProcessPendingSubscriptions();
     183             :     int GetPrimaryInstanceID(const std::string &s,
     184             :                              bool expect_prefix_len) const;
     185             : 
     186             : protected:
     187             :     struct InstanceMembershipRequestState {
     188         182 :         InstanceMembershipRequestState(int instance_id = -1,
     189             :             bool no_ribout = false)
     190         182 :             : instance_id(instance_id), no_ribout(no_ribout) {
     191         182 :         };
     192             : 
     193             :         int instance_id;
     194             :         bool no_ribout;
     195             :     };
     196             : 
     197             : 
     198             :     XmppChannel *channel_;
     199             : 
     200             : private:
     201             :     friend class BgpXmppChannelMock;
     202             :     friend class BgpXmppChannelManager;
     203             :     friend class BgpXmppParseTest;
     204             :     friend class BgpXmppUnitTest;
     205             :     class XmppPeer;
     206             :     class PeerClose;
     207             :     class PeerStats;
     208             : 
     209             :     //
     210             :     // State the instance id received in Membership subscription request
     211             :     // Also remember we received unregister request
     212             :     //
     213             :     enum RequestType {
     214             :         NONE,
     215             :         SUBSCRIBE,
     216             :         UNSUBSCRIBE,
     217             :     };
     218             :     struct TableMembershipRequestState {
     219       72819 :         TableMembershipRequestState(RequestType current, int id,
     220             :             bool no_ribout = false)
     221       72819 :             : current_req(current),
     222       72819 :               instance_id(id),
     223       72819 :               pending_req(current),
     224       72819 :               no_ribout(no_ribout) {
     225       72819 :         }
     226             :         RequestType current_req;
     227             :         int instance_id;
     228             :         RequestType pending_req;
     229             :         bool no_ribout;
     230             :     };
     231             : 
     232             :     // Map of routing instances to which this BgpXmppChannel is subscribed.
     233             :     struct SubscriptionState {
     234             :         enum State {
     235             :             NONE = 0,
     236             :             GR_STALE = 1 << 0,
     237             :             LLGR_STALE = 1 << 1
     238             :         };
     239             : 
     240             :         SubscriptionState() : index(-1), state(NONE) { }
     241       14231 :         SubscriptionState(const RoutingInstance::RouteTargetList &targets,
     242             :                           int index)
     243       14231 :                 : targets(targets), index(index), state(NONE) { }
     244             : 
     245       10497 :         bool IsGrStale() const { return((state & GR_STALE) != 0); }
     246         740 :         void SetGrStale() { state |= GR_STALE; }
     247             : 
     248         121 :         void SetLlgrStale() { state |= LLGR_STALE; }
     249        9690 :         bool IsLlgrStale() const { return((state & LLGR_STALE) != 0); }
     250             : 
     251         251 :         void ClearStale() { state &= ~(GR_STALE | LLGR_STALE); }
     252             : 
     253             :         RoutingInstance::RouteTargetList targets;
     254             :         int index;
     255             :         uint32_t state;
     256             :     };
     257             : 
     258             :     typedef std::map<RoutingInstance *, SubscriptionState>
     259             :         SubscribedRoutingInstanceList;
     260             : 
     261             :     // Map of table name to subscription request state.
     262             :     typedef std::map<std::string,
     263             :         TableMembershipRequestState> TableMembershipRequestMap;
     264             : 
     265             :     // Map of routing-instance name to subscription request state. This map
     266             :     // maintains list of requests that are received for subscription before
     267             :     // routing instance is created or recreated (in case the instance is in
     268             :     // the process of being deleted).
     269             :     typedef std::map<std::string,
     270             :         InstanceMembershipRequestState> InstanceMembershipRequestMap;
     271             : 
     272             :     // The code assumes that multimap preserves insertion order for duplicate
     273             :     // values of same key.
     274             :     typedef std::pair<const std::string, const std::string> VrfTableName;
     275             :     typedef std::multimap<VrfTableName, DBRequest *> DeferQ;
     276             : 
     277             :     virtual void ReceiveUpdate(const XmppStanza::XmppMessage *msg);
     278             : 
     279             :     virtual bool GetMembershipInfo(BgpTable *table,
     280             :         int *instance_id, uint64_t *subscribed_at, RequestType *req_type);
     281             :     bool VerifyMembership(const std::string &vrf_name, Address::Family family,
     282             :         BgpTable **table, int *instance_id, uint64_t *subscribed_at,
     283             :         bool *subscribe_pending, bool add_change);
     284             : 
     285             :     void AddTableMembershipState(const std::string &table_name,
     286             :         TableMembershipRequestState tmr_state);
     287             :     bool DeleteTableMembershipState(const std::string &table_name);
     288             :     TableMembershipRequestState *GetTableMembershipState(
     289             :         const std::string &table_name);
     290             :     const TableMembershipRequestState *GetTableMembershipState(
     291             :         const std::string &table_name) const;
     292             : 
     293             :     void AddInstanceMembershipState(const std::string &instance,
     294             :         InstanceMembershipRequestState imr_state);
     295             :     bool DeleteInstanceMembershipState(const std::string &instance);
     296             :     virtual const InstanceMembershipRequestState *GetInstanceMembershipState(
     297             :         const std::string &instance) const;
     298             : 
     299             :     bool ProcessItem(std::string vrf_name, const pugi::xml_node &node,
     300             :                      bool add_change, int primary_instance_id = 0);
     301             :     bool ProcessInet6Item(std::string vrf_name, const pugi::xml_node &node,
     302             :                           bool add_change);
     303             :     bool ProcessMcastItem(std::string vrf_name,
     304             :                           const pugi::xml_node &item, bool add_change);
     305             :     bool ProcessMvpnItem(std::string vrf_name,
     306             :                          const pugi::xml_node &item, bool add_change);
     307             :     void CreateType7MvpnRouteRequest(IpAddress grp_address,
     308             :         IpAddress src_address, bool add_change, uint64_t subscription_gen_id,
     309             :         DBRequest &req);
     310             :     void CreateType5MvpnRouteRequest(IpAddress grp_address,
     311             :         IpAddress src_address, bool add_change, uint64_t subscription_gen_id,
     312             :         int instance_id, DBRequest &req, const autogen::MvpnNextHopType &nh);
     313             :     bool ProcessEnetItem(std::string vrf_name,
     314             :                          const pugi::xml_node &item, bool add_change);
     315             :     void ProcessSubscriptionRequest(std::string rt_instance,
     316             :                                     const XmppStanza::XmppMessageIq *iq,
     317             :                                     bool add_change);
     318             :     void AddSubscriptionState(RoutingInstance *rt_instance, int index);
     319             :     void DeleteSubscriptionState(RoutingInstance *rt_instance);
     320             :     SubscriptionState *GetSubscriptionState(RoutingInstance *rt_instance);
     321             :     const SubscriptionState *GetSubscriptionState(
     322             :         RoutingInstance *rt_instance) const;
     323             : 
     324             :     void RegisterTable(int line, BgpTable *table,
     325             :         const TableMembershipRequestState *tmr_state);
     326             :     void UnregisterTable(int line, BgpTable *table);
     327             :     void MembershipRequestCallback(BgpTable *table);
     328             :     void DequeueRequest(const std::string &table_name, DBRequest *request);
     329             :     bool XmppDecodeAddress(int af, const std::string &address,
     330             :                            IpAddress *addrp, bool zero_ok = false);
     331             :     bool ResumeClose();
     332             :     void FlushDeferQ(std::string vrf_name);
     333             :     void FlushDeferQ(std::string vrf_name, std::string table_name);
     334             :     void ProcessDeferredSubscribeRequest(RoutingInstance *rt_instance,
     335             :         const InstanceMembershipRequestState &imr_state);
     336             :     void ClearStaledSubscription(RoutingInstance *rt_instance,
     337             :                                  SubscriptionState *sub_state);
     338             :     bool ProcessMembershipResponse(std::string table_name,
     339             :                                    TableMembershipRequestState *tmr_state);
     340             :     bool EndOfRibReceiveTimerExpired();
     341             :     void EndOfRibTimerErrorHandler(std::string error_name,
     342             :                                    std::string error_message);
     343             :     void SendEndOfRIB();
     344             :     virtual time_t GetEndOfRibSendTime() const;
     345             : 
     346             :     xmps::PeerId peer_id_;
     347             :     boost::scoped_ptr<BgpXmppRTargetManager> rtarget_manager_;
     348             :     BgpServer *bgp_server_;
     349             :     boost::scoped_ptr<XmppPeer> peer_;
     350             :     boost::scoped_ptr<BgpXmppPeerClose> peer_close_;
     351             :     boost::scoped_ptr<PeerCloseManager> close_manager_;
     352             :     boost::scoped_ptr<PeerStats> peer_stats_;
     353             :     RibExportPolicy bgp_policy_;
     354             : 
     355             :     // DB Requests pending membership request response.
     356             :     DeferQ defer_q_;
     357             : 
     358             :     TableMembershipRequestMap table_membership_request_map_;
     359             :     InstanceMembershipRequestMap instance_membership_request_map_;
     360             :     BgpXmppChannelManager *manager_;
     361             :     bool delete_in_progress_;
     362             :     bool deleted_;
     363             :     bool defer_peer_close_;
     364             :     bool skip_update_send_;
     365             :     bool skip_update_send_cached_;
     366             :     bool eor_sent_;
     367             :     Timer *eor_receive_timer_;
     368             :     Timer *eor_send_timer_;
     369             :     time_t eor_receive_timer_start_time_;
     370             :     time_t eor_send_timer_start_time_;
     371             :     WorkQueue<std::string> membership_response_worker_;
     372             :     SubscribedRoutingInstanceList routing_instances_;
     373             : 
     374             :     // statistics
     375             :     Stats stats_[2];
     376             :     ChannelStats channel_stats_;
     377             :     ErrorStats error_stats_;
     378             : 
     379             :     // Label block manager for multicast labels.
     380             :     LabelBlockManagerPtr lb_mgr_;
     381             : 
     382             :     DISALLOW_COPY_AND_ASSIGN(BgpXmppChannel);
     383             : };
     384             : 
     385             : class BgpXmppChannelManager {
     386             : public:
     387             :     typedef std::map<const XmppChannel *, BgpXmppChannel *> XmppChannelMap;
     388             :     typedef std::map<std::string, BgpXmppChannel *> XmppChannelNameMap;
     389             :     typedef XmppChannelNameMap::const_iterator const_name_iterator;
     390             :     typedef boost::function<void(BgpXmppChannel *)> VisitorFn;
     391             : 
     392             :     BgpXmppChannelManager(XmppServer *, BgpServer *);
     393             :     virtual ~BgpXmppChannelManager();
     394             : 
     395             :     const_name_iterator name_cbegin() const {
     396             :         return channel_name_map_.begin();
     397             :     }
     398        2506 :     const_name_iterator name_cend() const {
     399        2506 :         return channel_name_map_.end();
     400             :     }
     401         386 :     const_name_iterator name_clower_bound(const std::string &name) const {
     402         386 :         return channel_name_map_.lower_bound(name);
     403             :     }
     404             : 
     405             :     void VisitChannels(BgpXmppChannelManager::VisitorFn);
     406             :     void VisitChannels(BgpXmppChannelManager::VisitorFn) const;
     407             :     BgpXmppChannel *FindChannel(const XmppChannel *channel);
     408             :     BgpXmppChannel *FindChannel(std::string client);
     409             :     void RemoveChannel(XmppChannel *channel);
     410             :     virtual void XmppHandleChannelEvent(XmppChannel *channel,
     411             :                                         xmps::PeerState state);
     412             : 
     413         260 :     const XmppChannelMap &channel_map() const { return channel_map_; }
     414        6178 :     void Enqueue(BgpXmppChannel *bx_channel) {
     415        6178 :         queue_.Enqueue(bx_channel);
     416        6178 :     }
     417             :     bool IsReadyForDeletion();
     418             :     void SetQueueDisable(bool disabled);
     419             :     size_t GetQueueSize() const;
     420             :     void AdminDownCallback();
     421             :     void ASNUpdateCallback(as_t old_asn, as_t old_local_asn);
     422             :     void IdentifierUpdateCallback(Ip4Address old_identifier);
     423             :     void RoutingInstanceCallback(std::string vrf_name, int op);
     424             :     void DSCPUpdateCallback(uint8_t value);
     425             : 
     426         177 :     uint32_t count() const {
     427         177 :         std::scoped_lock lock(mutex_);
     428         354 :         return channel_map_.size();
     429         177 :     }
     430         641 :     uint32_t NumUpPeer() const {
     431         641 :         std::scoped_lock lock(mutex_);
     432        1282 :         return channel_map_.size();
     433         641 :     }
     434             : 
     435         177 :     int32_t deleting_count() const { return deleting_count_; }
     436        6178 :     void increment_deleting_count() { deleting_count_++; }
     437        6178 :     void decrement_deleting_count() {
     438        6178 :         assert(deleting_count_);
     439        6178 :         deleting_count_--;
     440        6178 :     }
     441             : 
     442         343 :     BgpServer *bgp_server() { return bgp_server_; }
     443       18826 :     XmppServer *xmpp_server() { return xmpp_server_; }
     444       19195 :     const XmppServer *xmpp_server() const { return xmpp_server_; }
     445       68251 :     uint64_t get_subscription_gen_id() {
     446      136502 :         return subscription_gen_id_.fetch_add(1);
     447             :     }
     448             :     bool CollectStats(BgpRouterState *state, bool first) const;
     449             : 
     450             : protected:
     451             :     virtual BgpXmppChannel *CreateChannel(XmppChannel *channel);
     452             :     virtual bool DeleteChannel(BgpXmppChannel *channel);
     453             : 
     454             : private:
     455             :     friend class BgpXmppChannelManagerMock;
     456             :     friend class BgpXmppUnitTest;
     457             : 
     458             :     void FillPeerInfo(const BgpXmppChannel *channel) const;
     459             : 
     460             :     XmppServer *xmpp_server_;
     461             :     BgpServer *bgp_server_;
     462             :     WorkQueue<BgpXmppChannel *> queue_;
     463             :     mutable std::mutex mutex_;
     464             :     XmppChannelMap channel_map_;
     465             :     XmppChannelNameMap channel_name_map_;
     466             :     int id_;
     467             :     int admin_down_listener_id_;
     468             :     int asn_listener_id_;
     469             :     int identifier_listener_id_;
     470             :     int dscp_listener_id_;
     471             :     std::atomic<int32_t> deleting_count_;
     472             :     // Generation number for subscription tracking
     473             :     std::atomic<uint64_t> subscription_gen_id_;
     474             : 
     475             :     DISALLOW_COPY_AND_ASSIGN(BgpXmppChannelManager);
     476             : };
     477             : 
     478             : #endif  // SRC_BGP_BGP_XMPP_CHANNEL_H_