[PATCH v2]: teamsyncd

Makefile.am

@@ -1,4 +1,4 @@
-SUBDIRS = common neighsyncd intfsyncd portsyncd orchagent
+SUBDIRS = common neighsyncd intfsyncd portsyncd orchagent teamsyncd
 
 if TESTS
 SUBDIRS += tests

common/select.cpp

@@ -13,6 +13,16 @@ void Select::addSelectable(Selectable *c)
     m_objects.push_back(c);
 }
 
+void Select::removeSelectable(Selectable *c)
+{
+    for (auto i = m_objects.begin(); i != m_objects.end(); i++)
+        if (*i == c)
+        {
+            m_objects.erase(i);
+            return;
+        }
+}
+
 void Select::addFd(int fd)
 {
     m_fds.push_back(fd);

common/select.h

@@ -12,8 +12,9 @@ namespace swss {
 class Select
 {
 public:
-    /* Add object for select */
+    /* Add object for select can be called during Selectable::readMe */
     void addSelectable(Selectable *c);
+    void removeSelectable(Selectable *c);
 
     /* Add file-descriptor for select  */
     void addFd(int fd);

configure.ac

@@ -59,6 +59,7 @@ AC_CONFIG_FILES([
     neighsyncd/Makefile
     intfsyncd/Makefile
     portsyncd/Makefile
+    teamsyncd/Makefile
 ])
 
 AC_OUTPUT

teamsyncd/Makefile.am

@@ -0,0 +1,16 @@
+INCLUDES = -I $(top_srcdir)
+
+bin_PROGRAMS = teamsyncd
+
+if DEBUG
+DBGFLAGS = -ggdb -DDEBUG
+else
+DBGFLAGS = -g
+endif
+
+teamsyncd_SOURCES = teamsyncd.cpp teamsync.cpp
+
+teamsyncd_CFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON)
+teamsyncd_CPPFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON)
+teamsyncd_LDADD = -lnl-3 -lnl-route-3 -lhiredis -lteam \
+	-L$(top_srcdir)/common -lswsscommon

teamsyncd/teamsync.cpp

@@ -0,0 +1,191 @@
+#include <string.h>
+#include <errno.h>
+#include <system_error>
+#include <sys/socket.h>
+#include <linux/if.h>
+#include <netlink/route/link.h>
+#include "common/logger.h"
+#include "common/netmsg.h"
+#include "common/dbconnector.h"
+#include "common/producertable.h"
+#include "common/scheme.h"
+#include "teamsyncd/teamsync.h"
+
+using namespace std;
+using namespace swss;
+
+/* Taken from drivers/net/team/team.c */
+#define TEAM_DRV_NAME "team"
+
+TeamSync::TeamSync(DBConnector *db, Select *select) :
+    m_select(select),
+    m_lagTable(db, APP_LAG_TABLE_NAME)
+{
+}
+
+void TeamSync::onMsg(int nlmsg_type, struct nl_object *obj)
+{
+    struct rtnl_link *link = (struct rtnl_link *)obj;
+    if ((nlmsg_type != RTM_NEWLINK) && (nlmsg_type != RTM_DELLINK))
+        return;
+
+    string lagName = rtnl_link_get_name(link);
+    /* Listens to LAG messages */
+    char *type = rtnl_link_get_type(link);
+    if (!type || (strcmp(type, TEAM_DRV_NAME) != 0))
+        return;
+
+    bool tracked = m_teamPorts.find(lagName) != m_teamPorts.end();
+
+    if ((nlmsg_type == RTM_DELLINK) && tracked)
+    {
+        /* Remove LAG ports and delete LAG */
+        removeLag(lagName);
+        return;
+    }
+
+    if ((nlmsg_type == RTM_NEWLINK) && tracked)
+        return;
+
+    /*
+     * New LAG was dedcated for the first time. Sync admin and oper state since
+     * portsyncd reflects only changes
+     */
+    addLag(lagName, rtnl_link_get_ifindex(link),
+           rtnl_link_get_flags(link) & IFF_UP,
+           rtnl_link_get_flags(link) & IFF_LOWER_UP,
+           rtnl_link_get_mtu(link));
+}
+
+void TeamSync::addLag(const string &lagName, int ifindex, bool admin_state,
+                      bool oper_state, unsigned int mtu)
+{
+    /* First add the LAG itself */
+    std::vector<FieldValueTuple> fvVector;
+    FieldValueTuple a("admin_status", admin_state ? "up" : "down");
+    FieldValueTuple o("oper_status", oper_state ? "up" : "down");
+    FieldValueTuple m("mtu", to_string(mtu));
+    fvVector.push_back(a);
+    fvVector.push_back(o);
+    fvVector.push_back(m);
+    m_lagTable.set(lagName, fvVector);
+
+    /* Start adding ports to LAG */
+    TeamPortSync *sync = new TeamPortSync(lagName, ifindex, &m_lagTable);
+    m_select->addSelectable(sync);
+    m_teamPorts[lagName] = shared_ptr<TeamPortSync>(sync);
+}
+
+void TeamSync::removeLag(const string &lagName)
+{
+    m_select->removeSelectable(m_teamPorts[lagName].get());
+    m_teamPorts.erase(lagName);
+    m_lagTable.del(lagName);
+}
+
+const struct team_change_handler TeamSync::TeamPortSync::gPortChangeHandler = {
+    .func       = TeamSync::TeamPortSync::teamdHandler,
+    .type_mask  = TEAM_PORT_CHANGE
+};
+
+TeamSync::TeamPortSync::TeamPortSync(const string &lagName, int ifindex,
+                                     ProducerTable *lagTable) :
+    m_lagTable(lagTable),
+    m_lagName(lagName),
+    m_ifindex(ifindex)
+{
+    m_team = team_alloc();
+    if (!m_team)
+    {
+        SWSS_LOG_ERROR("Unable to allocated team socket");
+        throw system_error(make_error_code(errc::address_not_available),
+                           "Unable to allocated team socket");
+    }
+
+    int err = team_init(m_team, ifindex);
+    if (err) {
+        team_free(m_team);
+        m_team = NULL;
+        SWSS_LOG_ERROR("Unable to init team socket");
+        throw system_error(make_error_code(errc::address_not_available),
+                           "Unable to init team socket");
+    }
+
+    err = team_change_handler_register(m_team, &gPortChangeHandler, this);
+    if (err) {
+        team_free(m_team);
+        m_team = NULL;
+        SWSS_LOG_ERROR("Unable to register port change event");
+        throw system_error(make_error_code(errc::address_not_available),
+                           "Unable to register port change event");
+    }
+
+    onPortChange(true);
+}
+
+TeamSync::TeamPortSync::~TeamPortSync()
+{
+    if (m_team)
+    {
+        team_change_handler_unregister(m_team, &gPortChangeHandler, this);
+        team_free(m_team);
+    }
+}
+
+int TeamSync::TeamPortSync::onPortChange(bool isInit)
+{
+    struct team_port *port;
+    team_for_each_port(port, m_team)
+    {
+        if (isInit || team_is_port_changed(port))
+        {
+            string key = m_lagName;
+            key += ":";
+            uint32_t ifindex = team_get_port_ifindex(port);
+            char ifname[MAX_IFNAME + 1] = {0};
+            key += team_ifindex2ifname(m_team, ifindex, ifname, MAX_IFNAME);
+
+            if (team_is_port_removed(port))
+            {
+                m_lagTable->del(key);
+            } else
+            {
+                std::vector<FieldValueTuple> fvVector;
+                FieldValueTuple l("linkup", team_is_port_link_up(port) ? "up" : "down");
+                FieldValueTuple s("speed", to_string(team_get_port_speed(port)) + "Mbit");
+                FieldValueTuple d("duplex", team_get_port_duplex(port) ? "full" : "half");
+                fvVector.push_back(l);
+                fvVector.push_back(s);
+                fvVector.push_back(d);
+                m_lagTable->set(key, fvVector);
+            }
+        }
+    }
+    return 0;
+}
+
+int TeamSync::TeamPortSync::teamdHandler(struct team_handle *team, void *arg,
+                                         team_change_type_mask_t type_mask)
+{
+    return ((TeamSync::TeamPortSync *)arg)->onPortChange(false);
+}
+
+void TeamSync::TeamPortSync::addFd(fd_set *fd)
+{
+    FD_SET(team_get_event_fd(m_team), fd);
+}
+
+bool TeamSync::TeamPortSync::isMe(fd_set *fd)
+{
+    return FD_ISSET(team_get_event_fd(m_team), fd);
+}
+
+int TeamSync::TeamPortSync::readCache()
+{
+    return NODATA;
+}
+
+void TeamSync::TeamPortSync::readMe()
+{
+    team_handle_events(m_team);
+}

teamsyncd/teamsync.h

@@ -0,0 +1,65 @@
+#ifndef __TEAMSYNC__
+#define __TEAMSYNC__
+
+#include <map>
+#include <string>
+#include <memory>
+#include "common/dbconnector.h"
+#include "common/producertable.h"
+#include "common/selectable.h"
+#include "common/select.h"
+#include "common/netmsg.h"
+#include <team.h>
+
+namespace swss {
+
+class TeamSync : public NetMsg
+{
+public:
+    TeamSync(DBConnector *db, Select *select);
+
+    /*
+     * Listens to RTM_NEWLINK and RTM_DELLINK to undestand if there is a new
+     * team device
+     */
+    virtual void onMsg(int nlmsg_type, struct nl_object *obj);
+
+    class TeamPortSync : public Selectable
+    {
+    public:
+        enum { MAX_IFNAME = 64 };
+        TeamPortSync(const std::string &lagName, int ifindex,
+                     ProducerTable *lagTable);
+        ~TeamPortSync();
+
+        virtual void addFd(fd_set *fd);
+        virtual bool isMe(fd_set *fd);
+        virtual int readCache();
+        virtual void readMe();
+
+    protected:
+        int onPortChange(bool isInit);
+        static int teamdHandler(struct team_handle *th, void *arg,
+                                team_change_type_mask_t type_mask);
+        static const struct team_change_handler gPortChangeHandler;
+    private:
+        ProducerTable *m_lagTable;
+        struct team_handle *m_team;
+        std::string m_lagName;
+        int m_ifindex;
+    };
+
+protected:
+    void addLag(const std::string &lagName, int ifindex, bool admin_state,
+                bool oper_state, unsigned int mtu);
+    void removeLag(const std::string &lagName);
+
+private:
+    Select *m_select;
+    ProducerTable m_lagTable;
+    std::map<std::string, std::shared_ptr<TeamPortSync> > m_teamPorts;
+};
+
+}
+
+#endif

teamsyncd/teamsyncd.cpp

@@ -0,0 +1,47 @@
+#include <iostream>
+#include <team.h>
+#include "common/logger.h"
+#include "common/select.h"
+#include "common/netdispatcher.h"
+#include "common/netlink.h"
+#include "teamsyncd/teamsync.h"
+
+using namespace std;
+using namespace swss;
+
+int main(int argc, char **argv)
+{
+    DBConnector db(APPL_DB, "localhost", 6379, 0);
+    Select s;
+    TeamSync sync(&db, &s);
+
+    NetDispatcher::getInstance().registerMessageHandler(RTM_NEWLINK, &sync);
+    NetDispatcher::getInstance().registerMessageHandler(RTM_DELLINK, &sync);
+
+    while (1)
+    {
+        try
+        {
+            NetLink netlink;
+
+            netlink.registerGroup(RTNLGRP_LINK);
+            cout << "Listens to teamd events..." << endl;
+            netlink.dumpRequest(RTM_GETLINK);
+
+            s.addSelectable(&netlink);
+            while (true)
+            {
+                Selectable *temps;
+                int tempfd;
+                s.select(&temps, &tempfd);
+            }
+        }
+        catch (...)
+        {
+            cout << "Exception had been thrown in deamon" << endl;
+            return 0;
+        }
+    }
+
+    return 1;
+}