Add app source, add vendoring and module support

vendor/github.com/osrg/gobgp/pkg/server/bmp.go

@@ -0,0 +1,358 @@
+// Copyright (C) 2015-2016 Nippon Telegraph and Telephone Corporation.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"fmt"
+	"net"
+	"strconv"
+	"time"
+
+	"github.com/osrg/gobgp/internal/pkg/config"
+	"github.com/osrg/gobgp/internal/pkg/table"
+	"github.com/osrg/gobgp/pkg/packet/bgp"
+	"github.com/osrg/gobgp/pkg/packet/bmp"
+	log "github.com/sirupsen/logrus"
+)
+
+type ribout map[string][]*table.Path
+
+func newribout() ribout {
+	return make(map[string][]*table.Path)
+}
+
+// return true if we need to send the path to the BMP server
+func (r ribout) update(p *table.Path) bool {
+	key := p.GetNlri().String() // TODO expose (*Path).getPrefix()
+	l := r[key]
+	if p.IsWithdraw {
+		if len(l) == 0 {
+			return false
+		}
+		n := make([]*table.Path, 0, len(l))
+		for _, q := range l {
+			if p.GetSource() == q.GetSource() {
+				continue
+			}
+			n = append(n, q)
+		}
+		if len(n) == 0 {
+			delete(r, key)
+		} else {
+			r[key] = n
+		}
+		return true
+	}
+
+	if len(l) == 0 {
+		r[key] = []*table.Path{p}
+		return true
+	}
+
+	doAppend := true
+	for idx, q := range l {
+		if p.GetSource() == q.GetSource() {
+			// if we have sent the same path, don't send it again
+			if p.Equal(q) {
+				return false
+			}
+			l[idx] = p
+			doAppend = false
+		}
+	}
+	if doAppend {
+		r[key] = append(r[key], p)
+	}
+	return true
+}
+
+func (b *bmpClient) tryConnect() *net.TCPConn {
+	interval := 1
+	for {
+		log.WithFields(log.Fields{"Topic": "bmp"}).Debugf("Connecting BMP server:%s", b.host)
+		conn, err := net.Dial("tcp", b.host)
+		if err != nil {
+			select {
+			case <-b.dead:
+				return nil
+			default:
+			}
+			time.Sleep(time.Duration(interval) * time.Second)
+			if interval < 30 {
+				interval *= 2
+			}
+		} else {
+			log.WithFields(log.Fields{"Topic": "bmp"}).Infof("BMP server is connected:%s", b.host)
+			return conn.(*net.TCPConn)
+		}
+	}
+}
+
+func (b *bmpClient) Stop() {
+	close(b.dead)
+}
+
+func (b *bmpClient) loop() {
+	for {
+		conn := b.tryConnect()
+		if conn == nil {
+			break
+		}
+
+		if func() bool {
+			ops := []WatchOption{WatchPeerState(true)}
+			if b.c.RouteMonitoringPolicy == config.BMP_ROUTE_MONITORING_POLICY_TYPE_BOTH {
+				log.WithFields(
+					log.Fields{"Topic": "bmp"},
+				).Warn("both option for route-monitoring-policy is obsoleted")
+			}
+			if b.c.RouteMonitoringPolicy == config.BMP_ROUTE_MONITORING_POLICY_TYPE_PRE_POLICY || b.c.RouteMonitoringPolicy == config.BMP_ROUTE_MONITORING_POLICY_TYPE_ALL {
+				ops = append(ops, WatchUpdate(true))
+			}
+			if b.c.RouteMonitoringPolicy == config.BMP_ROUTE_MONITORING_POLICY_TYPE_POST_POLICY || b.c.RouteMonitoringPolicy == config.BMP_ROUTE_MONITORING_POLICY_TYPE_ALL {
+				ops = append(ops, WatchPostUpdate(true))
+			}
+			if b.c.RouteMonitoringPolicy == config.BMP_ROUTE_MONITORING_POLICY_TYPE_LOCAL_RIB || b.c.RouteMonitoringPolicy == config.BMP_ROUTE_MONITORING_POLICY_TYPE_ALL {
+				ops = append(ops, WatchBestPath(true))
+			}
+			if b.c.RouteMirroringEnabled {
+				ops = append(ops, WatchMessage(false))
+			}
+			w := b.s.Watch(ops...)
+			defer w.Stop()
+
+			var tickerCh <-chan time.Time
+			if b.c.StatisticsTimeout == 0 {
+				log.WithFields(log.Fields{"Topic": "bmp"}).Debug("statistics reports disabled")
+			} else {
+				t := time.NewTicker(time.Duration(b.c.StatisticsTimeout) * time.Second)
+				defer t.Stop()
+				tickerCh = t.C
+			}
+
+			write := func(msg *bmp.BMPMessage) error {
+				buf, _ := msg.Serialize()
+				_, err := conn.Write(buf)
+				if err != nil {
+					log.Warnf("failed to write to bmp server %s", b.host)
+				}
+				return err
+			}
+
+			if err := write(bmp.NewBMPInitiation([]bmp.BMPInfoTLVInterface{})); err != nil {
+				return false
+			}
+
+			for {
+				select {
+				case ev := <-w.Event():
+					switch msg := ev.(type) {
+					case *WatchEventUpdate:
+						info := &table.PeerInfo{
+							Address: msg.PeerAddress,
+							AS:      msg.PeerAS,
+							ID:      msg.PeerID,
+						}
+						if msg.Payload == nil {
+							var pathList []*table.Path
+							if msg.Init {
+								pathList = msg.PathList
+							} else {
+								for _, p := range msg.PathList {
+									if b.ribout.update(p) {
+										pathList = append(pathList, p)
+									}
+								}
+							}
+							for _, path := range pathList {
+								for _, u := range table.CreateUpdateMsgFromPaths([]*table.Path{path}) {
+									payload, _ := u.Serialize()
+									if err := write(bmpPeerRoute(bmp.BMP_PEER_TYPE_GLOBAL, msg.PostPolicy, 0, true, info, msg.Timestamp.Unix(), payload)); err != nil {
+										return false
+									}
+								}
+							}
+						} else {
+							if err := write(bmpPeerRoute(bmp.BMP_PEER_TYPE_GLOBAL, msg.PostPolicy, 0, msg.FourBytesAs, info, msg.Timestamp.Unix(), msg.Payload)); err != nil {
+								return false
+							}
+						}
+					case *WatchEventBestPath:
+						info := &table.PeerInfo{
+							Address: net.ParseIP("0.0.0.0").To4(),
+							AS:      b.s.bgpConfig.Global.Config.As,
+							ID:      net.ParseIP(b.s.bgpConfig.Global.Config.RouterId).To4(),
+						}
+						for _, p := range msg.PathList {
+							u := table.CreateUpdateMsgFromPaths([]*table.Path{p})[0]
+							if payload, err := u.Serialize(); err != nil {
+								return false
+							} else if err = write(bmpPeerRoute(bmp.BMP_PEER_TYPE_LOCAL_RIB, false, 0, true, info, p.GetTimestamp().Unix(), payload)); err != nil {
+								return false
+							}
+						}
+					case *WatchEventPeerState:
+						if msg.State == bgp.BGP_FSM_ESTABLISHED {
+							if err := write(bmpPeerUp(msg, bmp.BMP_PEER_TYPE_GLOBAL, false, 0)); err != nil {
+								return false
+							}
+						} else {
+							if err := write(bmpPeerDown(msg, bmp.BMP_PEER_TYPE_GLOBAL, false, 0)); err != nil {
+								return false
+							}
+						}
+					case *WatchEventMessage:
+						info := &table.PeerInfo{
+							Address: msg.PeerAddress,
+							AS:      msg.PeerAS,
+							ID:      msg.PeerID,
+						}
+						if err := write(bmpPeerRouteMirroring(bmp.BMP_PEER_TYPE_GLOBAL, 0, info, msg.Timestamp.Unix(), msg.Message)); err != nil {
+							return false
+						}
+					}
+				case <-tickerCh:
+					neighborList := b.s.getNeighbor("", true)
+					for _, n := range neighborList {
+						if n.State.SessionState != config.SESSION_STATE_ESTABLISHED {
+							continue
+						}
+						if err := write(bmpPeerStats(bmp.BMP_PEER_TYPE_GLOBAL, 0, 0, n)); err != nil {
+							return false
+						}
+					}
+				case <-b.dead:
+					term := bmp.NewBMPTermination([]bmp.BMPTermTLVInterface{
+						bmp.NewBMPTermTLV16(bmp.BMP_TERM_TLV_TYPE_REASON, bmp.BMP_TERM_REASON_PERMANENTLY_ADMIN),
+					})
+					if err := write(term); err != nil {
+						return false
+					}
+					conn.Close()
+					return true
+				}
+			}
+		}() {
+			return
+		}
+	}
+}
+
+type bmpClient struct {
+	s      *BgpServer
+	dead   chan struct{}
+	host   string
+	c      *config.BmpServerConfig
+	ribout ribout
+}
+
+func bmpPeerUp(ev *WatchEventPeerState, t uint8, policy bool, pd uint64) *bmp.BMPMessage {
+	var flags uint8 = 0
+	if policy {
+		flags |= bmp.BMP_PEER_FLAG_POST_POLICY
+	}
+	ph := bmp.NewBMPPeerHeader(t, flags, pd, ev.PeerAddress.String(), ev.PeerAS, ev.PeerID.String(), float64(ev.Timestamp.Unix()))
+	return bmp.NewBMPPeerUpNotification(*ph, ev.LocalAddress.String(), ev.LocalPort, ev.PeerPort, ev.SentOpen, ev.RecvOpen)
+}
+
+func bmpPeerDown(ev *WatchEventPeerState, t uint8, policy bool, pd uint64) *bmp.BMPMessage {
+	var flags uint8 = 0
+	if policy {
+		flags |= bmp.BMP_PEER_FLAG_POST_POLICY
+	}
+	ph := bmp.NewBMPPeerHeader(t, flags, pd, ev.PeerAddress.String(), ev.PeerAS, ev.PeerID.String(), float64(ev.Timestamp.Unix()))
+	return bmp.NewBMPPeerDownNotification(*ph, uint8(ev.StateReason.PeerDownReason), ev.StateReason.BGPNotification, ev.StateReason.Data)
+}
+
+func bmpPeerRoute(t uint8, policy bool, pd uint64, fourBytesAs bool, peeri *table.PeerInfo, timestamp int64, payload []byte) *bmp.BMPMessage {
+	var flags uint8 = 0
+	if policy {
+		flags |= bmp.BMP_PEER_FLAG_POST_POLICY
+	}
+	if !fourBytesAs {
+		flags |= bmp.BMP_PEER_FLAG_TWO_AS
+	}
+	ph := bmp.NewBMPPeerHeader(t, flags, pd, peeri.Address.String(), peeri.AS, peeri.ID.String(), float64(timestamp))
+	m := bmp.NewBMPRouteMonitoring(*ph, nil)
+	body := m.Body.(*bmp.BMPRouteMonitoring)
+	body.BGPUpdatePayload = payload
+	return m
+}
+
+func bmpPeerStats(peerType uint8, peerDist uint64, timestamp int64, neighConf *config.Neighbor) *bmp.BMPMessage {
+	var peerFlags uint8 = 0
+	ph := bmp.NewBMPPeerHeader(peerType, peerFlags, peerDist, neighConf.State.NeighborAddress, neighConf.State.PeerAs, neighConf.State.RemoteRouterId, float64(timestamp))
+	return bmp.NewBMPStatisticsReport(
+		*ph,
+		[]bmp.BMPStatsTLVInterface{
+			bmp.NewBMPStatsTLV64(bmp.BMP_STAT_TYPE_ADJ_RIB_IN, uint64(neighConf.State.AdjTable.Accepted)),
+			bmp.NewBMPStatsTLV64(bmp.BMP_STAT_TYPE_LOC_RIB, uint64(neighConf.State.AdjTable.Advertised+neighConf.State.AdjTable.Filtered)),
+			bmp.NewBMPStatsTLV32(bmp.BMP_STAT_TYPE_WITHDRAW_UPDATE, neighConf.State.Messages.Received.WithdrawUpdate),
+			bmp.NewBMPStatsTLV32(bmp.BMP_STAT_TYPE_WITHDRAW_PREFIX, neighConf.State.Messages.Received.WithdrawPrefix),
+		},
+	)
+}
+
+func bmpPeerRouteMirroring(peerType uint8, peerDist uint64, peerInfo *table.PeerInfo, timestamp int64, msg *bgp.BGPMessage) *bmp.BMPMessage {
+	var peerFlags uint8 = 0
+	ph := bmp.NewBMPPeerHeader(peerType, peerFlags, peerDist, peerInfo.Address.String(), peerInfo.AS, peerInfo.ID.String(), float64(timestamp))
+	return bmp.NewBMPRouteMirroring(
+		*ph,
+		[]bmp.BMPRouteMirrTLVInterface{
+			// RFC7854: BGP Message TLV MUST occur last in the list of TLVs
+			bmp.NewBMPRouteMirrTLVBGPMsg(bmp.BMP_ROUTE_MIRRORING_TLV_TYPE_BGP_MSG, msg),
+		},
+	)
+}
+
+func (b *bmpClientManager) addServer(c *config.BmpServerConfig) error {
+	host := net.JoinHostPort(c.Address, strconv.Itoa(int(c.Port)))
+	if _, y := b.clientMap[host]; y {
+		return fmt.Errorf("bmp client %s is already configured", host)
+	}
+	b.clientMap[host] = &bmpClient{
+		s:      b.s,
+		dead:   make(chan struct{}),
+		host:   host,
+		c:      c,
+		ribout: newribout(),
+	}
+	go b.clientMap[host].loop()
+	return nil
+}
+
+func (b *bmpClientManager) deleteServer(c *config.BmpServerConfig) error {
+	host := net.JoinHostPort(c.Address, strconv.Itoa(int(c.Port)))
+	if c, y := b.clientMap[host]; !y {
+		return fmt.Errorf("bmp client %s isn't found", host)
+	} else {
+		c.Stop()
+		delete(b.clientMap, host)
+	}
+	return nil
+}
+
+type bmpClientManager struct {
+	s         *BgpServer
+	clientMap map[string]*bmpClient
+}
+
+func newBmpClientManager(s *BgpServer) *bmpClientManager {
+	return &bmpClientManager{
+		s:         s,
+		clientMap: make(map[string]*bmpClient),
+	}
+}

vendor/github.com/osrg/gobgp/pkg/server/collector.go

@@ -0,0 +1,222 @@
+// Copyright (C) 2016 Nippon Telegraph and Telephone Corporation.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/influxdata/influxdb/client/v2"
+	"github.com/osrg/gobgp/internal/pkg/table"
+	"github.com/osrg/gobgp/pkg/packet/bgp"
+	log "github.com/sirupsen/logrus"
+)
+
+type Collector struct {
+	s        *BgpServer
+	url      string
+	dbName   string
+	interval uint64
+	client   client.Client
+}
+
+const (
+	MEATUREMENT_UPDATE = "update"
+	MEATUREMENT_PEER   = "peer"
+	MEATUREMENT_TABLE  = "table"
+)
+
+func (c *Collector) writePoints(points []*client.Point) error {
+	bp, _ := client.NewBatchPoints(client.BatchPointsConfig{
+		Database:  c.dbName,
+		Precision: "ms",
+	})
+	bp.AddPoints(points)
+	return c.client.Write(bp)
+}
+
+func (c *Collector) writePeer(msg *WatchEventPeerState) error {
+	var state string
+	switch msg.State {
+	case bgp.BGP_FSM_ESTABLISHED:
+		state = "Established"
+	case bgp.BGP_FSM_IDLE:
+		state = "Idle"
+	default:
+		return fmt.Errorf("unexpected fsm state %v", msg.State)
+	}
+
+	tags := map[string]string{
+		"PeerAddress": msg.PeerAddress.String(),
+		"PeerAS":      fmt.Sprintf("%v", msg.PeerAS),
+		"State":       state,
+	}
+
+	fields := map[string]interface{}{
+		"PeerID": msg.PeerID.String(),
+	}
+
+	pt, err := client.NewPoint(MEATUREMENT_PEER, tags, fields, msg.Timestamp)
+	if err != nil {
+		return err
+	}
+	return c.writePoints([]*client.Point{pt})
+}
+
+func path2data(path *table.Path) (map[string]interface{}, map[string]string) {
+	fields := map[string]interface{}{
+		"RouterID": path.GetSource().ID,
+	}
+	if asPath := path.GetAsPath(); asPath != nil {
+		fields["ASPath"] = asPath.String()
+	}
+	if origin, err := path.GetOrigin(); err == nil {
+		typ := "-"
+		switch origin {
+		case bgp.BGP_ORIGIN_ATTR_TYPE_IGP:
+			typ = "i"
+		case bgp.BGP_ORIGIN_ATTR_TYPE_EGP:
+			typ = "e"
+		case bgp.BGP_ORIGIN_ATTR_TYPE_INCOMPLETE:
+			typ = "?"
+		}
+		fields["Origin"] = typ
+	}
+	if med, err := path.GetMed(); err == nil {
+		fields["Med"] = med
+	}
+
+	tags := map[string]string{
+		"PeerAddress": path.GetSource().Address.String(),
+		"PeerAS":      fmt.Sprintf("%v", path.GetSource().AS),
+		"Timestamp":   path.GetTimestamp().String(),
+	}
+	if nexthop := path.GetNexthop(); len(nexthop) > 0 {
+		fields["NextHop"] = nexthop.String()
+	}
+	if originAS := path.GetSourceAs(); originAS != 0 {
+		fields["OriginAS"] = fmt.Sprintf("%v", originAS)
+	}
+
+	if err := bgp.FlatUpdate(tags, path.GetNlri().Flat()); err != nil {
+		log.WithFields(log.Fields{"Type": "collector", "Error": err}).Error("NLRI FlatUpdate failed")
+	}
+	for _, p := range path.GetPathAttrs() {
+		if err := bgp.FlatUpdate(tags, p.Flat()); err != nil {
+			log.WithFields(log.Fields{"Type": "collector", "Error": err}).Error("PathAttr FlatUpdate failed")
+		}
+	}
+	return fields, tags
+}
+
+func (c *Collector) writeUpdate(msg *WatchEventUpdate) error {
+	if len(msg.PathList) == 0 {
+		// EOR
+		return nil
+	}
+	now := time.Now()
+	points := make([]*client.Point, 0, len(msg.PathList))
+	for _, path := range msg.PathList {
+		fields, tags := path2data(path)
+		tags["Withdraw"] = fmt.Sprintf("%v", path.IsWithdraw)
+		pt, err := client.NewPoint(MEATUREMENT_UPDATE, tags, fields, now)
+		if err != nil {
+			return fmt.Errorf("failed to write update, %v", err)
+		}
+		points = append(points, pt)
+	}
+	return c.writePoints(points)
+}
+
+func (c *Collector) writeTable(msg *WatchEventAdjIn) error {
+	now := time.Now()
+	points := make([]*client.Point, 0, len(msg.PathList))
+	for _, path := range msg.PathList {
+		fields, tags := path2data(path)
+		pt, err := client.NewPoint(MEATUREMENT_TABLE, tags, fields, now)
+		if err != nil {
+			return fmt.Errorf("failed to write table, %v", err)
+		}
+		points = append(points, pt)
+	}
+	return c.writePoints(points)
+}
+
+func (c *Collector) loop() {
+	w := c.s.Watch(WatchPeerState(true), WatchUpdate(false))
+	defer w.Stop()
+
+	ticker := func() *time.Ticker {
+		if c.interval == 0 {
+			return &time.Ticker{}
+		}
+		return time.NewTicker(time.Second * time.Duration(c.interval))
+	}()
+
+	for {
+		select {
+		case <-ticker.C:
+			w.Generate(WATCH_EVENT_TYPE_PRE_UPDATE)
+		case ev := <-w.Event():
+			switch msg := ev.(type) {
+			case *WatchEventUpdate:
+				if err := c.writeUpdate(msg); err != nil {
+					log.WithFields(log.Fields{"Type": "collector", "Error": err}).Error("Failed to write update event message")
+				}
+			case *WatchEventPeerState:
+				if err := c.writePeer(msg); err != nil {
+					log.WithFields(log.Fields{"Type": "collector", "Error": err}).Error("Failed to write state changed event message")
+				}
+			case *WatchEventAdjIn:
+				if err := c.writeTable(msg); err != nil {
+					log.WithFields(log.Fields{"Type": "collector", "Error": err}).Error("Failed to write Adj-In event message")
+				}
+			}
+		}
+	}
+}
+
+func NewCollector(s *BgpServer, url, dbName string, interval uint64) (*Collector, error) {
+	c, err := client.NewHTTPClient(client.HTTPConfig{
+		Addr: url,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	_, _, err = c.Ping(0)
+	if err != nil {
+		log.Error("can not connect to InfluxDB")
+		log.WithFields(log.Fields{"Type": "collector", "Error": err}).Error("Failed to connect to InfluxDB")
+		return nil, err
+	}
+
+	q := client.NewQuery("CREATE DATABASE "+dbName, "", "")
+	if response, err := c.Query(q); err != nil || response.Error() != nil {
+		log.WithFields(log.Fields{"Type": "collector", "Error": err}).Errorf("Failed to create database:%s", dbName)
+		return nil, err
+	}
+
+	collector := &Collector{
+		s:        s,
+		url:      url,
+		dbName:   dbName,
+		interval: interval,
+		client:   c,
+	}
+	go collector.loop()
+	return collector, nil
+}

vendor/github.com/osrg/gobgp/pkg/server/mrt.go

@@ -0,0 +1,409 @@
+// Copyright (C) 2016 Nippon Telegraph and Telephone Corporation.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"bytes"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/osrg/gobgp/internal/pkg/config"
+	"github.com/osrg/gobgp/internal/pkg/table"
+	"github.com/osrg/gobgp/pkg/packet/bgp"
+	"github.com/osrg/gobgp/pkg/packet/mrt"
+
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	MIN_ROTATION_INTERVAL = 60
+	MIN_DUMP_INTERVAL     = 60
+)
+
+type mrtWriter struct {
+	dead             chan struct{}
+	s                *BgpServer
+	c                *config.MrtConfig
+	file             *os.File
+	rotationInterval uint64
+	dumpInterval     uint64
+}
+
+func (m *mrtWriter) Stop() {
+	close(m.dead)
+}
+
+func (m *mrtWriter) loop() error {
+	ops := []WatchOption{}
+	switch m.c.DumpType {
+	case config.MRT_TYPE_UPDATES:
+		ops = append(ops, WatchUpdate(false))
+	case config.MRT_TYPE_TABLE:
+		if len(m.c.TableName) > 0 {
+			ops = append(ops, WatchTableName(m.c.TableName))
+		}
+	}
+	w := m.s.Watch(ops...)
+	rotator := func() *time.Ticker {
+		if m.rotationInterval == 0 {
+			return &time.Ticker{}
+		}
+		return time.NewTicker(time.Second * time.Duration(m.rotationInterval))
+	}()
+	dump := func() *time.Ticker {
+		if m.dumpInterval == 0 {
+			return &time.Ticker{}
+		}
+		return time.NewTicker(time.Second * time.Duration(m.dumpInterval))
+	}()
+
+	defer func() {
+		if m.file != nil {
+			m.file.Close()
+		}
+		if m.rotationInterval != 0 {
+			rotator.Stop()
+		}
+		if m.dumpInterval != 0 {
+			dump.Stop()
+		}
+		w.Stop()
+	}()
+
+	for {
+		serialize := func(ev WatchEvent) []*mrt.MRTMessage {
+			msg := make([]*mrt.MRTMessage, 0, 1)
+			switch e := ev.(type) {
+			case *WatchEventUpdate:
+				if e.Init {
+					return nil
+				}
+				mp := mrt.NewBGP4MPMessage(e.PeerAS, e.LocalAS, 0, e.PeerAddress.String(), e.LocalAddress.String(), e.FourBytesAs, nil)
+				mp.BGPMessagePayload = e.Payload
+				isAddPath := e.Neighbor.IsAddPathReceiveEnabled(e.PathList[0].GetRouteFamily())
+				subtype := mrt.MESSAGE
+				switch {
+				case isAddPath && e.FourBytesAs:
+					subtype = mrt.MESSAGE_AS4_ADDPATH
+				case isAddPath:
+					subtype = mrt.MESSAGE_ADDPATH
+				case e.FourBytesAs:
+					subtype = mrt.MESSAGE_AS4
+				}
+				if bm, err := mrt.NewMRTMessage(uint32(e.Timestamp.Unix()), mrt.BGP4MP, subtype, mp); err != nil {
+					log.WithFields(log.Fields{
+						"Topic": "mrt",
+						"Data":  e,
+						"Error": err,
+					}).Warnf("Failed to create MRT BGP4MP message (subtype %d)", subtype)
+				} else {
+					msg = append(msg, bm)
+				}
+			case *WatchEventTable:
+				t := uint32(time.Now().Unix())
+
+				peers := make([]*mrt.Peer, 1, len(e.Neighbor)+1)
+				// Adding dummy Peer record for locally generated routes
+				peers[0] = mrt.NewPeer("0.0.0.0", "0.0.0.0", 0, true)
+				neighborMap := make(map[string]*config.Neighbor)
+				for _, pconf := range e.Neighbor {
+					peers = append(peers, mrt.NewPeer(pconf.State.RemoteRouterId, pconf.State.NeighborAddress, pconf.Config.PeerAs, true))
+					neighborMap[pconf.State.NeighborAddress] = pconf
+				}
+
+				if bm, err := mrt.NewMRTMessage(t, mrt.TABLE_DUMPv2, mrt.PEER_INDEX_TABLE, mrt.NewPeerIndexTable(e.RouterId, "", peers)); err != nil {
+					log.WithFields(log.Fields{
+						"Topic": "mrt",
+						"Data":  e,
+						"Error": err,
+					}).Warnf("Failed to create MRT TABLE_DUMPv2 message (subtype %d)", mrt.PEER_INDEX_TABLE)
+					break
+				} else {
+					msg = append(msg, bm)
+				}
+
+				idx := func(p *table.Path) uint16 {
+					for i, pconf := range e.Neighbor {
+						if p.GetSource().Address.String() == pconf.State.NeighborAddress {
+							return uint16(i)
+						}
+					}
+					return uint16(len(e.Neighbor))
+				}
+
+				subtype := func(p *table.Path, isAddPath bool) mrt.MRTSubTypeTableDumpv2 {
+					t := mrt.RIB_GENERIC
+					switch p.GetRouteFamily() {
+					case bgp.RF_IPv4_UC:
+						t = mrt.RIB_IPV4_UNICAST
+					case bgp.RF_IPv4_MC:
+						t = mrt.RIB_IPV4_MULTICAST
+					case bgp.RF_IPv6_UC:
+						t = mrt.RIB_IPV6_UNICAST
+					case bgp.RF_IPv6_MC:
+						t = mrt.RIB_IPV6_MULTICAST
+					}
+					if isAddPath {
+						// Shift non-additional-path version to *_ADDPATH
+						t += 6
+					}
+					return t
+				}
+
+				seq := uint32(0)
+				appendTableDumpMsg := func(path *table.Path, entries []*mrt.RibEntry, isAddPath bool) {
+					st := subtype(path, isAddPath)
+					if bm, err := mrt.NewMRTMessage(t, mrt.TABLE_DUMPv2, st, mrt.NewRib(seq, path.GetNlri(), entries)); err != nil {
+						log.WithFields(log.Fields{
+							"Topic": "mrt",
+							"Data":  e,
+							"Error": err,
+						}).Warnf("Failed to create MRT TABLE_DUMPv2 message (subtype %d)", st)
+					} else {
+						msg = append(msg, bm)
+						seq++
+					}
+				}
+				for _, pathList := range e.PathList {
+					entries := make([]*mrt.RibEntry, 0, len(pathList))
+					entriesAddPath := make([]*mrt.RibEntry, 0, len(pathList))
+					for _, path := range pathList {
+						isAddPath := false
+						if path.IsLocal() {
+							isAddPath = true
+						} else if neighbor, ok := neighborMap[path.GetSource().Address.String()]; ok {
+							isAddPath = neighbor.IsAddPathReceiveEnabled(path.GetRouteFamily())
+						}
+						if !isAddPath {
+							entries = append(entries, mrt.NewRibEntry(idx(path), uint32(path.GetTimestamp().Unix()), 0, path.GetPathAttrs(), false))
+						} else {
+							entriesAddPath = append(entriesAddPath, mrt.NewRibEntry(idx(path), uint32(path.GetTimestamp().Unix()), path.GetNlri().PathIdentifier(), path.GetPathAttrs(), true))
+						}
+					}
+					if len(entries) > 0 {
+						appendTableDumpMsg(pathList[0], entries, false)
+					}
+					if len(entriesAddPath) > 0 {
+						appendTableDumpMsg(pathList[0], entriesAddPath, true)
+					}
+				}
+			}
+			return msg
+		}
+
+		drain := func(ev WatchEvent) {
+			events := make([]WatchEvent, 0, 1+len(w.Event()))
+			if ev != nil {
+				events = append(events, ev)
+			}
+
+			for len(w.Event()) > 0 {
+				events = append(events, <-w.Event())
+			}
+
+			w := func(buf []byte) {
+				if _, err := m.file.Write(buf); err == nil {
+					m.file.Sync()
+				} else {
+					log.WithFields(log.Fields{
+						"Topic": "mrt",
+						"Error": err,
+					}).Warn("Can't write to destination MRT file")
+				}
+			}
+
+			var b bytes.Buffer
+			for _, e := range events {
+				for _, m := range serialize(e) {
+					if buf, err := m.Serialize(); err != nil {
+						log.WithFields(log.Fields{
+							"Topic": "mrt",
+							"Data":  e,
+							"Error": err,
+						}).Warn("Failed to serialize event")
+					} else {
+						b.Write(buf)
+						if b.Len() > 1*1000*1000 {
+							w(b.Bytes())
+							b.Reset()
+						}
+					}
+				}
+			}
+			if b.Len() > 0 {
+				w(b.Bytes())
+			}
+		}
+		rotate := func() {
+			m.file.Close()
+			file, err := mrtFileOpen(m.c.FileName, m.rotationInterval)
+			if err == nil {
+				m.file = file
+			} else {
+				log.WithFields(log.Fields{
+					"Topic": "mrt",
+					"Error": err,
+				}).Warn("can't rotate MRT file")
+			}
+		}
+
+		select {
+		case <-m.dead:
+			drain(nil)
+			return nil
+		case e := <-w.Event():
+			drain(e)
+			if m.c.DumpType == config.MRT_TYPE_TABLE && m.rotationInterval != 0 {
+				rotate()
+			}
+		case <-rotator.C:
+			if m.c.DumpType == config.MRT_TYPE_UPDATES {
+				rotate()
+			} else {
+				w.Generate(WATCH_EVENT_TYPE_TABLE)
+			}
+		case <-dump.C:
+			w.Generate(WATCH_EVENT_TYPE_TABLE)
+		}
+	}
+}
+
+func mrtFileOpen(filename string, interval uint64) (*os.File, error) {
+	realname := filename
+	if interval != 0 {
+		realname = time.Now().Format(filename)
+	}
+	log.WithFields(log.Fields{
+		"Topic":         "mrt",
+		"Filename":      realname,
+		"Dump Interval": interval,
+	}).Debug("Setting new MRT destination file")
+
+	i := len(realname)
+	for i > 0 && os.IsPathSeparator(realname[i-1]) {
+		// skip trailing path separators
+		i--
+	}
+	j := i
+
+	for j > 0 && !os.IsPathSeparator(realname[j-1]) {
+		j--
+	}
+
+	if j > 0 {
+		if err := os.MkdirAll(realname[0:j-1], 0755); err != nil {
+			log.WithFields(log.Fields{
+				"Topic": "mrt",
+				"Error": err,
+			}).Warn("can't create MRT destination directory")
+			return nil, err
+		}
+	}
+
+	file, err := os.OpenFile(realname, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0644)
+	if err != nil {
+		log.WithFields(log.Fields{
+			"Topic": "mrt",
+			"Error": err,
+		}).Warn("can't create MRT destination file")
+	}
+	return file, err
+}
+
+func newMrtWriter(s *BgpServer, c *config.MrtConfig, rInterval, dInterval uint64) (*mrtWriter, error) {
+	file, err := mrtFileOpen(c.FileName, rInterval)
+	if err != nil {
+		return nil, err
+	}
+	m := mrtWriter{
+		s:                s,
+		c:                c,
+		file:             file,
+		rotationInterval: rInterval,
+		dumpInterval:     dInterval,
+	}
+	go m.loop()
+	return &m, nil
+}
+
+type mrtManager struct {
+	bgpServer *BgpServer
+	writer    map[string]*mrtWriter
+}
+
+func (m *mrtManager) enable(c *config.MrtConfig) error {
+	if _, ok := m.writer[c.FileName]; ok {
+		return fmt.Errorf("%s already exists", c.FileName)
+	}
+
+	rInterval := c.RotationInterval
+	dInterval := c.DumpInterval
+
+	setRotationMin := func() {
+		if rInterval < MIN_ROTATION_INTERVAL {
+			log.WithFields(log.Fields{
+				"Topic": "MRT",
+			}).Infof("minimum mrt rotation interval is %d seconds", MIN_ROTATION_INTERVAL)
+			rInterval = MIN_ROTATION_INTERVAL
+		}
+	}
+
+	if c.DumpType == config.MRT_TYPE_TABLE {
+		if rInterval == 0 {
+			if dInterval < MIN_DUMP_INTERVAL {
+				log.WithFields(log.Fields{
+					"Topic": "MRT",
+				}).Infof("minimum mrt dump interval is %d seconds", MIN_DUMP_INTERVAL)
+				dInterval = MIN_DUMP_INTERVAL
+			}
+		} else if dInterval == 0 {
+			setRotationMin()
+		} else {
+			return fmt.Errorf("can't specify both intervals in the table dump type")
+		}
+	} else if c.DumpType == config.MRT_TYPE_UPDATES {
+		// ignore the dump interval
+		dInterval = 0
+		if len(c.TableName) > 0 {
+			return fmt.Errorf("can't specify the table name with the update dump type")
+		}
+		setRotationMin()
+	}
+
+	w, err := newMrtWriter(m.bgpServer, c, rInterval, dInterval)
+	if err == nil {
+		m.writer[c.FileName] = w
+	}
+	return err
+}
+
+func (m *mrtManager) disable(c *config.MrtConfig) error {
+	w, ok := m.writer[c.FileName]
+	if !ok {
+		return fmt.Errorf("%s doesn't exists", c.FileName)
+	}
+	w.Stop()
+	delete(m.writer, c.FileName)
+	return nil
+}
+
+func newMrtManager(s *BgpServer) *mrtManager {
+	return &mrtManager{
+		bgpServer: s,
+		writer:    make(map[string]*mrtWriter),
+	}
+}

vendor/github.com/osrg/gobgp/pkg/server/peer.go

@@ -0,0 +1,596 @@
+// Copyright (C) 2014-2016 Nippon Telegraph and Telephone Corporation.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"fmt"
+	"net"
+	"time"
+
+	"github.com/osrg/gobgp/internal/pkg/config"
+	"github.com/osrg/gobgp/internal/pkg/table"
+	"github.com/osrg/gobgp/pkg/packet/bgp"
+
+	"github.com/eapache/channels"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	FLOP_THRESHOLD    = time.Second * 30
+	MIN_CONNECT_RETRY = 10
+)
+
+type PeerGroup struct {
+	Conf             *config.PeerGroup
+	members          map[string]config.Neighbor
+	dynamicNeighbors map[string]*config.DynamicNeighbor
+}
+
+func NewPeerGroup(c *config.PeerGroup) *PeerGroup {
+	return &PeerGroup{
+		Conf:             c,
+		members:          make(map[string]config.Neighbor),
+		dynamicNeighbors: make(map[string]*config.DynamicNeighbor),
+	}
+}
+
+func (pg *PeerGroup) AddMember(c config.Neighbor) {
+	pg.members[c.State.NeighborAddress] = c
+}
+
+func (pg *PeerGroup) DeleteMember(c config.Neighbor) {
+	delete(pg.members, c.State.NeighborAddress)
+}
+
+func (pg *PeerGroup) AddDynamicNeighbor(c *config.DynamicNeighbor) {
+	pg.dynamicNeighbors[c.Config.Prefix] = c
+}
+
+func newDynamicPeer(g *config.Global, neighborAddress string, pg *config.PeerGroup, loc *table.TableManager, policy *table.RoutingPolicy) *Peer {
+	conf := config.Neighbor{
+		Config: config.NeighborConfig{
+			PeerGroup: pg.Config.PeerGroupName,
+		},
+		State: config.NeighborState{
+			NeighborAddress: neighborAddress,
+		},
+		Transport: config.Transport{
+			Config: config.TransportConfig{
+				PassiveMode: true,
+			},
+		},
+	}
+	if err := config.OverwriteNeighborConfigWithPeerGroup(&conf, pg); err != nil {
+		log.WithFields(log.Fields{
+			"Topic": "Peer",
+			"Key":   neighborAddress,
+		}).Debugf("Can't overwrite neighbor config: %s", err)
+		return nil
+	}
+	if err := config.SetDefaultNeighborConfigValues(&conf, pg, g); err != nil {
+		log.WithFields(log.Fields{
+			"Topic": "Peer",
+			"Key":   neighborAddress,
+		}).Debugf("Can't set default config: %s", err)
+		return nil
+	}
+	peer := NewPeer(g, &conf, loc, policy)
+	peer.fsm.lock.Lock()
+	peer.fsm.state = bgp.BGP_FSM_ACTIVE
+	peer.fsm.lock.Unlock()
+	return peer
+}
+
+type Peer struct {
+	tableId           string
+	fsm               *FSM
+	adjRibIn          *table.AdjRib
+	outgoing          *channels.InfiniteChannel
+	policy            *table.RoutingPolicy
+	localRib          *table.TableManager
+	prefixLimitWarned map[bgp.RouteFamily]bool
+	llgrEndChs        []chan struct{}
+}
+
+func NewPeer(g *config.Global, conf *config.Neighbor, loc *table.TableManager, policy *table.RoutingPolicy) *Peer {
+	peer := &Peer{
+		outgoing:          channels.NewInfiniteChannel(),
+		localRib:          loc,
+		policy:            policy,
+		fsm:               NewFSM(g, conf, policy),
+		prefixLimitWarned: make(map[bgp.RouteFamily]bool),
+	}
+	if peer.isRouteServerClient() {
+		peer.tableId = conf.State.NeighborAddress
+	} else {
+		peer.tableId = table.GLOBAL_RIB_NAME
+	}
+	rfs, _ := config.AfiSafis(conf.AfiSafis).ToRfList()
+	peer.adjRibIn = table.NewAdjRib(rfs)
+	return peer
+}
+
+func (peer *Peer) AS() uint32 {
+	peer.fsm.lock.RLock()
+	defer peer.fsm.lock.RUnlock()
+	return peer.fsm.pConf.State.PeerAs
+}
+
+func (peer *Peer) ID() string {
+	peer.fsm.lock.RLock()
+	defer peer.fsm.lock.RUnlock()
+	return peer.fsm.pConf.State.NeighborAddress
+}
+
+func (peer *Peer) TableID() string {
+	return peer.tableId
+}
+
+func (peer *Peer) isIBGPPeer() bool {
+	return peer.fsm.pConf.State.PeerAs == peer.fsm.gConf.Config.As
+}
+
+func (peer *Peer) isRouteServerClient() bool {
+	peer.fsm.lock.RLock()
+	defer peer.fsm.lock.RUnlock()
+	return peer.fsm.pConf.RouteServer.Config.RouteServerClient
+}
+
+func (peer *Peer) isRouteReflectorClient() bool {
+	peer.fsm.lock.RLock()
+	defer peer.fsm.lock.RUnlock()
+	return peer.fsm.pConf.RouteReflector.Config.RouteReflectorClient
+}
+
+func (peer *Peer) isGracefulRestartEnabled() bool {
+	peer.fsm.lock.RLock()
+	defer peer.fsm.lock.RUnlock()
+	return peer.fsm.pConf.GracefulRestart.State.Enabled
+}
+
+func (peer *Peer) getAddPathMode(family bgp.RouteFamily) bgp.BGPAddPathMode {
+	peer.fsm.lock.RLock()
+	defer peer.fsm.lock.RUnlock()
+	if mode, y := peer.fsm.rfMap[family]; y {
+		return mode
+	}
+	return bgp.BGP_ADD_PATH_NONE
+}
+
+func (peer *Peer) isAddPathReceiveEnabled(family bgp.RouteFamily) bool {
+	return (peer.getAddPathMode(family) & bgp.BGP_ADD_PATH_RECEIVE) > 0
+}
+
+func (peer *Peer) isAddPathSendEnabled(family bgp.RouteFamily) bool {
+	return (peer.getAddPathMode(family) & bgp.BGP_ADD_PATH_SEND) > 0
+}
+
+func (peer *Peer) isDynamicNeighbor() bool {
+	peer.fsm.lock.RLock()
+	defer peer.fsm.lock.RUnlock()
+	return peer.fsm.pConf.Config.NeighborAddress == "" && peer.fsm.pConf.Config.NeighborInterface == ""
+}
+
+func (peer *Peer) recvedAllEOR() bool {
+	peer.fsm.lock.RLock()
+	defer peer.fsm.lock.RUnlock()
+	for _, a := range peer.fsm.pConf.AfiSafis {
+		if s := a.MpGracefulRestart.State; s.Enabled && !s.EndOfRibReceived {
+			return false
+		}
+	}
+	return true
+}
+
+func (peer *Peer) configuredRFlist() []bgp.RouteFamily {
+	peer.fsm.lock.RLock()
+	defer peer.fsm.lock.RUnlock()
+	rfs, _ := config.AfiSafis(peer.fsm.pConf.AfiSafis).ToRfList()
+	return rfs
+}
+
+func (peer *Peer) negotiatedRFList() []bgp.RouteFamily {
+	peer.fsm.lock.RLock()
+	defer peer.fsm.lock.RUnlock()
+	l := make([]bgp.RouteFamily, 0, len(peer.fsm.rfMap))
+	for family := range peer.fsm.rfMap {
+		l = append(l, family)
+	}
+	return l
+}
+
+func (peer *Peer) toGlobalFamilies(families []bgp.RouteFamily) []bgp.RouteFamily {
+	peer.fsm.lock.RLock()
+	defer peer.fsm.lock.RUnlock()
+	if peer.fsm.pConf.Config.Vrf != "" {
+		fs := make([]bgp.RouteFamily, 0, len(families))
+		for _, f := range families {
+			switch f {
+			case bgp.RF_IPv4_UC:
+				fs = append(fs, bgp.RF_IPv4_VPN)
+			case bgp.RF_IPv6_UC:
+				fs = append(fs, bgp.RF_IPv6_VPN)
+			default:
+				log.WithFields(log.Fields{
+					"Topic":  "Peer",
+					"Key":    peer.ID(),
+					"Family": f,
+					"VRF":    peer.fsm.pConf.Config.Vrf,
+				}).Warn("invalid family configured for neighbor with vrf")
+			}
+		}
+		families = fs
+	}
+	return families
+}
+
+func classifyFamilies(all, part []bgp.RouteFamily) ([]bgp.RouteFamily, []bgp.RouteFamily) {
+	a := []bgp.RouteFamily{}
+	b := []bgp.RouteFamily{}
+	for _, f := range all {
+		p := true
+		for _, g := range part {
+			if f == g {
+				p = false
+				a = append(a, f)
+				break
+			}
+		}
+		if p {
+			b = append(b, f)
+		}
+	}
+	return a, b
+}
+
+func (peer *Peer) forwardingPreservedFamilies() ([]bgp.RouteFamily, []bgp.RouteFamily) {
+	peer.fsm.lock.RLock()
+	list := []bgp.RouteFamily{}
+	for _, a := range peer.fsm.pConf.AfiSafis {
+		if s := a.MpGracefulRestart.State; s.Enabled && s.Received {
+			list = append(list, a.State.Family)
+		}
+	}
+	peer.fsm.lock.RUnlock()
+	return classifyFamilies(peer.configuredRFlist(), list)
+}
+
+func (peer *Peer) llgrFamilies() ([]bgp.RouteFamily, []bgp.RouteFamily) {
+	peer.fsm.lock.RLock()
+	list := []bgp.RouteFamily{}
+	for _, a := range peer.fsm.pConf.AfiSafis {
+		if a.LongLivedGracefulRestart.State.Enabled {
+			list = append(list, a.State.Family)
+		}
+	}
+	peer.fsm.lock.RUnlock()
+	return classifyFamilies(peer.configuredRFlist(), list)
+}
+
+func (peer *Peer) isLLGREnabledFamily(family bgp.RouteFamily) bool {
+	peer.fsm.lock.RLock()
+	llgrEnabled := peer.fsm.pConf.GracefulRestart.Config.LongLivedEnabled
+	peer.fsm.lock.RUnlock()
+	if !llgrEnabled {
+		return false
+	}
+	fs, _ := peer.llgrFamilies()
+	for _, f := range fs {
+		if f == family {
+			return true
+		}
+	}
+	return false
+}
+
+func (peer *Peer) llgrRestartTime(family bgp.RouteFamily) uint32 {
+	peer.fsm.lock.RLock()
+	defer peer.fsm.lock.RUnlock()
+	for _, a := range peer.fsm.pConf.AfiSafis {
+		if a.State.Family == family {
+			return a.LongLivedGracefulRestart.State.PeerRestartTime
+		}
+	}
+	return 0
+}
+
+func (peer *Peer) llgrRestartTimerExpired(family bgp.RouteFamily) bool {
+	peer.fsm.lock.RLock()
+	defer peer.fsm.lock.RUnlock()
+	all := true
+	for _, a := range peer.fsm.pConf.AfiSafis {
+		if a.State.Family == family {
+			a.LongLivedGracefulRestart.State.PeerRestartTimerExpired = true
+		}
+		s := a.LongLivedGracefulRestart.State
+		if s.Received && !s.PeerRestartTimerExpired {
+			all = false
+		}
+	}
+	return all
+}
+
+func (peer *Peer) markLLGRStale(fs []bgp.RouteFamily) []*table.Path {
+	paths := peer.adjRibIn.PathList(fs, true)
+	for i, p := range paths {
+		doStale := true
+		for _, c := range p.GetCommunities() {
+			if c == uint32(bgp.COMMUNITY_NO_LLGR) {
+				doStale = false
+				p = p.Clone(true)
+				break
+			}
+		}
+		if doStale {
+			p = p.Clone(false)
+			p.SetCommunities([]uint32{uint32(bgp.COMMUNITY_LLGR_STALE)}, false)
+		}
+		paths[i] = p
+	}
+	return paths
+}
+
+func (peer *Peer) stopPeerRestarting() {
+	peer.fsm.lock.Lock()
+	defer peer.fsm.lock.Unlock()
+	peer.fsm.pConf.GracefulRestart.State.PeerRestarting = false
+	for _, ch := range peer.llgrEndChs {
+		close(ch)
+	}
+	peer.llgrEndChs = make([]chan struct{}, 0)
+
+}
+
+func (peer *Peer) filterPathFromSourcePeer(path, old *table.Path) *table.Path {
+	if peer.ID() != path.GetSource().Address.String() {
+		return path
+	}
+
+	// Note: Multiple paths having the same prefix could exist the withdrawals
+	// list in the case of Route Server setup with import policies modifying
+	// paths. In such case, gobgp sends duplicated update messages; withdraw
+	// messages for the same prefix.
+	if !peer.isRouteServerClient() {
+		if peer.isRouteReflectorClient() && path.GetRouteFamily() == bgp.RF_RTC_UC {
+			// When the peer is a Route Reflector client and the given path
+			// contains the Route Tartget Membership NLRI, the path should not
+			// be withdrawn in order to signal the client to distribute routes
+			// with the specific RT to Route Reflector.
+			return path
+		} else if !path.IsWithdraw && old != nil && old.GetSource().Address.String() != peer.ID() {
+			// Say, peer A and B advertized same prefix P, and best path
+			// calculation chose a path from B as best. When B withdraws prefix
+			// P, best path calculation chooses the path from A as best. For
+			// peers other than A, this path should be advertised (as implicit
+			// withdrawal). However for A, we should advertise the withdrawal
+			// path. Thing is same when peer A and we advertized prefix P (as
+			// local route), then, we withdraws the prefix.
+			return old.Clone(true)
+		}
+	}
+	log.WithFields(log.Fields{
+		"Topic": "Peer",
+		"Key":   peer.ID(),
+		"Data":  path,
+	}).Debug("From me, ignore.")
+	return nil
+}
+
+func (peer *Peer) doPrefixLimit(k bgp.RouteFamily, c *config.PrefixLimitConfig) *bgp.BGPMessage {
+	if maxPrefixes := int(c.MaxPrefixes); maxPrefixes > 0 {
+		count := peer.adjRibIn.Count([]bgp.RouteFamily{k})
+		pct := int(c.ShutdownThresholdPct)
+		if pct > 0 && !peer.prefixLimitWarned[k] && count > (maxPrefixes*pct/100) {
+			peer.prefixLimitWarned[k] = true
+			log.WithFields(log.Fields{
+				"Topic":         "Peer",
+				"Key":           peer.ID(),
+				"AddressFamily": k.String(),
+			}).Warnf("prefix limit %d%% reached", pct)
+		}
+		if count > maxPrefixes {
+			log.WithFields(log.Fields{
+				"Topic":         "Peer",
+				"Key":           peer.ID(),
+				"AddressFamily": k.String(),
+			}).Warnf("prefix limit reached")
+			return bgp.NewBGPNotificationMessage(bgp.BGP_ERROR_CEASE, bgp.BGP_ERROR_SUB_MAXIMUM_NUMBER_OF_PREFIXES_REACHED, nil)
+		}
+	}
+	return nil
+
+}
+
+func (peer *Peer) updatePrefixLimitConfig(c []config.AfiSafi) error {
+	peer.fsm.lock.RLock()
+	x := peer.fsm.pConf.AfiSafis
+	peer.fsm.lock.RUnlock()
+	y := c
+	if len(x) != len(y) {
+		return fmt.Errorf("changing supported afi-safi is not allowed")
+	}
+	m := make(map[bgp.RouteFamily]config.PrefixLimitConfig)
+	for _, e := range x {
+		m[e.State.Family] = e.PrefixLimit.Config
+	}
+	for _, e := range y {
+		if p, ok := m[e.State.Family]; !ok {
+			return fmt.Errorf("changing supported afi-safi is not allowed")
+		} else if !p.Equal(&e.PrefixLimit.Config) {
+			log.WithFields(log.Fields{
+				"Topic":                   "Peer",
+				"Key":                     peer.ID(),
+				"AddressFamily":           e.Config.AfiSafiName,
+				"OldMaxPrefixes":          p.MaxPrefixes,
+				"NewMaxPrefixes":          e.PrefixLimit.Config.MaxPrefixes,
+				"OldShutdownThresholdPct": p.ShutdownThresholdPct,
+				"NewShutdownThresholdPct": e.PrefixLimit.Config.ShutdownThresholdPct,
+			}).Warnf("update prefix limit configuration")
+			peer.prefixLimitWarned[e.State.Family] = false
+			if msg := peer.doPrefixLimit(e.State.Family, &e.PrefixLimit.Config); msg != nil {
+				sendFsmOutgoingMsg(peer, nil, msg, true)
+			}
+		}
+	}
+	peer.fsm.lock.Lock()
+	peer.fsm.pConf.AfiSafis = c
+	peer.fsm.lock.Unlock()
+	return nil
+}
+
+func (peer *Peer) handleUpdate(e *FsmMsg) ([]*table.Path, []bgp.RouteFamily, *bgp.BGPMessage) {
+	m := e.MsgData.(*bgp.BGPMessage)
+	update := m.Body.(*bgp.BGPUpdate)
+	log.WithFields(log.Fields{
+		"Topic":       "Peer",
+		"Key":         peer.fsm.pConf.State.NeighborAddress,
+		"nlri":        update.NLRI,
+		"withdrawals": update.WithdrawnRoutes,
+		"attributes":  update.PathAttributes,
+	}).Debug("received update")
+	peer.fsm.lock.Lock()
+	peer.fsm.pConf.Timers.State.UpdateRecvTime = time.Now().Unix()
+	peer.fsm.lock.Unlock()
+	if len(e.PathList) > 0 {
+		paths := make([]*table.Path, 0, len(e.PathList))
+		eor := []bgp.RouteFamily{}
+		for _, path := range e.PathList {
+			if path.IsEOR() {
+				family := path.GetRouteFamily()
+				log.WithFields(log.Fields{
+					"Topic":         "Peer",
+					"Key":           peer.ID(),
+					"AddressFamily": family,
+				}).Debug("EOR received")
+				eor = append(eor, family)
+				continue
+			}
+			// RFC4271 9.1.2 Phase 2: Route Selection
+			//
+			// If the AS_PATH attribute of a BGP route contains an AS loop, the BGP
+			// route should be excluded from the Phase 2 decision function.
+			if aspath := path.GetAsPath(); aspath != nil {
+				peer.fsm.lock.RLock()
+				localAS := peer.fsm.peerInfo.LocalAS
+				allowOwnAS := int(peer.fsm.pConf.AsPathOptions.Config.AllowOwnAs)
+				peer.fsm.lock.RUnlock()
+				if hasOwnASLoop(localAS, allowOwnAS, aspath) {
+					path.SetAsLooped(true)
+					continue
+				}
+			}
+			// RFC4456 8. Avoiding Routing Information Loops
+			// A router that recognizes the ORIGINATOR_ID attribute SHOULD
+			// ignore a route received with its BGP Identifier as the ORIGINATOR_ID.
+			peer.fsm.lock.RLock()
+			isIBGPPeer := peer.isIBGPPeer()
+			routerId := peer.fsm.gConf.Config.RouterId
+			peer.fsm.lock.RUnlock()
+			if isIBGPPeer {
+				if id := path.GetOriginatorID(); routerId == id.String() {
+					log.WithFields(log.Fields{
+						"Topic":        "Peer",
+						"Key":          peer.ID(),
+						"OriginatorID": id,
+						"Data":         path,
+					}).Debug("Originator ID is mine, ignore")
+					continue
+				}
+			}
+			paths = append(paths, path)
+		}
+		peer.adjRibIn.Update(e.PathList)
+		peer.fsm.lock.RLock()
+		peerAfiSafis := peer.fsm.pConf.AfiSafis
+		peer.fsm.lock.RUnlock()
+		for _, af := range peerAfiSafis {
+			if msg := peer.doPrefixLimit(af.State.Family, &af.PrefixLimit.Config); msg != nil {
+				return nil, nil, msg
+			}
+		}
+		return paths, eor, nil
+	}
+	return nil, nil, nil
+}
+
+func (peer *Peer) startFSMHandler(incoming *channels.InfiniteChannel, stateCh chan *FsmMsg) {
+	handler := NewFSMHandler(peer.fsm, incoming, stateCh, peer.outgoing)
+	peer.fsm.lock.Lock()
+	peer.fsm.h = handler
+	peer.fsm.lock.Unlock()
+}
+
+func (peer *Peer) StaleAll(rfList []bgp.RouteFamily) []*table.Path {
+	return peer.adjRibIn.StaleAll(rfList)
+}
+
+func (peer *Peer) PassConn(conn *net.TCPConn) {
+	select {
+	case peer.fsm.connCh <- conn:
+	default:
+		conn.Close()
+		log.WithFields(log.Fields{
+			"Topic": "Peer",
+			"Key":   peer.ID(),
+		}).Warn("accepted conn is closed to avoid be blocked")
+	}
+}
+
+func (peer *Peer) DropAll(rfList []bgp.RouteFamily) {
+	peer.adjRibIn.Drop(rfList)
+}
+
+func (peer *Peer) stopFSM() error {
+	failed := false
+	peer.fsm.lock.RLock()
+	addr := peer.fsm.pConf.State.NeighborAddress
+	peer.fsm.lock.RUnlock()
+	t1 := time.AfterFunc(time.Minute*5, func() {
+		log.WithFields(log.Fields{
+			"Topic": "Peer",
+		}).Warnf("Failed to free the fsm.h.t for %s", addr)
+		failed = true
+	})
+
+	peer.fsm.h.t.Kill(nil)
+	peer.fsm.h.t.Wait()
+	t1.Stop()
+	if !failed {
+		log.WithFields(log.Fields{
+			"Topic": "Peer",
+			"Key":   addr,
+		}).Debug("freed fsm.h.t")
+		cleanInfiniteChannel(peer.outgoing)
+	}
+	failed = false
+	t2 := time.AfterFunc(time.Minute*5, func() {
+		log.WithFields(log.Fields{
+			"Topic": "Peer",
+		}).Warnf("Failed to free the fsm.t for %s", addr)
+		failed = true
+	})
+	peer.fsm.t.Kill(nil)
+	peer.fsm.t.Wait()
+	t2.Stop()
+	if !failed {
+		log.WithFields(log.Fields{
+			"Topic": "Peer",
+			"Key":   addr,
+		}).Debug("freed fsm.t")
+		return nil
+	}
+	return fmt.Errorf("Failed to free FSM for %s", addr)
+}

vendor/github.com/osrg/gobgp/pkg/server/rpki.go

@@ -0,0 +1,712 @@
+// Copyright (C) 2015,2016 Nippon Telegraph and Telephone Corporation.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"encoding/binary"
+	"fmt"
+	"io"
+	"net"
+	"sort"
+	"strconv"
+	"time"
+
+	"github.com/osrg/gobgp/internal/pkg/config"
+	"github.com/osrg/gobgp/internal/pkg/table"
+	"github.com/osrg/gobgp/pkg/packet/bgp"
+	"github.com/osrg/gobgp/pkg/packet/rtr"
+
+	"github.com/armon/go-radix"
+	log "github.com/sirupsen/logrus"
+	"golang.org/x/net/context"
+)
+
+const (
+	CONNECT_RETRY_INTERVAL = 30
+)
+
+func before(a, b uint32) bool {
+	return int32(a-b) < 0
+}
+
+type RoaBucket struct {
+	Prefix  *table.IPPrefix
+	entries []*table.ROA
+}
+
+func (r *RoaBucket) GetEntries() []*table.ROA {
+	return r.entries
+}
+
+type roas []*table.ROA
+
+func (r roas) Len() int {
+	return len(r)
+}
+
+func (r roas) Swap(i, j int) {
+	r[i], r[j] = r[j], r[i]
+}
+
+func (r roas) Less(i, j int) bool {
+	r1 := r[i]
+	r2 := r[j]
+
+	if r1.MaxLen < r2.MaxLen {
+		return true
+	} else if r1.MaxLen > r2.MaxLen {
+		return false
+	}
+
+	if r1.AS < r2.AS {
+		return true
+	}
+	return false
+}
+
+type ROAEventType uint8
+
+const (
+	CONNECTED ROAEventType = iota
+	DISCONNECTED
+	RTR
+	LIFETIMEOUT
+)
+
+type ROAEvent struct {
+	EventType ROAEventType
+	Src       string
+	Data      []byte
+	conn      *net.TCPConn
+}
+
+type roaManager struct {
+	AS        uint32
+	Roas      map[bgp.RouteFamily]*radix.Tree
+	eventCh   chan *ROAEvent
+	clientMap map[string]*roaClient
+}
+
+func NewROAManager(as uint32) (*roaManager, error) {
+	m := &roaManager{
+		AS:   as,
+		Roas: make(map[bgp.RouteFamily]*radix.Tree),
+	}
+	m.Roas[bgp.RF_IPv4_UC] = radix.New()
+	m.Roas[bgp.RF_IPv6_UC] = radix.New()
+	m.eventCh = make(chan *ROAEvent)
+	m.clientMap = make(map[string]*roaClient)
+	return m, nil
+}
+
+func (c *roaManager) enabled() bool {
+	return len(c.clientMap) != 0
+}
+
+func (m *roaManager) SetAS(as uint32) error {
+	if m.AS != 0 {
+		return fmt.Errorf("AS was already configured")
+	}
+	m.AS = as
+	return nil
+}
+
+func (m *roaManager) AddServer(host string, lifetime int64) error {
+	address, port, err := net.SplitHostPort(host)
+	if err != nil {
+		return err
+	}
+	if lifetime == 0 {
+		lifetime = 3600
+	}
+	if _, ok := m.clientMap[host]; ok {
+		return fmt.Errorf("ROA server exists %s", host)
+	}
+	m.clientMap[host] = NewRoaClient(address, port, m.eventCh, lifetime)
+	return nil
+}
+
+func (m *roaManager) DeleteServer(host string) error {
+	client, ok := m.clientMap[host]
+	if !ok {
+		return fmt.Errorf("ROA server doesn't exists %s", host)
+	}
+	client.stop()
+	m.deleteAllROA(host)
+	delete(m.clientMap, host)
+	return nil
+}
+
+func (m *roaManager) deleteAllROA(network string) {
+	for _, tree := range m.Roas {
+		deleteKeys := make([]string, 0, tree.Len())
+		tree.Walk(func(s string, v interface{}) bool {
+			b, _ := v.(*RoaBucket)
+			newEntries := make([]*table.ROA, 0, len(b.entries))
+			for _, r := range b.entries {
+				if r.Src != network {
+					newEntries = append(newEntries, r)
+				}
+			}
+			if len(newEntries) > 0 {
+				b.entries = newEntries
+			} else {
+				deleteKeys = append(deleteKeys, s)
+			}
+			return false
+		})
+		for _, key := range deleteKeys {
+			tree.Delete(key)
+		}
+	}
+}
+
+func (m *roaManager) Enable(address string) error {
+	for network, client := range m.clientMap {
+		add, _, _ := net.SplitHostPort(network)
+		if add == address {
+			client.enable(client.serialNumber)
+			return nil
+		}
+	}
+	return fmt.Errorf("ROA server not found %s", address)
+}
+
+func (m *roaManager) Disable(address string) error {
+	for network, client := range m.clientMap {
+		add, _, _ := net.SplitHostPort(network)
+		if add == address {
+			client.reset()
+			m.deleteAllROA(add)
+			return nil
+		}
+	}
+	return fmt.Errorf("ROA server not found %s", address)
+}
+
+func (m *roaManager) Reset(address string) error {
+	return m.Disable(address)
+}
+
+func (m *roaManager) SoftReset(address string) error {
+	for network, client := range m.clientMap {
+		add, _, _ := net.SplitHostPort(network)
+		if add == address {
+			client.softReset()
+			m.deleteAllROA(network)
+			return nil
+		}
+	}
+	return fmt.Errorf("ROA server not found %s", address)
+}
+
+func (c *roaManager) ReceiveROA() chan *ROAEvent {
+	return c.eventCh
+}
+
+func (c *roaClient) lifetimeout() {
+	c.eventCh <- &ROAEvent{
+		EventType: LIFETIMEOUT,
+		Src:       c.host,
+	}
+}
+
+func (m *roaManager) HandleROAEvent(ev *ROAEvent) {
+	client, y := m.clientMap[ev.Src]
+	if !y {
+		if ev.EventType == CONNECTED {
+			ev.conn.Close()
+		}
+		log.WithFields(log.Fields{"Topic": "rpki"}).Errorf("Can't find %s ROA server configuration", ev.Src)
+		return
+	}
+	switch ev.EventType {
+	case DISCONNECTED:
+		log.WithFields(log.Fields{"Topic": "rpki"}).Infof("ROA server %s is disconnected", ev.Src)
+		client.state.Downtime = time.Now().Unix()
+		// clear state
+		client.endOfData = false
+		client.pendingROAs = make([]*table.ROA, 0)
+		client.state.RpkiMessages = config.RpkiMessages{}
+		client.conn = nil
+		go client.tryConnect()
+		client.timer = time.AfterFunc(time.Duration(client.lifetime)*time.Second, client.lifetimeout)
+		client.oldSessionID = client.sessionID
+	case CONNECTED:
+		log.WithFields(log.Fields{"Topic": "rpki"}).Infof("ROA server %s is connected", ev.Src)
+		client.conn = ev.conn
+		client.state.Uptime = time.Now().Unix()
+		go client.established()
+	case RTR:
+		m.handleRTRMsg(client, &client.state, ev.Data)
+	case LIFETIMEOUT:
+		// a) already reconnected but hasn't received
+		// EndOfData -> needs to delete stale ROAs
+		// b) not reconnected -> needs to delete stale ROAs
+		//
+		// c) already reconnected and received EndOfData so
+		// all stale ROAs were deleted -> timer was cancelled
+		// so should not be here.
+		if client.oldSessionID != client.sessionID {
+			log.WithFields(log.Fields{"Topic": "rpki"}).Infof("Reconnected to %s. Ignore timeout", client.host)
+		} else {
+			log.WithFields(log.Fields{"Topic": "rpki"}).Infof("Deleting all ROAs due to timeout with:%s", client.host)
+			m.deleteAllROA(client.host)
+		}
+	}
+}
+
+func (m *roaManager) roa2tree(roa *table.ROA) (*radix.Tree, string) {
+	tree := m.Roas[bgp.RF_IPv4_UC]
+	if roa.Family == bgp.AFI_IP6 {
+		tree = m.Roas[bgp.RF_IPv6_UC]
+	}
+	return tree, table.IpToRadixkey(roa.Prefix.Prefix, roa.Prefix.Length)
+}
+
+func (m *roaManager) deleteROA(roa *table.ROA) {
+	tree, key := m.roa2tree(roa)
+	b, _ := tree.Get(key)
+	if b != nil {
+		bucket := b.(*RoaBucket)
+		newEntries := make([]*table.ROA, 0, len(bucket.entries))
+		for _, r := range bucket.entries {
+			if !r.Equal(roa) {
+				newEntries = append(newEntries, r)
+			}
+		}
+		if len(newEntries) != len(bucket.entries) {
+			bucket.entries = newEntries
+			if len(newEntries) == 0 {
+				tree.Delete(key)
+			}
+			return
+		}
+	}
+	log.WithFields(log.Fields{
+		"Topic":         "rpki",
+		"Prefix":        roa.Prefix.Prefix.String(),
+		"Prefix Length": roa.Prefix.Length,
+		"AS":            roa.AS,
+		"Max Length":    roa.MaxLen,
+	}).Info("Can't withdraw a ROA")
+}
+
+func (m *roaManager) DeleteROA(roa *table.ROA) {
+	m.deleteROA(roa)
+}
+
+func (m *roaManager) addROA(roa *table.ROA) {
+	tree, key := m.roa2tree(roa)
+	b, _ := tree.Get(key)
+	var bucket *RoaBucket
+	if b == nil {
+		bucket = &RoaBucket{
+			Prefix:  roa.Prefix,
+			entries: make([]*table.ROA, 0),
+		}
+		tree.Insert(key, bucket)
+	} else {
+		bucket = b.(*RoaBucket)
+		for _, r := range bucket.entries {
+			if r.Equal(roa) {
+				// we already have the same one
+				return
+			}
+		}
+	}
+	bucket.entries = append(bucket.entries, roa)
+}
+
+func (m *roaManager) AddROA(roa *table.ROA) {
+	m.addROA(roa)
+}
+
+func (c *roaManager) handleRTRMsg(client *roaClient, state *config.RpkiServerState, buf []byte) {
+	received := &state.RpkiMessages.RpkiReceived
+
+	m, err := rtr.ParseRTR(buf)
+	if err == nil {
+		switch msg := m.(type) {
+		case *rtr.RTRSerialNotify:
+			if before(client.serialNumber, msg.RTRCommon.SerialNumber) {
+				client.enable(client.serialNumber)
+			} else if client.serialNumber == msg.RTRCommon.SerialNumber {
+				// nothing
+			} else {
+				// should not happen. try to get the whole ROAs.
+				client.softReset()
+			}
+			received.SerialNotify++
+		case *rtr.RTRSerialQuery:
+		case *rtr.RTRResetQuery:
+		case *rtr.RTRCacheResponse:
+			received.CacheResponse++
+			client.endOfData = false
+		case *rtr.RTRIPPrefix:
+			family := bgp.AFI_IP
+			if msg.Type == rtr.RTR_IPV4_PREFIX {
+				received.Ipv4Prefix++
+			} else {
+				family = bgp.AFI_IP6
+				received.Ipv6Prefix++
+			}
+			roa := table.NewROA(family, msg.Prefix, msg.PrefixLen, msg.MaxLen, msg.AS, client.host)
+			if (msg.Flags & 1) == 1 {
+				if client.endOfData {
+					c.addROA(roa)
+				} else {
+					client.pendingROAs = append(client.pendingROAs, roa)
+				}
+			} else {
+				c.deleteROA(roa)
+			}
+		case *rtr.RTREndOfData:
+			received.EndOfData++
+			if client.sessionID != msg.RTRCommon.SessionID {
+				// remove all ROAs related with the
+				// previous session
+				c.deleteAllROA(client.host)
+			}
+			client.sessionID = msg.RTRCommon.SessionID
+			client.serialNumber = msg.RTRCommon.SerialNumber
+			client.endOfData = true
+			if client.timer != nil {
+				client.timer.Stop()
+				client.timer = nil
+			}
+			for _, roa := range client.pendingROAs {
+				c.addROA(roa)
+			}
+			client.pendingROAs = make([]*table.ROA, 0)
+		case *rtr.RTRCacheReset:
+			client.softReset()
+			received.CacheReset++
+		case *rtr.RTRErrorReport:
+			received.Error++
+		}
+	} else {
+		log.WithFields(log.Fields{
+			"Topic": "rpki",
+			"Host":  client.host,
+			"Error": err,
+		}).Info("Failed to parse an RTR message")
+	}
+}
+
+func (c *roaManager) GetServers() []*config.RpkiServer {
+	f := func(tree *radix.Tree) (map[string]uint32, map[string]uint32) {
+		records := make(map[string]uint32)
+		prefixes := make(map[string]uint32)
+
+		tree.Walk(func(s string, v interface{}) bool {
+			b, _ := v.(*RoaBucket)
+			tmpRecords := make(map[string]uint32)
+			for _, roa := range b.entries {
+				tmpRecords[roa.Src]++
+			}
+
+			for src, r := range tmpRecords {
+				if r > 0 {
+					records[src] += r
+					prefixes[src]++
+				}
+			}
+			return false
+		})
+		return records, prefixes
+	}
+
+	recordsV4, prefixesV4 := f(c.Roas[bgp.RF_IPv4_UC])
+	recordsV6, prefixesV6 := f(c.Roas[bgp.RF_IPv6_UC])
+
+	l := make([]*config.RpkiServer, 0, len(c.clientMap))
+	for _, client := range c.clientMap {
+		state := &client.state
+
+		if client.conn == nil {
+			state.Up = false
+		} else {
+			state.Up = true
+		}
+		f := func(m map[string]uint32, key string) uint32 {
+			if r, ok := m[key]; ok {
+				return r
+			}
+			return 0
+		}
+		state.RecordsV4 = f(recordsV4, client.host)
+		state.RecordsV6 = f(recordsV6, client.host)
+		state.PrefixesV4 = f(prefixesV4, client.host)
+		state.PrefixesV6 = f(prefixesV6, client.host)
+		state.SerialNumber = client.serialNumber
+
+		addr, port, _ := net.SplitHostPort(client.host)
+		l = append(l, &config.RpkiServer{
+			Config: config.RpkiServerConfig{
+				Address: addr,
+				// Note: RpkiServerConfig.Port is uint32 type, but the TCP/UDP
+				// port is 16-bit length.
+				Port: func() uint32 { p, _ := strconv.ParseUint(port, 10, 16); return uint32(p) }(),
+			},
+			State: client.state,
+		})
+	}
+	return l
+}
+
+func (c *roaManager) GetRoa(family bgp.RouteFamily) ([]*table.ROA, error) {
+	if len(c.clientMap) == 0 {
+		return []*table.ROA{}, fmt.Errorf("RPKI server isn't configured.")
+	}
+	var rfList []bgp.RouteFamily
+	switch family {
+	case bgp.RF_IPv4_UC:
+		rfList = []bgp.RouteFamily{bgp.RF_IPv4_UC}
+	case bgp.RF_IPv6_UC:
+		rfList = []bgp.RouteFamily{bgp.RF_IPv6_UC}
+	default:
+		rfList = []bgp.RouteFamily{bgp.RF_IPv4_UC, bgp.RF_IPv6_UC}
+	}
+	l := make([]*table.ROA, 0)
+	for _, rf := range rfList {
+		if tree, ok := c.Roas[rf]; ok {
+			tree.Walk(func(s string, v interface{}) bool {
+				b, _ := v.(*RoaBucket)
+				var roaList roas
+				for _, r := range b.entries {
+					roaList = append(roaList, r)
+				}
+				sort.Sort(roaList)
+				for _, roa := range roaList {
+					l = append(l, roa)
+				}
+				return false
+			})
+		}
+	}
+	return l, nil
+}
+
+func ValidatePath(ownAs uint32, tree *radix.Tree, cidr string, asPath *bgp.PathAttributeAsPath) *table.Validation {
+	var as uint32
+
+	validation := &table.Validation{
+		Status:          config.RPKI_VALIDATION_RESULT_TYPE_NOT_FOUND,
+		Reason:          table.RPKI_VALIDATION_REASON_TYPE_NONE,
+		Matched:         make([]*table.ROA, 0),
+		UnmatchedLength: make([]*table.ROA, 0),
+		UnmatchedAs:     make([]*table.ROA, 0),
+	}
+
+	if asPath == nil || len(asPath.Value) == 0 {
+		as = ownAs
+	} else {
+		param := asPath.Value[len(asPath.Value)-1]
+		switch param.GetType() {
+		case bgp.BGP_ASPATH_ATTR_TYPE_SEQ:
+			asList := param.GetAS()
+			if len(asList) == 0 {
+				as = ownAs
+			} else {
+				as = asList[len(asList)-1]
+			}
+		case bgp.BGP_ASPATH_ATTR_TYPE_CONFED_SET, bgp.BGP_ASPATH_ATTR_TYPE_CONFED_SEQ:
+			as = ownAs
+		default:
+			return validation
+		}
+	}
+	_, n, _ := net.ParseCIDR(cidr)
+	ones, _ := n.Mask.Size()
+	prefixLen := uint8(ones)
+	key := table.IpToRadixkey(n.IP, prefixLen)
+	_, b, _ := tree.LongestPrefix(key)
+	if b == nil {
+		return validation
+	}
+
+	var bucket *RoaBucket
+	fn := radix.WalkFn(func(k string, v interface{}) bool {
+		bucket, _ = v.(*RoaBucket)
+		for _, r := range bucket.entries {
+			if prefixLen <= r.MaxLen {
+				if r.AS != 0 && r.AS == as {
+					validation.Matched = append(validation.Matched, r)
+				} else {
+					validation.UnmatchedAs = append(validation.UnmatchedAs, r)
+				}
+			} else {
+				validation.UnmatchedLength = append(validation.UnmatchedLength, r)
+			}
+		}
+		return false
+	})
+	tree.WalkPath(key, fn)
+
+	if len(validation.Matched) != 0 {
+		validation.Status = config.RPKI_VALIDATION_RESULT_TYPE_VALID
+		validation.Reason = table.RPKI_VALIDATION_REASON_TYPE_NONE
+	} else if len(validation.UnmatchedAs) != 0 {
+		validation.Status = config.RPKI_VALIDATION_RESULT_TYPE_INVALID
+		validation.Reason = table.RPKI_VALIDATION_REASON_TYPE_AS
+	} else if len(validation.UnmatchedLength) != 0 {
+		validation.Status = config.RPKI_VALIDATION_RESULT_TYPE_INVALID
+		validation.Reason = table.RPKI_VALIDATION_REASON_TYPE_LENGTH
+	} else {
+		validation.Status = config.RPKI_VALIDATION_RESULT_TYPE_NOT_FOUND
+		validation.Reason = table.RPKI_VALIDATION_REASON_TYPE_NONE
+	}
+
+	return validation
+}
+
+func (c *roaManager) validate(path *table.Path) *table.Validation {
+	if len(c.clientMap) == 0 || path.IsWithdraw || path.IsEOR() {
+		// RPKI isn't enabled or invalid path
+		return nil
+	}
+	if tree, ok := c.Roas[path.GetRouteFamily()]; ok {
+		return ValidatePath(c.AS, tree, path.GetNlri().String(), path.GetAsPath())
+	}
+	return nil
+}
+
+type roaClient struct {
+	host         string
+	conn         *net.TCPConn
+	state        config.RpkiServerState
+	eventCh      chan *ROAEvent
+	sessionID    uint16
+	oldSessionID uint16
+	serialNumber uint32
+	timer        *time.Timer
+	lifetime     int64
+	endOfData    bool
+	pendingROAs  []*table.ROA
+	cancelfnc    context.CancelFunc
+	ctx          context.Context
+}
+
+func NewRoaClient(address, port string, ch chan *ROAEvent, lifetime int64) *roaClient {
+	ctx, cancel := context.WithCancel(context.Background())
+	c := &roaClient{
+		host:        net.JoinHostPort(address, port),
+		eventCh:     ch,
+		lifetime:    lifetime,
+		pendingROAs: make([]*table.ROA, 0),
+		ctx:         ctx,
+		cancelfnc:   cancel,
+	}
+	go c.tryConnect()
+	return c
+}
+
+func (c *roaClient) enable(serial uint32) error {
+	if c.conn != nil {
+		r := rtr.NewRTRSerialQuery(c.sessionID, serial)
+		data, _ := r.Serialize()
+		_, err := c.conn.Write(data)
+		if err != nil {
+			return err
+		}
+		c.state.RpkiMessages.RpkiSent.SerialQuery++
+	}
+	return nil
+}
+
+func (c *roaClient) softReset() error {
+	if c.conn != nil {
+		r := rtr.NewRTRResetQuery()
+		data, _ := r.Serialize()
+		_, err := c.conn.Write(data)
+		if err != nil {
+			return err
+		}
+		c.state.RpkiMessages.RpkiSent.ResetQuery++
+		c.endOfData = false
+		c.pendingROAs = make([]*table.ROA, 0)
+	}
+	return nil
+}
+
+func (c *roaClient) reset() {
+	if c.conn != nil {
+		c.conn.Close()
+	}
+}
+
+func (c *roaClient) stop() {
+	c.cancelfnc()
+	c.reset()
+}
+
+func (c *roaClient) tryConnect() {
+	for {
+		select {
+		case <-c.ctx.Done():
+			return
+		default:
+		}
+		if conn, err := net.Dial("tcp", c.host); err != nil {
+			// better to use context with timeout
+			time.Sleep(CONNECT_RETRY_INTERVAL * time.Second)
+		} else {
+			c.eventCh <- &ROAEvent{
+				EventType: CONNECTED,
+				Src:       c.host,
+				conn:      conn.(*net.TCPConn),
+			}
+			return
+		}
+	}
+}
+
+func (c *roaClient) established() (err error) {
+	defer func() {
+		c.conn.Close()
+		c.eventCh <- &ROAEvent{
+			EventType: DISCONNECTED,
+			Src:       c.host,
+		}
+	}()
+
+	if err := c.softReset(); err != nil {
+		return err
+	}
+
+	for {
+		header := make([]byte, rtr.RTR_MIN_LEN)
+		if _, err = io.ReadFull(c.conn, header); err != nil {
+			return err
+		}
+		totalLen := binary.BigEndian.Uint32(header[4:8])
+		if totalLen < rtr.RTR_MIN_LEN {
+			return fmt.Errorf("too short header length %v", totalLen)
+		}
+
+		body := make([]byte, totalLen-rtr.RTR_MIN_LEN)
+		if _, err = io.ReadFull(c.conn, body); err != nil {
+			return
+		}
+
+		c.eventCh <- &ROAEvent{
+			EventType: RTR,
+			Src:       c.host,
+			Data:      append(header, body...),
+		}
+	}
+}

vendor/github.com/osrg/gobgp/pkg/server/sockopt.go

@@ -0,0 +1,90 @@
+// Copyright (C) 2016 Nippon Telegraph and Telephone Corporation.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+// +build !linux,!openbsd
+
+package server
+
+import (
+	"fmt"
+	"net"
+
+	log "github.com/sirupsen/logrus"
+)
+
+func SetTcpMD5SigSockopt(l *net.TCPListener, address string, key string) error {
+	return setTcpMD5SigSockopt(l, address, key)
+}
+
+func SetListenTcpTTLSockopt(l *net.TCPListener, ttl int) error {
+	return setListenTcpTTLSockopt(l, ttl)
+}
+
+func SetTcpTTLSockopt(conn *net.TCPConn, ttl int) error {
+	return setTcpTTLSockopt(conn, ttl)
+}
+
+func SetTcpMinTTLSockopt(conn *net.TCPConn, ttl int) error {
+	return setTcpMinTTLSockopt(conn, ttl)
+}
+
+type TCPDialer struct {
+	net.Dialer
+
+	// MD5 authentication password.
+	AuthPassword string
+
+	// The TTL value to set outgoing connection.
+	Ttl uint8
+
+	// The minimum TTL value for incoming packets.
+	TtlMin uint8
+}
+
+func (d *TCPDialer) DialTCP(addr string, port int) (*net.TCPConn, error) {
+	if d.AuthPassword != "" {
+		log.WithFields(log.Fields{
+			"Topic": "Peer",
+			"Key":   addr,
+		}).Warn("setting md5 for active connection is not supported")
+	}
+	if d.Ttl != 0 {
+		log.WithFields(log.Fields{
+			"Topic": "Peer",
+			"Key":   addr,
+		}).Warn("setting ttl for active connection is not supported")
+	}
+	if d.TtlMin != 0 {
+		log.WithFields(log.Fields{
+			"Topic": "Peer",
+			"Key":   addr,
+		}).Warn("setting min ttl for active connection is not supported")
+	}
+
+	raddr, err := net.ResolveTCPAddr("tcp", net.JoinHostPort(addr, fmt.Sprintf("%d", port)))
+	if err != nil {
+		return nil, fmt.Errorf("invalid remote address: %s", err)
+	}
+	laddr, err := net.ResolveTCPAddr("tcp", d.LocalAddr.String())
+	if err != nil {
+		return nil, fmt.Errorf("invalid local address: %s", err)
+	}
+
+	dialer := net.Dialer{LocalAddr: laddr, Timeout: d.Timeout}
+	conn, err := dialer.Dial("tcp", raddr.String())
+	if err != nil {
+		return nil, err
+	}
+	return conn.(*net.TCPConn), nil
+}

vendor/github.com/osrg/gobgp/pkg/server/sockopt_bsd.go

@@ -0,0 +1,69 @@
+// Copyright (C) 2016 Nippon Telegraph and Telephone Corporation.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+// +build dragonfly freebsd netbsd
+
+package server
+
+import (
+	"net"
+	"syscall"
+)
+
+const (
+	TCP_MD5SIG       = 0x10 // TCP MD5 Signature (RFC2385)
+	IPV6_MINHOPCOUNT = 73   // Generalized TTL Security Mechanism (RFC5082)
+)
+
+func setTcpMD5SigSockopt(l *net.TCPListener, address string, key string) error {
+	sc, err := l.SyscallConn()
+	if err != nil {
+		return err
+	}
+	// always enable and assumes that the configuration is done by setkey()
+	return setsockOptInt(sc, syscall.IPPROTO_TCP, TCP_MD5SIG, 1)
+}
+
+func setListenTcpTTLSockopt(l *net.TCPListener, ttl int) error {
+	family := extractFamilyFromTCPListener(l)
+	sc, err := l.SyscallConn()
+	if err != nil {
+		return err
+	}
+	return setsockoptIpTtl(sc, family, ttl)
+}
+
+func setTcpTTLSockopt(conn *net.TCPConn, ttl int) error {
+	family := extractFamilyFromTCPConn(conn)
+	sc, err := conn.SyscallConn()
+	if err != nil {
+		return err
+	}
+	return setsockoptIpTtl(sc, family, ttl)
+}
+
+func setTcpMinTTLSockopt(conn *net.TCPConn, ttl int) error {
+	family := extractFamilyFromTCPConn(conn)
+	sc, err := conn.SyscallConn()
+	if err != nil {
+		return err
+	}
+	level := syscall.IPPROTO_IP
+	name := syscall.IP_MINTTL
+	if family == syscall.AF_INET6 {
+		level = syscall.IPPROTO_IPV6
+		name = IPV6_MINHOPCOUNT
+	}
+	return setsockOptInt(sc, level, name, ttl)
+}

vendor/github.com/osrg/gobgp/pkg/server/sockopt_darwin.go

@@ -0,0 +1,53 @@
+// Copyright (C) 2016-2017 Nippon Telegraph and Telephone Corporation.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+// +build darwin
+
+package server
+
+import (
+	"fmt"
+	"net"
+	"strings"
+	"syscall"
+)
+
+func setTcpMD5SigSockopt(l *net.TCPListener, address string, key string) error {
+	return fmt.Errorf("setting md5 is not supported")
+}
+
+func setListenTcpTTLSockopt(l *net.TCPListener, ttl int) error {
+	family := extractFamilyFromTCPListener(l)
+	sc, err := l.SyscallConn()
+	if err != nil {
+		return err
+	}
+	return setsockoptIpTtl(sc, family, ttl)
+}
+
+func setTcpTTLSockopt(conn *net.TCPConn, ttl int) error {
+	family := syscall.AF_INET
+	if strings.Contains(conn.RemoteAddr().String(), "[") {
+		family = syscall.AF_INET6
+	}
+	sc, err := conn.SyscallConn()
+	if err != nil {
+		return err
+	}
+	return setsockoptIpTtl(sc, family, ttl)
+}
+
+func setTcpMinTTLSockopt(conn *net.TCPConn, ttl int) error {
+	return fmt.Errorf("setting min ttl is not supported")
+}

vendor/github.com/osrg/gobgp/pkg/server/sockopt_linux.go

@@ -0,0 +1,289 @@
+// Copyright (C) 2016 Nippon Telegraph and Telephone Corporation.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+// +build linux
+
+package server
+
+import (
+	"fmt"
+	"net"
+	"os"
+	"syscall"
+	"unsafe"
+)
+
+const (
+	TCP_MD5SIG       = 14 // TCP MD5 Signature (RFC2385)
+	IPV6_MINHOPCOUNT = 73 // Generalized TTL Security Mechanism (RFC5082)
+)
+
+type tcpmd5sig struct {
+	ss_family uint16
+	ss        [126]byte
+	// padding the struct
+	_      uint16
+	keylen uint16
+	// padding the struct
+	_   uint32
+	key [80]byte
+}
+
+func buildTcpMD5Sig(address string, key string) (tcpmd5sig, error) {
+	t := tcpmd5sig{}
+	addr := net.ParseIP(address)
+	if addr.To4() != nil {
+		t.ss_family = syscall.AF_INET
+		copy(t.ss[2:], addr.To4())
+	} else {
+		t.ss_family = syscall.AF_INET6
+		copy(t.ss[6:], addr.To16())
+	}
+
+	t.keylen = uint16(len(key))
+	copy(t.key[0:], []byte(key))
+
+	return t, nil
+}
+
+func SetTcpMD5SigSockopt(l *net.TCPListener, address string, key string) error {
+	t, err := buildTcpMD5Sig(address, key)
+	if err != nil {
+		return err
+	}
+	b := *(*[unsafe.Sizeof(t)]byte)(unsafe.Pointer(&t))
+
+	sc, err := l.SyscallConn()
+	if err != nil {
+		return err
+	}
+	return setsockOptString(sc, syscall.IPPROTO_TCP, TCP_MD5SIG, string(b[:]))
+}
+
+func SetListenTcpTTLSockopt(l *net.TCPListener, ttl int) error {
+	family := extractFamilyFromTCPListener(l)
+	sc, err := l.SyscallConn()
+	if err != nil {
+		return err
+	}
+	return setsockoptIpTtl(sc, family, ttl)
+}
+
+func SetTcpTTLSockopt(conn *net.TCPConn, ttl int) error {
+	family := extractFamilyFromTCPConn(conn)
+	sc, err := conn.SyscallConn()
+	if err != nil {
+		return err
+	}
+	return setsockoptIpTtl(sc, family, ttl)
+}
+
+func SetTcpMinTTLSockopt(conn *net.TCPConn, ttl int) error {
+	family := extractFamilyFromTCPConn(conn)
+	sc, err := conn.SyscallConn()
+	if err != nil {
+		return err
+	}
+	level := syscall.IPPROTO_IP
+	name := syscall.IP_MINTTL
+	if family == syscall.AF_INET6 {
+		level = syscall.IPPROTO_IPV6
+		name = IPV6_MINHOPCOUNT
+	}
+	return setsockOptInt(sc, level, name, ttl)
+}
+
+func setsockoptTcpMD5Sig(fd int, address string, key string) error {
+	t, err := buildTcpMD5Sig(address, key)
+	if err != nil {
+		return err
+	}
+	b := *(*[unsafe.Sizeof(t)]byte)(unsafe.Pointer(&t))
+	return os.NewSyscallError("setsockopt", syscall.SetsockoptString(fd, syscall.IPPROTO_TCP, TCP_MD5SIG, string(b[:])))
+}
+
+func setsockoptIpTtl2(fd int, family int, value int) error {
+	level := syscall.IPPROTO_IP
+	name := syscall.IP_TTL
+	if family == syscall.AF_INET6 {
+		level = syscall.IPPROTO_IPV6
+		name = syscall.IPV6_UNICAST_HOPS
+	}
+	return os.NewSyscallError("setsockopt", syscall.SetsockoptInt(fd, level, name, value))
+}
+
+func setsockoptIpMinTtl(fd int, family int, value int) error {
+	level := syscall.IPPROTO_IP
+	name := syscall.IP_MINTTL
+	if family == syscall.AF_INET6 {
+		level = syscall.IPPROTO_IPV6
+		name = IPV6_MINHOPCOUNT
+	}
+	return os.NewSyscallError("setsockopt", syscall.SetsockoptInt(fd, level, name, value))
+}
+
+type TCPDialer struct {
+	net.Dialer
+
+	// MD5 authentication password.
+	AuthPassword string
+
+	// The TTL value to set outgoing connection.
+	Ttl uint8
+
+	// The minimum TTL value for incoming packets.
+	TtlMin uint8
+}
+
+func (d *TCPDialer) DialTCP(addr string, port int) (*net.TCPConn, error) {
+	var family int
+	var ra, la syscall.Sockaddr
+
+	raddr, err := net.ResolveTCPAddr("tcp", net.JoinHostPort(addr, fmt.Sprintf("%d", port)))
+	if err != nil {
+		return nil, fmt.Errorf("invalid remote address: %s", err)
+	}
+	laddr, err := net.ResolveTCPAddr("tcp", d.LocalAddr.String())
+	if err != nil {
+		return nil, fmt.Errorf("invalid local address: %s", err)
+	}
+	if raddr.IP.To4() != nil {
+		family = syscall.AF_INET
+		rsockaddr := &syscall.SockaddrInet4{Port: port}
+		copy(rsockaddr.Addr[:], raddr.IP.To4())
+		ra = rsockaddr
+		lsockaddr := &syscall.SockaddrInet4{}
+		copy(lsockaddr.Addr[:], laddr.IP.To4())
+		la = lsockaddr
+	} else {
+		family = syscall.AF_INET6
+		rsockaddr := &syscall.SockaddrInet6{Port: port}
+		copy(rsockaddr.Addr[:], raddr.IP.To16())
+		ra = rsockaddr
+		var zone uint32
+		if laddr.Zone != "" {
+			if intf, err := net.InterfaceByName(laddr.Zone); err != nil {
+				return nil, err
+			} else {
+				zone = uint32(intf.Index)
+			}
+		}
+		lsockaddr := &syscall.SockaddrInet6{ZoneId: zone}
+		copy(lsockaddr.Addr[:], laddr.IP.To16())
+		la = lsockaddr
+	}
+
+	sockType := syscall.SOCK_STREAM | syscall.SOCK_CLOEXEC | syscall.SOCK_NONBLOCK
+	proto := 0
+	fd, err := syscall.Socket(family, sockType, proto)
+	if err != nil {
+		return nil, err
+	}
+	fi := os.NewFile(uintptr(fd), "")
+	defer fi.Close()
+	// A new socket was created so we must close it before this
+	// function returns either on failure or success. On success,
+	// net.FileConn() in newTCPConn() increases the refcount of
+	// the socket so this fi.Close() doesn't destroy the socket.
+	// The caller must call Close() with the file later.
+	// Note that the above os.NewFile() doesn't play with the
+	// refcount.
+
+	if err = syscall.SetsockoptInt(fd, syscall.SOL_SOCKET, syscall.SO_BROADCAST, 1); err != nil {
+		return nil, os.NewSyscallError("setsockopt", err)
+	}
+
+	if err = syscall.SetsockoptInt(fd, syscall.IPPROTO_TCP, syscall.TCP_NODELAY, 1); err != nil {
+		return nil, os.NewSyscallError("setsockopt", err)
+	}
+
+	if d.AuthPassword != "" {
+		if err = setsockoptTcpMD5Sig(fd, addr, d.AuthPassword); err != nil {
+			return nil, err
+		}
+	}
+
+	if d.Ttl != 0 {
+		if err = setsockoptIpTtl2(fd, family, int(d.Ttl)); err != nil {
+			return nil, err
+		}
+	}
+
+	if d.TtlMin != 0 {
+		if err = setsockoptIpMinTtl(fd, family, int(d.Ttl)); err != nil {
+			return nil, err
+		}
+	}
+
+	if err = syscall.Bind(fd, la); err != nil {
+		return nil, os.NewSyscallError("bind", err)
+	}
+
+	newTCPConn := func(fi *os.File) (*net.TCPConn, error) {
+		if conn, err := net.FileConn(fi); err != nil {
+			return nil, err
+		} else {
+			return conn.(*net.TCPConn), err
+		}
+	}
+
+	err = syscall.Connect(fd, ra)
+	switch err {
+	case syscall.EINPROGRESS, syscall.EALREADY, syscall.EINTR:
+		// do timeout handling
+	case nil, syscall.EISCONN:
+		return newTCPConn(fi)
+	default:
+		return nil, os.NewSyscallError("connect", err)
+	}
+
+	epfd, e := syscall.EpollCreate1(syscall.EPOLL_CLOEXEC)
+	if e != nil {
+		return nil, e
+	}
+	defer syscall.Close(epfd)
+
+	var event syscall.EpollEvent
+	events := make([]syscall.EpollEvent, 1)
+
+	event.Events = syscall.EPOLLIN | syscall.EPOLLOUT | syscall.EPOLLPRI
+	event.Fd = int32(fd)
+	if e = syscall.EpollCtl(epfd, syscall.EPOLL_CTL_ADD, fd, &event); e != nil {
+		return nil, e
+	}
+
+	for {
+		nevents, e := syscall.EpollWait(epfd, events, int(d.Timeout/1000000) /*msec*/)
+		if e != nil {
+			return nil, e
+		}
+		if nevents == 0 {
+			return nil, fmt.Errorf("timeout")
+		} else if nevents == 1 && events[0].Fd == int32(fd) {
+			nerr, err := syscall.GetsockoptInt(fd, syscall.SOL_SOCKET, syscall.SO_ERROR)
+			if err != nil {
+				return nil, os.NewSyscallError("getsockopt", err)
+			}
+			switch err := syscall.Errno(nerr); err {
+			case syscall.EINPROGRESS, syscall.EALREADY, syscall.EINTR:
+			case syscall.Errno(0), syscall.EISCONN:
+				return newTCPConn(fi)
+			default:
+				return nil, os.NewSyscallError("getsockopt", err)
+			}
+		} else {
+			return nil, fmt.Errorf("unexpected epoll behavior")
+		}
+	}
+}

vendor/github.com/osrg/gobgp/pkg/server/sockopt_openbsd.go

@@ -0,0 +1,454 @@
+// Copyright (C) 2016 Nippon Telegraph and Telephone Corporation.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+// +build openbsd
+
+package server
+
+import (
+	"encoding/binary"
+	"fmt"
+	"net"
+	"os"
+	"syscall"
+	"unsafe"
+
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	PF_KEY_V2 = 2
+
+	SADB_X_SATYPE_TCPSIGNATURE = 8
+
+	SADB_EXT_SA          = 1
+	SADB_EXT_ADDRESS_SRC = 5
+	SADB_EXT_ADDRESS_DST = 6
+	SADB_EXT_KEY_AUTH    = 8
+	SADB_EXT_SPIRANGE    = 16
+
+	SADB_GETSPI = 1
+	SADB_UPDATE = 2
+	SADB_DELETE = 4
+
+	SADB_X_EALG_AES = 12
+
+	SADB_SASTATE_MATURE = 1
+)
+
+type sadbMsg struct {
+	sadbMsgVersion  uint8
+	sadbMsgType     uint8
+	sadbMsgErrno    uint8
+	sadbMsgSatype   uint8
+	sadbMsgLen      uint16
+	sadbMsgReserved uint16
+	sadbMsgSeq      uint32
+	sadbMsgPid      uint32
+}
+
+func (s *sadbMsg) DecodeFromBytes(data []byte) error {
+	if len(data) < SADB_MSG_SIZE {
+		fmt.Errorf("too short for sadbMsg %d", len(data))
+	}
+	s.sadbMsgVersion = data[0]
+	s.sadbMsgType = data[1]
+	s.sadbMsgErrno = data[2]
+	s.sadbMsgSatype = data[3]
+	s.sadbMsgLen = binary.LittleEndian.Uint16(data[4:6])
+	s.sadbMsgSeq = binary.LittleEndian.Uint32(data[8:12])
+	s.sadbMsgPid = binary.LittleEndian.Uint32(data[12:16])
+	return nil
+}
+
+type sadbSpirange struct {
+	sadbSpirangeLen      uint16
+	sadbSpirangeExttype  uint16
+	sadbSpirangeMin      uint32
+	sadbSpirangeMax      uint32
+	sadbSpirangeReserved uint32
+}
+
+type sadbAddress struct {
+	sadbAddressLen      uint16
+	sadbAddressExttype  uint16
+	sadbAddressReserved uint32
+}
+
+type sadbExt struct {
+	sadbExtLen  uint16
+	sadbExtType uint16
+}
+
+type sadbSa struct {
+	sadbSaLen     uint16
+	sadbSaExttype uint16
+	sadbSaSpi     uint32
+	sadbSaReplay  uint8
+	sadbSaState   uint8
+	sadbSaAuth    uint8
+	sadbSaEncrypt uint8
+	sadbSaFlags   uint32
+}
+
+type sadbKey struct {
+	sadbKeyLen      uint16
+	sadbKeyExttype  uint16
+	sadbKeyBits     uint16
+	sadbKeyReserved uint16
+}
+
+const (
+	SADB_MSG_SIZE      = int(unsafe.Sizeof(sadbMsg{}))
+	SADB_SPIRANGE_SIZE = int(unsafe.Sizeof(sadbSpirange{}))
+	SADB_ADDRESS_SIZE  = int(unsafe.Sizeof(sadbAddress{}))
+	SADB_SA_SIZE       = int(unsafe.Sizeof(sadbSa{}))
+	SADB_KEY_SIZE      = int(unsafe.Sizeof(sadbKey{}))
+)
+
+type sockaddrIn struct {
+	ssLen    uint8
+	ssFamily uint8
+	ssPort   uint16
+	ssAddr   uint32
+	pad      [8]byte
+}
+
+func newSockaddrIn(addr string) sockaddrIn {
+	if len(addr) == 0 {
+		return sockaddrIn{
+			ssLen: 16,
+		}
+	}
+	v := net.ParseIP(addr).To4()
+	return sockaddrIn{
+		ssAddr:   uint32(v[3])<<24 | uint32(v[2])<<16 | uint32(v[1])<<8 | uint32(v[0]),
+		ssLen:    16,
+		ssFamily: syscall.AF_INET,
+	}
+}
+
+func roundUp(v int) int {
+	if v%8 != 0 {
+		v += 8 - v%8
+	}
+	return v
+}
+
+func b(p unsafe.Pointer, length int) []byte {
+	buf := make([]byte, length)
+	for i := 0; i < length; i++ {
+		buf[i] = *(*byte)(p)
+		p = unsafe.Pointer(uintptr(p) + 1)
+	}
+	return buf
+}
+
+var seq uint32
+var fd int
+
+var spiInMap map[string]uint32 = map[string]uint32{}
+var spiOutMap map[string]uint32 = map[string]uint32{}
+
+func pfkeyReply() (spi uint32, err error) {
+	buf := make([]byte, SADB_MSG_SIZE)
+	if count, _, _, _, _ := syscall.Recvmsg(fd, buf, nil, syscall.MSG_PEEK); count != len(buf) {
+		return spi, fmt.Errorf("incomplete sadb msg %d %d", len(buf), count)
+	}
+	h := sadbMsg{}
+	h.DecodeFromBytes(buf)
+	if h.sadbMsgErrno != 0 {
+		return spi, fmt.Errorf("sadb msg reply error %d", h.sadbMsgErrno)
+	}
+
+	if h.sadbMsgSeq != seq {
+		return spi, fmt.Errorf("sadb msg sequence doesn't match %d %d", h.sadbMsgSeq, seq)
+	}
+
+	if h.sadbMsgPid != uint32(os.Getpid()) {
+		return spi, fmt.Errorf("sadb msg pid doesn't match %d %d", h.sadbMsgPid, os.Getpid())
+	}
+
+	buf = make([]byte, int(8*h.sadbMsgLen))
+	if count, _, _, _, _ := syscall.Recvmsg(fd, buf, nil, 0); count != len(buf) {
+		return spi, fmt.Errorf("incomplete sadb msg body %d %d", len(buf), count)
+	}
+
+	buf = buf[SADB_MSG_SIZE:]
+
+	for len(buf) >= 4 {
+		l := binary.LittleEndian.Uint16(buf[0:2]) * 8
+		t := binary.LittleEndian.Uint16(buf[2:4])
+		if t == SADB_EXT_SA {
+			return binary.LittleEndian.Uint32(buf[4:8]), nil
+		}
+
+		if len(buf) <= int(l) {
+			break
+		}
+		buf = buf[l:]
+	}
+	return spi, err
+}
+
+func sendSadbMsg(msg *sadbMsg, body []byte) (err error) {
+	if fd == 0 {
+		fd, err = syscall.Socket(syscall.AF_KEY, syscall.SOCK_RAW, PF_KEY_V2)
+		if err != nil {
+			return err
+		}
+	}
+
+	seq++
+	msg.sadbMsgSeq = seq
+	msg.sadbMsgLen = uint16((len(body) + SADB_MSG_SIZE) / 8)
+
+	buf := append(b(unsafe.Pointer(msg), SADB_MSG_SIZE), body...)
+
+	r, err := syscall.Write(fd, buf)
+	if r != len(buf) {
+		return fmt.Errorf("short write %d %d", r, len(buf))
+	}
+	return err
+}
+
+func rfkeyRequest(msgType uint8, src, dst string, spi uint32, key string) error {
+	h := sadbMsg{
+		sadbMsgVersion: PF_KEY_V2,
+		sadbMsgType:    msgType,
+		sadbMsgSatype:  SADB_X_SATYPE_TCPSIGNATURE,
+		sadbMsgPid:     uint32(os.Getpid()),
+	}
+
+	ssrc := newSockaddrIn(src)
+	sa_src := sadbAddress{
+		sadbAddressExttype: SADB_EXT_ADDRESS_SRC,
+		sadbAddressLen:     uint16(SADB_ADDRESS_SIZE+roundUp(int(ssrc.ssLen))) / 8,
+	}
+
+	sdst := newSockaddrIn(dst)
+	sa_dst := sadbAddress{
+		sadbAddressExttype: SADB_EXT_ADDRESS_DST,
+		sadbAddressLen:     uint16(SADB_ADDRESS_SIZE+roundUp(int(sdst.ssLen))) / 8,
+	}
+
+	buf := make([]byte, 0)
+	switch msgType {
+	case SADB_UPDATE, SADB_DELETE:
+		sa := sadbSa{
+			sadbSaLen:     uint16(SADB_SA_SIZE / 8),
+			sadbSaExttype: SADB_EXT_SA,
+			sadbSaSpi:     spi,
+			sadbSaState:   SADB_SASTATE_MATURE,
+			sadbSaEncrypt: SADB_X_EALG_AES,
+		}
+		buf = append(buf, b(unsafe.Pointer(&sa), SADB_SA_SIZE)...)
+	case SADB_GETSPI:
+		spirange := sadbSpirange{
+			sadbSpirangeLen:     uint16(SADB_SPIRANGE_SIZE) / 8,
+			sadbSpirangeExttype: SADB_EXT_SPIRANGE,
+			sadbSpirangeMin:     0x100,
+			sadbSpirangeMax:     0xffffffff,
+		}
+		buf = append(buf, b(unsafe.Pointer(&spirange), SADB_SPIRANGE_SIZE)...)
+	}
+
+	buf = append(buf, b(unsafe.Pointer(&sa_dst), SADB_ADDRESS_SIZE)...)
+	buf = append(buf, b(unsafe.Pointer(&sdst), roundUp(int(sdst.ssLen)))...)
+	buf = append(buf, b(unsafe.Pointer(&sa_src), SADB_ADDRESS_SIZE)...)
+	buf = append(buf, b(unsafe.Pointer(&ssrc), roundUp(int(ssrc.ssLen)))...)
+
+	switch msgType {
+	case SADB_UPDATE:
+		keylen := roundUp(len(key))
+		sa_akey := sadbKey{
+			sadbKeyLen:     uint16((SADB_KEY_SIZE + keylen) / 8),
+			sadbKeyExttype: SADB_EXT_KEY_AUTH,
+			sadbKeyBits:    uint16(len(key) * 8),
+		}
+		k := []byte(key)
+		if pad := keylen - len(k); pad != 0 {
+			k = append(k, make([]byte, pad)...)
+		}
+		buf = append(buf, b(unsafe.Pointer(&sa_akey), SADB_KEY_SIZE)...)
+		buf = append(buf, k...)
+	}
+
+	return sendSadbMsg(&h, buf)
+}
+
+func saAdd(address, key string) error {
+	f := func(src, dst string) error {
+		if err := rfkeyRequest(SADB_GETSPI, src, dst, 0, ""); err != nil {
+			return err
+		}
+		spi, err := pfkeyReply()
+		if err != nil {
+			return err
+		}
+		if src == "" {
+			spiOutMap[address] = spi
+		} else {
+			spiInMap[address] = spi
+		}
+
+		if err := rfkeyRequest(SADB_UPDATE, src, dst, spi, key); err != nil {
+			return err
+		}
+		_, err = pfkeyReply()
+		return err
+	}
+
+	if err := f(address, ""); err != nil {
+		return err
+	}
+
+	return f("", address)
+}
+
+func saDelete(address string) error {
+	if spi, y := spiInMap[address]; y {
+		if err := rfkeyRequest(SADB_DELETE, address, "", spi, ""); err != nil {
+			log.WithFields(log.Fields{
+				"Topic": "Peer",
+				"Key":   address,
+			}).Info("failed to delete md5 for incoming")
+		} else {
+			log.WithFields(log.Fields{
+				"Topic": "Peer",
+				"Key":   address,
+			}).Info("can't find spi for md5 for incoming")
+		}
+	}
+	if spi, y := spiOutMap[address]; y {
+		if err := rfkeyRequest(SADB_DELETE, "", address, spi, ""); err != nil {
+			log.WithFields(log.Fields{
+				"Topic": "Peer",
+				"Key":   address,
+			}).Info("failed to delete md5 for outgoing")
+		} else {
+			log.WithFields(log.Fields{
+				"Topic": "Peer",
+				"Key":   address,
+			}).Info("can't find spi for md5 for outgoing")
+		}
+	}
+	return nil
+}
+
+const (
+	TCP_MD5SIG       = 0x4 // TCP MD5 Signature (RFC2385)
+	IPV6_MINHOPCOUNT = 73  // Generalized TTL Security Mechanism (RFC5082)
+)
+
+func setsockoptTcpMD5Sig(sc syscall.RawConn, address string, key string) error {
+	if err := setsockOptInt(sc, syscall.IPPROTO_TCP, TCP_MD5SIG, 1); err != nil {
+		return err
+	}
+	if len(key) > 0 {
+		return saAdd(address, key)
+	}
+	return saDelete(address)
+}
+
+func SetTcpMD5SigSockopt(l *net.TCPListener, address string, key string) error {
+	sc, err := l.SyscallConn()
+	if err != nil {
+		return err
+	}
+	return setsockoptTcpMD5Sig(sc, address, key)
+}
+
+func SetListenTcpTTLSockopt(l *net.TCPListener, ttl int) error {
+	family := extractFamilyFromTCPListener(l)
+	sc, err := l.SyscallConn()
+	if err != nil {
+		return err
+	}
+	return setsockoptIpTtl(sc, family, ttl)
+}
+
+func SetTcpTTLSockopt(conn *net.TCPConn, ttl int) error {
+	family := extractFamilyFromTCPConn(conn)
+	sc, err := conn.SyscallConn()
+	if err != nil {
+		return err
+	}
+	return setsockoptIpTtl(sc, family, ttl)
+}
+
+func SetTcpMinTTLSockopt(conn *net.TCPConn, ttl int) error {
+	family := extractFamilyFromTCPConn(conn)
+	sc, err := conn.SyscallConn()
+	if err != nil {
+		return err
+	}
+	level := syscall.IPPROTO_IP
+	name := syscall.IP_MINTTL
+	if family == syscall.AF_INET6 {
+		level = syscall.IPPROTO_IPV6
+		name = IPV6_MINHOPCOUNT
+	}
+	return setsockOptInt(sc, level, name, ttl)
+}
+
+type TCPDialer struct {
+	net.Dialer
+
+	// MD5 authentication password.
+	AuthPassword string
+
+	// The TTL value to set outgoing connection.
+	Ttl uint8
+
+	// The minimum TTL value for incoming packets.
+	TtlMin uint8
+}
+
+func (d *TCPDialer) DialTCP(addr string, port int) (*net.TCPConn, error) {
+	if d.AuthPassword != "" {
+		log.WithFields(log.Fields{
+			"Topic": "Peer",
+			"Key":   addr,
+		}).Warn("setting md5 for active connection is not supported")
+	}
+	if d.Ttl != 0 {
+		log.WithFields(log.Fields{
+			"Topic": "Peer",
+			"Key":   addr,
+		}).Warn("setting ttl for active connection is not supported")
+	}
+	if d.TtlMin != 0 {
+		log.WithFields(log.Fields{
+			"Topic": "Peer",
+			"Key":   addr,
+		}).Warn("setting min ttl for active connection is not supported")
+	}
+
+	raddr, err := net.ResolveTCPAddr("tcp", net.JoinHostPort(addr, fmt.Sprintf("%d", port)))
+	if err != nil {
+		return nil, fmt.Errorf("invalid remote address: %s", err)
+	}
+	laddr, err := net.ResolveTCPAddr("tcp", d.LocalAddr.String())
+	if err != nil {
+		return nil, fmt.Errorf("invalid local address: %s", err)
+	}
+
+	dialer := net.Dialer{LocalAddr: laddr, Timeout: d.Timeout}
+	conn, err := dialer.Dial("tcp", raddr.String())
+	if err != nil {
+		return nil, err
+	}
+	return conn.(*net.TCPConn), nil
+}

vendor/github.com/osrg/gobgp/pkg/server/sockopt_stub.go

@@ -0,0 +1,38 @@
+// Copyright (C) 2016 Nippon Telegraph and Telephone Corporation.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+// +build !linux,!dragonfly,!freebsd,!netbsd,!openbsd,!darwin
+
+package server
+
+import (
+	"fmt"
+	"net"
+)
+
+func setTcpMD5SigSockopt(l *net.TCPListener, address string, key string) error {
+	return fmt.Errorf("setting md5 is not supported")
+}
+
+func setListenTcpTTLSockopt(l *net.TCPListener, ttl int) error {
+	return fmt.Errorf("setting ttl is not supported")
+}
+
+func setTcpTTLSockopt(conn *net.TCPConn, ttl int) error {
+	return fmt.Errorf("setting ttl is not supported")
+}
+
+func setTcpMinTTLSockopt(conn *net.TCPConn, ttl int) error {
+	return fmt.Errorf("setting min ttl is not supported")
+}

vendor/github.com/osrg/gobgp/pkg/server/util.go

@@ -0,0 +1,116 @@
+// Copyright (C) 2016 Nippon Telegraph and Telephone Corporation.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"net"
+	"strings"
+	"syscall"
+
+	"github.com/eapache/channels"
+
+	"github.com/osrg/gobgp/pkg/packet/bgp"
+)
+
+func cleanInfiniteChannel(ch *channels.InfiniteChannel) {
+	ch.Close()
+	// drain all remaining items
+	for range ch.Out() {
+	}
+}
+
+// Returns the binary formatted Administrative Shutdown Communication from the
+// given string value.
+func newAdministrativeCommunication(communication string) (data []byte) {
+	if communication == "" {
+		return nil
+	}
+	com := []byte(communication)
+	if len(com) > bgp.BGP_ERROR_ADMINISTRATIVE_COMMUNICATION_MAX {
+		data = []byte{bgp.BGP_ERROR_ADMINISTRATIVE_COMMUNICATION_MAX}
+		data = append(data, com[:bgp.BGP_ERROR_ADMINISTRATIVE_COMMUNICATION_MAX]...)
+	} else {
+		data = []byte{byte(len(com))}
+		data = append(data, com...)
+	}
+	return data
+}
+
+// Parses the given NOTIFICATION message data as a binary value and returns
+// the Administrative Shutdown Communication in string and the rest binary.
+func decodeAdministrativeCommunication(data []byte) (string, []byte) {
+	if len(data) == 0 {
+		return "", data
+	}
+	communicationLen := int(data[0])
+	if communicationLen > bgp.BGP_ERROR_ADMINISTRATIVE_COMMUNICATION_MAX {
+		communicationLen = bgp.BGP_ERROR_ADMINISTRATIVE_COMMUNICATION_MAX
+	}
+	if communicationLen > len(data)+1 {
+		communicationLen = len(data) + 1
+	}
+	return string(data[1 : communicationLen+1]), data[communicationLen+1:]
+}
+
+func extractFamilyFromTCPListener(l *net.TCPListener) int {
+	family := syscall.AF_INET
+	if strings.Contains(l.Addr().String(), "[") {
+		family = syscall.AF_INET6
+	}
+	return family
+}
+
+func extractFamilyFromTCPConn(conn *net.TCPConn) int {
+	family := syscall.AF_INET
+	if strings.Contains(conn.RemoteAddr().String(), "[") {
+		family = syscall.AF_INET6
+	}
+	return family
+}
+
+func setsockOptString(sc syscall.RawConn, level int, opt int, str string) error {
+	var opterr error
+	fn := func(s uintptr) {
+		opterr = syscall.SetsockoptString(int(s), level, opt, str)
+	}
+	err := sc.Control(fn)
+	if opterr == nil {
+		return err
+	}
+	return opterr
+}
+
+func setsockOptInt(sc syscall.RawConn, level, name, value int) error {
+	var opterr error
+	fn := func(s uintptr) {
+		opterr = syscall.SetsockoptInt(int(s), level, name, value)
+	}
+	err := sc.Control(fn)
+	if opterr == nil {
+		return err
+	}
+	return opterr
+}
+
+func setsockoptIpTtl(sc syscall.RawConn, family int, value int) error {
+	level := syscall.IPPROTO_IP
+	name := syscall.IP_TTL
+	if family == syscall.AF_INET6 {
+		level = syscall.IPPROTO_IPV6
+		name = syscall.IPV6_UNICAST_HOPS
+	}
+	return setsockOptInt(sc, level, name, value)
+}

vendor/github.com/osrg/gobgp/pkg/server/zclient.go

@@ -0,0 +1,456 @@
+// Copyright (C) 2015 Nippon Telegraph and Telephone Corporation.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"fmt"
+	"math"
+	"net"
+	"strconv"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/osrg/gobgp/internal/pkg/table"
+	"github.com/osrg/gobgp/internal/pkg/zebra"
+	"github.com/osrg/gobgp/pkg/packet/bgp"
+
+	log "github.com/sirupsen/logrus"
+)
+
+// nexthopStateCache stores a map of nexthop IP to metric value. Especially,
+// the metric value of math.MaxUint32 means the nexthop is unreachable.
+type nexthopStateCache map[string]uint32
+
+func (m nexthopStateCache) applyToPathList(paths []*table.Path) []*table.Path {
+	updated := make([]*table.Path, 0, len(paths))
+	for _, path := range paths {
+		if path == nil || path.IsWithdraw {
+			continue
+		}
+		metric, ok := m[path.GetNexthop().String()]
+		if !ok {
+			continue
+		}
+		isNexthopInvalid := metric == math.MaxUint32
+		med, err := path.GetMed()
+		if err == nil && med == metric && path.IsNexthopInvalid == isNexthopInvalid {
+			// If the nexthop state of the given path is already up to date,
+			// skips this path.
+			continue
+		}
+		newPath := path.Clone(false)
+		if isNexthopInvalid {
+			newPath.IsNexthopInvalid = true
+		} else {
+			newPath.IsNexthopInvalid = false
+			newPath.SetMed(int64(metric), true)
+		}
+		updated = append(updated, newPath)
+	}
+	return updated
+}
+
+func (m nexthopStateCache) updateByNexthopUpdate(body *zebra.NexthopUpdateBody) (updated bool) {
+	if len(body.Nexthops) == 0 {
+		// If NEXTHOP_UPDATE message does not contain any nexthop, the given
+		// nexthop is unreachable.
+		if _, ok := m[body.Prefix.Prefix.String()]; !ok {
+			// Zebra will send an empty NEXTHOP_UPDATE message as the fist
+			// response for the NEXTHOP_REGISTER message. Here ignores it.
+			return false
+		}
+		m[body.Prefix.Prefix.String()] = math.MaxUint32 // means unreachable
+	} else {
+		m[body.Prefix.Prefix.String()] = body.Metric
+	}
+	return true
+}
+
+func (m nexthopStateCache) filterPathToRegister(paths []*table.Path) []*table.Path {
+	filteredPaths := make([]*table.Path, 0, len(paths))
+	for _, path := range paths {
+		// Here filters out:
+		// - Nil path
+		// - Withdrawn path
+		// - External path (advertised from Zebra) in order avoid sending back
+		// - Unspecified nexthop address
+		// - Already registered nexthop
+		if path == nil || path.IsWithdraw || path.IsFromExternal() {
+			continue
+		} else if nexthop := path.GetNexthop(); nexthop.IsUnspecified() {
+			continue
+		} else if _, ok := m[nexthop.String()]; ok {
+			continue
+		}
+		filteredPaths = append(filteredPaths, path)
+	}
+	return filteredPaths
+}
+
+func filterOutExternalPath(paths []*table.Path) []*table.Path {
+	filteredPaths := make([]*table.Path, 0, len(paths))
+	for _, path := range paths {
+		// Here filters out:
+		// - Nil path
+		// - External path (advertised from Zebra) in order avoid sending back
+		// - Unreachable path because invalidated by Zebra
+		if path == nil || path.IsFromExternal() || path.IsNexthopInvalid {
+			continue
+		}
+		filteredPaths = append(filteredPaths, path)
+	}
+	return filteredPaths
+}
+
+func newIPRouteBody(dst []*table.Path) (body *zebra.IPRouteBody, isWithdraw bool) {
+	paths := filterOutExternalPath(dst)
+	if len(paths) == 0 {
+		return nil, false
+	}
+	path := paths[0]
+
+	l := strings.SplitN(path.GetNlri().String(), "/", 2)
+	var prefix net.IP
+	//nexthops := make([]net.IP, 0, len(paths))
+	var nexthop zebra.Nexthop
+	nexthops := make([]zebra.Nexthop, 0, len(paths))
+	switch path.GetRouteFamily() {
+	case bgp.RF_IPv4_UC, bgp.RF_IPv4_VPN:
+		if path.GetRouteFamily() == bgp.RF_IPv4_UC {
+			prefix = path.GetNlri().(*bgp.IPAddrPrefix).IPAddrPrefixDefault.Prefix.To4()
+		} else {
+			prefix = path.GetNlri().(*bgp.LabeledVPNIPAddrPrefix).IPAddrPrefixDefault.Prefix.To4()
+		}
+		for _, p := range paths {
+			nexthop.Gate = p.GetNexthop().To4()
+			nexthops = append(nexthops, nexthop)
+		}
+	case bgp.RF_IPv6_UC, bgp.RF_IPv6_VPN:
+		if path.GetRouteFamily() == bgp.RF_IPv6_UC {
+			prefix = path.GetNlri().(*bgp.IPv6AddrPrefix).IPAddrPrefixDefault.Prefix.To16()
+		} else {
+			prefix = path.GetNlri().(*bgp.LabeledVPNIPv6AddrPrefix).IPAddrPrefixDefault.Prefix.To16()
+		}
+		for _, p := range paths {
+			nexthop.Gate = p.GetNexthop().To16()
+			nexthops = append(nexthops, nexthop)
+		}
+	default:
+		return nil, false
+	}
+	msgFlags := zebra.MESSAGE_NEXTHOP
+	plen, _ := strconv.ParseUint(l[1], 10, 8)
+	med, err := path.GetMed()
+	if err == nil {
+		msgFlags |= zebra.MESSAGE_METRIC
+	}
+	var flags zebra.FLAG
+	info := path.GetSource()
+	if info.AS == info.LocalAS {
+		flags = zebra.FLAG_IBGP | zebra.FLAG_INTERNAL
+	} else if info.MultihopTtl > 0 {
+		flags = zebra.FLAG_INTERNAL
+	}
+	return &zebra.IPRouteBody{
+		Type:    zebra.ROUTE_BGP,
+		Flags:   flags,
+		SAFI:    zebra.SAFI_UNICAST,
+		Message: msgFlags,
+		Prefix: zebra.Prefix{
+			Prefix:    prefix,
+			PrefixLen: uint8(plen),
+		},
+		Nexthops: nexthops,
+		Metric:   med,
+	}, path.IsWithdraw
+}
+
+func newNexthopRegisterBody(paths []*table.Path, nexthopCache nexthopStateCache) *zebra.NexthopRegisterBody {
+	paths = nexthopCache.filterPathToRegister(paths)
+	if len(paths) == 0 {
+		return nil
+	}
+	path := paths[0]
+
+	family := path.GetRouteFamily()
+	nexthops := make([]*zebra.RegisteredNexthop, 0, len(paths))
+	for _, p := range paths {
+		nexthop := p.GetNexthop()
+		var nh *zebra.RegisteredNexthop
+		switch family {
+		case bgp.RF_IPv4_UC, bgp.RF_IPv4_VPN:
+			nh = &zebra.RegisteredNexthop{
+				Family: syscall.AF_INET,
+				Prefix: nexthop.To4(),
+			}
+		case bgp.RF_IPv6_UC, bgp.RF_IPv6_VPN:
+			nh = &zebra.RegisteredNexthop{
+				Family: syscall.AF_INET6,
+				Prefix: nexthop.To16(),
+			}
+		default:
+			continue
+		}
+		nexthops = append(nexthops, nh)
+	}
+
+	// If no nexthop needs to be registered or unregistered, skips to send
+	// message.
+	if len(nexthops) == 0 {
+		return nil
+	}
+
+	return &zebra.NexthopRegisterBody{
+		Nexthops: nexthops,
+	}
+}
+
+func newNexthopUnregisterBody(family uint16, prefix net.IP) *zebra.NexthopRegisterBody {
+	return &zebra.NexthopRegisterBody{
+		Nexthops: []*zebra.RegisteredNexthop{{
+			Family: family,
+			Prefix: prefix,
+		}},
+	}
+}
+
+func newPathFromIPRouteMessage(m *zebra.Message, version uint8) *table.Path {
+	header := m.Header
+	body := m.Body.(*zebra.IPRouteBody)
+	family := body.RouteFamily(version)
+	isWithdraw := body.IsWithdraw(version)
+
+	var nlri bgp.AddrPrefixInterface
+	pattr := make([]bgp.PathAttributeInterface, 0)
+	origin := bgp.NewPathAttributeOrigin(bgp.BGP_ORIGIN_ATTR_TYPE_IGP)
+	pattr = append(pattr, origin)
+
+	log.WithFields(log.Fields{
+		"Topic":        "Zebra",
+		"RouteType":    body.Type.String(),
+		"Flag":         body.Flags.String(),
+		"Message":      body.Message,
+		"Family":       body.Prefix.Family,
+		"Prefix":       body.Prefix.Prefix,
+		"PrefixLength": body.Prefix.PrefixLen,
+		"Nexthop":      body.Nexthops,
+		"Metric":       body.Metric,
+		"Distance":     body.Distance,
+		"Mtu":          body.Mtu,
+		"api":          header.Command.String(),
+	}).Debugf("create path from ip route message.")
+
+	switch family {
+	case bgp.RF_IPv4_UC:
+		nlri = bgp.NewIPAddrPrefix(body.Prefix.PrefixLen, body.Prefix.Prefix.String())
+		if len(body.Nexthops) > 0 {
+			pattr = append(pattr, bgp.NewPathAttributeNextHop(body.Nexthops[0].Gate.String()))
+		}
+	case bgp.RF_IPv6_UC:
+		nlri = bgp.NewIPv6AddrPrefix(body.Prefix.PrefixLen, body.Prefix.Prefix.String())
+		nexthop := ""
+		if len(body.Nexthops) > 0 {
+			nexthop = body.Nexthops[0].Gate.String()
+		}
+		pattr = append(pattr, bgp.NewPathAttributeMpReachNLRI(nexthop, []bgp.AddrPrefixInterface{nlri}))
+	default:
+		log.WithFields(log.Fields{
+			"Topic": "Zebra",
+		}).Errorf("unsupport address family: %s", family)
+		return nil
+	}
+
+	med := bgp.NewPathAttributeMultiExitDisc(body.Metric)
+	pattr = append(pattr, med)
+
+	path := table.NewPath(nil, nlri, isWithdraw, pattr, time.Now(), false)
+	path.SetIsFromExternal(true)
+	return path
+}
+
+type zebraClient struct {
+	client       *zebra.Client
+	server       *BgpServer
+	nexthopCache nexthopStateCache
+	dead         chan struct{}
+}
+
+func (z *zebraClient) getPathListWithNexthopUpdate(body *zebra.NexthopUpdateBody) []*table.Path {
+	rib := &table.TableManager{
+		Tables: make(map[bgp.RouteFamily]*table.Table),
+	}
+
+	var rfList []bgp.RouteFamily
+	switch body.Prefix.Family {
+	case syscall.AF_INET:
+		rfList = []bgp.RouteFamily{bgp.RF_IPv4_UC, bgp.RF_IPv4_VPN}
+	case syscall.AF_INET6:
+		rfList = []bgp.RouteFamily{bgp.RF_IPv6_UC, bgp.RF_IPv6_VPN}
+	}
+
+	for _, rf := range rfList {
+		tbl, _, err := z.server.getRib("", rf, nil)
+		if err != nil {
+			log.WithFields(log.Fields{
+				"Topic":  "Zebra",
+				"Family": rf.String(),
+				"Error":  err,
+			}).Error("failed to get global rib")
+			continue
+		}
+		rib.Tables[rf] = tbl
+	}
+
+	return rib.GetPathListWithNexthop(table.GLOBAL_RIB_NAME, rfList, body.Prefix.Prefix)
+}
+
+func (z *zebraClient) updatePathByNexthopCache(paths []*table.Path) {
+	paths = z.nexthopCache.applyToPathList(paths)
+	if len(paths) > 0 {
+		if err := z.server.updatePath("", paths); err != nil {
+			log.WithFields(log.Fields{
+				"Topic":    "Zebra",
+				"PathList": paths,
+			}).Error("failed to update nexthop reachability")
+		}
+	}
+}
+
+func (z *zebraClient) loop() {
+	w := z.server.Watch([]WatchOption{
+		WatchBestPath(true),
+		WatchPostUpdate(true),
+	}...)
+	defer w.Stop()
+
+	for {
+		select {
+		case <-z.dead:
+			return
+		case msg := <-z.client.Receive():
+			switch body := msg.Body.(type) {
+			case *zebra.IPRouteBody:
+				if path := newPathFromIPRouteMessage(msg, z.client.Version); path != nil {
+					if err := z.server.addPathList("", []*table.Path{path}); err != nil {
+						log.WithFields(log.Fields{
+							"Topic": "Zebra",
+							"Path":  path,
+							"Error": err,
+						}).Error("failed to add path from zebra")
+					}
+				}
+			case *zebra.NexthopUpdateBody:
+				if updated := z.nexthopCache.updateByNexthopUpdate(body); !updated {
+					continue
+				}
+				paths := z.getPathListWithNexthopUpdate(body)
+				if len(paths) == 0 {
+					// If there is no path bound for the given nexthop, send
+					// NEXTHOP_UNREGISTER message.
+					z.client.SendNexthopRegister(msg.Header.VrfId, newNexthopUnregisterBody(uint16(body.Prefix.Family), body.Prefix.Prefix), true)
+					delete(z.nexthopCache, body.Prefix.Prefix.String())
+				}
+				z.updatePathByNexthopCache(paths)
+			}
+		case ev := <-w.Event():
+			switch msg := ev.(type) {
+			case *WatchEventBestPath:
+				if table.UseMultiplePaths.Enabled {
+					for _, paths := range msg.MultiPathList {
+						z.updatePathByNexthopCache(paths)
+						if body, isWithdraw := newIPRouteBody(paths); body != nil {
+							z.client.SendIPRoute(0, body, isWithdraw)
+						}
+						if body := newNexthopRegisterBody(paths, z.nexthopCache); body != nil {
+							z.client.SendNexthopRegister(0, body, false)
+						}
+					}
+				} else {
+					z.updatePathByNexthopCache(msg.PathList)
+					for _, path := range msg.PathList {
+						vrfs := []uint32{0}
+						if msg.Vrf != nil {
+							if v, ok := msg.Vrf[path.GetNlri().String()]; ok {
+								vrfs = append(vrfs, v)
+							}
+						}
+						for _, i := range vrfs {
+							if body, isWithdraw := newIPRouteBody([]*table.Path{path}); body != nil {
+								z.client.SendIPRoute(i, body, isWithdraw)
+							}
+							if body := newNexthopRegisterBody([]*table.Path{path}, z.nexthopCache); body != nil {
+								z.client.SendNexthopRegister(i, body, false)
+							}
+						}
+					}
+				}
+			case *WatchEventUpdate:
+				if body := newNexthopRegisterBody(msg.PathList, z.nexthopCache); body != nil {
+					vrfID := uint32(0)
+					for _, vrf := range z.server.listVrf() {
+						if vrf.Name == msg.Neighbor.Config.Vrf {
+							vrfID = uint32(vrf.Id)
+						}
+					}
+					z.client.SendNexthopRegister(vrfID, body, false)
+				}
+			}
+		}
+	}
+}
+
+func newZebraClient(s *BgpServer, url string, protos []string, version uint8, nhtEnable bool, nhtDelay uint8) (*zebraClient, error) {
+	l := strings.SplitN(url, ":", 2)
+	if len(l) != 2 {
+		return nil, fmt.Errorf("unsupported url: %s", url)
+	}
+	var cli *zebra.Client
+	var err error
+	for _, ver := range []uint8{version, 2, 3, 4, 5} {
+		cli, err = zebra.NewClient(l[0], l[1], zebra.ROUTE_BGP, ver)
+		if err == nil {
+			break
+		}
+		// Retry with another Zebra message version
+		log.WithFields(log.Fields{
+			"Topic": "Zebra",
+		}).Warnf("cannot connect to Zebra with message version %d. going to retry another version...", ver)
+	}
+	if cli == nil {
+		return nil, err
+	}
+	// Note: HELLO/ROUTER_ID_ADD messages are automatically sent to negotiate
+	// the Zebra message version in zebra.NewClient().
+	// cli.SendHello()
+	// cli.SendRouterIDAdd()
+	cli.SendInterfaceAdd()
+	for _, typ := range protos {
+		t, err := zebra.RouteTypeFromString(typ, version)
+		if err != nil {
+			return nil, err
+		}
+		cli.SendRedistribute(t, zebra.VRF_DEFAULT)
+	}
+	w := &zebraClient{
+		client:       cli,
+		server:       s,
+		nexthopCache: make(nexthopStateCache),
+		dead:         make(chan struct{}),
+	}
+	go w.loop()
+	return w, nil
+}