internal/pkg/nfnNotify/server.go

   1 package nfn
   2
   3 import (
   4         "fmt"
   5         "net"
   6         pb "ovn4nfv-k8s-plugin/internal/pkg/nfnNotify/proto"
   7         "ovn4nfv-k8s-plugin/internal/pkg/node"
   8         v1alpha1 "ovn4nfv-k8s-plugin/pkg/apis/k8s/v1alpha1"
   9         clientset "ovn4nfv-k8s-plugin/pkg/generated/clientset/versioned"
  10         "strings"
  11
  12         "google.golang.org/grpc"
  13         "google.golang.org/grpc/reflection"
  14         v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  15         "k8s.io/client-go/kubernetes"
  16         "k8s.io/client-go/rest"
  17         logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
  18 )
  19
  20 var log = logf.Log.WithName("rpc-server")
  21
  22 type client struct {
  23         context *pb.SubscribeContext
  24         stream  pb.NfnNotify_SubscribeServer
  25 }
  26
  27 type serverDB struct {
  28         name       string
  29         clientList map[string]client
  30 }
  31
  32 var notifServer *serverDB
  33 var stopChan chan interface{}
  34
  35 var pnClientset *clientset.Clientset
  36 var kubeClientset *kubernetes.Clientset
  37
  38 func newServer() *serverDB {
  39         return &serverDB{name: "nfnNotifServer", clientList: make(map[string]client)}
  40 }
  41
  42 // Subscribe stores the client information & sends data
  43 func (s *serverDB) Subscribe(sc *pb.SubscribeContext, ss pb.NfnNotify_SubscribeServer) error {
  44         nodeName := sc.GetNodeName()
  45         log.Info("Subscribe request from node", "Node Name", nodeName)
  46         if nodeName == "" {
  47                 return fmt.Errorf("Node name can't be empty")
  48         }
  49
  50         nodeIntfIPAddr, nodeIntfMacAddr, err := node.AddNodeLogicalPorts(nodeName)
  51         if err != nil {
  52                 return fmt.Errorf("Error in creating node logical port for node- %s: %v", nodeName, err)
  53         }
  54         cp := client{
  55                 context: sc,
  56                 stream:  ss,
  57         }
  58         s.clientList[nodeName] = cp
  59
  60         providerNetworklist, err := pnClientset.K8sV1alpha1().ProviderNetworks("default").List(v1.ListOptions{})
  61         if err == nil {
  62                 for _, pn := range providerNetworklist.Items {
  63                         log.Info("Send message", "Provider Network", pn.GetName())
  64                         SendNotif(&pn, "create", nodeName)
  65                 }
  66         }
  67         inSyncMsg := pb.Notification{
  68                 CniType: "ovn4nfv",
  69                 Payload: &pb.Notification_InSync{
  70                         InSync: &pb.InSync{
  71                                 NodeIntfIpAddress:  nodeIntfIPAddr,
  72                                 NodeIntfMacAddress: nodeIntfMacAddr,
  73                         },
  74                 },
  75         }
  76         log.Info("Send Insync")
  77         if err = cp.stream.Send(&inSyncMsg); err != nil {
  78                 log.Error(err, "Unable to send sync", "node name", nodeName)
  79         }
  80         log.Info("Subscribe Completed")
  81         // Keep stream open
  82         for {
  83                 select {
  84                 case <-stopChan:
  85                 }
  86         }
  87 }
  88
  89 func (s *serverDB) GetClient(nodeName string) client {
  90         if val, ok := s.clientList[nodeName]; ok {
  91                 return val
  92         }
  93         return client{}
  94 }
  95
  96 func updatePnStatus(pn *v1alpha1.ProviderNetwork, status string) error {
  97         pnCopy := pn.DeepCopy()
  98         pnCopy.Status.State = status
  99         _, err := pnClientset.K8sV1alpha1().ProviderNetworks(pn.Namespace).Update(pnCopy)
 100         return err
 101 }
 102
 103 func createVlanMsg(pn *v1alpha1.ProviderNetwork) pb.Notification {
 104         msg := pb.Notification{
 105                 CniType: "ovn4nfv",
 106                 Payload: &pb.Notification_ProviderNwCreate{
 107                         ProviderNwCreate: &pb.ProviderNetworkCreate{
 108                                 ProviderNwName: pn.Name,
 109                                 Vlan: &pb.VlanInfo{
 110                                         VlanId:       pn.Spec.Vlan.VlanId,
 111                                         ProviderIntf: pn.Spec.Vlan.ProviderInterfaceName,
 112                                         LogicalIntf:  pn.Spec.Vlan.LogicalInterfaceName,
 113                                 },
 114                         },
 115                 },
 116         }
 117         return msg
 118 }
 119
 120 func deleteVlanMsg(pn *v1alpha1.ProviderNetwork) pb.Notification {
 121         msg := pb.Notification{
 122                 CniType: "ovn4nfv",
 123                 Payload: &pb.Notification_ProviderNwRemove{
 124                         ProviderNwRemove: &pb.ProviderNetworkRemove{
 125                                 ProviderNwName:  pn.Name,
 126                                 VlanLogicalIntf: pn.Spec.Vlan.LogicalInterfaceName,
 127                         },
 128                 },
 129         }
 130         return msg
 131 }
 132
 133 func createDirectMsg(pn *v1alpha1.ProviderNetwork) pb.Notification {
 134         msg := pb.Notification{
 135                 CniType: "ovn4nfv",
 136                 Payload: &pb.Notification_ProviderNwCreate{
 137                         ProviderNwCreate: &pb.ProviderNetworkCreate{
 138                                 ProviderNwName: pn.Name,
 139                                 Direct: &pb.DirectInfo{
 140                                         ProviderIntf: pn.Spec.Direct.ProviderInterfaceName,
 141                                 },
 142                         },
 143                 },
 144         }
 145         return msg
 146 }
 147
 148 func deleteDirectMsg(pn *v1alpha1.ProviderNetwork) pb.Notification {
 149         msg := pb.Notification{
 150                 CniType: "ovn4nfv",
 151                 Payload: &pb.Notification_ProviderNwRemove{
 152                         ProviderNwRemove: &pb.ProviderNetworkRemove{
 153                                 ProviderNwName:     pn.Name,
 154                                 DirectProviderIntf: pn.Spec.Direct.ProviderInterfaceName,
 155                         },
 156                 },
 157         }
 158         return msg
 159 }
 160
 161 //SendNotif to client
 162 func SendNotif(pn *v1alpha1.ProviderNetwork, msgType string, nodeReq string) error {
 163         var msg pb.Notification
 164         var err error
 165
 166         switch {
 167         case pn.Spec.CniType == "ovn4nfv":
 168                 switch {
 169                 case pn.Spec.ProviderNetType == "VLAN":
 170                         if msgType == "create" {
 171                                 msg = createVlanMsg(pn)
 172                         } else if msgType == "delete" {
 173                                 msg = deleteVlanMsg(pn)
 174                         }
 175                         if strings.EqualFold(pn.Spec.Vlan.VlanNodeSelector, "SPECIFIC") {
 176                                 for _, label := range pn.Spec.Vlan.NodeLabelList {
 177                                         l := strings.Split(label, "=")
 178                                         if len(l) == 0 {
 179                                                 log.Error(fmt.Errorf("Syntax error label: %v", label), "NodeListIterator")
 180                                                 return nil
 181                                         }
 182                                 }
 183                                 labels := strings.Join(pn.Spec.Vlan.NodeLabelList[:], ",")
 184                                 err = sendMsg(msg, labels, "specific", nodeReq)
 185                         } else if strings.EqualFold(pn.Spec.Vlan.VlanNodeSelector, "ALL") {
 186                                 err = sendMsg(msg, "", "all", nodeReq)
 187                         } else if strings.EqualFold(pn.Spec.Vlan.VlanNodeSelector, "ANY") {
 188                                 if pn.Status.State != v1alpha1.Created {
 189                                         err = sendMsg(msg, "", "any", nodeReq)
 190                                         if err == nil {
 191                                                 updatePnStatus(pn, v1alpha1.Created)
 192                                         }
 193                                 }
 194                         }
 195                 case pn.Spec.ProviderNetType == "DIRECT":
 196                         if msgType == "create" {
 197                                 msg = createDirectMsg(pn)
 198                         } else if msgType == "delete" {
 199                                 msg = deleteDirectMsg(pn)
 200                         }
 201                         if strings.EqualFold(pn.Spec.Direct.DirectNodeSelector, "SPECIFIC") {
 202                                 for _, label := range pn.Spec.Direct.NodeLabelList {
 203                                         l := strings.Split(label, "=")
 204                                         if len(l) == 0 {
 205                                                 log.Error(fmt.Errorf("Syntax error label: %v", label), "NodeListIterator")
 206                                                 return nil
 207                                         }
 208                                 }
 209                                 labels := strings.Join(pn.Spec.Direct.NodeLabelList[:], ",")
 210                                 err = sendMsg(msg, labels, "specific", nodeReq)
 211                         } else if strings.EqualFold(pn.Spec.Direct.DirectNodeSelector, "ALL") {
 212                                 err = sendMsg(msg, "", "all", nodeReq)
 213                         } else if strings.EqualFold(pn.Spec.Direct.DirectNodeSelector, "ANY") {
 214                                 if pn.Status.State != v1alpha1.Created {
 215                                         err = sendMsg(msg, "", "any", nodeReq)
 216                                         if err == nil {
 217                                                 updatePnStatus(pn, v1alpha1.Created)
 218                                         }
 219                                 }
 220                         }
 221                 default:
 222                         return fmt.Errorf("Unsupported Provider Network type")
 223                 }
 224         default:
 225                 return fmt.Errorf("Unsupported CNI type")
 226         }
 227         return err
 228 }
 229
 230 // sendMsg send notification to client
 231 func sendMsg(msg pb.Notification, labels string, option string, nodeReq string) error {
 232         if option == "all" {
 233                 for name, client := range notifServer.clientList {
 234                         if nodeReq != "" && nodeReq != name {
 235                                 continue
 236                         }
 237                         if client.stream != nil {
 238                                 if err := client.stream.Send(&msg); err != nil {
 239                                         log.Error(err, "Msg Send failed", "Node name", name)
 240                                 }
 241                         }
 242                 }
 243                 return nil
 244         } else if option == "any" {
 245                 // Always select the first
 246                 for _, client := range notifServer.clientList {
 247                         if client.stream != nil {
 248                                 if err := client.stream.Send(&msg); err != nil {
 249                                         return err
 250                                 }
 251                                 // return after first successful send
 252                                 return nil
 253                         }
 254                 }
 255                 return nil
 256         }
 257         // This is specific case
 258         for name := range nodeListIterator(labels) {
 259                 if nodeReq != "" && nodeReq != name {
 260                         continue
 261                 }
 262                 client := notifServer.GetClient(name)
 263                 if client.stream != nil {
 264                         if err := client.stream.Send(&msg); err != nil {
 265                                 return err
 266                         }
 267                 }
 268         }
 269         return nil
 270 }
 271
 272 func nodeListIterator(labels string) <-chan string {
 273         ch := make(chan string)
 274
 275         lo := v1.ListOptions{LabelSelector: labels}
 276         // List the Nodes matching the Labels
 277         nodes, err := kubeClientset.CoreV1().Nodes().List(lo)
 278         if err != nil {
 279                 log.Info("No Nodes found with labels", "list:", lo)
 280                 return nil
 281         }
 282         go func() {
 283                 for _, node := range nodes.Items {
 284                         log.Info("Send message to", " node:", node.ObjectMeta.Name)
 285                         ch <- node.ObjectMeta.Name
 286                 }
 287                 close(ch)
 288         }()
 289         return ch
 290 }
 291
 292 //SetupNotifServer initilizes the gRpc nfn notif server
 293 func SetupNotifServer(kConfig *rest.Config) {
 294
 295         log.Info("Starting Notif Server")
 296         var err error
 297
 298         // creates the clientset
 299         pnClientset, err = clientset.NewForConfig(kConfig)
 300         if err != nil {
 301                 log.Error(err, "Error building clientset")
 302         }
 303         kubeClientset, err = kubernetes.NewForConfig(kConfig)
 304         if err != nil {
 305                 log.Error(err, "Error building Kuberenetes clientset")
 306         }
 307
 308         stopChan = make(chan interface{})
 309
 310         // Start GRPC server
 311         lis, err := net.Listen("tcp", ":50000")
 312         if err != nil {
 313                 log.Error(err, "failed to listen")
 314         }
 315
 316         s := grpc.NewServer()
 317         // Intialize Notify server
 318         notifServer = newServer()
 319         pb.RegisterNfnNotifyServer(s, notifServer)
 320
 321         reflection.Register(s)
 322         log.Info("Initialization Completed")
 323         if err := s.Serve(lis); err != nil {
 324                 log.Error(err, "failed to serve")
 325         }
 326 }