1801

pending completion

Build # 1801

Build Type

Pull #856

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/contiv/vpp/commit/<a class=hub.com/contiv/vpp/commit/<a class="double-link" href="https://git"><a class=hub.com/contiv/vpp/commit/<a class="double-link" href="https://git"><a class=hub.com/contiv/vpp/commit/08aac07af4f4563582d2bad16e09fdd796df8d6a">08aac07af<a href="https://github.com/contiv/vpp/commit/08aac07af4f4563582d2bad16e09fdd796df8d6a">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/contiv/vpp/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/contiv/vpp/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/contiv/vpp/commit/08aac07af4f4563582d2bad16e09fdd796df8d6a&quot;&gt;08aac07af&lt;/a&gt;&lt;a href=&quot;https://github.com/contiv/vpp/commit/08aac07af4f4563582d2bad16e09fdd796df8d6a&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/contiv/vpp/commit/08aac07af4f4563582d2bad16e09fdd7&lt;/a&gt;96df8d6a">&quot;&gt;Merge &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/contiv/vpp/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/contiv/vpp/commit/<a class="double-link" href="https://github.com/contiv/vpp/commit/5e36210b16aad29f899c66aad4650851481d2aa7">5e36210b1</a><a href="https://github.com/contiv/vpp/commit/08aac07af4f4563582d2bad16e09fdd796df8d6a">&amp;quot;&amp;gt;5e36210b1&amp;lt;/a&amp;gt;&amp;quot;&amp;gt;5e36210b1&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/contiv/vpp/commit/08aac07af4f4563582d2bad16e09fdd796df8&lt;/a&gt;d6a&quot;&gt; into &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/contiv/vpp/commit/<a class="double-link" href="https://github.com/contiv/vpp/commit/30d3599c4">30d3599c4">30d3599c4</a>

Pull Request Pull Request #856:

Run Details

97 of 97 new or added lines in 4 files covered. (100.0%)

6388 of 8306 relevant lines covered (76.91%)

28.5 hits per line

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

88.03

/plugins/service/processor/processor_impl.go

/*
 * // Copyright (c) 2018 Cisco and/or its affiliates.
 * //
 * // 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 processor

import (
        "net"
        "strings"

        "github.com/ligato/cn-infra/datasync"
        "github.com/ligato/cn-infra/logging"
        "github.com/ligato/cn-infra/servicelabel"

        "github.com/contiv/vpp/plugins/contiv"
        nodemodel "github.com/contiv/vpp/plugins/contiv/model/node"
        epmodel "github.com/contiv/vpp/plugins/ksr/model/endpoints"
        podmodel "github.com/contiv/vpp/plugins/ksr/model/pod"
        svcmodel "github.com/contiv/vpp/plugins/ksr/model/service"
        "github.com/contiv/vpp/plugins/service/renderer"
)

// ServiceProcessor implements ServiceProcessorAPI.
type ServiceProcessor struct {
        Deps

        renderers []renderer.ServiceRendererAPI

        /* nodes */
        nodes map[int]*nodemodel.NodeInfo

        /* internal maps */
        services map[svcmodel.ID]*Service
        localEps map[podmodel.ID]*LocalEndpoint

        /* local frontend and backend interfaces */
        frontendIfs renderer.Interfaces
        backendIfs  renderer.Interfaces
}

// Deps lists dependencies of ServiceProcessor.
type Deps struct {
        Log          logging.Logger
        ServiceLabel servicelabel.ReaderAPI
        Contiv       contiv.API /* to get all interface names and pod IP network */
}

// LocalEndpoint represents a node-local endpoint.
type LocalEndpoint struct {
        ifName   string
        svcCount int /* number of services running on this endpoint. */
}

// Init initializes service processor.
func (sp *ServiceProcessor) Init() error {
        sp.reset()
        sp.Contiv.RegisterPodPreRemovalHook(sp.processDeletingPod)
        return nil
}

// AfterInit does nothing for the processor.
func (sp *ServiceProcessor) AfterInit() error {
        return nil
}

// reset clears the state of the processor.
func (sp *ServiceProcessor) reset() error {
        sp.nodes = make(map[int]*nodemodel.NodeInfo)
        sp.services = make(map[svcmodel.ID]*Service)
        sp.localEps = make(map[podmodel.ID]*LocalEndpoint)
        sp.frontendIfs = renderer.NewInterfaces()
        sp.backendIfs = renderer.NewInterfaces()
        return nil
}

// Update processes a datasync change event associated with the state data
// of K8s pods, endpoints, services and nodes.
// The data change is stored into the cache and all registered renderers
// are notified about any changes related to services that need to be
// reflected in the underlying network stack(s).
func (sp *ServiceProcessor) Update(dataChngEv datasync.ChangeEvent) error {
        return sp.propagateDataChangeEv(dataChngEv)
}

// Resync processes a datasync resync event associated with the state data
// of K8s pods, endpoints, services and nodes.
// The cache content is fully replaced and all registered renderers
// receive a full snapshot of Contiv Services at the present state to be
// (re)installed.
func (sp *ServiceProcessor) Resync(resyncEv datasync.ResyncEvent) error {
        resyncEvData := sp.parseResyncEv(resyncEv)
        return sp.processResyncEvent(resyncEvData)
}

// RegisterRenderer registers a new service renderer.
// The renderer will be receiving updates for all services on the cluster.
func (sp *ServiceProcessor) RegisterRenderer(renderer renderer.ServiceRendererAPI) error {
        sp.renderers = append(sp.renderers, renderer)
        return nil
}

func (sp *ServiceProcessor) processUpdatedPod(pod *podmodel.Pod) error {
        sp.Log.WithFields(logging.Fields{
                "pod": *pod,
        }).Debug("ServiceProcessor - processUpdatedPod()")
        podID := podmodel.ID{Name: pod.Name, Namespace: pod.Namespace}

        if pod.IpAddress == "" {
                return nil
        }
        podIPAddress := net.ParseIP(pod.IpAddress)
        if podIPAddress == nil || !sp.Contiv.GetPodNetwork().Contains(podIPAddress) {
                /* ignore pods deployed on other nodes */
                return nil
        }

        localEp := sp.getLocalEndpoint(podID)
        if localEp.ifName != "" {
                /* already processed */
                return nil
        }
        ifName, ifExists := sp.Contiv.GetIfName(podID.Namespace, podID.Name)
        if !ifExists {
                sp.Log.WithFields(logging.Fields{
                        "pod-ns":   podID.Namespace,
                        "pod-name": podID.Name,
                }).Warn("Failed to get pod interface name")
                return nil
        }

        localEp.ifName = ifName
        if localEp.svcCount > 0 {
                newBackendIfs := sp.backendIfs.Copy()
                newBackendIfs.Add(ifName)
                for _, renderer := range sp.renderers {
                        err := renderer.UpdateLocalBackendIfs(sp.backendIfs, newBackendIfs)
                        if err != nil {
                                return err
                        }
                }
                sp.backendIfs = newBackendIfs
        }
        newFrontendIfs := sp.frontendIfs.Copy()
        newFrontendIfs.Add(ifName)
        for _, renderer := range sp.renderers {
                err := renderer.UpdateLocalFrontendIfs(sp.frontendIfs, newFrontendIfs)
                if err != nil {
                        return err
                }
        }
        sp.frontendIfs = newFrontendIfs
        return nil
}

func (sp *ServiceProcessor) processDeletingPod(podNamespace string, podName string) error {
        podID := podmodel.ID{Name: podName, Namespace: podNamespace}
        sp.Log.WithFields(logging.Fields{
                "podID": podID,
        }).Debug("ServiceProcessor - processDeletingPod()")

        localEp, hasEntry := sp.localEps[podID]
        if !hasEntry {
                return nil
        }
        ifName := sp.localEps[podID].ifName
        if ifName == "" {
                return nil
        }

        if localEp.svcCount > 0 {
                newBackendIfs := sp.backendIfs.Copy()
                newBackendIfs.Del(ifName)
                for _, renderer := range sp.renderers {
                        /* ignore errors */
                        renderer.UpdateLocalBackendIfs(sp.backendIfs, newBackendIfs)
                }
                sp.backendIfs = newBackendIfs
        }
        newFrontendIfs := sp.frontendIfs.Copy()
        newFrontendIfs.Del(ifName)
        for _, renderer := range sp.renderers {
                /* ignore errors */
                renderer.UpdateLocalFrontendIfs(sp.frontendIfs, newFrontendIfs)
        }
        sp.frontendIfs = newFrontendIfs
        delete(sp.localEps, podID)
        return nil
}

func (sp *ServiceProcessor) processNewEndpoints(eps *epmodel.Endpoints) error {
        sp.Log.WithFields(logging.Fields{
                "eps": *eps,
        }).Debug("ServiceProcessor - processNewEndpoints()")

        svcID := svcmodel.ID{Namespace: eps.Namespace, Name: eps.Name}
        svc := sp.getService(svcID)
        svc.SetEndpoints(eps)
        return sp.renderService(svc, nil, []podmodel.ID{})
}

func (sp *ServiceProcessor) processUpdatedEndpoints(eps *epmodel.Endpoints) error {
        sp.Log.WithFields(logging.Fields{
                "eps": *eps,
        }).Debug("ServiceProcessor - processUpdatedEndpoints()")

        svcID := svcmodel.ID{Namespace: eps.Namespace, Name: eps.Name}
        svc := sp.getService(svcID)
        oldContivSvc := svc.GetContivService()
        oldBackends := svc.GetLocalBackends()
        svc.SetEndpoints(eps)
        return sp.renderService(svc, oldContivSvc, oldBackends)
}

func (sp *ServiceProcessor) processDeletedEndpoints(epsID epmodel.ID) error {
        sp.Log.WithFields(logging.Fields{
                "epsID": epsID,
        }).Debug("ServiceProcessor - processDeletedEndpoints()")

        svcID := svcmodel.ID{Namespace: epsID.Namespace, Name: epsID.Name}
        svc := sp.getService(svcID)
        oldContivSvc := svc.GetContivService()
        oldBackends := svc.GetLocalBackends()
        svc.SetEndpoints(nil)
        return sp.renderService(svc, oldContivSvc, oldBackends)
}

func (sp *ServiceProcessor) processNewService(service *svcmodel.Service) error {
        sp.Log.WithFields(logging.Fields{
                "service": *service,
        }).Debug("ServiceProcessor - processNewService()")

        svcID := svcmodel.ID{Namespace: service.Namespace, Name: service.Name}
        svc := sp.getService(svcID)
        svc.SetMetadata(service)
        return sp.renderService(svc, nil, []podmodel.ID{})
}

func (sp *ServiceProcessor) processUpdatedService(service *svcmodel.Service) error {
        sp.Log.WithFields(logging.Fields{
                "service": *service,
        }).Debug("ServiceProcessor - processUpdatedService()")

        svcID := svcmodel.ID{Namespace: service.Namespace, Name: service.Name}
        svc := sp.getService(svcID)
        oldContivSvc := svc.GetContivService()
        oldBackends := svc.GetLocalBackends()
        svc.SetMetadata(service)
        return sp.renderService(svc, oldContivSvc, oldBackends)
}

func (sp *ServiceProcessor) processDeletedService(serviceID svcmodel.ID) error {
        sp.Log.WithFields(logging.Fields{
                "serviceID": serviceID,
        }).Debug("ServiceProcessor - processDeletedService()")

        svcID := svcmodel.ID{Namespace: serviceID.Namespace, Name: serviceID.Name}
        svc := sp.getService(svcID)
        oldContivSvc := svc.GetContivService()
        oldBackends := svc.GetLocalBackends()
        svc.SetMetadata(nil)
        return sp.renderService(svc, oldContivSvc, oldBackends)
}

func (sp *ServiceProcessor) processNewNode(node *nodemodel.NodeInfo) error {
        sp.Log.WithFields(logging.Fields{
                "node": *node,
        }).Debug("ServiceProcessor - processNewNode()")

        sp.nodes[int(node.Id)] = node
        return sp.renderNodePorts()
}

func (sp *ServiceProcessor) processUpdatedNode(node *nodemodel.NodeInfo) error {
        sp.Log.WithFields(logging.Fields{
                "node": *node,
        }).Debug("ServiceProcessor - processUpdatedNode()")

        sp.nodes[int(node.Id)] = node
        return sp.renderNodePorts()
}

func (sp *ServiceProcessor) processDeletedNode(nodeID int) error {
        sp.Log.WithFields(logging.Fields{
                "nodeID": nodeID,
        }).Debug("ServiceProcessor - processDeletedNode()")

        if _, hasNode := sp.nodes[nodeID]; hasNode {
                delete(sp.nodes, nodeID)
                return sp.renderNodePorts()
        }
        return nil
}

// renderService reacts to added/removed/changed service.
func (sp *ServiceProcessor) renderService(svc *Service, oldContivSvc *renderer.ContivService,
        oldBackends []podmodel.ID) error {

        var err error
        newContivSvc := svc.GetContivService()
        newBackends := svc.GetLocalBackends()

        // Render service.
        if newContivSvc != nil {
                if oldContivSvc == nil {
                        for _, renderer := range sp.renderers {
                                if err = renderer.AddService(newContivSvc); err != nil {
                                        return err
                                }
                        }
                } else {
                        for _, renderer := range sp.renderers {
                                if err = renderer.UpdateService(oldContivSvc, newContivSvc); err != nil {
                                        return err
                                }
                        }
                }
        } else {
                if oldContivSvc != nil {
                        for _, renderer := range sp.renderers {
                                if err = renderer.DeleteService(oldContivSvc); err != nil {
                                        return err
                                }
                        }
                }
        }

        // Render local Backends.
        newBackendIfs := sp.backendIfs.Copy()
        updateBackends := false
        // -> handle new backend interfaces
        for _, newBackend := range newBackends {
                new := true
                for _, oldBackend := range oldBackends {
                        if newBackend == oldBackend {
                                new = false
                                break
                        }
                }
                if new {
                        localEp := sp.getLocalEndpoint(newBackend)
                        localEp.svcCount++
                        if localEp.ifName != "" && localEp.svcCount == 1 {
                                newBackendIfs.Add(localEp.ifName)
                                updateBackends = true
                        }
                }
        }
        // -> handle removed backend interfaces
        for _, oldBackend := range oldBackends {
                removed := true
                for _, newBackend := range newBackends {
                        if newBackend == oldBackend {
                                removed = false
                                break
                        }
                }
                if removed {
                        localEp := sp.getLocalEndpoint(oldBackend)
                        localEp.svcCount--
                        if localEp.ifName != "" && localEp.svcCount == 0 {
                                newBackendIfs.Del(localEp.ifName)
                                updateBackends = true
                        }
                }
        }
        // -> update local backends
        if updateBackends {
                for _, renderer := range sp.renderers {
                        err = renderer.UpdateLocalBackendIfs(sp.backendIfs, newBackendIfs)
                        if err != nil {
                                return err
                        }
                }
                sp.backendIfs = newBackendIfs
        }

        return err
}

// renderNodePorts re-renders all services with a node port.
func (sp *ServiceProcessor) renderNodePorts() error {
        sp.Log.Debug("ServiceProcessor - renderNodePorts()")

        var npServices []*renderer.ContivService
        for _, svc := range sp.services {
                contivSvc := svc.GetContivService()
                if contivSvc == nil {
                        continue
                }
                if contivSvc.HasNodePort() {
                        npServices = append(npServices, contivSvc)
                }
        }
        for _, renderer := range sp.renderers {
                err := renderer.UpdateNodePortServices(sp.getNodeIPs(), npServices)
                if err != nil {
                        return err
                }
        }
        return nil
}

// getNodeIPs returns a slice of IP addresses of all nodes in the cluster
// without duplicities.
func (sp *ServiceProcessor) getNodeIPs() *renderer.IPAddresses {
        nodeIPs := renderer.NewIPAddresses()
        addedNodeIPs := map[string]struct{}{}

        for _, node := range sp.nodes {
                // Node IP (VPP)
                ipAddr := sp.trimIPAddrPrefix(node.IpAddress)
                sp.Log.WithField("IPAddr", ipAddr).Debug("Node IP")
                if _, duplicate := addedNodeIPs[ipAddr]; !duplicate {
                        nodeIP := net.ParseIP(ipAddr)
                        if nodeIP != nil {
                                nodeIPs.Add(nodeIP)
                        }
                }
                addedNodeIPs[ipAddr] = struct{}{}
                // Node management IP (K8s, host)
                ipAddr = sp.trimIPAddrPrefix(node.ManagementIpAddress)
                sp.Log.WithField("IPAddr", ipAddr).Debug("Node mgmt IP")
                if _, duplicate := addedNodeIPs[ipAddr]; !duplicate {
                        nodeMgmtIP := net.ParseIP(ipAddr)
                        if nodeMgmtIP != nil {
                                nodeIPs.Add(nodeMgmtIP)
                        }
                }
                addedNodeIPs[ipAddr] = struct{}{}
        }

        return nodeIPs
}

func (sp *ServiceProcessor) trimIPAddrPrefix(ip string) string {
        if strings.Contains(ip, "/") {
                return ip[:strings.Index(ip, "/")]
        }
        return ip
}

func (sp *ServiceProcessor) processResyncEvent(resyncEv *ResyncEventData) error {
        sp.Log.WithFields(logging.Fields{
                "resyncEv": resyncEv,
        }).Debug("ServiceProcessor - processResyncEvent()")
        sp.reset()

        // Re-build the current state.
        confResyncEv := renderer.NewResyncEventData()

        // Replace the map of node IDs & IP addresses.
        sp.nodes = resyncEv.Nodes
        confResyncEv.NodeIPs = sp.getNodeIPs()

        // Fill up the set of frontend/backend interfaces and local endpoints.
        // With physical interfaces also build SNAT configuration.
        // -> VXLAN BVI interface
        vxlanBVIIf := sp.Contiv.GetVxlanBVIIfName()
        if vxlanBVIIf != "" {
                sp.frontendIfs.Add(vxlanBVIIf)
                sp.backendIfs.Add(vxlanBVIIf)
        }
        // -> main physical interfaces
        mainPhysIf := sp.Contiv.GetMainPhysicalIfName()
        if mainPhysIf != "" {
                if vxlanBVIIf == "" {
                        sp.backendIfs.Add(mainPhysIf)
                }
                sp.frontendIfs.Add(mainPhysIf)
        }
        // -> other physical interfaces
        for _, physIf := range sp.Contiv.GetOtherPhysicalIfNames() {
                sp.frontendIfs.Add(physIf)
        }
        // -> host interconnect
        hostInterconnect := sp.Contiv.GetHostInterconnectIfName()
        if hostInterconnect != "" {
                sp.frontendIfs.Add(hostInterconnect)
                sp.backendIfs.Add(hostInterconnect)
        }
        // -> pods
        for _, pod := range resyncEv.Pods {
                podID := podmodel.ID{Name: pod.Name, Namespace: pod.Namespace}
                if pod.IpAddress == "" {
                        continue
                }
                podIPAddress := net.ParseIP(pod.IpAddress)
                if podIPAddress == nil || !sp.Contiv.GetPodNetwork().Contains(podIPAddress) {
                        continue
                }

                // -> pod interface
                ifName, ifExists := sp.Contiv.GetIfName(podID.Namespace, podID.Name)
                if !ifExists {
                        sp.Log.WithFields(logging.Fields{
                                "pod-ns":   podID.Namespace,
                                "pod-name": podID.Name,
                        }).Warn("Failed to get pod interface name")
                        continue
                }
                localEp := sp.getLocalEndpoint(podID)
                localEp.ifName = ifName
                sp.frontendIfs.Add(ifName)
        }

        // Combine the service metadata with endpoints.
        for _, eps := range resyncEv.Endpoints {
                svcID := svcmodel.ID{Namespace: eps.Namespace, Name: eps.Name}
                svc := sp.getService(svcID)
                svc.SetEndpoints(eps)
        }
        for _, service := range resyncEv.Services {
                svcID := svcmodel.ID{Namespace: service.Namespace, Name: service.Name}
                svc := sp.getService(svcID)
                svc.SetMetadata(service)
        }

        // Iterate over services with complete data to get backend interfaces.
        for _, svc := range sp.services {
                contivSvc := svc.GetContivService()
                backends := svc.GetLocalBackends()
                if contivSvc == nil {
                        continue
                }
                confResyncEv.Services = append(confResyncEv.Services, contivSvc)
                for _, backend := range backends {
                        localEp := sp.getLocalEndpoint(backend)
                        localEp.svcCount++
                        if localEp.ifName != "" {
                                sp.backendIfs.Add(localEp.ifName)
                        }
                }
        }

        // Build resync data for service renderers.
        confResyncEv.FrontendIfs = sp.frontendIfs
        confResyncEv.BackendIfs = sp.backendIfs
        for _, renderer := range sp.renderers {
                if err := renderer.Resync(confResyncEv); err != nil {
                        return err
                }
        }
        return nil
}

// Close deallocates resource held by the processor.
func (sp *ServiceProcessor) Close() error {
        return nil
}

/**** Helper methods ****/

func (sp *ServiceProcessor) getService(svcID svcmodel.ID) *Service {
        _, hasEntry := sp.services[svcID]
        if !hasEntry {
                sp.services[svcID] = NewService(sp)
        }
        return sp.services[svcID]
}

func (sp *ServiceProcessor) getLocalEndpoint(podID podmodel.ID) *LocalEndpoint {
        _, hasEntry := sp.localEps[podID]
        if !hasEntry {
                sp.localEps[podID] = &LocalEndpoint{}
        }
        return sp.localEps[podID]
}

1	/*
2	* // Copyright (c) 2018 Cisco and/or its affiliates.
3	* //
4	* // Licensed under the Apache License, Version 2.0 (the "License");
5	* // you may not use this file except in compliance with the License.
6	* // You may obtain a copy of the License at:
7	* //
8	* // http://www.apache.org/licenses/LICENSE-2.0
9	* //
10	* // Unless required by applicable law or agreed to in writing, software
11	* // distributed under the License is distributed on an "AS IS" BASIS,
12	* // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* // See the License for the specific language governing permissions and
14	* // limitations under the License.
15	*/
16
17	package processor
18
19	import (
20	"net"
21	"strings"
22
23	"github.com/ligato/cn-infra/datasync"
24	"github.com/ligato/cn-infra/logging"
25	"github.com/ligato/cn-infra/servicelabel"
26
27	"github.com/contiv/vpp/plugins/contiv"
28	nodemodel "github.com/contiv/vpp/plugins/contiv/model/node"
29	epmodel "github.com/contiv/vpp/plugins/ksr/model/endpoints"
30	podmodel "github.com/contiv/vpp/plugins/ksr/model/pod"
31	svcmodel "github.com/contiv/vpp/plugins/ksr/model/service"
32	"github.com/contiv/vpp/plugins/service/renderer"
33	)
34
35	// ServiceProcessor implements ServiceProcessorAPI.
36	type ServiceProcessor struct {
37	Deps
38
39	renderers []renderer.ServiceRendererAPI
40
41	/* nodes */
42	nodes map[int]*nodemodel.NodeInfo
43
44	/* internal maps */
45	services map[svcmodel.ID]*Service
46	localEps map[podmodel.ID]*LocalEndpoint
47
48	/* local frontend and backend interfaces */
49	frontendIfs renderer.Interfaces
50	backendIfs renderer.Interfaces
51	}
52
53	// Deps lists dependencies of ServiceProcessor.
54	type Deps struct {
55	Log logging.Logger
56	ServiceLabel servicelabel.ReaderAPI
57	Contiv contiv.API /* to get all interface names and pod IP network */
58	}
59
60	// LocalEndpoint represents a node-local endpoint.
61	type LocalEndpoint struct {
62	ifName string
63	svcCount int /* number of services running on this endpoint. */
64	}
65
66	// Init initializes service processor.
67	func (sp *ServiceProcessor) Init() error {	8✔
68	sp.reset()	8✔
69	sp.Contiv.RegisterPodPreRemovalHook(sp.processDeletingPod)	8✔
70	return nil	8✔
71	}	8✔
72
73	// AfterInit does nothing for the processor.
74	func (sp *ServiceProcessor) AfterInit() error {	×
75	return nil	×
76	}	×
77
78	// reset clears the state of the processor.
79	func (sp *ServiceProcessor) reset() error {	17✔
80	sp.nodes = make(map[int]*nodemodel.NodeInfo)	17✔
81	sp.services = make(map[svcmodel.ID]*Service)	17✔
82	sp.localEps = make(map[podmodel.ID]*LocalEndpoint)	17✔
83	sp.frontendIfs = renderer.NewInterfaces()	17✔
84	sp.backendIfs = renderer.NewInterfaces()	17✔
85	return nil	17✔
86	}	17✔
87
88	// Update processes a datasync change event associated with the state data
89	// of K8s pods, endpoints, services and nodes.
90	// The data change is stored into the cache and all registered renderers
91	// are notified about any changes related to services that need to be
92	// reflected in the underlying network stack(s).
93	func (sp *ServiceProcessor) Update(dataChngEv datasync.ChangeEvent) error {	25✔
94	return sp.propagateDataChangeEv(dataChngEv)	25✔
95	}	25✔
96
97	// Resync processes a datasync resync event associated with the state data
98	// of K8s pods, endpoints, services and nodes.
99	// The cache content is fully replaced and all registered renderers
100	// receive a full snapshot of Contiv Services at the present state to be
101	// (re)installed.
102	func (sp *ServiceProcessor) Resync(resyncEv datasync.ResyncEvent) error {	9✔
103	resyncEvData := sp.parseResyncEv(resyncEv)	9✔
104	return sp.processResyncEvent(resyncEvData)	9✔
105	}	9✔
106
107	// RegisterRenderer registers a new service renderer.
108	// The renderer will be receiving updates for all services on the cluster.
109	func (sp *ServiceProcessor) RegisterRenderer(renderer renderer.ServiceRendererAPI) error {	8✔
110	sp.renderers = append(sp.renderers, renderer)	8✔
111	return nil	8✔
112	}	8✔
113
114	func (sp ServiceProcessor) processUpdatedPod(pod podmodel.Pod) error {	8✔
115	sp.Log.WithFields(logging.Fields{	8✔
116	"pod": *pod,	8✔
117	}).Debug("ServiceProcessor - processUpdatedPod()")	8✔
118	podID := podmodel.ID{Name: pod.Name, Namespace: pod.Namespace}	8✔
119		8✔
120	if pod.IpAddress == "" {	8✔
121	return nil	×
122	}	×
123	podIPAddress := net.ParseIP(pod.IpAddress)	8✔
124	if podIPAddress == nil \|\| !sp.Contiv.GetPodNetwork().Contains(podIPAddress) {	10✔
125	/* ignore pods deployed on other nodes */	2✔
126	return nil	2✔
127	}	2✔
128
129	localEp := sp.getLocalEndpoint(podID)	6✔
130	if localEp.ifName != "" {	6✔
131	/* already processed */	×
132	return nil	×
133	}	×
134	ifName, ifExists := sp.Contiv.GetIfName(podID.Namespace, podID.Name)	6✔
135	if !ifExists {	6✔
136	sp.Log.WithFields(logging.Fields{	×
137	"pod-ns": podID.Namespace,	×
138	"pod-name": podID.Name,	×
139	}).Warn("Failed to get pod interface name")	×
140	return nil	×
141	}	×
142
143	localEp.ifName = ifName	6✔
144	if localEp.svcCount > 0 {	6✔
145	newBackendIfs := sp.backendIfs.Copy()	×
146	newBackendIfs.Add(ifName)	×
147	for _, renderer := range sp.renderers {	×
148	err := renderer.UpdateLocalBackendIfs(sp.backendIfs, newBackendIfs)	×
149	if err != nil {	×
150	return err	×
151	}	×
152	}
153	sp.backendIfs = newBackendIfs	×
154	}
155	newFrontendIfs := sp.frontendIfs.Copy()	6✔
156	newFrontendIfs.Add(ifName)	6✔
157	for _, renderer := range sp.renderers {	12✔
158	err := renderer.UpdateLocalFrontendIfs(sp.frontendIfs, newFrontendIfs)	6✔
159	if err != nil {	6✔
160	return err	×
161	}	×
162	}
163	sp.frontendIfs = newFrontendIfs	6✔
164	return nil	6✔
165	}
166
167	func (sp *ServiceProcessor) processDeletingPod(podNamespace string, podName string) error {	2✔
168	podID := podmodel.ID{Name: podName, Namespace: podNamespace}	2✔
169	sp.Log.WithFields(logging.Fields{	2✔
170	"podID": podID,	2✔
171	}).Debug("ServiceProcessor - processDeletingPod()")	2✔
172		2✔
173	localEp, hasEntry := sp.localEps[podID]	2✔
174	if !hasEntry {	2✔
175	return nil	×
176	}	×
177	ifName := sp.localEps[podID].ifName	2✔
178	if ifName == "" {	2✔
179	return nil	×
180	}	×
181
182	if localEp.svcCount > 0 {	4✔
183	newBackendIfs := sp.backendIfs.Copy()	2✔
184	newBackendIfs.Del(ifName)	2✔
185	for _, renderer := range sp.renderers {	4✔
186	/* ignore errors */	2✔
187	renderer.UpdateLocalBackendIfs(sp.backendIfs, newBackendIfs)	2✔
188	}	2✔
189	sp.backendIfs = newBackendIfs	2✔
190	}
191	newFrontendIfs := sp.frontendIfs.Copy()	2✔
192	newFrontendIfs.Del(ifName)	2✔
193	for _, renderer := range sp.renderers {	4✔
194	/* ignore errors */	2✔
195	renderer.UpdateLocalFrontendIfs(sp.frontendIfs, newFrontendIfs)	2✔
196	}	2✔
197	sp.frontendIfs = newFrontendIfs	2✔
198	delete(sp.localEps, podID)	2✔
199	return nil	2✔
200	}
201
202	func (sp ServiceProcessor) processNewEndpoints(eps epmodel.Endpoints) error {	4✔
203	sp.Log.WithFields(logging.Fields{	4✔
204	"eps": *eps,	4✔
205	}).Debug("ServiceProcessor - processNewEndpoints()")	4✔
206		4✔
207	svcID := svcmodel.ID{Namespace: eps.Namespace, Name: eps.Name}	4✔
208	svc := sp.getService(svcID)	4✔
209	svc.SetEndpoints(eps)	4✔
210	return sp.renderService(svc, nil, []podmodel.ID{})	4✔
211	}	4✔
212
213	func (sp ServiceProcessor) processUpdatedEndpoints(eps epmodel.Endpoints) error {	2✔
214	sp.Log.WithFields(logging.Fields{	2✔
215	"eps": *eps,	2✔
216	}).Debug("ServiceProcessor - processUpdatedEndpoints()")	2✔
217		2✔
218	svcID := svcmodel.ID{Namespace: eps.Namespace, Name: eps.Name}	2✔
219	svc := sp.getService(svcID)	2✔
220	oldContivSvc := svc.GetContivService()	2✔
221	oldBackends := svc.GetLocalBackends()	2✔
222	svc.SetEndpoints(eps)	2✔
223	return sp.renderService(svc, oldContivSvc, oldBackends)	2✔
224	}	2✔
225
226	func (sp *ServiceProcessor) processDeletedEndpoints(epsID epmodel.ID) error {	1✔
227	sp.Log.WithFields(logging.Fields{	1✔
228	"epsID": epsID,	1✔
229	}).Debug("ServiceProcessor - processDeletedEndpoints()")	1✔
230		1✔
231	svcID := svcmodel.ID{Namespace: epsID.Namespace, Name: epsID.Name}	1✔
232	svc := sp.getService(svcID)	1✔
233	oldContivSvc := svc.GetContivService()	1✔
234	oldBackends := svc.GetLocalBackends()	1✔
235	svc.SetEndpoints(nil)	1✔
236	return sp.renderService(svc, oldContivSvc, oldBackends)	1✔
237	}	1✔
238
239	func (sp ServiceProcessor) processNewService(service svcmodel.Service) error {	4✔
240	sp.Log.WithFields(logging.Fields{	4✔
241	"service": *service,	4✔
242	}).Debug("ServiceProcessor - processNewService()")	4✔
243		4✔
244	svcID := svcmodel.ID{Namespace: service.Namespace, Name: service.Name}	4✔
245	svc := sp.getService(svcID)	4✔
246	svc.SetMetadata(service)	4✔
247	return sp.renderService(svc, nil, []podmodel.ID{})	4✔
248	}	4✔
249
250	func (sp ServiceProcessor) processUpdatedService(service svcmodel.Service) error {	1✔
251	sp.Log.WithFields(logging.Fields{	1✔
252	"service": *service,	1✔
253	}).Debug("ServiceProcessor - processUpdatedService()")	1✔
254		1✔
255	svcID := svcmodel.ID{Namespace: service.Namespace, Name: service.Name}	1✔
256	svc := sp.getService(svcID)	1✔
257	oldContivSvc := svc.GetContivService()	1✔
258	oldBackends := svc.GetLocalBackends()	1✔
259	svc.SetMetadata(service)	1✔
260	return sp.renderService(svc, oldContivSvc, oldBackends)	1✔
261	}	1✔
262
263	func (sp *ServiceProcessor) processDeletedService(serviceID svcmodel.ID) error {	1✔
264	sp.Log.WithFields(logging.Fields{	1✔
265	"serviceID": serviceID,	1✔
266	}).Debug("ServiceProcessor - processDeletedService()")	1✔
267		1✔
268	svcID := svcmodel.ID{Namespace: serviceID.Namespace, Name: serviceID.Name}	1✔
269	svc := sp.getService(svcID)	1✔
270	oldContivSvc := svc.GetContivService()	1✔
271	oldBackends := svc.GetLocalBackends()	1✔
272	svc.SetMetadata(nil)	1✔
273	return sp.renderService(svc, oldContivSvc, oldBackends)	1✔
274	}	1✔
275
276	func (sp ServiceProcessor) processNewNode(node nodemodel.NodeInfo) error {	2✔
277	sp.Log.WithFields(logging.Fields{	2✔
278	"node": *node,	2✔
279	}).Debug("ServiceProcessor - processNewNode()")	2✔
280		2✔
281	sp.nodes[int(node.Id)] = node	2✔
282	return sp.renderNodePorts()	2✔
283	}	2✔
284
285	func (sp ServiceProcessor) processUpdatedNode(node nodemodel.NodeInfo) error {	1✔
286	sp.Log.WithFields(logging.Fields{	1✔
287	"node": *node,	1✔
288	}).Debug("ServiceProcessor - processUpdatedNode()")	1✔
289		1✔
290	sp.nodes[int(node.Id)] = node	1✔
291	return sp.renderNodePorts()	1✔
292	}	1✔
293
294	func (sp *ServiceProcessor) processDeletedNode(nodeID int) error {	1✔
295	sp.Log.WithFields(logging.Fields{	1✔
296	"nodeID": nodeID,	1✔
297	}).Debug("ServiceProcessor - processDeletedNode()")	1✔
298		1✔
299	if _, hasNode := sp.nodes[nodeID]; hasNode {	2✔
300	delete(sp.nodes, nodeID)	1✔
301	return sp.renderNodePorts()	1✔
302	}	1✔
303	return nil	×
304	}
305
306	// renderService reacts to added/removed/changed service.
307	func (sp ServiceProcessor) renderService(svc Service, oldContivSvc *renderer.ContivService,
308	oldBackends []podmodel.ID) error {	13✔
309		13✔
310	var err error	13✔
311	newContivSvc := svc.GetContivService()	13✔
312	newBackends := svc.GetLocalBackends()	13✔
313		13✔
314	// Render service.	13✔
315	if newContivSvc != nil {	20✔
316	if oldContivSvc == nil {	11✔
317	for _, renderer := range sp.renderers {	8✔
318	if err = renderer.AddService(newContivSvc); err != nil {	4✔
319	return err	×
320	}	×
321	}
322	} else {	3✔
323	for _, renderer := range sp.renderers {	6✔
324	if err = renderer.UpdateService(oldContivSvc, newContivSvc); err != nil {	3✔
325	return err	×
326	}	×
327	}
328	}
329	} else {	6✔
330	if oldContivSvc != nil {	8✔
331	for _, renderer := range sp.renderers {	4✔
332	if err = renderer.DeleteService(oldContivSvc); err != nil {	2✔
333	return err	×
334	}	×
335	}
336	}
337	}
338
339	// Render local Backends.
340	newBackendIfs := sp.backendIfs.Copy()	13✔
341	updateBackends := false	13✔
342	// -> handle new backend interfaces	13✔
343	for _, newBackend := range newBackends {	24✔
344	new := true	11✔
345	for _, oldBackend := range oldBackends {	16✔
346	if newBackend == oldBackend {	9✔
347	new = false	4✔
348	break	4✔
349	}
350	}
351	if new {	18✔
352	localEp := sp.getLocalEndpoint(newBackend)	7✔
353	localEp.svcCount++	7✔
354	if localEp.ifName != "" && localEp.svcCount == 1 {	13✔
355	newBackendIfs.Add(localEp.ifName)	6✔
356	updateBackends = true	6✔
357	}	6✔
358	}
359	}
360	// -> handle removed backend interfaces
361	for _, oldBackend := range oldBackends {	21✔
362	removed := true	8✔
363	for _, newBackend := range newBackends {	14✔
364	if newBackend == oldBackend {	10✔
365	removed = false	4✔
366	break	4✔
367	}
368	}
369	if removed {	12✔
370	localEp := sp.getLocalEndpoint(oldBackend)	4✔
371	localEp.svcCount--	4✔
372	if localEp.ifName != "" && localEp.svcCount == 0 {	6✔
373	newBackendIfs.Del(localEp.ifName)	2✔
374	updateBackends = true	2✔
375	}	2✔
376	}
377	}
378	// -> update local backends
379	if updateBackends {	17✔
380	for _, renderer := range sp.renderers {	8✔
381	err = renderer.UpdateLocalBackendIfs(sp.backendIfs, newBackendIfs)	4✔
382	if err != nil {	4✔
383	return err	×
384	}	×
385	}
386	sp.backendIfs = newBackendIfs	4✔
387	}
388
389	return err	13✔
390	}
391
392	// renderNodePorts re-renders all services with a node port.
393	func (sp *ServiceProcessor) renderNodePorts() error {	4✔
394	sp.Log.Debug("ServiceProcessor - renderNodePorts()")	4✔
395		4✔
396	var npServices []*renderer.ContivService	4✔
397	for _, svc := range sp.services {	12✔
398	contivSvc := svc.GetContivService()	8✔
399	if contivSvc == nil {	8✔
400	continue	×
401	}
402	if contivSvc.HasNodePort() {	12✔
403	npServices = append(npServices, contivSvc)	4✔
404	}	4✔
405	}
406	for _, renderer := range sp.renderers {	8✔
407	err := renderer.UpdateNodePortServices(sp.getNodeIPs(), npServices)	4✔
408	if err != nil {	4✔
409	return err	×
410	}	×
411	}
412	return nil	4✔
413	}
414
415	// getNodeIPs returns a slice of IP addresses of all nodes in the cluster
416	// without duplicities.
417	func (sp ServiceProcessor) getNodeIPs() renderer.IPAddresses {	13✔
418	nodeIPs := renderer.NewIPAddresses()	13✔
419	addedNodeIPs := map[string]struct{}{}	13✔
420		13✔
421	for _, node := range sp.nodes {	21✔
422	// Node IP (VPP)	8✔
423	ipAddr := sp.trimIPAddrPrefix(node.IpAddress)	8✔
424	sp.Log.WithField("IPAddr", ipAddr).Debug("Node IP")	8✔
425	if _, duplicate := addedNodeIPs[ipAddr]; !duplicate {	16✔
426	nodeIP := net.ParseIP(ipAddr)	8✔
427	if nodeIP != nil {	16✔
428	nodeIPs.Add(nodeIP)	8✔
429	}	8✔
430	}
431	addedNodeIPs[ipAddr] = struct{}{}	8✔
432	// Node management IP (K8s, host)	8✔
433	ipAddr = sp.trimIPAddrPrefix(node.ManagementIpAddress)	8✔
434	sp.Log.WithField("IPAddr", ipAddr).Debug("Node mgmt IP")	8✔
435	if _, duplicate := addedNodeIPs[ipAddr]; !duplicate {	16✔
436	nodeMgmtIP := net.ParseIP(ipAddr)	8✔
437	if nodeMgmtIP != nil {	15✔
438	nodeIPs.Add(nodeMgmtIP)	7✔
439	}	7✔
440	}
441	addedNodeIPs[ipAddr] = struct{}{}	8✔
442	}
443
444	return nodeIPs	13✔
445	}
446
447	func (sp *ServiceProcessor) trimIPAddrPrefix(ip string) string {	16✔
448	if strings.Contains(ip, "/") {	24✔
449	return ip[:strings.Index(ip, "/")]	8✔
450	}	8✔
451	return ip	8✔
452	}
453
454	func (sp ServiceProcessor) processResyncEvent(resyncEv ResyncEventData) error {	9✔
455	sp.Log.WithFields(logging.Fields{	9✔
456	"resyncEv": resyncEv,	9✔
457	}).Debug("ServiceProcessor - processResyncEvent()")	9✔
458	sp.reset()	9✔
459		9✔
460	// Re-build the current state.	9✔
461	confResyncEv := renderer.NewResyncEventData()	9✔
462		9✔
463	// Replace the map of node IDs & IP addresses.	9✔
464	sp.nodes = resyncEv.Nodes	9✔
465	confResyncEv.NodeIPs = sp.getNodeIPs()	9✔
466		9✔
467	// Fill up the set of frontend/backend interfaces and local endpoints.	9✔
468	// With physical interfaces also build SNAT configuration.	9✔
469	// -> VXLAN BVI interface	9✔
470	vxlanBVIIf := sp.Contiv.GetVxlanBVIIfName()	9✔
471	if vxlanBVIIf != "" {	17✔
472	sp.frontendIfs.Add(vxlanBVIIf)	8✔
473	sp.backendIfs.Add(vxlanBVIIf)	8✔
474	}	8✔
475	// -> main physical interfaces
476	mainPhysIf := sp.Contiv.GetMainPhysicalIfName()	9✔
477	if mainPhysIf != "" {	18✔
478	if vxlanBVIIf == "" {	10✔
479	sp.backendIfs.Add(mainPhysIf)	1✔
480	}	1✔
481	sp.frontendIfs.Add(mainPhysIf)	9✔
482	}
483	// -> other physical interfaces
484	for _, physIf := range sp.Contiv.GetOtherPhysicalIfNames() {	11✔
485	sp.frontendIfs.Add(physIf)	2✔
486	}	2✔
487	// -> host interconnect
488	hostInterconnect := sp.Contiv.GetHostInterconnectIfName()	9✔
489	if hostInterconnect != "" {	18✔
490	sp.frontendIfs.Add(hostInterconnect)	9✔
491	sp.backendIfs.Add(hostInterconnect)	9✔
492	}	9✔
493	// -> pods
494	for _, pod := range resyncEv.Pods {	17✔
495	podID := podmodel.ID{Name: pod.Name, Namespace: pod.Namespace}	8✔
496	if pod.IpAddress == "" {	8✔
497	continue	×
498	}
499	podIPAddress := net.ParseIP(pod.IpAddress)	8✔
500	if podIPAddress == nil \|\| !sp.Contiv.GetPodNetwork().Contains(podIPAddress) {	10✔
501	continue	2✔
502	}
503
504	// -> pod interface
505	ifName, ifExists := sp.Contiv.GetIfName(podID.Namespace, podID.Name)	6✔
506	if !ifExists {	6✔
507	sp.Log.WithFields(logging.Fields{	×
508	"pod-ns": podID.Namespace,	×
509	"pod-name": podID.Name,	×
510	}).Warn("Failed to get pod interface name")	×
511	continue	×
512	}
513	localEp := sp.getLocalEndpoint(podID)	6✔
514	localEp.ifName = ifName	6✔
515	sp.frontendIfs.Add(ifName)	6✔
516	}
517
518	// Combine the service metadata with endpoints.
519	for _, eps := range resyncEv.Endpoints {	14✔
520	svcID := svcmodel.ID{Namespace: eps.Namespace, Name: eps.Name}	5✔
521	svc := sp.getService(svcID)	5✔
522	svc.SetEndpoints(eps)	5✔
523	}	5✔
524	for _, service := range resyncEv.Services {	12✔
525	svcID := svcmodel.ID{Namespace: service.Namespace, Name: service.Name}	3✔
526	svc := sp.getService(svcID)	3✔
527	svc.SetMetadata(service)	3✔
528	}	3✔
529
530	// Iterate over services with complete data to get backend interfaces.
531	for _, svc := range sp.services {	14✔
532	contivSvc := svc.GetContivService()	5✔
533	backends := svc.GetLocalBackends()	5✔
534	if contivSvc == nil {	7✔
535	continue	2✔
536	}
537	confResyncEv.Services = append(confResyncEv.Services, contivSvc)	3✔
538	for _, backend := range backends {	8✔
539	localEp := sp.getLocalEndpoint(backend)	5✔
540	localEp.svcCount++	5✔
541	if localEp.ifName != "" {	10✔
542	sp.backendIfs.Add(localEp.ifName)	5✔
543	}	5✔
544	}
545	}
546
547	// Build resync data for service renderers.
548	confResyncEv.FrontendIfs = sp.frontendIfs	9✔
549	confResyncEv.BackendIfs = sp.backendIfs	9✔
550	for _, renderer := range sp.renderers {	18✔
551	if err := renderer.Resync(confResyncEv); err != nil {	9✔
552	return err	×
553	}	×
554	}
555	return nil	9✔
556	}
557
558	// Close deallocates resource held by the processor.
559	func (sp *ServiceProcessor) Close() error {	6✔
560	return nil	6✔
561	}	6✔
562
563	/** Helper methods **/
564
565	func (sp ServiceProcessor) getService(svcID svcmodel.ID) Service {	21✔
566	_, hasEntry := sp.services[svcID]	21✔
567	if !hasEntry {	30✔
568	sp.services[svcID] = NewService(sp)	9✔
569	}	9✔
570	return sp.services[svcID]	21✔
571	}
572
573	func (sp ServiceProcessor) getLocalEndpoint(podID podmodel.ID) LocalEndpoint {	28✔
574	_, hasEntry := sp.localEps[podID]	28✔
575	if !hasEntry {	42✔
576	sp.localEps[podID] = &LocalEndpoint{}	14✔
577	}	14✔
578	return sp.localEps[podID]	28✔
579	}

contiv / vpp / 1801

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