path: root/server/bmp.go

// 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"
	log "github.com/Sirupsen/logrus"
	"github.com/osrg/gobgp/config"
	"github.com/osrg/gobgp/packet"
	"github.com/osrg/gobgp/table"
	"gopkg.in/tomb.v2"
	"net"
	"strconv"
	"time"
)

type broadcastBMPMsg struct {
	ch      chan *broadcastBMPMsg
	msgList []*bgp.BMPMessage
	conn    *net.TCPConn
}

func (m *broadcastBMPMsg) send() {
	m.ch <- m
}

type bmpServer struct {
	conn *net.TCPConn
	host string
	typ  config.BmpRouteMonitoringPolicyType
}

type bmpConfig struct {
	config config.BmpServerConfig
	del    bool
	errCh  chan error
}

type bmpWatcher struct {
	t           tomb.Tomb
	ch          chan watcherEvent
	apiCh       chan *GrpcRequest
	newServerCh chan *bmpServer
	endCh       chan *net.TCPConn
	connMap     map[string]*bmpServer
	ctlCh       chan *bmpConfig
}

func (w *bmpWatcher) notify(t watcherEventType) chan watcherEvent {
	if t == WATCHER_EVENT_UPDATE_MSG || t == WATCHER_EVENT_POST_POLICY_UPDATE_MSG || t == WATCHER_EVENT_STATE_CHANGE {
		return w.ch
	}
	return nil
}

func (w *bmpWatcher) stop() {
	w.t.Kill(nil)
}

func (w *bmpWatcher) tryConnect(server *bmpServer) {
	interval := 1
	host := server.host
	for {
		log.Debug("connecting bmp server: ", host)
		conn, err := net.Dial("tcp", host)
		if err != nil {
			time.Sleep(time.Duration(interval) * time.Second)
			if interval < 30 {
				interval *= 2
			}
		} else {
			log.Info("bmp server is connected, ", host)
			server.conn = conn.(*net.TCPConn)
			go func() {
				buf := make([]byte, 1)
				for {
					_, err := conn.Read(buf)
					if err != nil {
						w.endCh <- conn.(*net.TCPConn)
						return
					}
				}
			}()
			w.newServerCh <- server
			break
		}
	}
}

func (w *bmpWatcher) loop() error {
	for {
		select {
		case <-w.t.Dying():
			for _, server := range w.connMap {
				if server.conn != nil {
					server.conn.Close()
				}
			}
			return nil
		case m := <-w.ctlCh:
			c := m.config
			if m.del {
				host := net.JoinHostPort(c.Address, strconv.Itoa(int(c.Port)))
				if _, y := w.connMap[host]; !y {
					m.errCh <- fmt.Errorf("bmp server %s doesn't exists", host)
					continue
				}
				conn := w.connMap[host].conn
				delete(w.connMap, host)
				conn.Close()
			} else {
				host := net.JoinHostPort(c.Address, strconv.Itoa(int(c.Port)))
				if _, y := w.connMap[host]; y {
					m.errCh <- fmt.Errorf("bmp server %s already exists", host)
					continue
				}
				server := &bmpServer{
					host: host,
					typ:  c.RouteMonitoringPolicy,
				}
				w.connMap[host] = server
				go w.tryConnect(server)
			}
			m.errCh <- nil
			close(m.errCh)
		case server := <-w.newServerCh:
			i := bgp.NewBMPInitiation([]bgp.BMPTLV{})
			buf, _ := i.Serialize()
			_, err := server.conn.Write(buf)
			if err != nil {
				log.Warnf("failed to write to bmp server %s", server.host)
			}
			req := &GrpcRequest{
				RequestType: REQ_BMP_NEIGHBORS,
				ResponseCh:  make(chan *GrpcResponse, 1),
			}
			w.apiCh <- req
			write := func(req *GrpcRequest) {
				for res := range req.ResponseCh {
					for _, msg := range res.Data.([]*bgp.BMPMessage) {
						buf, _ = msg.Serialize()
						_, err := server.conn.Write(buf)
						if err != nil {
							log.Warnf("failed to write to bmp server %s", server.host)
						}
					}
				}
			}
			write(req)
			if server.typ != config.BMP_ROUTE_MONITORING_POLICY_TYPE_POST_POLICY {
				req = &GrpcRequest{
					RequestType: REQ_BMP_ADJ_IN,
					ResponseCh:  make(chan *GrpcResponse, 1),
				}
				w.apiCh <- req
				write(req)
			}
			if server.typ != config.BMP_ROUTE_MONITORING_POLICY_TYPE_PRE_POLICY {
				req = &GrpcRequest{
					RequestType: REQ_BMP_GLOBAL,
					ResponseCh:  make(chan *GrpcResponse, 1),
				}
				w.apiCh <- req
				write(req)
			}
		case ev := <-w.ch:
			switch msg := ev.(type) {
			case *watcherEventUpdateMsg:
				info := &table.PeerInfo{
					Address: msg.peerAddress,
					AS:      msg.peerAS,
					ID:      msg.peerID,
				}
				buf, _ := bmpPeerRoute(bgp.BMP_PEER_TYPE_GLOBAL, msg.postPolicy, 0, info, msg.timestamp.Unix(), msg.payload).Serialize()
				for _, server := range w.connMap {
					if server.conn != nil {
						send := server.typ != config.BMP_ROUTE_MONITORING_POLICY_TYPE_POST_POLICY && !msg.postPolicy
						send = send || (server.typ != config.BMP_ROUTE_MONITORING_POLICY_TYPE_PRE_POLICY && msg.postPolicy)
						if send {
							_, err := server.conn.Write(buf)
							if err != nil {
								log.Warnf("failed to write to bmp server %s", server.host)
							}
						}
					}
				}
			case *watcherEventStateChangedMsg:
				var bmpmsg *bgp.BMPMessage
				info := &table.PeerInfo{
					Address: msg.peerAddress,
					AS:      msg.peerAS,
					ID:      msg.peerID,
				}
				if msg.state == bgp.BGP_FSM_ESTABLISHED {
					bmpmsg = bmpPeerUp(msg.localAddress.String(), msg.localPort, msg.peerPort, msg.sentOpen, msg.recvOpen, bgp.BMP_PEER_TYPE_GLOBAL, false, 0, info, msg.timestamp.Unix())
				} else {
					bmpmsg = bmpPeerDown(bgp.BMP_PEER_DOWN_REASON_UNKNOWN, bgp.BMP_PEER_TYPE_GLOBAL, false, 0, info, msg.timestamp.Unix())
				}
				buf, _ := bmpmsg.Serialize()
				for _, server := range w.connMap {
					if server.conn != nil {
						_, err := server.conn.Write(buf)
						if err != nil {
							log.Warnf("failed to write to bmp server %s", server.host)
						}
					}
				}
			default:
				log.Warnf("unknown watcher event")
			}
		case conn := <-w.endCh:
			host := conn.RemoteAddr().String()
			log.Debugf("bmp connection to %s killed", host)
			if _, y := w.connMap[host]; y {
				w.connMap[host].conn = nil
				go w.tryConnect(w.connMap[host])
			}
		}
	}
}

func (w *bmpWatcher) restart(string) error {
	return nil
}

func bmpPeerUp(laddr string, lport, rport uint16, sent, recv *bgp.BGPMessage, t uint8, policy bool, pd uint64, peeri *table.PeerInfo, timestamp int64) *bgp.BMPMessage {
	ph := bgp.NewBMPPeerHeader(t, policy, pd, peeri.Address.String(), peeri.AS, peeri.ID.String(), float64(timestamp))
	return bgp.NewBMPPeerUpNotification(*ph, laddr, lport, rport, sent, recv)
}

func bmpPeerDown(reason uint8, t uint8, policy bool, pd uint64, peeri *table.PeerInfo, timestamp int64) *bgp.BMPMessage {
	ph := bgp.NewBMPPeerHeader(t, policy, pd, peeri.Address.String(), peeri.AS, peeri.ID.String(), float64(timestamp))
	return bgp.NewBMPPeerDownNotification(*ph, reason, nil, []byte{})
}

func bmpPeerRoute(t uint8, policy bool, pd uint64, peeri *table.PeerInfo, timestamp int64, payload []byte) *bgp.BMPMessage {
	ph := bgp.NewBMPPeerHeader(t, policy, pd, peeri.Address.String(), peeri.AS, peeri.ID.String(), float64(timestamp))
	m := bgp.NewBMPRouteMonitoring(*ph, nil)
	body := m.Body.(*bgp.BMPRouteMonitoring)
	body.BGPUpdatePayload = payload
	return m
}

func (w *bmpWatcher) addServer(c config.BmpServerConfig) error {
	ch := make(chan error)
	w.ctlCh <- &bmpConfig{
		config: c,
		errCh:  ch,
	}
	return <-ch
}

func (w *bmpWatcher) deleteServer(c config.BmpServerConfig) error {
	ch := make(chan error)
	w.ctlCh <- &bmpConfig{
		config: c,
		del:    true,
		errCh:  ch,
	}
	return <-ch
}

func (w *bmpWatcher) watchingEventTypes() []watcherEventType {
	state := false
	pre := false
	post := false
	for _, server := range w.connMap {
		if server.conn != nil {
			state = true
			if server.typ != config.BMP_ROUTE_MONITORING_POLICY_TYPE_POST_POLICY {
				pre = true
			}
			if server.typ != config.BMP_ROUTE_MONITORING_POLICY_TYPE_PRE_POLICY {
				post = true
			}
		}
	}
	types := make([]watcherEventType, 0, 3)
	if state {
		types = append(types, WATCHER_EVENT_STATE_CHANGE)
	}
	if pre {
		types = append(types, WATCHER_EVENT_UPDATE_MSG)
	}
	if post {
		types = append(types, WATCHER_EVENT_POST_POLICY_UPDATE_MSG)
	}
	return types
}

func newBmpWatcher(grpcCh chan *GrpcRequest) (*bmpWatcher, error) {
	w := &bmpWatcher{
		ch:          make(chan watcherEvent),
		apiCh:       grpcCh,
		newServerCh: make(chan *bmpServer),
		endCh:       make(chan *net.TCPConn),
		connMap:     make(map[string]*bmpServer),
		ctlCh:       make(chan *bmpConfig),
	}
	w.t.Go(w.loop)
	return w, nil
}