feat: update Ingress LB status (#740)

cmd/ingress/ingress.go

@@ -140,6 +140,14 @@ the apisix cluster and others are created`,
 	cmd.PersistentFlags().StringVar(&cfg.LogOutput, "log-output", "stderr", "error log output file")
 	cmd.PersistentFlags().StringVar(&cfg.HTTPListen, "http-listen", ":8080", "the HTTP Server listen address")
 	cmd.PersistentFlags().StringVar(&cfg.HTTPSListen, "https-listen", ":8443", "the HTTPS Server listen address")
+	cmd.PersistentFlags().StringVar(&cfg.IngressPublishService, "ingress-publish-service", "",
+		`the controller will use the Endpoint of this Service to update the status information of the Ingress resource. 
+The format is "namespace/svc-name" to solve the situation that the data plane and the controller are not deployed in the same namespace.`)
+	cmd.PersistentFlags().StringSliceVar(&cfg.IngressStatusAddress, "ingress-status-address", []string{},
+		`when there is no available information on the Service used for publishing on the data plane,
+the static address provided here will be used to update the status information of Ingress.
+When ingress-publish-service is specified at the same time, ingress-status-address is preferred.
+For example, no available LB exists in the bare metal environment.`)
 	cmd.PersistentFlags().BoolVar(&cfg.EnableProfiling, "enable-profiling", true, "enable profiling via web interface host:port/debug/pprof")
 	cmd.PersistentFlags().StringVar(&cfg.Kubernetes.Kubeconfig, "kubeconfig", "", "Kubernetes configuration file (by default in-cluster configuration will be used)")
 	cmd.PersistentFlags().DurationVar(&cfg.Kubernetes.ResyncInterval.Duration, "resync-interval", time.Minute, "the controller resync (with Kubernetes) interval, the minimum resync interval is 30s")

conf/config-default.yaml

@@ -32,6 +32,16 @@ key_file: "/etc/webhook/certs/key.pem"   # the TLS key file path.
 
 http_listen: ":8080"   # the HTTP Server listen address, default is ":8080"
 https_listen: ":8443"   # the HTTPS Server listen address, default is ":8443"
+ingress_publish_service: ""  # the controller will use the Endpoint of this Service to
+                             # update the status information of the Ingress resource.
+                             # The format is "namespace/svc-name" to solve the situation that
+                             # the data plane and the controller are not deployed in the same namespace.
+ingress_status_address: []   # when there is no available information on the Service
+                             # used for publishing on the data plane,
+                             # the static address provided here will be
+                             # used to update the status information of Ingress.
+                             # When ingress-publish-service is specified at the same time, ingress-status-address is preferred.
+                             # For example, no available LB exists in the bare metal environment.
 enable_profiling: true # enable profiling via web interfaces
                        # host:port/debug/pprof, default is true.
 

pkg/config/config.go

@@ -66,15 +66,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"`
-	LogOutput       string           `json:"log_output" yaml:"log_output"`
-	HTTPListen      string           `json:"http_listen" yaml:"http_listen"`
-	HTTPSListen     string           `json:"https_listen" yaml:"https_listen"`
-	EnableProfiling bool             `json:"enable_profiling" yaml:"enable_profiling"`
-	Kubernetes      KubernetesConfig `json:"kubernetes" yaml:"kubernetes"`
-	APISIX          APISIXConfig     `json:"apisix" yaml:"apisix"`
+	CertFilePath          string           `json:"cert_file" yaml:"cert_file"`
+	KeyFilePath           string           `json:"key_file" yaml:"key_file"`
+	LogLevel              string           `json:"log_level" yaml:"log_level"`
+	LogOutput             string           `json:"log_output" yaml:"log_output"`
+	HTTPListen            string           `json:"http_listen" yaml:"http_listen"`
+	HTTPSListen           string           `json:"https_listen" yaml:"https_listen"`
+	IngressPublishService string           `json:"ingress_publish_service" yaml:"ingress_publish_service"`
+	IngressStatusAddress  []string         `json:"ingress_status_address" yaml:"ingress_status_address"`
+	EnableProfiling       bool             `json:"enable_profiling" yaml:"enable_profiling"`
+	Kubernetes            KubernetesConfig `json:"kubernetes" yaml:"kubernetes"`
+	APISIX                APISIXConfig     `json:"apisix" yaml:"apisix"`
 }
 
 // KubernetesConfig contains all Kubernetes related config items.
@@ -113,13 +115,15 @@ type APISIXConfig struct {
 // default value.
 func NewDefaultConfig() *Config {
 	return &Config{
-		LogLevel:        "warn",
-		LogOutput:       "stderr",
-		HTTPListen:      ":8080",
-		HTTPSListen:     ":8443",
-		CertFilePath:    "/etc/webhook/certs/cert.pem",
-		KeyFilePath:     "/etc/webhook/certs/key.pem",
-		EnableProfiling: true,
+		LogLevel:              "warn",
+		LogOutput:             "stderr",
+		HTTPListen:            ":8080",
+		HTTPSListen:           ":8443",
+		IngressPublishService: "",
+		IngressStatusAddress:  []string{},
+		CertFilePath:          "/etc/webhook/certs/cert.pem",
+		KeyFilePath:           "/etc/webhook/certs/key.pem",
+		EnableProfiling:       true,
 		Kubernetes: KubernetesConfig{
 			Kubeconfig:          "", // Use in-cluster configurations.
 			ResyncInterval:      types.TimeDuration{Duration: 6 * time.Hour},

pkg/config/config_test.go

@@ -28,13 +28,15 @@ import (
 
 func TestNewConfigFromFile(t *testing.T) {
 	cfg := &Config{
-		LogLevel:        "warn",
-		LogOutput:       "stdout",
-		HTTPListen:      ":9090",
-		HTTPSListen:     ":9443",
-		CertFilePath:    "/etc/webhook/certs/cert.pem",
-		KeyFilePath:     "/etc/webhook/certs/key.pem",
-		EnableProfiling: true,
+		LogLevel:              "warn",
+		LogOutput:             "stdout",
+		HTTPListen:            ":9090",
+		HTTPSListen:           ":9443",
+		IngressPublishService: "",
+		IngressStatusAddress:  []string{},
+		CertFilePath:          "/etc/webhook/certs/cert.pem",
+		KeyFilePath:           "/etc/webhook/certs/key.pem",
+		EnableProfiling:       true,
 		Kubernetes: KubernetesConfig{
 			ResyncInterval:     types.TimeDuration{Duration: time.Hour},
 			Kubeconfig:         "/path/to/foo/baz",
@@ -77,6 +79,8 @@ log_level: warn
 log_output: stdout
 http_listen: :9090
 https_listen: :9443
+ingress_publish_service: ""
+ingress_status_address: []
 enable_profiling: true
 kubernetes:
   kubeconfig: /path/to/foo/baz
@@ -105,13 +109,15 @@ apisix:
 
 func TestConfigWithEnvVar(t *testing.T) {
 	cfg := &Config{
-		LogLevel:        "warn",
-		LogOutput:       "stdout",
-		HTTPListen:      ":9090",
-		HTTPSListen:     ":9443",
-		CertFilePath:    "/etc/webhook/certs/cert.pem",
-		KeyFilePath:     "/etc/webhook/certs/key.pem",
-		EnableProfiling: true,
+		LogLevel:              "warn",
+		LogOutput:             "stdout",
+		HTTPListen:            ":9090",
+		HTTPSListen:           ":9443",
+		IngressPublishService: "",
+		IngressStatusAddress:  []string{},
+		CertFilePath:          "/etc/webhook/certs/cert.pem",
+		KeyFilePath:           "/etc/webhook/certs/key.pem",
+		EnableProfiling:       true,
 		Kubernetes: KubernetesConfig{
 			ResyncInterval:     types.TimeDuration{Duration: time.Hour},
 			Kubeconfig:         "",
@@ -143,6 +149,8 @@ func TestConfigWithEnvVar(t *testing.T) {
     "log_output": "stdout",
     "http_listen": ":9090",
     "https_listen": ":9443",
+	"ingress_publish_service": "",
+	"ingress_status_address": [],
     "enable_profiling": true,
     "kubernetes": {
         "kubeconfig": "{{.KUBECONFIG}}",
@@ -176,6 +184,8 @@ log_level: warn
 log_output: stdout
 http_listen: :9090
 https_listen: :9443
+ingress_publish_service: ""
+ingress_status_address: []
 enable_profiling: true
 kubernetes:
   resync_interval: 1h0m0s

pkg/ingress/apisix_route.go

@@ -133,7 +133,7 @@ func (c *apisixRouteController) sync(ctx context.Context, ev *types.Event) error
 		}
 		ar = ev.Tombstone.(kube.ApisixRoute)
 	}
-	//
+
 	switch obj.GroupVersion {
 	case kube.ApisixRouteV1:
 		tctx, err = c.controller.translator.TranslateRouteV1(ar.V1())

pkg/ingress/ingress.go

@@ -21,6 +21,7 @@ import (
 
 	"go.uber.org/zap"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
 
@@ -184,7 +185,42 @@ func (c *ingressController) sync(ctx context.Context, ev *types.Event) error {
 }
 
 func (c *ingressController) handleSyncErr(obj interface{}, err error) {
+	ev := obj.(*types.Event)
+	event := ev.Object.(kube.IngressEvent)
+	namespace, name, errLocal := cache.SplitMetaNamespaceKey(event.Key)
+	if errLocal != nil {
+		log.Errorf("invalid resource key: %s", event.Key)
+		return
+	}
+
+	var ing kube.Ingress
+	switch event.GroupVersion {
+	case kube.IngressV1:
+		ing, err = c.controller.ingressLister.V1(namespace, name)
+	case kube.IngressV1beta1:
+		ing, err = c.controller.ingressLister.V1beta1(namespace, name)
+	case kube.IngressExtensionsV1beta1:
+		ing, err = c.controller.ingressLister.ExtensionsV1beta1(namespace, name)
+	}
+
 	if err == nil {
+		// add status
+		if ev.Type != types.EventDelete {
+			if errLocal == nil {
+				switch ing.GroupVersion() {
+				case kube.IngressV1:
+					c.controller.recordStatus(ing.V1(), _resourceSynced, nil, metav1.ConditionTrue, ing.V1().GetGeneration())
+				case kube.IngressV1beta1:
+					c.controller.recordStatus(ing.V1beta1(), _resourceSynced, nil, metav1.ConditionTrue, ing.V1beta1().GetGeneration())
+				case kube.IngressExtensionsV1beta1:
+					c.controller.recordStatus(ing.ExtensionsV1beta1(), _resourceSynced, nil, metav1.ConditionTrue, ing.ExtensionsV1beta1().GetGeneration())
+				}
+			} else {
+				log.Errorw("failed split namespace/name",
+					zap.Error(errLocal),
+				)
+			}
+		}
 		c.workqueue.Forget(obj)
 		c.controller.MetricsCollector.IncrSyncOperation("ingress", "success")
 		return
@@ -193,6 +229,21 @@ func (c *ingressController) handleSyncErr(obj interface{}, err error) {
 		zap.Any("object", obj),
 		zap.Error(err),
 	)
+
+	if errLocal == nil {
+		switch ing.GroupVersion() {
+		case kube.IngressV1:
+			c.controller.recordStatus(ing.V1(), _resourceSyncAborted, err, metav1.ConditionTrue, ing.V1().GetGeneration())
+		case kube.IngressV1beta1:
+			c.controller.recordStatus(ing.V1beta1(), _resourceSyncAborted, err, metav1.ConditionTrue, ing.V1beta1().GetGeneration())
+		case kube.IngressExtensionsV1beta1:
+			c.controller.recordStatus(ing.ExtensionsV1beta1(), _resourceSyncAborted, err, metav1.ConditionTrue, ing.ExtensionsV1beta1().GetGeneration())
+		}
+	} else {
+		log.Errorw("failed split namespace/name",
+			zap.Error(errLocal),
+		)
+	}
 	c.workqueue.AddRateLimited(obj)
 	c.controller.MetricsCollector.IncrSyncOperation("ingress", "failure")
 }