summaryrefslogtreecommitdiffhomepage
diff options
context:
space:
mode:
Diffstat
-rw-r--r--api/gobgp.pb.go68
-rw-r--r--api/gobgp.proto1
-rw-r--r--server/grpc_server.go34
-rw-r--r--server/server.go26
4 files changed, 125 insertions, 4 deletions
diff --git a/api/gobgp.pb.go b/api/gobgp.pb.go
index 7d01de14..7aeb6e26 100644
--- a/api/gobgp.pb.go
+++ b/api/gobgp.pb.go
@@ -1182,6 +1182,7 @@ type GrpcClient interface {
GetNeighbor(ctx context.Context, in *Arguments, opts ...grpc.CallOption) (*Peer, error)
GetRib(ctx context.Context, in *Arguments, opts ...grpc.CallOption) (Grpc_GetRibClient, error)
GetAdjRib(ctx context.Context, in *Arguments, opts ...grpc.CallOption) (Grpc_GetAdjRibClient, error)
+ MonitorBestChanged(ctx context.Context, in *Arguments, opts ...grpc.CallOption) (Grpc_MonitorBestChangedClient, error)
Reset(ctx context.Context, in *Arguments, opts ...grpc.CallOption) (*Error, error)
SoftReset(ctx context.Context, in *Arguments, opts ...grpc.CallOption) (*Error, error)
SoftResetIn(ctx context.Context, in *Arguments, opts ...grpc.CallOption) (*Error, error)
@@ -1310,6 +1311,38 @@ func (x *grpcGetAdjRibClient) Recv() (*Path, error) {
return m, nil
}
+func (c *grpcClient) MonitorBestChanged(ctx context.Context, in *Arguments, opts ...grpc.CallOption) (Grpc_MonitorBestChangedClient, error) {
+ stream, err := grpc.NewClientStream(ctx, &_Grpc_serviceDesc.Streams[3], c.cc, "/api.Grpc/MonitorBestChanged", opts...)
+ if err != nil {
+ return nil, err
+ }
+ x := &grpcMonitorBestChangedClient{stream}
+ if err := x.ClientStream.SendMsg(in); err != nil {
+ return nil, err
+ }
+ if err := x.ClientStream.CloseSend(); err != nil {
+ return nil, err
+ }
+ return x, nil
+}
+
+type Grpc_MonitorBestChangedClient interface {
+ Recv() (*Path, error)
+ grpc.ClientStream
+}
+
+type grpcMonitorBestChangedClient struct {
+ grpc.ClientStream
+}
+
+func (x *grpcMonitorBestChangedClient) Recv() (*Path, error) {
+ m := new(Path)
+ if err := x.ClientStream.RecvMsg(m); err != nil {
+ return nil, err
+ }
+ return m, nil
+}
+
func (c *grpcClient) Reset(ctx context.Context, in *Arguments, opts ...grpc.CallOption) (*Error, error) {
out := new(Error)
err := grpc.Invoke(ctx, "/api.Grpc/Reset", in, out, c.cc, opts...)
@@ -1374,7 +1407,7 @@ func (c *grpcClient) Disable(ctx context.Context, in *Arguments, opts ...grpc.Ca
}
func (c *grpcClient) ModPath(ctx context.Context, opts ...grpc.CallOption) (Grpc_ModPathClient, error) {
- stream, err := grpc.NewClientStream(ctx, &_Grpc_serviceDesc.Streams[3], c.cc, "/api.Grpc/ModPath", opts...)
+ stream, err := grpc.NewClientStream(ctx, &_Grpc_serviceDesc.Streams[4], c.cc, "/api.Grpc/ModPath", opts...)
if err != nil {
return nil, err
}
@@ -1417,7 +1450,7 @@ func (c *grpcClient) GetNeighborPolicy(ctx context.Context, in *Arguments, opts
}
func (c *grpcClient) ModNeighborPolicy(ctx context.Context, opts ...grpc.CallOption) (Grpc_ModNeighborPolicyClient, error) {
- stream, err := grpc.NewClientStream(ctx, &_Grpc_serviceDesc.Streams[4], c.cc, "/api.Grpc/ModNeighborPolicy", opts...)
+ stream, err := grpc.NewClientStream(ctx, &_Grpc_serviceDesc.Streams[5], c.cc, "/api.Grpc/ModNeighborPolicy", opts...)
if err != nil {
return nil, err
}
@@ -1448,7 +1481,7 @@ func (x *grpcModNeighborPolicyClient) Recv() (*Error, error) {
}
func (c *grpcClient) GetPolicyRoutePolicies(ctx context.Context, in *PolicyArguments, opts ...grpc.CallOption) (Grpc_GetPolicyRoutePoliciesClient, error) {
- stream, err := grpc.NewClientStream(ctx, &_Grpc_serviceDesc.Streams[5], c.cc, "/api.Grpc/GetPolicyRoutePolicies", opts...)
+ stream, err := grpc.NewClientStream(ctx, &_Grpc_serviceDesc.Streams[6], c.cc, "/api.Grpc/GetPolicyRoutePolicies", opts...)
if err != nil {
return nil, err
}
@@ -1489,7 +1522,7 @@ func (c *grpcClient) GetPolicyRoutePolicy(ctx context.Context, in *PolicyArgumen
}
func (c *grpcClient) ModPolicyRoutePolicy(ctx context.Context, opts ...grpc.CallOption) (Grpc_ModPolicyRoutePolicyClient, error) {
- stream, err := grpc.NewClientStream(ctx, &_Grpc_serviceDesc.Streams[6], c.cc, "/api.Grpc/ModPolicyRoutePolicy", opts...)
+ stream, err := grpc.NewClientStream(ctx, &_Grpc_serviceDesc.Streams[7], c.cc, "/api.Grpc/ModPolicyRoutePolicy", opts...)
if err != nil {
return nil, err
}
@@ -1526,6 +1559,7 @@ type GrpcServer interface {
GetNeighbor(context.Context, *Arguments) (*Peer, error)
GetRib(*Arguments, Grpc_GetRibServer) error
GetAdjRib(*Arguments, Grpc_GetAdjRibServer) error
+ MonitorBestChanged(*Arguments, Grpc_MonitorBestChangedServer) error
Reset(context.Context, *Arguments) (*Error, error)
SoftReset(context.Context, *Arguments) (*Error, error)
SoftResetIn(context.Context, *Arguments) (*Error, error)
@@ -1620,6 +1654,27 @@ func (x *grpcGetAdjRibServer) Send(m *Path) error {
return x.ServerStream.SendMsg(m)
}
+func _Grpc_MonitorBestChanged_Handler(srv interface{}, stream grpc.ServerStream) error {
+ m := new(Arguments)
+ if err := stream.RecvMsg(m); err != nil {
+ return err
+ }
+ return srv.(GrpcServer).MonitorBestChanged(m, &grpcMonitorBestChangedServer{stream})
+}
+
+type Grpc_MonitorBestChangedServer interface {
+ Send(*Path) error
+ grpc.ServerStream
+}
+
+type grpcMonitorBestChangedServer struct {
+ grpc.ServerStream
+}
+
+func (x *grpcMonitorBestChangedServer) Send(m *Path) error {
+ return x.ServerStream.SendMsg(m)
+}
+
func _Grpc_Reset_Handler(srv interface{}, ctx context.Context, codec grpc.Codec, buf []byte) (interface{}, error) {
in := new(Arguments)
if err := codec.Unmarshal(buf, in); err != nil {
@@ -1889,6 +1944,11 @@ var _Grpc_serviceDesc = grpc.ServiceDesc{
ServerStreams: true,
},
{
+ StreamName: "MonitorBestChanged",
+ Handler: _Grpc_MonitorBestChanged_Handler,
+ ServerStreams: true,
+ },
+ {
StreamName: "ModPath",
Handler: _Grpc_ModPath_Handler,
ClientStreams: true,
diff --git a/api/gobgp.proto b/api/gobgp.proto
index e54eacca..9a7717f3 100644
--- a/api/gobgp.proto
+++ b/api/gobgp.proto
@@ -24,6 +24,7 @@ service Grpc {
rpc GetNeighbor(Arguments) returns (Peer) {}
rpc GetRib(Arguments) returns (stream Destination) {}
rpc GetAdjRib(Arguments) returns (stream Path) {}
+ rpc MonitorBestChanged(Arguments) returns (stream Path) {}
rpc Reset(Arguments) returns (Error) {}
rpc SoftReset(Arguments) returns (Error) {}
rpc SoftResetIn(Arguments) returns (Error) {}
diff --git a/server/grpc_server.go b/server/grpc_server.go
index b14845b2..aa36a264 100644
--- a/server/grpc_server.go
+++ b/server/grpc_server.go
@@ -48,6 +48,7 @@ const (
REQ_GLOBAL_RIB
REQ_GLOBAL_ADD
REQ_GLOBAL_DELETE
+ REQ_GLOBAL_MONITOR_BEST_CHANGED
REQ_POLICY_PREFIX
REQ_POLICY_PREFIXES
REQ_POLICY_PREFIX_ADD
@@ -205,6 +206,37 @@ func (s *Server) GetRib(arg *api.Arguments, stream api.Grpc_GetRibServer) error
return nil
}
+func (s *Server) MonitorBestChanged(arg *api.Arguments, stream api.Grpc_MonitorBestChangedServer) error {
+ var reqType int
+ switch arg.Resource {
+ case api.Resource_GLOBAL:
+ reqType = REQ_GLOBAL_MONITOR_BEST_CHANGED
+ default:
+ return fmt.Errorf("unsupported resource type: %v", arg.Resource)
+ }
+
+ rf, err := convertAf2Rf(arg.Af)
+ if err != nil {
+ return err
+ }
+
+ req := NewGrpcRequest(reqType, "", rf, nil)
+ s.bgpServerCh <- req
+
+ for res := range req.ResponseCh {
+ if err = res.Err(); err != nil {
+ log.Debug(err.Error())
+ goto END
+ }
+ if err = stream.Send(res.Data.(*api.Path)); err != nil {
+ goto END
+ }
+ }
+END:
+ req.EndCh <- struct{}{}
+ return err
+}
+
func (s *Server) neighbor(reqType int, arg *api.Arguments) (*api.Error, error) {
rf, err := convertAf2Rf(arg.Af)
if err != nil {
@@ -512,6 +544,7 @@ type GrpcRequest struct {
RemoteAddr string
RouteFamily bgp.RouteFamily
ResponseCh chan *GrpcResponse
+ EndCh chan struct{}
Err error
Data interface{}
}
@@ -522,6 +555,7 @@ func NewGrpcRequest(reqType int, remoteAddr string, rf bgp.RouteFamily, d interf
RouteFamily: rf,
RemoteAddr: remoteAddr,
ResponseCh: make(chan *GrpcResponse),
+ EndCh: make(chan struct{}),
Data: d,
}
return r
diff --git a/server/server.go b/server/server.go
index 939c2f56..cc19ecc7 100644
--- a/server/server.go
+++ b/server/server.go
@@ -52,6 +52,7 @@ type BgpServer struct {
policyUpdateCh chan config.RoutingPolicy
policyMap map[string]*policy.Policy
routingPolicy config.RoutingPolicy
+ broadcastReqs []*GrpcRequest
neighborMap map[string]*Peer
localRibMap map[string]*LocalRib
@@ -354,6 +355,9 @@ func (server *BgpServer) dropPeerAllRoutes(peer *Peer) []*SenderMsg {
if len(pathList) == 0 {
continue
}
+
+ server.broadcastBests(pathList)
+
msgList := table.CreateUpdateMsgFromPaths(pathList)
for _, targetPeer := range server.neighborMap {
if targetPeer.isRouteServerClient() || targetPeer.fsm.state != bgp.BGP_FSM_ESTABLISHED {
@@ -404,6 +408,24 @@ func applyPolicies(peer *Peer, loc *LocalRib, isExport bool, pathList []table.Pa
return ret
}
+func (server *BgpServer) broadcastBests(bests []table.Path) {
+ for _, path := range bests {
+ result := &GrpcResponse{
+ Data: path.ToApiStruct(),
+ }
+ remainReqs := make([]*GrpcRequest, 0, len(server.broadcastReqs))
+ for _, req := range server.broadcastReqs {
+ select {
+ case <-req.EndCh:
+ continue
+ case req.ResponseCh <- result:
+ }
+ remainReqs = append(remainReqs, req)
+ }
+ server.broadcastReqs = remainReqs
+ }
+}
+
func (server *BgpServer) propagateUpdate(neighborAddress string, RouteServerClient bool, pathList []table.Path) []*SenderMsg {
msgs := make([]*SenderMsg, 0)
@@ -432,6 +454,8 @@ func (server *BgpServer) propagateUpdate(neighborAddress string, RouteServerClie
return msgs
}
+ server.broadcastBests(sendPathList)
+
for _, targetPeer := range server.neighborMap {
if targetPeer.isRouteServerClient() || targetPeer.fsm.state != bgp.BGP_FSM_ESTABLISHED {
continue
@@ -790,6 +814,8 @@ func (server *BgpServer) handleGrpc(grpcReq *GrpcRequest) []*SenderMsg {
close(grpcReq.ResponseCh)
}
+ case REQ_GLOBAL_MONITOR_BEST_CHANGED:
+ server.broadcastReqs = append(server.broadcastReqs, grpcReq)
case REQ_NEIGHBORS:
for _, peer := range server.neighborMap {
result := &GrpcResponse{
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-16 01:29:32 +0000