Check fluentd deployment befure running cluster logging e2e tests

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