Merge pull request #106 from fdfzcq/auto-reattach-negs

Make it possible to reattach NEGs automatically

README.md

@@ -82,6 +82,10 @@ The controller parameters can be customized via changing the [controller deploym
   to false. The template defaults to `{name}-{port}`. It can contain `namespace`, `name`, `port` and `hash` and the non-hash values will be 
   truncated evenly if the full name is longer than 63 characters. `<hash>` is generated using full length `namespace`, `name` and 
   `port` to avoid name collisions when truncated.
+* `--max-rate-per-endpoint`: optional. Sets a default value for max-rate-per-endpoint that can be overridden by user config. Defaults to 0.
+* `--max-connections-per-endpoint`: optional. Same as above but for connections.
+* `--always-reconcile`: optional. Makes it possible to reconcile periodically even if the status annotations don't change. Defaults to false.
+* `--reconcile-period`: optional. Sets a reconciliation duration if always-reconcile mode is on. Defaults to 10 hours.
 
 ## IAM considerations
 

controllers/autoneg.go

@@ -91,14 +91,14 @@ func (s AutonegStatus) Backend(name string, port string, group string) compute.B
 }
 
 // NewBackendController takes the project name and an initialized *compute.Service
-func NewBackendController(project string, s *compute.Service) *BackendController {
-	return &BackendController{
+func NewBackendController(project string, s *compute.Service) *ProdBackendController {
+	return &ProdBackendController{
 		project: project,
 		s:       s,
 	}
 }
 
-func (b *BackendController) getBackendService(name string, region string) (svc *compute.BackendService, err error) {
+func (b *ProdBackendController) getBackendService(name string, region string) (svc *compute.BackendService, err error) {
 	if region == "" {
 		svc, err = compute.NewBackendServicesService(b.s).Get(b.project, name).Do()
 		if e, ok := err.(*googleapi.Error); ok {
@@ -118,7 +118,7 @@ func (b *BackendController) getBackendService(name string, region string) (svc *
 
 }
 
-func (b *BackendController) updateBackends(name string, region string, svc *compute.BackendService) error {
+func (b *ProdBackendController) updateBackends(name string, region string, svc *compute.BackendService) error {
 	if len(svc.Backends) == 0 {
 		svc.NullFields = []string{"Backends"}
 	}
@@ -181,10 +181,7 @@ func checkOperation(op *compute.Operation) error {
 
 // ReconcileBackends takes the actual and intended AutonegStatus
 // and attempts to apply the intended status or return an error
-func (b *BackendController) ReconcileBackends(actual, intended AutonegStatus) (err error) {
-	if b.s == nil { // test suite
-		return nil
-	}
+func (b *ProdBackendController) ReconcileBackends(actual, intended AutonegStatus) (err error) {
 	removes, upserts := ReconcileStatus(b.project, actual, intended)
 
 	for port, _removes := range removes {

controllers/controller_test.go

@@ -77,6 +77,16 @@ var _ = Describe("Run autoneg Controller", func() {
 			Expect(updatedAnnos[negStatusAnnotation]).To(BeEmpty())
 		})
 
+		Context("Reconciles periodically", func() {
+
+			It("should reconcile", func() {
+				timesReconciled := backendController.Counter
+				time.Sleep(2 * time.Second)
+				Expect(backendController.Counter-timesReconciled > 0).To(BeTrue(), "should have at least reconciled once.")
+			})
+
+		})
+
 		Context("Remove the service", func() {
 
 			It("should succeed", func() {

controllers/service_controller.go

@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"errors"
 	"reflect"
+	"time"
 
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -32,16 +33,22 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
 
+type BackendController interface {
+	ReconcileBackends(AutonegStatus, AutonegStatus) error
+}
+
 // ServiceReconciler reconciles a Service object
 type ServiceReconciler struct {
 	client.Client
 	Scheme *runtime.Scheme
-	*BackendController
+	BackendController
 	Recorder                         record.EventRecorder
 	ServiceNameTemplate              string
 	AllowServiceName                 bool
 	MaxRatePerEndpointDefault        float64
 	MaxConnectionsPerEndpointDefault float64
+	AlwaysReconcile                  bool
+	ReconcileDuration                *time.Duration
 }
 
 //+kubebuilder:rbac:groups=core,resources=services,verbs=get;list;watch;update;patch
@@ -66,21 +73,21 @@ func (r *ServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	if err != nil {
 		if apierrors.IsNotFound(err) {
 			// Object not found, return.
-			return reconcile.Result{}, nil
+			return r.reconcileResult(nil)
 		}
 		// Error reading the object - requeue the request.
-		return reconcile.Result{}, err
+		return r.reconcileResult(err)
 	}
 
 	status, ok, err := getStatuses(svc.Namespace, svc.Name, svc.ObjectMeta.Annotations, r)
 	// Is this service using autoneg?
 	if !ok {
-		return reconcile.Result{}, nil
+		return r.reconcileResult(nil)
 	}
 
 	if err != nil {
 		r.Recorder.Event(svc, "Warning", "ConfigError", err.Error())
-		return reconcile.Result{}, err
+		return r.reconcileResult(err)
 	}
 
 	deleting := false
@@ -102,9 +109,9 @@ func (r *ServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 
 	if deleting {
 		intendedStatus.BackendServices = make(map[string]map[string]AutonegNEGConfig, 0)
-	} else if reflect.DeepEqual(status.status, intendedStatus) {
+	} else if reflect.DeepEqual(status.status, intendedStatus) && !r.AlwaysReconcile {
 		// Equal, no reconciliation necessary
-		return reconcile.Result{}, nil
+		return r.reconcileResult(nil)
 	}
 
 	// Reconcile differences
@@ -114,11 +121,11 @@ func (r *ServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		var e *errNotFound
 		if !(deleting && errors.As(err, &e)) {
 			r.Recorder.Event(svc, "Warning", "BackendError", err.Error())
-			return reconcile.Result{}, err
+			return r.reconcileResult(err)
 		}
 		if deleting {
 			r.Recorder.Event(svc, "Warning", "BackendError while deleting", err.Error())
-			return reconcile.Result{}, err
+			return r.reconcileResult(err)
 		}
 	}
 
@@ -147,7 +154,7 @@ func (r *ServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		anStatus, err := json.Marshal(intendedStatus)
 		if err != nil {
 			logger.Error(err, "json marshal error")
-			return reconcile.Result{}, err
+			return r.reconcileResult(err)
 		}
 		svc.ObjectMeta.Annotations[autonegStatusAnnotation] = string(anStatus)
 
@@ -156,7 +163,7 @@ func (r *ServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 
 			if err != nil {
 				logger.Error(err, "json marshal error")
-				return reconcile.Result{}, err
+				return r.reconcileResult(err)
 			}
 
 			svc.ObjectMeta.Annotations[oldAutonegStatusAnnotation] = string(oldStatus)
@@ -168,7 +175,7 @@ func (r *ServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		if !apierrors.IsConflict(err) {
 			r.Recorder.Event(svc, "Warning", "BackendError", err.Error())
 		}
-		return reconcile.Result{}, err
+		return r.reconcileResult(err)
 	}
 
 	for port, endpointGroups := range intendedStatus.BackendServices {
@@ -190,7 +197,7 @@ func (r *ServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		}
 	}
 
-	return reconcile.Result{}, nil
+	return r.reconcileResult(nil)
 }
 
 // SetupWithManager sets up the controller with the Manager.
@@ -219,3 +226,10 @@ func removeString(slice []string, s string) (result []string) {
 	}
 	return
 }
+
+func (r *ServiceReconciler) reconcileResult(err error) (reconcile.Result, error) {
+	if r.ReconcileDuration != nil && r.AlwaysReconcile {
+		return reconcile.Result{RequeueAfter: *r.ReconcileDuration}, err
+	}
+	return reconcile.Result{}, err
+}

controllers/suite_test.go

@@ -18,14 +18,15 @@ package controllers
 
 import (
 	"context"
+	"fmt"
 	"path/filepath"
 	"testing"
+	"time"
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/kubernetes/scheme"
-	"k8s.io/client-go/rest"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
@@ -37,11 +38,11 @@ import (
 // These tests use Ginkgo (BDD-style Go testing framework). Refer to
 // http://onsi.github.io/ginkgo/ to learn more about Ginkgo.
 
-var cfg *rest.Config
 var k8sClient client.Client
 var k8sManager ctrl.Manager
 var testEnv *envtest.Environment
 var cancel context.CancelFunc
+var backendController *TestBackendController
 
 func TestAPIs(t *testing.T) {
 	RegisterFailHandler(Fail)
@@ -79,12 +80,17 @@ var _ = BeforeSuite(func() {
 	})
 	Expect(err).ToNot(HaveOccurred())
 
+	backendController = &TestBackendController{Counter: 0}
+	duration := 1 * time.Second
+
 	err = (&ServiceReconciler{
 		Client:              k8sManager.GetClient(),
-		BackendController:   &BackendController{},
+		BackendController:   backendController,
 		Recorder:            k8sManager.GetEventRecorderFor("autoneg-controller"),
 		ServiceNameTemplate: serviceNameTemplate,
 		AllowServiceName:    true,
+		AlwaysReconcile:     true,
+		ReconcileDuration:   &duration,
 	}).SetupWithManager(k8sManager)
 	Expect(err).ToNot(HaveOccurred())
 
@@ -95,6 +101,16 @@ var _ = BeforeSuite(func() {
 	}()
 }, 60)
 
+type TestBackendController struct {
+	Counter int
+}
+
+func (t *TestBackendController) ReconcileBackends(AutonegStatus, AutonegStatus) error {
+	t.Counter++
+	fmt.Print(t.Counter)
+	return nil
+}
+
 var _ = AfterSuite(func() {
 	cancel()
 	By("tearing down the test environment")

controllers/types.go

@@ -85,7 +85,7 @@ type Backends struct {
 }
 
 // BackendController manages operations on a GCLB backend service
-type BackendController struct {
+type ProdBackendController struct {
 	project string
 	s       *compute.Service
 }

main.go

@@ -21,6 +21,7 @@ import (
 	"flag"
 	"fmt"
 	"os"
+	"time"
 
 	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
 	// to ensure that exec-entrypoint and run can make use of them.
@@ -61,17 +62,21 @@ func main() {
 	var enableLeaderElection bool
 	var serviceNameTemplate string
 	var allowServiceName bool
+	var alwaysReconcile bool
+	var reconcilePeriod string
 	flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.")
 	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
-	flag.Float64Var(&maxRatePerEndpointDefault, "max-rate-per-endpoint", 0, "The address the probe endpoint binds to.")
-	flag.Float64Var(&maxConnectionsPerEndpointDefault, "max-connections-per-endpoint", 0, "The address the probe endpoint binds to.")
+	flag.Float64Var(&maxRatePerEndpointDefault, "max-rate-per-endpoint", 0, "Default max rate per endpoint. Can be overridden by user config.")
+	flag.Float64Var(&maxConnectionsPerEndpointDefault, "max-connections-per-endpoint", 0, "Default max connections per endpoint. Can be overridden by user config.")
 	flag.BoolVar(&enableLeaderElection, "leader-elect", false,
 		"Enable leader election for controller manager. "+
 			"Enabling this will ensure there is only one active controller manager.")
 	flag.StringVar(&serviceNameTemplate, "default-backendservice-name", "{name}-{port}",
 		"A naming template consists of {namespace}, {name}, {port} or {hash} separated by hyphens, "+
 			"where {hash} is the first 8 digits of a hash of other given information")
 	flag.BoolVar(&allowServiceName, "enable-custom-service-names", true, "Enable using custom service names in autoneg annotation.")
+	flag.BoolVar(&alwaysReconcile, "always-reconcile", false, "Periodically reconciles even if annotation statuses don't change.")
+	flag.StringVar(&reconcilePeriod, "reconcile-period", "", "The minimum frequency at which watched resources are reconciled, e.g. 10m. Defaults to 10h if not set.")
 	opts := zap.Options{
 		Development: true,
 	}
@@ -95,6 +100,15 @@ func main() {
 		os.Exit(1)
 	}
 
+	var reconcileDuration time.Duration
+	if reconcilePeriod != "" {
+		reconcileDuration, err = time.ParseDuration(reconcilePeriod)
+		if err != nil {
+			setupLog.Error(err, "Invalid reconcilePeriod")
+			os.Exit(1)
+		}
+	}
+
 	if !controllers.IsValidServiceNameTemplate(serviceNameTemplate) {
 		err = fmt.Errorf("invalid service name template %s", serviceNameTemplate)
 		setupLog.Error(err, "invalid service name template")
@@ -123,6 +137,8 @@ func main() {
 		AllowServiceName:                 allowServiceName,
 		MaxRatePerEndpointDefault:        maxRatePerEndpointDefault,
 		MaxConnectionsPerEndpointDefault: maxConnectionsPerEndpointDefault,
+		AlwaysReconcile:                  alwaysReconcile,
+		ReconcileDuration:                &reconcileDuration,
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "Service")
 		os.Exit(1)