Merge pull request #1241 from kubescape/hostsensor-improvements

host-scanner improvements

core/cautils/getter/testdata/policy.json

@@ -25789,7 +25789,7 @@
     },
     {
       "guid": "",
-      "name": "exclude-kubescape-host-scanner-resources",
+      "name": "exclude-host-scanner-resources",
       "attributes": {
         "systemException": true
       },
@@ -25804,7 +25804,7 @@
           "attributes": {
             "kind": "DaemonSet",
             "name": "host-scanner",
-            "namespace": "kubescape-host-scanner"
+            "namespace": "kubescape"
           }
         }
       ],

core/pkg/hostsensorutils/hostsensor.yaml

@@ -6,13 +6,13 @@ metadata:
     k8s-app: kubescape-host-scanner
     kubernetes.io/metadata.name: kubescape-host-scanner
     tier: kubescape-host-scanner-control-plane
-  name: kubescape-host-scanner
+  name: kubescape
 ---
 apiVersion: apps/v1
 kind: DaemonSet
 metadata:
   name: host-scanner
-  namespace: kubescape-host-scanner
+  namespace: kubescape
   labels:
     app: host-scanner
     k8s-app: kubescape-host-scanner
@@ -32,7 +32,7 @@ spec:
       - operator: Exists
       containers:
       - name: host-sensor
-        image: quay.io/kubescape/host-scanner:v1.0.57
+        image: quay.io/kubescape/host-scanner:v1.0.59
         securityContext:
           allowPrivilegeEscalation: true
           privileged: true
@@ -52,12 +52,17 @@ spec:
         volumeMounts:
         - mountPath: /host_fs
           name: host-filesystem
-        readinessProbe:
+        startupProbe:
           httpGet:
-            path: /kernelVersion
+            path: /readyz
             port: 7888
-          initialDelaySeconds: 1
+          failureThreshold: 30
           periodSeconds: 1
+        livenessProbe:
+          httpGet:
+            path: /healthz
+            port: 7888
+          periodSeconds: 10
       terminationGracePeriodSeconds: 120
       dnsPolicy: ClusterFirstWithHostNet
       automountServiceAccountToken: false

core/pkg/hostsensorutils/hostsensordeploy.go

@@ -17,6 +17,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/watch"
 )
 
@@ -51,7 +52,7 @@ func NewHostSensorHandler(k8sObj *k8sinterface.KubernetesApi, hostSensorYAMLFile
 	if hostSensorYAMLFile != "" {
 		d, err := loadHostSensorFromFile(hostSensorYAMLFile)
 		if err != nil {
-			return nil, fmt.Errorf("failed to load host-scan yaml file, reason: %w", err)
+			return nil, fmt.Errorf("failed to load host-scanner yaml file, reason: %w", err)
 		}
 		hostSensorYAML = d
 	}
@@ -82,7 +83,11 @@ func (hsh *HostSensorHandler) Init(ctx context.Context) error {
 	// store pod names
 	// make sure all pods are running, after X seconds treat has running anyway, and log an error on the pods not running yet
 	logger.L().Info("Installing host scanner")
-	logger.L().Debug("The host scanner is a DaemonSet that runs on each node in the cluster. The DaemonSet will be running in it's own namespace and will be deleted once the scan is completed. If you do not wish to install the host scanner, please run the scan without the --enable-host-scan flag.")
+	logger.L().Debug("The host scanner is a DaemonSet that runs on each node in the cluster. The DaemonSet will be running in it's own Namespace and will be deleted once the scan is completed. If you do not wish to install the host scanner, please run the scan without the --enable-host-scan flag.")
+
+	// log is used to avoid log duplication
+	// coming from the different host-scanner instances
+	log := NewLogCoupling()
 
 	cautils.StartSpinner()
 	defer cautils.StopSpinner()
@@ -91,9 +96,9 @@ func (hsh *HostSensorHandler) Init(ctx context.Context) error {
 		return fmt.Errorf("failed to apply host scanner YAML, reason: %v", err)
 	}
 
-	hsh.populatePodNamesToNodeNames(ctx)
+	hsh.populatePodNamesToNodeNames(ctx, log)
 	if err := hsh.checkPodForEachNode(); err != nil {
-		logger.L().Ctx(ctx).Warning("failed to validate host-sensor pods status", helpers.Error(err))
+		logger.L().Ctx(ctx).Warning(failedToValidateHostSensorPodStatus, helpers.Error(err))
 	}
 
 	return nil
@@ -156,7 +161,7 @@ func (hsh *HostSensorHandler) applyYAML(ctx context.Context) error {
 			containers, err := w.GetContainers()
 			if err != nil {
 				if erra := hsh.tearDownNamespace(namespaceName); erra != nil {
-					logger.L().Ctx(ctx).Warning("failed to tear down namespace", helpers.Error(erra))
+					logger.L().Ctx(ctx).Warning(failedToTeardownNamespace, helpers.Error(erra))
 				}
 				return fmt.Errorf("container not found in DaemonSet: %v", err)
 			}
@@ -180,7 +185,7 @@ func (hsh *HostSensorHandler) applyYAML(ctx context.Context) error {
 		}
 		if e != nil {
 			if erra := hsh.tearDownNamespace(namespaceName); erra != nil {
-				logger.L().Ctx(ctx).Warning("failed to tear down namespace", helpers.Error(erra))
+				logger.L().Ctx(ctx).Warning(failedToTeardownNamespace, helpers.Error(erra))
 			}
 			return fmt.Errorf("failed to create/update YAML, reason: %v", e)
 		}
@@ -190,14 +195,14 @@ func (hsh *HostSensorHandler) applyYAML(ctx context.Context) error {
 			b, err := json.Marshal(newWorkload.GetObject())
 			if err != nil {
 				if erra := hsh.tearDownNamespace(namespaceName); erra != nil {
-					logger.L().Ctx(ctx).Warning("failed to tear down namespace", helpers.Error(erra))
+					logger.L().Ctx(ctx).Warning(failedToTeardownNamespace, helpers.Error(erra))
 				}
 				return fmt.Errorf("failed to Marshal YAML of DaemonSet, reason: %v", err)
 			}
 			var ds appsv1.DaemonSet
 			if err := json.Unmarshal(b, &ds); err != nil {
 				if erra := hsh.tearDownNamespace(namespaceName); erra != nil {
-					logger.L().Ctx(ctx).Warning("failed to tear down namespace", helpers.Error(erra))
+					logger.L().Ctx(ctx).Warning(failedToTeardownNamespace, helpers.Error(erra))
 				}
 				return fmt.Errorf("failed to Unmarshal YAML of DaemonSet, reason: %v", err)
 			}
@@ -228,7 +233,7 @@ func (hsh *HostSensorHandler) checkPodForEachNode() error {
 			hsh.podListLock.RLock()
 			podsMap := hsh.hostSensorPodNames
 			hsh.podListLock.RUnlock()
-			return fmt.Errorf("host-sensor pods number (%d) differ than nodes number (%d) after deadline exceeded. Kubescape will take data only from the pods below: %v",
+			return fmt.Errorf("host-scanner pods number (%d) differ than nodes number (%d) after deadline exceeded. Kubescape will take data only from the pods below: %v",
 				podsNum, len(nodesList.Items), podsMap)
 		}
 		time.Sleep(100 * time.Millisecond)
@@ -238,7 +243,7 @@ func (hsh *HostSensorHandler) checkPodForEachNode() error {
 }
 
 // initiating routine to keep pod list updated
-func (hsh *HostSensorHandler) populatePodNamesToNodeNames(ctx context.Context) {
+func (hsh *HostSensorHandler) populatePodNamesToNodeNames(ctx context.Context, log *LogsMap) {
 	go func() {
 		var watchRes watch.Interface
 		var err error
@@ -247,10 +252,10 @@ func (hsh *HostSensorHandler) populatePodNamesToNodeNames(ctx context.Context) {
 			LabelSelector: fmt.Sprintf("name=%s", hsh.daemonSet.Spec.Template.Labels["name"]),
 		})
 		if err != nil {
-			logger.L().Ctx(ctx).Warning("failed to watch over DaemonSet pods - are we missing watch pods permissions?", helpers.Error(err))
+			logger.L().Ctx(ctx).Warning(failedToWatchOverDaemonSetPods, helpers.Error(err))
 		}
 		if watchRes == nil {
-			logger.L().Ctx(ctx).Error("failed to watch over DaemonSet pods, will not be able to get host-sensor data")
+			logger.L().Ctx(ctx).Error("failed to watch over DaemonSet pods, will not be able to get host-scanner data")
 			return
 		}
 
@@ -259,12 +264,12 @@ func (hsh *HostSensorHandler) populatePodNamesToNodeNames(ctx context.Context) {
 			if !ok {
 				continue
 			}
-			go hsh.updatePodInListAtomic(ctx, eve.Type, pod)
+			go hsh.updatePodInListAtomic(ctx, eve.Type, pod, log)
 		}
 	}()
 }
 
-func (hsh *HostSensorHandler) updatePodInListAtomic(ctx context.Context, eventType watch.EventType, podObj *corev1.Pod) {
+func (hsh *HostSensorHandler) updatePodInListAtomic(ctx context.Context, eventType watch.EventType, podObj *corev1.Pod, log *LogsMap) {
 	hsh.podListLock.Lock()
 	defer hsh.podListLock.Unlock()
 
@@ -285,10 +290,11 @@ func (hsh *HostSensorHandler) updatePodInListAtomic(ctx context.Context, eventTy
 					len(podObj.Spec.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchFields[0].Values) > 0 {
 					nodeName = podObj.Spec.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchFields[0].Values[0]
 				}
-				logger.L().Ctx(ctx).Warning("One host-sensor pod is unable to schedule on node. We will fail to collect the data from this node",
-					helpers.String("message", podObj.Status.Conditions[0].Message),
-					helpers.String("nodeName", nodeName),
-					helpers.String("podName", podObj.ObjectMeta.Name))
+				if !log.isDuplicated(oneHostSensorPodIsUnabledToSchedule) {
+					logger.L().Ctx(ctx).Warning(oneHostSensorPodIsUnabledToSchedule,
+						helpers.String("message", podObj.Status.Conditions[0].Message))
+					log.update(oneHostSensorPodIsUnabledToSchedule)
+				}
 				if nodeName != "" {
 					hsh.hostSensorUnscheduledPodNames[podObj.ObjectMeta.Name] = nodeName
 				}
@@ -301,14 +307,84 @@ func (hsh *HostSensorHandler) updatePodInListAtomic(ctx context.Context, eventTy
 	}
 }
 
+// tearDownNamespace manage the host-scanner deletion.
+func (hsh *HostSensorHandler) tearDownHostScanner(namespace string) error {
+	client := hsh.k8sObj.KubernetesClient
+
+	// delete host-scanner DaemonSet
+	err := client.AppsV1().
+		DaemonSets(namespace).
+		Delete(
+			hsh.k8sObj.Context,
+			hsh.daemonSet.Name,
+			metav1.DeleteOptions{
+				GracePeriodSeconds: &hsh.gracePeriod,
+			},
+		)
+	if err != nil {
+		return fmt.Errorf("failed to delete host-scanner DaemonSet: %v", err)
+	}
+
+	// wait for DaemonSet to be deleted
+	err = hsh.waitHostScannerDeleted(hsh.k8sObj.Context)
+	if err != nil {
+		return fmt.Errorf("failed to delete host-scanner DaemonSet: %v", err)
+	}
+
+	return nil
+}
+
+// tearDownNamespace manage the given namespace deletion.
+// At first, it checks if the namespace was already present before installing host-scanner.
+// In that case skips the deletion.
+// If was not, then patches the namespace in order to remove the finalizers,
+// and finally delete the it.
 func (hsh *HostSensorHandler) tearDownNamespace(namespace string) error {
 	// if namespace was already present on kubernetes (before installing host-scanner),
 	// then we shouldn't delete it.
 	if hsh.namespaceWasPresent() {
 		return nil
 	}
-	if err := hsh.k8sObj.KubernetesClient.CoreV1().Namespaces().Delete(hsh.k8sObj.Context, namespace, metav1.DeleteOptions{GracePeriodSeconds: &hsh.gracePeriod}); err != nil {
-		return fmt.Errorf("failed to delete host-sensor namespace: %v", err)
+	// to make it more readable we store the object client in a variable
+	client := hsh.k8sObj.KubernetesClient
+
+	// prepare patch json to remove finalizers from namespace
+	patchData := `
+	[
+		{
+			"op": "replace",
+			"path": "/metadata/finalizers",
+			"value": []
+		}
+	]
+	`
+	// patch namespace object removing finalizers
+	_, err := client.CoreV1().
+		Namespaces().
+		Patch(
+			hsh.k8sObj.Context,
+			namespace,
+			types.JSONPatchType,
+			[]byte(patchData),
+			metav1.PatchOptions{},
+		)
+	if err != nil {
+		return fmt.Errorf("failed to remove finalizers from Namespace: %v", err)
+	}
+
+	// patch namespace object removing finalizers
+	// delete namespace object
+	err = client.CoreV1().
+		Namespaces().
+		Delete(
+			hsh.k8sObj.Context,
+			namespace,
+			metav1.DeleteOptions{
+				GracePeriodSeconds: &hsh.gracePeriod,
+			},
+		)
+	if err != nil {
+		return fmt.Errorf("failed to delete %s Namespace: %v", namespace, err)
 	}
 
 	return nil
@@ -317,14 +393,13 @@ func (hsh *HostSensorHandler) tearDownNamespace(namespace string) error {
 func (hsh *HostSensorHandler) TearDown() error {
 	namespace := hsh.GetNamespace()
 	// delete DaemonSet
-	if err := hsh.k8sObj.KubernetesClient.AppsV1().DaemonSets(hsh.GetNamespace()).Delete(hsh.k8sObj.Context, hsh.daemonSet.Name, metav1.DeleteOptions{GracePeriodSeconds: &hsh.gracePeriod}); err != nil {
-		return fmt.Errorf("failed to delete host-sensor daemonset: %v", err)
+	if err := hsh.tearDownHostScanner(namespace); err != nil {
+		return fmt.Errorf("failed to delete host-scanner DaemonSet: %v", err)
 	}
 	// delete Namespace
 	if err := hsh.tearDownNamespace(namespace); err != nil {
-		return fmt.Errorf("failed to delete host-sensor daemonset: %v", err)
+		return fmt.Errorf("failed to delete host-scanner Namespace: %v", err)
 	}
-	// TODO: wait for termination? may take up to 120 seconds!!!
 
 	return nil
 }
@@ -344,3 +419,32 @@ func loadHostSensorFromFile(hostSensorYAMLFile string) (string, error) {
 	// TODO - Add file validation
 	return string(dat), err
 }
+
+// waitHostScannerDeleted watch for host-scanner deletion.
+// In case it fails it returns an error.
+func (hsh *HostSensorHandler) waitHostScannerDeleted(ctx context.Context) error {
+	labelSelector := fmt.Sprintf("name=%s", hsh.daemonSet.Name)
+	opts := metav1.ListOptions{
+		TypeMeta:      metav1.TypeMeta{},
+		LabelSelector: labelSelector,
+		FieldSelector: "",
+	}
+	watcher, err := hsh.k8sObj.KubernetesClient.CoreV1().
+		Pods(hsh.daemonSet.Namespace).
+		Watch(ctx, opts)
+	if err != nil {
+		return err
+	}
+	defer watcher.Stop()
+
+	for {
+		select {
+		case event := <-watcher.ResultChan():
+			if event.Type == watch.Deleted {
+				return nil
+			}
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}

core/pkg/hostsensorutils/hostsensordeploy_test.go

@@ -34,7 +34,7 @@ func TestHostSensorHandler(t *testing.T) {
 			})
 
 			t.Run("should return namespace", func(t *testing.T) {
-				require.Equal(t, "kubescape-host-scanner", h.GetNamespace())
+				require.Equal(t, "kubescape", h.GetNamespace())
 			})
 
 			t.Run("should collect resources from pods - happy path", func(t *testing.T) {

core/pkg/hostsensorutils/hostsensorgetfrompod.go

@@ -84,12 +84,17 @@ func (hsh *HostSensorHandler) sendAllPodsHTTPGETRequest(ctx context.Context, pat
 	podList := hsh.getPodList()
 	res := make([]hostsensor.HostSensorDataEnvelope, 0, len(podList))
 	var wg sync.WaitGroup
+
 	// initialization of the channels
 	hsh.workerPool.init(len(podList))
 
+	// log is used to avoid log duplication
+	// coming from the different host-scanner instances
+	log := NewLogCoupling()
+
 	hsh.workerPool.hostSensorApplyJobs(podList, path, requestKind)
 	hsh.workerPool.hostSensorGetResults(&res)
-	hsh.workerPool.createWorkerPool(ctx, hsh, &wg)
+	hsh.workerPool.createWorkerPool(ctx, hsh, &wg, log)
 	hsh.workerPool.waitForDone(&wg)
 
 	return res, nil

core/pkg/hostsensorutils/hostsensorworkerpool.go

@@ -43,22 +43,23 @@ func (wp *workerPool) init(noOfPods ...int) {
 }
 
 // The worker takes a job out of the chan, executes the request, and pushes the result to the results chan
-func (wp *workerPool) hostSensorWorker(ctx context.Context, hsh *HostSensorHandler, wg *sync.WaitGroup) {
+func (wp *workerPool) hostSensorWorker(ctx context.Context, hsh *HostSensorHandler, wg *sync.WaitGroup, log *LogsMap) {
 	defer wg.Done()
 	for job := range wp.jobs {
 		hostSensorDataEnvelope, err := hsh.getResourcesFromPod(job.podName, job.nodeName, job.requestKind, job.path)
-		if err != nil {
-			logger.L().Ctx(ctx).Warning("failed to get data", helpers.String("path", job.path), helpers.String("podName", job.podName), helpers.Error(err))
+		if err != nil && !log.isDuplicated(failedToGetData) {
+			logger.L().Ctx(ctx).Warning(failedToGetData, helpers.String("path", job.path), helpers.Error(err))
+			log.update(failedToGetData)
 			continue
 		}
 		wp.results <- hostSensorDataEnvelope
 	}
 }
 
-func (wp *workerPool) createWorkerPool(ctx context.Context, hsh *HostSensorHandler, wg *sync.WaitGroup) {
+func (wp *workerPool) createWorkerPool(ctx context.Context, hsh *HostSensorHandler, wg *sync.WaitGroup, log *LogsMap) {
 	for i := 0; i < noOfWorkers; i++ {
 		wg.Add(1)
-		go wp.hostSensorWorker(ctx, hsh, wg)
+		go wp.hostSensorWorker(ctx, hsh, wg, log)
 	}
 }
 

core/pkg/hostsensorutils/log_coupling.go

@@ -0,0 +1,51 @@
+package hostsensorutils
+
+import "sync"
+
+type LogsMap struct {
+	// use sync.Mutex to avoid read/write
+	// access issues in multi-thread environments.
+	sync.Mutex
+	usedLogs map[string]int
+}
+
+// NewLogCoupling return an empty LogsMap struct ready to be used.
+func NewLogCoupling() *LogsMap {
+	return &LogsMap{
+		usedLogs: make(map[string]int),
+	}
+}
+
+// update add the logContent to the internal map
+// and set the occurrencty to 1 (if it has never been used before),
+// increment its values otherwise.
+func (lm *LogsMap) update(logContent string) {
+	lm.Lock()
+	_, ok := lm.usedLogs[logContent]
+	if !ok {
+		lm.usedLogs[logContent] = 1
+	} else {
+		lm.usedLogs[logContent]++
+	}
+	lm.Unlock()
+}
+
+// isDuplicated check if logContent is already present inside the internal map.
+// Return true in case logContent already exists, false otherwise.
+func (lm *LogsMap) isDuplicated(logContent string) bool {
+	lm.Lock()
+	_, ok := lm.usedLogs[logContent]
+	lm.Unlock()
+	return ok
+}
+
+// GgtOccurrence retrieve the number of occurrences logContent has been used.
+func (lm *LogsMap) getOccurrence(logContent string) int {
+	lm.Lock()
+	occurrence, ok := lm.usedLogs[logContent]
+	lm.Unlock()
+	if !ok {
+		return 0
+	}
+	return occurrence
+}

core/pkg/hostsensorutils/log_coupling_test.go

@@ -0,0 +1,100 @@
+package hostsensorutils
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestLogsMap_Update(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name        string
+		logs        []string
+		expectedLog string
+		expected    int
+	}{
+		{
+			name: "test_1",
+			logs: []string{
+				"log_1",
+				"log_1",
+				"log_1",
+			},
+			expectedLog: "log_1",
+			expected:    3,
+		},
+		{
+			name:        "test_2",
+			logs:        []string{},
+			expectedLog: "log_2",
+			expected:    0,
+		},
+		{
+			name: "test_3",
+			logs: []string{
+				"log_3",
+			},
+			expectedLog: "log_3",
+			expected:    1,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			lm := NewLogCoupling()
+			for _, log := range tt.logs {
+				lm.update(log)
+			}
+			if !assert.Equal(t, lm.getOccurrence(tt.expectedLog), tt.expected) {
+				t.Log("log occurrences are different")
+			}
+		})
+	}
+}
+
+func TestLogsMap_IsDuplicated(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name        string
+		logs        []string
+		expectedLog string
+		expected    bool
+	}{
+		{
+			name: "test_1",
+			logs: []string{
+				"log_1",
+				"log_1",
+				"log_1",
+			},
+			expectedLog: "log_1",
+			expected:    true,
+		},
+		{
+			name: "test_2",
+			logs: []string{
+				"log_1",
+				"log_1",
+			},
+			expectedLog: "log_2",
+			expected:    false,
+		},
+		{
+			name:        "test_3",
+			logs:        []string{},
+			expectedLog: "log_3",
+			expected:    false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			lm := NewLogCoupling()
+			for _, log := range tt.logs {
+				lm.update(log)
+			}
+			if !assert.Equal(t, lm.isDuplicated(tt.expectedLog), tt.expected) {
+				t.Log("duplication value differ from expected")
+			}
+		})
+	}
+}

core/pkg/hostsensorutils/logging_messages.go

@@ -0,0 +1,10 @@
+package hostsensorutils
+
+// messages used for warnings
+var (
+	failedToGetData                     = "failed to get data"
+	failedToTeardownNamespace           = "failed to teardown Namespace"
+	oneHostSensorPodIsUnabledToSchedule = "One host-sensor pod is unable to schedule on node. We will fail to collect the data from this node"
+	failedToWatchOverDaemonSetPods      = "failed to watch over DaemonSet pods"
+	failedToValidateHostSensorPodStatus = "failed to validate host-scanner pods status"
+)

core/pkg/resultshandling/reporter/v2/testdata/mock_opasessionobj.json

@@ -62743,7 +62743,7 @@
         },
         {
             "guid": "",
-            "name": "exclude-kubescape-host-scanner-resources",
+            "name": "exclude-host-scanner-resources",
             "attributes": {
                 "systemException": true
             },
@@ -62758,7 +62758,7 @@
                     "attributes": {
                         "kind": "DaemonSet",
                         "name": "host-scanner",
-                        "namespace": "kubescape-host-scanner"
+                        "namespace": "kubescape"
                     }
                 }
             ],