// 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 openswitch import ( "fmt" "net" "reflect" "strconv" "time" log "github.com/Sirupsen/logrus" api "github.com/osrg/gobgp/api" "github.com/osrg/gobgp/packet/bgp" "github.com/osrg/gobgp/table" "github.com/satori/go.uuid" ovsdb "github.com/socketplane/libovsdb" "golang.org/x/net/context" "google.golang.org/grpc" ) const ( TARGET_TABLE = "OpenSwitch" ) const ( NEXTHOP_TRANSACT_NUUID = "nexthop" ROUTE_TRANSACT_NUUID = "route" ) type Notifier struct { updateCh chan *ovsdb.TableUpdates } func (n Notifier) Update(context interface{}, tableUpdates ovsdb.TableUpdates) { n.updateCh <- &tableUpdates } func (n Notifier) Locked([]interface{}) { } func (n Notifier) Stolen([]interface{}) { } func (n Notifier) Echo([]interface{}) { } func (n Notifier) Disconnected(client *ovsdb.OvsdbClient) { } func NewNotifier(ch chan *ovsdb.TableUpdates) *Notifier { return &Notifier{ updateCh: ch, } } func (m *OpsManager) populateCache(updates ovsdb.TableUpdates) { for table, tableUpdate := range updates.Updates { if _, ok := m.cache[table]; !ok { m.cache[table] = make(map[string]ovsdb.Row) } for uuid, row := range tableUpdate.Rows { empty := ovsdb.Row{} if !reflect.DeepEqual(row.New, empty) { m.cache[table][uuid] = row.New } else { delete(m.cache[table], uuid) } } } } func extractUUID(v interface{}) uuid.UUID { vv, ok := v.([]interface{}) if !ok { return uuid.Nil } if len(vv) != 2 || vv[0].(string) != "uuid" { return uuid.Nil } return uuid.FromStringOrNil(vv[1].(string)) } func (m *OpsManager) getRootUUID() (uuid.UUID, error) { for k, _ := range m.cache[TARGET_TABLE] { return uuid.FromStringOrNil(k), nil } return uuid.Nil, fmt.Errorf("OpenSwitch table not found") } func (m *OpsManager) getVrfUUID() (uuid.UUID, error) { vrfs, ok := m.cache["VRF"] if !ok { return uuid.Nil, fmt.Errorf("VRF table not found") } for k, _ := range vrfs { return uuid.FromStringOrNil(k), nil } return uuid.Nil, fmt.Errorf("uuid not found in VRF table") } func (m *OpsManager) getBGPRouterUUID() (uint32, uuid.UUID, error) { var asn uint32 vrfs, ok := m.cache["VRF"] if !ok { return asn, uuid.Nil, fmt.Errorf("VRF table not found") } for _, v := range vrfs { if v.Fields["name"] == "vrf_default" { routers := v.Fields["bgp_routers"].(ovsdb.OvsMap).GoMap if len(routers) < 1 { return asn, uuid.Nil, fmt.Errorf("no bgp router configured") } if len(routers) > 1 { return asn, uuid.Nil, fmt.Errorf("default vrf has multiple bgp router setting") } for k, v := range routers { asn = uint32(k.(float64)) id := extractUUID(v) if id == uuid.Nil { return asn, uuid.Nil, fmt.Errorf("invalid bgp router schema") } return asn, id, nil } } } return asn, uuid.Nil, fmt.Errorf("row not found in vrf table") } func parseRouteToGobgp(route ovsdb.RowUpdate, nexthops map[string]ovsdb.Row) (*api.Path, bool, bool, error) { var nlri bgp.AddrPrefixInterface path := &api.Path{ IsFromExternal: true, Pattrs: make([][]byte, 0), } isWithdraw := false isFromGobgp := false prefix := route.New.Fields["prefix"].(string) safi := route.New.Fields["sub_address_family"].(string) afi := route.New.Fields["address_family"].(string) m := route.New.Fields["metric"].(float64) attrs := route.New.Fields["path_attributes"].(ovsdb.OvsMap).GoMap if attrs["IsFromGobgp"] == "true" { isFromGobgp = true } nh := make([]interface{}, 0) nhId, ok := route.New.Fields["bgp_nexthops"].(ovsdb.UUID) if ok { for id, n := range nexthops { if id == nhId.GoUUID { nh = append(nh, n.Fields["ip_address"]) } } } nexthop := "0.0.0.0" if afi == "ipv6" { nexthop = "::" } if len(nh) == 0 { log.WithFields(log.Fields{ "Topic": "openswitch", }).Debug("nexthop address does not exist") } else if len(nh) == 1 { if net.ParseIP(nh[0].(string)) == nil { return nil, isWithdraw, isFromGobgp, fmt.Errorf("invalid nexthop address") } else { nexthop = nh[0].(string) } } else { return nil, isWithdraw, isFromGobgp, fmt.Errorf("route has multiple nexthop address") } med, _ := bgp.NewPathAttributeMultiExitDisc(uint32(m)).Serialize() path.Pattrs = append(path.Pattrs, med) lpref, err := strconv.Atoi(attrs["BGP_loc_pref"].(string)) if err != nil { return nil, isWithdraw, isFromGobgp, err } localPref, _ := bgp.NewPathAttributeLocalPref(uint32(lpref)).Serialize() path.Pattrs = append(path.Pattrs, localPref) var origin_t int switch attrs["BGP_origin"].(string) { case "i": origin_t = bgp.BGP_ORIGIN_ATTR_TYPE_IGP case "e": origin_t = bgp.BGP_ORIGIN_ATTR_TYPE_EGP case "?": origin_t = bgp.BGP_ORIGIN_ATTR_TYPE_INCOMPLETE default: return nil, isWithdraw, isFromGobgp, fmt.Errorf("invalid origin") } origin, _ := bgp.NewPathAttributeOrigin(uint8(origin_t)).Serialize() path.Pattrs = append(path.Pattrs, origin) switch afi { case "ipv4", "ipv6": ip, net, err := net.ParseCIDR(prefix) if err != nil { return nil, isWithdraw, isFromGobgp, err } ones, _ := net.Mask.Size() if afi == "ipv4" { if ip.To4() == nil { return nil, isWithdraw, isFromGobgp, fmt.Errorf("invalid ipv4 prefix") } nlri = bgp.NewIPAddrPrefix(uint8(ones), ip.String()) } else { if ip.To16() == nil { return nil, isWithdraw, isFromGobgp, fmt.Errorf("invalid ipv6 prefix") } nlri = bgp.NewIPv6AddrPrefix(uint8(ones), ip.String()) } default: return nil, isWithdraw, isFromGobgp, fmt.Errorf("unsupported address family: %s", afi) } if afi == "ipv4" && safi == "unicast" { path.Nlri, _ = nlri.Serialize() n, _ := bgp.NewPathAttributeNextHop(nexthop).Serialize() path.Pattrs = append(path.Pattrs, n) } else { mpreach, _ := bgp.NewPathAttributeMpReachNLRI(nexthop, []bgp.AddrPrefixInterface{nlri}).Serialize() path.Pattrs = append(path.Pattrs, mpreach) } if attrs["BGP_flags"].(string) == "512" { isWithdraw = true } return path, isWithdraw, isFromGobgp, nil } func (m *OpsManager) getBGPNeighborUUIDs(id uuid.UUID) ([]net.IP, []uuid.UUID, error) { global, ok := m.cache["BGP_Router"] if !ok { return nil, nil, fmt.Errorf("BGP_Router table not found") } for k, v := range global { if uuid.Equal(id, uuid.FromStringOrNil(k)) { neighbors := v.Fields["bgp_neighbors"].(ovsdb.OvsMap).GoMap if len(neighbors) < 1 { return nil, nil, fmt.Errorf("no bgp neighbor configured") } addrs := make([]net.IP, 0, len(neighbors)) ids := make([]uuid.UUID, 0, len(neighbors)) for k, v := range neighbors { addrs = append(addrs, net.ParseIP(k.(string))) id := extractUUID(v) if id == uuid.Nil { return nil, nil, fmt.Errorf("invalid uuid schema") } ids = append(ids, id) } return addrs, ids, nil } } return nil, nil, fmt.Errorf("neighbor not found") } func (m *OpsManager) handleVrfUpdate(cli api.GobgpApiClient, update ovsdb.TableUpdate) { for _, v := range update.Rows { if len(v.Old.Fields) == 0 { log.WithFields(log.Fields{ "Topic": "openswitch", }).Debug("new vrf") } else if _, ok := v.Old.Fields["bgp_routers"]; ok { if _, _, err := m.getBGPRouterUUID(); err != nil { cli.StopServer(context.Background(), &api.StopServerRequest{}) return } } } } func (m *OpsManager) handleBgpRouterUpdate(cli api.GobgpApiClient, update ovsdb.TableUpdate) { asn, id, err := m.getBGPRouterUUID() if err != nil { log.WithFields(log.Fields{ "Topic": "openswitch", "Error": err, }).Debug("Could not get BGP Router UUID") return } for k, v := range update.Rows { if uuid.Equal(id, uuid.FromStringOrNil(k)) { initial := false if len(v.Old.Fields) == 0 { log.WithFields(log.Fields{ "Topic": "openswitch", }).Debug("new bgp router") initial = true } if _, ok := v.Old.Fields["router_id"]; initial || ok { r, ok := v.New.Fields["router_id"].(string) if !ok { log.WithFields(log.Fields{ "Topic": "openswitch", }).Debug("router-id is not configured yet") return } cli.StartServer(context.Background(), &api.StartServerRequest{ Global: &api.Global{ As: asn, RouterId: r, }, }) } if o, ok := v.Old.Fields["bgp_neighbors"]; ok { oldNeighMap := o.(ovsdb.OvsMap).GoMap newNeighMap := v.New.Fields["bgp_neighbors"].(ovsdb.OvsMap).GoMap for k, _ := range oldNeighMap { if _, ok := newNeighMap[k]; !ok { cli.DeleteNeighbor(context.Background(), &api.DeleteNeighborRequest{ Peer: &api.Peer{ Conf: &api.PeerConf{ NeighborAddress: k.(string), }, }, }) } } } } } } func (m *OpsManager) handleNeighborUpdate(cli api.GobgpApiClient, update ovsdb.TableUpdate) { _, id, _ := m.getBGPRouterUUID() addrs, ids, err := m.getBGPNeighborUUIDs(id) if err != nil { return } for k, v := range update.Rows { for idx, id := range ids { if uuid.Equal(id, uuid.FromStringOrNil(k)) { asn, ok := v.New.Fields["remote_as"].(float64) if !ok { log.WithFields(log.Fields{ "Topic": "openswitch", }).Debug("remote-as is not configured yet") continue } cli.AddNeighbor(context.Background(), &api.AddNeighborRequest{ Peer: &api.Peer{ Conf: &api.PeerConf{ NeighborAddress: addrs[idx].String(), PeerAs: uint32(asn), }, }, }) } } } } func (m *OpsManager) handleRouteUpdate(cli api.GobgpApiClient, update ovsdb.TableUpdate) { id, _ := m.getVrfUUID() for _, v := range update.Rows { vrf := v.New.Fields["vrf"] if vrf == nil { continue } idx := vrf.(ovsdb.UUID).GoUUID if uuid.Equal(id, uuid.FromStringOrNil(idx)) { path, isWithdraw, isFromGobgp, err := parseRouteToGobgp(v, m.cache["BGP_Nexthop"]) if err != nil { log.WithFields(log.Fields{ "Topic": "openswitch", "Path": path, "Error": err, }).Debug("failed to parse path") return } if isWithdraw { cli.DeletePath(context.Background(), &api.DeletePathRequest{ Resource: api.Resource_GLOBAL, Path: path, }) } else { if isFromGobgp { return } cli.AddPath(context.Background(), &api.AddPathRequest{ Resource: api.Resource_GLOBAL, Path: path, }) } } } } func parseRouteToOps(p *table.Path) (map[string]interface{}, bool, error) { route := map[string]interface{}{"metric": 0, "peer": "Remote announcement"} IsWithdraw := false var nexthop string pathAttr := map[string]string{"BGP_iBGP": "false", "BGP_flags": "16", "BGP_internal": "false", "BGP_loc_pref": "0", "IsFromGobgp": "true", } for _, a := range p.GetPathAttrs() { switch a.GetType() { case bgp.BGP_ATTR_TYPE_NEXT_HOP: nexthop = a.(*bgp.PathAttributeNextHop).Value.String() case bgp.BGP_ATTR_TYPE_MP_REACH_NLRI: n := a.(*bgp.PathAttributeMpReachNLRI).Nexthop if n != nil { nexthop = n.String() } else { nexthop = "" } case bgp.BGP_ATTR_TYPE_AS_PATH: pathAttr["BGP_AS_path"] = a.(*bgp.PathAttributeAsPath).String() case bgp.BGP_ATTR_TYPE_ORIGIN: origin := "-" switch a.(*bgp.PathAttributeOrigin).Value[0] { case bgp.BGP_ORIGIN_ATTR_TYPE_IGP: origin = "i" case bgp.BGP_ORIGIN_ATTR_TYPE_EGP: origin = "e" case bgp.BGP_ORIGIN_ATTR_TYPE_INCOMPLETE: origin = "?" } pathAttr["BGP_origin"] = origin case bgp.BGP_ATTR_TYPE_LOCAL_PREF: pathAttr["BGP_loc_pref"] = fmt.Sprintf("%v", a.(*bgp.PathAttributeLocalPref).Value) case bgp.BGP_ATTR_TYPE_MULTI_EXIT_DISC: route["metric"] = a.(*bgp.PathAttributeMultiExitDisc).Value default: continue } } IsWithdraw = p.IsWithdraw afi := "ipv4" if p.GetNlri().AFI() != bgp.AFI_IP { afi = "ipv6" } safi := "unicast" route["prefix"] = p.GetNlri().String() route["address_family"] = afi route["sub_address_family"] = safi route["bgp_nexthops"] = nexthop route["path_attributes"] = pathAttr return route, IsWithdraw, nil } func insertNextHop(opsRoute map[string]interface{}) ovsdb.Operation { nexthop := make(map[string]interface{}) nexthop["ip_address"] = opsRoute["bgp_nexthops"] nexthop["type"] = opsRoute["sub_address_family"] insNextHopOp := ovsdb.Operation{ Op: "insert", Table: "BGP_Nexthop", Row: nexthop, UUIDName: NEXTHOP_TRANSACT_NUUID, } return insNextHopOp } func insertRoute(vrfId uuid.UUID, opsRoute map[string]interface{}) (ovsdb.Operation, error) { v := []ovsdb.UUID{ovsdb.UUID{vrfId.String()}} vrfSet, _ := ovsdb.NewOvsSet(v) opsRoute["vrf"] = vrfSet nexthop := []ovsdb.UUID{ovsdb.UUID{NEXTHOP_TRANSACT_NUUID}} nexthopSet, _ := ovsdb.NewOvsSet(nexthop) opsRoute["bgp_nexthops"] = nexthopSet attrMap, err := ovsdb.NewOvsMap(opsRoute["path_attributes"]) if err != nil { return ovsdb.Operation{}, err } opsRoute["path_attributes"] = attrMap insRouteOp := ovsdb.Operation{ Op: "insert", Table: "BGP_Route", Row: opsRoute, UUIDName: ROUTE_TRANSACT_NUUID, } return insRouteOp, nil } func deleteRoute(opsRoute map[string]interface{}) ovsdb.Operation { condition := ovsdb.NewCondition("prefix", "==", opsRoute["prefix"]) deleteOp := ovsdb.Operation{ Op: "delete", Table: "BGP_Route", Where: []interface{}{condition}, } return deleteOp } func (m *OpsManager) TransactPreparation(p *table.Path) (*OpsOperation, error) { v, err := m.getVrfUUID() if err != nil { return nil, err } opsRoute, isWithdraw, err := parseRouteToOps(p) if err != nil { return nil, err } var o []ovsdb.Operation if !isWithdraw { insNextHopOp := insertNextHop(opsRoute) insRouteOp, err := insertRoute(v, opsRoute) if err != nil { return nil, err } o = []ovsdb.Operation{insNextHopOp, insRouteOp} } else { delRouteOp := deleteRoute(opsRoute) o = []ovsdb.Operation{delRouteOp} } oOperation := &OpsOperation{ operations: o, } return oOperation, nil } func (m *OpsManager) Transact(operations []ovsdb.Operation) error { ops := m.ops reply, err := ops.Transact(TARGET_TABLE, operations...) if err != nil { return err } if len(reply) < len(operations) { return fmt.Errorf("number of replies should be atleast equal to number of Operations") } var repErr error for i, o := range reply { if o.Error != "" && i < len(operations) { repErr = fmt.Errorf("transaction failed due to an error :", o.Error, " details:", o.Details, " in ", operations[i]) } else if o.Error != "" { repErr = fmt.Errorf("transaction failed due to an error :", o.Error) } } if repErr != nil { return repErr } return nil } func (m *OpsManager) GobgpMonitor(target string) { time.Sleep(time.Duration(time.Second * 2)) conn, err := grpc.Dial(target, grpc.WithTimeout(time.Second), grpc.WithBlock(), grpc.WithInsecure()) if err != nil { log.Fatal(err) } cli := api.NewGobgpApiClient(conn) rsp, err := cli.GetServer(context.Background(), &api.GetServerRequest{}) if err != nil { log.Fatal(err) } nativeOption := api.ToNativeOption{ LocalAS: uint32(rsp.Global.As), LocalID: net.ParseIP(rsp.Global.RouterId), } stream, err := cli.MonitorRib(context.Background(), &api.Table{ Type: api.Resource_GLOBAL, Family: uint32(bgp.RF_IPv4_UC), }) if err != nil { log.Fatal(err) } for { d, err := stream.Recv() if err != nil { log.Fatal(err) } dst, err := d.ToNativeDestination(nativeOption) if err != nil { log.WithFields(log.Fields{ "Topic": "openswitch", }).Error(err) continue } path := dst.GetAllKnownPathList()[0] if path.IsLocal() && !path.IsWithdraw { continue } o, err := m.TransactPreparation(path) if err != nil { log.WithFields(log.Fields{ "Topic": "openswitch", "Type": "Monitor", "Error": err, }).Error("failed transact preparation of ops") } m.opsCh <- o } } func (m *OpsManager) OpsServe(target string) error { initial, err := m.ops.MonitorAll(TARGET_TABLE, "") if err != nil { return err } go func() { m.opsUpdateCh <- initial }() conn, err := grpc.Dial(target, grpc.WithTimeout(time.Second), grpc.WithBlock(), grpc.WithInsecure()) if err != nil { log.Fatal(err) } cli := api.NewGobgpApiClient(conn) for { select { case updates := <-m.opsUpdateCh: m.populateCache(*updates) if t, ok := updates.Updates["VRF"]; ok { m.handleVrfUpdate(cli, t) } if t, ok := updates.Updates["BGP_Router"]; ok { m.handleBgpRouterUpdate(cli, t) } if t, ok := updates.Updates["BGP_Neighbor"]; ok { m.handleNeighborUpdate(cli, t) } if t, ok := updates.Updates["BGP_Route"]; ok { m.handleRouteUpdate(cli, t) } case r := <-m.opsCh: if err := m.Transact(r.operations); err != nil { } } } return nil } func (m *OpsManager) Serve() error { go m.OpsServe(m.target) go m.GobgpMonitor(m.target) return nil } type OpsOperation struct { operations []ovsdb.Operation } type OpsChs struct { opsCh chan *OpsOperation opsUpdateCh chan *ovsdb.TableUpdates } type OpsManager struct { ops *ovsdb.OvsdbClient opsCh chan *OpsOperation opsUpdateCh chan *ovsdb.TableUpdates bgpReady bool cache map[string]map[string]ovsdb.Row target string } func NewOpsManager(target string) (*OpsManager, error) { ops, err := ovsdb.Connect("", 0) if err != nil { return nil, err } opsUpdateCh := make(chan *ovsdb.TableUpdates) n := NewNotifier(opsUpdateCh) ops.Register(n) return &OpsManager{ ops: ops, opsCh: make(chan *OpsOperation, 1024), opsUpdateCh: opsUpdateCh, bgpReady: false, cache: make(map[string]map[string]ovsdb.Row), target: target, }, nil }