8852

Committed 21 Mar 2024 03:12PM UTC coverage: 70.881% (-0.4%) from 71.311%

Build # 8852

Build Type

Pull #1287

travis-pro

Committed by

akhilamohanan

Commit Message

Do VLAN programming upfront for OpenShift on OpenStack

For OpenShift on OpenStack clusters, vnsRsCIfPathAtt is created for each
OpenStack compute hosts along with OpenShift nodes so that VLAN will
already be there when a vm is migrated from one compute host to other.

The change is done to avoid datapath issues after vm migration.

Pull Request Pull Request #1287: Do VLAN programming upfront for OpenShift on OpenStack

Run Details

27 of 120 new or added lines in 2 files covered. (22.5%)

175 existing lines in 4 files now uncovered.

10703 of 15100 relevant lines covered (70.88%)

0.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.86

/pkg/hostagent/pods.go

// Copyright 2016 Cisco Systems, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

// Handlers for pod updates.  Pods map to opflex endpoints

package hostagent

import (
        "context"
        "encoding/json"
        "fmt"
        "os"
        "path/filepath"
        "reflect"
        "strings"
        "time"

        "github.com/sirupsen/logrus"

        aciv1 "github.com/noironetworks/aci-containers/pkg/gbpcrd/apis/acipolicy/v1"
        nodepodif "github.com/noironetworks/aci-containers/pkg/nodepodif/apis/acipolicy/v1"
        v1 "k8s.io/api/core/v1"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/fields"
        "k8s.io/apimachinery/pkg/labels"
        "k8s.io/apimachinery/pkg/runtime"
        "k8s.io/apimachinery/pkg/watch"
        "k8s.io/client-go/kubernetes"
        "k8s.io/client-go/tools/cache"

        "k8s.io/kubernetes/pkg/controller"

        "github.com/google/uuid"
        "github.com/noironetworks/aci-containers/pkg/metadata"
        cnimeta "github.com/noironetworks/aci-containers/pkg/metadata"
        "github.com/noironetworks/aci-containers/pkg/util"
        apierrors "k8s.io/apimachinery/pkg/api/errors"
)

const NullMac = "null-mac"

type epTimeStamps struct {
        podCreatationNoticationTimestamp   time.Time
        podUpdateNotificationsTimestamps   []time.Time
        epFileUpdateTimestamps             map[time.Time]string
        epFileCreateTimestamp              time.Time
        createNotifyToLastEpFileUpdateDiff time.Duration
        lastNotifyToLastEpFileUpdateDiff   time.Duration
        podCreationToLastEpFileUpdateDiff  time.Duration
        podCreationTimestamp               time.Time
        podStartTimestamp                  time.Time
}

type opflexEndpoint struct {
        Uuid string `json:"uuid"`

        EgPolicySpace string                 `json:"eg-policy-space,omitempty"`
        EndpointGroup string                 `json:"endpoint-group-name,omitempty"`
        SecurityGroup []metadata.OpflexGroup `json:"security-group,omitempty"`
        QosPolicy     metadata.OpflexGroup   `json:"qos-policy,omitempty"`

        IpAddress  []string `json:"ip,omitempty"`
        MacAddress string   `json:"mac,omitempty"`

        AccessIface       string `json:"access-interface,omitempty"`
        AccessUplinkIface string `json:"access-uplink-interface,omitempty"`
        IfaceName         string `json:"interface-name,omitempty"`

        Attributes        map[string]string `json:"attributes,omitempty"`
        SnatUuid          []string          `json:"snat-uuids,omitempty"`
        ServiceClusterIps []string          `json:"service-ip,omitempty"`
        registered        bool
}

func (agent *HostAgent) getPodIFName(ns, podName string) string {
        return fmt.Sprintf("%s.%s.%s", ns, podName, agent.vtepIP)
}

func (agent *HostAgent) EPRegAdd(ep *opflexEndpoint) bool {
        if agent.crdClient == nil {
                ep.registered = true
                return false // crd not used
        }

        // force the mask to /32
        ipRemEP := strings.Split(ep.IpAddress[0], "/")[0] + "/32"
        remEP := &aciv1.PodIF{
                Status: aciv1.PodIFStatus{
                        PodNS:       ep.Attributes["namespace"],
                        PodName:     ep.Attributes["vm-name"],
                        ContainerID: ep.Uuid,
                        MacAddr:     ep.MacAddress,
                        IPAddr:      ipRemEP,
                        EPG:         ep.EndpointGroup,
                        VTEP:        agent.vtepIP,
                        IFName:      ep.IfaceName,
                },
        }
        remEP.ObjectMeta.Name = agent.getPodIFName(ep.Attributes["namespace"], ep.Attributes["vm-name"])

        podif, err := agent.crdClient.PodIFs("kube-system").Get(context.TODO(), remEP.ObjectMeta.Name, metav1.GetOptions{})
        if err != nil {
                // create podif
                _, err := agent.crdClient.PodIFs("kube-system").Create(context.TODO(), remEP, metav1.CreateOptions{})
                if err != nil {
                        logrus.Errorf("Create error %v, podif: %+v", err, remEP)
                        return true
                }
        } else {
                // update it
                podif.Status = remEP.Status
                _, err := agent.crdClient.PodIFs("kube-system").Update(context.TODO(), podif, metav1.UpdateOptions{})
                if err != nil {
                        logrus.Errorf("Update error %v, podif: %+v", err, remEP)
                        return true
                }
        }
        ep.registered = true
        opflexEpLogger(agent.log, ep).Info("Updated podif")
        return false
}

func (agent *HostAgent) EPRegDelEP(name string) {
        if agent.crdClient == nil {
                return // crd not used
        }
        err := agent.crdClient.PodIFs("kube-system").Delete(context.TODO(), name, metav1.DeleteOptions{})
        if err != nil {
                agent.log.Errorf("Error %v, podif: %s", err, name)
                return
        }
        agent.log.Infof("podif: %s deleted", name)
}

func (agent *HostAgent) getNodePodIFName(nodeName string) string {
        return fmt.Sprintf("%s.%s", nodeName, agent.vtepIP)
}

func (agent *HostAgent) NodeEPRegAdd(nodePodIfEPs map[string]*opflexEndpoint) bool {
        if agent.nodePodIFClient == nil {
                agent.log.Debug("NodePodIF client or Kube clients are not intialized")
                return false // crd not used
        }

        var podifs []nodepodif.PodIF
        for _, ep := range nodePodIfEPs {
                ipRemEP := strings.Split(ep.IpAddress[0], "/")[0] + "/32"
                opflexEpLogger(agent.log, ep).Debug("ipRemEP")
                var podif nodepodif.PodIF
                podif.PodNS = ep.Attributes["namespace"]
                podif.PodName = ep.Attributes["vm-name"]
                podif.ContainerID = ep.Uuid
                podif.MacAddr = ep.MacAddress
                podif.IPAddr = ipRemEP
                // could change
                podif.EPG = ep.EndpointGroup
                podif.VTEP = agent.vtepIP
                podif.IFName = ep.IfaceName
                podifs = append(podifs, podif)
        }

        nodePodif, err := agent.nodePodIFClient.NodePodIFs("kube-system").Get(context.TODO(), agent.getNodePodIFName(agent.config.NodeName), metav1.GetOptions{})
        if err != nil {
                // create nodepodif
                if apierrors.IsNotFound(err) {
                        remEP := &nodepodif.NodePodIF{
                                ObjectMeta: metav1.ObjectMeta{
                                        Name:      agent.getNodePodIFName(agent.config.NodeName),
                                        Namespace: "kube-system",
                                },
                                Spec: nodepodif.NodePodIFSpec{
                                        PodIFs: podifs,
                                },
                        }

                        _, err := agent.nodePodIFClient.NodePodIFs("kube-system").Create(context.TODO(), remEP, metav1.CreateOptions{})
                        agent.log.Debugf("nodepodif: %s created for node", remEP.ObjectMeta.Name)
                        if err != nil {
                                logrus.Errorf("Create error %v, nodepodif: %+v", err, remEP)
                                return true
                        }
                }
        } else {
                // update nodepodif
                if !reflect.DeepEqual(nodePodif.Spec.PodIFs, podifs) {
                        nodePodif.Spec.PodIFs = podifs
                        _, err := agent.nodePodIFClient.NodePodIFs("kube-system").Update(context.TODO(), nodePodif, metav1.UpdateOptions{})
                        if err != nil {
                                logrus.Errorf("Update error %v, nodepodif: %+v", err, nodePodif)
                                return true
                        }
                }
        }
        if err == nil {
                agent.log.Debugf("nodepodif: %s updated for node", nodePodif.ObjectMeta.Name)
                return true
        }
        return false
}

func (agent *HostAgent) initPodInformerFromClient(
        kubeClient *kubernetes.Clientset) {
        agent.initPodInformerBase(
                &cache.ListWatch{
                        ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
                                options.FieldSelector =
                                        fields.Set{"spec.nodeName": agent.config.NodeName}.String()
                                obj, err := kubeClient.CoreV1().Pods(metav1.NamespaceAll).List(context.TODO(), options)
                                if err != nil {
                                        agent.log.Fatal("Failed to list Pods during initialization of PodInformer")
                                }
                                return obj, err
                        },
                        WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
                                options.FieldSelector =
                                        fields.Set{"spec.nodeName": agent.config.NodeName}.String()
                                obj, err := kubeClient.CoreV1().Pods(metav1.NamespaceAll).Watch(context.TODO(), options)
                                if err != nil {
                                        agent.log.Fatal("Failed to watch Pods during initialization of PodInformer")
                                }
                                return obj, err
                        },
                })

        agent.initControllerInformerBase(
                &cache.ListWatch{
                        ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
                                options.LabelSelector = labels.Set{"name": "aci-containers-controller"}.String()
                                obj, err := kubeClient.CoreV1().Pods(metav1.NamespaceAll).List(context.TODO(), options)
                                if err != nil {
                                        agent.log.Fatal("Failed to list Pods during initialization of ControllerInformer")
                                }
                                return obj, err
                        },
                        WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
                                options.LabelSelector = labels.Set{"name": "aci-containers-controller"}.String()
                                obj, err := kubeClient.CoreV1().Pods(metav1.NamespaceAll).Watch(context.TODO(), options)
                                if err != nil {
                                        agent.log.Fatal("Failed to watch Pods during initialization of ControllerInformer")
                                }
                                return obj, err
                        },
                })
}

func (agent *HostAgent) initPodInformerBase(listWatch *cache.ListWatch) {
        agent.podInformer = cache.NewSharedIndexInformer(
                listWatch,
                &v1.Pod{},
                controller.NoResyncPeriodFunc(),
                cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
        )
        agent.podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
                AddFunc: func(obj interface{}) {
                        agent.initPodNotifyTimeStamp(obj.(*v1.Pod))
                        agent.podUpdated(obj)
                },
                UpdateFunc: func(_ interface{}, obj interface{}) {
                        agent.updatePodNotifyTimeStamp(obj.(*v1.Pod))
                        agent.podUpdated(obj)
                },
                DeleteFunc: func(obj interface{}) {
                        agent.podDeleted(obj)
                },
        })
}

func (agent *HostAgent) initControllerInformerBase(listWatch *cache.ListWatch) {
        agent.controllerInformer = cache.NewSharedIndexInformer(
                listWatch,
                &v1.Pod{},
                controller.NoResyncPeriodFunc(),
                cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
        )
        agent.controllerInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
                AddFunc: func(obj interface{}) {
                        agent.updateGbpServerInfo(obj.(*v1.Pod))
                },
                UpdateFunc: func(_ interface{}, obj interface{}) {
                        agent.updateGbpServerInfo(obj.(*v1.Pod))
                },
                DeleteFunc: func(obj interface{}) {
                },
        })
}

func getEp(epfile string) (string, error) {
        raw, err := os.ReadFile(epfile)
        if err != nil {
                return "", err
        }
        return string(raw), err
}

func readEp(epfile string) (*opflexEndpoint, error) {
        epStr, err := getEp(epfile)
        if err != nil {
                return nil, err
        }

        ep := &opflexEndpoint{}
        err = json.Unmarshal([]byte(epStr), ep)
        return ep, err
}

func touchFileIfExists(epfile string) (bool, error) {
        var touch bool
        _, err := os.Stat(epfile)
        if err == nil {
                currentTime := time.Now().Local()
                er := os.Chtimes(epfile, currentTime, currentTime)
                if er != nil {
                        return touch, er
                }
                touch = true
        } else if !os.IsNotExist(err) {
                return touch, err
        }
        return touch, nil
}

func writeEp(epfile string, ep *opflexEndpoint) (bool, error) {
        newdata, err := json.MarshalIndent(ep, "", "  ")
        if err != nil {
                return true, err
        }
        existingdata, err := os.ReadFile(epfile)
        if err == nil && reflect.DeepEqual(existingdata, newdata) {
                return false, nil
        }
        err = os.WriteFile(epfile, newdata, 0644)
        return true, err
}

func podLogger(log *logrus.Logger, pod *v1.Pod) *logrus.Entry {
        return log.WithFields(logrus.Fields{
                "namespace": pod.ObjectMeta.Namespace,
                "name":      pod.ObjectMeta.Name,
                "node":      pod.Spec.NodeName,
        })
}

func opflexEpLogger(log *logrus.Logger, ep *opflexEndpoint) *logrus.Entry {
        return log.WithFields(logrus.Fields{
                "Uuid":      ep.Uuid,
                "name":      ep.Attributes["vm-name"],
                "namespace": ep.Attributes["namespace"],
        })
}

func (agent *HostAgent) FormEPFilePath(uuid string) string {
        return filepath.Join(agent.config.OpFlexEndpointDir, uuid+".ep")
}

func (agent *HostAgent) syncOpflexServer() bool {
        grpcAddr := fmt.Sprintf("%s:%d", agent.gbpServerIP, agent.config.GRPCPort)
        srvCfg := &OpflexServerConfig{GRPCAddress: grpcAddr}

        err := os.MkdirAll(filepath.Dir(agent.config.OpFlexServerConfigFile), os.ModeDir|0664)
        if err != nil {
                agent.log.Errorf("Failed to create directory: %s", filepath.Dir(agent.config.OpFlexServerConfigFile))
        }

        data, _ := json.MarshalIndent(srvCfg, "", "  ")
        err = os.WriteFile(agent.config.OpFlexServerConfigFile, data, 0644)
        if err != nil {
                agent.log.Errorf("Failed to create file: %s", agent.config.OpFlexServerConfigFile)
        } else {
                agent.log.Infof("Updated grpc addr to %s", grpcAddr)
        }

        return false
}

func (agent *HostAgent) syncEps() bool {
        if !agent.syncEnabled {
                return false
        }

        agent.log.Debug("Syncing endpoints")
        agent.indexMutex.Lock()
        opflexEps := make(map[string][]*opflexEndpoint)
        for k, v := range agent.opflexEps {
                ep := []*opflexEndpoint{}
                for _, op := range v {
                        val := &opflexEndpoint{}
                        err := util.DeepCopyObj(op, val)
                        if err != nil {
                                continue
                        }
                        ep = append(ep, val)
                }
                opflexEps[k] = ep
        }
        agent.indexMutex.Unlock()

        files, err := os.ReadDir(agent.config.OpFlexEndpointDir)
        if err != nil {
                agent.log.WithFields(
                        logrus.Fields{"endpointDir": agent.config.OpFlexEndpointDir},
                ).Error("Could not read directory ", err)
                return true
        }

        needRetry := false
        seen := make(map[string]bool)
        nullMacFile := false
        nullMacCheck := agent.getEpFileName(agent.config.DefaultEg.Name)
        for _, f := range files {
                if !strings.HasSuffix(f.Name(), ".ep") ||
                        strings.Contains(f.Name(), "veth_host_ac") {
                        continue
                }

                if f.Name() == nullMacCheck {
                        nullMacFile = true
                        continue
                }

                epfile := filepath.Join(agent.config.OpFlexEndpointDir, f.Name())
                epidstr := f.Name()
                epidstr = epidstr[:len(epidstr)-3]
                epid := strings.Split(epidstr, "_")

                if len(epid) < 3 {
                        agent.log.Warn("Removing invalid endpoint:", f.Name())
                        os.Remove(epfile)
                        continue
                }
                poduuid := epid[0]
                contid := epid[1]
                contiface := epid[2]

                logger := agent.log.WithFields(
                        logrus.Fields{
                                "PodUuid":   poduuid,
                                "ContId":    contid,
                                "ContIFace": contiface,
                        },
                )

                existing, ok := opflexEps[poduuid]
                if ok {
                        for _, ep := range existing {
                                if ep.Uuid != epidstr {
                                        continue
                                }
                                ep.SnatUuid, err = agent.getSnatUuids(poduuid, epfile)
                                if err != nil {
                                        agent.log.Error("Error while getting snat uuids")
                                        needRetry = true
                                        continue
                                }
                                ep.ServiceClusterIps = agent.getServiceIPs(poduuid)

                                existingEp, err := readEp(epfile)
                                if err != nil {
                                        agent.log.Error("Error while reading epfile: ", epidstr)
                                        continue
                                }
                                changed_fields := compareOpflexEndpoints(existingEp, ep)
                                if changed_fields != "" {
                                        update_ts := time.Now().UTC()
                                        agent.indexMutex.Lock()
                                        agent.podNameToTimeStamps[poduuid].epFileUpdateTimestamps[update_ts] = changed_fields
                                        agent.podNameToTimeStamps[poduuid].createNotifyToLastEpFileUpdateDiff = update_ts.Sub(agent.podNameToTimeStamps[poduuid].podCreatationNoticationTimestamp)
                                        agent.podNameToTimeStamps[poduuid].podCreationToLastEpFileUpdateDiff = update_ts.Sub(agent.podNameToTimeStamps[poduuid].podCreationTimestamp)
                                        if len(agent.podNameToTimeStamps[poduuid].podUpdateNotificationsTimestamps) > 0 {
                                                agent.podNameToTimeStamps[poduuid].lastNotifyToLastEpFileUpdateDiff = update_ts.Sub(agent.podNameToTimeStamps[poduuid].podUpdateNotificationsTimestamps[len(agent.podNameToTimeStamps[poduuid].podUpdateNotificationsTimestamps)-1])
                                        } else {
                                                agent.podNameToTimeStamps[poduuid].lastNotifyToLastEpFileUpdateDiff = update_ts.Sub(agent.podNameToTimeStamps[poduuid].podCreatationNoticationTimestamp)
                                        }
                                        agent.indexMutex.Unlock()
                                }

                                agent.epfileMutex.Lock()
                                wrote, err := writeEp(epfile, ep)
                                agent.epfileMutex.Unlock()
                                if err != nil {
                                        opflexEpLogger(agent.log, ep).
                                                Error("Error writing EP file: ", err)
                                } else if wrote || !ep.registered {
                                        needRetry = agent.EPRegAdd(ep)
                                }

                                if wrote && err == nil {
                                        agent.indexMutex.Lock()
                                        agent.log.Debugf("ep file updated for pod: %s, Timestamps: {podCreatationNoticationTimestamp=%s, "+
                                                "podUpdateNotificationsTimestamps=[%s], epFileUpdateTimestamps=[%s], epFileCreateTimestamp=%s, "+
                                                "createNotifyToLastEpFileUpdateDiff=%s, lastNotifyToLastEpFileUpdateDiff=%s, "+
                                                "podCreationToLastEpFileUpdateDiff=%s, podCreationTimestamp=%s, "+
                                                "podStartTimestamp=%s}", ep.Attributes["vm-name"],
                                                agent.podNameToTimeStamps[poduuid].podCreatationNoticationTimestamp.Format("2006-01-02T15:04:05Z"),
                                                formatTimestamps(agent.podNameToTimeStamps[poduuid].podUpdateNotificationsTimestamps),
                                                formatMapTimestamps(agent.podNameToTimeStamps[poduuid].epFileUpdateTimestamps),
                                                agent.podNameToTimeStamps[poduuid].epFileCreateTimestamp.Format("2006-01-02T15:04:05Z"),
                                                normalizeToMinutes(agent.podNameToTimeStamps[poduuid].createNotifyToLastEpFileUpdateDiff),
                                                normalizeToMinutes(agent.podNameToTimeStamps[poduuid].lastNotifyToLastEpFileUpdateDiff),
                                                normalizeToMinutes(agent.podNameToTimeStamps[poduuid].podCreationToLastEpFileUpdateDiff),
                                                agent.podNameToTimeStamps[poduuid].podCreationTimestamp.Format("2006-01-02T15:04:05Z"),
                                                agent.podNameToTimeStamps[poduuid].podStartTimestamp.Format("2006-01-02T15:04:05Z"))
                                        agent.indexMutex.Unlock()
                                }

                                if _, ok := agent.nodePodIfEPs[ep.Uuid]; !ok {
                                        agent.nodePodIfEPs[ep.Uuid] = ep
                                }
                                seen[epidstr] = true
                        }
                }

                if !ok || (ok && !seen[epidstr]) {
                        staleEp, err := readEp(epfile)
                        if err == nil {
                                k := agent.getPodIFName(staleEp.Attributes["namespace"], staleEp.Attributes["vm-name"])
                                agent.EPRegDelEP(k)
                                delete(agent.nodePodIfEPs, staleEp.Uuid)
                        }
                        logger.Info("Removing endpoint")
                        os.Remove(epfile)
                }
        }

        for _, eps := range opflexEps {
                for _, ep := range eps {
                        if seen[ep.Uuid] {
                                continue
                        }
                        epfile := agent.FormEPFilePath(ep.Uuid)
                        poduuid := strings.Split(ep.Uuid, "_")[0]
                        ep.SnatUuid, err = agent.getSnatUuids(poduuid, epfile)
                        if err != nil {
                                agent.log.Error("Error while getting snat uuids")
                                needRetry = true
                                continue
                        }
                        ep.ServiceClusterIps = agent.getServiceIPs(poduuid)
                        opflexEpLogger(agent.log, ep).Info("Adding endpoint")
                        agent.indexMutex.Lock()
                        agent.podNameToTimeStamps[poduuid].epFileCreateTimestamp = time.Now().UTC()
                        agent.podNameToTimeStamps[poduuid].createNotifyToLastEpFileUpdateDiff = agent.podNameToTimeStamps[poduuid].epFileCreateTimestamp.Sub(agent.podNameToTimeStamps[poduuid].podCreatationNoticationTimestamp)
                        agent.podNameToTimeStamps[poduuid].lastNotifyToLastEpFileUpdateDiff = agent.podNameToTimeStamps[poduuid].epFileCreateTimestamp.Sub(agent.podNameToTimeStamps[poduuid].podCreatationNoticationTimestamp)
                        agent.podNameToTimeStamps[poduuid].podCreationToLastEpFileUpdateDiff = agent.podNameToTimeStamps[poduuid].epFileCreateTimestamp.Sub(agent.podNameToTimeStamps[poduuid].podCreationTimestamp)
                        agent.indexMutex.Unlock()

                        agent.epfileMutex.Lock()
                        _, err = writeEp(epfile, ep)
                        agent.epfileMutex.Unlock()
                        if err != nil {
                                opflexEpLogger(agent.log, ep).
                                        Error("Error writing EP file: ", err)
                                needRetry = true
                        } else {
                                needRetry = agent.EPRegAdd(ep)
                                agent.nodePodIfEPs[ep.Uuid] = ep
                        }
                        agent.indexMutex.Lock()
                        agent.log.Debugf("ep file Created for pod: %s, Timestamps: {podCreatationNoticationTimestamp=%s, "+
                                "podUpdateNotificationsTimestamps=[%s],  epFileUpdateTimestamps=[%s],  epFileCreateTimestamp=%s, "+
                                "createNotifyToLastEpFileUpdateDiff=%s, lastNotifyToLastEpFileUpdateDiff=%s, "+
                                "podCreationToLastEpFileUpdateDiff=%s, podCreationTimestamp=%s, "+
                                "podStartTimestamp=%s}", ep.Attributes["vm-name"],
                                agent.podNameToTimeStamps[poduuid].podCreatationNoticationTimestamp.Format("2006-01-02T15:04:05Z"),
                                formatTimestamps(agent.podNameToTimeStamps[poduuid].podUpdateNotificationsTimestamps),
                                formatMapTimestamps(agent.podNameToTimeStamps[poduuid].epFileUpdateTimestamps),
                                agent.podNameToTimeStamps[poduuid].epFileCreateTimestamp.Format("2006-01-02T15:04:05Z"),
                                normalizeToMinutes(agent.podNameToTimeStamps[poduuid].createNotifyToLastEpFileUpdateDiff),
                                normalizeToMinutes(agent.podNameToTimeStamps[poduuid].lastNotifyToLastEpFileUpdateDiff),
                                normalizeToMinutes(agent.podNameToTimeStamps[poduuid].podCreationToLastEpFileUpdateDiff),
                                agent.podNameToTimeStamps[poduuid].podCreationTimestamp.Format("2006-01-02T15:04:05Z"),
                                agent.podNameToTimeStamps[poduuid].podStartTimestamp.Format("2006-01-02T15:04:05Z"))
                        agent.indexMutex.Unlock()
                }
        }

        if !nullMacFile {
                agent.creatNullMacEp()
        }
        agent.log.Debug("Finished endpoint sync")
        return needRetry
}

// syncNodePodIfs syncs the NodePodIfs with Eps
func (agent *HostAgent) syncNodePodIfs() bool {
        if !agent.syncEnabled || agent.config.ChainedMode {
                return false
        }
        agent.log.Debug("Syncing NodePodIfs")
        agent.indexMutex.Lock()
        defer agent.indexMutex.Unlock()
        agent.NodeEPRegAdd(agent.nodePodIfEPs)
        agent.log.Debug("Finished NodePodIfs sync")
        return false
}

func (agent *HostAgent) getEpFileName(epGroupName string) string {
        temp := strings.Split(epGroupName, "|")
        var EpFileName string
        if len(temp) == 1 {
                EpFileName = epGroupName + "_" + NullMac + ".ep"
        } else {
                EpFileName = temp[1] + "_" + NullMac + ".ep"
        }
        return EpFileName
}

func (agent *HostAgent) creatNullMacEp() {
        epGroup := agent.config.DefaultEg
        EpFileName := agent.getEpFileName(epGroup.Name)
        EpFilePath := filepath.Join(agent.config.OpFlexEndpointDir, EpFileName)
        ep_file_exists := fileExists(EpFilePath)
        if ep_file_exists {
                return
        }
        ep := &opflexEndpoint{
                Uuid:          uuid.New().String(),
                EgPolicySpace: epGroup.PolicySpace,
                EndpointGroup: epGroup.Name,
                MacAddress:    "00:00:00:00:00:00",
        }
        wrote, err := writeEp(EpFilePath, ep)
        if err != nil {
                agent.log.Debug("Unable to write null mac Ep file")
        } else if wrote {
                agent.log.Debug("Created null mac Ep file")
        }
}

func podFilter(pod *v1.Pod) bool {
        return !(pod.Spec.HostNetwork)
}

func (agent *HostAgent) podUpdated(obj interface{}) {
        agent.log.Infof("Pod updated: namespace=%s name=%s", obj.(*v1.Pod).Namespace, obj.(*v1.Pod).Name)
        agent.indexMutex.Lock()
        defer agent.indexMutex.Unlock()
        agent.depPods.UpdatePodNoCallback(obj.(*v1.Pod))
        agent.rcPods.UpdatePodNoCallback(obj.(*v1.Pod))
        if !agent.config.ChainedMode {
                agent.netPolPods.UpdatePodNoCallback(obj.(*v1.Pod))
                agent.qosPolPods.UpdatePodNoCallback(obj.(*v1.Pod))
        }
        agent.updatePodStausTimeStamps(obj.(*v1.Pod))
        if !agent.config.ChainedMode {
                agent.handleObjectUpdateForSnat(obj)
        }
        agent.podChangedLocked(obj)
}

func (agent *HostAgent) initPodNotifyTimeStamp(pod *v1.Pod) {
        tsMap := &epTimeStamps{
                podCreatationNoticationTimestamp:   time.Time{},
                podUpdateNotificationsTimestamps:   make([]time.Time, 0),
                epFileUpdateTimestamps:             make(map[time.Time]string),
                epFileCreateTimestamp:              time.Time{},
                createNotifyToLastEpFileUpdateDiff: 0,
                lastNotifyToLastEpFileUpdateDiff:   0,
                podCreationToLastEpFileUpdateDiff:  0,
                podCreationTimestamp:               time.Time{},
                podStartTimestamp:                  time.Time{},
        }
        tsMap.podCreatationNoticationTimestamp = time.Now().UTC()
        tsKey := string(pod.ObjectMeta.UID)
        agent.indexMutex.Lock()
        agent.podNameToTimeStamps[tsKey] = tsMap
        agent.indexMutex.Unlock()
}

func (agent *HostAgent) updatePodStausTimeStamps(pod *v1.Pod) {
        tsKey := string(pod.ObjectMeta.UID)
        agent.podNameToTimeStamps[tsKey].podCreationTimestamp = pod.GetCreationTimestamp().UTC()
        if pod.Status.StartTime != nil {
                agent.podNameToTimeStamps[tsKey].podStartTimestamp = pod.Status.StartTime.UTC()
        }
}

func (agent *HostAgent) updatePodNotifyTimeStamp(pod *v1.Pod) {
        tsKey := string(pod.ObjectMeta.UID)
        agent.indexMutex.Lock()
        agent.podNameToTimeStamps[tsKey].podUpdateNotificationsTimestamps = append(agent.podNameToTimeStamps[tsKey].podUpdateNotificationsTimestamps, time.Now().UTC())
        agent.indexMutex.Unlock()
}

func (agent *HostAgent) podChanged(podkey *string) {
        podobj, exists, err := agent.podInformer.GetStore().GetByKey(*podkey)
        if err != nil {
                agent.log.Error("Could not lookup pod: ", err)
        }
        if !exists || podobj == nil {
                agent.log.Debug("Object doesn't exist yet ", *podkey)
                return
        }

        agent.indexMutex.Lock()
        defer agent.indexMutex.Unlock()
        agent.podChangedLocked(podobj)
}

// Must have index lock
func (agent *HostAgent) podChangedPostLock(podkey *string) {
        podobj, exists, err := agent.podInformer.GetStore().GetByKey(*podkey)
        if err != nil {
                agent.log.Error("Could not lookup pod: ", err)
        }
        if !exists || podobj == nil {
                agent.log.Debug("Object doesn't exist yet ", *podkey)
                return
        }

        agent.podChangedLocked(podobj)
}

func (agent *HostAgent) podChangedLocked(podobj interface{}) {
        pod := podobj.(*v1.Pod)
        logger := podLogger(agent.log, pod)

        if agent.config.ChainedMode {
                return
        }
        epMetaKey := fmt.Sprintf("%s/%s", pod.ObjectMeta.Namespace, pod.ObjectMeta.Name)
        epUuid := string(pod.ObjectMeta.UID)

        if !podFilter(pod) {
                agent.epDeleted(&epUuid)
                return
        }
        agent.cniToPodID[epMetaKey] = epUuid
        if pod.Status.PodIP != "" {
                type void struct{}
                var ipexists void
                podIPs := make(map[string]void)
                podIPs[pod.Status.PodIP] = ipexists
                podIPsField := reflect.ValueOf(pod.Status).FieldByName("PodIPs")
                if podIPsField.IsValid() {
                        for _, ip := range pod.Status.PodIPs {
                                podIPs[ip.IP] = ipexists
                        }
                }
                for ip := range podIPs {
                        agent.podIpToName[ip] = epMetaKey
                }
        }
        agent.podUidToName[epUuid] = epMetaKey
        epGroup, secGroup, qpGroup, _ := agent.assignGroups(pod)
        epAttributes := pod.ObjectMeta.Labels
        if epAttributes == nil {
                epAttributes = make(map[string]string)
        }
        epAttributes["vm-name"] = pod.ObjectMeta.Name
        epAttributes["namespace"] = pod.ObjectMeta.Namespace

        agent.epChanged(&epUuid, &epMetaKey, &epGroup, secGroup, qpGroup, epAttributes, logger)
        var netAttachDef []string
        if pod.ObjectMeta.Annotations[metadata.NetAttDefAnnotation] != "" {
                netAttachDef = strings.Split(pod.ObjectMeta.Annotations[metadata.NetAttDefAnnotation], ",")
        } else if pod.ObjectMeta.Annotations[metadata.MultusNetAnnotation] != "" {
                netAttachDef = strings.Split(pod.ObjectMeta.Annotations[metadata.MultusNetAnnotation], ",")
        }
        if netAttachDef != nil {
                podKey := pod.ObjectMeta.Name + "-" + pod.ObjectMeta.Namespace
                agent.podToNetAttachDef[podKey] = netAttachDef
        }
}

func (agent *HostAgent) fillEpFields(ep *opflexEndpoint, podkey string) error {
        podobj, exists, err := agent.podInformer.GetStore().GetByKey(podkey)
        if err != nil {
                return err
        }
        if !exists {
                return fmt.Errorf("Failed to get pod object of %s ", podkey)
        }
        pod := podobj.(*v1.Pod)
        epAttributes := pod.ObjectMeta.Labels
        if epAttributes == nil {
                epAttributes = make(map[string]string)
        }
        epAttributes["vm-name"] = pod.ObjectMeta.Name
        epAttributes["namespace"] = pod.ObjectMeta.Namespace
        epGroup, epSecGroups, epQosPolicy, _ := agent.assignGroups(pod)
        ep.Attributes = make(map[string]string)
        for k, v := range epAttributes {
                ep.Attributes[k] = v
        }
        if epGroup.Tenant != "" {
                ep.EgPolicySpace = epGroup.Tenant
        } else {
                ep.EgPolicySpace = epGroup.PolicySpace
        }
        if epGroup.AppProfile != "" {
                ep.EndpointGroup = epGroup.AppProfile + "|" + epGroup.Name
        } else {
                ep.EndpointGroup = epGroup.Name
        }
        ep.SecurityGroup = epSecGroups
        ep.QosPolicy = epQosPolicy
        return nil
}

func (agent *HostAgent) epChanged(epUuid *string, epMetaKey *string, epGroup *metadata.OpflexGroup,
        epSecGroups []metadata.OpflexGroup, epQosPolicy metadata.OpflexGroup, epAttributes map[string]string,
        logger *logrus.Entry) {
        if logger == nil {
                logger = agent.log.WithFields(logrus.Fields{})
        }
        if agent.config.ChainedMode {
                return
        }
        logger.Debug("epChanged...")
        epmetadata, ok := agent.epMetadata[*epMetaKey]
        if !ok {
                logger.Debug("No metadata found for ep: ")
                delete(agent.opflexEps, *epUuid)
                agent.log.Debug("syncEP scheduled")
                agent.scheduleSyncEps()
                return
        }

        var neweps []*opflexEndpoint

        for _, epmeta := range epmetadata {
                for _, iface := range epmeta.Ifaces {
                        patchIntName, patchAccessName :=
                                metadata.GetIfaceNames(iface.HostVethName)

                        ips := make([]string, 0)
                        for _, ip := range iface.IPs {
                                if ip.Address.IP == nil {
                                        continue
                                }
                                ips = append(ips, ip.Address.IP.String())
                        }

                        epidstr := *epUuid + "_" + epmeta.Id.ContId + "_" + iface.HostVethName
                        ep := &opflexEndpoint{
                                Uuid:              epidstr,
                                MacAddress:        iface.Mac,
                                IpAddress:         ips,
                                AccessIface:       iface.HostVethName,
                                AccessUplinkIface: patchAccessName,
                                IfaceName:         patchIntName,
                        }

                        ep.Attributes = make(map[string]string)
                        for k, v := range epAttributes {
                                ep.Attributes[k] = v
                        }

                        ep.Attributes["interface-name"] = iface.HostVethName
                        if epGroup.Tenant != "" {
                                ep.EgPolicySpace = epGroup.Tenant
                        } else {
                                ep.EgPolicySpace = epGroup.PolicySpace
                        }
                        if epGroup.AppProfile != "" {
                                ep.EndpointGroup = epGroup.AppProfile + "|" + epGroup.Name
                        } else {
                                ep.EndpointGroup = epGroup.Name
                        }
                        ep.SecurityGroup = epSecGroups
                        ep.QosPolicy = epQosPolicy

                        neweps = append(neweps, ep)
                }
        }

        existing, ok := agent.opflexEps[*epUuid]
        for _, ep := range existing {
                for _, newep := range neweps {
                        if ep.Uuid == newep.Uuid {
                                newep.registered = ep.registered
                        }
                }
        }

        if (ok && !reflect.DeepEqual(existing, neweps)) || !ok {
                logger.WithFields(logrus.Fields{
                        "id": *epMetaKey,
                        "ep": neweps,
                }).Debug("Updated endpoints for pod")
                logger.Infof("EP updated: %+v", neweps[0])
                agent.log.Debug("syncEP scheduled")

                agent.opflexEps[*epUuid] = neweps
                agent.scheduleSyncEps()
        }
}

func (agent *HostAgent) epDeleted(epUuid *string) {
        if _, ok := agent.opflexEps[*epUuid]; ok {
                delete(agent.opflexEps, *epUuid)
                agent.log.Debug("syncEP scheduled")
                agent.scheduleSyncEps()
        }
}

func (agent *HostAgent) podDeleted(obj interface{}) {
        pod, isPod := obj.(*v1.Pod)
        if !isPod {
                deletedState, ok := obj.(cache.DeletedFinalStateUnknown)
                if !ok {
                        agent.log.Error("Received unexpected object: ", obj)
                        return
                }
                pod, ok = deletedState.Obj.(*v1.Pod)
                if !ok {
                        agent.log.Error("DeletedFinalStateUnknown contained non-Pod object: ", deletedState.Obj)
                        return
                }
        }
        agent.log.Infof("Pod deleted: name %s namespace %s", pod.ObjectMeta.Name, pod.ObjectMeta.Namespace)
        podns, podname := pod.ObjectMeta.Namespace, pod.ObjectMeta.Name
        podid := podns + "/" + podname
        mdmap, ok := agent.epMetadata[podid]
        if ok {
                for contid, v := range mdmap {
                        metadata := cnimeta.ContainerMetadata{
                                Id: cnimeta.ContainerId{
                                        ContId:    contid,
                                        Namespace: podns,
                                        Pod:       podname,
                                },
                                Network: cnimeta.ContainerNetworkMetadata{
                                        NetworkName: v.Network.NetworkName,
                                        ChainedMode: agent.config.ChainedMode,
                                },
                        }
                        err := agent.unconfigureContainerIfaces(&metadata)
                        if err != nil {
                                agent.log.Warnf("Could not unconfigure container:%s, %s ", contid, err)
                        }
                }
        }
        agent.indexMutex.Lock()
        defer agent.indexMutex.Unlock()
        agent.podDeletedLocked(obj)
        agent.depPods.DeletePod(obj.(*v1.Pod))
        agent.rcPods.DeletePod(obj.(*v1.Pod))
        if !agent.config.ChainedMode {
                agent.qosPolPods.DeletePod(obj.(*v1.Pod))
        }
        agent.netPolPods.DeletePod(obj.(*v1.Pod))
        if !agent.config.ChainedMode {
                agent.handleObjectDeleteForSnat(obj)
        }
}

func (agent *HostAgent) podDeletedLocked(obj interface{}) {
        pod := obj.(*v1.Pod)
        u := string(pod.ObjectMeta.UID)
        if _, ok := agent.opflexEps[u]; ok {
                agent.log.Infof("podDeletedLocked: delete %s/%s", pod.ObjectMeta.Namespace, pod.ObjectMeta.Name)
                delete(agent.opflexEps, u)
                agent.log.Debug("syncEP scheduled")
                agent.scheduleSyncEps()
        }
        if pod.Status.PodIP != "" {
                delete(agent.podIpToName, pod.Status.PodIP)
        }
        delete(agent.podUidToName, u)
        agent.epDeleted(&u)
}

func (agent *HostAgent) cniEpDelete(cniKey string) {
        agent.indexMutex.Lock()
        defer agent.indexMutex.Unlock()

        epUuid, ok := agent.cniToPodID[cniKey]
        if !ok {
                agent.log.Warnf("cniEpDelete: PodID not found for %s", cniKey)
                return
        }
        delete(agent.cniToPodID, cniKey)

        if _, ok := agent.opflexEps[epUuid]; ok {
                agent.log.Infof("cniEpDelete: delete %s", cniKey)
                delete(agent.opflexEps, epUuid)
                agent.log.Debug("syncEP scheduled")
                agent.scheduleSyncEps()
        }
}

func (agent *HostAgent) updateGbpServerInfo(pod *v1.Pod) {
        if pod.ObjectMeta.Labels == nil {
                return
        }
        agent.indexMutex.Lock()
        nameVal := pod.ObjectMeta.Labels["name"]
        if nameVal == "aci-containers-controller" {
                if agent.gbpServerIP != pod.Status.PodIP {
                        agent.log.Infof("gbpServerIPChanged to %s", pod.Status.PodIP)
                        agent.gbpServerIP = pod.Status.PodIP
                        agent.scheduleSyncOpflexServer()
                }
        }
        agent.indexMutex.Unlock()
}

func formatTimestamps(timestamps []time.Time) string {
        var formattedTimestamps string
        for i, timestamp := range timestamps {
                if i > 0 {
                        formattedTimestamps += ", "
                }
                formattedTimestamps += timestamp.Format("2006-01-02T15:04:05Z")
        }
        return formattedTimestamps
}

func formatMapTimestamps(timestamps map[time.Time]string) string {
        var formattedTimestamps string
        for key, value := range timestamps {
                formattedTimestamps += key.Format("2006-01-02T15:04:05Z") + ": " + value + ", "
        }

        if len(formattedTimestamps) > 2 {
                formattedTimestamps = formattedTimestamps[:len(formattedTimestamps)-2]
        }
        return formattedTimestamps
}

func compareOpflexEndpoints(oldEndpoint, newEndpoint *opflexEndpoint) string {
        var changedFields []string

        oldEndpointJSON, _ := json.Marshal(oldEndpoint)
        newEndpointJSON, _ := json.Marshal(newEndpoint)

        var oldMap, newMap map[string]interface{}
        json.Unmarshal(oldEndpointJSON, &oldMap)
        json.Unmarshal(newEndpointJSON, &newMap)

        for key, oldValue := range oldMap {
                newValue, exists := newMap[key]
                if !exists || !reflect.DeepEqual(oldValue, newValue) {
                        changedFields = append(changedFields, key)
                }
        }

        for key := range newMap {
                if _, exists := oldMap[key]; !exists {
                        changedFields = append(changedFields, key)
                }
        }

        return strings.Join(changedFields, ", ")
}

func normalizeToMinutes(duration time.Duration) string {
        return fmt.Sprintf("%.2f min", duration.Minutes())
}

noironetworks / aci-containers / 8852

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous