mirror of
https://github.com/kyverno/kyverno.git
synced 2025-03-06 16:06:56 +00:00
314 lines
9.5 KiB
Go
314 lines
9.5 KiB
Go
package controller
|
|
|
|
import (
|
|
"fmt"
|
|
"reflect"
|
|
"time"
|
|
|
|
jsonpatch "github.com/evanphx/json-patch"
|
|
|
|
"github.com/nirmata/kyverno/pkg/annotations"
|
|
"github.com/nirmata/kyverno/pkg/info"
|
|
|
|
"github.com/nirmata/kyverno/pkg/engine"
|
|
|
|
"github.com/golang/glog"
|
|
v1alpha1 "github.com/nirmata/kyverno/pkg/apis/policy/v1alpha1"
|
|
lister "github.com/nirmata/kyverno/pkg/client/listers/policy/v1alpha1"
|
|
client "github.com/nirmata/kyverno/pkg/dclient"
|
|
"github.com/nirmata/kyverno/pkg/event"
|
|
"github.com/nirmata/kyverno/pkg/sharedinformer"
|
|
violation "github.com/nirmata/kyverno/pkg/violation"
|
|
"k8s.io/apimachinery/pkg/api/errors"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
|
|
"k8s.io/apimachinery/pkg/util/wait"
|
|
"k8s.io/client-go/tools/cache"
|
|
"k8s.io/client-go/util/workqueue"
|
|
)
|
|
|
|
//PolicyController to manage Policy CRD
|
|
type PolicyController struct {
|
|
client *client.Client
|
|
policyLister lister.PolicyLister
|
|
policySynced cache.InformerSynced
|
|
violationBuilder violation.Generator
|
|
eventController event.Generator
|
|
annotationsController annotations.Controller
|
|
queue workqueue.RateLimitingInterface
|
|
}
|
|
|
|
// NewPolicyController from cmd args
|
|
func NewPolicyController(client *client.Client,
|
|
policyInformer sharedinformer.PolicyInformer,
|
|
violationBuilder violation.Generator,
|
|
eventController event.Generator,
|
|
annotationsController annotations.Controller) *PolicyController {
|
|
|
|
controller := &PolicyController{
|
|
client: client,
|
|
policyLister: policyInformer.GetLister(),
|
|
policySynced: policyInformer.GetInfomer().HasSynced,
|
|
violationBuilder: violationBuilder,
|
|
eventController: eventController,
|
|
annotationsController: annotationsController,
|
|
queue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), policyWorkQueueName),
|
|
}
|
|
|
|
policyInformer.GetInfomer().AddEventHandler(cache.ResourceEventHandlerFuncs{
|
|
AddFunc: controller.createPolicyHandler,
|
|
UpdateFunc: controller.updatePolicyHandler,
|
|
DeleteFunc: controller.deletePolicyHandler,
|
|
})
|
|
return controller
|
|
}
|
|
|
|
func (pc *PolicyController) createPolicyHandler(resource interface{}) {
|
|
pc.enqueuePolicy(resource)
|
|
}
|
|
|
|
func (pc *PolicyController) updatePolicyHandler(oldResource, newResource interface{}) {
|
|
newPolicy := newResource.(*v1alpha1.Policy)
|
|
oldPolicy := oldResource.(*v1alpha1.Policy)
|
|
newPolicy.Status = v1alpha1.Status{}
|
|
oldPolicy.Status = v1alpha1.Status{}
|
|
newPolicy.ResourceVersion = ""
|
|
oldPolicy.ResourceVersion = ""
|
|
if reflect.DeepEqual(newPolicy, oldPolicy) {
|
|
return
|
|
}
|
|
pc.enqueuePolicy(newResource)
|
|
}
|
|
|
|
func (pc *PolicyController) deletePolicyHandler(resource interface{}) {
|
|
var object metav1.Object
|
|
var ok bool
|
|
if object, ok = resource.(metav1.Object); !ok {
|
|
glog.Error("error decoding object, invalid type")
|
|
return
|
|
}
|
|
//TODO: need to clear annotations on the resources
|
|
cleanAnnotations(pc.client, resource)
|
|
glog.Infof("policy deleted: %s", object.GetName())
|
|
}
|
|
|
|
func (pc *PolicyController) enqueuePolicy(obj interface{}) {
|
|
var key string
|
|
var err error
|
|
if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {
|
|
glog.Error(err)
|
|
return
|
|
}
|
|
pc.queue.Add(key)
|
|
}
|
|
|
|
// Run is main controller thread
|
|
func (pc *PolicyController) Run(stopCh <-chan struct{}) error {
|
|
defer utilruntime.HandleCrash()
|
|
|
|
if ok := cache.WaitForCacheSync(stopCh, pc.policySynced); !ok {
|
|
return fmt.Errorf("failed to wait for caches to sync")
|
|
}
|
|
|
|
for i := 0; i < policyControllerWorkerCount; i++ {
|
|
go wait.Until(pc.runWorker, time.Second, stopCh)
|
|
}
|
|
glog.Info("started policy controller workers")
|
|
|
|
return nil
|
|
}
|
|
|
|
//Stop to perform actions when controller is stopped
|
|
func (pc *PolicyController) Stop() {
|
|
defer pc.queue.ShutDown()
|
|
glog.Info("shutting down policy controller workers")
|
|
}
|
|
func (pc *PolicyController) runWorker() {
|
|
for pc.processNextWorkItem() {
|
|
}
|
|
}
|
|
|
|
func (pc *PolicyController) processNextWorkItem() bool {
|
|
obj, shutdown := pc.queue.Get()
|
|
if shutdown {
|
|
return false
|
|
}
|
|
|
|
err := func(obj interface{}) error {
|
|
defer pc.queue.Done(obj)
|
|
err := pc.syncHandler(obj)
|
|
pc.handleErr(err, obj)
|
|
return nil
|
|
}(obj)
|
|
if err != nil {
|
|
glog.Error(err)
|
|
return true
|
|
}
|
|
return true
|
|
}
|
|
|
|
func (pc *PolicyController) handleErr(err error, key interface{}) {
|
|
if err == nil {
|
|
pc.queue.Forget(key)
|
|
return
|
|
}
|
|
// This controller retries if something goes wrong. After that, it stops trying.
|
|
if pc.queue.NumRequeues(key) < policyWorkQueueRetryLimit {
|
|
glog.Warningf("Error syncing events %v: %v", key, err)
|
|
// Re-enqueue the key rate limited. Based on the rate limiter on the
|
|
// queue and the re-enqueue history, the key will be processed later again.
|
|
pc.queue.AddRateLimited(key)
|
|
return
|
|
}
|
|
pc.queue.Forget(key)
|
|
glog.Error(err)
|
|
glog.Warningf("Dropping the key out of the queue: %v", err)
|
|
}
|
|
|
|
func (pc *PolicyController) syncHandler(obj interface{}) error {
|
|
var key string
|
|
var ok bool
|
|
if key, ok = obj.(string); !ok {
|
|
return fmt.Errorf("expected string in workqueue but got %#v", obj)
|
|
}
|
|
_, name, err := cache.SplitMetaNamespaceKey(key)
|
|
if err != nil {
|
|
glog.Errorf("invalid policy key: %s", key)
|
|
return nil
|
|
}
|
|
|
|
// Get Policy resource with namespace/name
|
|
policy, err := pc.policyLister.Get(name)
|
|
if err != nil {
|
|
if errors.IsNotFound(err) {
|
|
glog.Errorf("policy '%s' in work queue no longer exists", key)
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
// process policy on existing resource
|
|
// get the violations and pass to violation Builder
|
|
// get the events and pass to event Builder
|
|
//TODO: processPolicy
|
|
glog.Infof("process policy %s on existing resources", policy.GetName())
|
|
policyInfos := engine.ProcessExisting(pc.client, policy)
|
|
events, violations := pc.createEventsAndViolations(policyInfos)
|
|
pc.eventController.Add(events...)
|
|
err = pc.violationBuilder.Add(violations...)
|
|
if err != nil {
|
|
glog.Error(err)
|
|
}
|
|
|
|
// add annotations to resources
|
|
pc.createAnnotations(policyInfos)
|
|
|
|
return nil
|
|
}
|
|
|
|
func (pc *PolicyController) createAnnotations(policyInfos []*info.PolicyInfo) {
|
|
for _, pi := range policyInfos {
|
|
var patch []byte
|
|
//get resource
|
|
obj, err := pc.client.GetResource(pi.RKind, pi.RNamespace, pi.RName)
|
|
if err != nil {
|
|
glog.Error(err)
|
|
continue
|
|
}
|
|
// add annotation for policy application
|
|
ann := obj.GetAnnotations()
|
|
// Mutation rules
|
|
ann, mpatch, err := annotations.AddPolicyJSONPatch(ann, pi, info.Mutation)
|
|
if err != nil {
|
|
glog.Error(err)
|
|
continue
|
|
}
|
|
// Validation rules
|
|
ann, vpatch, err := annotations.AddPolicyJSONPatch(ann, pi, info.Validation)
|
|
if err != nil {
|
|
glog.Error(err)
|
|
}
|
|
if mpatch == nil && mpatch == nil {
|
|
//nothing to patch
|
|
continue
|
|
}
|
|
// merge the patches
|
|
if mpatch != nil && vpatch != nil {
|
|
patch, err = jsonpatch.MergePatch(mpatch, vpatch)
|
|
if err != nil {
|
|
glog.Error(err)
|
|
continue
|
|
}
|
|
}
|
|
if mpatch == nil {
|
|
patch = vpatch
|
|
} else {
|
|
patch = mpatch
|
|
}
|
|
// add the anotation to the resource
|
|
_, err = pc.client.PatchResource(pi.RKind, pi.RNamespace, pi.RName, patch)
|
|
if err != nil {
|
|
glog.Error(err)
|
|
continue
|
|
}
|
|
}
|
|
}
|
|
|
|
func (pc *PolicyController) createEventsAndViolations(policyInfos []*info.PolicyInfo) ([]*event.Info, []*violation.Info) {
|
|
events := []*event.Info{}
|
|
violations := []*violation.Info{}
|
|
// Create events from the policyInfo
|
|
for _, policyInfo := range policyInfos {
|
|
frules := []v1alpha1.FailedRule{}
|
|
sruleNames := []string{}
|
|
|
|
for _, rule := range policyInfo.Rules {
|
|
if !rule.IsSuccessful() {
|
|
e := &event.Info{}
|
|
frule := v1alpha1.FailedRule{Name: rule.Name}
|
|
switch rule.RuleType {
|
|
case info.Mutation, info.Validation, info.Generation:
|
|
// Events
|
|
e = event.NewEvent(policyInfo.RKind, policyInfo.RNamespace, policyInfo.RName, event.PolicyViolation, event.FProcessRule, rule.Name, policyInfo.Name)
|
|
switch rule.RuleType {
|
|
case info.Mutation:
|
|
frule.Type = info.Mutation.String()
|
|
case info.Validation:
|
|
frule.Type = info.Validation.String()
|
|
case info.Generation:
|
|
frule.Type = info.Generation.String()
|
|
}
|
|
default:
|
|
glog.Info("Unsupported Rule type")
|
|
}
|
|
frules = append(frules, frule)
|
|
events = append(events, e)
|
|
} else {
|
|
sruleNames = append(sruleNames, rule.Name)
|
|
}
|
|
}
|
|
|
|
if !policyInfo.IsSuccessful() {
|
|
e := event.NewEvent("Policy", "", policyInfo.Name, event.PolicyViolation, event.FResourcePolcy, policyInfo.RNamespace+"/"+policyInfo.RName, concatFailedRules(frules))
|
|
events = append(events, e)
|
|
// Violation
|
|
// TODO: Violation is currently create at policy, level not resource level
|
|
// As we create violation, we check if the
|
|
v := violation.BuldNewViolation(policyInfo.Name, policyInfo.RKind, policyInfo.RNamespace, policyInfo.RName, event.PolicyViolation.String(), frules)
|
|
violations = append(violations, v)
|
|
} else {
|
|
// clean up violations
|
|
pc.violationBuilder.RemoveInactiveViolation(policyInfo.Name, policyInfo.RKind, policyInfo.RNamespace, policyInfo.RName, info.Mutation)
|
|
pc.violationBuilder.RemoveInactiveViolation(policyInfo.Name, policyInfo.RKind, policyInfo.RNamespace, policyInfo.RName, info.Validation)
|
|
}
|
|
|
|
// else {
|
|
// // Policy was processed succesfully
|
|
// e := event.NewEvent("Policy", "", policyInfo.Name, event.PolicyApplied, event.SPolicyApply, policyInfo.Name)
|
|
// events = append(events, e)
|
|
// // Policy applied succesfully on resource
|
|
// e = event.NewEvent(policyInfo.RKind, policyInfo.RNamespace, policyInfo.RName, event.PolicyApplied, event.SRuleApply, strings.Join(sruleNames, ";"), policyInfo.RName)
|
|
// events = append(events, e)
|
|
// }
|
|
}
|
|
return events, violations
|
|
}
|