fix: hanging the adc program can cause the k8s client to exceed its limit (#104)

Author: AlinsRan

config/samples/config.yaml

@@ -15,3 +15,6 @@ leader_election:
   retry_period: 2s                      # retry_period is the time in seconds that the acting controller
                                         # will wait between tries of actions with the controller.
   disable: false                        # Whether to disable leader election.
+
+exec_adc_timeout: 15s                   # The timeout for the ADC to execute.
+                                        # The default value is 15 seconds.

examples/httpbin/ingress.yaml

@@ -34,7 +34,7 @@ spec:
 apiVersion: networking.k8s.io/v1
 kind: Ingress
 metadata:
-  name: api7-ingress-tls
+  name: api7-ingress
 spec:
   ingressClassName: api7
   rules:

internal/controller/config/config.go

@@ -31,6 +31,7 @@ func NewDefaultConfig() *Config {
 		ProbeAddr:        DefaultProbeAddr,
 		MetricsAddr:      DefaultMetricsAddr,
 		LeaderElection:   NewLeaderElection(),
+		ExecADCTimeout:   types.TimeDuration{Duration: 15 * time.Second},
 	}
 }
 

internal/controller/config/types.go

@@ -24,16 +24,17 @@ const (
 // Config contains all config items which are necessary for
 // apisix-ingress-controller's running.
 type Config struct {
-	CertFilePath     string          `json:"cert_file" yaml:"cert_file"`
-	KeyFilePath      string          `json:"key_file" yaml:"key_file"`
-	LogLevel         string          `json:"log_level" yaml:"log_level"`
-	ControllerName   string          `json:"controller_name" yaml:"controller_name"`
-	LeaderElectionID string          `json:"leader_election_id" yaml:"leader_election_id"`
-	MetricsAddr      string          `json:"metrics_addr" yaml:"metrics_addr"`
-	EnableHTTP2      bool            `json:"enable_http2" yaml:"enable_http2"`
-	ProbeAddr        string          `json:"probe_addr" yaml:"probe_addr"`
-	SecureMetrics    bool            `json:"secure_metrics" yaml:"secure_metrics"`
-	LeaderElection   *LeaderElection `json:"leader_election" yaml:"leader_election"`
+	CertFilePath     string             `json:"cert_file" yaml:"cert_file"`
+	KeyFilePath      string             `json:"key_file" yaml:"key_file"`
+	LogLevel         string             `json:"log_level" yaml:"log_level"`
+	ControllerName   string             `json:"controller_name" yaml:"controller_name"`
+	LeaderElectionID string             `json:"leader_election_id" yaml:"leader_election_id"`
+	MetricsAddr      string             `json:"metrics_addr" yaml:"metrics_addr"`
+	EnableHTTP2      bool               `json:"enable_http2" yaml:"enable_http2"`
+	ProbeAddr        string             `json:"probe_addr" yaml:"probe_addr"`
+	SecureMetrics    bool               `json:"secure_metrics" yaml:"secure_metrics"`
+	LeaderElection   *LeaderElection    `json:"leader_election" yaml:"leader_election"`
+	ExecADCTimeout   types.TimeDuration `json:"exec_adc_timeout" yaml:"exec_adc_timeout"`
 }
 
 type GatewayConfig struct {

internal/provider/adc/adc.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"os/exec"
 	"sync"
+	"time"
 
 	"go.uber.org/zap"
 	networkingv1 "k8s.io/api/networking/v1"
@@ -16,6 +17,7 @@ import (
 
 	adctypes "github.com/api7/api7-ingress-controller/api/adc"
 	"github.com/api7/api7-ingress-controller/api/v1alpha1"
+	"github.com/api7/api7-ingress-controller/internal/controller/config"
 	"github.com/api7/api7-ingress-controller/internal/controller/label"
 	"github.com/api7/api7-ingress-controller/internal/provider"
 	"github.com/api7/api7-ingress-controller/internal/provider/adc/translator"
@@ -38,6 +40,8 @@ type adcClient struct {
 	configs map[provider.ResourceKind]adcConfig
 	// httproute/consumer/ingress/gateway -> gateway/ingressclass
 	parentRefs map[provider.ResourceKind][]provider.ResourceKind
+
+	syncTimeout time.Duration
 }
 
 type Task struct {
@@ -50,9 +54,10 @@ type Task struct {
 
 func New() (provider.Provider, error) {
 	return &adcClient{
-		translator: &translator.Translator{},
-		configs:    make(map[provider.ResourceKind]adcConfig),
-		parentRefs: make(map[provider.ResourceKind][]provider.ResourceKind),
+		syncTimeout: config.ControllerConfig.ExecADCTimeout.Duration,
+		translator:  &translator.Translator{},
+		configs:     make(map[provider.ResourceKind]adcConfig),
+		parentRefs:  make(map[provider.ResourceKind][]provider.ResourceKind),
 	}, nil
 }
 
@@ -101,7 +106,7 @@ func (d *adcClient) Update(ctx context.Context, tctx *provider.TranslateContext,
 
 	// sync delete
 	if len(deleteConfigs) > 0 {
-		err = d.sync(Task{
+		err = d.sync(ctx, Task{
 			Name:          obj.GetName(),
 			Labels:        label.GenLabel(obj),
 			ResourceTypes: resourceTypes,
@@ -113,7 +118,7 @@ func (d *adcClient) Update(ctx context.Context, tctx *provider.TranslateContext,
 	}
 
 	// sync update
-	err = d.sync(Task{
+	err = d.sync(ctx, Task{
 		Name:   obj.GetName(),
 		Labels: label.GenLabel(obj),
 		Resources: adctypes.Resources{
@@ -160,7 +165,7 @@ func (d *adcClient) Delete(ctx context.Context, obj client.Object) error {
 
 	configs := d.getConfigs(rk)
 
-	err := d.sync(Task{
+	err := d.sync(ctx, Task{
 		Name:          obj.GetName(),
 		Labels:        labels,
 		ResourceTypes: resourceTypes,
@@ -174,7 +179,7 @@ func (d *adcClient) Delete(ctx context.Context, obj client.Object) error {
 	return nil
 }
 
-func (d *adcClient) sync(task Task) error {
+func (d *adcClient) sync(ctx context.Context, task Task) error {
 	log.Debugw("syncing resources", zap.Any("task", task))
 
 	if len(task.configs) == 0 {
@@ -216,15 +221,17 @@ func (d *adcClient) sync(task Task) error {
 
 	log.Debugw("syncing resources with multiple configs", zap.Any("configs", task.configs))
 	for _, config := range task.configs {
-		if err := d.execADC(config, args); err != nil {
+		if err := d.execADC(ctx, config, args); err != nil {
 			return err
 		}
 	}
 
 	return nil
 }
 
-func (d *adcClient) execADC(config adcConfig, args []string) error {
+func (d *adcClient) execADC(ctx context.Context, config adcConfig, args []string) error {
+	ctxWithTimeout, cancel := context.WithTimeout(ctx, d.syncTimeout)
+	defer cancel()
 	// todo: use adc config
 	serverAddr := d.ServerAddr
 	if config.ServerAddr != "" {
@@ -244,7 +251,7 @@ func (d *adcClient) execADC(config adcConfig, args []string) error {
 	}
 
 	var stdout, stderr bytes.Buffer
-	cmd := exec.Command("adc", args...)
+	cmd := exec.CommandContext(ctxWithTimeout, "adc", args...)
 	cmd.Stdout = &stdout
 	cmd.Stderr = &stderr
 	cmd.Env = append(cmd.Env, os.Environ()...)