Merge pull request #43559 from crassirostris/cluster-logging-check-deployment

Automatic merge from submit-queue

Check fluentd deployment befure running cluster logging e2e tests

There were changes to the way cluster logging is deployed to the cluster.

PR adds logic to the cluster logging e2e tests to check that cluster has fluentd and that there's only one fluentd instance on the node. This will verify the correctness of the deployment method.

test/e2e/cluster_logging_es.go

@@ -44,6 +44,9 @@ var _ = framework.KubeDescribe("Cluster level logging using Elasticsearch [Featu
 		err = esLogsProvider.EnsureWorking()
 		framework.ExpectNoError(err, "Elasticsearch is not working")
 
+		err = ensureSingleFluentdOnEachNode(f, esLogsProvider.FluentdApplicationName())
+		framework.ExpectNoError(err, "Fluentd deployed incorrectly")
+
 		By("Running synthetic logger")
 		pod := createLoggingPod(f, podName, 10*60, 10*time.Minute)
 		defer f.PodClient().Delete(podName, &meta_v1.DeleteOptions{})

test/e2e/cluster_logging_gcl.go

@@ -42,6 +42,9 @@ var _ = framework.KubeDescribe("Cluster level logging using GCL", func() {
 		err = gclLogsProvider.EnsureWorking()
 		framework.ExpectNoError(err, "GCL is not working")
 
+		err = ensureSingleFluentdOnEachNode(f, gclLogsProvider.FluentdApplicationName())
+		framework.ExpectNoError(err, "Fluentd deployed incorrectly")
+
 		By("Running synthetic logger")
 		pod := createLoggingPod(f, podName, 10*60, 10*time.Minute)
 		defer f.PodClient().Delete(podName, &meta_v1.DeleteOptions{})

test/e2e/cluster_logging_utils.go

@@ -284,6 +284,31 @@ func getMissingLinesCount(logsProvider logsProvider, pod *loggingPod) (int, erro
 	return pod.ExpectedLinesNumber - len(pod.Occurrences), nil
 }
 
+func ensureSingleFluentdOnEachNode(f *framework.Framework, fluentdApplicationName string) error {
+	fluentdPodList, err := getFluentdPods(f, fluentdApplicationName)
+	if err != nil {
+		return err
+	}
+
+	fluentdPodsPerNode := make(map[string]int)
+	for _, fluentdPod := range fluentdPodList.Items {
+		fluentdPodsPerNode[fluentdPod.Spec.NodeName]++
+	}
+
+	nodeList := framework.GetReadySchedulableNodesOrDie(f.ClientSet)
+	for _, node := range nodeList.Items {
+		fluentdPodCount, ok := fluentdPodsPerNode[node.Name]
+
+		if !ok {
+			return fmt.Errorf("node %s doesn't have fluentd instance", node.Name)
+		} else if fluentdPodCount != 1 {
+			return fmt.Errorf("node %s contains %d fluentd instaces, expected exactly one", node.Name, fluentdPodCount)
+		}
+	}
+
+	return nil
+}
+
 func getFluentdPods(f *framework.Framework, fluentdApplicationName string) (*api_v1.PodList, error) {
 	label := labels.SelectorFromSet(labels.Set(map[string]string{"k8s-app": fluentdApplicationName}))
 	options := meta_v1.ListOptions{LabelSelector: label.String()}