identity: Use timed ctx for WaitForInitialGlobalIdentities

* Add error check at the function's invocations.
* Make ip allocation and kvstore timeout configurable

Fixes cilium#8368

Signed-off-by: ifeanyi <ify1992@yahoo.com>

Documentation/cmdref/cilium-agent.md

@@ -76,6 +76,7 @@ cilium-agent [flags]
       --identity-allocation-mode string            Method to use for identity allocation (default "kvstore")
       --identity-change-grace-period duration      Time to wait before using new identity on endpoint identity change (default 5s)
       --install-iptables-rules                     Install base iptables rules for cilium to mainly interact with kube-proxy (and masquerading) (default true)
+      --ip-allocation-timeout duration             Time after which an incomplete CIDR allocation is considered failed (default 2m0s)
       --ipam string                                Backend to use for IPAM
       --ipsec-key-file string                      Path to IPSec key file
       --ipv4-cluster-cidr-mask-size int            Mask size for the cluster wide CIDR (default 8)
@@ -99,6 +100,7 @@ cilium-agent [flags]
       --keep-bpf-templates                         Do not restore BPF template files from binary
       --keep-config                                When restoring state, keeps containers' configuration in place
       --kvstore string                             Key-value store type
+      --kvstore-connectivity-timeout duration      Time after which an incomplete kvstore operation  is considered failed (default 2m0s)
       --kvstore-opt map                            Key-value store options (default map[])
       --kvstore-periodic-sync duration             Periodic KVstore synchronization interval (default 5m0s)
       --label-prefix-file string                   Valid label prefixes file path

daemon/daemon_main.go

@@ -606,6 +606,12 @@ func init() {
 	flags.Duration(option.KVstorePeriodicSync, defaults.KVstorePeriodicSync, "Periodic KVstore synchronization interval")
 	option.BindEnv(option.KVstorePeriodicSync)
 
+	flags.Duration(option.KVstoreConnectivityTimeout, defaults.KVstoreConnectivityTimeout, "Time after which an incomplete kvstore operation  is considered failed")
+	option.BindEnv(option.KVstoreConnectivityTimeout)
+
+	flags.Duration(option.IPAllocationTimeout, defaults.IPAllocationTimeout, "Time after which an incomplete CIDR allocation is considered failed")
+	option.BindEnv(option.IPAllocationTimeout)
+
 	flags.Var(option.NewNamedMapOptions(option.KVStoreOpt, &option.Config.KVStoreOpt, nil),
 		option.KVStoreOpt, "Key-value store options")
 	option.BindEnv(option.KVStoreOpt)

daemon/state.go

@@ -269,7 +269,9 @@ func (d *Daemon) regenerateRestoredEndpoints(state *endpointRestoreState) (resto
 			l, _ := labels.FilterLabels(ep.OpLabels.AllLabels())
 			ep.RUnlock()
 
-			identity, _, err := cache.AllocateIdentity(context.Background(), d, l)
+			allocateCtx, cancel := context.WithTimeout(context.Background(), option.Config.KVstoreConnectivityTimeout)
+			defer cancel()
+			identity, _, err := cache.AllocateIdentity(allocateCtx, d, l)
 
 			if err != nil {
 				scopedLog.WithError(err).Warn("Unable to restore endpoint")
@@ -282,7 +284,14 @@ func (d *Daemon) regenerateRestoredEndpoints(state *endpointRestoreState) (resto
 			// endpoints that don't have a fixed identity or are
 			// not well known.
 			if !identity.IsFixed() && !identity.IsWellKnown() {
-				cache.WaitForInitialGlobalIdentities(context.Background())
+				identityCtx, cancel := context.WithTimeout(context.Background(), option.Config.KVstoreConnectivityTimeout)
+				defer cancel()
+
+				err = cache.WaitForInitialGlobalIdentities(identityCtx)
+				if err != nil {
+					scopedLog.WithError(err).Warn("Failed while waiting for initial global identities")
+					return
+				}
 				ipcache.WaitForInitialSync()
 			}
 

pkg/defaults/defaults.go

@@ -171,6 +171,12 @@ const (
 	// KVstorePeriodicSync is the default kvstore periodic sync interval
 	KVstorePeriodicSync = 5 * time.Minute
 
+	// KVstoreConnectivityTimeout is the timeout when performing kvstore operations
+	KVstoreConnectivityTimeout = 2 * time.Minute
+
+	// IPAllocationTimeout is the timeout when allocating CIDRs
+	IPAllocationTimeout = 2 * time.Minute
+
 	// PolicyQueueSize is the default queue size for policy-related events.
 	PolicyQueueSize = 100
 

pkg/endpoint/endpoint.go

@@ -1336,7 +1336,11 @@ func (e *Endpoint) LeaveLocked(owner regeneration.Owner, proxyWaitGroup *complet
 
 	if !conf.NoIdentityRelease && e.SecurityIdentity != nil {
 		identitymanager.Remove(e.SecurityIdentity)
-		_, err := cache.Release(context.Background(), owner, e.SecurityIdentity)
+
+		releaseCtx, cancel := context.WithTimeout(context.Background(), option.Config.KVstoreConnectivityTimeout)
+		defer cancel()
+
+		_, err := cache.Release(releaseCtx, owner, e.SecurityIdentity)
 		if err != nil {
 			errors = append(errors, fmt.Errorf("unable to release identity: %s", err))
 		}
@@ -1967,7 +1971,10 @@ func (e *Endpoint) identityLabelsChanged(ctx context.Context, owner regeneration
 	e.RUnlock()
 	elog.Debug("Resolving identity for labels")
 
-	identity, _, err := cache.AllocateIdentity(ctx, owner, newLabels)
+	allocateCtx, cancel := context.WithTimeout(context.Background(), option.Config.KVstoreConnectivityTimeout)
+	defer cancel()
+
+	identity, _, err := cache.AllocateIdentity(allocateCtx, owner, newLabels)
 	if err != nil {
 		err = fmt.Errorf("unable to resolve identity: %s", err)
 		e.LogStatus(Other, Warning, fmt.Sprintf("%s (will retry)", err.Error()))

pkg/identity/cache/allocator.go

@@ -300,7 +300,10 @@ func AllocateIdentity(ctx context.Context, owner IdentityAllocatorOwner, lbls la
 
 	// This will block until the kvstore can be accessed and all identities
 	// were successfully synced
-	WaitForInitialGlobalIdentities(ctx)
+	err = WaitForInitialGlobalIdentities(ctx)
+	if err != nil {
+		return nil, false, err
+	}
 
 	if IdentityAllocator == nil {
 		return nil, false, fmt.Errorf("allocator not initialized")
@@ -352,7 +355,10 @@ func Release(ctx context.Context, owner IdentityAllocatorOwner, id *identity.Ide
 
 	// This will block until the kvstore can be accessed and all identities
 	// were successfully synced
-	WaitForInitialGlobalIdentities(ctx)
+	err = WaitForInitialGlobalIdentities(ctx)
+	if err != nil {
+		return false, err
+	}
 
 	if IdentityAllocator == nil {
 		return false, fmt.Errorf("allocator not initialized")

pkg/ipcache/cidr.go

@@ -24,6 +24,7 @@ import (
 	"github.com/cilium/cilium/pkg/ip"
 	"github.com/cilium/cilium/pkg/labels"
 	"github.com/cilium/cilium/pkg/labels/cidr"
+	"github.com/cilium/cilium/pkg/option"
 )
 
 // AllocateCIDRs attempts to allocate identities for a list of CIDRs. If any
@@ -61,8 +62,12 @@ func allocateCIDRs(prefixes []*net.IPNet) ([]*identity.Identity, error) {
 		}
 
 		prefixStr := prefix.String()
+
 		// Figure out if this call needs to be able to update the selector cache synchronously.
-		id, isNew, err := cache.AllocateIdentity(context.Background(), nil, cidr.GetCIDRLabels(prefix))
+		allocateCtx, cancel := context.WithTimeout(context.Background(), option.Config.IPAllocationTimeout)
+		defer cancel()
+
+		id, isNew, err := cache.AllocateIdentity(allocateCtx, nil, cidr.GetCIDRLabels(prefix))
 		if err != nil {
 			cache.ReleaseSlice(context.Background(), nil, usedIdentities)
 			return nil, fmt.Errorf("failed to allocate identity for cidr %s: %s", prefixStr, err)
@@ -104,7 +109,10 @@ func ReleaseCIDRs(prefixes []*net.IPNet) {
 		}
 
 		if id := cache.LookupIdentity(cidr.GetCIDRLabels(prefix)); id != nil {
-			released, err := cache.Release(context.Background(), nil, id)
+			releaseCtx, cancel := context.WithTimeout(context.Background(), option.Config.KVstoreConnectivityTimeout)
+			defer cancel()
+
+			released, err := cache.Release(releaseCtx, nil, id)
 			if err != nil {
 				log.WithError(err).Warningf("Unable to release identity for CIDR %s. Ignoring error. Identity may be leaked", prefix.String())
 			}

pkg/option/config.go

@@ -452,6 +452,12 @@ const (
 	// synchronization with the kvstore occurs
 	KVstorePeriodicSync = "kvstore-periodic-sync"
 
+	// KVstoreConnectivityTimeout is the timeout when performing kvstore operations
+	KVstoreConnectivityTimeout = "kvstore-connectivity-timeout"
+
+	// IPAllocationTimeout is the timeout when allocating CIDRs
+	IPAllocationTimeout = "ip-allocation-timeout"
+
 	// IdentityChangeGracePeriod is the name of the
 	// IdentityChangeGracePeriod option
 	IdentityChangeGracePeriod = "identity-change-grace-period"
@@ -1013,6 +1019,12 @@ type DaemonConfig struct {
 	// synchronization with the kvstore occurs
 	KVstorePeriodicSync time.Duration
 
+	// KVstoreConnectivityTimeout is the timeout when performing kvstore operations
+	KVstoreConnectivityTimeout time.Duration
+
+	// IPAllocationTimeout is the timeout when allocating CIDRs
+	IPAllocationTimeout time.Duration
+
 	// IdentityChangeGracePeriod is the grace period that needs to pass
 	// before an endpoint that has changed its identity will start using
 	// that new identity. During the grace period, the new identity has
@@ -1144,6 +1156,8 @@ var (
 		EnableIPv6:                   defaults.EnableIPv6,
 		ToFQDNsMaxIPsPerHost:         defaults.ToFQDNsMaxIPsPerHost,
 		KVstorePeriodicSync:          defaults.KVstorePeriodicSync,
+		KVstoreConnectivityTimeout:   defaults.KVstoreConnectivityTimeout,
+		IPAllocationTimeout:          defaults.IPAllocationTimeout,
 		IdentityChangeGracePeriod:    defaults.IdentityChangeGracePeriod,
 		ContainerRuntimeEndpoint:     make(map[string]string),
 		FixedIdentityMapping:         make(map[string]string),
@@ -1496,6 +1510,8 @@ func (c *DaemonConfig) Populate() {
 	c.KVstoreLeaseTTL = viper.GetDuration(KVstoreLeaseTTL)
 	c.KVstoreKeepAliveInterval = c.KVstoreLeaseTTL / defaults.KVstoreKeepAliveIntervalFactor
 	c.KVstorePeriodicSync = viper.GetDuration(KVstorePeriodicSync)
+	c.KVstoreConnectivityTimeout = viper.GetDuration(KVstoreConnectivityTimeout)
+	c.IPAllocationTimeout = viper.GetDuration(IPAllocationTimeout)
 	c.LabelPrefixFile = viper.GetString(LabelPrefixFile)
 	c.Labels = viper.GetStringSlice(Labels)
 	c.LBInterface = viper.GetString(LB)