pkg/controller/pod/pod_controller.go

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