pkg/controller/pod/pod_controller.go

   1 package pod
   2
   3 import (
   4         "context"
   5         "encoding/json"
   6         "fmt"
   7         "ovn4nfv-k8s-plugin/internal/pkg/ovn"
   8
   9         corev1 "k8s.io/api/core/v1"
  10         "k8s.io/apimachinery/pkg/api/errors"
  11         "k8s.io/apimachinery/pkg/runtime"
  12         "k8s.io/apimachinery/pkg/types"
  13         "sigs.k8s.io/controller-runtime/pkg/client"
  14         "sigs.k8s.io/controller-runtime/pkg/controller"
  15         "sigs.k8s.io/controller-runtime/pkg/event"
  16         "sigs.k8s.io/controller-runtime/pkg/handler"
  17         "sigs.k8s.io/controller-runtime/pkg/manager"
  18         "sigs.k8s.io/controller-runtime/pkg/predicate"
  19         "sigs.k8s.io/controller-runtime/pkg/reconcile"
  20         logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
  21         "sigs.k8s.io/controller-runtime/pkg/source"
  22 )
  23
  24 var log = logf.Log.WithName("controller_pod")
  25
  26 const (
  27         nfnNetworkAnnotation = "k8s.plugin.opnfv.org/nfn-network"
  28 )
  29
  30 type nfnNetwork struct {
  31         Type      string                   "json:\"type\""
  32         Interface []map[string]interface{} "json:\"interface\""
  33 }
  34
  35 var enableOvnDefaultIntf bool = true
  36
  37 // Add creates a new Pod Controller and adds it to the Manager. The Manager will set fields on the Controller
  38 // and Start it when the Manager is Started.
  39 func Add(mgr manager.Manager) error {
  40         return add(mgr, newReconciler(mgr))
  41 }
  42
  43 // newReconciler returns a new reconcile.Reconciler
  44 func newReconciler(mgr manager.Manager) reconcile.Reconciler {
  45         return &ReconcilePod{client: mgr.GetClient(), scheme: mgr.GetScheme()}
  46 }
  47
  48 // add adds a new Controller to mgr with r as the reconcile.Reconciler
  49 func add(mgr manager.Manager, r reconcile.Reconciler) error {
  50
  51         // Create a new Controller that will call the provided Reconciler function in response
  52         // to events.
  53         c, err := controller.New("pod-controller", mgr, controller.Options{Reconciler: r})
  54         if err != nil {
  55                 return err
  56         }
  57         // Define Predicates On Create and Update function
  58         p := predicate.Funcs{
  59                 UpdateFunc: func(e event.UpdateEvent) bool {
  60                         annotaion := e.MetaNew.GetAnnotations()
  61                         // The object doesn't contain annotation ,nfnNetworkAnnotation so the event will be
  62                         // ignored.
  63                         if _, ok := annotaion[nfnNetworkAnnotation]; !ok {
  64                                 return false
  65                         }
  66                         // If pod is already processed by OVN don't add event
  67                         if _, ok := annotaion[ovn.Ovn4nfvAnnotationTag]; ok {
  68                                 return false
  69                         }
  70                         return true
  71                 },
  72                 CreateFunc: func(e event.CreateEvent) bool {
  73                         // The object doesn't contain annotation ,nfnNetworkAnnotation so the event will be
  74                         // ignored.
  75                         /*annotaion := e.Meta.GetAnnotations()
  76                         if _, ok := annotaion[nfnNetworkAnnotation]; !ok {
  77                                 return false
  78                         }*/
  79                         return true
  80                 },
  81                 DeleteFunc: func(e event.DeleteEvent) bool {
  82                         // The object doesn't contain annotation ,nfnNetworkAnnotation so the event will be
  83                         // ignored.
  84                         /*annotaion := e.Meta.GetAnnotations()
  85                         if _, ok := annotaion[nfnNetworkAnnotation]; !ok {
  86                                 return false
  87                         }*/
  88                         return true
  89                 },
  90         }
  91
  92         // Watch for Pod create / update / delete events and call Reconcile
  93         err = c.Watch(&source.Kind{Type: &corev1.Pod{}}, &handler.EnqueueRequestForObject{}, p)
  94         if err != nil {
  95                 return err
  96         }
  97         return nil
  98 }
  99
 100 // blank assignment to verify that ReconcuilePod implements reconcile.Reconciler
 101 var _ reconcile.Reconciler = &ReconcilePod{}
 102
 103 // ReconcilePod reconciles a ProviderNetwork object
 104 type ReconcilePod struct {
 105         // This client, initialized using mgr.Client() above, is a split client
 106         // that reads objects from the cache and writes to the apiserver
 107         client client.Client
 108         scheme *runtime.Scheme
 109 }
 110
 111 // Reconcile function
 112 // The Controller will requeue the Request to be processed again if the returned error is non-nil or
 113 // Result.Requeue is true, otherwise upon completion it will remove the work from the queue.
 114 func (r *ReconcilePod) Reconcile(request reconcile.Request) (reconcile.Result, error) {
 115         reqLogger := log.WithValues("Request.Namespace", request.Namespace, "Request.Name", request.Name)
 116         reqLogger.Info("Enter Reconciling Pod")
 117
 118         // Fetch the Pod instance
 119         instance := &corev1.Pod{}
 120         err := r.client.Get(context.TODO(), request.NamespacedName, instance)
 121
 122         if err != nil {
 123                 if errors.IsNotFound(err) {
 124                         // Request object not found, could have been deleted after reconcile request.
 125                         // Owned objects are automatically garbage collected. For additional cleanup logic use finalizers.
 126                         // Return and don't requeue
 127                         reqLogger.Info("Delete Pod", "request", request)
 128                         r.deleteLogicalPorts(request.Name, request.Namespace)
 129                         reqLogger.Info("Exit Reconciling Pod")
 130                         return reconcile.Result{}, nil
 131                 }
 132                 // Error reading the object - requeue the request.
 133                 return reconcile.Result{}, err
 134         }
 135         if instance.Name == "" || instance.Namespace == "" {
 136                 return reconcile.Result{}, nil
 137         }
 138         if instance.Spec.HostNetwork {
 139                 return reconcile.Result{}, nil
 140         }
 141
 142         if instance.Spec.NodeName == "" {
 143                 return reconcile.Result{
 144                         Requeue: true,
 145                 }, nil
 146         }
 147
 148         err = r.addLogicalPorts(instance)
 149         if err != nil && err.Error() == "Failed to add ports" {
 150                 // Requeue the object
 151                 return reconcile.Result{}, err
 152         }
 153         reqLogger.Info("Exit Reconciling Pod")
 154         return reconcile.Result{}, nil
 155 }
 156
 157 // annotatePod annotates pod with the given annotations
 158 func (r *ReconcilePod) setPodAnnotation(pod *corev1.Pod, key, value string) error {
 159
 160         patchData := fmt.Sprintf(`{"metadata":{"annotations":{"%s":"%s"}}}`, key, value)
 161         err := r.client.Patch(context.TODO(), pod, client.ConstantPatch(types.MergePatchType, []byte(patchData)))
 162         if err != nil {
 163                 log.Error(err, "Updating pod failed", "pod", pod, "key", key, "value", value)
 164                 return err
 165         }
 166         return nil
 167 }
 168
 169 func (r *ReconcilePod) addLogicalPorts(pod *corev1.Pod) error {
 170
 171         nfn, err := r.readPodAnnotation(pod)
 172         if err != nil {
 173                 // No annotation for multiple interfaces
 174                 nfn = &nfnNetwork{Interface: nil}
 175                 if enableOvnDefaultIntf == true {
 176                         nfn.Type = "ovn4nfv"
 177                 } else {
 178                         return err
 179                 }
 180         }
 181
 182         switch {
 183         case nfn.Type == "ovn4nfv":
 184                 ovnCtl, err := ovn.GetOvnController()
 185                 if err != nil {
 186                         return err
 187                 }
 188                 if _, ok := pod.Annotations[ovn.Ovn4nfvAnnotationTag]; ok {
 189                         return fmt.Errorf("Pod annotation found")
 190                 }
 191                 key, value := ovnCtl.AddLogicalPorts(pod, nfn.Interface)
 192                 if len(key) > 0 {
 193                         return r.setPodAnnotation(pod, key, value)
 194                 }
 195                 return fmt.Errorf("Failed to add ports")
 196         default:
 197                 return fmt.Errorf("Unsupported Networking type %s", nfn.Type)
 198                 // Add other types here
 199         }
 200 }
 201
 202 func (r *ReconcilePod) deleteLogicalPorts(name, namesapce string) error {
 203
 204         // Run delete for all controllers; pod annonations inaccessible
 205         ovnCtl, err := ovn.GetOvnController()
 206         if err != nil {
 207                 return err
 208         }
 209         log.Info("Calling DeleteLogicalPorts")
 210         ovnCtl.DeleteLogicalPorts(name, namesapce)
 211         return nil
 212         // Add other types here
 213 }
 214
 215 func (r *ReconcilePod) readPodAnnotation(pod *corev1.Pod) (*nfnNetwork, error) {
 216         annotaion, ok := pod.Annotations[nfnNetworkAnnotation]
 217         if !ok {
 218                 return nil, fmt.Errorf("Invalid annotations")
 219         }
 220         var nfn nfnNetwork
 221         err := json.Unmarshal([]byte(annotaion), &nfn)
 222         if err != nil {
 223                 log.Error(err, "Invalid nfn annotaion", "annotaiton", annotaion)
 224                 return nil, err
 225         }
 226         return &nfn, nil
 227 }