Merge pull request #77 from prachidamle/fix_race

Set security-runner job properties and ensure pod cleanup, also use current scan name to avoid multiple scan jobs

Author: prachidamle

pkg/securityscan/controller.go

@@ -14,8 +14,12 @@ import (
 	detector "github.com/rancher/kubernetes-provider-detector"
 	"github.com/rancher/wrangler/pkg/apply"
 	"github.com/rancher/wrangler/pkg/crd"
+	appsctl "github.com/rancher/wrangler/pkg/generated/controllers/apps"
+	appsctlv1 "github.com/rancher/wrangler/pkg/generated/controllers/apps/v1"
 	batchctl "github.com/rancher/wrangler/pkg/generated/controllers/batch"
+	batchctlv1 "github.com/rancher/wrangler/pkg/generated/controllers/batch/v1"
 	corectl "github.com/rancher/wrangler/pkg/generated/controllers/core"
+	corectlv1 "github.com/rancher/wrangler/pkg/generated/controllers/core/v1"
 	"github.com/rancher/wrangler/pkg/start"
 
 	"sync"
@@ -24,6 +28,7 @@ import (
 
 	cisoperatorapiv1 "github.com/rancher/cis-operator/pkg/apis/cis.cattle.io/v1"
 	cisoperatorctl "github.com/rancher/cis-operator/pkg/generated/controllers/cis.cattle.io"
+	cisoperatorctlv1 "github.com/rancher/cis-operator/pkg/generated/controllers/cis.cattle.io/v1"
 	"github.com/rancher/cis-operator/pkg/securityscan/scan"
 )
 
@@ -38,11 +43,13 @@ type Controller struct {
 	xcs              *kubeapiext.Clientset
 	coreFactory      *corectl.Factory
 	batchFactory     *batchctl.Factory
+	appsFactory      *appsctl.Factory
 	cisFactory       *cisoperatorctl.Factory
 	apply            apply.Apply
 	monitoringClient v1monitoringclient.MonitoringV1Interface
 
-	mu *sync.Mutex
+	mu              *sync.Mutex
+	currentScanName string
 
 	numTestsFailed   *prometheus.GaugeVec
 	numScansComplete *prometheus.CounterVec
@@ -51,6 +58,16 @@ type Controller struct {
 	numTestsNA       *prometheus.GaugeVec
 	numTestsPassed   *prometheus.GaugeVec
 	numTestsWarn     *prometheus.GaugeVec
+
+	scans          cisoperatorctlv1.ClusterScanController
+	jobs           batchctlv1.JobController
+	configmaps     corectlv1.ConfigMapController
+	configMapCache corectlv1.ConfigMapCache
+	services       corectlv1.ServiceController
+	pods           corectlv1.PodController
+	podCache       corectlv1.PodCache
+	daemonsets     appsctlv1.DaemonSetController
+	daemonsetCache appsctlv1.DaemonSetCache
 }
 
 func NewController(ctx context.Context, cfg *rest.Config, namespace, name string, imgConfig *cisoperatorapiv1.ScanImageConfig) (ctl *Controller, err error) {
@@ -112,6 +129,11 @@ func NewController(ctx context.Context, cfg *rest.Config, namespace, name string
 		return nil, fmt.Errorf("Error building core NewFactoryFromConfig: %s", err.Error())
 	}
 
+	ctl.appsFactory, err = appsctl.NewFactoryFromConfig(cfg)
+	if err != nil {
+		return nil, fmt.Errorf("Error building apps NewFactoryFromConfig: %s", err.Error())
+	}
+
 	ctl.monitoringClient, err = v1monitoringclient.NewForConfig(cfg)
 	if err != nil {
 		return nil, fmt.Errorf("Error building v1 monitoring client from config: %s", err.Error())
@@ -121,6 +143,17 @@ func NewController(ctx context.Context, cfg *rest.Config, namespace, name string
 	if err != nil {
 		return nil, fmt.Errorf("Error registering CIS Metrics: %s", err.Error())
 	}
+
+	ctl.scans = ctl.cisFactory.Cis().V1().ClusterScan()
+	ctl.jobs = ctl.batchFactory.Batch().V1().Job()
+	ctl.configmaps = ctl.coreFactory.Core().V1().ConfigMap()
+	ctl.configMapCache = ctl.coreFactory.Core().V1().ConfigMap().Cache()
+	ctl.services = ctl.coreFactory.Core().V1().Service()
+	ctl.pods = ctl.coreFactory.Core().V1().Pod()
+	ctl.podCache = ctl.coreFactory.Core().V1().Pod().Cache()
+	ctl.daemonsets = ctl.appsFactory.Apps().V1().DaemonSet()
+	ctl.daemonsetCache = ctl.appsFactory.Apps().V1().DaemonSet().Cache()
+
 	return ctl, nil
 }
 

pkg/securityscan/job/job.go

@@ -21,28 +21,34 @@ import (
 const (
 	defaultTerminationGracePeriodSeconds = int64(0)
 	defaultBackoffLimit                  = int32(0)
+	defaultTTLSecondsAfterFinished       = int32(0)
 )
 
 var (
 	ConditionComplete = condition.Cond(batchv1.JobComplete)
 	ConditionFailed   = condition.Cond(batchv1.JobFailed)
 
-	BackoffLimit = func(defaultValue int32) int32 {
-		if str, ok := os.LookupEnv("SYSTEM_UPGRADE_JOB_BACKOFF_LIMIT"); ok {
-			if i, err := strconv.ParseInt(str, 10, 32); err != nil {
-				logrus.Errorf("failed to parse $%s: %v", "SYSTEM_UPGRADE_JOB_BACKOFF_LIMIT", err)
-			} else {
-				return int32(i)
-			}
-		}
-		return defaultValue
-	}(defaultBackoffLimit)
+	backoffLimit = readFromEnv("CIS_JOB_BACKOFF_LIMIT", defaultBackoffLimit)
 
 	TerminationGracePeriodSeconds = func(defaultValue int64) int64 {
 		return defaultValue
 	}(defaultTerminationGracePeriodSeconds)
+
+	ttlSecondsAfterFinished = readFromEnv("CIS_JOB_TTL_SECONDS_AFTER_FINISH", defaultTTLSecondsAfterFinished)
 )
 
+func readFromEnv(key string, defaultValue int32) int32 {
+	if str, ok := os.LookupEnv(key); ok {
+		i, err := strconv.ParseInt(str, 10, 32)
+		if err != nil {
+			logrus.Errorf("failed to parse $%s: %v", key, err)
+			return defaultValue
+		}
+		return int32(i)
+	}
+	return defaultValue
+}
+
 func New(clusterscan *cisoperatorapiv1.ClusterScan, clusterscanprofile *cisoperatorapiv1.ClusterScanProfile, controllerName string, imageConfig *cisoperatorapiv1.ScanImageConfig) *batchv1.Job {
 	privileged := true
 	job := &batchv1.Job{
@@ -63,7 +69,8 @@ func New(clusterscan *cisoperatorapiv1.ClusterScan, clusterscanprofile *cisopera
 			}},
 		},
 		Spec: batchv1.JobSpec{
-			BackoffLimit: &BackoffLimit,
+			BackoffLimit:            &backoffLimit,
+			TTLSecondsAfterFinished: &ttlSecondsAfterFinished,
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: labels.Set{

pkg/securityscan/jobHandler.go

@@ -22,6 +22,8 @@ import (
 	"github.com/rancher/wrangler/pkg/name"
 )
 
+var sonobuoyWorkerLabel = map[string]string{"sonobuoy-plugin": "rancher-kube-bench"}
+
 // job events (successful completions) should remove the job after validatinf Done annotation and Output CM
 func (c *Controller) handleJobs(ctx context.Context) error {
 	scans := c.cisFactory.Cis().V1().ClusterScan()
@@ -59,7 +61,6 @@ func (c *Controller) handleJobs(ctx context.Context) error {
 
 		// if the scan has completed then delete the job
 		if v1.ClusterScanConditionComplete.IsTrue(scan) {
-			c.ensureCleanup(scan)
 			if !v1.ClusterScanConditionFailed.IsTrue(scan) {
 				logrus.Infof("Marking ClusterScanConditionAlerted for scan: %v", scanName)
 				v1.ClusterScanConditionAlerted.Unknown(scan)
@@ -71,12 +72,21 @@ func (c *Controller) handleJobs(ctx context.Context) error {
 				c.rescheduleScan(scan)
 				c.purgeOldClusterScanReports(scan)
 			}
+			err := c.deleteJob(jobs, obj, metav1.DeletePropagationBackground)
+			if err != nil {
+				return obj, fmt.Errorf("error deleting job: %v", err)
+			}
+			err = c.ensureCleanup(scan)
+			if err != nil {
+				return obj, err
+			}
 			//update scan
 			_, err = scans.UpdateStatus(scan)
 			if err != nil {
 				return nil, fmt.Errorf("error updating condition of cluster scan object: %v", scanName)
 			}
-			return obj, c.deleteJob(jobs, obj, metav1.DeletePropagationBackground)
+			c.currentScanName = ""
+			return obj, nil
 		}
 
 		if v1.ClusterScanConditionRunCompleted.IsTrue(scan) {
@@ -187,9 +197,38 @@ func (c *Controller) createClusterScanReport(outputBytes []byte, scan *v1.Cluste
 
 func (c *Controller) ensureCleanup(scan *v1.ClusterScan) error {
 	var err error
-	configmaps := c.coreFactory.Core().V1().ConfigMap()
+	// Delete the dameonset
+	dsPrefix := "sonobuoy-rancher-kube-bench-daemon-set"
+	dsList, err := c.daemonsetCache.List(v1.ClusterScanNS, labels.Set(sonobuoyWorkerLabel).AsSelector())
+	if err != nil {
+		return fmt.Errorf("cis: ensureCleanup: error listing daemonsets: %v", err)
+	}
+	for _, ds := range dsList {
+		if !strings.HasPrefix(ds.Name, dsPrefix) {
+			continue
+		}
+		if e := c.daemonsets.Delete(v1.ClusterScanNS, ds.Name, &metav1.DeleteOptions{}); e != nil && !errors.IsNotFound(e) {
+			return fmt.Errorf("cis: ensureCleanup: error deleting daemonset %v: %v", ds.Name, e)
+		}
+	}
+
+	// Delete the pod
+	podPrefix := name.SafeConcatName("security-scan-runner", scan.Name)
+	podList, err := c.podCache.List(v1.ClusterScanNS, labels.Set(SonobuoyMasterLabel).AsSelector())
+	if err != nil {
+		return fmt.Errorf("cis: ensureCleanup: error listing pods: %v", err)
+	}
+	for _, pod := range podList {
+		if !strings.HasPrefix(pod.Name, podPrefix) {
+			continue
+		}
+		if e := c.pods.Delete(v1.ClusterScanNS, pod.Name, &metav1.DeleteOptions{}); e != nil && !errors.IsNotFound(e) {
+			return fmt.Errorf("cis: ensureCleanup: error deleting pod %v: %v", pod.Name, e)
+		}
+	}
+
 	// Delete cms
-	cms, err := configmaps.Cache().List(v1.ClusterScanNS, labels.NewSelector())
+	cms, err := c.configMapCache.List(v1.ClusterScanNS, labels.NewSelector())
 	if err != nil {
 		return fmt.Errorf("cis: ensureCleanup: error listing cm: %v", err)
 	}
@@ -198,7 +237,7 @@ func (c *Controller) ensureCleanup(scan *v1.ClusterScan) error {
 			continue
 		}
 
-		if e := configmaps.Delete(v1.ClusterScanNS, cm.Name, &metav1.DeleteOptions{}); e != nil && !errors.IsNotFound(e) {
+		if e := c.configmaps.Delete(v1.ClusterScanNS, cm.Name, &metav1.DeleteOptions{}); e != nil && !errors.IsNotFound(e) {
 			return fmt.Errorf("cis: ensureCleanup: error deleting cm %v: %v", cm.Name, e)
 		}
 	}