feat(mgmt): adding Mgmt server api for netlink DB dump

Signed-off-by: atulpatel261194 <[email protected]>

pkg/netlink/common.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"log"
 	"net"
-	"os"
 	"reflect"
 	"sync/atomic"
 
@@ -162,31 +161,20 @@ func netMaskToInt(mask int) (netmaskint [4]uint8) {
 }
 
 // dumpDBs dumps the databse
-func dumpDBs() {
-	file, err := os.OpenFile("netlink_dump", os.O_WRONLY|os.O_CREATE, 0600)
-	if err != nil {
-		panic(err)
-	}
-	if err := os.Truncate("netlink_dump", 0); err != nil {
-		log.Printf("netlink: Failed to truncate: %v", err)
-	}
+func dumpDBs() (string, error) {
 	str := dumpRouteDB()
-	log.Printf("\n")
+
 	str += dumpNexthDB()
-	log.Printf("\n")
+
 	str += dumpNeighDB()
-	log.Printf("\n")
+
 	str += dumpFDB()
-	log.Printf("\n")
+
 	str += dumpL2NexthDB()
-	_, err = file.WriteString(str)
-	if err != nil {
-		log.Printf("netlink: %v", err)
-	}
-	err = file.Close()
-	if err != nil {
-		log.Printf("netlink: error closing file: %v", err)
+	if str == "" {
+		return str, fmt.Errorf("no entries in database")
 	}
+	return str, nil
 }
 
 // checkProto checks the proto type

pkg/netlink/fdb.go

@@ -202,15 +202,12 @@ func (fdb *FdbEntryStruct) deepEqual(fdbOld *FdbEntryStruct, nc bool) bool {
 // dumpFDB dump the fdb entries
 func dumpFDB() string {
 	var s string
-	log.Printf("netlink: Dump fDB table:\n")
 	s = "fDB table:\n"
-	for _, n := range latestFDB {
+	for _, n := range fDB {
 		str := fmt.Sprintf("MacAddr(vlan=%d mac=%s state=%s type=%d l2nh_id=%d) ", n.VlanID, n.Mac, n.State, n.Type, n.Nexthop.ID)
-		log.Println(str)
 		s += str
 		s += "\n"
 	}
-	log.Printf("\n\n\n")
 	s += "\n\n"
 	return s
 }

pkg/netlink/l2nexthop.go

@@ -174,21 +174,18 @@ func (l2n *L2NexthopStruct) deepEqual(l2nOld *L2NexthopStruct, nc bool) bool {
 // dumpL2NexthDB dump the l2 nexthop entries
 func dumpL2NexthDB() string {
 	var s string
-	log.Printf("netlink: Dump L2 Nexthop table:\n")
 	s = "L2 Nexthop table:\n"
 	var ip string
-	for _, n := range latestL2Nexthop {
+	for _, n := range l2Nexthops {
 		if n.Dst == nil {
 			ip = strNone
 		} else {
 			ip = n.Dst.String()
 		}
 		str := fmt.Sprintf("L2Nexthop(id=%d dev=%s vlan=%d dst=%s type=%d #fDB entries=%d Resolved=%t) ", n.ID, n.Dev, n.VlanID, ip, n.Type, len(n.FdbRefs), n.Resolved)
-		log.Println(str)
 		s += str
 		s += "\n"
 	}
-	log.Printf("\n\n\n")
 	s += "\n\n"
 	return s
 }

pkg/netlink/mgmtServer.go

@@ -0,0 +1,92 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright (c) 2022-2023 Intel Corporation, or its subsidiaries.
+// Copyright (C) 2023 Nordix Foundation.
+
+// Package netlink handles the netlink related functionality
+package netlink
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"net"
+
+	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/logging"
+	"github.com/opiproject/opi-evpn-bridge/pkg/config"
+	pm "github.com/opiproject/opi-evpn-bridge/pkg/netlink/proto/gen/go"
+	"github.com/opiproject/opi-evpn-bridge/pkg/utils"
+	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/trace"
+	grpc "google.golang.org/grpc"
+	"google.golang.org/grpc/reflection"
+)
+
+// Server represents the Server object
+type Server struct {
+	pm.UnimplementedManagementServiceServer
+	Pagination map[string]int
+	tracer     trace.Tracer
+}
+
+// NewServer creates initialized instance of EVPN Mgmt server
+func NewServer() *Server {
+	return &Server{
+		Pagination: make(map[string]int),
+		tracer:     otel.Tracer(""),
+	}
+}
+
+// RunMgmtGrpcServer start the grpc server for all the components
+func RunMgmtGrpcServer(grpcPort uint16) {
+	if config.GlobalConfig.Tracer {
+		tp := utils.InitTracerProvider("opi-evpn-bridge")
+		defer func() {
+			if err := tp.Shutdown(context.Background()); err != nil {
+				log.Panicf("Tracer Provider Shutdown: %v", err)
+			}
+		}()
+	}
+
+	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", grpcPort))
+	if err != nil {
+		log.Panicf("failed to listen: %v", err)
+	}
+
+	var serverOptions []grpc.ServerOption
+	serverOptions = append(serverOptions,
+		grpc.StatsHandler(otelgrpc.NewServerHandler()),
+		grpc.UnaryInterceptor(
+			logging.UnaryServerInterceptor(utils.InterceptorLogger(log.Default()),
+				logging.WithLogOnEvents(
+					logging.StartCall,
+					logging.FinishCall,
+					logging.PayloadReceived,
+					logging.PayloadSent,
+				),
+			)),
+	)
+	s := grpc.NewServer(serverOptions...)
+
+	ManagementServer := NewServer()
+	pm.RegisterManagementServiceServer(s, ManagementServer)
+
+	reflection.Register(s)
+
+	log.Printf("gRPC server listening at %v", lis.Addr())
+	if err := s.Serve(lis); err != nil {
+		log.Panicf("failed to serve: %v", err)
+	}
+}
+
+// DumpNetlinkDB executes the dumping of netlink database
+func (s *Server) DumpNetlinkDB(_ context.Context, in *pm.DumpNetlinkDbRequest) (*pm.DumpNetlinkDbResult, error) {
+	if in != nil {
+		dump, err := DumpDatabases()
+		if err != nil {
+			return &pm.DumpNetlinkDbResult{}, err
+		}
+		return &pm.DumpNetlinkDbResult{Details: dump}, nil
+	}
+	return &pm.DumpNetlinkDbResult{}, nil
+}

pkg/netlink/neighbor.go

@@ -291,7 +291,6 @@ func (neigh NeighStruct) neighborAnnotate() NeighStruct {
 // dumpNeighDB dump the neighbor entries
 func dumpNeighDB() string {
 	var s string
-	log.Printf("netlink: Dump Neighbor table:\n")
 	s = "Neighbor table:\n"
 	for _, n := range latestNeighbors {
 		var Proto string
@@ -301,7 +300,6 @@ func dumpNeighDB() string {
 			Proto = n.Protocol
 		}
 		str := fmt.Sprintf("Neighbor(vrf=%s dst=%s lladdr=%s dev=%s proto=%s state=%s Type : %d) ", n.VrfName, n.Neigh0.IP.String(), n.Neigh0.HardwareAddr.String(), nameIndex[n.Neigh0.LinkIndex], Proto, getStateStr(n.Neigh0.State), n.Type)
-		log.Println(str)
 		s += str
 		s += "\n"
 	}

pkg/netlink/netlink_watcher.go

@@ -8,6 +8,7 @@ package netlink
 import (
 	"context"
 	"log"
+	sync "sync"
 
 	"time"
 
@@ -18,6 +19,9 @@ import (
 	"github.com/opiproject/opi-evpn-bridge/pkg/utils"
 )
 
+// Define a global mutex
+var mu sync.Mutex
+
 // deleteLatestDB deletes the latest db snap
 func deleteLatestDB() {
 	latestRoutes = make(map[RouteKey]*RouteStruct)
@@ -98,6 +102,14 @@ func annotateDBEntries() {
 	annotateMap(latestL2Nexthop)
 }
 
+// DumpDatabases reads the latest netlink state
+func DumpDatabases() (string, error) {
+	mu.Lock()
+	defer mu.Unlock() // Ensure the mutex is unlocked when the function exits
+	dump, err := dumpDBs()
+	return dump, err
+}
+
 // readLatestNetlinkState reads the latest netlink state
 func readLatestNetlinkState() {
 	grdVrf, err := infradb.GetVrf("//network.opiproject.org/vrfs/GRD")
@@ -116,11 +128,13 @@ func readLatestNetlinkState() {
 			m[i].addFdbEntry()
 		}
 	}
-	dumpDBs()
 }
 
 // resyncWithKernel fun resyncs with kernal db
 func resyncWithKernel() {
+	mu.Lock()
+	defer mu.Unlock() // Ensure the mutex is unlocked when the function exits
+
 	// Build a new DB snapshot from netlink and other sources
 	readLatestNetlinkState()
 	// Annotate the latest DB entries
@@ -133,6 +147,7 @@ func resyncWithKernel() {
 	nexthops = latestNexthop
 	fDB = latestFDB
 	l2Nexthops = latestL2Nexthop
+
 	deleteLatestDB()
 }
 
@@ -188,6 +203,8 @@ func getlink() {
 	}
 }
 
+const grpcPort uint16 = 50152
+
 // Initialize function intializes config
 func Initialize() {
 	pollInterval = config.GlobalConfig.Netlink.PollInterval
@@ -201,6 +218,8 @@ func Initialize() {
 		log.Printf("netlink: netlink_monitor disabled")
 		return
 	}
+
+	go RunMgmtGrpcServer(grpcPort)
 	for i := 0; i < len(config.GlobalConfig.Interfaces.PhyPorts); i++ {
 		phyPorts[config.GlobalConfig.Interfaces.PhyPorts[i].Rep] = config.GlobalConfig.Interfaces.PhyPorts[i].Vsi
 	}

pkg/netlink/nexthop.go

@@ -167,7 +167,7 @@ func NHAssignID(key NexthopKey) int {
 
 // addNexthop adds the nexthop
 //
-//nolint
+// nolint
 func (nexthop *NexthopStruct) addNexthop(r *RouteStruct) *RouteStruct {
 	if len(r.Nexthops) > 0 && !enableEcmp {
 		log.Printf("ECMP disabled: Ignoring additional nexthop of route")
@@ -195,7 +195,7 @@ func (nexthop *NexthopStruct) addNexthop(r *RouteStruct) *RouteStruct {
 
 // ParseNexthop parses the neighbor
 //
-//nolint
+// nolint
 func (nexthop *NexthopStruct) ParseNexthop(v *infradb.Vrf, rc RouteCmdInfo) {
 	var phyFlag bool
 	phyFlag = false
@@ -418,15 +418,12 @@ func (nexthop *NexthopStruct) GetVrfOperStatus() infradb.VrfOperStatus {
 // dumpNexthDB dump the nexthop entries
 func dumpNexthDB() string {
 	var s string
-	log.Printf("netlink: Dump Nexthop table:\n")
 	s = "Nexthop table:\n"
-	for _, n := range latestNexthop {
+	for _, n := range nexthops {
 		str := fmt.Sprintf("Nexthop(id=%d vrf=%s dst=%s dev=%s Local=%t weight=%d flags=[%s] #routes=%d Resolved=%t neighbor=%s) ", n.ID, n.Vrf.Name, n.nexthop.Gw.String(), nameIndex[n.nexthop.LinkIndex], n.Local, n.Weight, getFlagString(n.nexthop.Flags), len(n.RouteRefs), n.Resolved, n.Neighbor.printNeigh())
-		log.Println(str)
 		s += str
 		s += "\n"
 	}
-	log.Printf("\n\n\n")
 	s += "\n\n"
 	return s
 }