Merge remote-tracking branch 'upstream/dev'

hostsensorutils/hostsensor.yaml

@@ -42,7 +42,7 @@ spec:
             containerPort: 7888
         resources:
           limits:
-            cpu: 1m
+            cpu: 0.1m
             memory: 200Mi
           requests:
             cpu: 1m

hostsensorutils/hostsensordeploy.go

@@ -27,13 +27,14 @@ var (
 )
 
 type HostSensorHandler struct {
-	HostSensorPort     int32
-	HostSensorPodNames map[string]string //map from pod names to node names
-	IsReady            <-chan bool       //readonly chan
-	k8sObj             *k8sinterface.KubernetesApi
-	DaemonSet          *appsv1.DaemonSet
-	podListLock        sync.RWMutex
-	gracePeriod        int64
+	HostSensorPort              int32
+	HostSensorPodNames          map[string]string //map from pod names to node names
+	HostSensorUnshedulePodNames map[string]string //map from pod names to node names
+	IsReady                     <-chan bool       //readonly chan
+	k8sObj                      *k8sinterface.KubernetesApi
+	DaemonSet                   *appsv1.DaemonSet
+	podListLock                 sync.RWMutex
+	gracePeriod                 int64
 }
 
 func NewHostSensorHandler(k8sObj *k8sinterface.KubernetesApi) (*HostSensorHandler, error) {
@@ -42,9 +43,10 @@ func NewHostSensorHandler(k8sObj *k8sinterface.KubernetesApi) (*HostSensorHandle
 		return nil, fmt.Errorf("nil k8s interface received")
 	}
 	hsh := &HostSensorHandler{
-		k8sObj:             k8sObj,
-		HostSensorPodNames: map[string]string{},
-		gracePeriod:        int64(15),
+		k8sObj:                      k8sObj,
+		HostSensorPodNames:          map[string]string{},
+		HostSensorUnshedulePodNames: map[string]string{},
+		gracePeriod:                 int64(15),
 	}
 	// Don't deploy on cluster with no nodes. Some cloud providers prevents 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 {
@@ -140,12 +142,17 @@ func (hsh *HostSensorHandler) checkPodForEachNode() error {
 		}
 		hsh.podListLock.RLock()
 		podsNum := len(hsh.HostSensorPodNames)
+		unschedPodNum := len(hsh.HostSensorUnshedulePodNames)
 		hsh.podListLock.RUnlock()
-		if len(nodesList.Items) == podsNum {
+		if len(nodesList.Items) <= podsNum+unschedPodNum {
 			break
 		}
 		if time.Now().After(deadline) {
-			return fmt.Errorf("host-sensor pods number (%d) differ than nodes number (%d) after deadline exceded", podsNum, len(nodesList.Items))
+			hsh.podListLock.RLock()
+			podsMap := hsh.HostSensorPodNames
+			hsh.podListLock.RUnlock()
+			return fmt.Errorf("host-sensor pods number (%d) differ than nodes number (%d) after deadline exceded. We will take data only from the pods below: %v",
+				podsNum, len(nodesList.Items), podsMap)
 		}
 		time.Sleep(100 * time.Millisecond)
 	}
@@ -156,12 +163,17 @@ func (hsh *HostSensorHandler) checkPodForEachNode() error {
 func (hsh *HostSensorHandler) populatePodNamesToNodeNames() {
 
 	go func() {
-		watchRes, err := hsh.k8sObj.KubernetesClient.CoreV1().Pods(hsh.DaemonSet.Namespace).Watch(hsh.k8sObj.Context, metav1.ListOptions{
+		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().Error("failed to watch over daemonset pods", helpers.Error(err))
+			logger.L().Error("failed to watch over daemonset pods - are we missing watch pods permissions?", helpers.Error(err))
+		}
+		if watchRes == nil {
+			return
 		}
 		for eve := range watchRes.ResultChan() {
 			pod, ok := eve.Object.(*corev1.Pod)
@@ -179,10 +191,31 @@ func (hsh *HostSensorHandler) updatePodInListAtomic(eventType watch.EventType, p
 
 	switch eventType {
 	case watch.Added, watch.Modified:
-		if podObj.Status.Phase == corev1.PodRunning && podObj.Status.ContainerStatuses[0].Ready {
+		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.HostSensorUnshedulePodNames, podObj.ObjectMeta.Name)
 		} else {
-			delete(hsh.HostSensorPodNames, podObj.ObjectMeta.Name)
+			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().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.HostSensorUnshedulePodNames[podObj.ObjectMeta.Name] = nodeName
+				}
+			} else {
+				delete(hsh.HostSensorPodNames, podObj.ObjectMeta.Name)
+			}
 		}
 	default:
 		delete(hsh.HostSensorPodNames, podObj.ObjectMeta.Name)