Conditional gatherer of logs of unhealthy pods (#509)

* Cond. gather of unhealthy pods' logs

* Add Previous switch/prop to log msg filter struct

* Add FieldSelector to log gathering helper func

* Add missing fields to gatherer params

* Improve cond. gather of unhealthy pod logs

* Add tests for unhealthy pod logs cond. gatherer

* Update cond. gatherer rules JSON in sample archive

* Simplify unhealthy pod logs cond. gatherer

* Update sample cond. gatherer rules JSON

* Refactor pod log test to satisfy linter

* Make linter even happier (pod log cond. gather)

* Fix linter long lines in pod log cond. gather

* Add alert name pattern to pod log cond gather conf

* Couple of tweaks to unhealthy pod log cond. gahter

* Update unhealthy pod log cond. gatherer tests

* Update Gathered Data docs

Author: natiiix

docs/gathered-data.md

@@ -356,6 +356,15 @@ Response see:
   * 4.9+
 
 
+## LogsOfUnhealthyPods
+
+collects either current or previous logs for pods firing one of the configured alerts.
+
+* Location in archive: conditional/namespaces/<namespace>/pods/<pod>/containers/<container>/<logs|logs-previous>/last-<tail length>-lines.log
+* Since versions:
+  * 4.10+
+
+
 ## MachineAutoscalers
 
 collects MachineAutoscalers definition

docs/insights-archive-sample/insights-operator/conditional-gatherer-rules.json

@@ -9,14 +9,62 @@
       }
     ],
     "gathering_functions": {
-      "image_stream_definitions_of_namespace": {
+      "image_streams_of_namespace": {
         "namespace": "openshift-cluster-samples-operator"
       },
       "logs_of_namespace": {
-        "label_selector": "",
         "namespace": "openshift-cluster-samples-operator",
         "tail_lines": 100
       }
     }
+  },
+  {
+    "conditions": [
+      {
+        "type": "alert_is_firing",
+        "params": {
+          "name": "APIRemovedInNextEUSReleaseInUse"
+        }
+      }
+    ],
+    "gathering_functions": {
+      "api_request_counts_of_resource_from_alert": {
+        "alert_name": "APIRemovedInNextEUSReleaseInUse"
+      }
+    }
+  },
+  {
+    "conditions": [
+      {
+        "type": "alert_is_firing",
+        "params": {
+          "name": "KubePodCrashLooping"
+        }
+      }
+    ],
+    "gathering_functions": {
+      "logs_of_unhealthy_pods": {
+        "alert_name": "KubePodCrashLooping",
+        "tail_lines": 20,
+        "previous": true
+      }
+    }
+  },
+  {
+    "conditions": [
+      {
+        "type": "alert_is_firing",
+        "params": {
+          "name": "KubePodNotReady"
+        }
+      }
+    ],
+    "gathering_functions": {
+      "logs_of_unhealthy_pods": {
+        "alert_name": "KubePodNotReady",
+        "tail_lines": 100,
+        "previous": false
+      }
+    }
   }
 ]

pkg/gatherers/common/gather_logs.go

@@ -21,6 +21,7 @@ import (
 type LogContainersFilter struct {
 	Namespace                string
 	LabelSelector            string
+	FieldSelector            string
 	ContainerNameRegexFilter string
 }
 
@@ -31,6 +32,7 @@ type LogMessagesFilter struct {
 	SinceSeconds     int64
 	LimitBytes       int64
 	TailLines        int64
+	Previous         bool
 }
 
 // CollectLogsFromContainers collects logs from containers
@@ -65,6 +67,7 @@ func CollectLogsFromContainers( //nolint:gocyclo
 
 	pods, err := coreClient.Pods(containersFilter.Namespace).List(ctx, metav1.ListOptions{
 		LabelSelector: containersFilter.LabelSelector,
+		FieldSelector: containersFilter.FieldSelector,
 	})
 	if err != nil {
 		return nil, err
@@ -114,6 +117,7 @@ func CollectLogsFromContainers( //nolint:gocyclo
 				SinceSeconds: sinceSeconds,
 				LimitBytes:   limitBytes,
 				TailLines:    tailLines,
+				Previous:     messagesFilter.Previous,
 				Timestamps:   true,
 			})
 

pkg/gatherers/conditional/conditional_gatherer.go

@@ -28,6 +28,7 @@ var gatheringFunctionBuilders = map[GatheringFunctionName]GathererFunctionBuilde
 	GatherLogsOfNamespace:         (*Gatherer).BuildGatherLogsOfNamespace,
 	GatherImageStreamsOfNamespace: (*Gatherer).BuildGatherImageStreamsOfNamespace,
 	GatherAPIRequestCounts:        (*Gatherer).BuildGatherAPIRequestCounts,
+	GatherLogsOfUnhealthyPods:     (*Gatherer).BuildGatherLogsOfUnhealthyPods,
 }
 
 // gatheringRules contains all the rules used to run conditional gatherings.
@@ -93,6 +94,40 @@ var defaultGatheringRules = []GatheringRule{
 			},
 		},
 	},
+	{
+		Conditions: []ConditionWithParams{
+			{
+				Type: AlertIsFiring,
+				Params: AlertIsFiringConditionParams{
+					Name: "KubePodCrashLooping",
+				},
+			},
+		},
+		GatheringFunctions: GatheringFunctions{
+			GatherLogsOfUnhealthyPods: GatherLogsOfUnhealthyPodsParams{
+				AlertName: "KubePodCrashLooping",
+				TailLines: 20,
+				Previous:  true,
+			},
+		},
+	},
+	{
+		Conditions: []ConditionWithParams{
+			{
+				Type: AlertIsFiring,
+				Params: AlertIsFiringConditionParams{
+					Name: "KubePodNotReady",
+				},
+			},
+		},
+		GatheringFunctions: GatheringFunctions{
+			GatherLogsOfUnhealthyPods: GatherLogsOfUnhealthyPodsParams{
+				AlertName: "KubePodNotReady",
+				TailLines: 100,
+				Previous:  false,
+			},
+		},
+	},
 }
 
 const canConditionalGathererFail = false
@@ -230,24 +265,20 @@ func (g *Gatherer) updateAlertsCacheFromClient(ctx context.Context, metricsClien
 	if err != nil {
 		return err
 	}
-
 	var parser expfmt.TextParser
 	metricFamilies, err := parser.TextToMetricFamilies(bytes.NewReader(data))
 	if err != nil {
 		return err
 	}
-
 	if len(metricFamilies) > 1 {
 		// just log cuz everything would still work
 		klog.Warning(logPrefix + "unexpected output from prometheus metrics parser")
 	}
-
 	metricFamily, found := metricFamilies["ALERTS"]
 	if !found {
 		klog.Info(logPrefix + "no alerts are firing")
 		return nil
 	}
-
 	for _, metric := range metricFamily.GetMetric() {
 		if metric == nil {
 			klog.Info(logPrefix + "metric is nil")

pkg/gatherers/conditional/conditional_gatherer_test.go

@@ -166,7 +166,7 @@ func Test_Gatherer_GatherConditionalGathererRules(t *testing.T) {
 	err = json.Unmarshal(item, &gotGatheringRules)
 	assert.NoError(t, err)
 
-	assert.Len(t, gotGatheringRules, 2)
+	assert.Len(t, gotGatheringRules, 4)
 }
 
 func newFakeClientWithMetrics(metrics string) *fake.RESTClient {

pkg/gatherers/conditional/gather_logs_of_unhealthy_pods.go

@@ -0,0 +1,110 @@
+package conditional
+
+import (
+	"context"
+	"fmt"
+
+	"k8s.io/client-go/kubernetes"
+	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/klog/v2"
+
+	"github.com/openshift/insights-operator/pkg/gatherers"
+	"github.com/openshift/insights-operator/pkg/gatherers/common"
+	"github.com/openshift/insights-operator/pkg/record"
+)
+
+// BuildGatherLogsOfUnhealthyPods collects either current or previous logs for pods firing one of the configured alerts.
+//
+// * Location in archive: conditional/namespaces/<namespace>/pods/<pod>/containers/<container>/<logs|logs-previous>/last-<tail length>-lines.log
+// * Since versions:
+//   * 4.10+
+func (g *Gatherer) BuildGatherLogsOfUnhealthyPods(paramsInterface interface{}) (gatherers.GatheringClosure, error) {
+	params, ok := paramsInterface.(GatherLogsOfUnhealthyPodsParams)
+	if !ok {
+		return gatherers.GatheringClosure{}, fmt.Errorf(
+			"unexpected type in paramsInterface, expected %T, got %T",
+			GatherLogsOfUnhealthyPodsParams{}, paramsInterface,
+		)
+	}
+
+	return gatherers.GatheringClosure{
+		Run: func(ctx context.Context) ([]record.Record, []error) {
+			kubeClient, err := kubernetes.NewForConfig(g.gatherProtoKubeConfig)
+			if err != nil {
+				return nil, []error{err}
+			}
+			return g.gatherLogsOfUnhealthyPods(ctx, kubeClient.CoreV1(), params)
+		},
+		CanFail: canConditionalGathererFail,
+	}, nil
+}
+
+func (g *Gatherer) gatherLogsOfUnhealthyPods(
+	ctx context.Context, coreClient v1.CoreV1Interface, params GatherLogsOfUnhealthyPodsParams,
+) ([]record.Record, []error) {
+	errs := []error{}
+	records := []record.Record{}
+
+	alertInstances, ok := g.firingAlerts[params.AlertName]
+	if !ok {
+		return nil, []error{fmt.Errorf("conditional gatherer triggered, but specified alert %q is not firing", params.AlertName)}
+	}
+	for _, alertLabels := range alertInstances {
+		alertNamespace, ok := alertLabels["namespace"]
+		if !ok {
+			newErr := fmt.Errorf("alert is missing 'namespace' label")
+			klog.Warningln(newErr.Error())
+			errs = append(errs, newErr)
+			continue
+		}
+		alertPod, ok := alertLabels["pod"]
+		if !ok {
+			newErr := fmt.Errorf("alert is missing 'pod' label")
+			klog.Warningln(newErr.Error())
+			errs = append(errs, newErr)
+			continue
+		}
+		// The container label may not be present for all alerts (e.g., KubePodNotReady).
+		containerFilter := ""
+		if alertContainer, ok := alertLabels["container"]; ok && alertContainer != "" {
+			containerFilter = fmt.Sprintf("^%s$", alertContainer)
+		}
+
+		logRecords, err := common.CollectLogsFromContainers(ctx, coreClient,
+			common.LogContainersFilter{
+				Namespace:                alertNamespace,
+				FieldSelector:            fmt.Sprintf("metadata.name=%s", alertPod),
+				ContainerNameRegexFilter: containerFilter,
+			},
+			common.LogMessagesFilter{
+				TailLines: params.TailLines,
+				Previous:  params.Previous,
+			},
+			func(namespace string, podName string, containerName string) string {
+				logDirName := "logs"
+				if params.Previous {
+					logDirName = "logs-previous"
+				}
+				return fmt.Sprintf(
+					"%s/namespaces/%s/pods/%s/containers/%s/%s/last-%d-lines.log",
+					g.GetName(),
+					namespace,
+					podName,
+					containerName,
+					logDirName,
+					params.TailLines,
+				)
+			})
+		if err != nil {
+			// This can happen when the pod is destroyed but the alert still exists.
+			newErr := fmt.Errorf("unable to get container logs: %v", err)
+			klog.Warningln(newErr.Error())
+			errs = append(errs, newErr)
+			continue
+		}
+
+		records = append(records, logRecords...)
+	}
+
+	return records, errs
+}