gerrit.onap Code Review - so/adapters/so-cnf-adapter.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2023 Nordix Foundation.
   4  * ================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  * SPDX-License-Identifier: Apache-2.0
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.so.cnfm.lcm.bpmn.flows.extclients.kubernetes;
  22
  23 import static org.onap.so.cnfm.lcm.bpmn.flows.Constants.KIND_DAEMON_SET;
  24 import static org.onap.so.cnfm.lcm.bpmn.flows.Constants.KIND_DEPLOYMENT;
  25 import static org.onap.so.cnfm.lcm.bpmn.flows.Constants.KIND_JOB;
  26 import static org.onap.so.cnfm.lcm.bpmn.flows.Constants.KIND_POD;
  27 import static org.onap.so.cnfm.lcm.bpmn.flows.Constants.KIND_REPLICA_SET;
  28 import static org.onap.so.cnfm.lcm.bpmn.flows.Constants.KIND_SERVICE;
  29 import static org.onap.so.cnfm.lcm.bpmn.flows.Constants.KIND_STATEFUL_SET;
  30 import java.io.IOException;
  31 import java.lang.reflect.Type;
  32 import java.net.SocketTimeoutException;
  33 import java.util.ArrayList;
  34 import java.util.Collections;
  35 import java.util.HashMap;
  36 import java.util.List;
  37 import java.util.Map;
  38 import java.util.Map.Entry;
  39 import java.util.Optional;
  40 import java.util.function.Predicate;
  41 import java.util.stream.Collectors;
  42 import org.onap.so.cnfm.lcm.bpmn.flows.exceptions.KubernetesRequestProcessingException;
  43 import org.onap.so.cnfm.lcm.bpmn.flows.exceptions.KubernetesRequestTimeOut;
  44 import org.slf4j.Logger;
  45 import org.slf4j.LoggerFactory;
  46 import org.springframework.beans.factory.annotation.Value;
  47 import org.springframework.stereotype.Service;
  48 import com.google.gson.reflect.TypeToken;
  49 import io.kubernetes.client.apimachinery.GroupVersion;
  50 import io.kubernetes.client.common.KubernetesListObject;
  51 import io.kubernetes.client.common.KubernetesObject;
  52 import io.kubernetes.client.custom.IntOrString;
  53 import io.kubernetes.client.openapi.ApiClient;
  54 import io.kubernetes.client.openapi.ApiException;
  55 import io.kubernetes.client.openapi.apis.AppsV1Api;
  56 import io.kubernetes.client.openapi.apis.BatchV1Api;
  57 import io.kubernetes.client.openapi.apis.CoreV1Api;
  58 import io.kubernetes.client.openapi.models.V1DaemonSet;
  59 import io.kubernetes.client.openapi.models.V1DaemonSetList;
  60 import io.kubernetes.client.openapi.models.V1DaemonSetSpec;
  61 import io.kubernetes.client.openapi.models.V1DaemonSetStatus;
  62 import io.kubernetes.client.openapi.models.V1Deployment;
  63 import io.kubernetes.client.openapi.models.V1DeploymentList;
  64 import io.kubernetes.client.openapi.models.V1DeploymentSpec;
  65 import io.kubernetes.client.openapi.models.V1DeploymentStatus;
  66 import io.kubernetes.client.openapi.models.V1Job;
  67 import io.kubernetes.client.openapi.models.V1JobCondition;
  68 import io.kubernetes.client.openapi.models.V1JobList;
  69 import io.kubernetes.client.openapi.models.V1ObjectMeta;
  70 import io.kubernetes.client.openapi.models.V1Pod;
  71 import io.kubernetes.client.openapi.models.V1PodCondition;
  72 import io.kubernetes.client.openapi.models.V1PodList;
  73 import io.kubernetes.client.openapi.models.V1ReplicaSet;
  74 import io.kubernetes.client.openapi.models.V1ReplicaSetList;
  75 import io.kubernetes.client.openapi.models.V1ReplicaSetSpec;
  76 import io.kubernetes.client.openapi.models.V1ReplicaSetStatus;
  77 import io.kubernetes.client.openapi.models.V1RollingUpdateStatefulSetStrategy;
  78 import io.kubernetes.client.openapi.models.V1Service;
  79 import io.kubernetes.client.openapi.models.V1ServiceList;
  80 import io.kubernetes.client.openapi.models.V1StatefulSet;
  81 import io.kubernetes.client.openapi.models.V1StatefulSetList;
  82 import io.kubernetes.client.openapi.models.V1StatefulSetSpec;
  83 import io.kubernetes.client.openapi.models.V1StatefulSetStatus;
  84 import io.kubernetes.client.openapi.models.V1StatefulSetUpdateStrategy;
  85 import io.kubernetes.client.util.Watch;
  86 import io.kubernetes.client.util.Watch.Response;
  87 import okhttp3.Call;
  88
  89 /**
  90  *
  91  * @author Waqas Ikram (waqas.ikram@est.tech)
  92  *
  93  */
  94 @Service
  95 public class KubernetesClientImpl implements KubernetesClient {
  96     private static final String ROLLING_UPDATE = "RollingUpdate";
  97     private static final String EVENT_TYPE_ERROR = "ERROR";
  98     private static final String EVENT_TYPE_DELETED = "DELETED";
  99     private static final String EVENT_TYPE_MODIFIED = "MODIFIED";
 100     private static final String EVENT_TYPE_ADDED = "ADDED";
 101     private static final String TRUE_STRING = Boolean.TRUE.toString();
 102     private static final String JOB_FAILED = "Failed";
 103     private static final String JOB_COMPLETE = "Complete";
 104     private static final boolean DISABLE_WATCH = false;
 105     private static final boolean ENABLE_WATCH = true;
 106     private static final String POD_READY = "Ready";
 107
 108     private static final Logger logger = LoggerFactory.getLogger(KubernetesClientImpl.class);
 109
 110     /**
 111      * Event Listener timeout in seconds Note: this should be less then the timeout camunda task execution
 112      */
 113     @Value("${kubernetes.client.http-request.timeoutSeconds:5}")
 114     private Integer timeoutSeconds;
 115
 116     @Override
 117     public boolean isJobReady(final ApiClient apiClient, final String labelSelector)
 118             throws KubernetesRequestProcessingException {
 119         logger.debug("Will check if Job is ready using labelSelector: {}", labelSelector);
 120         try {
 121             final BatchV1Api batchV1Api = new BatchV1Api(apiClient);
 122             final Call call = batchV1Api.listJobForAllNamespacesCall(null, null, null, labelSelector, null, null, null,
 123                     null, timeoutSeconds, ENABLE_WATCH, null);
 124
 125             final Map<V1Job, String> readyResources =
 126                     getReadyResources(apiClient, call, new TypeToken<Response<V1Job>>() {}.getType());
 127
 128             if (!readyResources.isEmpty()) {
 129                 final List<Entry<V1Job, String>> jobNotReadyList = readyResources.entrySet().stream()
 130                         .filter(entry -> !isResourceReady(entry.getKey(), entry.getValue(), this::isJobReady))
 131                         .collect(Collectors.toList());
 132
 133                 if (jobNotReadyList.isEmpty()) {
 134                     logger.debug("JobList is ready ...");
 135                     return true;
 136                 }
 137                 logger.debug("JobList is not yet Ready: {}", jobNotReadyList);
 138                 return false;
 139
 140             }
 141
 142             logger.warn("No items found in jobList : {}", readyResources);
 143             return false;
 144
 145         } catch (final ApiException exception) {
 146             handleApiException(KIND_JOB, labelSelector, exception);
 147         } catch (final RuntimeException runtimeException) {
 148             handleRuntimeException(KIND_JOB, labelSelector, runtimeException);
 149         }
 150         logger.debug("Returning false as Job is not ready ...");
 151         return false;
 152     }
 153
 154     @Override
 155     public boolean isPodReady(final ApiClient apiClient, final String labelSelector)
 156             throws KubernetesRequestProcessingException {
 157         logger.debug("Will check if Pod is ready using labelSelector: {}", labelSelector);
 158         try {
 159             final CoreV1Api coreV1Api = new CoreV1Api(apiClient);
 160             final Call call = coreV1Api.listPodForAllNamespacesCall(null, null, null, labelSelector, null, null, null,
 161                     null, timeoutSeconds, ENABLE_WATCH, null);
 162
 163             final Map<V1Pod, String> readyResources =
 164                     getReadyResources(apiClient, call, new TypeToken<Response<V1Pod>>() {}.getType());
 165
 166             if (!readyResources.isEmpty()) {
 167                 final List<Entry<V1Pod, String>> podNotReadyList = readyResources.entrySet().stream()
 168                         .filter(entry -> !isResourceReady(entry.getKey(), entry.getValue(), this::isPodReady))
 169                         .collect(Collectors.toList());
 170
 171                 if (podNotReadyList.isEmpty()) {
 172                     logger.debug("PodList is ready ...");
 173                     return true;
 174                 }
 175                 logger.debug("PodList is not yet Ready: {}", podNotReadyList);
 176                 return false;
 177
 178             }
 179
 180             logger.warn("No items found in podList : {}", readyResources);
 181             return false;
 182
 183         } catch (final ApiException exception) {
 184             handleApiException(KIND_POD, labelSelector, exception);
 185         } catch (final RuntimeException runtimeException) {
 186             handleRuntimeException(KIND_POD, labelSelector, runtimeException);
 187         }
 188
 189         logger.debug("Returning false as Pod is not ready ...");
 190
 191         return false;
 192     }
 193
 194     @Override
 195     public boolean isServiceReady(final ApiClient apiClient, final String labelSelector)
 196             throws KubernetesRequestProcessingException {
 197         logger.debug("Will check if Service is ready using labelSelector: {}", labelSelector);
 198         try {
 199             final CoreV1Api api = new CoreV1Api(apiClient);
 200             final Call call = api.listServiceForAllNamespacesCall(null, null, null, labelSelector, null, null, null,
 201                     null, timeoutSeconds, ENABLE_WATCH, null);
 202
 203             final Map<V1Service, String> readyResources =
 204                     getReadyResources(apiClient, call, new TypeToken<Response<V1Service>>() {}.getType());
 205
 206             if (!readyResources.isEmpty()) {
 207                 final List<Entry<V1Service, String>> serviceNotReadyList = readyResources.entrySet().stream()
 208                         .filter(entry -> !isResourceReady(entry.getKey(), entry.getValue(), this::isServiceReady))
 209                         .collect(Collectors.toList());
 210
 211                 if (serviceNotReadyList.isEmpty()) {
 212                     logger.debug("ServiceList is ready ...");
 213                     return true;
 214                 }
 215                 logger.debug("ServiceList is not yet Ready: {}", serviceNotReadyList);
 216                 return false;
 217
 218             }
 219
 220             logger.warn("No items found in serviceList : {}", readyResources);
 221             return false;
 222
 223         } catch (final ApiException exception) {
 224             handleApiException(KIND_SERVICE, labelSelector, exception);
 225         } catch (final RuntimeException runtimeException) {
 226             handleRuntimeException(KIND_SERVICE, labelSelector, runtimeException);
 227         }
 228
 229         logger.debug("Returning false as Service is not ready ...");
 230         return false;
 231     }
 232
 233     @Override
 234     public boolean isDeploymentReady(final ApiClient apiClient, final String labelSelector)
 235             throws KubernetesRequestProcessingException {
 236         logger.debug("Will check if Deployment is ready using labelSelector: {}", labelSelector);
 237
 238         try {
 239             final AppsV1Api appsV1Api = new AppsV1Api(apiClient);
 240             final Call call = appsV1Api.listDeploymentForAllNamespacesCall(null, null, null, labelSelector, null, null,
 241                     null, null, timeoutSeconds, ENABLE_WATCH, null);
 242
 243             final Map<V1Deployment, String> readyResources =
 244                     getReadyResources(apiClient, call, new TypeToken<Response<V1Deployment>>() {}.getType());
 245
 246             if (!readyResources.isEmpty()) {
 247                 final List<Entry<V1Deployment, String>> deploymentNotReadyList = readyResources.entrySet().stream()
 248                         .filter(entry -> !isResourceReady(entry.getKey(), entry.getValue(), this::isDeploymentReady))
 249                         .collect(Collectors.toList());
 250
 251                 if (deploymentNotReadyList.isEmpty()) {
 252                     logger.debug("DeploymentList is ready ...");
 253                     return true;
 254                 }
 255                 logger.debug("DeploymentList is not yet Ready: {}", deploymentNotReadyList);
 256                 return false;
 257
 258             }
 259
 260             logger.warn("No items found in deploymentList : {}", readyResources);
 261             return false;
 262
 263         } catch (final ApiException exception) {
 264             handleApiException(KIND_DEPLOYMENT, labelSelector, exception);
 265         } catch (final RuntimeException runtimeException) {
 266             handleRuntimeException(KIND_DEPLOYMENT, labelSelector, runtimeException);
 267         }
 268
 269         logger.debug("Returning false as Deployment is not ready ...");
 270         return false;
 271     }
 272
 273     @Override
 274     public boolean isReplicaSetReady(final ApiClient apiClient, final String labelSelector)
 275             throws KubernetesRequestProcessingException {
 276         logger.debug("Will check if ReplicaSet is ready using labelSelector: {}", labelSelector);
 277         try {
 278             final AppsV1Api appsV1Api = new AppsV1Api(apiClient);
 279             final Call call = appsV1Api.listReplicaSetForAllNamespacesCall(null, null, null, labelSelector, null, null,
 280                     null, null, timeoutSeconds, ENABLE_WATCH, null);
 281
 282             final Map<V1ReplicaSet, String> readyResources =
 283                     getReadyResources(apiClient, call, new TypeToken<Response<V1ReplicaSet>>() {}.getType());
 284
 285             if (!readyResources.isEmpty()) {
 286                 final List<Entry<V1ReplicaSet, String>> replicaSet = readyResources.entrySet().stream()
 287                         .filter(entry -> !isResourceReady(entry.getKey(), entry.getValue(), this::isReplicaSetReady))
 288                         .collect(Collectors.toList());
 289
 290                 if (replicaSet.isEmpty()) {
 291                     logger.debug("ReplicaSetList is ready ...");
 292                     return true;
 293                 }
 294                 logger.debug("ReplicaSetList is not yet Ready: {}", replicaSet);
 295                 return false;
 296
 297             }
 298
 299             logger.warn("No items found in replicaSetList : {}", readyResources);
 300             return false;
 301
 302         } catch (final ApiException exception) {
 303             handleApiException(KIND_REPLICA_SET, labelSelector, exception);
 304         } catch (final RuntimeException runtimeException) {
 305             handleRuntimeException(KIND_REPLICA_SET, labelSelector, runtimeException);
 306         }
 307         logger.debug("Returning false as ReplicaSet is not ready ...");
 308         return false;
 309     }
 310
 311     @Override
 312     public boolean isDaemonSetReady(final ApiClient apiClient, final String labelSelector)
 313             throws KubernetesRequestProcessingException {
 314         logger.debug("Will check if DaemonSet is ready using labelSelector: {}", labelSelector);
 315         try {
 316             final AppsV1Api appsV1Api = new AppsV1Api(apiClient);
 317             final Call call = appsV1Api.listDaemonSetForAllNamespacesCall(null, null, null, labelSelector, null, null,
 318                     null, null, timeoutSeconds, ENABLE_WATCH, null);
 319
 320             final Map<V1DaemonSet, String> readyResources =
 321                     getReadyResources(apiClient, call, new TypeToken<Response<V1DaemonSet>>() {}.getType());
 322
 323             if (!readyResources.isEmpty()) {
 324                 final List<Entry<V1DaemonSet, String>> daemonSetNotReadyList = readyResources.entrySet().stream()
 325                         .filter(entry -> !isResourceReady(entry.getKey(), entry.getValue(), this::isDaemonSetReady))
 326                         .collect(Collectors.toList());
 327
 328                 if (daemonSetNotReadyList.isEmpty()) {
 329                     logger.debug("DaemonSetList is ready ...");
 330                     return true;
 331                 }
 332                 logger.debug("DaemonSetList is not yet Ready: {}", daemonSetNotReadyList);
 333                 return false;
 334
 335             }
 336
 337             logger.warn("No items found in daemonSetList : {}", readyResources);
 338             return false;
 339
 340         } catch (final ApiException exception) {
 341             handleApiException(KIND_DAEMON_SET, labelSelector, exception);
 342         } catch (final RuntimeException runtimeException) {
 343             handleRuntimeException(KIND_DAEMON_SET, labelSelector, runtimeException);
 344         }
 345         logger.debug("Returning false as DaemonSet is not ready ...");
 346         return false;
 347     }
 348
 349     @Override
 350     public boolean isStatefulSetReady(final ApiClient apiClient, final String labelSelector)
 351             throws KubernetesRequestProcessingException {
 352         logger.debug("Will check if StatefulSet is ready using labelSelector: {}", labelSelector);
 353         try {
 354             final AppsV1Api appsV1Api = new AppsV1Api(apiClient);
 355             final Call call = appsV1Api.listStatefulSetForAllNamespacesCall(null, null, null, labelSelector, null, null,
 356                     null, null, timeoutSeconds, ENABLE_WATCH, null);
 357
 358             final Map<V1StatefulSet, String> readyResources =
 359                     getReadyResources(apiClient, call, new TypeToken<Response<V1StatefulSet>>() {}.getType());
 360
 361             if (!readyResources.isEmpty()) {
 362                 final List<Entry<V1StatefulSet, String>> statefulSetNotReadyList = readyResources.entrySet().stream()
 363                         .filter(entry -> !isResourceReady(entry.getKey(), entry.getValue(), this::isStatefulSetReady))
 364                         .collect(Collectors.toList());
 365
 366                 if (statefulSetNotReadyList.isEmpty()) {
 367                     logger.debug("StatefulSetList is ready ...");
 368                     return true;
 369                 }
 370                 logger.debug("StatefulSetList is not yet Ready: {}", statefulSetNotReadyList);
 371                 return false;
 372
 373             }
 374
 375             logger.warn("No items found in statefulSetList : {}", readyResources);
 376             return false;
 377
 378         } catch (final ApiException exception) {
 379             handleApiException(KIND_STATEFUL_SET, labelSelector, exception);
 380         } catch (final RuntimeException runtimeException) {
 381             handleRuntimeException(KIND_STATEFUL_SET, labelSelector, runtimeException);
 382         }
 383         logger.debug("Returning false as StatefulSet is not ready ...");
 384         return false;
 385     }
 386
 387     @Override
 388     public boolean isServiceDeleted(final ApiClient apiClient, final String labelSelector)
 389             throws KubernetesRequestProcessingException {
 390         logger.debug("Check is Service deleted by using labelSelector: {}", labelSelector);
 391         try {
 392             final CoreV1Api coreV1Api = new CoreV1Api(apiClient);
 393             final V1ServiceList v1ServiceList = coreV1Api.listServiceForAllNamespaces(null, null, null, labelSelector,
 394                     null, null, null, null, timeoutSeconds, DISABLE_WATCH);
 395             logger.debug("Response from list service for all Namespaces: {}", v1ServiceList);
 396             return v1ServiceList.getItems().isEmpty();
 397         } catch (final ApiException exception) {
 398             handleApiException(KIND_SERVICE, labelSelector, exception);
 399         } catch (final RuntimeException runtimeException) {
 400             handleRuntimeException(KIND_SERVICE, labelSelector, runtimeException);
 401         }
 402         logger.debug("Returning false as Service is not Deleted ...");
 403         return false;
 404     }
 405
 406     @Override
 407     public boolean isPodDeleted(final ApiClient apiClient, final String labelSelector)
 408             throws KubernetesRequestProcessingException {
 409         logger.debug("Check is Pod deleted by using labelSelector: {}", labelSelector);
 410         try {
 411             final CoreV1Api coreV1Api = new CoreV1Api(apiClient);
 412             final V1PodList v1PodList = coreV1Api.listPodForAllNamespaces(null, null, null, labelSelector, null, null,
 413                     null, null, timeoutSeconds, DISABLE_WATCH);
 414             logger.debug("Response from list Pod for all Namespaces: {}", v1PodList);
 415             return v1PodList.getItems().isEmpty();
 416         } catch (final ApiException exception) {
 417             handleApiException(KIND_POD, labelSelector, exception);
 418         } catch (final RuntimeException runtimeException) {
 419             handleRuntimeException(KIND_POD, labelSelector, runtimeException);
 420         }
 421         logger.debug("Returning false as Pod is not Deleted ...");
 422         return false;
 423     }
 424
 425     @Override
 426     public boolean isJobDeleted(final ApiClient apiClient, final String labelSelector)
 427             throws KubernetesRequestProcessingException {
 428         logger.debug("Check is Job deleted by using labelSelector: {}", labelSelector);
 429         try {
 430             final BatchV1Api batchV1Api = new BatchV1Api(apiClient);
 431             final V1JobList v1JobList = batchV1Api.listJobForAllNamespaces(null, null, null, labelSelector, null, null,
 432                     null, null, timeoutSeconds, DISABLE_WATCH);
 433             logger.debug("Response from list Job for all Namespaces: {}", v1JobList);
 434             return v1JobList.getItems().isEmpty();
 435         } catch (final ApiException exception) {
 436             handleApiException(KIND_JOB, labelSelector, exception);
 437         } catch (final RuntimeException runtimeException) {
 438             handleRuntimeException(KIND_JOB, labelSelector, runtimeException);
 439         }
 440         logger.debug("Returning false as Job is not Deleted ...");
 441         return false;
 442     }
 443
 444     @Override
 445     public boolean isDeploymentDeleted(final ApiClient apiClient, final String labelSelector)
 446             throws KubernetesRequestProcessingException {
 447         logger.debug("Check is Deployment deleted by using labelSelector: {}", labelSelector);
 448         try {
 449             final AppsV1Api batchV1Api = new AppsV1Api(apiClient);
 450             final V1DeploymentList v1DeploymentList = batchV1Api.listDeploymentForAllNamespaces(null, null, null,
 451                     labelSelector, null, null, null, null, timeoutSeconds, DISABLE_WATCH);
 452             logger.debug("Response from list Deployment for all Namespaces: {}", v1DeploymentList);
 453             return v1DeploymentList.getItems().isEmpty();
 454         } catch (final ApiException exception) {
 455             handleApiException(KIND_DEPLOYMENT, labelSelector, exception);
 456         } catch (final RuntimeException runtimeException) {
 457             handleRuntimeException(KIND_DEPLOYMENT, labelSelector, runtimeException);
 458         }
 459         logger.debug("Returning false as Deployment is not Deleted ...");
 460         return false;
 461     }
 462
 463     @Override
 464     public boolean isReplicaSetDeleted(final ApiClient apiClient, final String labelSelector)
 465             throws KubernetesRequestProcessingException {
 466         logger.debug("Check is ReplicaSet deleted by using labelSelector: {}", labelSelector);
 467         try {
 468             final AppsV1Api batchV1Api = new AppsV1Api(apiClient);
 469             final V1ReplicaSetList v1ReplicaSetList = batchV1Api.listReplicaSetForAllNamespaces(null, null, null,
 470                     labelSelector, null, null, null, null, timeoutSeconds, DISABLE_WATCH);
 471             logger.debug("Response from list ReplicaSet for all Namespaces: {}", v1ReplicaSetList);
 472             return v1ReplicaSetList.getItems().isEmpty();
 473         } catch (final ApiException exception) {
 474             handleApiException(KIND_REPLICA_SET, labelSelector, exception);
 475         } catch (final RuntimeException runtimeException) {
 476             handleRuntimeException(KIND_REPLICA_SET, labelSelector, runtimeException);
 477         }
 478         logger.debug("Returning false as ReplicaSet is not Deleted ...");
 479         return false;
 480     }
 481
 482     @Override
 483     public boolean isDaemonSetDeleted(final ApiClient apiClient, final String labelSelector)
 484             throws KubernetesRequestProcessingException {
 485         logger.debug("Check is DaemonSet deleted by using labelSelector: {}", labelSelector);
 486         try {
 487             final AppsV1Api batchV1Api = new AppsV1Api(apiClient);
 488             final V1DaemonSetList v1DaemonSetList = batchV1Api.listDaemonSetForAllNamespaces(null, null, null,
 489                     labelSelector, null, null, null, null, timeoutSeconds, DISABLE_WATCH);
 490             logger.debug("Response from list DaemonSet for all Namespaces: {}", v1DaemonSetList);
 491             return v1DaemonSetList.getItems().isEmpty();
 492         } catch (final ApiException exception) {
 493             handleApiException(KIND_DAEMON_SET, labelSelector, exception);
 494         } catch (final RuntimeException runtimeException) {
 495             handleRuntimeException(KIND_DAEMON_SET, labelSelector, runtimeException);
 496         }
 497         logger.debug("Returning false as DaemonSet is not Deleted ...");
 498         return false;
 499     }
 500
 501     @Override
 502     public boolean isStatefulSetDeleted(final ApiClient apiClient, final String labelSelector)
 503             throws KubernetesRequestProcessingException {
 504         logger.debug("Check is StatefulSet deleted by using labelSelector: {}", labelSelector);
 505         try {
 506             final AppsV1Api batchV1Api = new AppsV1Api(apiClient);
 507             final V1StatefulSetList v1StatefulSetList = batchV1Api.listStatefulSetForAllNamespaces(null, null, null,
 508                     labelSelector, null, null, null, null, timeoutSeconds, DISABLE_WATCH);
 509             logger.debug("Response from list StatefulSet for all Namespaces: {}", v1StatefulSetList);
 510             return v1StatefulSetList.getItems().isEmpty();
 511         } catch (final ApiException exception) {
 512             handleApiException(KIND_STATEFUL_SET, labelSelector, exception);
 513         } catch (final RuntimeException runtimeException) {
 514             handleRuntimeException(KIND_STATEFUL_SET, labelSelector, runtimeException);
 515         }
 516         logger.debug("Returning false as StatefulSet is not Deleted ...");
 517         return false;
 518     }
 519
 520
 521     @Override
 522     public List<KubernetesResource> getJobResources(final ApiClient apiClient, final String labelSelector)
 523             throws KubernetesRequestProcessingException {
 524         logger.debug("Retrieving Jobs using labelSelector: {}", labelSelector);
 525         try {
 526             final BatchV1Api batchV1Api = new BatchV1Api(apiClient);
 527             final V1JobList jobList = batchV1Api.listJobForAllNamespaces(null, null, null, labelSelector, null, null,
 528                     null, null, timeoutSeconds, DISABLE_WATCH);
 529
 530             logger.debug("Received Jobs: {}", jobList);
 531             return getKubernetesResource(jobList);
 532
 533         } catch (final ApiException exception) {
 534             handleApiException(KIND_JOB, labelSelector, exception);
 535         } catch (final IllegalArgumentException illegalArgumentException) {
 536             handleIllegalArgumentException(KIND_JOB, labelSelector, illegalArgumentException);
 537         } catch (final RuntimeException runtimeException) {
 538             handleRuntimeException(KIND_JOB, labelSelector, runtimeException);
 539         }
 540
 541         logger.error("Unable to find any job resources ...");
 542         return Collections.emptyList();
 543     }
 544
 545     @Override
 546     public List<KubernetesResource> getDeploymentResources(final ApiClient apiClient, final String labelSelector)
 547             throws KubernetesRequestProcessingException {
 548         logger.debug("Retrieving Deployment using labelSelector: {}", labelSelector);
 549         try {
 550             final AppsV1Api appsV1Api = new AppsV1Api(apiClient);
 551             final V1DeploymentList deploymentList = appsV1Api.listDeploymentForAllNamespaces(null, null, null,
 552                     labelSelector, null, null, null, null, timeoutSeconds, DISABLE_WATCH);
 553
 554             logger.debug("Received Deployments: {}", deploymentList);
 555             return getKubernetesResource(deploymentList);
 556
 557         } catch (final ApiException exception) {
 558             handleApiException(KIND_DEPLOYMENT, labelSelector, exception);
 559         } catch (final IllegalArgumentException illegalArgumentException) {
 560             handleIllegalArgumentException(KIND_DEPLOYMENT, labelSelector, illegalArgumentException);
 561         } catch (final RuntimeException runtimeException) {
 562             handleRuntimeException(KIND_DEPLOYMENT, labelSelector, runtimeException);
 563         }
 564
 565         logger.error("Unable to find any Deployment resources ...");
 566         return Collections.emptyList();
 567     }
 568
 569     @Override
 570     public List<KubernetesResource> getPodResources(final ApiClient apiClient, final String labelSelector)
 571             throws KubernetesRequestProcessingException {
 572         logger.debug("Retrieving Pod using labelSelector: {}", labelSelector);
 573         try {
 574             final CoreV1Api coreV1Api = new CoreV1Api(apiClient);
 575             final V1PodList podList = coreV1Api.listPodForAllNamespaces(null, null, null, labelSelector, null, null,
 576                     null, null, timeoutSeconds, DISABLE_WATCH);
 577
 578             logger.debug("Received Pods: {}", podList);
 579             return getKubernetesResource(podList);
 580
 581         } catch (final ApiException exception) {
 582             handleApiException(KIND_POD, labelSelector, exception);
 583         } catch (final IllegalArgumentException illegalArgumentException) {
 584             handleIllegalArgumentException(KIND_POD, labelSelector, illegalArgumentException);
 585         } catch (final RuntimeException runtimeException) {
 586             handleRuntimeException(KIND_POD, labelSelector, runtimeException);
 587         }
 588
 589         logger.error("Unable to find any Pod resources ...");
 590         return Collections.emptyList();
 591     }
 592
 593     @Override
 594     public List<KubernetesResource> getServiceResources(final ApiClient apiClient, final String labelSelector)
 595             throws KubernetesRequestProcessingException {
 596         logger.debug("Retrieving Service using labelSelector: {}", labelSelector);
 597         try {
 598             final CoreV1Api coreV1Api = new CoreV1Api(apiClient);
 599             final V1ServiceList serviceList = coreV1Api.listServiceForAllNamespaces(null, null, null, labelSelector,
 600                     null, null, null, null, timeoutSeconds, DISABLE_WATCH);
 601
 602             logger.debug("Received Services: {}", serviceList);
 603             return getKubernetesResource(serviceList);
 604
 605         } catch (final ApiException exception) {
 606             handleApiException(KIND_SERVICE, labelSelector, exception);
 607         } catch (final IllegalArgumentException illegalArgumentException) {
 608             handleIllegalArgumentException(KIND_SERVICE, labelSelector, illegalArgumentException);
 609         } catch (final RuntimeException runtimeException) {
 610             handleRuntimeException(KIND_SERVICE, labelSelector, runtimeException);
 611         }
 612
 613         logger.error("Unable to find any Service resources ...");
 614         return Collections.emptyList();
 615     }
 616
 617     @Override
 618     public List<KubernetesResource> getReplicaSetResources(final ApiClient apiClient, final String labelSelector)
 619             throws KubernetesRequestProcessingException {
 620         logger.debug("Retrieving ReplicaSet using labelSelector: {}", labelSelector);
 621         try {
 622             final AppsV1Api appsV1Api = new AppsV1Api(apiClient);
 623             final V1ReplicaSetList replicaSetList = appsV1Api.listReplicaSetForAllNamespaces(null, null, null,
 624                     labelSelector, null, null, null, null, timeoutSeconds, DISABLE_WATCH);
 625
 626             logger.debug("Received ReplicaSets: {}", replicaSetList);
 627             return getKubernetesResource(replicaSetList);
 628
 629         } catch (final ApiException exception) {
 630             handleApiException(KIND_REPLICA_SET, labelSelector, exception);
 631         } catch (final IllegalArgumentException illegalArgumentException) {
 632             handleIllegalArgumentException(KIND_REPLICA_SET, labelSelector, illegalArgumentException);
 633         } catch (final RuntimeException runtimeException) {
 634             handleRuntimeException(KIND_REPLICA_SET, labelSelector, runtimeException);
 635         }
 636
 637         logger.error("Unable to find any ReplicaSet resources ...");
 638         return Collections.emptyList();
 639     }
 640
 641     @Override
 642     public List<KubernetesResource> getDaemonSetResources(final ApiClient apiClient, final String labelSelector)
 643             throws KubernetesRequestProcessingException {
 644         logger.debug("Retrieving DaemonSet using labelSelector: {}", labelSelector);
 645         try {
 646             final AppsV1Api appsV1Api = new AppsV1Api(apiClient);
 647
 648             final V1DaemonSetList daemonSetList = appsV1Api.listDaemonSetForAllNamespaces(null, null, null,
 649                     labelSelector, null, null, null, null, timeoutSeconds, DISABLE_WATCH);
 650
 651             logger.debug("Received DaemonSets: {}", daemonSetList);
 652             return getKubernetesResource(daemonSetList);
 653
 654         } catch (final ApiException exception) {
 655             handleApiException(KIND_DAEMON_SET, labelSelector, exception);
 656         } catch (final IllegalArgumentException illegalArgumentException) {
 657             handleIllegalArgumentException(KIND_DAEMON_SET, labelSelector, illegalArgumentException);
 658         } catch (final RuntimeException runtimeException) {
 659             handleRuntimeException(KIND_DAEMON_SET, labelSelector, runtimeException);
 660         }
 661
 662         logger.error("Unable to find any DaemonSet resources ...");
 663         return Collections.emptyList();
 664     }
 665
 666     @Override
 667     public List<KubernetesResource> getStatefulSetResources(final ApiClient apiClient, final String labelSelector)
 668             throws KubernetesRequestProcessingException {
 669         logger.debug("Retrieving StatefulSet using labelSelector: {}", labelSelector);
 670         try {
 671             final AppsV1Api appsV1Api = new AppsV1Api(apiClient);
 672
 673             final V1StatefulSetList statefulSetList = appsV1Api.listStatefulSetForAllNamespaces(null, null, null,
 674                     labelSelector, null, null, null, null, timeoutSeconds, DISABLE_WATCH);
 675
 676             logger.debug("Received StatefulSets: {}", statefulSetList);
 677             return getKubernetesResource(statefulSetList);
 678
 679         } catch (final ApiException exception) {
 680             handleApiException(KIND_STATEFUL_SET, labelSelector, exception);
 681         } catch (final IllegalArgumentException illegalArgumentException) {
 682             handleIllegalArgumentException(KIND_STATEFUL_SET, labelSelector, illegalArgumentException);
 683         } catch (final RuntimeException runtimeException) {
 684             handleRuntimeException(KIND_STATEFUL_SET, labelSelector, runtimeException);
 685         }
 686
 687         logger.error("Unable to find any StatefulSet resources ...");
 688         return Collections.emptyList();
 689
 690     }
 691
 692     private List<KubernetesResource> getKubernetesResource(final KubernetesListObject kubernetesListObject) {
 693         if (kubernetesListObject != null && kubernetesListObject.getItems() != null) {
 694             final List<KubernetesResource> kubernetesResources = new ArrayList<>();
 695             final List<? extends KubernetesObject> items = kubernetesListObject.getItems();
 696             items.forEach(item -> {
 697                 final String apiVersion =
 698                         item.getApiVersion() != null ? item.getApiVersion() : kubernetesListObject.getApiVersion();
 699                 final String kind = item.getKind() != null ? item.getKind() : kubernetesListObject.getKind();
 700                 kubernetesResources.add(getKubernetesResource(apiVersion, kind, item.getMetadata()));
 701             });
 702             logger.debug("KubernetesResources found: {}", kubernetesResources);
 703             return kubernetesResources;
 704         }
 705         logger.error("kubernetesListObject or items is null {}", kubernetesListObject);
 706         return Collections.emptyList();
 707     }
 708
 709     private KubernetesResource getKubernetesResource(final String apiVersion, final String kind,
 710             final V1ObjectMeta metadata) {
 711         final GroupVersion groupVersion = GroupVersion.parse(apiVersion);
 712         return new KubernetesResource().id(metadata.getUid()).name(metadata.getName()).group(groupVersion.getGroup())
 713                 .version(groupVersion.getVersion()).kind(kind).resourceVersion(metadata.getResourceVersion())
 714                 .namespace(metadata.getNamespace() != null ? metadata.getNamespace() : "")
 715                 .labels(getLabels(metadata.getLabels()));
 716     }
 717
 718     private List<String> getLabels(final Map<String, String> labels) {
 719         if (labels != null) {
 720             final List<String> result = new ArrayList<>();
 721             labels.entrySet().forEach(entry -> result.add(entry.getKey() + "=" + entry.getValue()));
 722             return result;
 723         }
 724         return Collections.emptyList();
 725
 726     }
 727
 728     private boolean isJobReady(final V1Job job) {
 729         if (job.getStatus() != null && job.getStatus().getConditions() != null) {
 730             logger.debug("Received Job with conditions ..");
 731             for (final V1JobCondition condition : job.getStatus().getConditions()) {
 732                 if (JOB_COMPLETE.equalsIgnoreCase(condition.getType())
 733                         && TRUE_STRING.equalsIgnoreCase(condition.getStatus())) {
 734                     logger.debug("Job completed successfully ...");
 735                     return true;
 736                 }
 737                 if (JOB_FAILED.equalsIgnoreCase(condition.getType())
 738                         && TRUE_STRING.equalsIgnoreCase(condition.getStatus())) {
 739                     final String message = "Job failed with reason: " + condition.getReason();
 740                     logger.error(message);
 741                     throw new KubernetesRequestProcessingException(message);
 742
 743                 }
 744             }
 745         }
 746
 747         logger.debug("Job is not ready ...");
 748         return false;
 749     }
 750
 751     private boolean isPodReady(final V1Pod pod) {
 752         final Optional<V1PodCondition> optional = getPodReadyCondition(pod);
 753         if (optional.isPresent()) {
 754             final V1PodCondition condition = optional.get();
 755             if (TRUE_STRING.equalsIgnoreCase(condition.getStatus())) {
 756                 logger.debug("Pod is Ready...");
 757                 return true;
 758             }
 759
 760         }
 761
 762         logger.debug("Pod is not ready ...");
 763         return false;
 764     }
 765
 766     private boolean isServiceReady(final V1Service service) {
 767         logger.debug("V1Service is Ready ...");
 768         return true;
 769     }
 770
 771     private boolean isDeploymentReady(final V1Deployment deployment) {
 772         final V1DeploymentSpec spec = deployment.getSpec();
 773         final V1DeploymentStatus status = deployment.getStatus();
 774
 775         if (status != null && status.getReplicas() != null && status.getAvailableReplicas() != null && spec != null
 776                 && spec.getReplicas() != null) {
 777             if (spec.getReplicas().intValue() == status.getReplicas().intValue()
 778                     && status.getAvailableReplicas().intValue() <= spec.getReplicas().intValue()) {
 779                 logger.debug("Deployment is Ready...");
 780                 return true;
 781             }
 782         }
 783
 784         logger.debug("Deployment is not ready ...");
 785         return false;
 786     }
 787
 788     private boolean isReplicaSetReady(final V1ReplicaSet replicaSet) {
 789         final V1ReplicaSetSpec spec = replicaSet.getSpec();
 790         final V1ReplicaSetStatus status = replicaSet.getStatus();
 791
 792         if (status != null && status.getReadyReplicas() != null && spec != null && spec.getReplicas() != null) {
 793             if (spec.getReplicas().intValue() == status.getReadyReplicas().intValue()) {
 794                 logger.debug("ReplicaSet is Ready...");
 795                 return true;
 796             }
 797         }
 798
 799         logger.debug("ReplicaSet is not ready ...");
 800         return false;
 801     }
 802
 803     private boolean isDaemonSetReady(final V1DaemonSet daemonSet) {
 804
 805         final V1DaemonSetSpec spec = daemonSet.getSpec();
 806         final V1DaemonSetStatus status = daemonSet.getStatus();
 807
 808         if (status == null || spec == null) {
 809             logger.debug("Found null status/spec \n DaemonSet: {}", daemonSet);
 810             return false;
 811         }
 812
 813         if (spec.getUpdateStrategy() != null && spec.getUpdateStrategy().getType() != null) {
 814             if (!ROLLING_UPDATE.equalsIgnoreCase(spec.getUpdateStrategy().getType())) {
 815                 logger.debug("Type is {} returning true", spec.getUpdateStrategy().getType());
 816                 return true;
 817             }
 818         }
 819
 820         if (status.getDesiredNumberScheduled() != null && status.getUpdatedNumberScheduled() != null) {
 821             if (status.getUpdatedNumberScheduled().intValue() != status.getDesiredNumberScheduled().intValue()) {
 822                 logger.debug("DaemonSet is not ready {} out of {} expected pods have been scheduled",
 823                         status.getUpdatedNumberScheduled(), status.getDesiredNumberScheduled());
 824                 return false;
 825             }
 826
 827             if (spec.getUpdateStrategy() != null && spec.getUpdateStrategy().getRollingUpdate() != null
 828                     && status.getNumberReady() != null) {
 829
 830                 final Integer maxUnavailable =
 831                         getMaxUnavailable(spec.getUpdateStrategy().getRollingUpdate().getMaxUnavailable(),
 832                                 status.getDesiredNumberScheduled());
 833
 834                 final int expectedReady = status.getDesiredNumberScheduled().intValue() - maxUnavailable.intValue();
 835                 final int numberReady = status.getNumberReady().intValue();
 836                 if (!(numberReady >= expectedReady)) {
 837                     logger.debug("DaemonSet is not ready {} out of {} expected pods are ready", numberReady,
 838                             expectedReady);
 839                     return false;
 840                 }
 841                 logger.debug("DaemonSet is ready {} out of {} expected pods are ready", numberReady, expectedReady);
 842                 logger.debug("DaemonSet is Ready...");
 843                 return true;
 844             }
 845
 846         }
 847
 848         logger.debug("DaemonSet is not ready ...");
 849         return false;
 850     }
 851
 852     private boolean isStatefulSetReady(final V1StatefulSet statefulSet) {
 853         final V1StatefulSetSpec spec = statefulSet.getSpec();
 854         final V1StatefulSetStatus status = statefulSet.getStatus();
 855
 856         if (status == null || spec == null) {
 857             logger.debug("Found null status/spec \n StatefulSet: {}", statefulSet);
 858             return false;
 859         }
 860
 861         final V1StatefulSetUpdateStrategy updateStrategy = spec.getUpdateStrategy();
 862         if (updateStrategy != null && updateStrategy.getType() != null) {
 863             if (!ROLLING_UPDATE.equalsIgnoreCase(updateStrategy.getType())) {
 864                 logger.debug("Type is {} returning true", updateStrategy.getType());
 865                 return true;
 866             }
 867
 868             // Dereference all the pointers because StatefulSets like them
 869             int partition = 0;
 870             // 1 is the default for replicas if not set
 871             int replicas = 1;
 872             final V1RollingUpdateStatefulSetStrategy rollingUpdate = updateStrategy.getRollingUpdate();
 873             if (rollingUpdate != null && rollingUpdate.getPartition() != null) {
 874                 partition = updateStrategy.getRollingUpdate().getPartition().intValue();
 875             }
 876
 877             if (spec.getReplicas() != null) {
 878                 replicas = spec.getReplicas().intValue();
 879             }
 880
 881             final int expectedReplicas = replicas - partition;
 882
 883             if (status.getUpdatedReplicas() != null && status.getUpdatedReplicas().intValue() < expectedReplicas) {
 884                 logger.debug("StatefulSet is not ready. {} out of {} expected pods have been scheduled",
 885                         status.getUpdatedReplicas(), expectedReplicas);
 886                 return false;
 887             }
 888
 889             if (status.getReadyReplicas() != null && status.getReadyReplicas().intValue() != replicas) {
 890                 logger.debug("StatefulSet is not ready. {} out of {} expected pods are ready",
 891                         status.getReadyReplicas(), replicas);
 892                 return false;
 893             }
 894
 895             logger.debug("{} out of {} expected pods are ready", status.getReadyReplicas(), replicas);
 896             logger.debug("StatefulSet is Ready...");
 897             return true;
 898         }
 899
 900         logger.debug("StatefulSet is not ready ...");
 901         return false;
 902     }
 903
 904     private Integer getMaxUnavailable(final IntOrString maxUnavailable, final Integer desiredNumberScheduled) {
 905         if (maxUnavailable == null) {
 906             logger.debug("maxUnavailable value is {}", maxUnavailable);
 907             return desiredNumberScheduled;
 908         }
 909
 910         if (maxUnavailable.isInteger()) {
 911             logger.debug("maxUnavailable is Integer: {}", maxUnavailable);
 912             return maxUnavailable.getIntValue();
 913         }
 914
 915         if (!maxUnavailable.isInteger()) {
 916             final Integer maxUnavailableIntValue = getIntegerValue(maxUnavailable);
 917             if (maxUnavailableIntValue != null) {
 918                 return (maxUnavailableIntValue.intValue() * desiredNumberScheduled.intValue()) / 100;
 919             }
 920
 921             logger.debug("maxUnavailableIntValue is null {}", maxUnavailableIntValue);
 922         }
 923         logger.debug("Returning desiredNumberScheduled: {}", desiredNumberScheduled);
 924         return desiredNumberScheduled;
 925     }
 926
 927     private Integer getIntegerValue(final IntOrString maxUnavailable) {
 928         try {
 929             final String strValue = maxUnavailable.getStrValue();
 930             if (strValue != null && strValue.length() > 1) {
 931                 if (strValue.contains("%")) {
 932                     final String val = strValue.trim().replace("%", "");
 933                     return Integer.valueOf(val);
 934                 }
 935                 logger.debug("maxUnavailable is not a percentage");
 936             }
 937         } catch (final Exception exception) {
 938             logger.error("Unable to parse maxUnavailable value: {}", maxUnavailable);
 939         }
 940         return null;
 941     }
 942
 943     private void closeWatchSilently(final Watch<?> watch) {
 944         try {
 945             watch.close();
 946         } catch (final IOException exception) {
 947             logger.warn("Unexpected IOException while closing watch suppressing exception");
 948         }
 949     }
 950
 951     private void handleRuntimeException(final String resourceType, final String labelSelector,
 952             final RuntimeException runtimeException) {
 953         if (runtimeException.getCause() instanceof SocketTimeoutException) {
 954             final Throwable cause = runtimeException.getCause();
 955             final String message = "Unexpected SocketTimeoutException occurred while getting " + resourceType
 956                     + " status using labelSelector: " + labelSelector + " message: " + cause.getMessage();
 957             logger.error(message, cause);
 958             throw new KubernetesRequestTimeOut(message, cause);
 959         }
 960         final String message = "Unexpected RuntimeException occurred while getting " + resourceType
 961                 + " status using labelSelector: " + labelSelector;
 962         logger.error(message, runtimeException);
 963         throw new KubernetesRequestProcessingException(message, runtimeException);
 964     }
 965
 966     private void handleApiException(final String resourceType, final String labelSelector,
 967             final ApiException exception) {
 968         final String message = "Unexpected ApiException occurred while getting " + resourceType
 969                 + " status using labelSelector: " + labelSelector + " \n response code: " + exception.getCode()
 970                 + " \n response body: " + exception.getResponseBody();
 971         logger.error(message, exception);
 972         throw new KubernetesRequestProcessingException(message, exception);
 973     }
 974
 975     private void handleIllegalArgumentException(final String resourceType, final String labelSelector,
 976             final IllegalArgumentException illegalArgumentException) {
 977         final String message = "Unexpected IllegalArgumentException occurred while getting " + resourceType
 978                 + " resource using labelSelector: " + labelSelector;
 979         logger.error(message, illegalArgumentException);
 980         throw new KubernetesRequestProcessingException(message, illegalArgumentException);
 981     }
 982
 983     private Optional<V1PodCondition> getPodReadyCondition(final V1Pod pod) {
 984         if (pod.getStatus() != null && pod.getStatus().getConditions() != null) {
 985             final List<V1PodCondition> conditions = pod.getStatus().getConditions();
 986             return conditions.stream().filter(condition -> POD_READY.equals(condition.getType()))
 987                     .peek(condition -> logger.debug("Found {}", condition)).findFirst();
 988
 989         }
 990         logger.warn("Unable to find a {} condition {}", POD_READY, pod.getStatus());
 991         return Optional.empty();
 992     }
 993
 994     /**
 995      * Capturing resource events and objects
 996      *
 997      * @param <T>
 998      * @param apiClient
 999      * @param call
1000      * @param type
1001      * @return
1002      * @throws ApiException
1003      */
1004     private <T> Map<T, String> getReadyResources(final ApiClient apiClient, final Call call, final Type type)
1005             throws ApiException {
1006         final Watch<T> watch = Watch.createWatch(apiClient, call, type);
1007         logger.debug("Listening for {} events ....", type.getTypeName());
1008
1009         final Map<T, String> resources = new HashMap<>();
1010         try {
1011             while (watch.hasNext()) {
1012                 final Response<T> next = watch.next();
1013                 final T object = next.object;
1014                 logger.debug("Received object: {}", object);
1015                 final String event = next.type;
1016                 logger.debug("Received Event: {}", event);
1017                 resources.put(object, event);
1018             }
1019
1020         } finally {
1021             closeWatchSilently(watch);
1022         }
1023         logger.debug("Finished Listening for {} events ....", type.getTypeName());
1024         return resources;
1025
1026     }
1027
1028     private static <T> boolean isResourceReady(final T object, final String eventType, final Predicate<T> predicate) {
1029
1030         switch (eventType) {
1031             case EVENT_TYPE_ADDED:
1032             case EVENT_TYPE_MODIFIED:
1033                 return predicate.test(object);
1034             case EVENT_TYPE_DELETED:
1035                 logger.debug("{} event received marking it as successfully", EVENT_TYPE_DELETED);
1036                 return true;
1037             case EVENT_TYPE_ERROR:
1038                 final String message = "Error event received for " + (object != null ? object.getClass() : "null");
1039                 logger.error(message);
1040                 logger.debug("{} received: {}", (object != null ? object.getClass() : "null"), object);
1041                 throw new KubernetesRequestProcessingException(message);
1042
1043             default:
1044                 logger.warn("Unhandled event received ... ");
1045                 return false;
1046         }
1047
1048     }
1049 }