Skip to content

chore: migrate from exit/err channels to context #900

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Jun 23, 2021
Merged

chore: migrate from exit/err channels to context #900

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
bfcc22f
chore: migrate from exit/err channels to context
rbtr Jun 13, 2021
b97f731
pass context instead of storing
rbtr Jun 16, 2021
dc6da75
rename controller packages
rbtr Jun 21, 2021
38050a1
comment init
rbtr Jun 21, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
256 changes: 256 additions & 0 deletions build/tools/go.sum
View file
Open in desktop

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion cns/api.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ const (
// HTTPService describes the min API interface that every service should have.
type HTTPService interface {
common.ServiceAPI
SendNCSnapShotPeriodically(int, chan bool)
SendNCSnapShotPeriodically(context.Context, int)
SetNodeOrchestrator(*SetOrchestratorTypeRequest)
SyncNodeStatus(string, string, string, json.RawMessage) (int, string)
GetPendingProgramIPConfigs() []IPConfigurationStatus
Expand Down
6 changes: 3 additions & 3 deletions cns/common/service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,10 @@ package common

import (
"errors"
"github.com/Azure/azure-container-networking/server/tls"

"github.com/Azure/azure-container-networking/cns/logger"
acn "github.com/Azure/azure-container-networking/common"
"github.com/Azure/azure-container-networking/server/tls"
"github.com/Azure/azure-container-networking/store"
)

Expand All @@ -17,7 +17,7 @@ type Service struct {
Name string
Version string
Options map[string]interface{}
ErrChan chan error
ErrChan chan<- error
Store store.KeyValueStore
ChannelMode string
}
Expand All @@ -36,7 +36,7 @@ type ServiceConfig struct {
Name string
Version string
Listener *acn.Listener
ErrChan chan error
ErrChan chan<- error
Store store.KeyValueStore
ChannelMode string
TlsSettings tls.TlsSettings
Expand Down
18 changes: 5 additions & 13 deletions cns/fakes/cnsfake.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,8 +96,6 @@ func (ipm *IPStateManager) AddIPConfigs(ipconfigs []cns.IPConfigurationStatus) {
ipm.PendingReleaseIPConfigState[ipconfigs[i].ID] = ipconfigs[i]
}
}

return
}

func (ipm *IPStateManager) RemovePendingReleaseIPConfigs(ipconfigNames []string) {
Expand All @@ -107,8 +105,6 @@ func (ipm *IPStateManager) RemovePendingReleaseIPConfigs(ipconfigNames []string)
for i := 0; i < len(ipconfigNames); i++ {
delete(ipm.PendingReleaseIPConfigState, ipconfigNames[i])
}

return
}

func (ipm *IPStateManager) ReserveIPConfig() (cns.IPConfigurationStatus, error) {
Expand Down Expand Up @@ -176,6 +172,8 @@ func (ipm *IPStateManager) MarkIPAsPendingRelease(numberOfIPsToMark int) (map[st
return pendingRelease, nil
}

var _ cns.HTTPService = (*HTTPServiceFake)(nil)

type HTTPServiceFake struct {
IPStateManager IPStateManager
PoolMonitor cns.IPAMPoolMonitor
Expand Down Expand Up @@ -220,22 +218,16 @@ func (fake *HTTPServiceFake) SetNumberOfAllocatedIPs(desiredAllocatedIPCount int
return nil
}

func (fake *HTTPServiceFake) SendNCSnapShotPeriodically(int, chan bool) {

}

func (fake *HTTPServiceFake) SetNodeOrchestrator(*cns.SetOrchestratorTypeRequest) {
func (fake *HTTPServiceFake) SendNCSnapShotPeriodically(context.Context, int) {}

}
func (fake *HTTPServiceFake) SetNodeOrchestrator(*cns.SetOrchestratorTypeRequest) {}

func (fake *HTTPServiceFake) SyncNodeStatus(string, string, string, json.RawMessage) (int, string) {
return 0, ""
}

// SyncHostNCVersion will update HostVersion in containerstatus.
func (fake *HTTPServiceFake) SyncHostNCVersion(context.Context, string, time.Duration) {
return
}
func (fake *HTTPServiceFake) SyncHostNCVersion(context.Context, string, time.Duration) {}

func (fake *HTTPServiceFake) GetPendingProgramIPConfigs() []cns.IPConfigurationStatus {
ipconfigs := []cns.IPConfigurationStatus{}
Expand Down
18 changes: 11 additions & 7 deletions cns/fakes/requestcontrollerfake.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,13 @@ import (
"net"

"github.com/Azure/azure-container-networking/cns"
"github.com/Azure/azure-container-networking/cns/singletenantcontroller"
nnc "github.com/Azure/azure-container-networking/nodenetworkconfig/api/v1alpha"
"github.com/google/uuid"
)

var _ singletenantcontroller.RequestController = (*RequestControllerFake)(nil)

type RequestControllerFake struct {
fakecns *HTTPServiceFake
cachedCRD nnc.NodeNetworkConfig
Expand All @@ -22,9 +25,11 @@ func NewRequestControllerFake(cnsService *HTTPServiceFake, scalar nnc.Scaler, su
Spec: nnc.NodeNetworkConfigSpec{},
Status: nnc.NodeNetworkConfigStatus{
Scaler: scalar,
NetworkContainers: []nnc.NetworkContainer{nnc.NetworkContainer{
SubnetAddressSpace: subnetAddressSpace,
}},
NetworkContainers: []nnc.NetworkContainer{
{
SubnetAddressSpace: subnetAddressSpace,
},
},
},
},
}
Expand Down Expand Up @@ -62,21 +67,20 @@ func (rc *RequestControllerFake) CarveIPConfigsAndAddToStatusAndCNS(numberOfIPCo
return cnsIPConfigs
}

func (rc *RequestControllerFake) InitRequestController() error {
func (rc *RequestControllerFake) Init(context.Context) error {
return nil
}

func (rc *RequestControllerFake) StartRequestController(exitChan <-chan struct{}) error {
func (rc *RequestControllerFake) Start(context.Context) error {
return nil
}

func (rc *RequestControllerFake) IsStarted() bool {
return true
}

func (rc *RequestControllerFake) UpdateCRDSpec(cntxt context.Context, desiredSpec nnc.NodeNetworkConfigSpec) error {
func (rc *RequestControllerFake) UpdateCRDSpec(_ context.Context, desiredSpec nnc.NodeNetworkConfigSpec) error {
rc.cachedCRD.Spec = desiredSpec

return nil
}

Expand Down
53 changes: 20 additions & 33 deletions cns/ipampoolmonitor/ipampoolmonitor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ import (

"github.com/Azure/azure-container-networking/cns"
"github.com/Azure/azure-container-networking/cns/logger"
"github.com/Azure/azure-container-networking/cns/requestcontroller"
"github.com/Azure/azure-container-networking/cns/singletenantcontroller"
nnc "github.com/Azure/azure-container-networking/nodenetworkconfig/api/v1alpha"
)

Expand All @@ -17,21 +17,18 @@ const (
)

type CNSIPAMPoolMonitor struct {
pendingRelease bool

MaximumFreeIps int64
MinimumFreeIps int64
cachedNNC nnc.NodeNetworkConfig
updatingIpsNotInUseCount int
httpService cns.HTTPService
mu sync.RWMutex
pendingRelease bool
rc singletenantcontroller.RequestController
scalarUnits nnc.Scaler

httpService cns.HTTPService
rc requestcontroller.RequestController
MinimumFreeIps int64
MaximumFreeIps int64

mu sync.RWMutex
updatingIpsNotInUseCount int
}

func NewCNSIPAMPoolMonitor(httpService cns.HTTPService, rc requestcontroller.RequestController) *CNSIPAMPoolMonitor {
func NewCNSIPAMPoolMonitor(httpService cns.HTTPService, rc singletenantcontroller.RequestController) *CNSIPAMPoolMonitor {
logger.Printf("NewCNSIPAMPoolMonitor: Create IPAM Pool Monitor")
return &CNSIPAMPoolMonitor{
pendingRelease: false,
Expand All @@ -40,16 +37,6 @@ func NewCNSIPAMPoolMonitor(httpService cns.HTTPService, rc requestcontroller.Req
}
}

func stopReconcile(ch <-chan struct{}) bool {
select {
case <-ch:
return true
default:
}

return false
}

func (pm *CNSIPAMPoolMonitor) Start(ctx context.Context, poolMonitorRefreshMilliseconds int) error {
logger.Printf("[ipam-pool-monitor] Starting CNS IPAM Pool Monitor")

Expand All @@ -60,15 +47,15 @@ func (pm *CNSIPAMPoolMonitor) Start(ctx context.Context, poolMonitorRefreshMilli
case <-ctx.Done():
return fmt.Errorf("[ipam-pool-monitor] CNS IPAM Pool Monitor received cancellation signal")
case <-ticker.C:
err := pm.Reconcile()
err := pm.Reconcile(ctx)
if err != nil {
logger.Printf("[ipam-pool-monitor] Reconcile failed with err %v", err)
}
}
}
}

func (pm *CNSIPAMPoolMonitor) Reconcile() error {
func (pm *CNSIPAMPoolMonitor) Reconcile(ctx context.Context) error {
cnsPodIPConfigCount := len(pm.httpService.GetPodIPConfigState())
pendingProgramCount := len(pm.httpService.GetPendingProgramIPConfigs()) // TODO: add pending program count to real cns
allocatedPodIPCount := len(pm.httpService.GetAllocatedIPConfigs())
Expand All @@ -90,18 +77,18 @@ func (pm *CNSIPAMPoolMonitor) Reconcile() error {
}

logger.Printf("[ipam-pool-monitor] Increasing pool size...%s ", msg)
return pm.increasePoolSize()
return pm.increasePoolSize(ctx)

// pod count is decreasing
case freeIPConfigCount > pm.MaximumFreeIps:
logger.Printf("[ipam-pool-monitor] Decreasing pool size...%s ", msg)
return pm.decreasePoolSize(pendingReleaseIPCount)
return pm.decreasePoolSize(ctx, pendingReleaseIPCount)

// CRD has reconciled CNS state, and target spec is now the same size as the state
// free to remove the IP's from the CRD
case pm.pendingRelease && int(pm.cachedNNC.Spec.RequestedIPCount) == cnsPodIPConfigCount:
logger.Printf("[ipam-pool-monitor] Removing Pending Release IP's from CRD...%s ", msg)
return pm.cleanPendingRelease()
return pm.cleanPendingRelease(ctx)

// no pods scheduled
case allocatedPodIPCount == 0:
Expand All @@ -112,7 +99,7 @@ func (pm *CNSIPAMPoolMonitor) Reconcile() error {
return nil
}

func (pm *CNSIPAMPoolMonitor) increasePoolSize() error {
func (pm *CNSIPAMPoolMonitor) increasePoolSize(ctx context.Context) error {
defer pm.mu.Unlock()
pm.mu.Lock()

Expand Down Expand Up @@ -143,7 +130,7 @@ func (pm *CNSIPAMPoolMonitor) increasePoolSize() error {

logger.Printf("[ipam-pool-monitor] Increasing pool size, Current Pool Size: %v, Updated Requested IP Count: %v, Pods with IP's:%v, ToBeDeleted Count: %v", len(pm.httpService.GetPodIPConfigState()), tempNNCSpec.RequestedIPCount, len(pm.httpService.GetAllocatedIPConfigs()), len(tempNNCSpec.IPsNotInUse))

err = pm.rc.UpdateCRDSpec(context.Background(), tempNNCSpec)
err = pm.rc.UpdateCRDSpec(ctx, tempNNCSpec)
if err != nil {
// caller will retry to update the CRD again
return err
Expand All @@ -155,7 +142,7 @@ func (pm *CNSIPAMPoolMonitor) increasePoolSize() error {
return nil
}

func (pm *CNSIPAMPoolMonitor) decreasePoolSize(existingPendingReleaseIPCount int) error {
func (pm *CNSIPAMPoolMonitor) decreasePoolSize(ctx context.Context, existingPendingReleaseIPCount int) error {
defer pm.mu.Unlock()
pm.mu.Lock()

Expand Down Expand Up @@ -215,7 +202,7 @@ func (pm *CNSIPAMPoolMonitor) decreasePoolSize(existingPendingReleaseIPCount int
tempNNCSpec.RequestedIPCount -= int64(len(pendingIpAddresses))
logger.Printf("[ipam-pool-monitor] Decreasing pool size, Current Pool Size: %v, Requested IP Count: %v, Pods with IP's: %v, ToBeDeleted Count: %v", len(pm.httpService.GetPodIPConfigState()), tempNNCSpec.RequestedIPCount, len(pm.httpService.GetAllocatedIPConfigs()), len(tempNNCSpec.IPsNotInUse))

err = pm.rc.UpdateCRDSpec(context.Background(), tempNNCSpec)
err = pm.rc.UpdateCRDSpec(ctx, tempNNCSpec)
if err != nil {
// caller will retry to update the CRD again
return err
Expand All @@ -236,7 +223,7 @@ func (pm *CNSIPAMPoolMonitor) decreasePoolSize(existingPendingReleaseIPCount int

// if cns pending ip release map is empty, request controller has already reconciled the CNS state,
// so we can remove it from our cache and remove the IP's from the CRD
func (pm *CNSIPAMPoolMonitor) cleanPendingRelease() error {
func (pm *CNSIPAMPoolMonitor) cleanPendingRelease(ctx context.Context) error {
defer pm.mu.Unlock()
pm.mu.Lock()

Expand All @@ -247,7 +234,7 @@ func (pm *CNSIPAMPoolMonitor) cleanPendingRelease() error {
return err
}

err = pm.rc.UpdateCRDSpec(context.Background(), tempNNCSpec)
err = pm.rc.UpdateCRDSpec(ctx, tempNNCSpec)
if err != nil {
// caller will retry to update the CRD again
return err
Expand Down
Loading