|
| 1 | +// Copyright © 2021 Cisco Systems, Inc. and/or its affiliates |
| 2 | +// |
| 3 | +// Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | +// you may not use this file except in compliance with the License. |
| 5 | +// You may obtain a copy of the License at |
| 6 | +// |
| 7 | +// http://www.apache.org/licenses/LICENSE-2.0 |
| 8 | +// |
| 9 | +// Unless required by applicable law or agreed to in writing, software |
| 10 | +// distributed under the License is distributed on an "AS IS" BASIS, |
| 11 | +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 | +// See the License for the specific language governing permissions and |
| 13 | +// limitations under the License. |
| 14 | + |
| 15 | +package syslong_ng_aggregator |
| 16 | + |
| 17 | +import ( |
| 18 | + "context" |
| 19 | + "fmt" |
| 20 | + "os" |
| 21 | + "os/exec" |
| 22 | + "path/filepath" |
| 23 | + "strings" |
| 24 | + "testing" |
| 25 | + "time" |
| 26 | + |
| 27 | + "github.com/cisco-open/operator-tools/pkg/typeoverride" |
| 28 | + "github.com/cisco-open/operator-tools/pkg/utils" |
| 29 | + "github.com/stretchr/testify/require" |
| 30 | + appsv1 "k8s.io/api/apps/v1" |
| 31 | + batchv1 "k8s.io/api/batch/v1" |
| 32 | + corev1 "k8s.io/api/core/v1" |
| 33 | + rbacv1 "k8s.io/api/rbac/v1" |
| 34 | + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" |
| 35 | + "k8s.io/apimachinery/pkg/api/resource" |
| 36 | + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" |
| 37 | + "k8s.io/apimachinery/pkg/runtime" |
| 38 | + "sigs.k8s.io/controller-runtime/pkg/client" |
| 39 | + "sigs.k8s.io/controller-runtime/pkg/cluster" |
| 40 | + |
| 41 | + "github.com/kube-logging/logging-operator/pkg/resources/syslogng" |
| 42 | + "github.com/kube-logging/logging-operator/pkg/sdk/logging/api/v1beta1" |
| 43 | + "github.com/kube-logging/logging-operator/pkg/sdk/logging/model/syslogng/filter" |
| 44 | + syslogngoutput "github.com/kube-logging/logging-operator/pkg/sdk/logging/model/syslogng/output" |
| 45 | + |
| 46 | + "github.com/kube-logging/logging-operator/e2e/common" |
| 47 | + "github.com/kube-logging/logging-operator/e2e/common/cond" |
| 48 | + "github.com/kube-logging/logging-operator/e2e/common/setup" |
| 49 | +) |
| 50 | + |
| 51 | +var TestTempDir string |
| 52 | + |
| 53 | +func init() { |
| 54 | + var ok bool |
| 55 | + TestTempDir, ok = os.LookupEnv("PROJECT_DIR") |
| 56 | + if !ok { |
| 57 | + TestTempDir = "../.." |
| 58 | + } |
| 59 | + TestTempDir = filepath.Join(TestTempDir, "build/_test") |
| 60 | + err := os.MkdirAll(TestTempDir, os.FileMode(0755)) |
| 61 | + if err != nil { |
| 62 | + panic(err) |
| 63 | + } |
| 64 | +} |
| 65 | + |
| 66 | +func TestSyslogNGIsRunningAndForwardingLogs(t *testing.T) { |
| 67 | + ns := "syslog-ng-1" |
| 68 | + common.WithCluster(t, func(t *testing.T, c common.Cluster) { |
| 69 | + setup.LoggingOperator(t, c, setup.LoggingOperatorOptionFunc(func(options *setup.LoggingOperatorOptions) { |
| 70 | + options.Config.DisableWebhook = true |
| 71 | + options.Config.Namespace = ns |
| 72 | + })) |
| 73 | + |
| 74 | + consumer := setup.LogConsumer(t, c.GetClient(), setup.LogConsumerOptionFunc(func(options *setup.LogConsumerOptions) { |
| 75 | + options.Namespace = ns |
| 76 | + })) |
| 77 | + |
| 78 | + ctx := context.Background() |
| 79 | + |
| 80 | + logging := v1beta1.Logging{ |
| 81 | + ObjectMeta: metav1.ObjectMeta{ |
| 82 | + Name: "syslog-ng-aggregator-test", |
| 83 | + Namespace: ns, |
| 84 | + }, |
| 85 | + Spec: v1beta1.LoggingSpec{ |
| 86 | + EnableRecreateWorkloadOnImmutableFieldChange: true, |
| 87 | + ControlNamespace: ns, |
| 88 | + FluentbitSpec: &v1beta1.FluentbitSpec{ |
| 89 | + Network: &v1beta1.FluentbitNetwork{ |
| 90 | + Keepalive: utils.BoolPointer(false), |
| 91 | + }, |
| 92 | + }, |
| 93 | + SyslogNGSpec: &v1beta1.SyslogNGSpec{ |
| 94 | + StatefulSetOverrides: &typeoverride.StatefulSet{ |
| 95 | + Spec: typeoverride.StatefulSetSpec{ |
| 96 | + Template: typeoverride.PodTemplateSpec{ |
| 97 | + Spec: typeoverride.PodSpec{ |
| 98 | + Containers: []corev1.Container{ |
| 99 | + { |
| 100 | + Name: syslogng.ContainerName, |
| 101 | + Resources: corev1.ResourceRequirements{ |
| 102 | + Limits: corev1.ResourceList{ |
| 103 | + corev1.ResourceCPU: resource.MustParse("50m"), |
| 104 | + corev1.ResourceMemory: resource.MustParse("20M"), |
| 105 | + }, |
| 106 | + Requests: corev1.ResourceList{ |
| 107 | + corev1.ResourceCPU: resource.MustParse("25m"), |
| 108 | + corev1.ResourceMemory: resource.MustParse("10M"), |
| 109 | + }, |
| 110 | + }, |
| 111 | + }, |
| 112 | + }, |
| 113 | + }, |
| 114 | + }, |
| 115 | + }, |
| 116 | + }, |
| 117 | + }, |
| 118 | + }, |
| 119 | + } |
| 120 | + require.NoError(t, c.GetClient().Create(ctx, &logging)) |
| 121 | + output := v1beta1.SyslogNGOutput{ |
| 122 | + ObjectMeta: metav1.ObjectMeta{ |
| 123 | + Name: "test-output", |
| 124 | + Namespace: ns, |
| 125 | + }, |
| 126 | + Spec: v1beta1.SyslogNGOutputSpec{ |
| 127 | + HTTP: &syslogngoutput.HTTPOutput{ |
| 128 | + URL: consumer.InputURL(), |
| 129 | + DiskBuffer: &syslogngoutput.DiskBuffer{ |
| 130 | + DiskBufSize: 100 * 1024 * 1024, |
| 131 | + Reliable: true, |
| 132 | + Dir: syslogng.BufferPath, |
| 133 | + }, |
| 134 | + }, |
| 135 | + }, |
| 136 | + } |
| 137 | + require.NoError(t, c.GetClient().Create(ctx, &output)) |
| 138 | + flow := v1beta1.SyslogNGFlow{ |
| 139 | + ObjectMeta: metav1.ObjectMeta{ |
| 140 | + Name: "test-flow", |
| 141 | + Namespace: ns, |
| 142 | + }, |
| 143 | + Spec: v1beta1.SyslogNGFlowSpec{ |
| 144 | + Match: &v1beta1.SyslogNGMatch{ |
| 145 | + Regexp: &filter.RegexpMatchExpr{ |
| 146 | + Pattern: "log-producer", |
| 147 | + Value: "json.kubernetes.labels.my-unique-label", |
| 148 | + Type: "string", |
| 149 | + }, |
| 150 | + }, |
| 151 | + LocalOutputRefs: []string{output.Name}, |
| 152 | + }, |
| 153 | + } |
| 154 | + require.NoError(t, c.GetClient().Create(ctx, &flow)) |
| 155 | + |
| 156 | + meta := logging.SyslogNGObjectMeta(syslogng.StatefulSetName, syslogng.ComponentSyslogNG) |
| 157 | + aggergatorPodName := meta.Name + "-0" |
| 158 | + require.Eventually(t, cond.PodShouldBeRunning(t, c.GetClient(), client.ObjectKey{Namespace: ns, Name: aggergatorPodName}), 5*time.Minute, 5*time.Second) |
| 159 | + |
| 160 | + setup.LogProducer(t, c.GetClient(), setup.LogProducerOptionFunc(func(options *setup.LogProducerOptions) { |
| 161 | + options.Namespace = ns |
| 162 | + options.Labels = map[string]string{ |
| 163 | + "my-unique-label": "log-producer", |
| 164 | + } |
| 165 | + })) |
| 166 | + |
| 167 | + require.Eventually(t, func() bool { |
| 168 | + rawOut, err := exec.Command("kubectl", "-n", consumer.PodKey.Namespace, "logs", consumer.PodKey.Name).Output() |
| 169 | + if err != nil { |
| 170 | + t.Logf("failed to get log consumer logs: %v", err) |
| 171 | + return false |
| 172 | + } |
| 173 | + t.Logf("log consumer logs: %s", rawOut) |
| 174 | + return strings.Contains(string(rawOut), "got request") |
| 175 | + }, 5*time.Minute, 2*time.Second) |
| 176 | + |
| 177 | + require.NoError(t, exec.Command("kubectl", "-n", consumer.PodKey.Namespace, "exec", consumer.PodKey.Name, "--", "curl", "-sS", "http://localhost:8082/off").Run()) |
| 178 | + |
| 179 | + require.Eventually(t, cond.PodShouldBeRunning(t, c.GetClient(), client.ObjectKey{Namespace: ns, Name: aggergatorPodName}), 30*time.Second, time.Second/2) |
| 180 | + |
| 181 | + require.NoError(t, exec.Command("kubectl", "-n", consumer.PodKey.Namespace, "exec", consumer.PodKey.Name, "--", "curl", "-sS", "http://localhost:8082/on").Run()) |
| 182 | + |
| 183 | + }, func(t *testing.T, c common.Cluster) error { |
| 184 | + path := filepath.Join(TestTempDir, fmt.Sprintf("cluster-%s.log", t.Name())) |
| 185 | + t.Logf("Printing cluster logs to %s", path) |
| 186 | + return c.PrintLogs(common.PrintLogConfig{ |
| 187 | + Namespaces: []string{ns, "default"}, |
| 188 | + FilePath: path, |
| 189 | + Limit: 100 * 1000, |
| 190 | + }) |
| 191 | + }, func(o *cluster.Options) { |
| 192 | + if o.Scheme == nil { |
| 193 | + o.Scheme = runtime.NewScheme() |
| 194 | + } |
| 195 | + require.NoError(t, v1beta1.AddToScheme(o.Scheme)) |
| 196 | + require.NoError(t, apiextensionsv1.AddToScheme(o.Scheme)) |
| 197 | + require.NoError(t, appsv1.AddToScheme(o.Scheme)) |
| 198 | + require.NoError(t, batchv1.AddToScheme(o.Scheme)) |
| 199 | + require.NoError(t, corev1.AddToScheme(o.Scheme)) |
| 200 | + require.NoError(t, rbacv1.AddToScheme(o.Scheme)) |
| 201 | + }) |
| 202 | +} |
0 commit comments