src/k8splugin/internal/statuscheck/ready.go

   1 /*
   2 Copyright The Helm 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 statuscheck // import "helm.sh/helm/v3/pkg/kube"
  18
  19 import (
  20         "context"
  21         "github.com/onap/multicloud-k8s/src/k8splugin/internal/utils"
  22         appsv1 "k8s.io/api/apps/v1"
  23         appsv1beta1 "k8s.io/api/apps/v1beta1"
  24         appsv1beta2 "k8s.io/api/apps/v1beta2"
  25         batchv1 "k8s.io/api/batch/v1"
  26         corev1 "k8s.io/api/core/v1"
  27         extensionsv1beta1 "k8s.io/api/extensions/v1beta1"
  28         apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
  29         apiextv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
  30         metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  31         "k8s.io/apimachinery/pkg/runtime"
  32         "k8s.io/apimachinery/pkg/util/intstr"
  33         "k8s.io/cli-runtime/pkg/resource"
  34         "k8s.io/client-go/kubernetes"
  35         "k8s.io/client-go/kubernetes/scheme"
  36         "log"
  37 )
  38
  39 // ReadyCheckerOption is a function that configures a ReadyChecker.
  40 type ReadyCheckerOption func(*ReadyChecker)
  41
  42 // PausedAsReady returns a ReadyCheckerOption that configures a ReadyChecker
  43 // to consider paused resources to be ready. For example a Deployment
  44 // with spec.paused equal to true would be considered ready.
  45 func PausedAsReady(pausedAsReady bool) ReadyCheckerOption {
  46         return func(c *ReadyChecker) {
  47                 c.pausedAsReady = pausedAsReady
  48         }
  49 }
  50
  51 // CheckJobs returns a ReadyCheckerOption that configures a ReadyChecker
  52 // to consider readiness of Job resources.
  53 func CheckJobs(checkJobs bool) ReadyCheckerOption {
  54         return func(c *ReadyChecker) {
  55                 c.checkJobs = checkJobs
  56         }
  57 }
  58
  59 // NewReadyChecker creates a new checker. Passed ReadyCheckerOptions can
  60 // be used to override defaults.
  61 func NewReadyChecker(cl kubernetes.Interface, opts ...ReadyCheckerOption) ReadyChecker {
  62         c := ReadyChecker{
  63                 client: cl,
  64         }
  65
  66         for _, opt := range opts {
  67                 opt(&c)
  68         }
  69
  70         return c
  71 }
  72
  73 // ReadyChecker is a type that can check core Kubernetes types for readiness.
  74 type ReadyChecker struct {
  75         client        kubernetes.Interface
  76         checkJobs     bool
  77         pausedAsReady bool
  78 }
  79
  80 // IsReady checks if v is ready. It supports checking readiness for pods,
  81 // deployments, persistent volume claims, services, daemon sets, custom
  82 // resource definitions, stateful sets, replication controllers, and replica
  83 // sets. All other resource kinds are always considered ready.
  84 //
  85 // IsReady will fetch the latest state of the object from the server prior to
  86 // performing readiness checks, and it will return any error encountered.
  87 func (c *ReadyChecker) IsReady(ctx context.Context, v *resource.Info) (bool, error) {
  88         var (
  89                 // This defaults to true, otherwise we get to a point where
  90                 // things will always return false unless one of the objects
  91                 // that manages pods has been hit
  92                 ok  = true
  93                 err error
  94         )
  95         switch value := AsVersioned(v).(type) {
  96         case *corev1.Pod:
  97                 pod, err := c.client.CoreV1().Pods(v.Namespace).Get(ctx, v.Name, metav1.GetOptions{})
  98                 if err != nil || !c.isPodReady(pod) {
  99                         return false, err
 100                 }
 101         case *batchv1.Job:
 102                 if c.checkJobs {
 103                         job, err := c.client.BatchV1().Jobs(v.Namespace).Get(ctx, v.Name, metav1.GetOptions{})
 104                         if err != nil || !c.jobReady(job) {
 105                                 return false, err
 106                         }
 107                 }
 108         case *appsv1.Deployment, *appsv1beta1.Deployment, *appsv1beta2.Deployment, *extensionsv1beta1.Deployment:
 109                 currentDeployment, err := c.client.AppsV1().Deployments(v.Namespace).Get(ctx, v.Name, metav1.GetOptions{})
 110                 if err != nil {
 111                         return false, err
 112                 }
 113                 // If paused deployment will never be ready
 114                 if currentDeployment.Spec.Paused {
 115                         return c.pausedAsReady, nil
 116                 }
 117                 // Find RS associated with deployment
 118                 newReplicaSet, err := utils.GetNewReplicaSet(currentDeployment, c.client.AppsV1())
 119                 if err != nil || newReplicaSet == nil {
 120                         return false, err
 121                 }
 122                 if !c.deploymentReady(newReplicaSet, currentDeployment) {
 123                         return false, nil
 124                 }
 125         case *corev1.PersistentVolumeClaim:
 126                 claim, err := c.client.CoreV1().PersistentVolumeClaims(v.Namespace).Get(ctx, v.Name, metav1.GetOptions{})
 127                 if err != nil {
 128                         return false, err
 129                 }
 130                 if !c.volumeReady(claim) {
 131                         return false, nil
 132                 }
 133         case *corev1.Service:
 134                 svc, err := c.client.CoreV1().Services(v.Namespace).Get(ctx, v.Name, metav1.GetOptions{})
 135                 if err != nil {
 136                         return false, err
 137                 }
 138                 if !c.serviceReady(svc) {
 139                         return false, nil
 140                 }
 141         case *extensionsv1beta1.DaemonSet, *appsv1.DaemonSet, *appsv1beta2.DaemonSet:
 142                 ds, err := c.client.AppsV1().DaemonSets(v.Namespace).Get(ctx, v.Name, metav1.GetOptions{})
 143                 if err != nil {
 144                         return false, err
 145                 }
 146                 if !c.daemonSetReady(ds) {
 147                         return false, nil
 148                 }
 149         case *apiextv1beta1.CustomResourceDefinition:
 150                 if err := v.Get(); err != nil {
 151                         return false, err
 152                 }
 153                 crd := &apiextv1beta1.CustomResourceDefinition{}
 154                 if err := scheme.Scheme.Convert(v.Object, crd, nil); err != nil {
 155                         return false, err
 156                 }
 157                 if !c.crdBetaReady(*crd) {
 158                         return false, nil
 159                 }
 160         case *apiextv1.CustomResourceDefinition:
 161                 if err := v.Get(); err != nil {
 162                         return false, err
 163                 }
 164                 crd := &apiextv1.CustomResourceDefinition{}
 165                 if err := scheme.Scheme.Convert(v.Object, crd, nil); err != nil {
 166                         return false, err
 167                 }
 168                 if !c.crdReady(*crd) {
 169                         return false, nil
 170                 }
 171         case *appsv1.StatefulSet, *appsv1beta1.StatefulSet, *appsv1beta2.StatefulSet:
 172                 sts, err := c.client.AppsV1().StatefulSets(v.Namespace).Get(ctx, v.Name, metav1.GetOptions{})
 173                 if err != nil {
 174                         return false, err
 175                 }
 176                 if !c.statefulSetReady(sts) {
 177                         return false, nil
 178                 }
 179         case *corev1.ReplicationController, *extensionsv1beta1.ReplicaSet, *appsv1beta2.ReplicaSet, *appsv1.ReplicaSet:
 180                 ok, err = c.podsReadyForObject(ctx, v.Namespace, value)
 181         }
 182         if !ok || err != nil {
 183                 return false, err
 184         }
 185         return true, nil
 186 }
 187
 188 func (c *ReadyChecker) podsReadyForObject(ctx context.Context, namespace string, obj runtime.Object) (bool, error) {
 189         pods, err := c.podsforObject(ctx, namespace, obj)
 190         if err != nil {
 191                 return false, err
 192         }
 193         for _, pod := range pods {
 194                 if !c.isPodReady(&pod) {
 195                         return false, nil
 196                 }
 197         }
 198         return true, nil
 199 }
 200
 201 func (c *ReadyChecker) podsforObject(ctx context.Context, namespace string, obj runtime.Object) ([]corev1.Pod, error) {
 202         selector, err := SelectorsForObject(obj)
 203         if err != nil {
 204                 return nil, err
 205         }
 206         list, err := getPods(ctx, c.client, namespace, selector.String())
 207         return list, err
 208 }
 209
 210 // isPodReady returns true if a pod is ready; false otherwise.
 211 func (c *ReadyChecker) isPodReady(pod *corev1.Pod) bool {
 212         for _, c := range pod.Status.Conditions {
 213                 if c.Type == corev1.PodReady && c.Status == corev1.ConditionTrue {
 214                         return true
 215                 }
 216         }
 217         log.Printf("Pod is not ready: %s/%s", pod.GetNamespace(), pod.GetName())
 218         return false
 219 }
 220
 221 func (c *ReadyChecker) jobReady(job *batchv1.Job) bool {
 222         if job.Status.Failed > *job.Spec.BackoffLimit {
 223                 log.Printf("Job is failed: %s/%s", job.GetNamespace(), job.GetName())
 224                 return false
 225         }
 226         if job.Status.Succeeded < *job.Spec.Completions {
 227                 log.Printf("Job is not completed: %s/%s", job.GetNamespace(), job.GetName())
 228                 return false
 229         }
 230         return true
 231 }
 232
 233 func (c *ReadyChecker) serviceReady(s *corev1.Service) bool {
 234         // ExternalName Services are external to cluster so helm shouldn't be checking to see if they're 'ready' (i.e. have an IP Set)
 235         if s.Spec.Type == corev1.ServiceTypeExternalName {
 236                 return true
 237         }
 238
 239         // Ensure that the service cluster IP is not empty
 240         if s.Spec.ClusterIP == "" {
 241                 log.Printf("Service does not have cluster IP address: %s/%s", s.GetNamespace(), s.GetName())
 242                 return false
 243         }
 244
 245         // This checks if the service has a LoadBalancer and that balancer has an Ingress defined
 246         if s.Spec.Type == corev1.ServiceTypeLoadBalancer {
 247                 // do not wait when at least 1 external IP is set
 248                 if len(s.Spec.ExternalIPs) > 0 {
 249                         log.Printf("Service %s/%s has external IP addresses (%v), marking as ready", s.GetNamespace(), s.GetName(), s.Spec.ExternalIPs)
 250                         return true
 251                 }
 252
 253                 if s.Status.LoadBalancer.Ingress == nil {
 254                         log.Printf("Service does not have load balancer ingress IP address: %s/%s", s.GetNamespace(), s.GetName())
 255                         return false
 256                 }
 257         }
 258
 259         return true
 260 }
 261
 262 func (c *ReadyChecker) volumeReady(v *corev1.PersistentVolumeClaim) bool {
 263         if v.Status.Phase != corev1.ClaimBound {
 264                 log.Printf("PersistentVolumeClaim is not bound: %s/%s", v.GetNamespace(), v.GetName())
 265                 return false
 266         }
 267         return true
 268 }
 269
 270 func (c *ReadyChecker) deploymentReady(rs *appsv1.ReplicaSet, dep *appsv1.Deployment) bool {
 271         expectedReady := *dep.Spec.Replicas - utils.MaxUnavailable(*dep)
 272         if !(rs.Status.ReadyReplicas >= expectedReady) {
 273                 log.Printf("Deployment is not ready: %s/%s. %d out of %d expected pods are ready", dep.Namespace, dep.Name, rs.Status.ReadyReplicas, expectedReady)
 274                 return false
 275         }
 276         return true
 277 }
 278
 279 func (c *ReadyChecker) daemonSetReady(ds *appsv1.DaemonSet) bool {
 280         // If the update strategy is not a rolling update, there will be nothing to wait for
 281         if ds.Spec.UpdateStrategy.Type != appsv1.RollingUpdateDaemonSetStrategyType {
 282                 return true
 283         }
 284
 285         // Make sure all the updated pods have been scheduled
 286         if ds.Status.UpdatedNumberScheduled != ds.Status.DesiredNumberScheduled {
 287                 log.Printf("DaemonSet is not ready: %s/%s. %d out of %d expected pods have been scheduled", ds.Namespace, ds.Name, ds.Status.UpdatedNumberScheduled, ds.Status.DesiredNumberScheduled)
 288                 return false
 289         }
 290         maxUnavailable, err := intstr.GetValueFromIntOrPercent(ds.Spec.UpdateStrategy.RollingUpdate.MaxUnavailable, int(ds.Status.DesiredNumberScheduled), true)
 291         if err != nil {
 292                 // If for some reason the value is invalid, set max unavailable to the
 293                 // number of desired replicas. This is the same behavior as the
 294                 // `MaxUnavailable` function in deploymentutil
 295                 maxUnavailable = int(ds.Status.DesiredNumberScheduled)
 296         }
 297
 298         expectedReady := int(ds.Status.DesiredNumberScheduled) - maxUnavailable
 299         if !(int(ds.Status.NumberReady) >= expectedReady) {
 300                 log.Printf("DaemonSet is not ready: %s/%s. %d out of %d expected pods are ready", ds.Namespace, ds.Name, ds.Status.NumberReady, expectedReady)
 301                 return false
 302         }
 303         return true
 304 }
 305
 306 // Because the v1 extensions API is not available on all supported k8s versions
 307 // yet and because Go doesn't support generics, we need to have a duplicate
 308 // function to support the v1beta1 types
 309 func (c *ReadyChecker) crdBetaReady(crd apiextv1beta1.CustomResourceDefinition) bool {
 310         for _, cond := range crd.Status.Conditions {
 311                 switch cond.Type {
 312                 case apiextv1beta1.Established:
 313                         if cond.Status == apiextv1beta1.ConditionTrue {
 314                                 return true
 315                         }
 316                 case apiextv1beta1.NamesAccepted:
 317                         if cond.Status == apiextv1beta1.ConditionFalse {
 318                                 // This indicates a naming conflict, but it's probably not the
 319                                 // job of this function to fail because of that. Instead,
 320                                 // we treat it as a success, since the process should be able to
 321                                 // continue.
 322                                 return true
 323                         }
 324                 }
 325         }
 326         return false
 327 }
 328
 329 func (c *ReadyChecker) crdReady(crd apiextv1.CustomResourceDefinition) bool {
 330         for _, cond := range crd.Status.Conditions {
 331                 switch cond.Type {
 332                 case apiextv1.Established:
 333                         if cond.Status == apiextv1.ConditionTrue {
 334                                 return true
 335                         }
 336                 case apiextv1.NamesAccepted:
 337                         if cond.Status == apiextv1.ConditionFalse {
 338                                 // This indicates a naming conflict, but it's probably not the
 339                                 // job of this function to fail because of that. Instead,
 340                                 // we treat it as a success, since the process should be able to
 341                                 // continue.
 342                                 return true
 343                         }
 344                 }
 345         }
 346         return false
 347 }
 348
 349 func (c *ReadyChecker) statefulSetReady(sts *appsv1.StatefulSet) bool {
 350         // If the update strategy is not a rolling update, there will be nothing to wait for
 351         if sts.Spec.UpdateStrategy.Type != appsv1.RollingUpdateStatefulSetStrategyType {
 352                 return true
 353         }
 354
 355         // Dereference all the pointers because StatefulSets like them
 356         var partition int
 357         // 1 is the default for replicas if not set
 358         var replicas = 1
 359         // For some reason, even if the update strategy is a rolling update, the
 360         // actual rollingUpdate field can be nil. If it is, we can safely assume
 361         // there is no partition value
 362         if sts.Spec.UpdateStrategy.RollingUpdate != nil && sts.Spec.UpdateStrategy.RollingUpdate.Partition != nil {
 363                 partition = int(*sts.Spec.UpdateStrategy.RollingUpdate.Partition)
 364         }
 365         if sts.Spec.Replicas != nil {
 366                 replicas = int(*sts.Spec.Replicas)
 367         }
 368
 369         // Because an update strategy can use partitioning, we need to calculate the
 370         // number of updated replicas we should have. For example, if the replicas
 371         // is set to 3 and the partition is 2, we'd expect only one pod to be
 372         // updated
 373         expectedReplicas := replicas - partition
 374
 375         // Make sure all the updated pods have been scheduled
 376         if int(sts.Status.UpdatedReplicas) != expectedReplicas {
 377                 log.Printf("StatefulSet is not ready: %s/%s. %d out of %d expected pods have been scheduled", sts.Namespace, sts.Name, sts.Status.UpdatedReplicas, expectedReplicas)
 378                 return false
 379         }
 380
 381         if int(sts.Status.ReadyReplicas) != replicas {
 382                 log.Printf("StatefulSet is not ready: %s/%s. %d out of %d expected pods are ready", sts.Namespace, sts.Name, sts.Status.ReadyReplicas, replicas)
 383                 return false
 384         }
 385         return true
 386 }
 387
 388 func getPods(ctx context.Context, client kubernetes.Interface, namespace, selector string) ([]corev1.Pod, error) {
 389         list, err := client.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{
 390                 LabelSelector: selector,
 391         })
 392         return list.Items, err
 393 }