|
| 1 | +package fluentd |
| 2 | + |
| 3 | +import ( |
| 4 | + "context" |
| 5 | + "crypto/rand" |
| 6 | + "encoding/hex" |
| 7 | + "fmt" |
| 8 | + "time" |
| 9 | + |
| 10 | + . "github.com/onsi/ginkgo/v2" |
| 11 | + . "github.com/onsi/gomega" |
| 12 | + appsv1 "k8s.io/api/apps/v1" |
| 13 | + corev1 "k8s.io/api/core/v1" |
| 14 | + apierrors "k8s.io/apimachinery/pkg/api/errors" |
| 15 | + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" |
| 16 | + "k8s.io/apimachinery/pkg/types" |
| 17 | + "k8s.io/utils/ptr" |
| 18 | + "sigs.k8s.io/controller-runtime/pkg/client" |
| 19 | + |
| 20 | + fluentdv1alpha1 "github.com/fluent/fluent-operator/v3/apis/fluentd/v1alpha1" |
| 21 | + "github.com/fluent/fluent-operator/v3/apis/fluentd/v1alpha1/plugins/input" |
| 22 | + "github.com/fluent/fluent-operator/v3/apis/fluentd/v1alpha1/plugins/output" |
| 23 | +) |
| 24 | + |
| 25 | +// generateRandomID creates a cryptographically random hex string |
| 26 | +func generateRandomID() string { |
| 27 | + b := make([]byte, 4) |
| 28 | + _, _ = rand.Read(b) |
| 29 | + return hex.EncodeToString(b) |
| 30 | +} |
| 31 | + |
| 32 | +var _ = Describe("Fluentd E2E Deployment Test", func() { |
| 33 | + var cancel context.CancelFunc |
| 34 | + var ctx context.Context |
| 35 | + |
| 36 | + BeforeEach(func() { |
| 37 | + // Create context with timeout to prevent hung tests |
| 38 | + ctx, cancel = context.WithTimeout(context.Background(), 5*time.Minute) |
| 39 | + }) |
| 40 | + |
| 41 | + AfterEach(func() { |
| 42 | + if cancel != nil { |
| 43 | + cancel() |
| 44 | + } |
| 45 | + }) |
| 46 | + |
| 47 | + Describe("Deploying Fluentd CR", func() { |
| 48 | + var ( |
| 49 | + fluentdCR *fluentdv1alpha1.Fluentd |
| 50 | + fluentdConfig *fluentdv1alpha1.FluentdConfig |
| 51 | + clusterOutput *fluentdv1alpha1.ClusterOutput |
| 52 | + namespace string |
| 53 | + ) |
| 54 | + |
| 55 | + BeforeEach(func() { |
| 56 | + // Generate a unique namespace using crypto/rand for true isolation |
| 57 | + namespace = fmt.Sprintf("fluentd-e2e-%s", generateRandomID()) |
| 58 | + ns := &corev1.Namespace{ |
| 59 | + ObjectMeta: metav1.ObjectMeta{ |
| 60 | + Name: namespace, |
| 61 | + }, |
| 62 | + } |
| 63 | + // Handle case where namespace might already exist from crashed previous run |
| 64 | + err := k8sClient.Create(ctx, ns) |
| 65 | + if err != nil && !apierrors.IsAlreadyExists(err) { |
| 66 | + Fail(fmt.Sprintf("Failed to create namespace: %v", err)) |
| 67 | + } |
| 68 | + |
| 69 | + // Create Fluentd CR with proper GlobalInputs type |
| 70 | + fluentdCR = &fluentdv1alpha1.Fluentd{ |
| 71 | + ObjectMeta: metav1.ObjectMeta{ |
| 72 | + Name: "fluentd-instance", |
| 73 | + Namespace: namespace, |
| 74 | + Labels: map[string]string{ |
| 75 | + "app.kubernetes.io/name": "fluentd", |
| 76 | + "app.kubernetes.io/instance": "fluentd-instance", |
| 77 | + }, |
| 78 | + }, |
| 79 | + Spec: fluentdv1alpha1.FluentdSpec{ |
| 80 | + Replicas: ptr.To(int32(1)), |
| 81 | + GlobalInputs: []input.Input{ |
| 82 | + { |
| 83 | + Forward: &input.Forward{ |
| 84 | + Bind: ptr.To("0.0.0.0"), |
| 85 | + Port: ptr.To(int32(24224)), |
| 86 | + }, |
| 87 | + }, |
| 88 | + }, |
| 89 | + // Explicitly set image as operator doesn't provide a default yet |
| 90 | + Image: "ghcr.io/fluent/fluent-operator/fluentd:v1.19.1", |
| 91 | + // Use EmptyDir for buffers to avoid PVC provisioning issues in CI |
| 92 | + BufferVolume: &fluentdv1alpha1.BufferVolume{ |
| 93 | + EmptyDir: &corev1.EmptyDirVolumeSource{}, |
| 94 | + }, |
| 95 | + FluentdCfgSelector: metav1.LabelSelector{ |
| 96 | + MatchLabels: map[string]string{ |
| 97 | + "config.fluentd.fluent.io/enabled": "true", |
| 98 | + }, |
| 99 | + }, |
| 100 | + }, |
| 101 | + } |
| 102 | + |
| 103 | + // Create a ClusterOutput for stdout (minimal working config) |
| 104 | + clusterOutput = &fluentdv1alpha1.ClusterOutput{ |
| 105 | + ObjectMeta: metav1.ObjectMeta{ |
| 106 | + Name: "fluentd-output-stdout", |
| 107 | + Labels: map[string]string{ |
| 108 | + "output.fluentd.fluent.io/enabled": "true", |
| 109 | + }, |
| 110 | + }, |
| 111 | + Spec: fluentdv1alpha1.ClusterOutputSpec{ |
| 112 | + Outputs: []output.Output{ |
| 113 | + { |
| 114 | + Stdout: &output.Stdout{}, |
| 115 | + }, |
| 116 | + }, |
| 117 | + }, |
| 118 | + } |
| 119 | + |
| 120 | + // Create FluentdConfig to wire everything together |
| 121 | + fluentdConfig = &fluentdv1alpha1.FluentdConfig{ |
| 122 | + ObjectMeta: metav1.ObjectMeta{ |
| 123 | + Name: "fluentd-config", |
| 124 | + Namespace: namespace, |
| 125 | + Labels: map[string]string{ |
| 126 | + "config.fluentd.fluent.io/enabled": "true", |
| 127 | + }, |
| 128 | + }, |
| 129 | + Spec: fluentdv1alpha1.FluentdConfigSpec{ |
| 130 | + ClusterOutputSelector: &metav1.LabelSelector{ |
| 131 | + MatchLabels: map[string]string{ |
| 132 | + "output.fluentd.fluent.io/enabled": "true", |
| 133 | + }, |
| 134 | + }, |
| 135 | + }, |
| 136 | + } |
| 137 | + |
| 138 | + DeferCleanup(func() { |
| 139 | + // Use a fresh context for cleanup to avoid timeout issues |
| 140 | + cleanupCtx := context.Background() |
| 141 | + |
| 142 | + // Delete all CRs (ignore NotFound errors for idempotency) |
| 143 | + if clusterOutput != nil { |
| 144 | + _ = client.IgnoreNotFound(k8sClient.Delete(cleanupCtx, clusterOutput)) |
| 145 | + } |
| 146 | + if fluentdConfig != nil { |
| 147 | + _ = client.IgnoreNotFound(k8sClient.Delete(cleanupCtx, fluentdConfig)) |
| 148 | + } |
| 149 | + if fluentdCR != nil { |
| 150 | + _ = client.IgnoreNotFound(k8sClient.Delete(cleanupCtx, fluentdCR)) |
| 151 | + } |
| 152 | + |
| 153 | + // Wait for StatefulSet to be deleted (find by label, not name) |
| 154 | + if fluentdCR != nil { |
| 155 | + Eventually(func() bool { |
| 156 | + stsList := &appsv1.StatefulSetList{} |
| 157 | + err := k8sClient.List(cleanupCtx, stsList, client.InNamespace(namespace)) |
| 158 | + if err != nil { |
| 159 | + return false |
| 160 | + } |
| 161 | + // StatefulSet should be gone |
| 162 | + return len(stsList.Items) == 0 |
| 163 | + }, time.Minute, time.Second).Should(BeTrue(), "StatefulSet should be deleted") |
| 164 | + } |
| 165 | + |
| 166 | + // Delete namespace and wait for it to be gone |
| 167 | + ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}} |
| 168 | + _ = client.IgnoreNotFound(k8sClient.Delete(cleanupCtx, ns)) |
| 169 | + Eventually(func() bool { |
| 170 | + err := k8sClient.Get(cleanupCtx, types.NamespacedName{Name: namespace}, &corev1.Namespace{}) |
| 171 | + return apierrors.IsNotFound(err) |
| 172 | + }, 2*time.Minute, time.Second).Should(BeTrue(), "Namespace should be deleted") |
| 173 | + }) |
| 174 | + }) |
| 175 | + |
| 176 | + It("Should create a healthy Fluentd StatefulSet", func() { |
| 177 | + By("Creating the ClusterOutput") |
| 178 | + Expect(k8sClient.Create(ctx, clusterOutput)).To(Succeed()) |
| 179 | + |
| 180 | + By("Creating the FluentdConfig") |
| 181 | + Expect(k8sClient.Create(ctx, fluentdConfig)).To(Succeed()) |
| 182 | + |
| 183 | + By("Creating the Fluentd Custom Resource") |
| 184 | + Expect(k8sClient.Create(ctx, fluentdCR)).To(Succeed()) |
| 185 | + |
| 186 | + By("Verifying StatefulSet creation and readiness") |
| 187 | + |
| 188 | + // Find StatefulSet by label instead of assuming name |
| 189 | + Eventually(func() bool { |
| 190 | + stsList := &appsv1.StatefulSetList{} |
| 191 | + err := k8sClient.List(ctx, stsList, |
| 192 | + client.InNamespace(namespace), |
| 193 | + client.MatchingLabels{"app.kubernetes.io/name": fluentdCR.Name}) |
| 194 | + if err != nil || len(stsList.Items) == 0 { |
| 195 | + return false |
| 196 | + } |
| 197 | + return true |
| 198 | + }, time.Minute, time.Second).Should(BeTrue(), "StatefulSet should be created by the Operator") |
| 199 | + |
| 200 | + // Check for Ready Replicas (Real Workload Health) |
| 201 | + Eventually(func() int32 { |
| 202 | + // Refresh StatefulSet status |
| 203 | + stsList := &appsv1.StatefulSetList{} |
| 204 | + _ = k8sClient.List(ctx, stsList, |
| 205 | + client.InNamespace(namespace), |
| 206 | + client.MatchingLabels{"app.kubernetes.io/name": fluentdCR.Name}) |
| 207 | + if len(stsList.Items) == 0 { |
| 208 | + return 0 |
| 209 | + } |
| 210 | + return stsList.Items[0].Status.ReadyReplicas |
| 211 | + }, 5*time.Minute, 2*time.Second).Should(Equal(*fluentdCR.Spec.Replicas), |
| 212 | + "StatefulSet should have expected number of ready replicas") |
| 213 | + |
| 214 | + By("Verifying Pod Status and Container Readiness") |
| 215 | + Eventually(func() bool { |
| 216 | + podList := &corev1.PodList{} |
| 217 | + // List pods owned by the StatefulSet |
| 218 | + err := k8sClient.List(ctx, podList, client.InNamespace(namespace)) |
| 219 | + if err != nil { |
| 220 | + return false |
| 221 | + } |
| 222 | + for _, pod := range podList.Items { |
| 223 | + // Check if pod is owned by a StatefulSet |
| 224 | + for _, owner := range pod.OwnerReferences { |
| 225 | + if owner.Kind == "StatefulSet" { |
| 226 | + // Verify pod is running |
| 227 | + if pod.Status.Phase != corev1.PodRunning { |
| 228 | + continue |
| 229 | + } |
| 230 | + |
| 231 | + // Check ALL containers are ready (not just pod condition) |
| 232 | + allContainersReady := true |
| 233 | + for _, cs := range pod.Status.ContainerStatuses { |
| 234 | + if !cs.Ready { |
| 235 | + allContainersReady = false |
| 236 | + break |
| 237 | + } |
| 238 | + } |
| 239 | + |
| 240 | + // Also check pod ready condition |
| 241 | + podReady := false |
| 242 | + for _, condition := range pod.Status.Conditions { |
| 243 | + if condition.Type == corev1.PodReady && condition.Status == corev1.ConditionTrue { |
| 244 | + podReady = true |
| 245 | + break |
| 246 | + } |
| 247 | + } |
| 248 | + |
| 249 | + if allContainersReady && podReady { |
| 250 | + return true |
| 251 | + } |
| 252 | + } |
| 253 | + } |
| 254 | + } |
| 255 | + return false |
| 256 | + }, 5*time.Minute, 2*time.Second).Should(BeTrue(), |
| 257 | + "At least one Fluentd Pod should be Running with all containers Ready") |
| 258 | + }) |
| 259 | + }) |
| 260 | +}) |
0 commit comments