|
| 1 | +/* |
| 2 | +Copyright 2024 The Kubernetes Authors. |
| 3 | +
|
| 4 | +Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | +you may not use this file except in compliance with the License. |
| 6 | +You may obtain a copy of the License at |
| 7 | +
|
| 8 | + http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | +
|
| 10 | +Unless required by applicable law or agreed to in writing, software |
| 11 | +distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | +See the License for the specific language governing permissions and |
| 14 | +limitations under the License. |
| 15 | +*/ |
| 16 | + |
| 17 | +package csimock |
| 18 | + |
| 19 | +import ( |
| 20 | + "context" |
| 21 | + "strings" |
| 22 | + "time" |
| 23 | + |
| 24 | + "google.golang.org/grpc/codes" |
| 25 | + |
| 26 | + csipbv1 "github.com/container-storage-interface/spec/lib/go/csi" |
| 27 | + v1 "k8s.io/api/core/v1" |
| 28 | + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" |
| 29 | + "k8s.io/apimachinery/pkg/util/wait" |
| 30 | + "k8s.io/kubernetes/pkg/features" |
| 31 | + kubeletmetrics "k8s.io/kubernetes/pkg/kubelet/metrics" |
| 32 | + "k8s.io/kubernetes/test/e2e/feature" |
| 33 | + "k8s.io/kubernetes/test/e2e/framework" |
| 34 | + e2emetrics "k8s.io/kubernetes/test/e2e/framework/metrics" |
| 35 | + e2epod "k8s.io/kubernetes/test/e2e/framework/pod" |
| 36 | + e2epv "k8s.io/kubernetes/test/e2e/framework/pv" |
| 37 | + "k8s.io/kubernetes/test/e2e/storage/drivers" |
| 38 | + "k8s.io/kubernetes/test/e2e/storage/utils" |
| 39 | + admissionapi "k8s.io/pod-security-admission/api" |
| 40 | + |
| 41 | + "github.com/onsi/ginkgo/v2" |
| 42 | +) |
| 43 | + |
| 44 | +var _ = utils.SIGDescribe("CSI Mock Node Volume Health", feature.CSIVolumeHealth, framework.WithFeatureGate(features.CSIVolumeHealth), func() { |
| 45 | + f := framework.NewDefaultFramework("csi-mock-node-volume-health") |
| 46 | + f.NamespacePodSecurityLevel = admissionapi.LevelPrivileged |
| 47 | + m := newMockDriverSetup(f) |
| 48 | + |
| 49 | + f.Context("CSI Mock Node Volume Health", f.WithSlow(), func() { |
| 50 | + trackedCalls := []string{ |
| 51 | + "NodeGetVolumeStats", |
| 52 | + } |
| 53 | + tests := []struct { |
| 54 | + name string |
| 55 | + expectedCalls []csiCall |
| 56 | + nodeVolumeConditionRequired bool |
| 57 | + nodeAbnormalVolumeCondition bool |
| 58 | + }{ |
| 59 | + { |
| 60 | + name: "return normal volume stats", |
| 61 | + expectedCalls: []csiCall{ |
| 62 | + { |
| 63 | + expectedMethod: "NodeGetVolumeStats", |
| 64 | + expectedError: codes.OK, |
| 65 | + }, |
| 66 | + }, |
| 67 | + nodeVolumeConditionRequired: true, |
| 68 | + nodeAbnormalVolumeCondition: false, |
| 69 | + }, |
| 70 | + { |
| 71 | + name: "return normal volume stats without volume condition", |
| 72 | + expectedCalls: []csiCall{ |
| 73 | + { |
| 74 | + expectedMethod: "NodeGetVolumeStats", |
| 75 | + expectedError: codes.OK, |
| 76 | + }, |
| 77 | + }, |
| 78 | + nodeVolumeConditionRequired: false, |
| 79 | + nodeAbnormalVolumeCondition: false, |
| 80 | + }, |
| 81 | + { |
| 82 | + name: "return normal volume stats with abnormal volume condition", |
| 83 | + expectedCalls: []csiCall{ |
| 84 | + { |
| 85 | + expectedMethod: "NodeGetVolumeStats", |
| 86 | + expectedError: codes.OK, |
| 87 | + }, |
| 88 | + }, |
| 89 | + nodeVolumeConditionRequired: true, |
| 90 | + nodeAbnormalVolumeCondition: true, |
| 91 | + }, |
| 92 | + } |
| 93 | + for _, t := range tests { |
| 94 | + test := t |
| 95 | + ginkgo.It(test.name, func(ctx context.Context) { |
| 96 | + m.init(ctx, testParameters{ |
| 97 | + registerDriver: true, |
| 98 | + enableNodeVolumeCondition: test.nodeVolumeConditionRequired, |
| 99 | + hooks: createGetVolumeStatsHook(test.nodeAbnormalVolumeCondition), |
| 100 | + }) |
| 101 | + ginkgo.DeferCleanup(m.cleanup) |
| 102 | + _, claim, pod := m.createPod(ctx, pvcReference) |
| 103 | + if pod == nil { |
| 104 | + return |
| 105 | + } |
| 106 | + // Wait for PVC to get bound to make sure the CSI driver is fully started. |
| 107 | + err := e2epv.WaitForPersistentVolumeClaimPhase(ctx, v1.ClaimBound, f.ClientSet, f.Namespace.Name, claim.Name, time.Second, framework.ClaimProvisionTimeout) |
| 108 | + framework.ExpectNoError(err, "while waiting for PVC to get provisioned") |
| 109 | + |
| 110 | + ginkgo.By("Waiting for pod to be running") |
| 111 | + err = e2epod.WaitForPodNameRunningInNamespace(ctx, m.cs, pod.Name, pod.Namespace) |
| 112 | + framework.ExpectNoError(err, "wait for running pod") |
| 113 | + ginkgo.By("Waiting for all remaining expected CSI calls") |
| 114 | + err = wait.PollUntilContextTimeout(ctx, time.Second, csiNodeVolumeStatWaitPeriod, true, func(c context.Context) (done bool, err error) { |
| 115 | + var index int |
| 116 | + _, index, err = compareCSICalls(ctx, trackedCalls, test.expectedCalls, m.driver.GetCalls) |
| 117 | + if err != nil { |
| 118 | + return true, err |
| 119 | + } |
| 120 | + if index == 0 { |
| 121 | + // No CSI call received yet |
| 122 | + return false, nil |
| 123 | + } |
| 124 | + if len(test.expectedCalls) == index { |
| 125 | + // all calls received |
| 126 | + return true, nil |
| 127 | + } |
| 128 | + return false, nil |
| 129 | + }) |
| 130 | + framework.ExpectNoError(err, "while waiting for all CSI calls") |
| 131 | + // try to use ```csi.NewMetricsCsi(pv.handler).GetMetrics()``` to get metrics from csimock driver but failed. |
| 132 | + // the mocked csidriver register doesn't regist itself to normal csidriver. |
| 133 | + if test.nodeVolumeConditionRequired { |
| 134 | + pod, err := f.ClientSet.CoreV1().Pods(pod.Namespace).Get(ctx, pod.Name, metav1.GetOptions{}) |
| 135 | + framework.ExpectNoError(err, "get running pod") |
| 136 | + grabber, err := e2emetrics.NewMetricsGrabber(ctx, f.ClientSet, nil, f.ClientConfig(), true, false, false, false, false, false) |
| 137 | + framework.ExpectNoError(err, "creating the metrics grabber") |
| 138 | + waitErr := wait.PollUntilContextTimeout(ctx, 30*time.Second, csiNodeVolumeStatWaitPeriod, true, func(ctx context.Context) (bool, error) { |
| 139 | + framework.Logf("Grabbing Kubelet metrics") |
| 140 | + // Grab kubelet metrics from the node the pod was scheduled on |
| 141 | + var err error |
| 142 | + kubeMetrics, err := grabber.GrabFromKubelet(ctx, pod.Spec.NodeName) |
| 143 | + if err != nil { |
| 144 | + framework.Logf("Error fetching kubelet metrics err: %v", err) |
| 145 | + return false, err |
| 146 | + } |
| 147 | + if !findVolumeConditionMetrics(f.Namespace.Name, claim.Name, kubeMetrics, test.nodeAbnormalVolumeCondition) { |
| 148 | + return false, nil |
| 149 | + } |
| 150 | + return true, nil |
| 151 | + }) |
| 152 | + framework.ExpectNoError(waitErr, "call metrics should not have any error") |
| 153 | + } |
| 154 | + }) |
| 155 | + } |
| 156 | + |
| 157 | + }) |
| 158 | +}) |
| 159 | + |
| 160 | +func findVolumeConditionMetrics(pvcNamespace, pvcName string, kubeMetrics e2emetrics.KubeletMetrics, nodeAbnormalVolumeCondition bool) bool { |
| 161 | + |
| 162 | + found := false |
| 163 | + framework.Logf("Looking for sample tagged with namespace `%s`, PVC `%s`", pvcNamespace, pvcName) |
| 164 | + for key, value := range kubeMetrics { |
| 165 | + for _, sample := range value { |
| 166 | + framework.Logf("Found sample %++v with key: %s", sample, key) |
| 167 | + samplePVC, ok := sample.Metric["persistentvolumeclaim"] |
| 168 | + if !ok { |
| 169 | + break |
| 170 | + } |
| 171 | + sampleNS, ok := sample.Metric["namespace"] |
| 172 | + if !ok { |
| 173 | + break |
| 174 | + } |
| 175 | + |
| 176 | + if string(samplePVC) == pvcName && string(sampleNS) == pvcNamespace && strings.Contains(key, kubeletmetrics.VolumeStatsHealthStatusAbnormalKey) { |
| 177 | + if (nodeAbnormalVolumeCondition && sample.Value.String() == "1") || (!nodeAbnormalVolumeCondition && sample.Value.String() == "0") { |
| 178 | + found = true |
| 179 | + break |
| 180 | + } |
| 181 | + } |
| 182 | + } |
| 183 | + } |
| 184 | + return found |
| 185 | +} |
| 186 | + |
| 187 | +func createGetVolumeStatsHook(abnormalVolumeCondition bool) *drivers.Hooks { |
| 188 | + return &drivers.Hooks{ |
| 189 | + Pre: func(ctx context.Context, fullMethod string, request interface{}) (reply interface{}, err error) { |
| 190 | + if req, ok := request.(*csipbv1.NodeGetVolumeStatsRequest); ok { |
| 191 | + if abnormalVolumeCondition { |
| 192 | + req.VolumePath = "/tmp/csi/health/abnormal" |
| 193 | + } |
| 194 | + } |
| 195 | + return nil, nil |
| 196 | + }, |
| 197 | + } |
| 198 | + |
| 199 | +} |
0 commit comments