Updating stage branch (#695)

* OCPBUGS-2249: fix the schema checking conditional gathering rules (#687)

* refactor(workloads_info): improving code reability (#650)

This was the first try to improve workloads_info reability.
It still have space to improve, but at least it reduces the funlen.

* OCPBUGS-2915: Updated info link in insights recommendations (#683)

* Updated info link in insights recommendations

* updated AdvisorURI in insightsreport

* unit tests update

* updated code in case of GetClusterVersion() error

* small fix in recommendation link

* reworking how info link gets cluster ID

* update of GetClusterVersion() in case of error

* Updated where clusterID is being set, created interface for mocking insightsClient inside unit tests

* updated interface name and noop-ed function in unit tests

* Use cgroups memory usage data in the archive metadata (#693)

Co-authored-by: Tomas Remes <[email protected]>
Co-authored-by: Ricardo Lüders <[email protected]>
Co-authored-by: Richard Hrmo <[email protected]>

Author: 4 people

pkg/gather/gather.go

@@ -4,7 +4,9 @@ package gather
 import (
 	"context"
 	"fmt"
-	"runtime"
+	"os"
+	"strconv"
+	"strings"
 	"time"
 
 	"k8s.io/client-go/rest"
@@ -47,8 +49,9 @@ type GathererFunctionReport struct {
 type ArchiveMetadata struct {
 	// info about gathering functions.
 	StatusReports []GathererFunctionReport `json:"status_reports"`
-	// MemoryAlloc is the amount of memory taken by heap objects after processing the records
-	MemoryAlloc uint64 `json:"memory_alloc_bytes"`
+	// MemoryBytesUsage is the number of bytes of memory used by the container. The number is obtained
+	// from cgroups and is related to the Prometheus metric with the same name.
+	MemoryBytesUsage uint64 `json:"container_memory_bytes_usage"`
 	// Uptime is the number of seconds from the program start till the point when metadata was created
 	Uptime float64 `json:"uptime_seconds"`
 	// IsGlobalObfuscationEnabled shows if obfuscation(hiding IPs and cluster domain) is enabled
@@ -192,20 +195,31 @@ func recordGatheringFunctionResult(
 	}, allErrors
 }
 
+func readMemoryUsage() (int, error) {
+	b, err := os.ReadFile("/sys/fs/cgroup/memory/memory.usage_in_bytes")
+	if err != nil {
+		return 0, err
+	}
+	memUsage := strings.ReplaceAll(string(b), "\n", "")
+	return strconv.Atoi(memUsage)
+}
+
 // RecordArchiveMetadata records info about archive and gatherers' reports
 func RecordArchiveMetadata(
 	functionReports []GathererFunctionReport,
 	rec recorder.Interface,
 	anonymizer *anonymization.Anonymizer,
 ) error {
-	var m runtime.MemStats
-	runtime.ReadMemStats(&m)
+	memUsage, err := readMemoryUsage()
+	if err != nil {
+		klog.Warningf("can't read cgroups memory usage data: %v", err)
+	}
 
 	archiveMetadata := record.Record{
 		Name: recorder.MetadataRecordName,
 		Item: record.JSONMarshaller{Object: ArchiveMetadata{
 			StatusReports:              functionReports,
-			MemoryAlloc:                m.HeapAlloc,
+			MemoryBytesUsage:           uint64(memUsage),
 			Uptime:                     time.Since(programStartTime).Truncate(time.Millisecond).Seconds(),
 			IsGlobalObfuscationEnabled: anonymizer.IsObfuscationEnabled(),
 		}},

pkg/gatherers/conditional/gathering_rule.schema.json

@@ -98,12 +98,12 @@
                                     "version"
                                 ],
                                 "properties": {
-                                    "name": {
+                                    "version": {
                                         "type": "string",
                                         "title": "Version",
                                         "description": "Version contains a semantic versioning expression",
                                         "minLength": 1,
-                                        "maxLength": 32
+                                        "maxLength": 64
                                     }
                                 }
                             }

pkg/gatherers/workloads/workloads_info.go

@@ -63,37 +63,63 @@ func (g *Gatherer) GatherWorkloadInfo(ctx context.Context) ([]record.Record, []e
 	return gatherWorkloadInfo(ctx, gatherKubeClient.CoreV1(), gatherOpenShiftClient)
 }
 
-// nolint: funlen, gocyclo, gocritic
 func gatherWorkloadInfo(
 	ctx context.Context,
 	coreClient corev1client.CoreV1Interface,
 	imageClient imageclient.ImageV1Interface,
 ) ([]record.Record, []error) {
-	// load images as we find them
-	imageCh := make(chan string, workloadGatherPageSize)
-	imagesDoneCh := gatherWorkloadImageInfo(ctx, imageClient.Images(), imageCh)
+	imageCh, imagesDoneCh := gatherWorkloadImageInfo(ctx, imageClient.Images())
 
-	// load pods in order
 	start := time.Now()
+	limitReached, info, err := workloadInfo(ctx, coreClient, imageCh)
+	if err != nil {
+		return nil, []error{err}
+	}
+
+	workloadImageResize(info.PodCount)
+
+	records := []record.Record{
+		{
+			Name: "config/workload_info",
+			Item: record.JSONMarshaller{Object: &info},
+		},
+	}
+
+	// wait for as many images as we can find to load
+	handleWorkloadImageInfo(ctx, &info, start, imagesDoneCh)
+
+	if limitReached {
+		return records, []error{fmt.Errorf("the %d limit for number of pods gathered was reached", podsLimit)}
+	}
+	return records, nil
+}
+
+// nolint: funlen, gocritic, gocyclo
+func workloadInfo(
+	ctx context.Context,
+	coreClient corev1client.CoreV1Interface,
+	imageCh chan string,
+) (bool, workloadPods, error) {
+	defer close(imageCh)
 	limitReached := false
 
 	var info workloadPods
-	var namespace string
-	var namespaceHash string
+	var namespace, namespaceHash, continueValue string
 	var namespacePods workloadNamespacePods
 	h := sha256.New()
 
 	// Use the Limit and Continue fields to request the pod information in chunks.
-	var continueValue string
 	for {
 		pods, err := coreClient.Pods("").List(ctx, metav1.ListOptions{
 			Limit:    workloadGatherPageSize,
 			Continue: continueValue,
 		})
 		if err != nil {
-			return nil, []error{err}
+			return false, workloadPods{}, err
 		}
-		for _, pod := range pods.Items {
+
+		for podIdx := range pods.Items {
+			pod := pods.Items[podIdx]
 			// initialize the running state, including the namespace hash
 			if pod.Namespace != namespace {
 				if len(namespace) != 0 {
@@ -117,45 +143,35 @@ func gatherWorkloadInfo(
 			namespacePods.Count++
 
 			switch {
-			case pod.Status.Phase == corev1.PodSucceeded, pod.Status.Phase == corev1.PodFailed:
+			case isPodTerminated(&pod):
 				// track terminal pods but do not report their data
 				namespacePods.TerminalCount++
 				continue
-			case pod.Status.Phase != corev1.PodRunning && pod.Status.Phase != corev1.PodPending,
-				len(pod.Status.InitContainerStatuses) != len(pod.Spec.InitContainers),
-				len(pod.Status.ContainerStatuses) != len(pod.Spec.Containers):
+			case podCanBeIgnored(&pod):
 				// consider pods that are in a known state
 				// or pods without filled out status are invalid
 				namespacePods.IgnoredCount++
 				continue
 			}
 
-			var podShape workloadPodShape
-			var ok bool
-			podShape.InitContainers, ok = calculateWorkloadContainerShapes(h, pod.Spec.InitContainers, pod.Status.InitContainerStatuses)
+			podShape, ok := calculatePodShape(h, &pod)
 			if !ok {
 				namespacePods.InvalidCount++
 				continue
 			}
-			podShape.Containers, ok = calculateWorkloadContainerShapes(h, pod.Spec.Containers, pod.Status.ContainerStatuses)
-			if !ok {
-				namespacePods.InvalidCount++
-				continue
-			}
-
-			podShape.RestartsAlways = pod.Spec.RestartPolicy == corev1.RestartPolicyAlways
 
 			if index := workloadPodShapeIndex(namespacePods.Shapes, podShape); index != -1 {
 				namespacePods.Shapes[index].Duplicates++
-			} else {
-				namespacePods.Shapes = append(namespacePods.Shapes, podShape)
+				continue
+			}
+			namespacePods.Shapes = append(namespacePods.Shapes, podShape)
 
-				for _, container := range podShape.InitContainers {
-					imageCh <- container.ImageID
-				}
-				for _, container := range podShape.Containers {
-					imageCh <- container.ImageID
-				}
+			for _, container := range podShape.InitContainers {
+				imageCh <- container.ImageID
+			}
+
+			for _, container := range podShape.Containers {
+				imageCh <- container.ImageID
 			}
 		}
 
@@ -166,6 +182,7 @@ func gatherWorkloadInfo(
 		}
 		continueValue = pods.Continue
 	}
+
 	// add the last set of pods
 	if len(namespace) != 0 {
 		if info.Namespaces == nil {
@@ -175,23 +192,53 @@ func gatherWorkloadInfo(
 		info.PodCount += namespacePods.Count
 	}
 
-	workloadImageResize(info.PodCount)
+	return limitReached, info, nil
+}
 
-	records := []record.Record{
-		{
-			Name: "config/workload_info",
-			Item: record.JSONMarshaller{Object: &info},
-		},
+func isPodTerminated(pod *corev1.Pod) bool {
+	return pod.Status.Phase == corev1.PodSucceeded ||
+		pod.Status.Phase == corev1.PodFailed
+}
+
+func podCanBeIgnored(pod *corev1.Pod) bool {
+	return pod.Status.Phase != corev1.PodRunning && pod.Status.Phase != corev1.PodPending ||
+		len(pod.Status.InitContainerStatuses) != len(pod.Spec.InitContainers) ||
+		len(pod.Status.ContainerStatuses) != len(pod.Spec.Containers)
+}
+
+func calculatePodShape(h hash.Hash, pod *corev1.Pod) (workloadPodShape, bool) {
+	var podShape workloadPodShape
+	var ok bool
+	podShape.InitContainers, ok = calculateWorkloadContainerShapes(h, pod.Spec.InitContainers, pod.Status.InitContainerStatuses)
+	if !ok {
+		return workloadPodShape{}, false
 	}
 
-	// wait for as many images as we can find to load
+	podShape.Containers, ok = calculateWorkloadContainerShapes(h, pod.Spec.Containers, pod.Status.ContainerStatuses)
+	if !ok {
+		return workloadPodShape{}, false
+	}
+
+	podShape.RestartsAlways = pod.Spec.RestartPolicy == corev1.RestartPolicyAlways
+
+	return podShape, true
+}
+
+func handleWorkloadImageInfo(
+	ctx context.Context,
+	info *workloadPods,
+	start time.Time,
+	imagesDoneCh <-chan workloadImageInfo,
+) {
 	var imageInfo workloadImageInfo
+
 	// wait proportional to the number of pods + a floor
 	waitDuration := time.Second*time.Duration(info.PodCount)/10 + 15*time.Second
+
 	klog.V(2).Infof("Loaded pods in %s, will wait %s for image data",
 		time.Since(start).Round(time.Second).String(),
 		waitDuration.Round(time.Second).String())
-	close(imageCh)
+
 	select {
 	case <-ctx.Done():
 		select {
@@ -208,19 +255,15 @@ func gatherWorkloadInfo(
 
 	info.Images = imageInfo.images
 	info.ImageCount = imageInfo.count
-	if limitReached {
-		return records, []error{fmt.Errorf("the %d limit for number of pods gathered was reached", podsLimit)}
-	}
-	return records, nil
 }
 
-// nolint: gocyclo
+// nolint: gocritic
 func gatherWorkloadImageInfo(
 	ctx context.Context,
 	imageClient imageclient.ImageInterface,
-	imageCh <-chan string,
-) <-chan workloadImageInfo {
+) (chan string, <-chan workloadImageInfo) {
 	images := make(map[string]workloadImage)
+	imageCh := make(chan string, workloadGatherPageSize)
 	imagesDoneCh := make(chan workloadImageInfo)
 
 	go func() {
@@ -254,17 +297,8 @@ func gatherWorkloadImageInfo(
 				for k := range pendingIDs {
 					delete(pendingIDs, k)
 				}
-				if _, ok := images[imageID]; !ok {
-					pendingIDs[imageID] = struct{}{}
-				}
-				for l := len(imageCh); l > 0; l = len(imageCh) {
-					for i := 0; i < l; i++ {
-						imageID := <-imageCh
-						if _, ok := images[imageID]; !ok {
-							pendingIDs[imageID] = struct{}{}
-						}
-					}
-				}
+
+				readImageSHAs(pendingIDs, images, imageID, imageCh)
 
 				for imageID := range pendingIDs {
 					if _, ok := images[imageID]; ok {
@@ -274,29 +308,55 @@ func gatherWorkloadImageInfo(
 						images[imageID] = image
 						continue
 					}
+
 					images[imageID] = workloadImage{}
-					start := time.Now()
-					image, err := imageClient.Get(ctx, imageID, metav1.GetOptions{})
-					if errors.IsNotFound(err) {
-						klog.V(4).Infof("No image %s (%s)", imageID, time.Since(start).Round(time.Millisecond).String())
+					image := imageFromID(ctx, imageClient, imageID)
+					if image == nil {
 						continue
 					}
-					if err == context.Canceled {
-						return
-					}
-					if err != nil {
-						klog.Errorf("Unable to retrieve image %s", imageID)
-						continue
-					}
-					klog.V(4).Infof("Found image %s (%s)", imageID, time.Since(start).Round(time.Millisecond).String())
 					info := calculateWorkloadInfo(h, image)
 					images[imageID] = info
 					workloadImageAdd(imageID, info)
 				}
 			}
 		}
 	}()
-	return imagesDoneCh
+	return imageCh, imagesDoneCh
+}
+
+// readImageSHAs drains the channel of any image IDs
+func readImageSHAs(pendingIDs map[string]struct{}, images map[string]workloadImage, id string, imageCh <-chan string) {
+	if _, ok := images[id]; !ok {
+		pendingIDs[id] = struct{}{}
+	}
+	for l := len(imageCh); l > 0; l = len(imageCh) {
+		for i := 0; i < l; i++ {
+			imageID := <-imageCh
+			if _, ok := images[imageID]; !ok {
+				pendingIDs[imageID] = struct{}{}
+			}
+		}
+	}
+}
+
+// imageFromID gets the container image from given ID
+func imageFromID(ctx context.Context, client imageclient.ImageInterface, id string) *imagev1.Image {
+	start := time.Now()
+	image, err := client.Get(ctx, id, metav1.GetOptions{})
+	if errors.IsNotFound(err) {
+		klog.V(4).Infof("No image %s (%s)", id, time.Since(start).Round(time.Millisecond).String())
+		return nil
+	}
+	if err == context.Canceled {
+		return nil
+	}
+	if err != nil {
+		klog.Errorf("Unable to retrieve image %s", id)
+		return nil
+	}
+
+	klog.V(4).Infof("Found image %s (%s)", id, time.Since(start).Round(time.Millisecond).String())
+	return image
 }
 
 // workloadPodShapeIndex attempts to find an equivalent shape within the current