vnids_master.go

package plugin

import (
	"fmt"
	"strings"
	"sync"

	log "github.com/golang/glog"

	kapi "k8s.io/kubernetes/pkg/api"
	utilruntime "k8s.io/kubernetes/pkg/util/runtime"
	"k8s.io/kubernetes/pkg/util/sets"
	utilwait "k8s.io/kubernetes/pkg/util/wait"
	"k8s.io/kubernetes/pkg/watch"

	osapi "github.com/openshift/origin/pkg/sdn/api"
	pnetid "github.com/openshift/origin/pkg/sdn/plugin/netid"
)

type masterVNIDMap struct {
	// Synchronizes assign, revoke and update VNID
	lock         sync.Mutex
	ids          map[string]uint32
	netIDManager *pnetid.Allocator

	adminNamespaces sets.String
}

func newMasterVNIDMap() *masterVNIDMap {
	netIDRange, err := pnetid.NewNetIDRange(osapi.MinVNID, osapi.MaxVNID)
	if err != nil {
		panic(err)
	}

	return &masterVNIDMap{
		netIDManager:    pnetid.NewInMemory(netIDRange),
		adminNamespaces: sets.NewString(kapi.NamespaceDefault),
		ids:             make(map[string]uint32),
	}
}

func (vmap *masterVNIDMap) getVNID(name string) (uint32, bool) {
	id, found := vmap.ids[name]
	return id, found
}

func (vmap *masterVNIDMap) setVNID(name string, id uint32) {
	vmap.ids[name] = id
}

func (vmap *masterVNIDMap) unsetVNID(name string) (uint32, bool) {
	id, found := vmap.ids[name]
	delete(vmap.ids, name)
	return id, found
}

func (vmap *masterVNIDMap) getVNIDCount(id uint32) int {
	count := 0
	for _, netid := range vmap.ids {
		if id == netid {
			count = count + 1
		}
	}
	return count
}

func (vmap *masterVNIDMap) isAdminNamespace(nsName string) bool {
	if vmap.adminNamespaces.Has(nsName) {
		return true
	}
	return false
}

func (vmap *masterVNIDMap) populateVNIDs(registry *Registry) error {
	netnsList, err := registry.GetNetNamespaces()
	if err != nil {
		return err
	}

	for _, netns := range netnsList {
		vmap.setVNID(netns.NetName, netns.NetID)

		// Skip GlobalVNID, not part of netID allocation range
		if netns.NetID == osapi.GlobalVNID {
			continue
		}

		switch err := vmap.netIDManager.Allocate(netns.NetID); err {
		case nil: // Expected normal case
		case pnetid.ErrAllocated: // Expected when project networks are joined
		default:
			return fmt.Errorf("unable to allocate netid %d: %v", netns.NetID, err)
		}
	}
	return nil
}

func (vmap *masterVNIDMap) allocateNetID(nsName string) (uint32, bool, error) {
	// Nothing to do if the netid is in the vnid map
	exists := false
	if netid, found := vmap.getVNID(nsName); found {
		exists = true
		return netid, exists, nil
	}

	// NetNamespace not found, so allocate new NetID
	var netid uint32
	if vmap.isAdminNamespace(nsName) {
		netid = osapi.GlobalVNID
	} else {
		var err error
		netid, err = vmap.netIDManager.AllocateNext()
		if err != nil {
			return 0, exists, err
		}
	}

	vmap.setVNID(nsName, netid)
	log.Infof("Allocated netid %d for namespace %q", netid, nsName)
	return netid, exists, nil
}

func (vmap *masterVNIDMap) releaseNetID(nsName string) error {
	// Remove NetID from vnid map
	netid, found := vmap.unsetVNID(nsName)
	if !found {
		return fmt.Errorf("netid not found for namespace %q", nsName)
	}

	// Skip osapi.GlobalVNID as it is not part of NetID allocation
	if netid == osapi.GlobalVNID {
		return nil
	}

	// Check if this netid is used by any other namespaces
	// If not, then release the netid
	if count := vmap.getVNIDCount(netid); count == 0 {
		if err := vmap.netIDManager.Release(netid); err != nil {
			return fmt.Errorf("Error while releasing netid %d for namespace %q, %v", netid, nsName, err)
		}
		log.Infof("Released netid %d for namespace %q", netid, nsName)
	} else {
		log.V(5).Infof("netid %d for namespace %q is still in use", netid, nsName)
	}
	return nil
}

func (vmap *masterVNIDMap) updateNetID(nsName string, action osapi.PodNetworkAction, args string) (uint32, error) {
	var netid uint32
	allocated := false

	// Check if the given namespace exists or not
	oldnetid, found := vmap.getVNID(nsName)
	if !found {
		return 0, fmt.Errorf("netid not found for namespace %q", nsName)
	}

	// Determine new network ID
	switch action {
	case osapi.GlobalPodNetwork:
		netid = osapi.GlobalVNID
	case osapi.JoinPodNetwork:
		joinNsName := args
		var found bool
		if netid, found = vmap.getVNID(joinNsName); !found {
			return 0, fmt.Errorf("netid not found for namespace %q", joinNsName)
		}
	case osapi.IsolatePodNetwork:
		// Check if the given namespace is already isolated
		if count := vmap.getVNIDCount(oldnetid); count == 1 {
			return oldnetid, nil
		}

		var err error
		netid, err = vmap.netIDManager.AllocateNext()
		if err != nil {
			return 0, err
		}
		allocated = true
	default:
		return 0, fmt.Errorf("invalid pod network action: %v", action)
	}

	// Release old network ID
	if err := vmap.releaseNetID(nsName); err != nil {
		if allocated {
			vmap.netIDManager.Release(netid)
		}
		return 0, err
	}

	// Set new network ID
	vmap.setVNID(nsName, netid)
	log.Infof("Updated netid %d for namespace %q", netid, nsName)
	return netid, nil
}

// assignVNID, revokeVNID and updateVNID methods updates in-memory structs and persists etcd objects
func (vmap *masterVNIDMap) assignVNID(registry *Registry, nsName string) error {
	vmap.lock.Lock()
	defer vmap.lock.Unlock()

	netid, exists, err := vmap.allocateNetID(nsName)
	if err != nil {
		return err
	}

	if !exists {
		// Create NetNamespace Object and update vnid map
		err = registry.CreateNetNamespace(nsName, netid)
		if err != nil {
			vmap.releaseNetID(nsName)
			return err
		}
	}
	return nil
}

func (vmap *masterVNIDMap) revokeVNID(registry *Registry, nsName string) error {
	vmap.lock.Lock()
	defer vmap.lock.Unlock()

	// Delete NetNamespace object
	if err := registry.DeleteNetNamespace(nsName); err != nil {
		return err
	}

	if err := vmap.releaseNetID(nsName); err != nil {
		return err
	}
	return nil
}

func (vmap *masterVNIDMap) updateVNID(registry *Registry, netns *osapi.NetNamespace) error {
	action, args, err := osapi.GetChangePodNetworkAnnotation(netns)
	if err == osapi.ErrorPodNetworkAnnotationNotFound {
		// Nothing to update
		return nil
	}

	vmap.lock.Lock()
	defer vmap.lock.Unlock()

	netid, err := vmap.updateNetID(netns.NetName, action, args)
	if err != nil {
		return err
	}
	netns.NetID = netid
	osapi.DeleteChangePodNetworkAnnotation(netns)

	if _, err := registry.UpdateNetNamespace(netns); err != nil {
		return err
	}
	return nil
}

//--------------------- Master methods ----------------------

func (master *OsdnMaster) VnidStartMaster() error {
	err := master.vnids.populateVNIDs(master.registry)
	if err != nil {
		return err
	}

	go utilwait.Forever(master.watchNamespaces, 0)
	go utilwait.Forever(master.watchNetNamespaces, 0)
	return nil
}

func (master *OsdnMaster) watchNamespaces() {
	registry := master.registry
	eventQueue := registry.RunEventQueue(Namespaces)

	for {
		eventType, obj, err := eventQueue.Pop()
		if err != nil {
			utilruntime.HandleError(fmt.Errorf("EventQueue failed for namespaces: %v", err))
			return
		}
		ns := obj.(*kapi.Namespace)
		name := ns.ObjectMeta.Name

		log.V(5).Infof("Watch %s event for Namespace %q", strings.Title(string(eventType)), name)
		switch eventType {
		case watch.Added, watch.Modified:
			err = master.vnids.assignVNID(registry, name)
			if err != nil {
				log.Errorf("Error assigning netid: %v", err)
				continue
			}
		case watch.Deleted:
			err = master.vnids.revokeVNID(registry, name)
			if err != nil {
				log.Errorf("Error revoking netid: %v", err)
				continue
			}
		}
	}
}

func (master *OsdnMaster) watchNetNamespaces() {
	registry := master.registry
	eventQueue := registry.RunEventQueue(NetNamespaces)

	for {
		eventType, obj, err := eventQueue.Pop()
		if err != nil {
			utilruntime.HandleError(fmt.Errorf("EventQueue failed for network namespaces: %v", err))
			return
		}
		netns := obj.(*osapi.NetNamespace)
		name := netns.ObjectMeta.Name

		log.V(5).Infof("Watch %s event for NetNamespace %q", strings.Title(string(eventType)), name)
		switch eventType {
		case watch.Added, watch.Modified:
			err = master.vnids.updateVNID(registry, netns)
			if err != nil {
				log.Errorf("Error updating netid: %v", err)
				continue
			}
		}
	}
}