11995

Committed 29 May 2026 05:52AM UTC coverage: 63.172% (-0.04%) from 63.215%

Build # 11995

Build Type

Pull #1736

travis-pro

Committed by

web-flow

Commit Message

Merge e7bf1ba4e into 1654f3a56

Pull Request Pull Request #1736: Bump go version to 1.26.3 in .travis.yml

Coverage Stats

13318 of 21082 relevant lines covered (63.17%)

0.72 hits per line

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

63.51

/pkg/apicapi/apicapi.go

// Copyright 2017 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.

// Interface for connecting to APIC REST API using websockets
package apicapi

import (
        "bytes"
        "crypto/tls"
        "crypto/x509"
        "encoding/json"
        "errors"
        "fmt"
        "io"
        "math/rand"
        "net/http"
        "net/http/cookiejar"
        "regexp"
        "sort"
        "strconv"
        "strings"
        "time"

        "k8s.io/apimachinery/pkg/util/wait"
        "k8s.io/client-go/util/workqueue"

        "github.com/gorilla/websocket"
        "github.com/noironetworks/aci-containers/pkg/util"
        "github.com/sirupsen/logrus"
        "golang.org/x/time/rate"
)

const (
        // defaultConnectionRefresh is used as connection refresh interval if
        // RefreshInterval is set to 0
        defaultConnectionRefresh = 30 * time.Second
        cacheStaleAfterTime      = 60 * time.Minute
        maxRequestRetry          = 5
)

// ApicVersion - This global variable to be used when dealing with version-
// dependencies during APIC interaction. It gets filled with actual version
// as part of runConn()
var (
        ApicVersion = "3.1"
)

func complete(resp *http.Response) {
        if resp.StatusCode != http.StatusOK {
                rBody, err := io.ReadAll(resp.Body)
                if err != nil {
                        logrus.Errorf("ReadAll :%v", err)
                } else {
                        logrus.Infof("Resp: %s", rBody)
                }
        }
        resp.Body.Close()
}

// Yes, this is really stupid, but this is really how this works
func (conn *ApicConnection) sign(req *http.Request, uri string, body []byte) {
        if conn.signer == nil {
                return
        }

        sig, err := conn.signer.sign(req.Method, uri, body)
        if err != nil {
                conn.log.Error("Failed to sign request: ", err)
                return
        }

        req.Header.Set("Cookie", conn.apicSigCookie(sig, conn.token))
}

func (conn *ApicConnection) apicSigCookie(sig, token string) string {
        tokc := ""
        if token != "" {
                tokc = "; APIC-WebSocket-Session=" + token
        }
        return fmt.Sprintf("APIC-Request-Signature=%s; "+
                "APIC-Certificate-Algorithm=v1.0; "+
                "APIC-Certificate-DN=uni/userext/user-%s/usercert-%s.crt; "+
                "APIC-Certificate-Fingerprint=fingerprint%s",
                sig, conn.user, conn.user, tokc)
}

func (conn *ApicConnection) login() (string, error) {
        var path string
        var method string

        if conn.signer == nil {
                path = "aaaLogin"
                method = "POST"
        } else {
                path = "webtokenSession"
                method = "GET"
        }
        uri := fmt.Sprintf("/api/%s.json", path)
        var raw []byte
        var err error
        if conn.signer == nil {
                login := &ApicObject{
                        "aaaUser": &ApicObjectBody{
                                Attributes: map[string]interface{}{
                                        "name": conn.user,
                                        "pwd":  conn.password,
                                },
                        },
                }
                raw, err = json.Marshal(login)
                if err != nil {
                        return "", err
                }
        }
        conn.log.Infof("Req: %+v", &struct{ Method, URI string }{method, uri})
        resp, err := conn.sendHTTPSRequestToAPIC(method, uri, raw, "application/json", false, nil)
        if err != nil {
                return "", err
        }
        defer complete(resp)

        var apicresp ApicResponse
        err = json.NewDecoder(resp.Body).Decode(&apicresp)
        if err != nil {
                return "", err
        }

        for _, obj := range apicresp.Imdata {
                lresp, ok := obj["aaaLogin"]
                if !ok {
                        lresp, ok = obj["webtokenSession"]
                        if !ok {
                                continue
                        }
                }

                token, ok := lresp.Attributes["token"]
                if !ok {
                        return "", errors.New("Token not found in login response")
                }
                stoken, isStr := token.(string)
                if !isStr {
                        return "", errors.New("Token is not a string")
                }
                return stoken, nil
        }
        return "", errors.New("Login response not found")
}

func configureTls(cert []byte) (*tls.Config, error) {
        if cert == nil {
                return &tls.Config{InsecureSkipVerify: true}, nil
        }
        pool := x509.NewCertPool()
        if !pool.AppendCertsFromPEM(cert) {
                return nil, errors.New("Could not load CA certificates")
        }
        return &tls.Config{RootCAs: pool}, nil
}

func New(log *logrus.Logger, apic []string, user string,
        password string, privKey []byte, cert []byte,
        prefix string, refresh int, refreshTickerAdjust int,
        leafRebootCheckInterval int, subscriptionDelay int, vrfTenant string,
        lldpIfHldr func(dn, lldpIf string) bool, cnoEnabled bool) (*ApicConnection, error) {
        tls, err := configureTls(cert)
        if err != nil {
                return nil, err
        }

        var signer *signer
        if privKey != nil {
                signer, err = newSigner(privKey)
                if err != nil {
                        return nil, err
                }
        }

        dialer := &websocket.Dialer{
                TLSClientConfig: tls,
        }
        tr := &http.Transport{
                Proxy:           http.ProxyFromEnvironment,
                TLSClientConfig: dialer.TLSClientConfig,
        }
        jar, err := cookiejar.New(nil)
        if err != nil {
                return nil, err
        }
        client := &http.Client{
                Transport: tr,
                Jar:       jar,
                Timeout:   5 * time.Minute,
        }

        // When prefix is very long, AciNameForKey produces a pure hash with no
        // prefix embedded, making DN-based prefix filtering impossible.
        prefixFilterable := len(prefix) <= util.AciPrefixMaxLen

        conn := &ApicConnection{
                ReconnectInterval:       time.Duration(5) * time.Second,
                ReconnectRetryLimit:     5,
                RefreshInterval:         time.Duration(refresh) * time.Second,
                LeafRebootCheckInterval: time.Duration(leafRebootCheckInterval) * time.Second,
                RefreshTickerAdjust:     time.Duration(refreshTickerAdjust) * time.Second,
                SubscriptionDelay:       time.Duration(subscriptionDelay) * time.Millisecond,

                SyncDone:         false,
                signer:           signer,
                dialer:           dialer,
                logger:           log,
                log:              log.WithField("mod", "APICAPI"),
                Apic:             apic,
                user:             user,
                password:         password,
                prefix:           prefix,
                prefixFilterable: prefixFilterable,
                client:           client,
                vrfTenant:        vrfTenant,
                lldpIfHldr:       lldpIfHldr,
                cnoEnabled:       cnoEnabled,
                subscriptions: subIndex{
                        subs: make(map[string]*subscription),
                        ids:  make(map[string]string),
                },
                syncStore:          make(map[string]*syncObj),
                cacheOpflexOdev:    make(map[string]struct{}),
                desiredState:       make(map[string]ApicSlice),
                desiredStateDn:     make(map[string]ApicObject),
                keyHashes:          make(map[string]string),
                containerDns:       make(map[string]bool),
                cachedState:        make(map[string]ApicSlice),
                cacheDnSubIds:      make(map[string]map[string]bool),
                pendingSubDnUpdate: make(map[string]pendingChange),
                CachedSubnetDns:    make(map[string]string),
                cachedLLDPIfs:      make(map[string]string),
                cachedLeafDns:      make(map[string]*leafCacheEntry),
        }
        return conn, nil
}

func (conn *ApicConnection) addToQueue(queue workqueue.RateLimitingInterface, item interface{}) {
        if conn.cnoEnabled || conn.DisableRateLimit {
                queue.Add(item)
        } else {
                queue.AddRateLimited(item)
        }
}

func (conn *ApicConnection) handleSocketUpdate(apicresp *ApicResponse) {
        var subIds []string
        switch ids := apicresp.SubscriptionId.(type) {
        case string:
                subIds = append(subIds, ids)
        case []interface{}:
                for _, id := range ids {
                        subIds = append(subIds, id.(string))
                }
        }

        // Check if any subscription in this message has hooks (updateHook/deleteHook).
        // Subscriptions with hooks handle all notifications regardless of ownership
        // (e.g., fvSubnet class subscription tracking subnets across all BDs).
        // The prefix filter must be bypassed for these.
        hasHooks := false
        if !conn.cnoEnabled && conn.prefixFilterable && conn.vrfTenant != "" {
                conn.indexMutex.Lock()
                for _, id := range subIds {
                        if value, ok := conn.subscriptions.ids[id]; ok {
                                if sub, ok := conn.subscriptions.subs[value]; ok {
                                        if sub.updateHook != nil || sub.deleteHook != nil {
                                                hasHooks = true
                                                break
                                        }
                                }
                        }
                }
                conn.indexMutex.Unlock()
        }

        for _, obj := range apicresp.Imdata {
                for key, body := range obj {
                        if dn, ok := body.Attributes["dn"].(string); ok {
                                if status, isStr := body.Attributes["status"].(string); isStr {
                                        dnSlice := strings.Split(dn, "/")
                                        // Filter out notifications from other controllers sharing the same vrfTenant.
                                        // In normal mode, all MOs created by this controller have names generated via
                                        // AciNameForKey(prefix, ktype, key) which always produces "{prefix}_{ktype}_{...}",
                                        // so the DN will contain "{prefix}_" in one of its segments. Skip notifications
                                        // under our vrfTenant that don't carry our prefix — they belong to other controllers.
                                        // This filter is bypassed in CNO/chained mode, when AciNameForKey produces
                                        // pure-hash names (prefixFilterable is false), and when the subscription has
                                        // hooks that need to process all notifications regardless of ownership.
                                        if !hasHooks && !conn.cnoEnabled && conn.prefixFilterable &&
                                                len(dnSlice) > 1 && conn.vrfTenant != "" &&
                                                dnSlice[1] == "tn-"+conn.vrfTenant &&
                                                !strings.Contains(dn, "-"+conn.prefix+"_") {
                                                conn.log.Debug("Skipping websocket notification for dn: ", dn, " class: ", key)
                                                continue
                                        }
                                        var pendingKind int
                                        if status == "deleted" {
                                                pendingKind = pendingChangeDelete
                                        } else {
                                                pendingKind = pendingChangeUpdate
                                        }
                                        conn.indexMutex.Lock()

                                        conn.logger.WithFields(logrus.Fields{
                                                "mod": "APICAPI",
                                                "dn":  obj.GetDn(),
                                                "obj": obj,
                                        }).Debug("Processing websocket notification for:")

                                        conn.pendingSubDnUpdate[dn] = pendingChange{
                                                kind:    pendingKind,
                                                subIds:  subIds,
                                                isDirty: false,
                                        }
                                        if key == "opflexODev" && conn.odevQueue != nil {
                                                if conn.FilterOpflexDevice {
                                                        if conn.isPresentInOpflexOdevCache(dn, status, obj) {
                                                                conn.log.Info("Adding dn to odevQueue: ", dn)
                                                                conn.addToQueue(conn.odevQueue, dn)
                                                        }
                                                } else {
                                                        conn.log.Info("Adding dn to odevQueue: ", dn)
                                                        conn.addToQueue(conn.odevQueue, dn)
                                                }
                                        } else if isPriorityObject(dn) {
                                                conn.log.Debug("Adding dn to priorityQueue: ", dn)
                                                conn.addToQueue(conn.priorityQueue, dn)
                                        } else if isLLDPIfObject(dn) && conn.lldpIfQueue != nil {
                                                if lldpIf, ok := body.Attributes["portDesc"].(string); ok {
                                                        conn.cachedLLDPIfs[dn] = lldpIf
                                                }
                                                conn.log.Debug("Adding dn to lldpIfQueue: ", dn)
                                                conn.addToQueue(conn.lldpIfQueue, dn)
                                        } else if conn.deltaQueue != nil {
                                                conn.addToQueue(conn.deltaQueue, dn)
                                        }
                                        conn.indexMutex.Unlock()
                                }
                        }
                }
        }
}

func (conn *ApicConnection) restart() {
        conn.indexMutex.Lock()
        if conn.restartCh != nil {
                conn.log.Debug("Restarting connection")
                close(conn.restartCh)
                conn.restartCh = nil
        }
        conn.indexMutex.Unlock()
}

func (conn *ApicConnection) handleQueuedDn(dn string) bool {
        var respClasses []string
        var updateHandlers []ApicObjectHandler
        var deleteHandlers []ApicDnHandler
        var rootDn string

        handleId := func(id string) {
                conn.indexMutex.Lock()
                if value, ok := conn.subscriptions.ids[id]; ok {
                        if sub, ok := conn.subscriptions.subs[value]; ok {
                                if subComp, ok := sub.childSubs[id]; ok {
                                        respClasses =
                                                append(respClasses, subComp.respClasses...)
                                } else {
                                        respClasses =
                                                append(respClasses, sub.respClasses...)
                                }
                                if sub.updateHook != nil {
                                        updateHandlers = append(updateHandlers, sub.updateHook)
                                }
                                if sub.deleteHook != nil {
                                        deleteHandlers = append(deleteHandlers, sub.deleteHook)
                                }

                                if sub.kind == apicSubTree {
                                        rootDn = getRootDn(dn, value)
                                }
                        }
                } else {
                        conn.log.Warning("Unexpected subscription: ", id)
                }
                conn.indexMutex.Unlock()
        }

        var requeue bool
        conn.indexMutex.Lock()
        pending, hasPendingChange := conn.pendingSubDnUpdate[dn]
        conn.pendingSubDnUpdate[dn] = pendingChange{isDirty: true}
        obj, hasDesiredState := conn.desiredStateDn[dn]
        conn.indexMutex.Unlock()

        if hasPendingChange {
                for _, id := range pending.subIds {
                        handleId(id)
                }
        }

        if rootDn == "" {
                rootDn = dn
        }

        if hasDesiredState {
                if hasPendingChange {
                        if pending.kind == pendingChangeDelete {
                                conn.logger.WithFields(logrus.Fields{"mod": "APICAPI", "DN": dn}).
                                        Warning("Restoring unexpectedly deleted" +
                                                " ACI object")
                                requeue = conn.postDn(dn, obj)
                        } else {
                                conn.log.Debug("getSubtreeDn for:", rootDn)
                                conn.getSubtreeDn(rootDn, respClasses, updateHandlers)
                        }
                } else {
                        requeue = conn.postDn(dn, obj)
                }
        } else {
                if hasPendingChange {
                        if pending.kind == pendingChangeSyncDelete {
                                // Ownership already verified by sync/reconcile;
                                // delete without an extra APIC lookup.
                                requeue = conn.Delete(dn)
                        } else {
                                if pending.kind == pendingChangeDelete {
                                        for _, handler := range deleteHandlers {
                                                handler(dn)
                                        }
                                }

                                if (pending.kind != pendingChangeDelete) || (dn != rootDn) {
                                        conn.log.Debug("getSubtreeDn for:", rootDn)
                                        conn.getSubtreeDn(rootDn, respClasses, updateHandlers)
                                }
                        }
                } else {
                        if conn.isOwnedByController(dn) {
                                requeue = conn.Delete(dn)
                        } else {
                                conn.log.Debug("Skipping delete for unowned dn: ", dn)
                        }
                }
        }

        return requeue
}

func (conn *ApicConnection) processQueue(queue workqueue.RateLimitingInterface,
        queueStop <-chan struct{}, name string) {
        go wait.Until(func() {
                conn.log.Debug("Running processQueue for queue ", name)
                for {
                        dn, quit := queue.Get()
                        if quit {
                                break
                        }
                        conn.log.Debug("Processing queue for:", dn)
                        var requeue bool
                        if dn, ok := dn.(string); ok {
                                requeue = conn.handleQueuedDn(dn)
                        }
                        if requeue {
                                queue.AddRateLimited(dn)
                        } else {
                                conn.indexMutex.Lock()
                                if conn.pendingSubDnUpdate[dn.(string)].isDirty {
                                        delete(conn.pendingSubDnUpdate, dn.(string))
                                }
                                conn.indexMutex.Unlock()
                                queue.Forget(dn)
                        }
                        queue.Done(dn)
                }
        }, time.Second, queueStop)
        <-queueStop
        queue.ShutDown()
}

func (conn *ApicConnection) processLLDPIfQueue(queue workqueue.RateLimitingInterface,
        handler func(dn, lldpIf string) bool, stopCh <-chan struct{}) {
        go wait.Until(func() {
                for {
                        key, quit := queue.Get()
                        if quit {
                                break
                        }
                        var requeue bool
                        switch key := key.(type) {
                        case chan struct{}:
                                close(key)
                        case string:
                                if handler != nil {
                                        if lldpIf, ok := conn.cachedLLDPIfs[key]; ok {
                                                requeue = handler(key, lldpIf)
                                        }
                                }
                        }
                        if requeue {
                                queue.AddRateLimited(key)
                        } else {
                                queue.Forget(key)
                        }
                        queue.Done(key)

                }
        }, time.Second, stopCh)
        <-stopCh
        queue.ShutDown()
}

type fullSync struct{}

func (conn *ApicConnection) UnsubscribeImmediateDnLocked(dn string,
        targetClasses []string) {
        // Post local delete, subscription will not be refreshed
        // after refresh time-out
        delete(conn.subscriptions.subs, dn)
}

func (conn *ApicConnection) AddImmediateSubscriptionDnLocked(dn string,
        targetClasses []string, updateHook ApicObjectHandler, deleteHook ApicDnHandler) bool {
        conn.logger.WithFields(logrus.Fields{
                "mod": "APICAPI",
                "dn":  dn,
        }).Debug("Adding Subscription for Dn")

        conn.subscriptions.subs[dn] = &subscription{
                kind:          apicSubDn,
                childSubs:     make(map[string]subComponent),
                targetClasses: targetClasses,
                respClasses:   computeRespClasses(targetClasses),
        }
        if updateHook != nil {
                conn.subscriptions.subs[dn].updateHook = updateHook
        }
        if deleteHook != nil {
                conn.subscriptions.subs[dn].deleteHook = deleteHook
        }
        if conn.connection != nil {
                return conn.subscribe(dn, conn.subscriptions.subs[dn], true)
        } else {
                return false
        }
}

func (conn *ApicConnection) runConn(stopCh <-chan struct{}) {
        done := make(chan struct{})
        restart := make(chan struct{})
        queueStop := make(chan struct{})
        odevQueueStop := make(chan struct{})
        priorityQueueStop := make(chan struct{})
        syncHook := make(chan fullSync, 1)
        conn.restartCh = restart

        go func() {
                defer conn.connection.Close()
                defer close(done)

                for {
                        var apicresp ApicResponse
                        err := conn.connection.ReadJSON(&apicresp)
                        var closeErr *websocket.CloseError
                        if errors.As(err, &closeErr) {
                                conn.log.Info("Websocket connection closed: ", closeErr.Code)
                                conn.restart()
                                break
                        } else if err != nil {
                                conn.log.Error("Could not read web socket message:", err)
                                conn.restart()
                                break
                        } else {
                                conn.handleSocketUpdate(&apicresp)
                        }
                }
        }()

        if conn.VMMLiteSyncHook != nil {
                conn.VMMLiteSyncHook()
        }
        conn.indexMutex.Lock()
        oldState := conn.cacheDnSubIds
        conn.cachedState = make(map[string]ApicSlice)
        conn.cacheDnSubIds = make(map[string]map[string]bool)
        conn.deltaQueue = workqueue.NewNamedRateLimitingQueue(
                workqueue.NewMaxOfRateLimiter(
                        workqueue.NewItemExponentialFailureRateLimiter(5*time.Millisecond,
                                10*time.Second),
                        &workqueue.BucketRateLimiter{
                                Limiter: rate.NewLimiter(rate.Limit(10), int(30)),
                        },
                ),
                "delta")
        go conn.processQueue(conn.deltaQueue, queueStop, "delta")
        conn.odevQueue = workqueue.NewNamedRateLimitingQueue(
                workqueue.NewMaxOfRateLimiter(
                        workqueue.NewItemExponentialFailureRateLimiter(5*time.Millisecond,
                                10*time.Second),
                        &workqueue.BucketRateLimiter{
                                Limiter: rate.NewLimiter(rate.Limit(10), int(30)),
                        },
                ),
                "odev")
        go conn.processQueue(conn.odevQueue, odevQueueStop, "odev")
        conn.priorityQueue = workqueue.NewNamedRateLimitingQueue(
                workqueue.NewMaxOfRateLimiter(
                        workqueue.NewItemExponentialFailureRateLimiter(5*time.Millisecond,
                                10*time.Second),
                        &workqueue.BucketRateLimiter{
                                Limiter: rate.NewLimiter(rate.Limit(10), int(30)),
                        },
                ),
                "priority")
        go conn.processQueue(conn.priorityQueue, priorityQueueStop, "priority")
        conn.lldpIfQueue = workqueue.NewNamedRateLimitingQueue(
                workqueue.NewMaxOfRateLimiter(
                        workqueue.NewItemExponentialFailureRateLimiter(5*time.Millisecond,
                                10*time.Second),
                        &workqueue.BucketRateLimiter{
                                Limiter: rate.NewLimiter(rate.Limit(10), int(30)),
                        },
                ),
                "lldpIf")
        go conn.processLLDPIfQueue(conn.lldpIfQueue,
                conn.lldpIfHldr, stopCh)
        conn.indexMutex.Unlock()

        refreshInterval := conn.RefreshInterval
        if refreshInterval == 0 {
                refreshInterval = defaultConnectionRefresh
        }
        // Adjust refreshTickerInterval.
        // To refresh the subscriptions early than actual refresh timeout value
        refreshTickerInterval := refreshInterval - conn.RefreshTickerAdjust
        refreshTicker := time.NewTicker(refreshTickerInterval)
        defer refreshTicker.Stop()

        leafRebootCheckTicker := time.NewTicker(conn.LeafRebootCheckInterval)
        if conn.cnoEnabled {
                leafRebootCheckTicker.Stop()
        } else {
                conn.initializeLeafDnCache()
                defer leafRebootCheckTicker.Stop()
        }
        var hasErr bool
        for value, object := range conn.syncStore {
                if !(conn.getSyncObject(value, object)) {
                        hasErr = true
                        conn.restart()
                        break
                }
        }
        if !hasErr {
                for value, subscription := range conn.subscriptions.subs {
                        if !(conn.subscribe(value, subscription, false)) {
                                hasErr = true
                                conn.restart()
                                break
                        }
                }
        }
        if !hasErr {
                conn.checkDeletes(oldState)
                go func() {
                        if conn.FullSyncHook != nil {
                                conn.FullSyncHook()
                        }
                        syncHook <- fullSync{}
                }()
        }

        // Get APIC version if connection restarts
        if conn.version == "" && conn.checkVersion {
                go func() {
                        version, err := conn.GetVersion()
                        if err != nil {
                                conn.log.Error("Error while getting APIC version: ", err, " Restarting connection...")
                                conn.restart()
                        } else {
                                conn.log.Debug("Cached version:", conn.CachedVersion, " New version:", version)
                                if ApicVersion != version {
                                        ApicVersion = version
                                        if ApicVersion >= "6.0(4c)" {
                                                metadata["fvBD"].attributes["serviceBdRoutingDisable"] = "no"
                                        } else {
                                                delete(metadata["fvBD"].attributes, "serviceBdRoutingDisable")
                                        }
                                        if conn.VersionUpdateHook != nil {
                                                conn.VersionUpdateHook()
                                        }
                                }
                                conn.CachedVersion = version
                        }
                }()
        }

        closeConn := func(stop bool) {
                close(queueStop)
                close(odevQueueStop)

                conn.indexMutex.Lock()
                conn.deltaQueue = nil
                conn.odevQueue = nil
                conn.priorityQueue = nil
                conn.lldpIfQueue = nil
                conn.stopped = stop
                conn.syncEnabled = false
                conn.subscriptions.ids = make(map[string]string)
                conn.version = ""
                conn.indexMutex.Unlock()

                conn.log.Debug("Shutting down web socket")
                err := conn.connection.WriteMessage(websocket.CloseMessage,
                        websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
                if err != nil {
                        conn.log.Error("Error while closing socket: ", err)
                } else {
                        select {
                        case <-done:
                        case <-time.After(time.Second):
                        }
                }
                conn.connection.Close()
        }

loop:
        for {
                select {
                case <-syncHook:
                        conn.fullSync()
                case <-refreshTicker.C:
                        conn.refresh()
                case <-leafRebootCheckTicker.C:
                        conn.checkLeafReboot()
                case <-restart:
                        closeConn(false)
                        break loop
                case <-stopCh:
                        closeConn(true)
                        break loop
                }
        }

        conn.log.Debug("Exiting websocket handler")
}

// This function should only to be called before we make the first connection to APIC.
// Use the cached Apic version when determining the version elsewhere. This can lead to inconsistent tokens.
func (conn *ApicConnection) GetVersion() (string, error) {
        versionMo := "firmwareCtrlrRunning"

        if len(conn.Apic) == 0 {
                return "", errors.New("No APIC configuration")
        }

        conn.checkVersion = true // enable version check on websocket reconnect
        // To Handle unit-tests
        if strings.Contains(conn.Apic[conn.ApicIndex], "127.0.0.1") {
                conn.version = "4.2(4i)"
                conn.SnatPbrFltrChain = true
                conn.log.Debug("Returning APIC version 4.2(4i) for test server")
                return conn.version, nil
        }

        uri := fmt.Sprintf("/api/node/class/%s.json?&", versionMo)

        retries := 0
        for conn.version == "" {
                if retries <= conn.ReconnectRetryLimit {
                        // Wait before Retry.
                        time.Sleep(conn.ReconnectInterval)
                        retries++
                } else {
                        return "", fmt.Errorf("Failed to get APIC version after %d retries", retries)
                }

                token, err := conn.login()
                if err != nil {
                        conn.log.Error("Failed to log into APIC: ", err)
                        continue
                }
                conn.token = token

                resp, err := conn.sendHTTPSRequestToAPIC("GET", uri, nil, "", false, nil)
                if err != nil {
                        continue
                }
                defer complete(resp)

                var apicresp ApicResponse
                err = json.NewDecoder(resp.Body).Decode(&apicresp)
                if err != nil {
                        conn.log.Error("Could not parse APIC response: ", err)
                        continue
                }
                for _, obj := range apicresp.Imdata {
                        vresp := obj["firmwareCtrlrRunning"]
                        version, ok := vresp.Attributes["version"]
                        if !ok {
                                conn.log.Debug("No version attribute in the response??!")
                                conn.logger.WithFields(logrus.Fields{
                                        "mod":                            "APICAPI",
                                        "firmwareCtrlrRunning":           vresp,
                                        "firmwareCtrlRunning Attributes": vresp.Attributes,
                                }).Debug("Response:")
                        } else {
                                switch version := version.(type) {
                                default:
                                case string:
                                        version_split := strings.Split(version, "(")
                                        version_number, err := strconv.ParseFloat(version_split[0], 64)
                                        conn.log.Info("Actual APIC version:", version, " Stripped out version:", version_number)
                                        if err == nil {
                                                conn.version = version //return the actual version
                                        }
                                }
                        }
                }
        }
        return conn.version, nil
}

func (conn *ApicConnection) Run(stopCh <-chan struct{}) {
        if len(conn.Apic) == 0 {
                conn.log.Warning("APIC connection not configured")
                return
        }

        if conn.version >= "6.0(4c)" {
                metadata["fvBD"].attributes["serviceBdRoutingDisable"] = "no"
        }

        for !conn.stopped {
                func() {
                        defer func() {
                                conn.ApicIndex = (conn.ApicIndex + 1) % len(conn.Apic)
                                time.Sleep(conn.ReconnectInterval)
                        }()

                        conn.logger.WithFields(logrus.Fields{
                                "mod":  "APICAPI",
                                "host": conn.Apic[conn.ApicIndex],
                        }).Info("Connecting to APIC")

                        for dn := range conn.subscriptions.subs {
                                conn.subscriptions.subs[dn].childSubs = make(map[string]subComponent)
                        }
                        conn.subscriptions.ids = make(map[string]string)

                        token, err := conn.login()
                        if err != nil {
                                conn.log.Error("Failed to log into APIC: ", err)
                                return
                        }
                        conn.token = token

                        uri := fmt.Sprintf("/socket%s", token)
                        url := fmt.Sprintf("wss://%s%s",
                                conn.Apic[conn.ApicIndex], uri)
                        header := make(http.Header)
                        if conn.signer != nil {
                                sig, err := conn.signer.sign("GET", uri, nil)
                                if err != nil {
                                        conn.log.Error("Failed to sign request: ", err)
                                        return
                                }
                                header.Set("Cookie", conn.apicSigCookie(sig, token))
                        }

                        conn.connection, _, err = conn.dialer.Dial(url, header)
                        if err != nil {
                                conn.log.Error("Failed to open APIC websocket: ", err)
                                return
                        }
                        conn.log.Info("Websocket connected!")
                        conn.runConn(stopCh)
                }()
        }
}

func (conn *ApicConnection) refresh() {
        if conn.signer == nil {
                uri := "/api/aaaRefresh.json"
                resp, err := conn.sendHTTPSRequestToAPIC("GET", uri, nil, "", true, nil)
                if err != nil {
                        return
                }
                complete(resp)
                conn.log.Debugf("Refresh: url %v", resp.Request.URL)
        }

        for _, sub := range conn.subscriptions.subs {
                refreshId := func(id string) {
                        uri := fmt.Sprintf("/api/subscriptionRefresh.json?id=%s", id)
                        resp, err := conn.sendHTTPSRequestToAPIC("GET", uri, nil, "", true, nil)
                        if err != nil {
                                return
                        }
                        complete(resp)
                        conn.log.Debugf("Refresh sub: url %v", resp.Request.URL)
                        time.Sleep(conn.SubscriptionDelay)
                }
                if len(sub.childSubs) > 0 {
                        for id := range sub.childSubs {
                                refreshId(id)
                        }
                } else {
                        refreshId(sub.id)
                }
        }
}

func (conn *ApicConnection) getLeafDns() ApicSlice {
        uri := "/api/node/class/topSystem.json?query-target-filter=and(eq(topSystem.role,\"leaf\"))"
        apicresp, err := conn.GetApicResponse(uri)
        if err != nil {
                conn.log.Errorf("Unable to get leaf details. Error while getting apic response: %v", err)
                return nil
        }
        if len(apicresp.Imdata) == 0 {
                conn.log.Warnf("Zero leaf details found in apic response")
        }
        return apicresp.Imdata
}

func (conn *ApicConnection) initializeLeafDnCache() {
        imData := conn.getLeafDns()
        for _, obj := range imData {
                leafDn := obj.GetDn()
                lastRebootTimeStr := obj.GetAttrStr("lastRebootTime")
                lastRebootTime, err := time.Parse(time.RFC3339, lastRebootTimeStr)
                if err != nil {
                        conn.log.Warnf("Skipping leaf %v with unparsable lastRebootTime: %v", leafDn, err)
                        continue
                }
                conn.log.Debugf("Initializing cache with leaf %v with lastRebootTime: %v", leafDn, lastRebootTime)
                conn.cachedLeafDns[leafDn] = &leafCacheEntry{
                        lastRebootTime: lastRebootTime,
                        lastSeenTime:   time.Now(),
                }
        }
}

func (conn *ApicConnection) checkLeafReboot() {
        reboot := false
        imData := conn.getLeafDns()
        for _, obj := range imData {
                leafDn := obj.GetDn()
                lastRebootTimeStr := obj.GetAttrStr("lastRebootTime")
                lastRebootTime, err := time.Parse(time.RFC3339, lastRebootTimeStr)
                if err != nil {
                        conn.log.Warnf("Skipping leaf %v with unparsable lastRebootTime: %v", leafDn, err)
                        continue
                }
                if cacheEntry, present := conn.cachedLeafDns[leafDn]; present {
                        if !cacheEntry.lastRebootTime.Equal(lastRebootTime) {
                                conn.log.Debugf("Detected reboot of leaf %v. Current lastRebootTime:%v, cached lastRebootTime:%v", leafDn, lastRebootTime, cacheEntry.lastRebootTime)
                                conn.cachedLeafDns[leafDn].lastRebootTime = lastRebootTime
                                reboot = true
                        }
                } else {
                        conn.log.Debugf("Detected new leaf %v. Current lastRebootTime:%v", leafDn, lastRebootTime)
                        conn.cachedLeafDns[leafDn] = &leafCacheEntry{
                                lastRebootTime: lastRebootTime,
                        }
                        reboot = true
                }
                conn.cachedLeafDns[leafDn].lastSeenTime = time.Now()
        }

        for leafDn, cacheEntry := range conn.cachedLeafDns {
                if time.Since(cacheEntry.lastSeenTime) > cacheStaleAfterTime {
                        delete(conn.cachedLeafDns, leafDn)
                        conn.log.Debugf("Removing stale leaf cache entry: %v (last seen %v ago)", leafDn, time.Since(cacheEntry.lastSeenTime))
                }
        }
        if reboot {
                conn.log.Infof("Restarting APIC connection because of leaf reboot")
                conn.restart()
        }
}

// sendHTTPSRequestToAPIC sends an HTTPS request to APIC.
// - Retries on 503 when EnableRequestRetry is true, up to maxRequestRetry times, using exponential backoff with jitter.
// - If restart is true, the connection is restarted on request/transport errors, non-2xx responses, and after retry exhaustion.
// - Status codes in exceptionStatusCode are returned as-is for the caller to handle.
// Sets the Content-Type header when provided; pass nil for body and "" for contentType to omit them.
// Returns a non-nil error for non-2xx responses (except those listed in exceptionStatusCode).
func (conn *ApicConnection) sendHTTPSRequestToAPIC(method string, uri string, body []byte, contentType string,
        restart bool, exceptionStatusCode []int) (*http.Response, error) {
        url := fmt.Sprintf("https://%s%s", conn.Apic[conn.ApicIndex], uri)
        retry := 0

        for {
                // Construct new request for each retry.
                reqBody := bytes.NewReader(body)
                req, err := http.NewRequest(method, url, reqBody)
                if err != nil {
                        conn.log.Error("Could not create request:", err)
                        if restart {
                                conn.restart()
                        }
                        return nil, err
                }
                if contentType != "" {
                        req.Header.Set("Content-Type", contentType)
                }
                conn.sign(req, uri, body)

                if retry > 0 {
                        conn.log.Infof("Retrying request : %s, Attempt %d ", req.URL.String(), retry)
                }

                resp, err := conn.client.Do(req)
                if err != nil {
                        conn.log.Error("Failed to send request to APIC: ", err)
                        if restart {
                                conn.restart()
                        }
                        return nil, err
                }

                // If exceptionStatusCode is passed by caller function,
                // the caller is responsible for handling those status codes.
                for _, exception := range exceptionStatusCode {
                        if exception == resp.StatusCode {
                                return resp, nil
                        }
                }

                // Retry on 503 Service Unavailable
                if conn.EnableRequestRetry && resp.StatusCode == 503 {
                        conn.log.Error("Recieved Service Unavailable Response for url : ", req.URL.String())
                        retry++
                        if retry > maxRequestRetry {
                                conn.log.Errorf("Maximum retry for %s exceeded", req.URL.String())
                                complete(resp)
                                if restart {
                                        conn.restart()
                                }
                                return resp, fmt.Errorf("Got error response from APIC")
                        }
                        complete(resp)
                        // Exponential backoff with jitter
                        // With default minBase=20s (configurable via RequestRetryDelayBase):
                        //   retry intervals: [20s, 40s], [40s, 60s], [80s, 100s], [160s, 180s], [320s, 340s]
                        // Intervals are strictly non-overlapping (each min >= previous max).
                        // For all retries to complete - min time: 620s ~ 10 min 20 sec, max time: 720s = 12 min.
                        minBase := time.Duration(conn.RequestRetryDelayBase) * time.Second
                        exp := time.Duration(1 << uint(retry-1)) // 2^(retry-1): 1, 2, 4, 8, 16
                        base := minBase * exp
                        jitter := time.Duration(rand.Int63n(int64(20 * time.Second)))
                        backoffDuration := base + jitter
                        conn.log.Debugf("Waiting %v before retry attempt %d", backoffDuration, retry)
                        time.Sleep(backoffDuration)
                        continue
                }

                // Handle non-success status codes
                if resp.StatusCode < 200 || resp.StatusCode >= 300 {
                        conn.logErrorResp("Error response from APIC ", resp)
                        complete(resp)
                        if restart {
                                conn.restart()
                        }
                        return resp, fmt.Errorf("Got error response from APIC")
                }

                //success case
                return resp, nil
        }
}

func (conn *ApicConnection) logErrorResp(message string, resp *http.Response) {
        var apicresp ApicResponse
        err := json.NewDecoder(resp.Body).Decode(&apicresp)
        if err != nil {
                conn.log.Error("Could not parse APIC error response: ", err)
        } else {
                code := 0
                text := ""
                for _, o := range apicresp.Imdata {
                        if ob, ok := o["error"]; ok {
                                if ob.Attributes != nil {
                                        if t, isStr := ob.Attributes["text"].(string); isStr {
                                                text = t
                                        }
                                        if c, isInt := ob.Attributes["code"].(int); isInt {
                                                code = c
                                        }
                                }
                        }
                }
                conn.logger.WithFields(logrus.Fields{
                        "mod":    "APICAPI",
                        "text":   text,
                        "code":   code,
                        "url":    resp.Request.URL,
                        "status": resp.StatusCode,
                }).Error(message)
        }
}

// To make sure cluster's POD/NodeBDs and L3OUT are all mapped
// to same and correct VRF.
func (conn *ApicConnection) ValidateAciVrfAssociation(acivrfdn string, expectedVrfRelations []string) error {
        var aciVrfBdL3OuttDns []string
        args := []string{
                "query-target=subtree",
                "target-subtree-class=fvRtCtx,fvRtEctx",
        }

        uri := fmt.Sprintf("/api/mo/%s.json?%s", acivrfdn, strings.Join(args, "&"))
        resp, err := conn.sendHTTPSRequestToAPIC("GET", uri, nil, "", true, nil)
        if err != nil {
                return err
        }
        defer complete(resp)

        var apicresp ApicResponse
        err = json.NewDecoder(resp.Body).Decode(&apicresp)
        if err != nil {
                conn.log.Error("Could not parse APIC response: ", err)
                return err
        }

        for _, obj := range apicresp.Imdata {
                for _, body := range obj {
                        tDn, ok := body.Attributes["tDn"].(string)
                        if !ok {
                                continue
                        }
                        aciVrfBdL3OuttDns = append(aciVrfBdL3OuttDns, tDn)
                }
        }
        sort.Strings(aciVrfBdL3OuttDns)
        conn.log.Debug("aciVrfBdL3OuttDns:", aciVrfBdL3OuttDns)
        for _, expectedDn := range expectedVrfRelations {
                i := sort.SearchStrings(aciVrfBdL3OuttDns, expectedDn)
                if !(i < len(aciVrfBdL3OuttDns) && aciVrfBdL3OuttDns[i] == expectedDn) {
                        conn.log.Debug("Missing (or) Incorrect Vrf association: ", expectedDn)
                        return errors.New("Incorrect Pod/NodeBD/L3OUT VRF association")
                }
        }
        return nil
}

func (conn *ApicConnection) getSubtreeDn(dn string, respClasses []string,
        updateHandlers []ApicObjectHandler) {
        // If there are no update handlers, there is no point doing anything till sync is enabled as reconcileApicObject just returns
        if len(updateHandlers) == 0 {
                conn.indexMutex.Lock()
                if !conn.syncEnabled {
                        conn.indexMutex.Unlock()
                        return
                }
                conn.indexMutex.Unlock()
        }
        args := []string{
                "rsp-subtree=full",
        }

        if len(respClasses) > 0 {
                args = append(args, "rsp-subtree-class="+strings.Join(respClasses, ","))
        }
        // properly encoding the URI query parameters breaks APIC
        uri := fmt.Sprintf("/api/mo/%s.json?%s", dn, strings.Join(args, "&"))
        conn.log.Debugf("URL: %v", uri)
        resp, err := conn.sendHTTPSRequestToAPIC("GET", uri, nil, "", true, nil)
        if err != nil {
                return
        }
        defer complete(resp)

        var apicresp ApicResponse
        err = json.NewDecoder(resp.Body).Decode(&apicresp)
        if err != nil {
                conn.log.Error("Could not parse APIC response: ", err)
                return
        }
        if len(apicresp.Imdata) == 0 {
                conn.log.Debugf("No subtree found for dn %s", dn)
        }

        for _, obj := range apicresp.Imdata {
                conn.logger.WithFields(logrus.Fields{
                        "mod": "APICAPI",
                        "dn":  obj.GetDn(),
                        "obj": obj,
                }).Debug("Object updated on APIC")
                var count int
                prepareApicCache("", obj, &count)

                handled := false
                for _, handler := range updateHandlers {
                        if handler(obj) {
                                handled = true
                                break
                        }
                }
                if handled {
                        continue
                }
                conn.reconcileApicObject(obj)
        }
}

func (conn *ApicConnection) queuePriorityDn(dn string) {
        conn.indexMutex.Lock()
        if conn.priorityQueue != nil {
                conn.addToQueue(conn.priorityQueue, dn)
        }
        conn.indexMutex.Unlock()
}

func (conn *ApicConnection) queueDn(dn string) {
        conn.indexMutex.Lock()
        if conn.deltaQueue != nil {
                conn.addToQueue(conn.deltaQueue, dn)
        }
        conn.indexMutex.Unlock()
}

func (conn *ApicConnection) ForceRelogin() {
        conn.token = ""
}

/*
// Commented out - function no longer in use
func (conn *ApicConnection) PostDnInline(dn string, obj ApicObject) error {
        conn.logger.WithFields(logrus.Fields{
                "mod": "APICAPI",
                "dn":  dn,
                "obj": obj,
        }).Debug("Posting Dn Inline")
        if conn.token == "" {
                token, err := conn.login()
                if err != nil {
                        conn.log.Errorf("Login: %v", err)
                        return err
                }
                conn.token = token
        }
        uri := fmt.Sprintf("/api/mo/%s.json", dn)
        url := fmt.Sprintf("https://%s%s", conn.Apic[conn.ApicIndex], uri)
        raw, err := json.Marshal(obj)
        if err != nil {
                conn.log.Error("Could not serialize object for dn ", dn, ": ", err)
                return err
        }
        req, err := http.NewRequest("POST", url, bytes.NewBuffer(raw))
        if err != nil {
                conn.log.Error("Could not create request: ", err)
                return err
        }
        conn.sign(req, uri, raw)
        req.Header.Set("Content-Type", "application/json")
        conn.log.Infof("Post: %+v", req)
        resp, err := conn.client.Do(req)
        if err != nil {
                conn.log.Error("Could not update dn ", dn, ": ", err)
                return err
        }

        complete(resp)
        if resp.StatusCode != http.StatusOK {
                return fmt.Errorf("Status: %v", resp.StatusCode)
        }
        return nil
}

// Commented out - function no longer in use
func (conn *ApicConnection) DeleteDnInline(dn string) error {
        conn.logger.WithFields(logrus.Fields{
                "mod": "APICAPI",
                "dn":  dn,
        }).Debug("Deleting Dn Inline")
        uri := fmt.Sprintf("/api/mo/%s.json", dn)
        url := fmt.Sprintf("https://%s%s", conn.Apic[conn.ApicIndex], uri)
        req, err := http.NewRequest("DELETE", url, http.NoBody)
        if err != nil {
                conn.log.Error("Could not create delete request: ", err)
                return err
        }
        conn.sign(req, uri, nil)
        resp, err := conn.client.Do(req)
        if err != nil {
                conn.log.Error("Could not delete dn ", dn, ": ", err)
                return err
        }
        // TODO: handle resp statusCode
        defer complete(resp)
        return nil
}
*/

func (conn *ApicConnection) postDn(dn string, obj ApicObject) bool {
        conn.logger.WithFields(logrus.Fields{
                "mod": "APICAPI",
                "dn":  dn,
                "obj": obj,
        }).Debug("Posting Dn")

        uri := fmt.Sprintf("/api/mo/%s.json", dn)
        raw, err := json.Marshal(obj)
        if err != nil {
                conn.log.Error("Could not serialize object for dn ", dn, ": ", err)
        }
        resp, err := conn.sendHTTPSRequestToAPIC("POST", uri, raw, "application/json", true, []int{400})
        if err != nil {
                return false
        }
        defer complete(resp)
        if resp.StatusCode == 400 {
                return true
        }
        return false
}

func (conn *ApicConnection) Delete(dn string) bool {
        if dn == "" {
                conn.log.Debug("Skip delete for empty Dn: ")
                return false
        }
        dnSlice := strings.Split(dn, "/")
        identifier := dnSlice[len(dnSlice)-1]
        iSlice := strings.SplitN(identifier, "-", 2)
        if len(iSlice) == 2 {
                if iSlice[0] == "ip" {
                        addr := strings.Trim(iSlice[1], "[]")
                        obj := NewDeleteHostprotRemoteIp(addr)
                        conn.log.Debug("Posting delete of dn ", dn)
                        return conn.postDn(dn, obj)
                } else if iSlice[0] == "odev" {
                        conn.log.Debug("Skipping delete of opflexODev : ", dn)
                        return false
                }
        }
        return conn.DeleteDn(dn)
}

func (conn *ApicConnection) DeleteDn(dn string) bool {
        if dn == "" {
                conn.log.Debug("Skip delete for empty Dn: ")
                return false
        }
        conn.logger.WithFields(logrus.Fields{
                "mod": "APICAPI",
                "dn":  dn,
        }).Debug("Deleting Dn")
        uri := fmt.Sprintf("/api/mo/%s.json", dn)
        resp, err := conn.sendHTTPSRequestToAPIC("DELETE", uri, nil, "", true, nil)
        if err != nil {
                return false
        }
        defer complete(resp)
        return false
}

func doComputeRespClasses(targetClasses []string,
        visited map[string]bool) {
        for _, class := range targetClasses {
                if visited[class] {
                        continue
                }
                visited[class] = true
                if md, ok := metadata[class]; ok {
                        doComputeRespClasses(md.children, visited)
                }
        }
}

func computeRespClasses(targetClasses []string) []string {
        visited := make(map[string]bool)
        doComputeRespClasses(targetClasses, visited)

        var respClasses []string
        for class := range visited {
                respClasses = append(respClasses, class)
        }
        respClasses = append(respClasses, "tagAnnotation")
        return respClasses
}

// AddSubscriptionTree subscribe at a subtree level. class specifies
// the root. Changes will cause entire subtree of the rootdn to be fetched
func (conn *ApicConnection) AddSubscriptionTree(class string,
        targetClasses []string, targetFilter string) {
        if _, ok := classDepth[class]; !ok {
                errStr := fmt.Sprintf("classDepth not for class %s", class)
                panic(errStr)
        }

        conn.indexMutex.Lock()
        conn.subscriptions.subs[class] = &subscription{
                kind:          apicSubTree,
                childSubs:     make(map[string]subComponent),
                targetClasses: targetClasses,
                targetFilter:  targetFilter,
        }
        conn.indexMutex.Unlock()
}

func (conn *ApicConnection) AddSubscriptionClass(class string,
        targetClasses []string, targetFilter string) {
        conn.indexMutex.Lock()
        conn.subscriptions.subs[class] = &subscription{
                kind:          apicSubClass,
                childSubs:     make(map[string]subComponent),
                targetClasses: targetClasses,
                respClasses:   computeRespClasses(targetClasses),
                targetFilter:  targetFilter,
        }
        conn.indexMutex.Unlock()
}

func (conn *ApicConnection) AddSubscriptionDn(dn string,
        targetClasses []string) {
        conn.logger.WithFields(logrus.Fields{
                "mod": "APICAPI",
                "dn":  dn,
        }).Debug("Adding Subscription for Dn")

        conn.indexMutex.Lock()
        conn.subscriptions.subs[dn] = &subscription{
                kind:          apicSubDn,
                childSubs:     make(map[string]subComponent),
                targetClasses: targetClasses,
                respClasses:   computeRespClasses(targetClasses),
        }
        conn.indexMutex.Unlock()
}

func (conn *ApicConnection) AddSyncDn(dn string,
        targetClasses []string) {
        conn.logger.WithFields(logrus.Fields{
                "mod": "APICAPI",
                "dn":  dn,
        }).Debug("Adding Dn for syncing")
        conn.indexMutex.Lock()
        conn.syncStore[dn] = &syncObj{
                kind:          apicSubDn,
                targetClasses: targetClasses,
                respClasses:   computeRespClasses(targetClasses),
        }
        conn.indexMutex.Unlock()
}

func (conn *ApicConnection) SetSyncHooks(value string,
        updateHook ApicObjectHandler, deleteHook ApicDnHandler) {
        conn.indexMutex.Lock()
        if s, ok := conn.subscriptions.subs[value]; ok {
                s.updateHook = updateHook
                s.deleteHook = deleteHook
        }
        conn.indexMutex.Unlock()
}

func (conn *ApicConnection) SetSubscriptionHooks(value string,
        updateHook ApicObjectHandler, deleteHook ApicDnHandler) {
        conn.indexMutex.Lock()
        if s, ok := conn.subscriptions.subs[value]; ok {
                s.updateHook = updateHook
                s.deleteHook = deleteHook
        }
        conn.indexMutex.Unlock()
}

func (conn *ApicConnection) GetApicResponse(uri string) (ApicResponse, error) {
        conn.log.Debug("apicIndex: ", conn.Apic[conn.ApicIndex], " uri: ", uri)
        conn.log.Debug("Apic Get url: ", uri)
        var apicresp ApicResponse
        resp, err := conn.sendHTTPSRequestToAPIC("GET", uri, nil, "", false, nil)
        if err != nil {
                return apicresp, err
        }
        defer complete(resp)
        err = json.NewDecoder(resp.Body).Decode(&apicresp)
        if err != nil {
                conn.log.Error("Could not parse APIC response: ", err)
                return apicresp, err
        }
        return apicresp, nil
}

func (conn *ApicConnection) doSubscribe(args []string,
        kind, value, refresh_interval string, apicresp *ApicResponse) bool {
        // properly encoding the URI query parameters breaks APIC
        uri := fmt.Sprintf("/api/%s/%s.json?subscription=yes&%s%s",
                kind, value, refresh_interval, strings.Join(args, "&"))
        conn.log.Info("APIC connection URL: ", uri)
        resp, err := conn.sendHTTPSRequestToAPIC("GET", uri, nil, "", false, nil)
        if err != nil {
                return false
        }
        defer complete(resp)

        err = json.NewDecoder(resp.Body).Decode(apicresp)
        if err != nil {
                conn.log.Error("Could not decode APIC response", err)
                return false
        }
        time.Sleep(conn.SubscriptionDelay)
        return true
}

func (conn *ApicConnection) isPresentInOpflexOdevCache(dn, status string, obj ApicObject) bool {
        if _, ok := conn.cacheOpflexOdev[dn]; ok {
                return true
        }

        switch status {
        case "created":
                return conn.updateOpflexOdevCache(obj, false, true)
        case "deleted":
                return conn.updateOpflexOdevCache(obj, true, true)
        default:
                return false
        }
}

func (conn *ApicConnection) updateOpflexOdevCache(obj ApicObject, remove, lockHeld bool) bool {
        var updated bool
        dn := obj.GetDn()
        if !lockHeld {
                conn.indexMutex.Lock()
                defer conn.indexMutex.Unlock()
        }

        if remove {
                if _, ok := conn.cacheOpflexOdev[dn]; ok {
                        delete(conn.cacheOpflexOdev, dn)
                        updated = true
                        conn.log.Info("Removed dn from opflexOdev cache: ", dn)
                }
        } else if strings.Contains(obj.GetDomName(), conn.VmmDomain) ||
                (strings.Contains(conn.Flavor, "openstack") && strings.Contains(obj.GetCompHvDn(), "/prov-OpenStack/")) {
                conn.cacheOpflexOdev[dn] = struct{}{}
                updated = true
                conn.log.Info("Added dn in opflexOdev cache: ", dn)
        }
        return updated
}

func (conn *ApicConnection) computeArgSet(targetClasses, respClasses []string, targetFilter string, defaultArgs int) (splitTargetClasses, splitRespClasses, argSet [][]string, argCount int) {

        baseArgs := []string{
                "query-target=subtree",
                "rsp-subtree=full",
                "target-subtree-class=" + strings.Join(targetClasses, ","),
        }

        argCount = defaultArgs
        var combinableSubClasses, separableSubClasses []string
        argSet = make([][]string, defaultArgs)
        argSet[defaultArgs-1] = make([]string, len(baseArgs))
        copy(argSet[defaultArgs-1], baseArgs)
        if respClasses != nil {
                separateClasses := func(classes []string, combClasses, sepClasses *[]string) {
                        for i := range classes {
                                if classMeta, ok := metadata[classes[i]]; ok {
                                        if classes[i] == "tagAnnotation" {
                                                continue
                                        }
                                        if classMeta.hints != nil && classMeta.hints["cardinality"] == "high" {
                                                *sepClasses = append(*sepClasses, classes[i])
                                                continue
                                        }
                                        *combClasses = append(*combClasses, classes[i])
                                }
                        }
                }
                separateClasses(respClasses, &combinableSubClasses, &separableSubClasses)

                // In case there are high cardinality children, we register for all the classes individually.
                // The concept of target-subtree and rsp-subtree class cannot be used because of the tagAnnotation object
                // vmmInjectedLabel is added for every object, so getting it separately will not be scalable
                if len(separableSubClasses) > 0 {
                        separateClasses(targetClasses, &combinableSubClasses, &separableSubClasses)
                        separableSubClasses = append(separableSubClasses, combinableSubClasses...)
                        baseArgs = []string{
                                "query-target=subtree",
                                "rsp-subtree=children",
                        }
                        subscribingClasses := make(map[string]bool)
                        argSet = make([][]string, len(separableSubClasses))
                        splitTargetClasses = make([][]string, len(separableSubClasses))
                        splitRespClasses = make([][]string, len(separableSubClasses))

                        argCount = 0
                        for i := range separableSubClasses {
                                // Eliminate duplicates
                                if _, ok := subscribingClasses[separableSubClasses[i]]; ok {
                                        continue
                                }
                                subscribingClasses[separableSubClasses[i]] = true
                                argSet[argCount] = make([]string, len(baseArgs))
                                copy(argSet[argCount], baseArgs)
                                argSet[argCount] = append(argSet[argCount], "target-subtree-class="+separableSubClasses[i], "rsp-subtree-class=tagAnnotation")
                                splitTargetClasses[argCount] = append(splitTargetClasses[argCount], separableSubClasses[i])
                                splitRespClasses[argCount] = computeRespClasses([]string{separableSubClasses[i]})
                                argCount++
                        }
                } else {
                        argSet[defaultArgs-1] = append(argSet[defaultArgs-1], "rsp-subtree-class="+strings.Join(combinableSubClasses, ",")+",tagAnnotation")
                }
        }
        if targetFilter != "" {
                targetFilterArgs := "query-target-filter=" + targetFilter
                if len(separableSubClasses) == 0 {
                        argSet[defaultArgs-1] = append(argSet[defaultArgs-1], targetFilterArgs)
                } else {
                        for i := 0; i < argCount; i++ {
                                argSet[i] = append(argSet[i], targetFilterArgs)
                        }
                }
        }

        return
}

func (conn *ApicConnection) processImdata(value string, subId string, args []string, imdata ApicSlice, updateHook ApicObjectHandler, lockHeld bool) {
        var respObjCount int
        for _, obj := range imdata {
                dn := obj.GetDn()
                if dn == "" {
                        continue
                }
                if subId != "" {
                        if !lockHeld {
                                conn.indexMutex.Lock()
                        }
                        subIds, found := conn.cacheDnSubIds[dn]
                        if !found {
                                subIds = make(map[string]bool)
                                conn.cacheDnSubIds[dn] = subIds
                        }
                        subIds[subId] = true
                        if !lockHeld {
                                conn.indexMutex.Unlock()
                        }
                }
                if value == "opflexODev" && conn.FilterOpflexDevice {
                        conn.updateOpflexOdevCache(obj, false, lockHeld)
                }
                if updateHook != nil && updateHook(obj) {
                        continue
                }

                tag := obj.GetTag()
                if !conn.isSyncTag(tag) {
                        continue
                }

                conn.logger.WithFields(logrus.Fields{
                        "mod": "APICAPI",
                        "dn":  dn,
                        "tag": tag,
                        "obj": obj,
                }).Debug("Caching")
                var count int
                prepareApicCache("", obj, &count)
                respObjCount += count
                if !lockHeld {
                        conn.indexMutex.Lock()
                }
                conn.cachedState[tag] = append(conn.cachedState[tag], obj)
                if !lockHeld {
                        conn.indexMutex.Unlock()
                }
        }
        if respObjCount >= ApicSubscriptionResponseMoMaxCount/10 {
                conn.logger.WithFields(logrus.Fields{
                        "args":       args,
                        "moCount":    respObjCount,
                        "maxAllowed": ApicSubscriptionResponseMoMaxCount,
                }).Warning("Subscription response is significantly large. Each new object will add 2 Mos atleast and twice the number of labels on the object")
        } else {
                conn.logger.WithFields(logrus.Fields{
                        "moCount": respObjCount,
                }).Debug("ResponseObjCount")
        }
}

func (conn *ApicConnection) getSyncObject(value string, obj *syncObj) bool {
        const defaultArgs = 1
        _, _, argSet, argCount := conn.computeArgSet(obj.targetClasses, obj.respClasses, obj.targetFilter, defaultArgs)

        kind := "mo"
        if obj.kind == apicSubClass || obj.kind == apicSubTree {
                kind = "class"
        }

        for i := 0; i < argCount; i++ {
                // properly encoding the URI query parameters breaks APIC
                uri := fmt.Sprintf("/api/%s/%s.json?%s", kind, value, strings.Join(argSet[i], "&"))
                resp, err := conn.sendHTTPSRequestToAPIC("GET", uri, nil, "", false, nil)
                if err != nil {
                        return false
                }
                defer complete(resp)
                var apicresp ApicResponse
                err = json.NewDecoder(resp.Body).Decode(&apicresp)
                if err != nil {
                        conn.log.Error("Could not decode APIC response", err)
                        return false
                }
                conn.logger.WithFields(logrus.Fields{
                        "mod":   "APICAPI",
                        "value": value,
                        "kind":  kind,
                        "args":  argSet[i],
                }).Debug("Fetched")

                conn.processImdata(value, "", argSet[i], apicresp.Imdata, nil, false)
        }

        return true
}

func (conn *ApicConnection) subscribe(value string, sub *subscription, lockHeld bool) bool {
        const defaultArgs = 1
        splitTargetClasses, splitRespClasses, argSet, argCount := conn.computeArgSet(sub.targetClasses, sub.respClasses, sub.targetFilter, defaultArgs)

        kind := "mo"
        if sub.kind == apicSubClass || sub.kind == apicSubTree {
                kind = "class"
        }

        refresh_interval := ""
        if conn.RefreshInterval != 0 {
                refresh_interval = fmt.Sprintf("refresh-timeout=%v&",
                        conn.RefreshInterval.Seconds())
        }
        for i := 0; i < argCount; i++ {
                var apicresp ApicResponse
                if !conn.doSubscribe(argSet[i], kind, value, refresh_interval, &apicresp) {
                        return false
                }
                subId, ok := apicresp.SubscriptionId.(string)
                if !ok {
                        conn.log.Error("Subscription ID is not a string")
                        return false
                }

                conn.logger.WithFields(logrus.Fields{
                        "mod":   "APICAPI",
                        "value": value,
                        "kind":  kind,
                        "id":    subId,
                        "args":  argSet[i],
                }).Debug("Subscribed")
                if !lockHeld {
                        conn.indexMutex.Lock()
                }
                if argCount > defaultArgs {
                        sub.childSubs[subId] = subComponent{
                                targetClasses: splitTargetClasses[i],
                                respClasses:   splitRespClasses[i],
                        }
                } else {
                        conn.subscriptions.subs[value].id = subId
                }
                conn.subscriptions.ids[subId] = value
                if !lockHeld {
                        conn.indexMutex.Unlock()
                }

                conn.processImdata(value, subId, argSet[i], apicresp.Imdata, sub.updateHook, lockHeld)
        }

        return true
}

var tagRegexp = regexp.MustCompile(`[a-zA-Z0-9_]{1,31}-[a-f0-9]{32}`)

func (conn *ApicConnection) isSyncTag(tag string) bool {
        return tagRegexp.MatchString(tag) &&
                strings.HasPrefix(tag, conn.prefix+"-")
}

// isOwnedByController fetches the object from APIC and checks whether
// its tagAnnotation matches this controller's sync tag. Returns true
// only when the object exists on APIC and carries this controller's tag.
func (conn *ApicConnection) isOwnedByController(dn string) bool {
        uri := fmt.Sprintf("/api/mo/%s.json?rsp-subtree=children&rsp-subtree-class=tagAnnotation", dn)
        resp, err := conn.GetApicResponse(uri)
        if err != nil {
                conn.log.Warning("Failed to fetch object for tag check: ", dn, " err: ", err)
                return false
        }
        if len(resp.Imdata) == 0 {
                // Object no longer exists on APIC; nothing to protect.
                conn.log.Debug("Object not found on APIC for tag check: ", dn)
                return false
        }
        tag := resp.Imdata[0].GetTag()
        if conn.isSyncTag(tag) {
                return true
        }
        conn.log.Debug("Object tag does not match controller prefix, skipping: ",
                dn, " tag: ", tag)
        return false
}

func getRootDn(dn, rootClass string) string {
        depth := classDepth[rootClass]
        parts := strings.Split(dn, "/")
        parts = parts[:depth]
        return strings.Join(parts, "/")
}

func (conn *ApicConnection) PostApicObjects(uri string, payload ApicSlice) error {
        conn.log.Debug("apicIndex: ", conn.Apic[conn.ApicIndex], " uri: ", uri)
        conn.log.Debug("Apic POST url: ", uri)

        if conn.token == "" {
                token, err := conn.login()
                if err != nil {
                        conn.log.Errorf("Login: %v", err)
                        return err
                }
                conn.token = token
        }

        raw, err := json.Marshal(payload)
        if err != nil {
                conn.log.Error("Could not serialize object: ", err)
                return err
        }
        conn.log.Infof("Post: %+v", &struct{ Method, URI string }{"POST", uri})
        resp, err := conn.sendHTTPSRequestToAPIC("POST", uri, raw, "application/json", false, nil)
        if err != nil {
                return err
        }
        defer complete(resp)

        if resp.StatusCode != http.StatusOK {
                return fmt.Errorf("Status: %v", resp.StatusCode)
        }

        return nil
}

noironetworks / aci-containers / 11995

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