api/src/main/java/com/epam/pipeline/manager/cluster/PodMonitor.java
@SchedulerLock(name = "PodMonitor_updateStatus", lockAtMostForString = "PT5M")
        public void updateStatus() {
            LOGGER.debug(messageHelper.getMessage(MessageConstants.DEBUG_MONITOR_CHECK_RUNNING));
            List<PipelineRun> running = pipelineRunManager.loadRunningAndTerminatedPipelineRuns();
            for (PipelineRun run : running) {
                if (!run.getExecutionPreferences().getEnvironment().isMonitored()) {
                    if (run.getStatus().isFinal()) {
                        run.setTerminating(false);
                        pipelineRunManager.updatePipelineStatus(run);
                    }
                    LOGGER.debug("Skipping run {} in exec environment {}", run.getId(),
                            run.getExecutionPreferences().getEnvironment());
                    continue;
                }
                LOGGER.debug("RUN ID {} status {} terminating {}", run.getId(), run.getStatus(), run.isTerminating());
                try (KubernetesClient client = kubernetesManager.getKubernetesClient()) {
                    Pod pod = client.pods().inNamespace(kubeNamespace).withName(run.getPodId()).get();
                    //check maybe run was already processed with master node
                    PipelineRun currentRunState = pipelineRunManager.loadPipelineRun(run.getId());
                    if (pod == null && currentRunState.getStatus().isFinal()) {
                        LOGGER.debug("Run ID {} is already in final status {}",
                                run.getId(), currentRunState.getStatus());
                        setRunFinished(currentRunState, pod, client);
                        continue;
                    }
                    if (pod == null || run.isTerminating()) {
                        setRunFinished(run, pod, client);
                    } else {
                        PodStatus status = pod.getStatus();
                        // update pod IP, if it is not set yet
                        if (StringUtils.isEmpty(run.getPodIP())) {
                            if (StringUtils.isEmpty(status.getPodIP())) {
                                notifyIfExceedsThreshold(run, pod, NotificationType.LONG_INIT);
                            } else {
                                run.setPodIP(status.getPodIP());
                                pipelineRunManager.updatePodIP(run);
                            }
                        }

                        if (status.getPhase().equals(KubernetesConstants.POD_SUCCEEDED_PHASE)) {
                            run.setStatus(TaskStatus.SUCCESS);
                            run.setEndDate(DateUtils.now());
                            run.setTerminating(false);
                            //check that all tasks managed to reports its statuses
                            if (!checkChildrenPods(run, client, pod)) {
                                continue;
                            }
                        } else if (status.getPhase().equals(KubernetesConstants.POD_FAILED_PHASE) ||
                                (status.getReason() != null &&
                                        status.getReason().equals(KubernetesConstants.NODE_LOST))) {
                            setRunFinished(run, pod, client);
                        } else {
                            notifyIfExceedsThreshold(run, pod, NotificationType.LONG_RUNNING);
                            continue;
                        }
                    }
                    pipelineRunManager.updatePipelineStatus(run);
                } catch (Exception e) {
                    LOGGER.error(e.getMessage(), e);
                }
            }
            LOGGER.debug(messageHelper.getMessage(MessageConstants.DEBUG_MONITOR_CHECK_FINISHED));
        }
/**
         * Queries statuses of pods of running tasks and adjust task statuses corresponding to pods statuses
         */
Queries statuses of pods of running tasks and adjust task statuses corresponding to pods statuses
Queries statuses of pods of running tasks and adjust task statuses corresponding to pods statuses
@SchedulerLock(name = "PodMonitor_updateStatus", lockAtMostForString = "PT5M")
        public void updateStatus() {
            LOGGER.debug(messageHelper.getMessage(MessageConstants.DEBUG_MONITOR_CHECK_RUNNING));
            List<PipelineRun> running = pipelineRunManager.loadRunningAndTerminatedPipelineRuns();
            for (PipelineRun run : running) {
                if (!run.getExecutionPreferences().getEnvironment().isMonitored()) {
                    if (run.getStatus().isFinal()) {
                        run.setTerminating(false);
                        pipelineRunManager.updatePipelineStatus(run);
                    }
                    LOGGER.debug("Skipping run {} in exec environment {}", run.getId(),
                            run.getExecutionPreferences().getEnvironment());
                    continue;
                }
                LOGGER.debug("RUN ID {} status {} terminating {}", run.getId(), run.getStatus(), run.isTerminating());
                try (KubernetesClient client = kubernetesManager.getKubernetesClient()) {
                    Pod pod = client.pods().inNamespace(kubeNamespace).withName(run.getPodId()).get();
                    PipelineRun currentRunState = pipelineRunManager.loadPipelineRun(run.getId());
                    if (pod == null && currentRunState.getStatus().isFinal()) {
                        LOGGER.debug("Run ID {} is already in final status {}",
                                run.getId(), currentRunState.getStatus());
                        setRunFinished(currentRunState, pod, client);
                        continue;
                    }
                    if (pod == null || run.isTerminating()) {
                        setRunFinished(run, pod, client);
                    } else {
                        PodStatus status = pod.getStatus();
                        if (StringUtils.isEmpty(run.getPodIP())) {
                            if (StringUtils.isEmpty(status.getPodIP())) {
                                notifyIfExceedsThreshold(run, pod, NotificationType.LONG_INIT);
                            } else {
                                run.setPodIP(status.getPodIP());
                                pipelineRunManager.updatePodIP(run);
                            }
                        }
                        if (status.getPhase().equals(KubernetesConstants.POD_SUCCEEDED_PHASE)) {
                            run.setStatus(TaskStatus.SUCCESS);
                            run.setEndDate(DateUtils.now());
                            run.setTerminating(false);
                            if (!checkChildrenPods(run, client, pod)) {
                                continue;
                            }
                        } else if (status.getPhase().equals(KubernetesConstants.POD_FAILED_PHASE) ||
                                (status.getReason() != null &&
                                        status.getReason().equals(KubernetesConstants.NODE_LOST))) {
                            setRunFinished(run, pod, client);
                        } else {
                            notifyIfExceedsThreshold(run, pod, NotificationType.LONG_RUNNING);
                            continue;
                        }
                    }
                    pipelineRunManager.updatePipelineStatus(run);
                } catch (Exception e) {
                    LOGGER.error(e.getMessage(), e);
                }
            }
            LOGGER.debug(messageHelper.getMessage(MessageConstants.DEBUG_MONITOR_CHECK_FINISHED));
        }
@ SchedulerLock ( name = "PodMonitor_updateStatus" , lockAtMostForString = "PT5M" ) public void updateStatus ( ) { LOGGER . debug ( messageHelper . getMessage ( MessageConstants . DEBUG_MONITOR_CHECK_RUNNING ) ) ; List < PipelineRun > running = pipelineRunManager . loadRunningAndTerminatedPipelineRuns ( ) ; for ( PipelineRun run : running ) { if ( ! run . getExecutionPreferences ( ) . getEnvironment ( ) . isMonitored ( ) ) { if ( run . getStatus ( ) . isFinal ( ) ) { run . setTerminating ( false ) ; pipelineRunManager . updatePipelineStatus ( run ) ; } LOGGER . debug ( "Skipping run {} in exec environment {}" , run . getId ( ) , run . getExecutionPreferences ( ) . getEnvironment ( ) ) ; continue ; } LOGGER . debug ( "RUN ID {} status {} terminating {}" , run . getId ( ) , run . getStatus ( ) , run . isTerminating ( ) ) ; try ( KubernetesClient client = kubernetesManager . getKubernetesClient ( ) ) { Pod pod = client . pods ( ) . inNamespace ( kubeNamespace ) . withName ( run . getPodId ( ) ) . get ( ) ; PipelineRun currentRunState = pipelineRunManager . loadPipelineRun ( run . getId ( ) ) ; if ( pod == null && currentRunState . getStatus ( ) . isFinal ( ) ) { LOGGER . debug ( "Run ID {} is already in final status {}" , run . getId ( ) , currentRunState . getStatus ( ) ) ; setRunFinished ( currentRunState , pod , client ) ; continue ; } if ( pod == null || run . isTerminating ( ) ) { setRunFinished ( run , pod , client ) ; } else { PodStatus status = pod . getStatus ( ) ; if ( StringUtils . isEmpty ( run . getPodIP ( ) ) ) { if ( StringUtils . isEmpty ( status . getPodIP ( ) ) ) { notifyIfExceedsThreshold ( run , pod , NotificationType . LONG_INIT ) ; } else { run . setPodIP ( status . getPodIP ( ) ) ; pipelineRunManager . updatePodIP ( run ) ; } } if ( status . getPhase ( ) . equals ( KubernetesConstants . POD_SUCCEEDED_PHASE ) ) { run . setStatus ( TaskStatus . SUCCESS ) ; run . setEndDate ( DateUtils . now ( ) ) ; run . setTerminating ( false ) ; if ( ! checkChildrenPods ( run , client , pod ) ) { continue ; } } else if ( status . getPhase ( ) . equals ( KubernetesConstants . POD_FAILED_PHASE ) || ( status . getReason ( ) != null && status . getReason ( ) . equals ( KubernetesConstants . NODE_LOST ) ) ) { setRunFinished ( run , pod , client ) ; } else { notifyIfExceedsThreshold ( run , pod , NotificationType . LONG_RUNNING ) ; continue ; } } pipelineRunManager . updatePipelineStatus ( run ) ; } catch ( Exception e ) { LOGGER . error ( e . getMessage ( ) , e ) ; } } LOGGER . debug ( messageHelper . getMessage ( MessageConstants . DEBUG_MONITOR_CHECK_FINISHED ) ) ; }
Queries statuses of pods of running tasks and adjust task statuses corresponding to pods statuses
Queries statuses of pods of running tasks and adjust task statuses corresponding to pods statuses
//check maybe run was already processed with master node // update pod IP, if it is not set yet //check that all tasks managed to reports its statuses