kubescape/core/pkg/hostsensorutils/hostsensordeploy.go

package hostsensorutils

import (
	"context"
	_ "embed"
	"fmt"
	"os"
	"sync"
	"time"

	logger "github.com/kubescape/go-logger"
	"github.com/kubescape/go-logger/helpers"
	"github.com/kubescape/k8s-interface/k8sinterface"
	"github.com/kubescape/k8s-interface/workloadinterface"
	"github.com/kubescape/kubescape/v2/core/cautils"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/watch"
)

var (
	//go:embed hostsensor.yaml
	hostSensorYAML      string
	namespaceWasPresent bool
)

const portName string = "scanner"

// HostSensorHandler is a client that interacts with a host-scanner component deployed on nodes.
//
// The API exposed by the host sensor is defined here: https://github.com/kubescape/host-scanner
type HostSensorHandler struct {
	hostSensorPort                int32
	hostSensorPodNames            map[string]string //map from pod names to node names
	hostSensorUnscheduledPodNames map[string]string //map from pod names to node names
	k8sObj                        *k8sinterface.KubernetesApi
	daemonSet                     *appsv1.DaemonSet
	podListLock                   sync.RWMutex
	gracePeriod                   int64
	workerPool                    workerPool
}

// NewHostSensorHandler builds a new http client to the host-scanner API.
func NewHostSensorHandler(k8sObj *k8sinterface.KubernetesApi, hostSensorYAMLFile string) (*HostSensorHandler, error) {
	if k8sObj == nil {
		return nil, fmt.Errorf("nil k8s interface received")
	}

	if hostSensorYAMLFile != "" {
		d, err := loadHostSensorFromFile(hostSensorYAMLFile)
		if err != nil {
			return nil, fmt.Errorf("failed to load host-scan yaml file, reason: %w", err)
		}
		hostSensorYAML = d
	}

	hsh := &HostSensorHandler{
		k8sObj:                        k8sObj,
		hostSensorPodNames:            map[string]string{},
		hostSensorUnscheduledPodNames: map[string]string{},
		gracePeriod:                   int64(15),
		workerPool:                    newWorkerPool(),
	}

	// Don't deploy on a cluster with no nodes. Some cloud providers prevent the termination of K8s objects for cluster with no nodes!!!
	if nodeList, err := k8sObj.KubernetesClient.CoreV1().Nodes().List(k8sObj.Context, metav1.ListOptions{}); err != nil || len(nodeList.Items) == 0 {
		if err == nil {
			err = fmt.Errorf("no nodes to scan")
		}
		return hsh, fmt.Errorf("in NewHostSensorHandler, failed to get nodes list: %v", err)
	}

	return hsh, nil
}

// Init deploys the host-scanner and start watching the pods on the host.
func (hsh *HostSensorHandler) Init(ctx context.Context) error {
	// deploy the YAML
	// store namespace + port
	// 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.")

	cautils.StartSpinner()
	defer cautils.StopSpinner()

	if err := hsh.applyYAML(ctx); err != nil {
		return fmt.Errorf("failed to apply host scanner YAML, reason: %v", err)
	}

	hsh.populatePodNamesToNodeNames(ctx)
	if err := hsh.checkPodForEachNode(); err != nil {
		logger.L().Ctx(ctx).Warning("failed to validate host-sensor pods status", helpers.Error(err))
	}

	return nil
}

// checkNamespaceWasPresent check if the given namespace was already present on kubernetes and in "Active" state.
// Return true in case it find the namespace on the list, false otherwise.
// In case we have some error with the kubernetes APIs, it returns an error.
func (hsh *HostSensorHandler) checkNamespaceWasPresent(namespace string) bool {
	ns, err := hsh.k8sObj.KubernetesClient.
		CoreV1().
		Namespaces().
		Get(hsh.k8sObj.Context, namespace, metav1.GetOptions{})
	if err != nil {
		return false
	}

	// check also if it is in "Active" state.
	if ns.Status.Phase != corev1.NamespaceActive {
		return false
	}

	return true
}

// namespaceWasPresent return the namespaceWasPresent variable value.
func (hsh *HostSensorHandler) namespaceWasPresent() bool {
	return namespaceWasPresent
}

func (hsh *HostSensorHandler) applyYAML(ctx context.Context) error {
	workloads, err := cautils.ReadFile([]byte(hostSensorYAML), cautils.YAML_FILE_FORMAT)
	if err != nil {
		return fmt.Errorf("failed to read YAML files, reason: %v", err)
	}

	// Get namespace name
	namespaceName := ""
	for i := range workloads {
		if workloads[i].GetKind() == "Namespace" {
			namespaceName = workloads[i].GetName()
			break
		}
	}
	// check if namespace was already present on kubernetes
	namespaceWasPresent = hsh.checkNamespaceWasPresent(namespaceName)

	// Update workload data before applying
	for i := range workloads {
		w := workloadinterface.NewWorkloadObj(workloads[i].GetObject())
		if w == nil {
			return fmt.Errorf("invalid workload: %v", workloads[i].GetObject())
		}
		// set namespace in all objects
		if w.GetKind() != "Namespace" {
			w.SetNamespace(namespaceName)
		}
		// Get container port
		if w.GetKind() == "DaemonSet" {
			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))
				}
				return fmt.Errorf("container not found in DaemonSet: %v", err)
			}
			for j := range containers {
				for k := range containers[j].Ports {
					if containers[j].Ports[k].Name == portName {
						hsh.hostSensorPort = containers[j].Ports[k].ContainerPort
					}
				}
			}
		}

		// Apply workload
		var newWorkload k8sinterface.IWorkload
		var e error

		if g, err := hsh.k8sObj.GetWorkload(w.GetNamespace(), w.GetKind(), w.GetName()); err == nil && g != nil {
			newWorkload, e = hsh.k8sObj.UpdateWorkload(w)
		} else {
			newWorkload, e = hsh.k8sObj.CreateWorkload(w)
		}
		if e != nil {
			if erra := hsh.tearDownNamespace(namespaceName); erra != nil {
				logger.L().Ctx(ctx).Warning("failed to tear down namespace", helpers.Error(erra))
			}
			return fmt.Errorf("failed to create/update YAML, reason: %v", e)
		}

		// Save DaemonSet
		if newWorkload.GetKind() == "DaemonSet" {
			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))
				}
				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))
				}
				return fmt.Errorf("failed to Unmarshal YAML of DaemonSet, reason: %v", err)
			}
			hsh.daemonSet = &ds
		}
	}

	return nil
}

func (hsh *HostSensorHandler) checkPodForEachNode() error {
	deadline := time.Now().Add(time.Second * 100)
	for {
		nodesList, err := hsh.k8sObj.KubernetesClient.CoreV1().Nodes().List(hsh.k8sObj.Context, metav1.ListOptions{})
		if err != nil {
			return fmt.Errorf("in checkPodsForEveryNode, failed to get nodes list: %v", nodesList)
		}

		hsh.podListLock.RLock()
		podsNum := len(hsh.hostSensorPodNames)
		unschedPodNum := len(hsh.hostSensorUnscheduledPodNames)
		hsh.podListLock.RUnlock()
		if len(nodesList.Items) <= podsNum+unschedPodNum {
			break
		}

		if time.Now().After(deadline) {
			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",
				podsNum, len(nodesList.Items), podsMap)
		}
		time.Sleep(100 * time.Millisecond)
	}

	return nil
}

// initiating routine to keep pod list updated
func (hsh *HostSensorHandler) populatePodNamesToNodeNames(ctx context.Context) {
	go func() {
		var watchRes watch.Interface
		var err error
		watchRes, err = hsh.k8sObj.KubernetesClient.CoreV1().Pods(hsh.daemonSet.Namespace).Watch(hsh.k8sObj.Context, metav1.ListOptions{
			Watch:         true,
			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))
		}
		if watchRes == nil {
			logger.L().Ctx(ctx).Error("failed to watch over DaemonSet pods, will not be able to get host-sensor data")
			return
		}

		for eve := range watchRes.ResultChan() {
			pod, ok := eve.Object.(*corev1.Pod)
			if !ok {
				continue
			}
			go hsh.updatePodInListAtomic(ctx, eve.Type, pod)
		}
	}()
}

func (hsh *HostSensorHandler) updatePodInListAtomic(ctx context.Context, eventType watch.EventType, podObj *corev1.Pod) {
	hsh.podListLock.Lock()
	defer hsh.podListLock.Unlock()

	switch eventType {
	case watch.Added, watch.Modified:
		if podObj.Status.Phase == corev1.PodRunning && len(podObj.Status.ContainerStatuses) > 0 &&
			podObj.Status.ContainerStatuses[0].Ready {
			hsh.hostSensorPodNames[podObj.ObjectMeta.Name] = podObj.Spec.NodeName
			delete(hsh.hostSensorUnscheduledPodNames, podObj.ObjectMeta.Name)
		} else {
			if podObj.Status.Phase == corev1.PodPending && len(podObj.Status.Conditions) > 0 &&
				podObj.Status.Conditions[0].Reason == corev1.PodReasonUnschedulable {
				nodeName := ""
				if podObj.Spec.Affinity != nil && podObj.Spec.Affinity.NodeAffinity != nil &&
					podObj.Spec.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution != nil &&
					len(podObj.Spec.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms) > 0 &&
					len(podObj.Spec.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchFields) > 0 &&
					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 nodeName != "" {
					hsh.hostSensorUnscheduledPodNames[podObj.ObjectMeta.Name] = nodeName
				}
			} else {
				delete(hsh.hostSensorPodNames, podObj.ObjectMeta.Name)
			}
		}
	default:
		delete(hsh.hostSensorPodNames, podObj.ObjectMeta.Name)
	}
}

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)
	}

	return nil
}

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)
	}
	// delete Namespace
	if err := hsh.tearDownNamespace(namespace); err != nil {
		return fmt.Errorf("failed to delete host-sensor daemonset: %v", err)
	}
	// TODO: wait for termination? may take up to 120 seconds!!!

	return nil
}

func (hsh *HostSensorHandler) GetNamespace() string {
	if hsh.daemonSet == nil {
		return ""
	}
	return hsh.daemonSet.Namespace
}

func loadHostSensorFromFile(hostSensorYAMLFile string) (string, error) {
	dat, err := os.ReadFile(hostSensorYAMLFile)
	if err != nil {
		return "", err
	}
	// TODO - Add file validation
	return string(dat), err
}