9623

Committed 17 Aug 2024 08:14PM UTC coverage: 70.854% (-0.05%) from 70.908%

Build # 9623

Build Type

push

travis-pro

Committed by

abhis2112

Commit Message

Adding packet tracing feature for ovs datapath

Run Details

13035 of 18397 relevant lines covered (70.85%)

0.81 hits per line

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

66.1

/pkg/hostagent/services.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.

package hostagent

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

        "github.com/noironetworks/aci-containers/pkg/util"
        configv1 "github.com/openshift/api/config/v1"
        "github.com/sirupsen/logrus"
        v1 "k8s.io/api/core/v1"
        discovery "k8s.io/api/discovery/v1"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/labels"
        "k8s.io/apimachinery/pkg/runtime"
        "k8s.io/apimachinery/pkg/types"
        "k8s.io/apimachinery/pkg/watch"
        "k8s.io/client-go/kubernetes"
        "k8s.io/client-go/tools/cache"
        "k8s.io/kubernetes/pkg/controller"
        "sigs.k8s.io/controller-runtime/pkg/client"
        "sigs.k8s.io/controller-runtime/pkg/client/config"
)

type opflexServiceMapping struct {
        ServiceIp    string `json:"service-ip,omitempty"`
        ServiceProto string `json:"service-proto,omitempty"`
        ServicePort  uint16 `json:"service-port,omitempty"`

        NextHopIps  []string `json:"next-hop-ips"`
        NextHopPort uint16   `json:"next-hop-port,omitempty"`

        Conntrack       bool                         `json:"conntrack-enabled"`
        NodePort        uint16                       `json:"node-port,omitempty"`
        SessionAffinity *opflexSessionAffinityConfig `json:"session-affinity,omitempty"`
}

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

        DomainPolicySpace string `json:"domain-policy-space,omitempty"`
        DomainName        string `json:"domain-name,omitempty"`

        ServiceMode   string `json:"service-mode,omitempty"`
        ServiceMac    string `json:"service-mac,omitempty"`
        InterfaceName string `json:"interface-name,omitempty"`
        InterfaceIp   string `json:"interface-ip,omitempty"`
        InterfaceVlan uint16 `json:"interface-vlan,omitempty"`
        ServiceType   string `json:"service-type,omitempty"`

        ServiceMappings []opflexServiceMapping `json:"service-mapping"`

        Attributes map[string]string `json:"attributes,omitempty"`
}

type opflexSessionAffinityConfig struct {
        // clientIP contains the configurations of Client IP based session affinity.
        ClientIP opflexClientIPConfig `json:"client-ip,omitempty"`
}

// ClientIPConfig represents the configurations of Client IP based session affinity.
type opflexClientIPConfig struct {
        // timeoutSeconds specifies the seconds of ClientIP type session sticky time.
        // The value must be >0 && <=86400(for 1 day) if ServiceAffinity == "ClientIP".
        // Default value is 10800(for 3 hours).
        TimeoutSeconds int32 `json:"timeout-seconds,omitempty"`
}

// Default Session value is 10800(for 3 hours)
const (
        DefaultSessionAffinityTimer = 10800
        TempSessionAffinityTimer    = 1
)

// Name of the Openshift Service
const (
        RouterInternalDefault string = "router-internal-default"
)

// Namespace of Openshift Service
const (
        OpenShiftIngressNs string = "openshift-ingress"
)

// Represent the Openshift services
type opflexOcService struct {
        Name      string
        Namespace string
}

var Version = map[string]bool{
        "openshift-4.6-baremetal":              true,
        "openshift-4.7-baremetal":              true,
        "openshift-4.8-baremetal":              true,
        "openshift-4.9-baremetal":              true,
        "openshift-4.10-baremetal":             true,
        "openshift-4.11-baremetal":             true,
        "openshift-4.12-baremetal":             true,
        "openshift-4.13-baremetal":             true,
        "openshift-4.14-baremetal":             true,
        "openshift-4.15-baremetal":             true,
        "openshift-4.14-agent-based-baremetal": true,
        "openshift-4.4-esx":                    true,
        "openshift-4.5-esx":                    true,
        "openshift-4.6-esx":                    true,
        "openshift-4.7-esx":                    true,
        "openshift-4.8-esx":                    true,
        "openshift-4.9-esx":                    true,
        "openshift-4.10-esx":                   true,
        "openshift-4.11-esx":                   true,
        "openshift-4.12-esx":                   true,
        "openshift-4.13-esx":                   true,
        "openshift-4.14-esx":                   true,
        "openshift-4.15-esx":                   true,
        "openshift-4.14-agent-based-esx":       true,
}

func (agent *HostAgent) initEndpointsInformerFromClient(
        kubeClient *kubernetes.Clientset) {
        agent.initEndpointsInformerBase(
                &cache.ListWatch{
                        ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
                                obj, err := kubeClient.CoreV1().Endpoints(metav1.NamespaceAll).List(context.TODO(), options)
                                if err != nil {
                                        agent.log.Fatalf("Failed to list Endpoints during initialization of EndpointsInformer: %s", err)
                                }
                                return obj, err
                        },
                        WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
                                obj, err := kubeClient.CoreV1().Endpoints(metav1.NamespaceAll).Watch(context.TODO(), options)
                                if err != nil {
                                        agent.log.Fatalf("Failed to watch Endpoints during initialization of EndpointsInformer: %s", err)
                                }
                                return obj, err
                        },
                })
}

func (agent *HostAgent) initEndpointsInformerBase(listWatch *cache.ListWatch) {
        agent.endpointsInformer = cache.NewSharedIndexInformer(
                listWatch,
                &v1.Endpoints{},
                controller.NoResyncPeriodFunc(),
                cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
        )
        agent.endpointsInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
                AddFunc: func(obj interface{}) {
                        agent.endpointsChanged(obj)
                },
                UpdateFunc: func(_ interface{}, obj interface{}) {
                        agent.endpointsChanged(obj)
                },
                DeleteFunc: func(obj interface{}) {
                        agent.endpointsChanged(obj)
                },
        })
}

func (agent *HostAgent) initEndpointSliceInformerFromClient(
        kubeClient *kubernetes.Clientset) {
        agent.initEndpointSliceInformerBase(
                &cache.ListWatch{
                        ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
                                obj, err := kubeClient.DiscoveryV1().EndpointSlices(metav1.NamespaceAll).List(context.TODO(), options)
                                if err != nil {
                                        agent.log.Fatalf("Failed to list EndpointSlices during initialization of EndpointSliceInformer: %s", err)
                                }
                                return obj, err
                        },
                        WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
                                obj, err := kubeClient.DiscoveryV1().EndpointSlices(metav1.NamespaceAll).Watch(context.TODO(), options)
                                if err != nil {
                                        agent.log.Fatalf("Failed to watch EndpointSlices during initialization of EndpointSliceInformer: %s", err)
                                }
                                return obj, err
                        },
                })
}

func (agent *HostAgent) initEndpointSliceInformerBase(listWatch *cache.ListWatch) {
        agent.endpointSliceInformer = cache.NewSharedIndexInformer(
                listWatch,
                &discovery.EndpointSlice{},
                controller.NoResyncPeriodFunc(),
                cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
        )
        agent.endpointSliceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
                AddFunc: func(obj interface{}) {
                        agent.endpointSliceChanged(obj)
                },
                UpdateFunc: func(_ interface{}, obj interface{}) {
                        agent.endpointSliceChanged(obj)
                },
                DeleteFunc: func(obj interface{}) {
                        agent.endpointSliceChanged(obj)
                },
        })
}

func (agent *HostAgent) initServiceInformerFromClient(
        kubeClient *kubernetes.Clientset) {
        agent.initServiceInformerBase(
                &cache.ListWatch{
                        ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
                                obj, err := kubeClient.CoreV1().Services(metav1.NamespaceAll).List(context.TODO(), options)
                                if err != nil {
                                        agent.log.Fatalf("Failed to list Services during initialization of ServiceInformer: %s", err)
                                }
                                return obj, err
                        },
                        WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
                                obj, err := kubeClient.CoreV1().Services(metav1.NamespaceAll).Watch(context.TODO(), options)
                                if err != nil {
                                        agent.log.Fatalf("Failed to watch Services during initialization of ServiceInformer: %s", err)
                                }
                                return obj, err
                        },
                })
}

func (agent *HostAgent) initServiceInformerBase(listWatch *cache.ListWatch) {
        agent.serviceInformer = cache.NewSharedIndexInformer(
                listWatch,
                &v1.Service{},
                controller.NoResyncPeriodFunc(),
                cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
        )
        agent.serviceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
                AddFunc: func(obj interface{}) {
                        agent.serviceChanged(obj)
                },
                UpdateFunc: func(_ interface{}, obj interface{}) {
                        agent.serviceChanged(obj)
                },
                DeleteFunc: func(obj interface{}) {
                        agent.serviceDeleted(obj)
                },
        })
}

func writeAs(asfile string, as *opflexService) (bool, error) {
        newdata, err := json.MarshalIndent(as, "", "  ")
        if err != nil {
                return true, err
        }
        existingdata, err := os.ReadFile(asfile)
        if err == nil && reflect.DeepEqual(existingdata, newdata) {
                return false, nil
        }

        err = os.WriteFile(asfile, newdata, 0644)
        return true, err
}

func serviceLogger(log *logrus.Logger, as *v1.Service) *logrus.Entry {
        return log.WithFields(logrus.Fields{
                "namespace": as.ObjectMeta.Namespace,
                "name":      as.ObjectMeta.Name,
                "type":      as.Spec.Type,
        })
}

func opflexServiceLogger(log *logrus.Logger, as *opflexService) *logrus.Entry {
        return log.WithFields(logrus.Fields{
                "namespace": as.Attributes["namespace"],
                "name":      as.Attributes["name"],
                "uuid":      as.Uuid,
                "tenant":    as.DomainPolicySpace,
                "vrf":       as.DomainName,
        })
}

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

        agent.log.Debug("Syncing services")
        agent.indexMutex.Lock()
        opflexServices := make(map[string]*opflexService)
        for k, v := range agent.opflexServices {
                val := &opflexService{}
                err := util.DeepCopyObj(v, val)
                if err != nil {
                        continue
                }
                opflexServices[k] = val
        }
        agent.indexMutex.Unlock()

        files, err := os.ReadDir(agent.config.OpFlexServiceDir)
        if err != nil {
                agent.log.WithFields(
                        logrus.Fields{"serviceDir": agent.config.OpFlexServiceDir},
                ).Error("Could not read directory " + err.Error())
                return true
        }
        seen := make(map[string]bool)
        for _, f := range files {
                uuid := f.Name()
                if strings.HasSuffix(uuid, ".as") {
                        uuid = uuid[:len(uuid)-3]
                } else if strings.HasSuffix(f.Name(), ".service") {
                        uuid = uuid[:len(uuid)-8]
                } else {
                        continue
                }

                asfile := filepath.Join(agent.config.OpFlexServiceDir, f.Name())
                logger := agent.log.WithFields(
                        logrus.Fields{"Uuid": uuid},
                )
                existing, ok := opflexServices[uuid]
                if ok {
                        wrote, err := writeAs(asfile, existing)
                        if err != nil {
                                opflexServiceLogger(agent.log, existing).
                                        Error("Error writing service file: ", err)
                        } else if wrote {
                                opflexServiceLogger(agent.log, existing).Info("Updated service")
                        }
                        seen[uuid] = true
                } else {
                        logger.Info("Removing service")
                        os.Remove(asfile)
                }
        }

        for _, as := range opflexServices {
                if seen[as.Uuid] {
                        continue
                }

                opflexServiceLogger(agent.log, as).Info("Adding service")
                asfile :=
                        filepath.Join(agent.config.OpFlexServiceDir, as.Uuid+".service")
                _, err = writeAs(asfile, as)
                if err != nil {
                        opflexServiceLogger(agent.log, as).
                                Error("Error writing service file: ", err)
                }
        }

        agent.log.Debug("Finished service sync")
        return false
}

// Must have index lock
func (agent *HostAgent) updateServiceDesc(external bool, as *v1.Service, key string) bool {
        if as.Spec.ClusterIP == "None" {
                agent.log.Debugf("ClusterIP of service %s is set to None", as.ObjectMeta.Name)
                return true
        }

        ofas := &opflexService{
                Uuid:              string(as.ObjectMeta.UID),
                DomainPolicySpace: agent.config.AciVrfTenant,
                DomainName:        agent.config.AciVrf,
                ServiceMode:       "loadbalancer",
                ServiceMappings:   make([]opflexServiceMapping, 0),
        }
        switch as.Spec.Type {
        case v1.ServiceTypeClusterIP:
                ofas.ServiceType = "clusterIp"
        case v1.ServiceTypeNodePort:
                ofas.ServiceType = "nodePort"
        case v1.ServiceTypeLoadBalancer:
                ofas.ServiceType = "loadBalancer"
        case v1.ServiceTypeExternalName:
                ofas.ServiceType = "externalName"
        }

        if external {
                if agent.config.UplinkIface == "" ||
                        agent.serviceEp.Ipv4 == nil ||
                        agent.serviceEp.Mac == "" {
                        return false
                }

                ofas.InterfaceName = agent.config.UplinkIface
                ofas.InterfaceVlan = uint16(agent.config.ServiceVlan)
                // Directly using the Uplink MacAdress instead of using Opflex injected mac
                ofas.ServiceMac = agent.config.UplinkMacAdress
                ofas.InterfaceIp = agent.serviceEp.Ipv4.String()
                ofas.Uuid += "-external"
        }
        hasValidMapping := false
        for ix := range as.Spec.Ports {
                hasValidMapping = agent.serviceEndPoints.SetOpflexService(ofas, as, external, key, &as.Spec.Ports[ix])
        }

        id := fmt.Sprintf("%s_%s", as.ObjectMeta.Namespace, as.ObjectMeta.Name)
        ofas.Attributes = as.ObjectMeta.Labels
        if ofas.Attributes == nil {
                ofas.Attributes = make(map[string]string)
        }
        ofas.Attributes["namespace"] = as.ObjectMeta.Namespace
        ofas.Attributes["name"] = as.ObjectMeta.Name
        ofas.Attributes["service-name"] = id

        existing, ok := agent.opflexServices[ofas.Uuid]
        if hasValidMapping {
                if (ok && !reflect.DeepEqual(existing, ofas)) || !ok {
                        agent.opflexServices[ofas.Uuid] = ofas
                        // Check matching oc serivce and create a extra service file.
                        // This Change is specfic to Openshfit domain
                        agent.setOpenShfitService(as, external, ofas)
                }
                return true
        }
        if ok {
                delete(agent.opflexServices, ofas.Uuid)
                return true
        }

        return false
}

// must have index lock
func (agent *HostAgent) doUpdateService(key string) {
        asobj, exists, err := agent.serviceInformer.GetStore().GetByKey(key)
        if err != nil {
                agent.log.Error("Could not lookup service for " +
                        key + ": " + err.Error())
                return
        }
        if !exists || asobj == nil {
                return
        }
        as := asobj.(*v1.Service)
        doSync := false
        doSync = agent.updateServiceDesc(false, as, key) || doSync
        doSync = agent.updateServiceDesc(true, as, key) || doSync
        if doSync {
                agent.scheduleSyncServices()
                agent.updateEpFileWithClusterIp(as, false)
                agent.scheduleSyncEps()
        }
}

func (agent *HostAgent) endpointsChanged(obj interface{}) {
        agent.indexMutex.Lock()
        defer agent.indexMutex.Unlock()

        endpoints := obj.(*v1.Endpoints)
        agent.log.Debugf("Endpoint changed: name=%s namespace=%s",
                endpoints.ObjectMeta.Name, endpoints.ObjectMeta.Namespace)

        key, err := cache.MetaNamespaceKeyFunc(endpoints)
        if err != nil {
                agent.log.Error("Could not create key:" + err.Error())
                return
        }
        agent.doUpdateService(key)
}
func getServiceKey(endPointSlice *discovery.EndpointSlice) (string, bool) {
        serviceName, ok := endPointSlice.Labels[discovery.LabelServiceName]
        if !ok {
                return "", false
        }
        return endPointSlice.ObjectMeta.Namespace + "/" + serviceName, true
}

func (agent *HostAgent) endpointSliceChanged(obj interface{}) {
        agent.indexMutex.Lock()
        defer agent.indexMutex.Unlock()
        endpointslice := obj.(*discovery.EndpointSlice)
        agent.log.Debugf("endpointslice changed: name=%s namespace=%s",
                endpointslice.ObjectMeta.Name, endpointslice.ObjectMeta.Namespace)
        servicekey, ok := getServiceKey(endpointslice)
        if !ok {
                return
        }
        agent.doUpdateService(servicekey)
}

func (agent *HostAgent) serviceChanged(obj interface{}) {
        agent.indexMutex.Lock()
        defer agent.indexMutex.Unlock()

        as := obj.(*v1.Service)
        agent.log.Debugf("Service changed: name=%s namespace=%s",
                as.ObjectMeta.Name, as.ObjectMeta.Namespace)

        key, err := cache.MetaNamespaceKeyFunc(as)
        if err != nil {
                serviceLogger(agent.log, as).
                        Error("Could not create service object key:" + err.Error())
                return
        }
        agent.doUpdateService(key)
        agent.handleObjectUpdateForSnat(obj)
}

func (agent *HostAgent) serviceDeleted(obj interface{}) {
        agent.indexMutex.Lock()
        defer agent.indexMutex.Unlock()

        as, isService := obj.(*v1.Service)
        if !isService {
                deletedState, ok := obj.(cache.DeletedFinalStateUnknown)
                if !ok {
                        agent.log.Error("Received unexpected object: ", obj)
                        return
                }
                as, ok = deletedState.Obj.(*v1.Service)
                if !ok {
                        agent.log.Error("DeletedFinalStateUnknown contained non-Services object: ", deletedState.Obj)
                        return
                }
        }
        agent.log.Debugf("Service deleted: name=%s namespace=%s",
                as.ObjectMeta.Name, as.ObjectMeta.Namespace)

        u := string(as.ObjectMeta.UID)
        if _, ok := agent.opflexServices[u]; ok {
                delete(agent.opflexServices, u)
                delete(agent.opflexServices, u+"-external")
                for _, v := range agent.ocServices {
                        if v.Name == as.ObjectMeta.Name &&
                                v.Namespace == as.ObjectMeta.Namespace {
                                delete(agent.opflexServices, u+"-"+v.Name)
                        }
                }
                agent.scheduleSyncServices()
                agent.deleteServIpFromEp(u)
                agent.scheduleSyncEps()
        }
        agent.handleObjectDeleteForSnat(obj)
}

func (agent *HostAgent) updateAllServices() {
        if agent.serviceInformer == nil {
                return
        }
        store := agent.serviceInformer.GetStore()
        if store == nil {
                return
        }
        keys := agent.serviceInformer.GetStore().ListKeys()
        if keys == nil {
                return
        }

        agent.indexMutex.Lock()
        defer agent.indexMutex.Unlock()
        for _, key := range keys {
                agent.doUpdateService(key)
        }
}

// This API is get the OpenShift InfrastructreIp's
func (agent *HostAgent) getInfrastucreIp(serviceName string) string {
        if _, ok := Version[agent.config.Flavor]; ok {
                if serviceName == RouterInternalDefault {
                        return agent.config.InstallerProvlbIp
                }
                return ""
        }
        infraStructureInfo := &configv1.Infrastructure{
                TypeMeta:   metav1.TypeMeta{APIVersion: configv1.GroupVersion.String(), Kind: "Infrastructure"},
                ObjectMeta: metav1.ObjectMeta{Name: "cluster"},
        }
        cfg, _ := config.GetConfig()
        scheme := runtime.NewScheme()
        scheme.AddKnownTypes(configv1.SchemeGroupVersion, &configv1.Infrastructure{})
        scheme.AddKnownTypes(configv1.SchemeGroupVersion, &metav1.GetOptions{})
        rclient, err := client.New(cfg, client.Options{Scheme: scheme})
        if err != nil {
                agent.log.Error(err.Error())
                return ""
        }
        if rclient == nil {
                agent.log.Error("client is nil")
                return ""
        }
        err = rclient.Get(context.TODO(), types.NamespacedName{
                Name: "cluster"}, infraStructureInfo)
        if err != nil {
                agent.log.Error(err.Error())
                return ""
        }
        if infraStructureInfo.Status.Platform == configv1.OpenStackPlatformType {
                if infraStructureInfo.Status.PlatformStatus != nil &&
                        infraStructureInfo.Status.PlatformStatus.OpenStack != nil {
                        if serviceName == RouterInternalDefault {
                                return infraStructureInfo.Status.PlatformStatus.OpenStack.IngressIP
                        }
                }
        }
        return ""
}

func (agent *HostAgent) setOpenShfitService(as *v1.Service, external bool, ofas *opflexService) {
        if agent.config.AciVmmDomainType == "OpenShift" {
                if !external {
                        for _, v := range agent.ocServices {
                                // Check for Namespace is equal
                                if v.Namespace != as.ObjectMeta.Namespace {
                                        continue
                                }
                                // Check Service Name is equal
                                if v.Name != as.ObjectMeta.Name {
                                        continue
                                }
                                InfraIp := agent.getInfrastucreIp(as.ObjectMeta.Name)
                                agent.log.Debug("InfraIp: ", InfraIp)
                                if InfraIp == "" {
                                        continue
                                }
                                ocas := &opflexService{
                                        Uuid:              string(as.ObjectMeta.UID),
                                        DomainPolicySpace: agent.config.AciVrfTenant,
                                        DomainName:        agent.config.AciVrf,
                                        ServiceMode:       "loadbalancer",
                                        ServiceMappings:   make([]opflexServiceMapping, 0),
                                }
                                ocas.Uuid = ocas.Uuid + "-" + as.ObjectMeta.Name
                                for _, val := range ofas.ServiceMappings {
                                        val.ServiceIp = InfraIp
                                        ocas.ServiceMappings = append(ocas.ServiceMappings, val)
                                }
                                ocas.ServiceType = ofas.ServiceType
                                ocas.Attributes = ofas.Attributes
                                agent.opflexServices[ocas.Uuid] = ocas
                        }
                }
        }
}

func (sep *serviceEndpoint) SetOpflexService(ofas *opflexService, as *v1.Service,
        external bool, key string, sp *v1.ServicePort) bool {
        agent := sep.agent
        endpointsobj, exists, err :=
                agent.endpointsInformer.GetStore().GetByKey(key)
        if err != nil {
                agent.log.Error("Could not lookup endpoints for " +
                        key + ": " + err.Error())
                return false
        }
        if !exists || endpointsobj == nil {
                agent.log.Debugf("no endpoints for service %s/%s", as.Namespace, as.Name)
                return false
        }
        endpoints := endpointsobj.(*v1.Endpoints)
        hasValidMapping := false

        type void struct{}
        var ipexists void
        clusterIPs := make(map[string]void)
        clusterIPs[as.Spec.ClusterIP] = ipexists
        clusterIPsField := reflect.ValueOf(as.Spec).FieldByName("ClusterIPs")
        if clusterIPsField.IsValid() {
                for _, ip := range as.Spec.ClusterIPs {
                        clusterIPs[ip] = ipexists
                }
        }

        for clusterIP := range clusterIPs {
                for _, e := range endpoints.Subsets {
                        if len(e.Addresses) == 0 {
                                continue
                        }
                        parsedClusterIp := net.ParseIP(clusterIP)
                        parsedPodIp := net.ParseIP(e.Addresses[0].IP)

                        if parsedClusterIp == nil || parsedPodIp == nil {
                                agent.log.Info("Not a valid IP address..", parsedClusterIp, parsedPodIp)
                                continue
                        }
                        if parsedClusterIp.To4() != nil && parsedPodIp.To4() != nil {
                                agent.log.Info("Both are IPv4 addresses..", parsedClusterIp, parsedPodIp, "Adding to map..")
                        } else if parsedClusterIp.To4() == nil && parsedPodIp.To4() == nil {
                                agent.log.Info("Both are IPv6 addresses..", parsedClusterIp, parsedPodIp, "Adding to map..")
                        } else {
                                continue
                        }
                        for _, p := range e.Ports {
                                if p.Protocol != sp.Protocol {
                                        continue
                                }
                                if p.Name != sp.Name {
                                        continue
                                }

                                sm := &opflexServiceMapping{
                                        ServicePort:  uint16(sp.Port),
                                        ServiceProto: strings.ToLower(string(sp.Protocol)),
                                        NextHopIps:   make([]string, 0),
                                        NextHopPort:  uint16(p.Port),
                                        Conntrack:    true,
                                        NodePort:     uint16(sp.NodePort),
                                }

                                if external {
                                        if as.Spec.Type == v1.ServiceTypeLoadBalancer &&
                                                len(as.Status.LoadBalancer.Ingress) > 0 {
                                                for _, ip := range as.Status.LoadBalancer.Ingress {
                                                        LBIp := net.ParseIP(ip.IP)
                                                        if (LBIp.To4() != nil) == (parsedPodIp.To4() != nil) {
                                                                sm.ServiceIp = ip.IP
                                                                break
                                                        }
                                                }
                                        }
                                } else {
                                        sm.ServiceIp = clusterIP
                                }
                                sm.SessionAffinity = getSessionAffinity(as)
                                for _, a := range e.Addresses {
                                        if !external ||
                                                (a.NodeName != nil && *a.NodeName == agent.config.NodeName) {
                                                sm.NextHopIps = append(sm.NextHopIps, a.IP)
                                        }
                                }
                                if sm.ServiceIp != "" && len(sm.NextHopIps) > 0 {
                                        hasValidMapping = true
                                }

                                ofas.ServiceMappings = append(ofas.ServiceMappings, *sm)
                        }
                }
        }
        return hasValidMapping
}

func getSessionAffinity(as *v1.Service) *opflexSessionAffinityConfig {
        if as.Spec.SessionAffinityConfig != nil && as.Spec.SessionAffinity == "ClientIP" {
                config := as.Spec.SessionAffinityConfig
                if config.ClientIP != nil && config.ClientIP.TimeoutSeconds != nil {
                        return &opflexSessionAffinityConfig{ClientIP: opflexClientIPConfig{TimeoutSeconds: *config.ClientIP.TimeoutSeconds}}
                } else {
                        return &opflexSessionAffinityConfig{ClientIP: opflexClientIPConfig{TimeoutSeconds: DefaultSessionAffinityTimer}}
                }
        }
        return &opflexSessionAffinityConfig{ClientIP: opflexClientIPConfig{TimeoutSeconds: TempSessionAffinityTimer}}
}

func (seps *serviceEndpointSlice) SetOpflexService(ofas *opflexService, as *v1.Service,
        external bool, key string, sp *v1.ServicePort) bool {
        agent := seps.agent
        hasValidMapping := false
        var endpointSlices []*discovery.EndpointSlice
        label := map[string]string{discovery.LabelServiceName: as.ObjectMeta.Name}
        selector := labels.SelectorFromSet(labels.Set(label))
        cache.ListAllByNamespace(agent.endpointSliceInformer.GetIndexer(), as.ObjectMeta.Namespace, selector,
                func(endpointSliceobj interface{}) {
                        endpointSlices = append(endpointSlices, endpointSliceobj.(*discovery.EndpointSlice))
                })

        type void struct{}
        var ipexists void
        clusterIPs := make(map[string]void)
        clusterIPs[as.Spec.ClusterIP] = ipexists
        clusterIPsField := reflect.ValueOf(as.Spec).FieldByName("ClusterIPs")
        if clusterIPsField.IsValid() {
                for _, ip := range as.Spec.ClusterIPs {
                        clusterIPs[ip] = ipexists
                }
        }

        for clusterIP := range clusterIPs {
                for _, endpointSlice := range endpointSlices {
                        if !(len(endpointSlice.Endpoints) > 0 && len(endpointSlice.Endpoints[0].Addresses) > 0) {
                                continue
                        }
                        parsedClusterIp := net.ParseIP(clusterIP)
                        parsedPodIp := net.ParseIP(endpointSlice.Endpoints[0].Addresses[0])
                        if parsedClusterIp == nil || parsedPodIp == nil {
                                agent.log.Info("Not a valid IP address..", parsedClusterIp, parsedPodIp)
                                continue
                        }
                        if parsedClusterIp.To4() != nil && parsedPodIp.To4() != nil {
                                agent.log.Info("Both are IPv4 addresses..", parsedClusterIp, parsedPodIp, "Adding to map..")
                        } else if parsedClusterIp.To4() == nil && parsedPodIp.To4() == nil {
                                agent.log.Info("Both are IPv6 addresses..", parsedClusterIp, parsedPodIp, "Adding to map..")
                        } else {
                                continue
                        }
                        for _, p := range endpointSlice.Ports {
                                if p.Protocol != nil && *p.Protocol != sp.Protocol {
                                        continue
                                }

                                if p.Name != nil && *p.Name != sp.Name {
                                        continue
                                }

                                sm := &opflexServiceMapping{
                                        ServicePort:  uint16(sp.Port),
                                        ServiceProto: strings.ToLower(string(sp.Protocol)),
                                        NextHopIps:   make([]string, 0),
                                        NextHopPort:  uint16(*p.Port),
                                        Conntrack:    true,
                                        NodePort:     uint16(sp.NodePort),
                                }

                                if external {
                                        if as.Spec.Type == v1.ServiceTypeLoadBalancer &&
                                                len(as.Status.LoadBalancer.Ingress) > 0 {
                                                for _, ip := range as.Status.LoadBalancer.Ingress {
                                                        LBIp := net.ParseIP(ip.IP)
                                                        if (LBIp.To4() != nil) == (parsedPodIp.To4() != nil) {
                                                                sm.ServiceIp = ip.IP
                                                                break
                                                        }
                                                }
                                        }
                                } else {
                                        sm.ServiceIp = clusterIP
                                }
                                nexthops := make(map[string][]string)
                                var nodeZone string
                                for _, e := range endpointSlice.Endpoints {
                                        for _, a := range e.Addresses {
                                                if !external || (e.NodeName != nil && *e.NodeName == agent.config.NodeName) {
                                                        obj, exists, err := agent.nodeInformer.GetStore().GetByKey(agent.config.NodeName)
                                                        if err != nil {
                                                                agent.log.Error("Could not lookup node: ", err)
                                                                continue
                                                        }
                                                        if !exists && obj == nil {
                                                                agent.log.Error("Object nil")
                                                                continue
                                                        }
                                                        node := obj.(*v1.Node)
                                                        // Services need an annotation to inform the
                                                        // endpointslice controller to add hints
                                                        // Currently-1.22 only zones are used as hints.
                                                        // https://github.com/kubernetes/enhancements/tree/master/keps/sig-network/2433-topology-aware-hints
                                                        var hintsEnabled = false
                                                        if val1, ok1 := as.ObjectMeta.Annotations["service.kubernetes.io/topology-aware-routing"]; ok1 && (strings.ToLower(val1) == "auto") {
                                                                hintsEnabled = true
                                                        }
                                                        // annotation deprecated in 1.27 but still supported
                                                        if val2, ok2 := as.ObjectMeta.Annotations["service.kubernetes.io/topology-aware-hints"]; ok2 && (strings.ToLower(val2) == "auto") {
                                                                hintsEnabled = true
                                                        }
                                                        // annotation as of 1.27
                                                        if val3, ok3 := as.ObjectMeta.Annotations[v1.AnnotationTopologyMode]; ok3 && (strings.ToLower(val3) == "auto") {
                                                                hintsEnabled = true
                                                        }
                                                        zone, zoneOk := node.ObjectMeta.Labels[v1.LabelTopologyZone]
                                                        nodeZone = zone
                                                        if !external && zoneOk && hintsEnabled && e.Hints != nil {
                                                                for _, hintZone := range e.Hints.ForZones {
                                                                        if nodeZone == hintZone.Name && *e.Conditions.Ready {
                                                                                nexthops["topologyawarehints"] =
                                                                                        append(nexthops["topologyawarehints"], a)
                                                                        }
                                                                }
                                                        } else if *e.Conditions.Ready {
                                                                nexthops["any"] = append(nexthops["any"], a)
                                                        }
                                                }
                                        }
                                }
                                // Select the high priority keys as datapath doesn't have support for fallback
                                if _, ok := nexthops["topologyawarehints"]; ok {
                                        sm.NextHopIps = append(sm.NextHopIps, nexthops["topologyawarehints"]...)
                                        agent.log.Info("NextHopIps are", sm.NextHopIps)
                                        agent.log.Debugf("Topology matching hint: %s Nexthops: %s", nodeZone, sm.NextHopIps)
                                } else {
                                        sm.NextHopIps = append(sm.NextHopIps, nexthops["any"]...)
                                        agent.log.Info("NextHopIps are", sm.NextHopIps)
                                }
                                if sm.ServiceIp != "" && len(sm.NextHopIps) > 0 {
                                        hasValidMapping = true
                                }
                                sm.SessionAffinity = getSessionAffinity(as)
                                ofas.ServiceMappings = append(ofas.ServiceMappings, *sm)
                        }
                }
        }
        return hasValidMapping
}

func (agent *HostAgent) updateEpFileWithClusterIp(as *v1.Service, deleted bool) {
        suid := string(as.ObjectMeta.UID)
        ofas, ok := agent.opflexServices[suid]
        var dummy struct{}
        if ok {
                if as.Spec.ClusterIP == "None" {
                        return
                }
                podKeys := agent.getPodKeysFromSm(ofas.ServiceMappings)
                current := make(map[string]struct{})
                for _, key := range podKeys {
                        obj, exists, err := agent.podInformer.GetStore().GetByKey(key)
                        if err == nil && exists && (obj != nil) {
                                pod := obj.(*v1.Pod)
                                if agent.config.NodeName != pod.Spec.NodeName {
                                        continue
                                }
                                poduid := string(pod.ObjectMeta.UID)
                                if _, sok := agent.servicetoPodUids[suid]; !sok {
                                        agent.servicetoPodUids[suid] = make(map[string]struct{})
                                }
                                agent.servicetoPodUids[suid][poduid] = dummy
                                if _, podok := agent.podtoServiceUids[poduid]; !podok {
                                        agent.podtoServiceUids[poduid] = make(map[string][]string)
                                }

                                type void struct{}
                                var ipexists void
                                clusterIPs := make(map[string]void)
                                clusterIPs[as.Spec.ClusterIP] = ipexists
                                clusterIPsField := reflect.ValueOf(as.Spec).FieldByName("ClusterIPs")
                                if clusterIPsField.IsValid() {
                                        for _, ip := range as.Spec.ClusterIPs {
                                                clusterIPs[ip] = ipexists
                                        }
                                }

                                var listClusterIPs []string
                                for ip := range clusterIPs {
                                        listClusterIPs = append(listClusterIPs, ip)
                                }
                                agent.podtoServiceUids[poduid][suid] = listClusterIPs
                                agent.log.Info("EpUpdated: ", poduid, " with ClusterIp: ", listClusterIPs)
                                current[poduid] = dummy
                        }
                }
                // reconcile with the current pods matching the service
                // if there is any stale info remove that from service matching the pods
                // update the revese map for the pod to service
                poduids := agent.servicetoPodUids[suid]
                for id := range poduids {
                        if _, ok := current[id]; !ok {
                                delete(agent.servicetoPodUids[suid], id)
                                delete(agent.podtoServiceUids[id], suid)
                                if len(agent.podtoServiceUids[id]) == 0 {
                                        delete(agent.podtoServiceUids, id)
                                }
                        }
                }
        } else {
                agent.deleteServIpFromEp(suid)
        }
}

func (agent *HostAgent) getPodKeysFromSm(sm []opflexServiceMapping) []string {
        var podkeys []string
        if len(sm) > 0 {
                podIps := sm[0].NextHopIps
                for _, ip := range podIps {
                        podkey, ok := agent.podIpToName[ip]
                        if ok {
                                podkeys = append(podkeys, podkey)
                        }
                }
        }
        return podkeys
}

func (agent *HostAgent) getServiceIPs(poduid string) []string {
        var ips []string
        agent.indexMutex.Lock()
        v, ok := agent.podtoServiceUids[poduid]
        if ok {
                for _, service_ips := range v {
                        ips = append(ips, service_ips...)
                }
        }
        agent.indexMutex.Unlock()
        return ips
}

func (agent *HostAgent) deleteServIpFromEp(suid string) {
        v, ok := agent.servicetoPodUids[suid]
        if ok {
                for poduid := range v {
                        if _, ok := agent.podtoServiceUids[poduid]; ok {
                                delete(agent.podtoServiceUids[poduid], suid)
                                if len(agent.podtoServiceUids[poduid]) == 0 {
                                        delete(agent.podtoServiceUids, poduid)
                                }
                        }
                }
                delete(agent.servicetoPodUids, suid)
        }
}

noironetworks / aci-containers / 9623

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