mirror of
https://github.com/rancher/norman.git
synced 2025-07-31 06:42:05 +00:00
102 lines
2.7 KiB
Go
102 lines
2.7 KiB
Go
package controller
|
|
|
|
import (
|
|
"context"
|
|
"strings"
|
|
"time"
|
|
|
|
errors2 "github.com/pkg/errors"
|
|
"github.com/rancher/lasso/pkg/controller"
|
|
"github.com/rancher/norman/metrics"
|
|
"github.com/sirupsen/logrus"
|
|
"k8s.io/apimachinery/pkg/api/errors"
|
|
"k8s.io/apimachinery/pkg/api/meta"
|
|
"k8s.io/apimachinery/pkg/runtime"
|
|
"k8s.io/client-go/tools/cache"
|
|
)
|
|
|
|
type HandlerFunc func(key string, obj interface{}) (interface{}, error)
|
|
|
|
type GenericController interface {
|
|
Informer() cache.SharedIndexInformer
|
|
AddHandler(ctx context.Context, name string, handler HandlerFunc)
|
|
Enqueue(namespace, name string)
|
|
EnqueueAfter(namespace, name string, after time.Duration)
|
|
}
|
|
|
|
type genericController struct {
|
|
controller controller.SharedController
|
|
informer cache.SharedIndexInformer
|
|
name string
|
|
namespace string
|
|
}
|
|
|
|
func NewGenericController(namespace, name string, controller controller.SharedController) GenericController {
|
|
return &genericController{
|
|
controller: controller,
|
|
informer: controller.Informer(),
|
|
name: name,
|
|
namespace: namespace,
|
|
}
|
|
}
|
|
|
|
func (g *genericController) Informer() cache.SharedIndexInformer {
|
|
return g.informer
|
|
}
|
|
|
|
func (g *genericController) Enqueue(namespace, name string) {
|
|
g.controller.Enqueue(namespace, name)
|
|
}
|
|
|
|
func (g *genericController) EnqueueAfter(namespace, name string, after time.Duration) {
|
|
g.controller.EnqueueAfter(namespace, name, after)
|
|
}
|
|
|
|
func (g *genericController) AddHandler(ctx context.Context, name string, handler HandlerFunc) {
|
|
g.controller.RegisterHandler(ctx, name, controller.SharedControllerHandlerFunc(func(key string, obj runtime.Object) (runtime.Object, error) {
|
|
if !isNamespace(g.namespace, obj) {
|
|
return obj, nil
|
|
}
|
|
logrus.Tracef("%s calling handler %s %s", g.name, name, key)
|
|
metrics.IncTotalHandlerExecution(g.name, name)
|
|
result, err := handler(key, obj)
|
|
runtimeObject, _ := result.(runtime.Object)
|
|
if err != nil && !ignoreError(err, false) {
|
|
metrics.IncTotalHandlerFailure(g.name, name, key)
|
|
}
|
|
if _, ok := err.(*ForgetError); ok {
|
|
logrus.Tracef("%v %v completed with dropped err: %v", g.name, key, err)
|
|
return runtimeObject, controller.ErrIgnore
|
|
}
|
|
return runtimeObject, err
|
|
}))
|
|
}
|
|
|
|
func isNamespace(namespace string, obj runtime.Object) bool {
|
|
if namespace == "" {
|
|
return true
|
|
}
|
|
meta, err := meta.Accessor(obj)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
return meta.GetNamespace() == namespace
|
|
}
|
|
|
|
func ignoreError(err error, checkString bool) bool {
|
|
err = errors2.Cause(err)
|
|
if errors.IsConflict(err) {
|
|
return true
|
|
}
|
|
if err == controller.ErrIgnore {
|
|
return true
|
|
}
|
|
if _, ok := err.(*ForgetError); ok {
|
|
return true
|
|
}
|
|
if checkString {
|
|
return strings.HasSuffix(err.Error(), "please apply your changes to the latest version and try again")
|
|
}
|
|
return false
|
|
}
|