145 lines
4.3 KiB
Go
145 lines
4.3 KiB
Go
/*
|
|
|
|
Copyright 2020 The KubeSphere Authors.
|
|
|
|
Licensed under the Apache License, Version 2.0 (the "License");
|
|
you may not use this file except in compliance with the License.
|
|
You may obtain a copy of the License at
|
|
|
|
http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
Unless required by applicable law or agreed to in writing, software
|
|
distributed under the License is distributed on an "AS IS" BASIS,
|
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
See the License for the specific language governing permissions and
|
|
limitations under the License.
|
|
|
|
*/
|
|
|
|
package auditing
|
|
|
|
import (
|
|
"fmt"
|
|
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
|
|
"k8s.io/apimachinery/pkg/util/wait"
|
|
"k8s.io/client-go/kubernetes/scheme"
|
|
"k8s.io/client-go/tools/cache"
|
|
"k8s.io/client-go/util/workqueue"
|
|
"k8s.io/klog"
|
|
kubespherescheme "kubesphere.io/kubesphere/pkg/client/clientset/versioned/scheme"
|
|
auditinginformer "kubesphere.io/kubesphere/pkg/client/informers/externalversions/auditing/v1alpha1"
|
|
auditinglister "kubesphere.io/kubesphere/pkg/client/listers/auditing/v1alpha1"
|
|
"time"
|
|
)
|
|
|
|
type Controller struct {
|
|
auditingLister auditinglister.WebhookLister
|
|
auditingSynced cache.InformerSynced
|
|
workQueue workqueue.RateLimitingInterface
|
|
}
|
|
|
|
func NewController(informer auditinginformer.WebhookInformer) *Controller {
|
|
// Create auditing webhook informer
|
|
|
|
utilruntime.Must(kubespherescheme.AddToScheme(scheme.Scheme))
|
|
|
|
return &Controller{
|
|
auditingLister: informer.Lister(),
|
|
auditingSynced: informer.Informer().HasSynced,
|
|
workQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "auditing"),
|
|
}
|
|
}
|
|
|
|
func (c *Controller) Run(threadiness int, stopCh <-chan struct{}) error {
|
|
defer utilruntime.HandleCrash()
|
|
defer c.workQueue.ShutDown()
|
|
|
|
//init client
|
|
|
|
// Start the informer factories to begin populating the informer caches
|
|
klog.Info("Starting User controller")
|
|
|
|
// Wait for the caches to be synced before starting workers
|
|
klog.Info("Waiting for informer caches to sync")
|
|
if ok := cache.WaitForCacheSync(stopCh, c.auditingSynced); !ok {
|
|
return fmt.Errorf("failed to wait for caches to sync")
|
|
}
|
|
|
|
klog.Info("Starting workers")
|
|
// Launch two workers to process Foo resources
|
|
for i := 0; i < threadiness; i++ {
|
|
go wait.Until(c.runWorker, time.Second, stopCh)
|
|
}
|
|
|
|
klog.Info("Started workers")
|
|
<-stopCh
|
|
klog.Info("Shutting down workers")
|
|
return nil
|
|
}
|
|
|
|
func (c *Controller) enqueueUser(obj interface{}) {
|
|
var key string
|
|
var err error
|
|
if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {
|
|
utilruntime.HandleError(err)
|
|
return
|
|
}
|
|
c.workQueue.Add(key)
|
|
}
|
|
|
|
func (c *Controller) runWorker() {
|
|
for c.processNextWorkItem() {
|
|
}
|
|
}
|
|
|
|
func (c *Controller) processNextWorkItem() bool {
|
|
obj, shutdown := c.workQueue.Get()
|
|
|
|
if shutdown {
|
|
return false
|
|
}
|
|
|
|
// We wrap this block in a func so we can defer c.workqueue.Done.
|
|
err := func(obj interface{}) error {
|
|
// We call Done here so the workqueue knows we have finished
|
|
// processing this item. We also must remember to call Forget if we
|
|
// do not want this work item being re-queued. For example, we do
|
|
// not call Forget if a transient error occurs, instead the item is
|
|
// put back on the workqueue and attempted again after a back-off
|
|
// period.
|
|
defer c.workQueue.Done(obj)
|
|
var key string
|
|
var ok bool
|
|
// We expect strings to come off the workqueue. These are of the
|
|
// form namespace/name. We do this as the delayed nature of the
|
|
// workqueue means the items in the informer cache may actually be
|
|
// more up to date that when the item was initially put onto the
|
|
// workqueue.
|
|
if key, ok = obj.(string); !ok {
|
|
// As the item in the workqueue is actually invalid, we call
|
|
// Forget here else we'd go into a loop of attempting to
|
|
// process a work item that is invalid.
|
|
c.workQueue.Forget(obj)
|
|
utilruntime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
|
|
return nil
|
|
}
|
|
|
|
// Finally, if no error occurs we Forget this item so it does not
|
|
// get queued again until another change happens.
|
|
c.workQueue.Forget(obj)
|
|
klog.Infof("Successfully synced %s:%s", "key", key)
|
|
return nil
|
|
}(obj)
|
|
|
|
if err != nil {
|
|
utilruntime.HandleError(err)
|
|
return true
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
func (c *Controller) Start(stopCh <-chan struct{}) error {
|
|
return c.Run(4, stopCh)
|
|
}
|