add checker and fix lease create bug

Eikykun · Eikykun · commit f45671d014b8 · 2023-07-12T16:35:29.000+08:00
diff --git a/README.md b/README.md
@@ -16,7 +16,7 @@ The design architecture of this project is based on [openkruise/controllermesh](
 3. **Circuit breaker and rate limiter**: Not only Kubernetes operation requests, but also other external operation requests.
 4. **Multicluster routing and sharding**: This feature is supported by [kusionstack/kaera(karbour)]()
 
-<p align="center"><img width="800" src="./docs/img/img2.png"/></p>
+<p align="center"><img width="800" src="./docs/img/img4.png"/></p>
 
 ## Quick Start
 Visit [Quick Start]().
diff --git a/artifacts/images/manager.Dockerfile b/artifacts/images/manager.Dockerfile
@@ -1,5 +1,5 @@
 # Build the manager binary
-FROM golang:1.19 as builder
+FROM golang:1.20 as builder
 
 WORKDIR /workspace
 
diff --git a/artifacts/images/proxy.Dockerfile b/artifacts/images/proxy.Dockerfile
@@ -1,4 +1,4 @@
-FROM golang:1.19 as builder
+FROM golang:1.20 as builder
 
 WORKDIR /workspace
 
@@ -33,6 +33,6 @@ RUN useradd -m --uid 1359 kridge-proxy && \
   echo "kridge-proxy ALL=NOPASSWD: ALL" >> /etc/sudoers
 WORKDIR /
 COPY artifacts/scripts/proxy-poststart.sh /poststart.sh
-
+RUN mkdir /kridge && chmod 777 /kridge
 COPY --from=builder /workspace/kridge-proxy .
 ENTRYPOINT ["/kridge-proxy"]
diff --git a/docs/installation.md b/docs/installation.md
@@ -0,0 +1,18 @@
+
+## Installation
+
+**Install by helm**
+```shell
+# Firstly add charts repository if you haven't do this.
+$ helm repo add kusionstack https://kusionstack.github.io/charts/
+
+# [Optional]
+$ helm repo update
+
+# Install the latest version.
+$ helm install kridge kusionstack/kridge --version 0.0.x
+
+# Uninstall
+$ helm uninstall kridge
+```
+
diff --git a/e2e/customoperator/app/controller/pod_recorder.go b/e2e/customoperator/app/controller/pod_recorder.go
@@ -23,6 +23,7 @@ import (
 
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog/v2"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/event"
@@ -37,6 +38,7 @@ var (
 // ManagerStateReconciler reconciles a ManagerState object
 type PodReconciler struct {
 	client.Client
+	DirectorClient client.Client
 }
 
 func (r *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (_ ctrl.Result, err error) {
@@ -50,8 +52,8 @@ func (r *PodReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	// default handler.EnqueueRequestForObject
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&v1.Pod{}).
-		Watches(&source.Kind{Type: &v1.Pod{}}, &enqueueHandler{Client: r.Client, kind: "Pod"}).
-		Watches(&source.Kind{Type: &v1.ConfigMap{}}, &enqueueHandler{Client: r.Client, kind: "ConfigMap"}).
+		Watches(&source.Kind{Type: &v1.Pod{}}, &enqueueHandler{Client: r.DirectorClient, kind: "Pod"}).
+		Watches(&source.Kind{Type: &v1.ConfigMap{}}, &enqueueHandler{Client: r.DirectorClient, kind: "ConfigMap"}).
 		Complete(r)
 }
 
@@ -61,17 +63,24 @@ type enqueueHandler struct {
 }
 
 func (e *enqueueHandler) Create(event event.CreateEvent, q workqueue.RateLimitingInterface) {
-	add(e.Client, event.Object.GetNamespace(), e.kind)
+	Add(e.Client, event.Object.GetNamespace(), event.Object.GetName(), e.kind)
 }
 
 func (e *enqueueHandler) Update(event event.UpdateEvent, q workqueue.RateLimitingInterface) {
-	add(e.Client, event.ObjectNew.GetNamespace(), e.kind)
+	Add(e.Client, event.ObjectNew.GetNamespace(), event.ObjectNew.GetName(), e.kind)
 }
 
 func (e *enqueueHandler) Delete(event event.DeleteEvent, q workqueue.RateLimitingInterface) {
-	add(e.Client, event.Object.GetNamespace(), e.kind)
+	Add(e.Client, event.Object.GetNamespace(), event.Object.GetName(), e.kind)
 }
 
 func (e *enqueueHandler) Generic(event event.GenericEvent, q workqueue.RateLimitingInterface) {
-	add(e.Client, event.Object.GetNamespace(), e.kind)
+	Add(e.Client, event.Object.GetNamespace(), event.Object.GetName(), e.kind)
+}
+
+func Add(c client.Client, namespace, name, kind string) {
+	klog.Infof("handle event %s %s/%s", kind, namespace, name)
+	if err := add(c, namespace, kind); err != nil {
+		klog.Errorf("fail to record event %s %s/%s, %v", kind, namespace, name, err)
+	}
 }
diff --git a/e2e/customoperator/app/controller/util.go b/e2e/customoperator/app/controller/util.go
@@ -8,9 +8,9 @@ import (
 	"time"
 
 	v1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/util/retry"
 	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -25,21 +25,29 @@ func initConfigMap(c client.Client) *v1.ConfigMap {
 	cm := &v1.ConfigMap{}
 	cm.Name = cmName
 	cm.Namespace = podNamespace
-	c.Create(context.TODO(), cm)
+	cm.Data = map[string]string{}
+	if err := c.Create(context.TODO(), cm); err != nil {
+		klog.Errorf("Failed to create cm %v", err)
+	}
 	return cm
 }
 
+var Retry = wait.Backoff{
+	Steps:    50,
+	Duration: 10 * time.Millisecond,
+	Factor:   1.0,
+	Jitter:   0.1,
+}
+
 func add(c client.Client, namespace, kind string) error {
 	if namespace == "" || kind == "" {
 		return nil
 	}
 	key := podName
-	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
+	return retry.RetryOnConflict(Retry, func() error {
 		cm := &v1.ConfigMap{}
 		if err := c.Get(context.TODO(), types.NamespacedName{Name: cmName, Namespace: podNamespace}, cm); err != nil {
-			if !errors.IsNotFound(err) {
-				return err
-			}
+			klog.Errorf("Failed to get cm %v, try init", err)
 			cm = initConfigMap(c)
 		}
 		if cm.Data == nil {
@@ -78,7 +86,7 @@ func Checker(ctx context.Context, c client.Client) {
 		}
 
 		if err := c.Get(context.TODO(), types.NamespacedName{Name: cmName, Namespace: podNamespace}, cm); err != nil {
-			klog.Errorf("fail to get configMap %s", cmName)
+			klog.Errorf("fail to get configMap %s, %v", cmName, err)
 			continue
 		}
 		if cm.Data == nil {
diff --git a/e2e/customoperator/app/main.go b/e2e/customoperator/app/main.go
@@ -18,6 +18,7 @@ package main
 
 import (
 	"flag"
+	"fmt"
 	"math/rand"
 	_ "net/http/pprof"
 	"os"
@@ -28,10 +29,12 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
+	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/leaderelection/resourcelock"
 	"k8s.io/klog/v2"
 	"k8s.io/klog/v2/klogr"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 
@@ -85,11 +88,12 @@ func main() {
 		os.Exit(1)
 	}
 
-	go func() {
-		setupLog.Info("wait webhook ready")
+	directorClient := NewDirectorClientFromManager(mgr, "checker")
 
+	go func() {
 		if err = (&appcontroller.PodReconciler{
-			Client: mgr.GetClient(),
+			Client:         mgr.GetClient(),
+			DirectorClient: directorClient,
 		}).SetupWithManager(mgr); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "Recorder")
 			os.Exit(1)
@@ -98,11 +102,22 @@ func main() {
 
 	go func() {
 		<-time.After(20 * time.Second)
-		appcontroller.Checker(ctx, mgr.GetClient())
+		appcontroller.Checker(ctx, directorClient)
 	}()
 	setupLog.Info("starting manager")
 	if err := mgr.Start(ctx); err != nil {
 		setupLog.Error(err, "problem running manager")
 		os.Exit(1)
 	}
 }
+
+func NewDirectorClientFromManager(mgr manager.Manager, name string) client.Client {
+	cfg := rest.CopyConfig(mgr.GetConfig())
+	cfg.UserAgent = fmt.Sprintf("director-client/%s", name)
+
+	c, err := client.New(cfg, client.Options{Scheme: mgr.GetScheme(), Mapper: mgr.GetRESTMapper()})
+	if err != nil {
+		panic(err)
+	}
+	return c
+}
diff --git a/go.mod b/go.mod
@@ -1,6 +1,6 @@
 module github.com/KusionStack/kridge
 
-go 1.19
+go 1.20
 
 require (
 	github.com/go-logr/zapr v1.2.3
diff --git a/pkg/cmd/proxy/main.go b/pkg/cmd/proxy/main.go
@@ -82,6 +82,7 @@ func main() {
 	{
 		opts := proxyapiserver.NewOptions()
 		opts.Config = rest.CopyConfig(cfg)
+		// Certs generated by proxy-init.sh
 		opts.SecureServingOptions.ServerCert.CertKey.KeyFile = "/var/run/secrets/kubernetes.io/serviceaccount/kridge/tls.key"
 		opts.SecureServingOptions.ServerCert.CertKey.CertFile = "/var/run/secrets/kubernetes.io/serviceaccount/kridge/tls.crt"
 		opts.SecureServingOptions.BindAddress = net.ParseIP("127.0.0.1")
diff --git a/pkg/manager/controllers/patchrunnable/labelpatch_runnable.go b/pkg/manager/controllers/patchrunnable/labelpatch_runnable.go
@@ -140,6 +140,7 @@ func (p *PatchRunnable) Start(ctx context.Context) error {
 		klog.Errorf("fail to load group version kind, %v", err)
 		return err
 	}
+	klog.Infof("load group version kind %s", util.DumpJSON(gvks))
 	p.groupVersionKinds = gvks
 	p.invalid = sets.Set[string]{}
 
@@ -170,6 +171,7 @@ func (p *PatchRunnable) Start(ctx context.Context) error {
 		}()
 	}
 	wg.Wait()
+	klog.Infof("finished patching label")
 	return nil
 }
 
@@ -191,7 +193,14 @@ func (p *PatchRunnable) getAllResources(namespaceList *v1.NamespaceList, itemCh
 		}()
 	}
 	wg.Wait()
-	close(itemCh)
+	defer klog.Infof("finish collect resources")
+	for {
+		if len(itemCh) == 0 {
+			close(itemCh)
+			return
+		}
+		<-time.After(5 * time.Second)
+	}
 }
 
 func (p *PatchRunnable) getResourceInNamespace(namespace *v1.Namespace, itemCh chan<- patchItem) {
diff --git a/pkg/manager/controllers/patchrunnable/resource.go b/pkg/manager/controllers/patchrunnable/resource.go
@@ -49,6 +49,7 @@ func LoadGroupVersionKind(c client.Client, discoveryClient *discovery.DiscoveryC
 	cm := &v1.ConfigMap{}
 	if err := c.Get(context.TODO(), types.NamespacedName{Name: *configMapName, Namespace: configMapNamespace}, cm); err != nil {
 		if errors.IsNotFound(err) {
+			klog.Infof("ConfigMap %s/%s not found", configMapNamespace, *configMapName)
 			return result, nil
 		}
 		return result, err
diff --git a/pkg/proxy/apiserver/handler.go b/pkg/proxy/apiserver/handler.go
@@ -51,6 +51,7 @@ import (
 var (
 	upgradeSubresources = sets.NewString("exec", "attach")
 	enableIpTable       = os.Getenv(constants.EnvIPTable) == "true"
+	enableWebhookProxy  = os.Getenv(constants.EnvEnableWebHookProxy) == "true"
 )
 
 type Proxy struct {
@@ -62,8 +63,10 @@ type Proxy struct {
 
 func NewProxy(opts *Options) (*Proxy, error) {
 	var servingInfo *server.SecureServingInfo
-	if err := opts.ApplyTo(&servingInfo); err != nil {
-		return nil, fmt.Errorf("error apply options %s: %v", utils.DumpJSON(opts), err)
+	if enableIpTable {
+		if err := opts.ApplyTo(&servingInfo); err != nil {
+			return nil, fmt.Errorf("error apply options %s: %v", utils.DumpJSON(opts), err)
+		}
 	}
 
 	tp, err := rest.TransportFor(opts.Config)
@@ -131,10 +134,15 @@ type handler struct {
 	electionHandler leaderelection.Handler
 }
 
+func getReqInfoStr(r *apirequest.RequestInfo) string {
+	return fmt.Sprintf("RequestInfo: { Path: %s, APIGroup: %s, Resource: %s, Subresource: %s, Verb: %s, Namespace: %s, Name: %s, APIVersion: %s }", r.Path, r.APIGroup, r.Resource, r.Subresource, r.Verb, r.Namespace, r.Name, r.APIVersion)
+}
+
 func (h *handler) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
 	startTime := time.Now()
-
 	requestInfo, ok := apirequest.RequestInfoFrom(r.Context())
+	klog.Infof("handle http req %s", r.URL.String())
+	klog.Infof(getReqInfoStr(requestInfo))
 	if !ok {
 		klog.Errorf("%s %s %s, no request info in context", r.Method, r.Header.Get("Content-Type"), r.URL)
 		http.Error(rw, "no request info in context", http.StatusBadRequest)
diff --git a/pkg/proxy/leaderelection/handler.go b/pkg/proxy/leaderelection/handler.go
@@ -93,6 +93,7 @@ func (h *handler) Handle(req *request.RequestInfo, r *http.Request) (handled boo
 		}
 
 		if adp.GetName() != h.lockName {
+			adp.EncodeInto(r)
 			return false, nil, nil
 		}
 
diff --git a/pkg/proxy/proto/strorage.go b/pkg/proxy/proto/strorage.go
@@ -40,6 +40,12 @@ type storage struct {
 	currentSpecFile  *os.File
 }
 
+func init() {
+	if err := os.MkdirAll("/home/kridge-proxy/kridge", 0777); err != nil {
+		klog.Error(err)
+	}
+}
+
 func newStorage() (*storage, error) {
 	var err error
 	s := &storage{}
diff --git a/pkg/utils/http/reverse_proxy.go b/pkg/utils/http/reverse_proxy.go
@@ -429,7 +429,9 @@ func (p *ReverseProxy) flushInterval(req *http.Request, res *http.Response) time
 	if resCT == "text/event-stream" {
 		return -1 // negative means immediately
 	}
-
+	if res.ContentLength == -1 {
+		return -1
+	}
 	// TODO: more specific cases? e.g. res.ContentLength == -1?
 	return p.FlushInterval
 }
diff --git a/pkg/webhook/pod/injector.go b/pkg/webhook/pod/injector.go
@@ -142,8 +142,8 @@ func (h *MutatingHandler) injectByShardingConfig(ctx context.Context, pod *v1.Po
 		Name:            constants.ProxyContainerName,
 		Image:           *proxyImage,
 		ImagePullPolicy: imagePullPolicy,
-		Args: []string{
-			"--v=" + strconv.Itoa(int(*proxyLogLevel)),
+		Args:            []string{
+			//"--v=" + strconv.Itoa(int(*proxyLogLevel)),
 		},
 		Env: []v1.EnvVar{
 			{Name: constants.EnvPodName, ValueFrom: &v1.EnvVarSource{FieldRef: &v1.ObjectFieldSelector{FieldPath: "metadata.name"}}},
@@ -172,9 +172,9 @@ func (h *MutatingHandler) injectByShardingConfig(ctx context.Context, pod *v1.Po
 			},
 		},
 		SecurityContext: &v1.SecurityContext{
-			Privileged:             utilpointer.Bool(true), // This can be false, but true help us debug more easier.
-			ReadOnlyRootFilesystem: utilpointer.Bool(true),
-			RunAsUser:              utilpointer.Int64(int64(constants.ProxyUserID)),
+			Privileged: utilpointer.Bool(true), // This can be false, but true help us debug more easier.
+			//ReadOnlyRootFilesystem: utilpointer.Bool(true),
+			RunAsUser: utilpointer.Int64(int64(constants.ProxyUserID)),
 		},
 	}
 

-Original file line number
+Diff line change
@@ @@ -1,6 +1,6 @@ @@
 module github.com/KusionStack/kridge
 -go 1.19
 +go 1.20
 require (
 	github.com/go-logr/zapr v1.2.3
Original file line number	Diff line number	Diff line change
`@@ -82,6 +82,7 @@ func main() {`
`82`	`82`	`{`
`83`	`83`	`opts := proxyapiserver.NewOptions()`
`84`	`84`	`opts.Config = rest.CopyConfig(cfg)`
	`85`	`+ // Certs generated by proxy-init.sh`
`85`	`86`	`opts.SecureServingOptions.ServerCert.CertKey.KeyFile = "/var/run/secrets/kubernetes.io/serviceaccount/kridge/tls.key"`
`86`	`87`	`opts.SecureServingOptions.ServerCert.CertKey.CertFile = "/var/run/secrets/kubernetes.io/serviceaccount/kridge/tls.crt"`
`87`	`88`	`opts.SecureServingOptions.BindAddress = net.ParseIP("127.0.0.1")`
Original file line number	Diff line number	Diff line change
`@@ -140,6 +140,7 @@ func (p *PatchRunnable) Start(ctx context.Context) error {`
`140`	`140`	`klog.Errorf("fail to load group version kind, %v", err)`
`141`	`141`	`return err`
`142`	`142`	`}`
	`143`	`+ klog.Infof("load group version kind %s", util.DumpJSON(gvks))`
`143`	`144`	`p.groupVersionKinds = gvks`
`144`	`145`	`p.invalid = sets.Set[string]{}`
`145`	`146`
`@@ -170,6 +171,7 @@ func (p *PatchRunnable) Start(ctx context.Context) error {`
`170`	`171`	`}()`
`171`	`172`	`}`
`172`	`173`	`wg.Wait()`
	`174`	`+ klog.Infof("finished patching label")`
`173`	`175`	`return nil`
`174`	`176`	`}`
`175`	`177`
`@@ -191,7 +193,14 @@ func (p PatchRunnable) getAllResources(namespaceList v1.NamespaceList, itemCh`
`191`	`193`	`}()`
`192`	`194`	`}`
`193`	`195`	`wg.Wait()`
`194`		`- close(itemCh)`
	`196`	`+ defer klog.Infof("finish collect resources")`
	`197`	`+ for {`
	`198`	`+ if len(itemCh) == 0 {`
	`199`	`+ close(itemCh)`
	`200`	`+ return`
	`201`	`+ }`
	`202`	`+ <-time.After(5 * time.Second)`
	`203`	`+ }`
`195`	`204`	`}`
`196`	`205`
`197`	`206`	`func (p PatchRunnable) getResourceInNamespace(namespace v1.Namespace, itemCh chan<- patchItem) {`