Create expectation package

.travis.yml

@@ -14,7 +14,7 @@ install:
 script:
   - ./hack/verify-codegen.sh
   - go build ./...
-  - golangci-lint run ./...
+  - golangci-lint run --config=linter_config.yaml ./...
   # Here we run all tests in pkg and we have to use `-ignore`
   # since goveralls uses `filepath.Match` to match ignore files
   # and it does not support patterns like `**`.

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/google/btree v1.0.0 // indirect
 	github.com/googleapis/gnostic v0.2.0 // indirect
 	github.com/imdario/mergo v0.3.7 // indirect
-	github.com/prometheus/client_golang v1.5.1 // indirect
+	github.com/prometheus/client_golang v1.5.1
 	github.com/sirupsen/logrus v1.4.2
 	github.com/stretchr/testify v1.4.0
 	gopkg.in/inf.v0 v0.9.1 // indirect

linter_config.yaml

@@ -32,7 +32,6 @@ linters:
   enable:
     - bodyclose
     - deadcode
-    - errcheck
     - misspell
     - lll
     - typecheck

pkg/controller.v1/common/job_controller.go

@@ -21,8 +21,8 @@ import (
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
-	"k8s.io/kubernetes/pkg/controller"
 	"volcano.sh/volcano/pkg/apis/scheduling/v1beta1"
+	"github.com/kubeflow/common/pkg/controller.v1/expectation"
 	volcanoclient "volcano.sh/volcano/pkg/client/clientset/versioned"
 )
 
@@ -97,7 +97,7 @@ type JobController struct {
 	// - "tf-operator/tfjob-abc/ps/pods", expects 2 adds.
 	// - "tf-operator/tfjob-abc/worker/services", expects 4 adds.
 	// - "tf-operator/tfjob-abc/worker/pods", expects 4 adds.
-	Expectations controller.ControllerExpectationsInterface
+	Expectations expectation.ControllerExpectationsInterface
 
 	// WorkQueue is a rate limited work queue. This is used to queue work to be
 	// processed instead of performing it as soon as a change happens. This
@@ -136,7 +136,7 @@ func NewJobController(
 		Config:           jobControllerConfig,
 		KubeClientSet:    kubeClientSet,
 		VolcanoClientSet: volcanoClientSet,
-		Expectations:     controller.NewControllerExpectations(),
+		Expectations:     expectation.NewControllerExpectations(),
 		WorkQueue:        workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), workQueueName),
 		Recorder:         recorder,
 	}

pkg/controller.v1/common/pod.go

@@ -16,11 +16,12 @@ package common
 
 import (
 	"fmt"
-	"github.com/kubeflow/common/pkg/controller.v1/control"
 	"reflect"
 	"strconv"
 	"strings"
 
+	"github.com/kubeflow/common/pkg/controller.v1/control"
+	"github.com/kubeflow/common/pkg/controller.v1/expectation"
 	log "github.com/sirupsen/logrus"
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -85,7 +86,7 @@ func (jc *JobController) AddPod(obj interface{}) {
 		}
 
 		rtype := pod.Labels[apiv1.ReplicaTypeLabel]
-		expectationPodsKey := GenExpectationPodsKey(jobKey, rtype)
+		expectationPodsKey := expectation.GenExpectationPodsKey(jobKey, rtype)
 
 		jc.Expectations.CreationObserved(expectationPodsKey)
 		// TODO: we may need add backoff here
@@ -186,7 +187,7 @@ func (jc *JobController) DeletePod(obj interface{}) {
 	}
 
 	rtype := pod.Labels[apiv1.ReplicaTypeLabel]
-	expectationPodsKey := GenExpectationPodsKey(jobKey, rtype)
+	expectationPodsKey := expectation.GenExpectationPodsKey(jobKey, rtype)
 
 	jc.Expectations.DeletionObserved(expectationPodsKey)
 	// TODO: we may need add backoff here
@@ -364,7 +365,7 @@ func (jc *JobController) createNewPod(job interface{}, rt, index string, spec *a
 		utilruntime.HandleError(fmt.Errorf("couldn't get key for job object %#v: %v", job, err))
 		return err
 	}
-	expectationPodsKey := GenExpectationPodsKey(jobKey, rt)
+	expectationPodsKey := expectation.GenExpectationPodsKey(jobKey, rt)
 	err = jc.Expectations.ExpectCreations(expectationPodsKey, 1)
 	if err != nil {
 		return err

pkg/controller.v1/common/service.go

@@ -15,11 +15,12 @@ package common
 
 import (
 	"fmt"
-	"github.com/kubeflow/common/pkg/controller.v1/control"
 	"strconv"
 	"strings"
 
 	apiv1 "github.com/kubeflow/common/pkg/apis/common/v1"
+	"github.com/kubeflow/common/pkg/controller.v1/control"
+	"github.com/kubeflow/common/pkg/controller.v1/expectation"
 	commonutil "github.com/kubeflow/common/pkg/util"
 	log "github.com/sirupsen/logrus"
 	"k8s.io/api/core/v1"
@@ -73,7 +74,7 @@ func (jc *JobController) AddService(obj interface{}) {
 		}
 
 		rtype := service.Labels[apiv1.ReplicaTypeLabel]
-		expectationServicesKey := GenExpectationServicesKey(jobKey, rtype)
+		expectationServicesKey := expectation.GenExpectationServicesKey(jobKey, rtype)
 
 		jc.Expectations.CreationObserved(expectationServicesKey)
 		// TODO: we may need add backoff here
@@ -245,7 +246,7 @@ func (jc *JobController) CreateNewService(job metav1.Object, rtype apiv1.Replica
 
 	// Convert ReplicaType to lower string.
 	rt := strings.ToLower(string(rtype))
-	expectationServicesKey := GenExpectationServicesKey(jobKey, rt)
+	expectationServicesKey := expectation.GenExpectationServicesKey(jobKey, rt)
 	err = jc.Expectations.ExpectCreations(expectationServicesKey, 1)
 	if err != nil {
 		return err

pkg/controller.v1/common/util.go

@@ -43,14 +43,6 @@ func RecheckDeletionTimestamp(getObject func() (metav1.Object, error)) func() er
 	}
 }
 
-func GenExpectationPodsKey(jobKey, replicaType string) string {
-	return jobKey + "/" + strings.ToLower(replicaType) + "/pods"
-}
-
-func GenExpectationServicesKey(jobKey, replicaType string) string {
-	return jobKey + "/" + strings.ToLower(replicaType) + "/services"
-}
-
 func MaxInt(x, y int) int {
 	if x < y {
 		return y

pkg/controller.v1/expectation/expectation.go

@@ -0,0 +1,205 @@
+package expectation
+
+import (
+	"fmt"
+	log "github.com/sirupsen/logrus"
+	"sync/atomic"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/clock"
+	"k8s.io/client-go/tools/cache"
+)
+
+const (
+	// If a watch drops a delete event for a pod, it'll take this long
+	// before a dormant controller waiting for those packets is woken up anyway. It is
+	// specifically targeted at the case where some problem prevents an update
+	// of expectations, without it the controller could stay asleep forever. This should
+	// be set based on the expected latency of watch events.
+	//
+	// Currently a controller can service (create *and* observe the watch events for said
+	// creation) about 10 pods a second, so it takes about 1 min to service
+	// 500 pods. Just creation is limited to 20qps, and watching happens with ~10-30s
+	// latency/pod at the scale of 3000 pods over 100 nodes.
+	ExpectationsTimeout = 5 * time.Minute
+)
+
+// Expectations are a way for controllers to tell the controller manager what they expect. eg:
+//	ControllerExpectations: {
+//		controller1: expects  2 adds in 2 minutes
+//		controller2: expects  2 dels in 2 minutes
+//		controller3: expects -1 adds in 2 minutes => controller3's expectations have already been met
+//	}
+//
+// Implementation:
+//	ControlleeExpectation = pair of atomic counters to track controllee's creation/deletion
+//	ControllerExpectationsStore = TTLStore + a ControlleeExpectation per controller
+//
+// * Once set expectations can only be lowered
+// * A controller isn't synced till its expectations are either fulfilled, or expire
+// * Controllers that don't set expectations will get woken up for every matching controllee
+
+// ExpKeyFunc to parse out the key from a ControlleeExpectation
+var ExpKeyFunc = func(obj interface{}) (string, error) {
+	if e, ok := obj.(*ControlleeExpectations); ok {
+		return e.key, nil
+	}
+	return "", fmt.Errorf("could not find key for obj %#v", obj)
+}
+
+// ControllerExpectationsInterface is an interface that allows users to set and wait on expectations.
+// Only abstracted out for testing.
+// Warning: if using KeyFunc it is not safe to use a single ControllerExpectationsInterface with different
+// types of controllers, because the keys might conflict across types.
+type ControllerExpectationsInterface interface {
+	GetExpectations(controllerKey string) (*ControlleeExpectations, bool, error)
+	SatisfiedExpectations(controllerKey string) bool
+	DeleteExpectations(controllerKey string)
+	SetExpectations(controllerKey string, add, del int) error
+	ExpectCreations(controllerKey string, adds int) error
+	ExpectDeletions(controllerKey string, dels int) error
+	CreationObserved(controllerKey string)
+	DeletionObserved(controllerKey string)
+	RaiseExpectations(controllerKey string, add, del int)
+	LowerExpectations(controllerKey string, add, del int)
+}
+
+// ControllerExpectations is a cache mapping controllers to what they expect to see before being woken up for a sync.
+type ControllerExpectations struct {
+	cache.Store
+}
+
+// GetExpectations returns the ControlleeExpectations of the given controller.
+func (r *ControllerExpectations) GetExpectations(controllerKey string) (*ControlleeExpectations, bool, error) {
+	exp, exists, err := r.GetByKey(controllerKey)
+	if err == nil && exists {
+		return exp.(*ControlleeExpectations), true, nil
+	}
+	return nil, false, err
+}
+
+// DeleteExpectations deletes the expectations of the given controller from the TTLStore.
+func (r *ControllerExpectations) DeleteExpectations(controllerKey string) {
+	if exp, exists, err := r.GetByKey(controllerKey); err == nil && exists {
+		if err := r.Delete(exp); err != nil {
+			log.Infof("Error deleting expectations for controller %v: %v", controllerKey, err)
+		}
+	}
+}
+
+// SatisfiedExpectations returns true if the required adds/dels for the given controller have been observed.
+// Add/del counts are established by the controller at sync time, and updated as controllees are observed by the controller
+// manager.
+func (r *ControllerExpectations) SatisfiedExpectations(controllerKey string) bool {
+	if exp, exists, err := r.GetExpectations(controllerKey); exists {
+		if exp.Fulfilled() {
+			log.Infof("Controller expectations fulfilled %#v", exp)
+			return true
+		} else if exp.isExpired() {
+			log.Infof("Controller expectations expired %#v", exp)
+			return true
+		} else {
+			log.Infof("Controller still waiting on expectations %#v", exp)
+			return false
+		}
+	} else if err != nil {
+		log.Infof("Error encountered while checking expectations %#v, forcing sync", err)
+	} else {
+		// When a new controller is created, it doesn't have expectations.
+		// When it doesn't see expected watch events for > TTL, the expectations expire.
+		//	- In this case it wakes up, creates/deletes controllees, and sets expectations again.
+		// When it has satisfied expectations and no controllees need to be created/destroyed > TTL, the expectations expire.
+		//	- In this case it continues without setting expectations till it needs to create/delete controllees.
+		log.Infof("Controller %v either never recorded expectations, or the ttl expired.", controllerKey)
+	}
+	// Trigger a sync if we either encountered and error (which shouldn't happen since we're
+	// getting from local store) or this controller hasn't established expectations.
+	return true
+}
+
+// TODO: Extend ExpirationCache to support explicit expiration.
+// TODO: Make this possible to disable in tests.
+// TODO: Support injection of clock.
+func (exp *ControlleeExpectations) isExpired() bool {
+	return clock.RealClock{}.Since(exp.timestamp) > ExpectationsTimeout
+}
+
+// SetExpectations registers new expectations for the given controller. Forgets existing expectations.
+func (r *ControllerExpectations) SetExpectations(controllerKey string, add, del int) error {
+	exp := &ControlleeExpectations{add: int64(add), del: int64(del), key: controllerKey, timestamp: clock.RealClock{}.Now()}
+	log.Infof("Setting expectations %#v", exp)
+	return r.Add(exp)
+}
+
+func (r *ControllerExpectations) ExpectCreations(controllerKey string, adds int) error {
+	return r.SetExpectations(controllerKey, adds, 0)
+}
+
+func (r *ControllerExpectations) ExpectDeletions(controllerKey string, dels int) error {
+	return r.SetExpectations(controllerKey, 0, dels)
+}
+
+// Decrements the expectation counts of the given controller.
+func (r *ControllerExpectations) LowerExpectations(controllerKey string, add, del int) {
+	if exp, exists, err := r.GetExpectations(controllerKey); err == nil && exists {
+		exp.Add(int64(-add), int64(-del))
+		// The expectations might've been modified since the update on the previous line.
+		log.Infof("Lowered expectations %#v", exp)
+	}
+}
+
+// Increments the expectation counts of the given controller.
+func (r *ControllerExpectations) RaiseExpectations(controllerKey string, add, del int) {
+	if exp, exists, err := r.GetExpectations(controllerKey); err == nil && exists {
+		exp.Add(int64(add), int64(del))
+		// The expectations might've been modified since the update on the previous line.
+		log.Infof("Raised expectations %#v", exp)
+	}
+}
+
+// CreationObserved atomically decrements the `add` expectation count of the given controller.
+func (r *ControllerExpectations) CreationObserved(controllerKey string) {
+	r.LowerExpectations(controllerKey, 1, 0)
+}
+
+// DeletionObserved atomically decrements the `del` expectation count of the given controller.
+func (r *ControllerExpectations) DeletionObserved(controllerKey string) {
+	r.LowerExpectations(controllerKey, 0, 1)
+}
+
+// Expectations are either fulfilled, or expire naturally.
+type Expectations interface {
+	Fulfilled() bool
+}
+
+// ControlleeExpectations track controllee creates/deletes.
+type ControlleeExpectations struct {
+	// Important: Since these two int64 fields are using sync/atomic, they have to be at the top of the struct due to a bug on 32-bit platforms
+	// See: https://golang.org/pkg/sync/atomic/ for more information
+	add       int64
+	del       int64
+	key       string
+	timestamp time.Time
+}
+
+// Add increments the add and del counters.
+func (e *ControlleeExpectations) Add(add, del int64) {
+	atomic.AddInt64(&e.add, add)
+	atomic.AddInt64(&e.del, del)
+}
+
+// Fulfilled returns true if this expectation has been fulfilled.
+func (e *ControlleeExpectations) Fulfilled() bool {
+	// TODO: think about why this line being atomic doesn't matter
+	return atomic.LoadInt64(&e.add) <= 0 && atomic.LoadInt64(&e.del) <= 0
+}
+
+// GetExpectations returns the add and del expectations of the controllee.
+func (e *ControlleeExpectations) GetExpectations() (int64, int64) {
+	return atomic.LoadInt64(&e.add), atomic.LoadInt64(&e.del)
+}
+
+// NewControllerExpectations returns a store for ControllerExpectations.
+func NewControllerExpectations() *ControllerExpectations {
+	return &ControllerExpectations{cache.NewStore(ExpKeyFunc)}
+}